WIP: feat: svc-rabbitmq 支持 plan 管理配置

svc-rabbitmq/README.md

@@ -2,6 +2,19 @@
 
 `svc-rabbitmq` 是蓝鲸开发者中心的 RabbitMQ 增强服务模块。
 
+## 相关概念说明
+
+### 集群（Cluster）
+
+集群是 RabbitMQ 的基本单位，包含了 RabbitMQ 的基本信息和配置信息。
+
+### 增强服务方案（Plan）
+
+增强服务方案是增强服务的一种规格，包含了增强服务的基本信息和 RabbitMQ 集群配置信息。
+增强服务方案是增强服务的基本单位，用户可以根据自己的需求选择不同的增强服务方案。
+
+在创建增强服务方案时，需要在 config 中配置 RabbitMQ 集群信息，增强服务会根据集群信息自动创建 RabbitMQ 集群。
+
 ## 本地开发指引
 
 ### 1. 安装 Python 3.11
@@ -50,6 +63,12 @@ export DATABASE_PORT="3306"
 export BKKRILL_ENCRYPT_SECRET_KEY="请参考上面的命令生成"
 # Django Settings
 export DJANGO_SETTINGS_MODULE="svc_rabbitmq.settings"
+
+# redis 配置
+export REDIS_HOST="localhost"
+export REDIS_PORT=6379
+export REDIS_DB=0
+export REDIS_PASS=""
 ```
 
 ### 4. 初始化数据
@@ -59,11 +78,32 @@ export DJANGO_SETTINGS_MODULE="svc_rabbitmq.settings"
 ```bash
 python manage.py migrate
 
+# 通过初始化数据创建服务和方案
 # 初始化数据放在 /data/fixtures 目录下
 # 注意这里是社区版本的初始化数据，如果是其他版本，需要修改 default.json 中 region 的值
 python manage.py loaddata data/fixtures/default.json
 
-## 初始化 rabbitmq 集群，请根据实际情况修改参数的值
+# 启动 shell 环境
+python manage.py shell
+```
+
+在 shell 环境中执行以下命令，配置增强服务方案，会根据方案自动生成 rabbitmq 集群信息(推荐)：
+
+```python
+import json
+from paas_service.models import Plan
+
+# pk 由 /data/fixtures/default.json 确认 
+plan = Plan.objects.get(pk="843127a9-d7a2-4485-b985-076a9b6695d8")
+config = {"host":"10.0.0.1", "admin":"admin","password":"blueking", "port":5672,"api_port":15672}
+plan.config = json.dumps(config)
+plan.save(update_fields=["config"])
+```
+
+不通过方案管理集群，也可以手动创建 RabbitMQ Cluster，可以使用以下命令：
+
+```bash
+## 初始化 rabbitmq 集群，请根据实际情况修改参数的值(如果没有配置增强服务方案，也可以手动创建集群)
 python manage.py register_cluster \
 --name "builtin" \
 --host "10.0.0.1" \
@@ -81,6 +121,7 @@ python manage.py register_cluster \
 使用以下命令启动项目：
 
 ```bash
+python manage.py collectstatic --no-input
 python manage.py runserver 8005
 ```
 

svc-rabbitmq/vendor/apps.py

@@ -16,8 +16,13 @@
 We undertake not to change the open source license (MIT license) applicable
 to the current version of the project delivered to anyone in the future.
 """
+
 from django.apps import AppConfig
 
 
 class VendorConfig(AppConfig):
-    name = 'vendor'
+    name = "vendor"
+
+    def ready(self):
+        # Register signal handlers
+        from . import handlers  # noqa: F401

svc-rabbitmq/vendor/handlers.py

@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+# TencentBlueKing is pleased to support the open source community by making
+# 蓝鲸智云 - PaaS 平台 (BlueKing - PaaS System) available.
+# Copyright (C) 2017 THL A29 Limited, a Tencent company. All rights reserved.
+# Licensed under the MIT License (the "License"); you may not use this file except
+# in compliance with the License. You may obtain a copy of the License at
+#
+#     http://opensource.org/licenses/MIT
+#
+# Unless required by applicable law or agreed to in writing, software distributed under
+# the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+# either express or implied. See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# We undertake not to change the open source license (MIT license) applicable
+# to the current version of the project delivered to anyone in the future.
+
+import json
+import logging
+
+from django.db.models.signals import post_delete, post_save
+from django.dispatch import receiver
+from paas_service.models import Plan
+
+from vendor.models import Cluster, ClusterConfig
+from vendor.serializers import PlanConfigSerializer
+
+logger = logging.getLogger(__name__)
+
+
+@receiver(post_save, sender=Plan)
+def create_related_cluster(sender, instance: Plan, created, **kwargs):
+    plan_config = json.loads(instance.config)
+    slz = PlanConfigSerializer(data=plan_config)
+    if not slz.is_valid():
+        logger.error("serialize plan config error: %s", slz.errors)
+        return
+    config = slz.data
+
+    cluster_config = ClusterConfig(
+        name=f"{instance.name}-cluster",
+        host=config["host"],
+        port=config["port"],
+        management_api=config["management_api"],
+        admin=config["admin"],
+        password=config["password"],
+        cluster_version=config["cluster_version"],
+        plan_id=instance.uuid,
+        cluster_id=plan_config.get("cluster_id"),
+    )
+
+    cluster, created = Cluster.objects.update_or_create_by_cluster_config(cluster_config)
+
+    if created:
+        plan_config["cluster_id"] = cluster.id
+        instance.config = json.dumps(plan_config)
+        instance.save(update_fields=["config"])
+
+
+@receiver(post_delete, sender=Plan)
+def delete_related_cluster(sender, instance: Plan, **kwargs):
+    Cluster.objects.delete_by_plan(plan=instance)

svc-rabbitmq/vendor/models.py

@@ -16,30 +16,78 @@
 We undertake not to change the open source license (MIT license) applicable
 to the current version of the project delivered to anyone in the future.
 """
+
 import json
+import logging
 from contextlib import contextmanager
 from copy import deepcopy
+from dataclasses import dataclass
 from enum import Enum
-from typing import Callable
+from typing import Callable, Optional
 
 from blue_krill.models.fields import EncryptField
 from django.db import models
 from django.utils.functional import cached_property
 from jsonfield import JSONField
-from paas_service.models import AuditedModel, UuidAuditedModel
+from paas_service.models import AuditedModel, Plan, UuidAuditedModel
 
 from .constants import LinkType
 
+logger = logging.getLogger(__name__)
+
 
 class Tag(AuditedModel):
     class Meta(object):
         abstract = True
 
-    instance: 'models.Model'
+    instance: "models.Model"
     key = models.CharField("名称", max_length=64)
     value = models.CharField("值", max_length=128)
 
 
+@dataclass
+class ClusterConfig:
+    name: str
+    host: str
+    port: int
+    management_api: str
+    admin: str
+    password: str
+    cluster_version: str
+    plan_id: str
+    cluster_id: Optional[int] = None
+
+
+class ClusterManager(models.Manager):
+    """集群管理器"""
+
+    def update_or_create_by_cluster_config(self, config: ClusterConfig) -> ("Cluster", bool):
+        return self.update_or_create(
+            id=config.cluster_id,
+            defaults={
+                "name": config.name,
+                "host": config.host,
+                "port": config.port,
+                "management_api": config.management_api,
+                "admin": config.admin,
+                "password": config.password,
+                "version": config.cluster_version,
+                "extra": {"from_plan": config.plan_id},
+            },
+        )
+
+    def delete_by_plan(self, plan: Plan):
+        plan_config = json.loads(plan.config)
+        cluster = self.filter(id=plan_config.get("cluster_id"))
+        cluster.delete()
+
+    def filter_not_from_plan(self) -> models.QuerySet:
+        clusters = self.all()
+        # 筛选不是从 plan 中获取的集群
+        filtered_ids = [c.id for c in clusters if not c.extra.get("from_plan")]
+        return self.filter(id__in=filtered_ids)
+
+
 class Cluster(AuditedModel):
     name = models.CharField("名称", max_length=64)
     host = models.CharField("主机", max_length=64)
@@ -51,6 +99,8 @@ class Cluster(AuditedModel):
     enable = models.BooleanField("是否启用", default=True)
     extra = JSONField("额外信息", default=dict, blank=True, null=True)
 
+    objects = ClusterManager()
+
     def __str__(self):
         return f"{self.name}[{self.pk}]"
 
@@ -68,9 +118,9 @@ class Meta(object):
     link_type = models.IntegerField(
         "连接方式", default=LinkType.empty.value, choices=[(i.value, i.name) for i in LinkType]
     )
-    linked = models.ForeignKey('self', null=True, blank=True, on_delete=models.CASCADE, default=None)
+    linked = models.ForeignKey("self", null=True, blank=True, on_delete=models.CASCADE, default=None)
 
-    def resolve_extend(self, other: 'LinkableModel'):
+    def resolve_extend(self, other: "LinkableModel"):
         """处理继承连接合并细节"""
         for field in self._meta.fields:
             attname = field.attname
@@ -103,7 +153,7 @@ class PolicyTarget(Enum):
 class UserPolicy(LinkableModel):
     """集群下创建 vhost 默认策略，和具体 vhost 无关"""
 
-    resolve_link: Callable[[], 'UserPolicy']
+    resolve_link: Callable[[], "UserPolicy"]
 
     name = models.CharField("名称", max_length=64, null=True)
     enable = models.BooleanField("是否启用", default=True)
@@ -116,10 +166,10 @@ class UserPolicy(LinkableModel):
     cluster_id = models.IntegerField("集群id", blank=True, default=None)
 
     @cached_property
-    def cluster(self) -> 'Cluster':
+    def cluster(self) -> "Cluster":
         return Cluster.objects.filter(pk=self.cluster_id)
 
-    def resolve_extend(self, other: 'UserPolicy'):
+    def resolve_extend(self, other: "UserPolicy"):
         definitions = self.definitions or {}
         definitions.update(other.definitions or {})
         super().resolve_extend(other)
@@ -153,7 +203,7 @@ class LimitType(Enum):
 class LimitPolicy(LinkableModel):
     """集群下创建 vhost 限制机制，和具体 vhost 无关"""
 
-    resolve_link: Callable[[], 'LimitPolicy']
+    resolve_link: Callable[[], "LimitPolicy"]
 
     name = models.CharField("名称", max_length=64, null=True)
     enable = models.BooleanField("是否启用", default=True)
@@ -164,7 +214,7 @@ class LimitPolicy(LinkableModel):
     cluster_id = models.IntegerField("集群id", blank=True, default=None)
 
     @cached_property
-    def cluster(self) -> 'Cluster':
+    def cluster(self) -> "Cluster":
         return Cluster.objects.filter(pk=self.cluster_id)
 
     def __str__(self):
@@ -187,11 +237,11 @@ class InstanceBill(UuidAuditedModel):
     def get_context(self):
         return json.loads(self.context or "{}")
 
-    def set_context(self, context: 'dict'):
+    def set_context(self, context: "dict"):
         self.context = json.dumps(context)
 
     @contextmanager
-    def log_context(self) -> 'dict':
+    def log_context(self) -> "dict":
         context = self.get_context()
         try:
             yield context