feat: support incremental configuration synchronization client

CHANGES.md

@@ -18,5 +18,6 @@ Apollo 2.5.0
 * [Bugfix: Correct permission target format to appId+env+namespace/cluster](https://github.com/apolloconfig/apollo/pull/5407)
 * [Security: Hide password when registering or modifying users](https://github.com/apolloconfig/apollo/pull/5414)
 * [Fix: the logical judgment for configuration addition, deletion, and modification.](https://github.com/apolloconfig/apollo/pull/5432)
+* [Feature support incremental configuration synchronization client](https://github.com/apolloconfig/apollo/pull/5288)
 ------------------
 All issues and pull requests are [here](https://github.com/apolloconfig/apollo/milestone/16?closed=1)

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

@@ -267,6 +267,10 @@ public long getInstanceConfigAuditTimeThresholdInMilli() {
     return TimeUnit.MINUTES.toMillis(timeThreshold);
   }
 
+  public boolean isConfigServiceIncrementalChangeEnabled() {
+    return getBooleanProperty("config-service.incremental.change.enabled", false);
+  }
+
   int checkInt(int value, int min, int max, int defaultValue) {
     if (value >= min && value <= max) {
       return value;

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

@@ -37,13 +37,15 @@ Release findFirstByAppIdAndClusterNameAndNamespaceNameAndIsAbandonedFalseOrderBy
 
   Release findByIdAndIsAbandonedFalse(long id);
 
+  Release findByReleaseKey(String releaseKey);
+
   List<Release> findByAppIdAndClusterNameAndNamespaceNameOrderByIdDesc(String appId, String clusterName, String namespaceName, Pageable page);
 
   List<Release> findByAppIdAndClusterNameAndNamespaceNameAndIsAbandonedFalseOrderByIdDesc(String appId, String clusterName, String namespaceName, Pageable page);
 
   List<Release> findByAppIdAndClusterNameAndNamespaceNameAndIsAbandonedFalseAndIdBetweenOrderByIdDesc(String appId, String clusterName, String namespaceName, long fromId, long toId);
 
-  List<Release> findByReleaseKeyIn(Set<String> releaseKey);
+  List<Release> findByReleaseKeyIn(Set<String> releaseKeys);
 
   List<Release> findByIdIn(Set<Long> releaseIds);
 

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

@@ -119,6 +119,10 @@ public List<Release> findByReleaseKeys(Set<String> releaseKeys) {
     return releaseRepository.findByReleaseKeyIn(releaseKeys);
   }
 
+  public Release findByReleaseKey(String releaseKey) {
+    return releaseRepository.findByReleaseKey(releaseKey);
+  }
+
   public Release findLatestActiveRelease(Namespace namespace) {
     return findLatestActiveRelease(namespace.getAppId(),
                                    namespace.getClusterName(), namespace.getNamespaceName());

apollo-configservice/src/main/java/com/ctrip/framework/apollo/configservice/ConfigServiceAutoConfiguration.java

@@ -31,6 +31,8 @@
 import com.ctrip.framework.apollo.configservice.service.config.ConfigService;
 import com.ctrip.framework.apollo.configservice.service.config.ConfigServiceWithCache;
 import com.ctrip.framework.apollo.configservice.service.config.DefaultConfigService;
+import com.ctrip.framework.apollo.configservice.service.config.DefaultIncrementalSyncService;
+import com.ctrip.framework.apollo.configservice.service.config.IncrementalSyncService;
 import com.ctrip.framework.apollo.configservice.util.AccessKeyUtil;
 import io.micrometer.core.instrument.MeterRegistry;
 import org.springframework.boot.web.servlet.FilterRegistrationBean;
@@ -69,13 +71,18 @@ public GrayReleaseRulesHolder grayReleaseRulesHolder() {
 
   @Bean
   public ConfigService configService() {
+    // enable local cache
     if (bizConfig.isConfigServiceCacheEnabled()) {
       return new ConfigServiceWithCache(releaseService, releaseMessageService,
           grayReleaseRulesHolder(), bizConfig, meterRegistry);
     }
     return new DefaultConfigService(releaseService, grayReleaseRulesHolder());
   }
 
+  @Bean
+  public IncrementalSyncService incrementalSyncService() {
+    return new DefaultIncrementalSyncService();
+  }
   @Bean
   public static NoOpPasswordEncoder passwordEncoder() {
     return (NoOpPasswordEncoder) NoOpPasswordEncoder.getInstance();

apollo-configservice/src/main/java/com/ctrip/framework/apollo/configservice/controller/ConfigController.java

@@ -16,22 +16,29 @@
  */
 package com.ctrip.framework.apollo.configservice.controller;
 
+import com.ctrip.framework.apollo.biz.config.BizConfig;
 import com.ctrip.framework.apollo.biz.entity.Release;
 import com.ctrip.framework.apollo.common.entity.AppNamespace;
 import com.ctrip.framework.apollo.common.utils.WebUtils;
 import com.ctrip.framework.apollo.configservice.service.AppNamespaceServiceWithCache;
 import com.ctrip.framework.apollo.configservice.service.config.ConfigService;
+import com.ctrip.framework.apollo.configservice.service.config.IncrementalSyncService;
 import com.ctrip.framework.apollo.configservice.util.InstanceConfigAuditUtil;
 import com.ctrip.framework.apollo.configservice.util.NamespaceUtil;
 import com.ctrip.framework.apollo.core.ConfigConsts;
 import com.ctrip.framework.apollo.core.dto.ApolloConfig;
 import com.ctrip.framework.apollo.core.dto.ApolloNotificationMessages;
+import com.ctrip.framework.apollo.core.dto.ConfigurationChange;
+import com.ctrip.framework.apollo.core.enums.ConfigSyncType;
 import com.ctrip.framework.apollo.tracer.Tracer;
 import com.google.common.base.Strings;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import com.google.gson.Gson;
 import com.google.gson.reflect.TypeToken;
+import java.util.regex.Pattern;
+import org.springframework.util.CollectionUtils;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -42,6 +49,9 @@
 import javax.servlet.http.HttpServletResponse;
 import java.io.IOException;
 import java.lang.reflect.Type;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -55,25 +65,33 @@
 public class ConfigController {
 
   private final ConfigService configService;
+
+  private final IncrementalSyncService incrementalSyncService;
   private final AppNamespaceServiceWithCache appNamespaceService;
   private final NamespaceUtil namespaceUtil;
   private final InstanceConfigAuditUtil instanceConfigAuditUtil;
   private final Gson gson;
+  private final BizConfig bizConfig;
+
 
   private static final Type configurationTypeReference = new TypeToken<Map<String, String>>() {
-      }.getType();
+  }.getType();
 
   public ConfigController(
       final ConfigService configService,
+      final IncrementalSyncService incrementalSyncService,
       final AppNamespaceServiceWithCache appNamespaceService,
       final NamespaceUtil namespaceUtil,
       final InstanceConfigAuditUtil instanceConfigAuditUtil,
-      final Gson gson) {
+      final Gson gson,
+      final BizConfig bizConfig) {
     this.configService = configService;
+    this.incrementalSyncService = incrementalSyncService;
     this.appNamespaceService = appNamespaceService;
     this.namespaceUtil = namespaceUtil;
     this.instanceConfigAuditUtil = instanceConfigAuditUtil;
     this.gson = gson;
+    this.bizConfig = bizConfig;
   }
 
   @GetMapping(value = "/{appId}/{clusterName}/{namespace:.+}")
@@ -132,10 +150,10 @@ public ApolloConfig queryConfig(@PathVariable String appId, @PathVariable String
 
     auditReleases(appId, clusterName, dataCenter, clientIp, releases);
 
-    String mergedReleaseKey = releases.stream().map(Release::getReleaseKey)
-            .collect(Collectors.joining(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR));
+    String latestMergedReleaseKey = releases.stream().map(Release::getReleaseKey)
+        .collect(Collectors.joining(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR));
 
-    if (mergedReleaseKey.equals(clientSideReleaseKey)) {
+    if (latestMergedReleaseKey.equals(clientSideReleaseKey)) {
       // Client side configuration is the same with server side, return 304
       response.setStatus(HttpServletResponse.SC_NOT_MODIFIED);
       Tracer.logEvent("Apollo.Config.NotModified",
@@ -144,8 +162,55 @@ public ApolloConfig queryConfig(@PathVariable String appId, @PathVariable String
     }
 
     ApolloConfig apolloConfig = new ApolloConfig(appId, appClusterNameLoaded, originalNamespace,
-        mergedReleaseKey);
-    apolloConfig.setConfigurations(mergeReleaseConfigurations(releases));
+        latestMergedReleaseKey);
+
+    Map<String, String> latestConfigurations = mergeReleaseConfigurations(releases);
+
+    try {
+      if (bizConfig.isConfigServiceIncrementalChangeEnabled()) {
+        LinkedHashSet<String> clientSideReleaseKeys = Sets.newLinkedHashSet(
+            Arrays.stream(
+                    clientSideReleaseKey.split(Pattern.quote(ConfigConsts.CLUSTER_NAMESPACE_SEPARATOR)))
+                .collect(Collectors.toList()));
+
+        Map<String, Release> clientSideReleases = configService.findReleasesByReleaseKeys(
+            clientSideReleaseKeys);
+        //find history releases
+        if (!CollectionUtils.isEmpty(clientSideReleases)) {
+          //order by clientSideReleaseKeys
+          List<Release> historyReleasesWithOrder = new ArrayList<>();
+          for (String item : clientSideReleaseKeys) {
+            Release release = clientSideReleases.get(item);
+            if (release != null) {
+              historyReleasesWithOrder.add(release);
+            }
+          }
+
+          Map<String, String> clientSideConfigurations = mergeReleaseConfigurations
+              (historyReleasesWithOrder);
+
+          if (!CollectionUtils.isEmpty(clientSideConfigurations)) {
+            List<ConfigurationChange> configurationChanges = incrementalSyncService.getConfigurationChanges(
+                latestMergedReleaseKey,
+                latestConfigurations, clientSideReleaseKey, clientSideConfigurations);
+
+            apolloConfig.setConfigurationChanges(configurationChanges);
+
+            apolloConfig.setConfigSyncType(ConfigSyncType.INCREMENTAL_SYNC.getValue());
+            Tracer.logEvent("Apollo.Config.Found", assembleKey(appId, appClusterNameLoaded,
+                originalNamespace, dataCenter));
+            return apolloConfig;
+          }
+
+        }
+
+      }
+    } catch (Exception e) {
+      //fallback to full sync
+      Tracer.logError("Failed to do incremental sync, fallback to full sync", e);
+    }
+
+    apolloConfig.setConfigurations(latestConfigurations);
 
     Tracer.logEvent("Apollo.Config.Found", assembleKey(appId, appClusterNameLoaded,
         originalNamespace, dataCenter));

apollo-configservice/src/main/java/com/ctrip/framework/apollo/configservice/service/config/ConfigService.java

@@ -19,6 +19,8 @@
 import com.ctrip.framework.apollo.biz.entity.Release;
 import com.ctrip.framework.apollo.biz.message.ReleaseMessageListener;
 import com.ctrip.framework.apollo.core.dto.ApolloNotificationMessages;
+import java.util.Map;
+import java.util.Set;
 
 /**
  * @author Jason Song([email protected])
@@ -40,4 +42,11 @@ public interface ConfigService extends ReleaseMessageListener {
    */
   Release loadConfig(String clientAppId, String clientIp, String clientLabel, String configAppId, String
       configClusterName, String configNamespace, String dataCenter, ApolloNotificationMessages clientMessages);
+
+
+  /**
+   * @param releaseKeys
+   * @return the ReleaseMap
+   */
+  Map<String, Release> findReleasesByReleaseKeys(Set<String> releaseKeys);
 }

apollo-configservice/src/main/java/com/ctrip/framework/apollo/configservice/service/config/ConfigServiceWithCache.java

@@ -22,6 +22,7 @@
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 
 import com.ctrip.framework.apollo.biz.entity.Release;
@@ -37,8 +38,12 @@
 
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.binder.cache.GuavaCacheMetrics;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 
+import java.util.Set;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -61,6 +66,8 @@ public class ConfigServiceWithCache extends AbstractConfigService {
   private static final String TRACER_EVENT_CACHE_LOAD_ID = "ConfigCache.LoadFromDBById";
   private static final String TRACER_EVENT_CACHE_GET = "ConfigCache.Get";
   private static final String TRACER_EVENT_CACHE_GET_ID = "ConfigCache.GetById";
+  private static final String TRACER_EVENT_CACHE_LOAD_RELEASE_KEY = "ConfigCache.LoadFromDBByReleaseKey";
+
 
   private final ReleaseService releaseService;
   private final ReleaseMessageService releaseMessageService;
@@ -71,6 +78,8 @@ public class ConfigServiceWithCache extends AbstractConfigService {
 
   private LoadingCache<Long, Optional<Release>> configIdCache;
 
+  private LoadingCache<String, Optional<Long>> releaseKeyCache;
+
   private ConfigCacheEntry nullConfigCacheEntry;
 
   public ConfigServiceWithCache(final ReleaseService releaseService,
@@ -90,6 +99,7 @@ public ConfigServiceWithCache(final ReleaseService releaseService,
   void initialize() {
     buildConfigCache();
     buildConfigIdCache();
+    buildReleaseKeyCache();
   }
 
   @Override
@@ -149,6 +159,42 @@ public void handleMessage(ReleaseMessage message, String channel) {
     }
   }
 
+  @Override
+  public Map<String, Release> findReleasesByReleaseKeys(Set<String> releaseKeys)
+      {
+        try {
+          ImmutableMap<String, Optional<Long>> releaseKeyMap = releaseKeyCache.getAll(releaseKeys);
+          if (CollectionUtils.isEmpty(releaseKeyMap)) {
+            return Collections.emptyMap();
+          }
+
+          Map<String, Long> validReleaseKeyIdMap = new HashMap<>();
+          for (Map.Entry<String, Optional<Long>> entry : releaseKeyMap.entrySet()) {
+            entry.getValue().ifPresent(id -> validReleaseKeyIdMap.put(entry.getKey(), id));
+          }
+          if (validReleaseKeyIdMap.isEmpty()) {
+            return Collections.emptyMap();
+          }
+
+          Map<Long, Optional<Release>> releasesMap = configIdCache.getAll(validReleaseKeyIdMap.values());
+          if (CollectionUtils.isEmpty(releasesMap)) {
+            return Collections.emptyMap();
+          }
+
+          Map<String, Release> releases = new HashMap<>();
+          for (Map.Entry<String, Long> entry : validReleaseKeyIdMap.entrySet()) {
+            Optional<Release> releaseOpt = releasesMap.get(entry.getValue());
+            releaseOpt.ifPresent(release -> releases.put(entry.getKey(), release));
+          }
+
+          return releases.isEmpty() ? Collections.emptyMap() : ImmutableMap.copyOf(releases);
+        } catch (Exception e) {
+          Tracer.logError(e);
+          logger.error("Failed to invoke findReleasesByReleaseKeys {}", releaseKeys, e);
+        }
+        return null;
+      }
+
   private void buildConfigCache() {
     CacheBuilder configCacheBuilder = CacheBuilder.newBuilder()
         .expireAfterAccess(DEFAULT_EXPIRED_AFTER_ACCESS_IN_MINUTES, TimeUnit.MINUTES);
@@ -198,6 +244,39 @@ public ConfigCacheEntry load(String key) throws Exception {
 
   }
 
+  private void buildReleaseKeyCache() {
+    CacheBuilder releaseKeyCacheBuilder = CacheBuilder.newBuilder()
+        .expireAfterAccess(DEFAULT_EXPIRED_AFTER_ACCESS_IN_MINUTES, TimeUnit.MINUTES);
+    if (bizConfig.isConfigServiceCacheStatsEnabled()) {
+      releaseKeyCacheBuilder.recordStats();
+    }
+    releaseKeyCache = releaseKeyCacheBuilder.build(new CacheLoader<String, Optional<Long>>() {
+      @Override
+      public Optional<Long> load(String key) throws Exception {
+        Transaction transaction = Tracer.newTransaction(TRACER_EVENT_CACHE_LOAD_RELEASE_KEY,
+            String.valueOf(key));
+        try {
+          Release release = releaseService.findByReleaseKey(key);
+
+          transaction.setStatus(Transaction.SUCCESS);
+          if (release != null) {
+            return Optional.ofNullable(release.getId());
+          }
+          return Optional.empty();
+        } catch (Throwable ex) {
+          transaction.setStatus(ex);
+          throw ex;
+        } finally {
+          transaction.complete();
+        }
+      }
+    });
+
+    if (bizConfig.isConfigServiceCacheStatsEnabled()) {
+      GuavaCacheMetrics.monitor(meterRegistry, releaseKeyCache, "releaseKey_cache");
+    }
+
+  }
   private void buildConfigIdCache() {
     CacheBuilder configIdCacheBuilder = CacheBuilder.newBuilder()
         .expireAfterAccess(DEFAULT_EXPIRED_AFTER_ACCESS_IN_MINUTES, TimeUnit.MINUTES);

apollo-configservice/src/main/java/com/ctrip/framework/apollo/configservice/service/config/DefaultConfigService.java

@@ -21,6 +21,11 @@
 import com.ctrip.framework.apollo.biz.grayReleaseRule.GrayReleaseRulesHolder;
 import com.ctrip.framework.apollo.biz.service.ReleaseService;
 import com.ctrip.framework.apollo.core.dto.ApolloNotificationMessages;
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 /**
  * config service with no cache
@@ -55,4 +60,13 @@ protected Release findLatestActiveRelease(String configAppId, String configClust
   public void handleMessage(ReleaseMessage message, String channel) {
     // since there is no cache, so do nothing
   }
+
+  @Override
+  public Map<String, Release> findReleasesByReleaseKeys(Set<String> releaseKeys) {
+    List<Release> releasesMap = releaseService.findByReleaseKeys(releaseKeys);
+    if (releasesMap != null) {
+      return releasesMap.stream().collect(ImmutableMap.toImmutableMap(Release::getReleaseKey, release -> release));
+    }
+    return Collections.emptyMap();
+  }
 }