Add release history cleaning function

CHANGES.md

@@ -20,6 +20,7 @@ Apollo 2.2.0
 * [Misc dependency updates](https://github.com/apolloconfig/apollo/pull/4784)
 * [Fix the problem that the deletion failure of the system rights management page does not prompt](https://github.com/apolloconfig/apollo/pull/4803)
 * [Fix the issue of the system permission management page retrieving non-existent users](https://github.com/apolloconfig/apollo/pull/4802)
+* [Add release history cleaning function](https://github.com/apolloconfig/apollo/pull/4813)
 
 ------------------
 All issues and pull requests are [here](https://github.com/apolloconfig/apollo/milestone/13?closed=1)

apollo-biz/src/main/java/com/ctrip/framework/apollo/biz/config/BizConfig.java

@@ -46,12 +46,16 @@ public class BizConfig extends RefreshableConfig {
   private static final int DEFAULT_RELEASE_MESSAGE_NOTIFICATION_BATCH = 100;
   private static final int DEFAULT_RELEASE_MESSAGE_NOTIFICATION_BATCH_INTERVAL_IN_MILLI = 100;//100ms
   private static final int DEFAULT_LONG_POLLING_TIMEOUT = 60; //60s
+  public static final int DEFAULT_RELEASE_HISTORY_RETENTION_SIZE = -1;
 
   private static final Gson GSON = new Gson();
 
   private static final Type namespaceValueLengthOverrideTypeReference =
       new TypeToken<Map<Long, Integer>>() {
       }.getType();
+  private static final Type releaseHistoryRetentionSizeOverrideTypeReference =
+      new TypeToken<Map<String, Integer>>() {
+      }.getType();
 
   private final BizDBPropertySource propertySource;
 
@@ -154,6 +158,21 @@ public int accessKeyAuthTimeDiffTolerance() {
         DEFAULT_ACCESS_KEY_AUTH_TIME_DIFF_TOLERANCE);
   }
 
+  public int releaseHistoryRetentionSize() {
+    int count = getIntProperty("apollo.release-history.retention.size", DEFAULT_RELEASE_HISTORY_RETENTION_SIZE);
+    return checkInt(count, 1, Integer.MAX_VALUE, DEFAULT_RELEASE_HISTORY_RETENTION_SIZE);
+  }
+
+  public Map<String, Integer> releaseHistoryRetentionSizeOverride() {
+    String overrideString = getValue("apollo.release-history.retention.size.override");
+    Map<String, Integer> releaseHistoryRetentionSizeOverride = Maps.newHashMap();
+    if (!Strings.isNullOrEmpty(overrideString)) {
+      releaseHistoryRetentionSizeOverride =
+          GSON.fromJson(overrideString, releaseHistoryRetentionSizeOverrideTypeReference);
+    }
+    return releaseHistoryRetentionSizeOverride;
+  }
+
   public int releaseMessageCacheScanInterval() {
     int interval = getIntProperty("apollo.release-message-cache-scan.interval", DEFAULT_RELEASE_MESSAGE_CACHE_SCAN_INTERVAL);
     return checkInt(interval, 1, Integer.MAX_VALUE, DEFAULT_RELEASE_MESSAGE_CACHE_SCAN_INTERVAL);

apollo-biz/src/main/java/com/ctrip/framework/apollo/biz/repository/ReleaseHistoryRepository.java

@@ -18,6 +18,8 @@
 
 import com.ctrip.framework.apollo.biz.entity.ReleaseHistory;
 
+import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
@@ -42,4 +44,9 @@ Page<ReleaseHistory> findByAppIdAndClusterNameAndNamespaceNameOrderByIdDesc(Stri
   @Query("update ReleaseHistory set IsDeleted = true, DeletedAt = ROUND(UNIX_TIMESTAMP(NOW(4))*1000), DataChange_LastModifiedBy = ?4 where AppId=?1 and ClusterName=?2 and NamespaceName = ?3 and IsDeleted = false")
   int batchDelete(String appId, String clusterName, String namespaceName, String operator);
 
+  @Query(value = "select Id from ReleaseHistory where AppId=?1 and ClusterName=?2 and NamespaceName = ?3 and BranchName = ?4 and IsDeleted = false order by Id desc Limit 1 offset ?5", nativeQuery = true)
+  Optional<Long> findReleaseHistoryRetentionMaxId(String appId, String clusterName, String namespaceName, String branchName, int releaseHistoryLimit);
+
+  List<ReleaseHistory> findFirst100ByAppIdAndClusterNameAndNamespaceNameAndBranchNameAndIdLessThanEqualOrderByIdAsc(String appId, String clusterName, String namespaceName, String branchName, long maxId);
+
 }

apollo-biz/src/main/java/com/ctrip/framework/apollo/biz/service/ReleaseHistoryService.java

@@ -16,36 +16,91 @@
  */
 package com.ctrip.framework.apollo.biz.service;
 
+import static com.ctrip.framework.apollo.biz.config.BizConfig.DEFAULT_RELEASE_HISTORY_RETENTION_SIZE;
+
+import com.ctrip.framework.apollo.biz.config.BizConfig;
 import com.ctrip.framework.apollo.biz.entity.Audit;
 import com.ctrip.framework.apollo.biz.entity.ReleaseHistory;
 import com.ctrip.framework.apollo.biz.repository.ReleaseHistoryRepository;
+import com.ctrip.framework.apollo.biz.repository.ReleaseRepository;
+import com.ctrip.framework.apollo.core.utils.ApolloThreadFactory;
+import com.ctrip.framework.apollo.tracer.Tracer;
+import com.google.common.collect.Queues;
 import com.google.gson.Gson;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.stereotype.Service;
+import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.Date;
 import java.util.Map;
 import java.util.Set;
+import org.springframework.transaction.support.TransactionCallbackWithoutResult;
+import org.springframework.transaction.support.TransactionTemplate;
 
 /**
  * @author Jason Song(song_s@ctrip.com)
  */
 @Service
 public class ReleaseHistoryService {
+  private static final Logger logger = LoggerFactory.getLogger(ReleaseHistoryService.class);
   private static final Gson GSON = new Gson();
+  private static final int CLEAN_QUEUE_MAX_SIZE = 100;
+  private final BlockingQueue<ReleaseHistory> releaseClearQueue = Queues.newLinkedBlockingQueue(CLEAN_QUEUE_MAX_SIZE);
+  private final ExecutorService cleanExecutorService = Executors.newSingleThreadExecutor(
+      ApolloThreadFactory.create("ReleaseHistoryService", true));
+  private final AtomicBoolean cleanStopped = new AtomicBoolean(false);
 
   private final ReleaseHistoryRepository releaseHistoryRepository;
+  private final ReleaseRepository releaseRepository;
   private final AuditService auditService;
+  private final BizConfig bizConfig;
+  private final TransactionTemplate transactionManager;
 
   public ReleaseHistoryService(
       final ReleaseHistoryRepository releaseHistoryRepository,
-      final AuditService auditService) {
+      final ReleaseRepository releaseRepository,
+      final AuditService auditService,
+      final BizConfig bizConfig,
+      final TransactionTemplate transactionManager) {
     this.releaseHistoryRepository = releaseHistoryRepository;
+    this.releaseRepository = releaseRepository;
     this.auditService = auditService;
+    this.bizConfig = bizConfig;
+    this.transactionManager = transactionManager;
   }
 
+  @PostConstruct
+  private void initialize() {
+    cleanExecutorService.submit(() -> {
+      while (!cleanStopped.get() && !Thread.currentThread().isInterrupted()) {
+        try {
+          ReleaseHistory releaseHistory = releaseClearQueue.poll(1, TimeUnit.SECONDS);
+          if (releaseHistory != null) {
+            this.cleanReleaseHistory(releaseHistory);
+          } else {
+            TimeUnit.MINUTES.sleep(1);
+          }
+        } catch (Throwable ex) {
+          logger.error("Clean releaseHistory failed", ex);
+          Tracer.logError(ex);
+        }
+      }
+    });
+  }
 
   public Page<ReleaseHistory> findReleaseHistoriesByNamespace(String appId, String clusterName,
                                                               String namespaceName, Pageable
@@ -92,11 +147,67 @@ public ReleaseHistory createReleaseHistory(String appId, String clusterName, Str
     auditService.audit(ReleaseHistory.class.getSimpleName(), releaseHistory.getId(),
                        Audit.OP.INSERT, releaseHistory.getDataChangeCreatedBy());
 
+    int releaseHistoryRetentionLimit = this.getReleaseHistoryRetentionLimit(releaseHistory);
+    if (releaseHistoryRetentionLimit != DEFAULT_RELEASE_HISTORY_RETENTION_SIZE) {
+      if (!releaseClearQueue.offer(releaseHistory)) {
+        logger.warn("releaseClearQueue is full, failed to add task to clean queue, " +
+            "clean queue max size:{}", CLEAN_QUEUE_MAX_SIZE);
+      }
+    }
     return releaseHistory;
   }
 
   @Transactional
   public int batchDelete(String appId, String clusterName, String namespaceName, String operator) {
     return releaseHistoryRepository.batchDelete(appId, clusterName, namespaceName, operator);
   }
+
+  private void cleanReleaseHistory(ReleaseHistory cleanRelease) {
+    String appId = cleanRelease.getAppId();
+    String clusterName = cleanRelease.getClusterName();
+    String namespaceName = cleanRelease.getNamespaceName();
+    String branchName = cleanRelease.getBranchName();
+
+    int retentionLimit = this.getReleaseHistoryRetentionLimit(cleanRelease);
+    //Second check, if retentionLimit is default value, do not clean
+    if (retentionLimit == DEFAULT_RELEASE_HISTORY_RETENTION_SIZE) {
+      return;
+    }
+
+    Optional<Long> maxId = releaseHistoryRepository.findReleaseHistoryRetentionMaxId(appId, clusterName, namespaceName, branchName, retentionLimit);
+    if (!maxId.isPresent()) {
+      return;
+    }
+
+    boolean hasMore = true;
+    while (hasMore && !Thread.currentThread().isInterrupted()) {
+      List<ReleaseHistory> cleanReleaseHistoryList = releaseHistoryRepository.findFirst100ByAppIdAndClusterNameAndNamespaceNameAndBranchNameAndIdLessThanEqualOrderByIdAsc(
+          appId, clusterName, namespaceName, branchName, maxId.get());
+      Set<Long> releaseIds = cleanReleaseHistoryList.stream()
+          .map(ReleaseHistory::getReleaseId)
+          .collect(Collectors.toSet());
+
+      transactionManager.execute(new TransactionCallbackWithoutResult() {
+        @Override
+        protected void doInTransactionWithoutResult(TransactionStatus status) {
+          releaseHistoryRepository.deleteAll(cleanReleaseHistoryList);
+          releaseRepository.deleteAllById(releaseIds);
+        }
+      });
+      hasMore = cleanReleaseHistoryList.size() == 100;
+    }
+  }
+
+  private int getReleaseHistoryRetentionLimit(ReleaseHistory releaseHistory) {
+    String overrideKey = String.format("%s+%s+%s+%s", releaseHistory.getAppId(),
+        releaseHistory.getClusterName(), releaseHistory.getNamespaceName(), releaseHistory.getBranchName());
+
+    Map<String, Integer> overrideMap = bizConfig.releaseHistoryRetentionSizeOverride();
+    return overrideMap.getOrDefault(overrideKey, bizConfig.releaseHistoryRetentionSize());
+  }
+
+  @PreDestroy
+  void stopClean() {
+    cleanStopped.set(true);
+  }
 }

apollo-biz/src/test/java/com/ctrip/framework/apollo/biz/AbstractIntegrationTest.java

@@ -32,4 +32,9 @@
 )
 public abstract class AbstractIntegrationTest {
 
+  protected static final String APP_ID = "kl-app";
+  protected static final String CLUSTER_NAME = "default";
+  protected static final String NAMESPACE_NAME = "application";
+  protected static final String BRANCH_NAME = "default";
+
 }

apollo-biz/src/test/java/com/ctrip/framework/apollo/biz/config/BizConfigTest.java

@@ -69,6 +69,24 @@ public void testReleaseMessageNotificationBatchWithInvalidNumber() throws Except
     assertEquals(defaultBatch, bizConfig.releaseMessageNotificationBatch());
   }
 
+  @Test
+  public void testReleaseHistoryRetentionLimit() {
+    int someLimit = 20;
+    when(environment.getProperty("apollo.release-history.retention.size")).thenReturn(String.valueOf(someLimit));
+
+    assertEquals(someLimit, bizConfig.releaseHistoryRetentionSize());
+  }
+
+  @Test
+  public void testReleaseHistoryRetentionLimitOverride() {
+    int someOverrideLimit = 10;
+    String overrideValueString = "{'a+b+c+b':10}";
+    when(environment.getProperty("apollo.release-history.retention.size.override")).thenReturn(overrideValueString);
+    int  overrideValue = bizConfig.releaseHistoryRetentionSizeOverride().get("a+b+c+b");
+
+    assertEquals(someOverrideLimit, overrideValue);
+  }
+
   @Test
   public void testReleaseMessageNotificationBatchWithNAN() throws Exception {
     String someNAN = "someNAN";

apollo-biz/src/test/java/com/ctrip/framework/apollo/biz/repository/ReleaseHistoryRepositoryTest.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2023 Apollo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.
+ *
+ */
+package com.ctrip.framework.apollo.biz.repository;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import com.ctrip.framework.apollo.biz.AbstractIntegrationTest;
+import com.ctrip.framework.apollo.biz.entity.ReleaseHistory;
+import java.util.List;
+import java.util.Optional;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.jdbc.Sql;
+import org.springframework.test.context.jdbc.Sql.ExecutionPhase;
+
+/**
+ * @author kl (http://kailing.pub)
+ * @since 2023/3/23
+ */
+public class ReleaseHistoryRepositoryTest extends AbstractIntegrationTest {
+
+  @Autowired
+  private ReleaseHistoryRepository releaseHistoryRepository;
+
+  @Test
+  @Sql(scripts = "/sql/release-history-test.sql", executionPhase = ExecutionPhase.BEFORE_TEST_METHOD)
+  @Sql(scripts = "/sql/clean.sql", executionPhase = ExecutionPhase.AFTER_TEST_METHOD)
+  public void testFindReleaseHistoryRetentionMaxId() {
+    Optional<Long> maxId = releaseHistoryRepository.findReleaseHistoryRetentionMaxId(APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, 1);
+    assertEquals(5, maxId.orElse(0L).longValue());
+
+    maxId = releaseHistoryRepository.findReleaseHistoryRetentionMaxId(APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, 2);
+    assertEquals(4, maxId.orElse(0L).longValue());
+
+    maxId = releaseHistoryRepository.findReleaseHistoryRetentionMaxId(APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, 5);
+    assertEquals(1, maxId.orElse(0L).longValue());
+
+    releaseHistoryRepository.deleteAll();
+    maxId = releaseHistoryRepository.findReleaseHistoryRetentionMaxId(APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, 1);
+    assertFalse(maxId.isPresent());
+  }
+
+  @Test
+  @Sql(scripts = "/sql/release-history-test.sql", executionPhase = ExecutionPhase.BEFORE_TEST_METHOD)
+  @Sql(scripts = "/sql/clean.sql", executionPhase = ExecutionPhase.AFTER_TEST_METHOD)
+  public void testFindFirst100ByAppIdAndClusterNameAndNamespaceNameAndBranchNameAndIdLessThanEqualOrderByIdAsc() {
+
+    int releaseHistoryRetentionSize = 2;
+    Optional<Long> maxId = releaseHistoryRepository.findReleaseHistoryRetentionMaxId(APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, releaseHistoryRetentionSize);
+    List<ReleaseHistory> releaseHistories = releaseHistoryRepository.findFirst100ByAppIdAndClusterNameAndNamespaceNameAndBranchNameAndIdLessThanEqualOrderByIdAsc(
+        APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, maxId.orElse(0L));
+    assertEquals(4, releaseHistories.size());
+
+    releaseHistoryRetentionSize = 1;
+    maxId = releaseHistoryRepository.findReleaseHistoryRetentionMaxId(APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, releaseHistoryRetentionSize);
+    releaseHistories = releaseHistoryRepository.findFirst100ByAppIdAndClusterNameAndNamespaceNameAndBranchNameAndIdLessThanEqualOrderByIdAsc(
+        APP_ID, CLUSTER_NAME, NAMESPACE_NAME, BRANCH_NAME, maxId.orElse(0L));
+    assertEquals(5, releaseHistories.size());
+  }
+}