fix: 审计队列消息堆积问题 --story=119869959

1. 增加sops接口睡眠超时限制
2. 自动处理任务捕获等待超时异常
3. 增加运行日志
4. 修复单测运行环境异常

src/backend/api/bk_sops/default.py

@@ -33,13 +33,15 @@ class BKSOps(BkApiResource, abc.ABC):
     base_url = BK_SOPS_API_URL
     platform_authorization = True
     rate_limit = settings.SOPS_API_RATE_LIMIT
+    max_retry_duration = settings.SOPS_API_MAX_RETRY_DURATION
 
     def perform_request(self, validated_request_data):
         return Throttler(
             config=ThrottlerConfig(
                 func=super().perform_request,
                 key=f"{self.__module__}.{self.__class__.__name__}",
                 rate=self.rate_limit,
+                max_retry_duration=self.max_retry_duration,
             )
         )(validated_request_data)
 

src/backend/config/default.py

@@ -319,6 +319,7 @@
 )
 setup(throttler_config)
 SOPS_API_RATE_LIMIT = os.getenv("BKAPP_SOPS_API_RATE_LIMIT", "10/s")
+SOPS_API_MAX_RETRY_DURATION = float(os.getenv("BKAPP_SOPS_API_MAX_RETRY_DURATION", 60))
 SOPS_OPERATE_API_RATE_LIMIT = os.getenv("BKAPP_SOPS_OPERATE_API_RATE_LIMIT", "10/s")
 ITSM_API_RATE_LIMIT = os.getenv("BKAPP_ITSM_API_RATE_LIMIT", "10/s")
 

src/backend/services/web/risk/handlers/risk.py

@@ -70,6 +70,7 @@ def generate_risk_from_event(self, start_time: datetime.datetime, end_time: date
                     from services.web.risk.tasks import process_risk_ticket
 
                     process_risk_ticket(risk_id=risk.risk_id)
+                    logger.info("[CreateRiskSuccess] Risk %s", risk.risk_id)
             except Exception as err:  # NOCC:broad-except(需要处理所有错误)
                 logger.exception("[CreateRiskFailed] Event: %s; Error: %s", json.dumps(event), err)
                 ErrorMsgHandler(

src/backend/services/web/risk/handlers/ticket.py

@@ -23,6 +23,8 @@
 
 from bk_resource import api, resource
 from bk_resource.settings import bk_resource_settings
+from blueapps.utils.logger import logger
+from client_throttler.exceptions import RetryTimeout
 from django.conf import settings
 from django.db import transaction
 from django.utils import timezone
@@ -436,6 +438,16 @@ class AutoProcess(RiskFlowBaseHandler):
     name = gettext_lazy("自动套餐处理")
     allowed_status = [RiskStatus.AUTO_PROCESS, RiskStatus.FOR_APPROVE, RiskStatus.AWAIT_PROCESS]
 
+    def run(self, *args, **kwargs) -> None:
+        """
+        处理风险单
+        """
+
+        try:
+            return super().run(*args, **kwargs)
+        except RetryTimeout as e:
+            logger.warning("[Auto]Process] RetryTimeout: %s", e)
+
     def process(self, pa_config: dict = None, **kwargs) -> dict:
         """
         三种发起方式

src/backend/services/web/risk/tasks.py

@@ -128,6 +128,7 @@ def process_risk_ticket(*, risk_id: str = None):
     if settings.ENABLE_PROCESS_RISK_WHITELIST:
         risks = risks.filter(strategy_id__in=settings.PROCESS_RISK_WHITELIST)
 
+    logger_celery.info("[ProcessRiskTicket] Total %s", len(risks))
     # 逐个处理
     for risk in risks:
         if settings.ENABLE_MULTI_PROCESS_RISK:

src/backend/tests/core/test_render.py

@@ -1,10 +1,12 @@
 # -*- coding: utf-8 -*-
 from core.render import Jinja2Renderer
 from services.web.risk.render import RiskTitleUndefined
+from tests.base import TestCase
 
 
-def test_jinja_render():
-    template_value = "render: {{content.title}} {{content.content}} {{title}} {{test}}"
-    context = {"content": {"title": "title", "content": "content"}, "title": "title"}
-    actual = Jinja2Renderer(undefined=RiskTitleUndefined).jinja_render(template_value, context)
-    assert actual == "render: title content title (未获取到变量值:test)"
+class TestJinja2Render(TestCase):
+    def test_jinja_render(self):
+        template_value = "render: {{content.title}} {{content.content}} {{title}} {{test}}"
+        context = {"content": {"title": "title", "content": "content"}, "title": "title"}
+        actual = Jinja2Renderer(undefined=RiskTitleUndefined).jinja_render(template_value, context)
+        assert actual == "render: title content title (未获取到变量值:test)"