fix(redis): cluster架构不要重复发起自愈 #10042

1. 自愈添加消息独立推送（忽略自愈、自愈发起、自愈失败）
2. 自愈修复cluster 集群类型，在自愈进行时再次发起自愈单子
3. 自愈添加自愈状态监控

Author: xiepaup

dbm-ui/backend/db_periodic_task/local_tasks/redis_autofix.py

@@ -8,20 +8,26 @@
 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.
 """
+import json
 import logging
 
 from celery.schedules import crontab
+from django.utils import timezone
 from django.utils import timezone as datetime
+from django.utils.translation import ugettext as _
 
 from backend.db_periodic_task.local_tasks.register import register_periodic_task
 from backend.db_services.redis.autofix.bill import generate_autofix_ticket
 from backend.db_services.redis.autofix.enums import AutofixItem, AutofixStatus
+from backend.db_services.redis.autofix.message import send_msg_2_qywx
 from backend.db_services.redis.autofix.models import RedisAutofixCore, RedisAutofixCtl
 from backend.db_services.redis.autofix.watcher import (
     get_4_next_watch_ID,
     save_swithed_host_by_cluster,
     watcher_get_by_hosts,
 )
+from backend.ticket.constants import TicketStatus
+from backend.ticket.models import Ticket
 from backend.utils.time import datetime2str
 
 logger = logging.getLogger("root")
@@ -74,3 +80,49 @@ def start_autofix_flow():
         return
 
     generate_autofix_ticket(fixlists)
+
+
+@register_periodic_task(run_every=crontab(minute="*/1"))
+def watch_autofix_flow():
+    """监控自愈状态,已期进行流转"""
+
+    try:
+        fixlists = RedisAutofixCore.objects.filter(
+            deal_status__in=[
+                AutofixStatus.AF_WFLOW.value,
+                AutofixStatus.AF_REUSE.value,
+                AutofixStatus.AF_RUNNING.value,
+            ]
+        )
+    except RedisAutofixCore.DoesNotExist:
+        logger.info("waiting other flow items ... ")
+        return
+    if len(fixlists) == 0:
+        logger.info("waiting other flow  items ... ")
+        return
+
+    for flow in fixlists:
+        try:
+            ticket_obj = Ticket.objects.get(id=flow.ticket_id)
+
+            if ticket_obj.status == TicketStatus.RUNNING.value:
+                flow.deal_status = AutofixStatus.AF_RUNNING.value
+            elif ticket_obj.status == TicketStatus.SUCCEEDED.value:
+                flow.deal_status = AutofixStatus.AF_SUCC.value
+            else:
+                flow.deal_status = AutofixStatus.AF_FAIL.value
+                flow.status_version = ticket_obj.status
+                msgs, title = {}, _("{} - 自愈失败☹️".format(flow.immute_domain))
+                msgs["BKID"] = flow.bk_biz_id
+                msgs[_("集群类型")] = flow.cluster_type
+                msgs[_("故障机器")] = json.dumps(flow.fault_machines)
+                msgs[_("失败原因")] = flow.status_version
+                send_msg_2_qywx(title, msgs)
+
+            flow.update_at = datetime2str(datetime.datetime.now(timezone.utc))
+            flow.save(update_fields=["status_version", "deal_status", "update_at"])
+        except Exception as e:
+            flow.deal_status = AutofixStatus.AF_UNKOWN.value
+            flow.status_version = e
+            flow.update_at = datetime2str(datetime.datetime.now(timezone.utc))
+            flow.save(update_fields=["status_version", "deal_status", "update_at"])

dbm-ui/backend/db_periodic_task/local_tasks/redis_clusternodes_update/task.py

@@ -385,13 +385,15 @@ def start_autofix_flow(self, cluster: Cluster):
             if all([slave_obj.status == InstanceStatus.UNAVAILABLE.value for slave_obj in slave_objs]):
                 # 该机器所有实例都unrunning了
                 # 继续检测该机器 10 分钟内没被加入到自愈流程中
-                mins10_ago = datetime.now(timezone.utc) - timedelta(minutes=10)
+                # 需要根据 时间 + 单据状态 (10分钟不够)
+                mins10_ago = datetime.now(timezone.utc) - timedelta(minutes=600)
                 rows = RedisAutofixCore.objects.filter(cluster_id=cluster.id, create_at__gt=mins10_ago)
                 exists = False
                 for row in rows:
                     if ip in [item["ip"] for item in json.loads(row.fault_machines)]:
-                        exists = True
-                        break
+                        if row.deal_status in [AutofixStatus.AF_RUNNING.value, AutofixStatus.AF_TICKET.value]:
+                            exists = True
+                            break
                 if not exists:
                     fault_machines.append({"instance_type": slave_objs[0].machine_type, "ip": ip})
         if len(fault_machines) > 0:

dbm-ui/backend/db_services/redis/autofix/bill.py

@@ -33,6 +33,7 @@
 from backend.utils.time import datetime2str
 
 from .enums import AutofixStatus
+from .message import send_msg_2_qywx
 from .models import RedisAutofixCore
 
 logger = logging.getLogger("root")
@@ -151,6 +152,14 @@ def create_ticket(cluster: RedisAutofixCore, cluster_ids: list, redis_proxies: l
 
     # 初始化builder类
     try:
+        msgs, title = {}, _("{} - 发起自愈".format(cluster.immute_domain))
+        msgs[_("BKID")] = cluster.bk_biz_id
+        msgs[_("流程ID")] = ticket.id
+        msgs[_("集群IDS")] = cluster_ids
+        msgs[_("集群类型")] = cluster.cluster_type
+        msgs[_("故障机S")] = json.dumps(ips)
+        send_msg_2_qywx(title, msgs)
+
         builder = BuilderFactory.create_builder(ticket)
         builder.patch_ticket_detail()
         builder.init_ticket_flows()

dbm-ui/backend/db_services/redis/autofix/enums.py

@@ -27,6 +27,7 @@ class AutofixItem(str, StructuredEnum):
 
     AUTOFIX_ENABLE = EnumField("enable", _("自愈开关"))
     DBHA_ID = EnumField("last_id", _("监控到的id"))
+    CHAT_IDS = EnumField("chat_ids", _("群消息IDS"))
     IGNORE_APPS = EnumField("ignore_apps", _("忽略自愈的APP列表"))
     IGNORE_DOMAINS = EnumField("ignore_domains", _("忽略自愈的集群列表"))
 
@@ -36,8 +37,12 @@ class AutofixStatus(str, StructuredEnum):
 
     AF_INIT = EnumField("initautofix", _("初始化"))
     AF_TICKET = EnumField("initticket", _("创建单据"))
-    AF_SFLOW = EnumField("startflow", _("发起flow流程"))
-    AF_WFLOW = EnumField("watchflow", _("监控流程完成状态"))
-    AF_IGNORE = EnumField("ignore", _("不支持自愈"))
+    AF_REUSE = EnumField("reuse_host", _("尝试复用"))
+    AF_REUSE_SUCCESS = EnumField("reuse_succ", _("复用搞定"))
+    AF_REUSE_FAILED = EnumField("reuse_fail", _("尝试失败"))
+    AF_WFLOW = EnumField("watchflow", _("监控状态"))
+    AF_IGNORE = EnumField("ignore", _("不理解类型"))
     AF_SUCC = EnumField("success", _("自愈成功"))
+    AF_RUNNING = EnumField("running", _("拼命干活中"))
     AF_FAIL = EnumField("fail", _("自愈失败"))
+    AF_UNKOWN = EnumField("unkown", _("未至之境"))

dbm-ui/backend/db_services/redis/autofix/message.py

@@ -0,0 +1,56 @@
+# -*- 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.
+"""
+import datetime
+import json
+import logging
+
+from django.utils import timezone
+from django.utils.translation import ugettext as _
+
+from backend.configuration.constants import DBType
+from backend.configuration.models.dba import DBAdministrator
+from backend.core.notify.handlers import CmsiHandler
+from backend.db_meta.models import AppCache
+from backend.utils.time import date2str
+
+from .enums import AutofixItem
+from .models import RedisAutofixCtl
+
+logger = logging.getLogger("root")
+
+
+def send_msg_2_qywx(sub_title: str, msgs):
+    msg_ids = []
+    try:
+        msg_item = RedisAutofixCtl.objects.filter(ctl_name=AutofixItem.CHAT_IDS.value).get()
+        if msg_item:
+            msg_ids = json.loads(msg_item.ctl_value)
+    except RedisAutofixCtl.DoesNotExist:
+        RedisAutofixCtl.objects.create(
+            bk_cloud_id=0, bk_biz_id=0, ctl_value=json.dumps("[]"), ctl_name=AutofixItem.CHAT_IDS.value
+        ).save()
+
+    if len(msg_ids) == 0:
+        return
+
+    bk_biz_id = msgs["BKID"]
+    redis_DBA = DBAdministrator.get_biz_db_type_admins(bk_biz_id=bk_biz_id, db_type=DBType.Redis.value)
+    app_info = AppCache.objects.get(bk_biz_id=bk_biz_id)
+
+    content = _(">> Tendis-{}\n".format(sub_title))
+    for k, v in msgs.items():
+        if k == "BKID":
+            content += _("业务信息 : {}(#{},{})\n".format(app_info.bk_biz_name, app_info.bk_biz_id, app_info.db_app_abbr))
+            content += _("业务DBA : {}(@{})\n".format(redis_DBA[0], redis_DBA[0]))
+        else:
+            content += _("{} : {}\n".format(k, v))
+    content += _("消息时间 : {}\n".format(date2str(datetime.datetime.now(timezone.utc), "%Y-%m-%d %H:%M:%S")))
+    CmsiHandler(_("Tendis自愈"), content, msg_ids).send_wecom_robot()

dbm-ui/backend/db_services/redis/autofix/watcher.py

@@ -15,6 +15,7 @@
 
 from django.utils import timezone
 from django.utils.crypto import get_random_string
+from django.utils.translation import ugettext_lazy as _
 
 from backend.components.hadb.client import HADBApi
 from backend.constants import DEFAULT_BK_CLOUD_ID
@@ -25,6 +26,7 @@
 
 from .const import REDIS_SWITCH_WAITER, SWITCH_MAX_WAIT_SECONDS, SWITCH_SMALL, RedisSwitchHost, RedisSwitchWait
 from .enums import AutofixItem, AutofixStatus, DBHASwitchResult
+from .message import send_msg_2_qywx
 from .models import RedisAutofixCore, RedisAutofixCtl, RedisIgnoreAutofix
 
 logger = logging.getLogger("root")
@@ -243,7 +245,7 @@ def save_swithed_host_by_cluster(batch_small: int, switch_hosts: Dict):
 
 # 把需要忽略自愈的保存起来
 def save_ignore_host(switched_host: RedisSwitchHost, msg):
-    RedisIgnoreAutofix.objects.update_or_create(
+    rst = RedisIgnoreAutofix.objects.update_or_create(
         bk_cloud_id=DEFAULT_BK_CLOUD_ID,
         bk_biz_id=switched_host.bk_biz_id,
         cluster_id=switched_host.cluster_id,
@@ -259,3 +261,18 @@ def save_ignore_host(switched_host: RedisSwitchHost, msg):
         sw_result=json.dumps(switched_host.sw_result),
         ignore_msg=msg,
     )
+
+    if switched_host.cluster_type in [
+        ClusterType.TwemproxyTendisSSDInstance.value,
+        ClusterType.TendisTwemproxyRedisInstance.value,
+        ClusterType.TendisPredixyRedisCluster.value,
+        ClusterType.TendisPredixyTendisplusCluster.value,
+        ClusterType.TendisRedisInstance.value,
+    ]:
+        msgs, title = {}, _("{}-忽略自愈".format(switched_host.immute_domain))
+        msgs[_("BKID")] = switched_host.bk_biz_id
+        msgs[_("故障IP")] = switched_host.ip
+        msgs[_("实例类型")] = switched_host.instance_type
+        msgs[_("ByDBHA")] = json.dumps(switched_host.sw_result)
+        msgs[_("xxxxxx")] = rst
+        send_msg_2_qywx(title, msgs)