fix(backend): dumper相关协议和视图接口调整 #2180

dbm-ui/backend/core/encrypt/migrations/0001_initial.py

@@ -12,7 +12,7 @@
 
 from django.db import migrations, models
 
-from backend.core.encrypt.constants import AsymmetricCipherKeyType
+from backend.core.encrypt.constants import AsymmetricCipherConfigType, AsymmetricCipherKeyType
 
 
 class Migration(migrations.Migration):
@@ -33,7 +33,7 @@ class Migration(migrations.Migration):
                 (
                     "name",
                     models.CharField(
-                        choices=[("mysql", "MySQL的非对称秘钥"), ("proxypass", "透传接口的非对称秘钥"), ("cloud", "云区域服务的非对称秘钥")],
+                        choices=AsymmetricCipherConfigType.get_choices(),
                         max_length=128,
                         verbose_name="密钥名称",
                     ),

dbm-ui/backend/core/encrypt/migrations/0002_alter_asymmetriccipherkey_name.py

@@ -2,6 +2,8 @@
 
 from django.db import migrations, models
 
+from backend.core.encrypt.constants import AsymmetricCipherConfigType
+
 
 class Migration(migrations.Migration):
 
@@ -14,7 +16,7 @@ class Migration(migrations.Migration):
             model_name="asymmetriccipherkey",
             name="name",
             field=models.CharField(
-                choices=[("password", "平台密码的非对称秘钥"), ("proxypass", "透传接口的非对称秘钥"), ("cloud", "云区域服务的非对称秘钥")],
+                choices=AsymmetricCipherConfigType.get_choices(),
                 max_length=128,
                 verbose_name="密钥名称",
             ),

dbm-ui/backend/db_meta/admin.py

@@ -149,3 +149,16 @@ class SpecAdmin(admin.ModelAdmin):
         "storage_spec",
     )
     search_fields = ("spec_name", "spec_cluster_type", "spec_machine_type")
+
+
+@admin.register(models.extra_process.ExtraProcessInstance)
+class ExtraProcessInstanceAdmin(admin.ModelAdmin):
+    list_display = (
+        "bk_biz_id",
+        "cluster_id",
+        "bk_cloud_id",
+        "ip",
+        "listen_port",
+        "proc_type",
+    )
+    search_fields = ("cluster_id", "ip", "bk_biz_id", "proc_type")

dbm-ui/backend/db_meta/api/cluster/tendbha/handler.py

@@ -168,6 +168,7 @@ def add_tbinlogdumper(self, add_confs: list):
         添加TBinlogDumper实例的信息
         """
         master = self.cluster.storageinstance_set.get(instance_role=InstanceRole.BACKEND_MASTER)
+        new_dumper_instance_ids = []
         for conf in add_confs:
             tbinlogdumper = ExtraProcessInstance(
                 bk_biz_id=self.cluster.bk_biz_id,
@@ -197,7 +198,9 @@ def add_tbinlogdumper(self, add_confs: list):
                 },
             )
             tbinlogdumper.save()
-            # todo 关联tbinlogdumper订阅配置
+            new_dumper_instance_ids.append(tbinlogdumper.id)
+
+        return new_dumper_instance_ids
 
     @transaction.atomic()
     def switch_tbinlogdumper_for_cluster(self, switch_ids: list):

dbm-ui/backend/db_meta/migrations/0025_dumpersubscribeconfig.py

@@ -9,33 +9,4 @@ class Migration(migrations.Migration):
         ("db_meta", "0024_machine_bk_agent_id"),
     ]
 
-    operations = [
-        migrations.CreateModel(
-            name="DumperSubscribeConfig",
-            fields=[
-                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
-                ("creator", models.CharField(max_length=64, verbose_name="创建人")),
-                ("create_at", models.DateTimeField(auto_now_add=True, verbose_name="创建时间")),
-                ("updater", models.CharField(max_length=64, verbose_name="修改人")),
-                ("update_at", models.DateTimeField(auto_now=True, verbose_name="更新时间")),
-                ("bk_biz_id", models.IntegerField(help_text="关联的业务id，对应cmdb")),
-                ("name", models.CharField(help_text="订阅配置名", max_length=128)),
-                (
-                    "receiver_type",
-                    models.CharField(
-                        choices=[("redis", "redis"), ("kafka", "kafka")], help_text="数据接收端类型", max_length=32
-                    ),
-                ),
-                ("receiver", models.CharField(help_text="接收端域名(IP)", max_length=255)),
-                (
-                    "subscribe",
-                    models.JSONField(help_text="订阅库表信息 eg: [{'db_name': 'xx', 'table_names': [....]}, ...]"),
-                ),
-            ],
-            options={
-                "verbose_name": "dumper数据订阅配置",
-                "verbose_name_plural": "dumper数据订阅配置",
-                "unique_together": {("bk_biz_id", "name")},
-            },
-        ),
-    ]
+    operations = []

dbm-ui/backend/db_meta/migrations/0027_delete_dumpersubscribeconfig.py

@@ -0,0 +1,12 @@
+# Generated by Django 3.2.19 on 2023-11-28 02:54
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("db_meta", "0026_extraprocessinstance_db_meta_ext_cluster_2cbc7c_idx"),
+    ]
+
+    operations = []

dbm-ui/backend/db_meta/migrations/0028_merge_20231207_1619.py

@@ -0,0 +1,13 @@
+# Generated by Django 3.2.19 on 2023-12-07 08:19
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("db_meta", "0027_delete_dumpersubscribeconfig"),
+        ("db_meta", "0027_extraprocessinstance_phase"),
+    ]
+
+    operations = []

dbm-ui/backend/db_meta/models/__init__.py

@@ -14,6 +14,7 @@
 from .cluster_entry import CLBEntryDetail, ClusterEntry, PolarisEntryDetail
 from .cluster_monitor import AppMonitorTopo, ClusterMonitorTopo
 from .db_module import BKModule, DBModule
+from .extra_process import ExtraProcessInstance
 from .group import Group, GroupInstance
 from .instance import ProxyInstance, StorageInstance
 from .machine import Machine

dbm-ui/backend/db_services/mysql/dumper/admin.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-DB管理系统(BlueKing-BK-DBM) available.
+Copyright (C) 2017-2023 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 https://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.
+"""
+from django.contrib import admin
+
+from . import models
+
+
+@admin.register(models.DumperSubscribeConfig)
+class DumperSubscribeConfigAdmin(admin.ModelAdmin):
+    list_display = ("bk_biz_id", "name", "add_type")
+    list_filter = ("bk_biz_id", "name", "add_type")
+    search_fields = ("bk_biz_id", "name", "add_type")

dbm-ui/backend/db_services/mysql/dumper/filters.py

@@ -11,34 +11,75 @@
 
 from django.utils.translation import ugettext_lazy as _
 from django_filters import rest_framework as filters
+from django_mysql.models.functions import JSONExtract
 
 from backend.db_meta.models import Cluster
-from backend.db_meta.models.dumper import DumperSubscribeConfig
 from backend.db_meta.models.extra_process import ExtraProcessInstance
+from backend.db_services.mysql.dumper.models import DumperSubscribeConfig
+
+
+class DumperConfigListFilter(filters.FilterSet):
+    db_name = filters.CharFilter(field_name="db_name", method="filter_db_name", label=_("库名"))
+    table_name = filters.CharFilter(field_name="table_name", method="filter_table_name", label=_("表名"))
+
+    def filter_db_name(self, queryset, name, value):
+        return queryset.annotate(db_name=JSONExtract("repl_tables", "$[*].db_name")).filter(db_name__contains=value)
+
+    def filter_table_name(self, queryset, name, value):
+        return queryset.annotate(table_names=JSONExtract("repl_tables", "$[*].table_names[*]")).filter(
+            table_names__contains=value
+        )
+
+    class Meta:
+        model = DumperSubscribeConfig
+        fields = ["db_name", "table_name"]
 
 
 class DumperInstanceListFilter(filters.FilterSet):
+    config_name = filters.CharFilter(field_name="config_name", method="filter_config_name", label=_("dumper订阅规则"))
     ip = filters.CharFilter(field_name="ip", lookup_expr="icontains", label=_("实例IP"))
+    address = filters.CharFilter(field_name="address", method="filter_address", label=_("实例IP:Port"))
     dumper_id = filters.CharFilter(field_name="dumper_id", method="filter_dumper_id", label=_("dumper id"))
     source_cluster = filters.CharFilter(field_name="source_cluster", method="filter_source_cluster", label=_("源集群"))
-    receiver_type = filters.CharFilter(field_name="type", method="filter_receiver_type", label=_("接收端类型"))
+    protocol_type = filters.CharFilter(field_name="type", method="filter_protocol_type", label=_("接收端类型"))
+    target_address = filters.CharFilter(field_name="target_address", method="filter_target_address", label=_("接收端地址"))
     start_time = filters.DateTimeFilter(field_name="create_at", lookup_expr="lte")
     end_time = filters.DateTimeFilter(field_name="create_at", lookup_expr="gte")
 
+    def filter_config_name(self, queryset, name, value):
+        bk_biz_id = self.request.parser_context["kwargs"]["bk_biz_id"]
+        try:
+            dumper_config = DumperSubscribeConfig.objects.get(bk_biz_id=bk_biz_id, name=value)
+            return queryset.filter(id__in=dumper_config.dumper_process_ids)
+        except DumperSubscribeConfig.DoesNotExist:
+            return queryset
+
+    def filter_address(self, queryset, name, value):
+        ip, port = value.split(":")
+        return queryset.filter(ip=ip, listen_port=port)
+
     def filter_dumper_id(self, queryset, name, value):
-        return queryset.filter(extra_config__dumper_id__icontains=value)
+        return queryset.filter(extra_config__dumper_id=int(value))
 
     def filter_source_cluster(self, queryset, name, value):
         cluster_ids = list(Cluster.objects.filter(immute_domain__icontains=value).values_list("id", flat=True))
         return queryset.filter(cluster_id__in=cluster_ids)
 
-    def filter_receiver_type(self, queryset, name, value):
-        bk_biz_id = self.request.parser_context["kwargs"]["bk_biz_id"]
-        config_ids = list(
-            DumperSubscribeConfig.objects.filter(bk_biz_id=bk_biz_id, receiver_type=value).values_list("id", flat=True)
-        )
-        return queryset.filter(extra_config__dumper_config_id__in=config_ids)
+    def filter_protocol_type(self, queryset, name, value):
+        return queryset.filter(extra_config__protocol_type__icontains=value)
+
+    def filter_target_address(self, queryset, name, value):
+        return queryset.filter(extra_config__target_address__icontains=value)
 
     class Meta:
         model = ExtraProcessInstance
-        fields = ["ip", "dumper_id", "source_cluster", "receiver_type", "start_time", "end_time"]
+        fields = [
+            "config_name",
+            "ip",
+            "address",
+            "dumper_id",
+            "source_cluster",
+            "protocol_type",
+            "start_time",
+            "end_time",
+        ]

dbm-ui/backend/db_services/mysql/dumper/migrations/0001_initial.py

@@ -0,0 +1,41 @@
+# Generated by Django 3.2.19 on 2023-11-28 09:19
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="DumperSubscribeConfig",
+            fields=[
+                ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("creator", models.CharField(max_length=64, verbose_name="创建人")),
+                ("create_at", models.DateTimeField(auto_now_add=True, verbose_name="创建时间")),
+                ("updater", models.CharField(max_length=64, verbose_name="修改人")),
+                ("update_at", models.DateTimeField(auto_now=True, verbose_name="更新时间")),
+                ("bk_biz_id", models.IntegerField(help_text="关联的业务id，对应cmdb")),
+                ("name", models.CharField(help_text="订阅配置名", max_length=128)),
+                (
+                    "add_type",
+                    models.CharField(
+                        choices=[("full_sync", "全量同步"), ("incr_sync", "增量同步")], help_text="数据同步方式", max_length=32
+                    ),
+                ),
+                (
+                    "repl_tables",
+                    models.JSONField(help_text="订阅库表信息 eg: [{'db_name': 'xx', 'table_names': [....]}, ...]"),
+                ),
+                ("dumper_process_ids", models.JSONField(help_text="订阅配置关联的订阅实例列表")),
+            ],
+            options={
+                "verbose_name": "dumper数据订阅配置",
+                "verbose_name_plural": "dumper数据订阅配置",
+                "unique_together": {("bk_biz_id", "name")},
+            },
+        ),
+    ]

dbm-ui/backend/db_services/mysql/dumper/migrations/0002_alter_dumpersubscribeconfig_dumper_process_ids.py

@@ -0,0 +1,18 @@
+# Generated by Django 3.2.19 on 2023-12-01 07:01
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("dumper", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="dumpersubscribeconfig",
+            name="dumper_process_ids",
+            field=models.JSONField(blank=True, default=list, help_text="订阅配置关联的订阅实例列表", null=True),
+        ),
+    ]

dbm-ui/backend/db_meta/enums/dumper.py → dbm-ui/backend/db_services/mysql/dumper/migrations/__init__.py

@@ -8,14 +8,3 @@
 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.
 """
-
-from django.utils.translation import gettext_lazy as _
-
-from blue_krill.data_types.enum import EnumField, StructuredEnum
-
-
-class DumperReceiverType(str, StructuredEnum):
-    """dumper接收端枚举类型"""
-
-    REDIS = EnumField("redis", _("redis"))
-    KAFKA = EnumField("kafka", _("kafka"))

dbm-ui/backend/db_services/mysql/dumper/mock_data.py

@@ -9,6 +9,29 @@
 specific language governing permissions and limitations under the License.
 """
 
+DUMPER_CONFIG_DATA = {
+    "id": 1,
+    "repl_tables": [{"db_name": "db1", "table_names": ["a", "b"]}],
+    "instance_count": 4,
+    "dumper_instances": [
+        {
+            "source_cluster_domain": "spider.xxx.dba.db",
+            "protocol_type": "kafka",
+            "dumper_id": 1,
+            "target_address": "kafka.com:8000",
+        }
+    ],
+    "running_tickets": [1, 2, 3],
+    "creator": " ",
+    "create_at": "2000-01-01 19:11:11",
+    "updater": " ",
+    "update_at": "2000-01-01 19:11:11",
+    "bk_biz_id": 3,
+    "name": "fff",
+    "add_type": "kafka",
+    "dumper_process_ids": [1],
+}
+
 DUMPER_INSTANCE_LIST_DATA = [
     {
         "id": 1,
@@ -45,5 +68,10 @@
         },
         "dumper_id": 1,
         "area_name": 1,
+        "add_type": "null",
+        "l5_modid": "null",
+        "l5_cmdid": "null",
+        "kafka_user": "null",
+        "kafka_pwd": "null",
     }
 ]