任务数据清理添加特定类型数据清理逻辑 (TencentBlueKing#7119)

* optimization: 任务数据清理添加特定类型数据清理逻辑

* feature: 任务数据清理支持特定业务数据清理

* minor: 任务数据清理支持特定业务

config/default.py

@@ -773,6 +773,8 @@ def monitor_report_config():
 V2_TASK_VALIDITY_DAY = env.V2_TASK_VALIDITY_DAY
 CLEAN_EXPIRED_V2_TASK_BATCH_NUM = env.CLEAN_EXPIRED_V2_TASK_BATCH_NUM
 CLEAN_EXPIRED_V2_TASK_INSTANCE = env.CLEAN_EXPIRED_V2_TASK_INSTANCE
+CLEAN_EXPIRED_V2_TASK_CREATE_METHODS = env.CLEAN_EXPIRED_V2_TASK_CREATE_METHODS
+CLEAN_EXPIRED_V2_TASK_PROJECTS = env.CLEAN_EXPIRED_V2_TASK_PROJECTS
 
 # 是否启动swagger ui
 ENABLE_SWAGGER_UI = env.ENABLE_SWAGGER_UI

env.py

@@ -93,6 +93,9 @@
 V2_TASK_VALIDITY_DAY = int(os.getenv("BKAPP_V2_TASK_VALIDITY_DAY", 730))
 CLEAN_EXPIRED_V2_TASK_BATCH_NUM = int(os.getenv("BKAPP_CLEAN_EXPIRED_V2_TASK_BATCH_NUM", 100))
 CLEAN_EXPIRED_V2_TASK_INSTANCE = bool(os.getenv("BKAPP_CLEAN_EXPIRED_V2_TASK_INSTANCE", False))
+CLEAN_EXPIRED_V2_TASK_CREATE_METHODS = os.getenv("BKAPP_CLEAN_EXPIRED_V2_TASK_CREATE_METHODS", "periodic").split(",")
+# 没有配置则默认清除所有项目
+CLEAN_EXPIRED_V2_TASK_PROJECTS = json.loads(os.getenv("BKAPP_CLEAN_EXPIRED_V2_TASK_PROJECTS", "[]"))
 
 # 是否启动swagger ui
 ENABLE_SWAGGER_UI = os.getenv("BKAPP_ENABLE_SWAGGER_UI", False)

gcloud/contrib/cleaner/tasks.py

@@ -43,15 +43,19 @@ def clean_expired_v2_task_data():
 
         batch_num = settings.CLEAN_EXPIRED_V2_TASK_BATCH_NUM
 
-        ids = (
-            TaskFlowInstance.objects.filter(
-                pipeline_instance__create_time__lt=expire_time, engine_ver=2, pipeline_instance__is_expired=False
-            )
-            .order_by("id")
-            .values("id", "pipeline_instance__instance_id")[:batch_num]
+        qs = TaskFlowInstance.objects.filter(
+            pipeline_instance__create_time__lt=expire_time,
+            engine_ver=2,
+            pipeline_instance__is_finished=True,
+            pipeline_instance__is_revoked=False,
+            pipeline_instance__is_expired=False,
+            create_method__in=settings.CLEAN_EXPIRED_V2_TASK_CREATE_METHODS,
         )
+        if settings.CLEAN_EXPIRED_V2_TASK_PROJECTS:
+            qs = qs.filter(project_id__in=settings.CLEAN_EXPIRED_V2_TASK_PROJECTS)
+        ids = qs.order_by("id").values("id", "pipeline_instance__instance_id")[:batch_num]
         task_ids = [item["id"] for item in ids]
-        logger.info(f"Clean expired task data, task_ids: {task_ids}")
+        logger.info(f"[clean_expired_v2_task_data] Clean expired task data, task_ids: {task_ids}")
         pipeline_instance_ids = [item["pipeline_instance__instance_id"] for item in ids]
         data_to_clean = get_clean_pipeline_instance_data(pipeline_instance_ids)
         tasks = TaskFlowInstance.objects.filter(id__in=task_ids)
@@ -63,6 +67,9 @@ def clean_expired_v2_task_data():
         with transaction.atomic():
             for field, qs in data_to_clean.items():
                 if field not in instance_fields or settings.CLEAN_EXPIRED_V2_TASK_INSTANCE:
+                    logger.info(
+                        f"[clean_expired_v2_task_data] clean field: {field}, qs ids: {qs.values_list('id', flat=True)}"
+                    )
                     qs.delete()
                 elif field == "pipeline_instances":
                     qs.update(is_expired=True)