[waterflow] 添加流程数据定期清理机制

Author: FUNK6666

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/scheduletasks/CleanContextSchedule.java

@@ -0,0 +1,62 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved.
+ *  This file is a part of the ModelEngine Project.
+ *  Licensed under the MIT License. See License.txt in the project root for license information.
+ *--------------------------------------------------------------------------------------------*/
+
+package modelengine.fit.waterflow.flowsengine.biz.service.scheduletasks;
+
+import modelengine.fit.waterflow.common.utils.SleepUtil;
+import modelengine.fit.waterflow.flowsengine.domain.flows.context.repo.flowcontext.FlowContextRepo;
+import modelengine.fit.waterflow.flowsengine.domain.flows.context.repo.flowtrace.FlowTraceRepo;
+import modelengine.fitframework.annotation.Component;
+import modelengine.fitframework.annotation.Fit;
+import modelengine.fitframework.annotation.Value;
+import modelengine.fitframework.log.Logger;
+import modelengine.fitframework.schedule.annotation.Scheduled;
+import modelengine.fitframework.transaction.Transactional;
+
+import java.util.List;
+
+/**
+ * 定时清理流程中已完成的context
+ * 包括成功、失败、终止的流程数据
+ *
+ * @author 杨祥宇
+ * @since 2025/04/02
+ */
+@Component
+public class CleanContextSchedule {
+    private static final Logger log = Logger.get(CleanContextSchedule.class);
+    private static final int LIMIT = 1000;
+    private final FlowTraceRepo flowTraceRepo;
+    private final FlowContextRepo flowContextRepo;
+    private final int expiredDays;
+
+    public CleanContextSchedule(FlowTraceRepo flowTraceRepo, @Fit(alias = "flowContextPersistRepo") FlowContextRepo
+            flowContextRepo, @Value("${jane.flowsEngine.contextExpiredDays}") int expiredDays) {
+        this.flowTraceRepo = flowTraceRepo;
+        this.flowContextRepo = flowContextRepo;
+        this.expiredDays = expiredDays;
+    }
+
+    /**
+     * 每天凌晨3点定时清理超期EXPIRED_DAYS天的流程运行数据
+     */
+    @Scheduled(strategy = Scheduled.Strategy.CRON, value = "0 0 3 * * ?")
+    @Transactional
+    public void cleanContextSchedule() {
+        log.info("Start clean flow expired contexts");
+        try {
+            List<String> traceIds = flowTraceRepo.getExpiredTrace(expiredDays, LIMIT);
+            while (!traceIds.isEmpty()) {
+                flowContextRepo.deleteByTraceIdList(traceIds);
+                flowTraceRepo.deleteByIdList(traceIds);
+                traceIds = flowTraceRepo.getExpiredTrace(expiredDays, LIMIT);
+                SleepUtil.sleep(60000);
+            }
+        } catch (Exception ex) {
+            log.error("Clean context error, error message: {}" + ex.getMessage());
+        }
+    }
+}

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextPersistRepo.java

@@ -736,4 +736,12 @@ public boolean hasContextWithStatusAtPosition(List<String> statusList, String tr
     public String getTransIdByTrace(String traceId) {
         return contextMapper.getTransIdByTrace(traceId);
     }
+
+    @Override
+    public void deleteByTraceIdList(List<String> traceIdList) {
+        if (traceIdList.isEmpty()) {
+            return;
+        }
+        contextMapper.deleteByTraceIdList(traceIdList);
+    }
 }

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextRepo.java

@@ -671,5 +671,14 @@ default List<FlowContext<String>> getWithoutFlowDataByToBatch(List<String> toBat
      * @return trans id
      */
     String getTransIdByTrace(String traceId);
+
+    /**
+     * 根据trace列表删除对应的context数据
+     *
+     * @param traceIdList trace id列表
+     */
+    default void deleteByTraceIdList(List<String> traceIdList) {
+        throw new WaterflowException(ErrorCodes.FLOW_ENGINE_DATABASE_NOT_SUPPORT, "deleteByTraceIdList");
+    }
 }
 

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowtrace/DefaultFlowTraceRepo.java

@@ -101,6 +101,13 @@ public void deleteByIdList(List<String> traceIds) {
         flowTraceMapper.deleteByIdList(traceIds);
     }
 
+    @Override
+    public List<String> getExpiredTrace(int expiredDays, int limit) {
+        LocalDateTime now = LocalDateTime.now();
+        LocalDateTime expired = now.minusDays(expiredDays);
+        return flowTraceMapper.getExpiredTrace(expired, limit);
+    }
+
     private FlowTracePO serializer(FlowTrace flowTrace) {
         String contextPool = String.join(", ", flowTrace.getContextPool());
         return FlowTracePO.builder()

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowtrace/FlowTraceRepo.java

@@ -100,4 +100,13 @@ public interface FlowTraceRepo {
      * @param traceIds traceId列表
      */
     void deleteByIdList(List<String> traceIds);
+
+    /**
+     * 查询超期并且已完成的trace id
+     *
+     * @param expiredDays 超期天数
+     * @param limit 查询限制
+     * @return trace id列表
+     */
+    List<String> getExpiredTrace(int expiredDays, int limit);
 }

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/persist/mapper/FlowContextMapper.java

@@ -523,4 +523,11 @@ List<FlowContextPO> findFinishedContextsPagedByTraceId(String traceId, String en
      * @return trans id
      */
     String getTransIdByTrace(String traceId);
+
+    /**
+     * 根据trace列表删除对应的context
+     *
+     * @param traceIds trace id列表
+     */
+    void deleteByTraceIdList(List<String> traceIds);
 }

app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/persist/mapper/FlowTraceMapper.java

@@ -113,4 +113,13 @@ public interface FlowTraceMapper {
      * @return trace列表
      */
     List<String> findRunningTrace(List<String> applications);
+
+    /**
+     * 查询超期并且已完成的trace id
+     *
+     * @param expiredDays 超期天数
+     * @param limit 查询数量
+     * @return trace id列表
+     */
+    List<String> getExpiredTrace(LocalDateTime expiredDays, int limit);
 }

app-builder/waterflow/java/waterflow-service/src/main/resources/application.yml

@@ -16,5 +16,6 @@ jane:
       scheduleRate: 10000
       maxCount: 0
     isNeedFlowCallbackAdapt: false
+    contextExpiredDays: 7
 
 distributed-lock-provider: databaseDistributedLockProvider

app-builder/waterflow/java/waterflow-service/src/main/resources/mapper/FlowContextMapper.xml

@@ -765,4 +765,14 @@
         trace_id = #{traceId}
         LIMIT 1
     </select>
+
+    <delete id="deleteByTraceIdList">
+        DELETE FROM
+        <include refid="table"/>
+        WHERE
+        trace_id IN
+        <foreach item="id" collection="traceIds" open="(" separator="," close=")">
+            #{id}
+        </foreach>
+    </delete>
 </mapper>

app-builder/waterflow/java/waterflow-service/src/main/resources/mapper/FlowTraceMapper.xml

@@ -163,4 +163,15 @@
             #{traceId}
         </foreach>
     </delete>
+
+    <select id="getExpiredTrace" resultType="string">
+        SELECT
+        trace_id
+        FROM
+        <include refid="table"/>
+        WHERE
+        status != 'RUNNING'
+        AND end_time &lt; #{expiredDays}
+        LIMIT #{limit};
+    </select>
 </mapper>