feat: 进程状态页提示快照时间(closed #324)

apps/gsekit/meta/models.py

@@ -37,6 +37,7 @@ class KEYS:
         SYNC_BIZ_PROCESS_STATUS_TIMEOUT = "SYNC_BIZ_PROCESS_STATUS_TIMEOUT"
         # 记录所有业务ID,用于同步新业务到灰度列表对比使用
         ALL_BIZ_IDS = "ALL_BIZ_IDS"
+        SYNC_PROC_STATUS_TIME = "SYNC_PROC_STATUS_TIME"
 
     @classmethod
     def process_task_aggregate_info(cls, bk_biz_id: int) -> typing.Dict[str, str]:

apps/gsekit/periodic_tasks/sync_process.py

@@ -25,7 +25,10 @@
 
 @task(ignore_result=True)
 def sync_biz_process_task(bk_biz_id):
-    ProcessHandler(bk_biz_id=bk_biz_id).sync_biz_process()
+    logger.info(f"[sync_biz_process_task] start, bk_biz_id={bk_biz_id}")
+    process_related_infos = ProcessHandler(bk_biz_id=bk_biz_id).sync_biz_process()
+    ProcessHandler(bk_biz_id=bk_biz_id).sync_biz_process_status(process_related_infos=process_related_infos)
+    logger.info(f"[sync_biz_process_task] finished, bk_biz_id={bk_biz_id}")
 
 
 @periodic_task(run_every=django_celery_beat.tzcrontab.TzAwareCrontab(minute="*/10", tz=timezone.get_current_timezone()))
@@ -40,8 +43,6 @@ def sync_process(bk_biz_id=None):
         logger.info(f"[sync_process] start, bk_biz_id={biz_id}")
         countdown = calculate_countdown(count, index)
         sync_biz_process_task.apply_async((biz_id,), countdown=countdown)
-        # TODO 由于GSE接口存在延迟，此处暂停同步状态的周期任务，待GSE优化后再开启
-        # ProcessHandler(bk_biz_id=biz_id).sync_proc_status_to_db()
         logger.info(f"[sync_process] bk_biz_id={biz_id} will be run after {countdown} seconds.")
 
 

apps/gsekit/process/handlers/process.py

@@ -9,6 +9,7 @@
 See the License for the specific language governing permissions and limitations under the License.
 """
 import copy
+from datetime import datetime
 import json
 import operator
 import time
@@ -512,6 +513,8 @@ def sync_biz_process(self):
 
         self.create_process_inst(process_list)
 
+        return process_list
+
     def generate_process_inst_migrate_data(self, process_list: List) -> Dict:
         """计算准备好变更实例所需的数据"""
 
@@ -1109,11 +1112,11 @@ def get_proc_inst_status_infos(
             )
         return proc_inst_status_infos
 
-    def sync_biz_process_status(self):
+    def sync_biz_process_status(self, process_related_infos=None):
 
         begin_time = time.time()
-
-        process_related_infos = batch_request(CCApi.list_process_related_info, {"bk_biz_id": self.bk_biz_id})
+        if not process_related_infos:
+            process_related_infos = batch_request(CCApi.list_process_related_info, {"bk_biz_id": self.bk_biz_id})
         bk_process_ids = [process_info["process"]["bk_process_id"] for process_info in process_related_infos]
         proc_inst_map = defaultdict(list)
         for proc_inst in ProcessInst.objects.filter(bk_process_id__in=bk_process_ids).values(
@@ -1164,6 +1167,14 @@ def sync_biz_process_status(self):
 
         cost_time = time.time() - begin_time
         logger.info("[sync_proc_status] cost: {cost_time}s".format(cost_time=cost_time))
+        # 记录同步时间
+        with transaction.atomic():
+            sync_proc_status_time, _ = GlobalSettings.objects.select_for_update().get_or_create(
+                key=GlobalSettings.KEYS.SYNC_PROC_STATUS_TIME, defaults={"v_json": {}}
+            )
+            sync_proc_status_time.v_json[str(self.bk_biz_id)] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            sync_proc_status_time.save()
+
         return {"cost_time": cost_time}
 
     def process_instance_simple(
@@ -1207,3 +1218,9 @@ def process_info(self, bk_process_id):
             # 若有疑问请联系 CMDB 排查
             raise ProcessNotMatchException(user_bk_process_id=bk_process_id, cc_bk_process_id=cc_process_id)
         return process_list[0]
+
+    def sync_process_status_time(self):
+        sync_process_status_time: Dict[str:str] = GlobalSettings.get_config(
+            key=GlobalSettings.KEYS.SYNC_PROC_STATUS_TIME, default={}
+        )
+        return {"time": sync_process_status_time.get(str(self.bk_biz_id), "")}

apps/gsekit/process/views/process.py

@@ -240,3 +240,8 @@ def process_instance_simple(self, request, bk_biz_id, *args, **kwargs):
                 expression=self.validated_data.get("expression"),
             )
         )
+
+    @swagger_auto_schema(operation_summary="获取业务进程同步时间", tags=ProcessViewTags)
+    @action(detail=False, methods=["GET"])
+    def sync_process_status_time(self, request, bk_biz_id, *args, **kwargs):
+        return Response(ProcessHandler(bk_biz_id=bk_biz_id).sync_process_status_time())