feat: reduce conflicts when update configmap in k8s #89

CHANGES.md

@@ -14,6 +14,7 @@ Apollo Java 2.4.0
 * [Fix monitor arg cause npe](https://github.com/apolloconfig/apollo-java/pull/86)
 * [Fix the concurrent issue in SpringValueRegistry.scanAndClean](https://github.com/apolloconfig/apollo-java/pull/95)
 * [Feature support incremental configuration synchronization client](https://github.com/apolloconfig/apollo-java/pull/90)
+* [Feature reduce conflicts when update configmap in k8](https://github.com/apolloconfig/apollo-java/pull/93)
 
 ------------------
 All issues and pull requests are [here](https://github.com/apolloconfig/apollo-java/milestone/4?closed=1)

apollo-client/src/main/java/com/ctrip/framework/apollo/kubernetes/KubernetesManager.java

@@ -16,43 +16,67 @@
  */
 package com.ctrip.framework.apollo.kubernetes;
 
+import com.ctrip.framework.apollo.build.ApolloInjector;
 import com.ctrip.framework.apollo.core.utils.StringUtils;
 import com.ctrip.framework.apollo.exceptions.ApolloConfigException;
+import com.ctrip.framework.apollo.util.ConfigUtil;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
 import io.kubernetes.client.openapi.ApiClient;
 import io.kubernetes.client.openapi.ApiException;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
 import io.kubernetes.client.openapi.models.V1ConfigMap;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
 import io.kubernetes.client.util.Config;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
+/**
+ * Manages Kubernetes ConfigMap operations.
+ * Required Kubernetes permissions:
+ * - pods: [get, list] - For pod selection and write eligibility
+ * - configmaps: [get, create, update] - For ConfigMap operations
+ */
 @Service
 public class KubernetesManager {
     private static final Logger logger = LoggerFactory.getLogger(KubernetesManager.class);
 
+    private static final String RUNNING_POD_FIELD_SELECTOR = "status.phase=Running";
+
+    private static final int MAX_SEARCH_NUM = 1000;
+
     private ApiClient client;
     private CoreV1Api coreV1Api;
+    private int propertyKubernetesMaxWritePods;
+    private String localPodName = System.getenv("HOSTNAME");
 
     public KubernetesManager() {
         try {
             client = Config.defaultClient();
             coreV1Api = new CoreV1Api(client);
+            ConfigUtil configUtil = ApolloInjector.getInstance(ConfigUtil.class);
+            propertyKubernetesMaxWritePods = configUtil.getPropertyKubernetesMaxWritePods();
         } catch (Exception e) {
             String errorMessage = "Failed to initialize Kubernetes client: " + e.getMessage();
             logger.error(errorMessage, e);
             throw new RuntimeException(errorMessage, e);
         }
     }
 
-    public KubernetesManager(CoreV1Api coreV1Api) {
+    @VisibleForTesting
+    public KubernetesManager(CoreV1Api coreV1Api, String localPodName, int propertyKubernetesMaxWritePods) {
         this.coreV1Api = coreV1Api;
+        this.localPodName = localPodName;
+        this.propertyKubernetesMaxWritePods = propertyKubernetesMaxWritePods;
     }
 
     private V1ConfigMap buildConfigMap(String name, String namespace, Map<String, String> data) {
@@ -132,6 +156,10 @@ public boolean updateConfigMap(String k8sNamespace, String name, Map<String, Str
             return false;
         }
 
+        if (!isWritePod(k8sNamespace)) {
+            return true;
+        }
+
         int maxRetries = 5;
         int retryCount = 0;
         long waitTime = 100;
@@ -205,4 +233,43 @@ public boolean checkConfigMapExist(String k8sNamespace, String configMapName) {
             return false;
         }
     }
+
+    /**
+     * check pod whether pod can write configmap
+     *
+     * @param k8sNamespace config map namespace
+     * @return true if this pod can write configmap, false otherwise
+     */
+    private boolean isWritePod(String k8sNamespace) {
+        try {
+            if (Strings.isNullOrEmpty(localPodName)) {
+                return true;
+            }
+            V1Pod localPod = coreV1Api.readNamespacedPod(localPodName, k8sNamespace, null);
+            V1ObjectMeta localMetadata = localPod.getMetadata();
+            if (localMetadata == null || localMetadata.getLabels() == null) {
+                return true;
+            }
+            String appName = localMetadata.getLabels().get("app");
+            String labelSelector = "app=" + appName;
+
+            V1PodList v1PodList = coreV1Api.listNamespacedPod(k8sNamespace, null, null,
+                    null, RUNNING_POD_FIELD_SELECTOR, labelSelector,
+                    MAX_SEARCH_NUM, null, null
+                    , null, null);
+
+            return v1PodList.getItems().stream()
+                    .map(V1Pod::getMetadata)
+                    .filter(Objects::nonNull)
+                    //Make each node selects the same write nodes by sorting
+                    .filter(metadata -> metadata.getCreationTimestamp() != null)
+                    .sorted(Comparator.comparing(V1ObjectMeta::getCreationTimestamp))
+                    .map(V1ObjectMeta::getName)
+                    .limit(propertyKubernetesMaxWritePods)
+                    .anyMatch(localPodName::equals);
+        } catch (Exception e) {
+            logger.info("Error determining write pod eligibility:{}", e.getMessage(), e);
+            return true;
+        }
+    }
 }

apollo-client/src/main/java/com/ctrip/framework/apollo/util/ConfigUtil.java

@@ -72,12 +72,14 @@ public class ConfigUtil {
   private boolean propertyFileCacheEnabled = true;
   private boolean overrideSystemProperties = true;
   private boolean propertyKubernetesCacheEnabled = false;
+  private int propertyKubernetesMaxWritePods = 3;
   private boolean clientMonitorEnabled = false;
   private boolean clientMonitorJmxEnabled = false;
   private String monitorExternalType = "";
   private long monitorExternalExportPeriod = 10;
   private int monitorExceptionQueueSize = 25;
 
+
   public ConfigUtil() {
     warnLogRateLimiter = RateLimiter.create(0.017); // 1 warning log output per minute
     initRefreshInterval();
@@ -93,6 +95,7 @@ public ConfigUtil() {
     initPropertyFileCacheEnabled();
     initOverrideSystemProperties();
     initPropertyKubernetesCacheEnabled();
+    initPropertyKubernetesMaxWritePods();
     initClientMonitorEnabled();
     initClientMonitorJmxEnabled();
     initClientMonitorExternalType();
@@ -390,31 +393,44 @@ private String getDeprecatedCustomizedCacheRoot() {
   }
 
   public String getK8sNamespace() {
-    String k8sNamespace = getCacheKubernetesNamespace();
+    return getK8sConfigProperties(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE,
+            ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE_ENVIRONMENT_VARIABLES,
+            ConfigConsts.KUBERNETES_CACHE_CONFIG_MAP_NAMESPACE_DEFAULT);
+  }
 
-    if (!Strings.isNullOrEmpty(k8sNamespace)) {
-      return k8sNamespace;
+  private void initPropertyKubernetesMaxWritePods() {
+    String propertyKubernetesMaxWritePodsStr = getK8sConfigProperties(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_MAX_WRITE_PODS,
+            ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_MAX_WRITE_PODS_ENVIRONMENT_VARIABLES,
+            String.valueOf(propertyKubernetesMaxWritePods));
+    if (!Strings.isNullOrEmpty(propertyKubernetesMaxWritePodsStr)) {
+      try {
+        propertyKubernetesMaxWritePods = Integer.parseInt(propertyKubernetesMaxWritePodsStr);
+      } catch (Throwable ex) {
+        logger.error("Config for {} is invalid: {}",
+                ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE, propertyKubernetesMaxWritePodsStr);
+      }
     }
-
-    return ConfigConsts.KUBERNETES_CACHE_CONFIG_MAP_NAMESPACE_DEFAULT;
   }
 
-  private String getCacheKubernetesNamespace() {
+  private String getK8sConfigProperties(String key, String environmentKey, String defaultValue) {
     // 1. Get from System Property
-    String k8sNamespace = System.getProperty(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE);
+    String k8sNamespace = System.getProperty(key);
     if (Strings.isNullOrEmpty(k8sNamespace)) {
       // 2. Get from OS environment variable
-      k8sNamespace = System.getenv(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE_ENVIRONMENT_VARIABLES);
+      k8sNamespace = System.getenv(environmentKey);
     }
     if (Strings.isNullOrEmpty(k8sNamespace)) {
       // 3. Get from server.properties
-      k8sNamespace = Foundation.server().getProperty(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE, null);
+      k8sNamespace = Foundation.server().getProperty(key, null);
     }
     if (Strings.isNullOrEmpty(k8sNamespace)) {
       // 4. Get from app.properties
-      k8sNamespace = Foundation.app().getProperty(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE, null);
+      k8sNamespace = Foundation.app().getProperty(key, null);
+    }
+    if (!Strings.isNullOrEmpty(k8sNamespace)) {
+      return k8sNamespace;
     }
-    return k8sNamespace;
+    return defaultValue;
   }
 
   public boolean isInLocalMode() {
@@ -524,6 +540,10 @@ public boolean isPropertyKubernetesCacheEnabled() {
     return propertyKubernetesCacheEnabled;
   }
 
+  public int getPropertyKubernetesMaxWritePods() {
+    return propertyKubernetesMaxWritePods;
+  }
+
   public boolean isOverrideSystemProperties() {
     return overrideSystemProperties;
   }

apollo-client/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -126,6 +126,13 @@
       "description": "kubernetes configmap namespace.",
       "defaultValue": "default"
     },
+    {
+      "name": "apollo.cache.kubernetes.max-write-pods",
+      "type": "java.lang.String",
+      "sourceType": "com.ctrip.framework.apollo.util.ConfigUtil",
+      "description": "max number of pods that can write the configmap cache in Kubernetes.",
+      "defaultValue": "3"
+    },
     {
       "name": "apollo.property.order.enable",
       "type": "java.lang.Boolean",

apollo-client/src/test/java/com/ctrip/framework/apollo/kubernetes/KubernetesManagerTest.java

@@ -21,14 +21,30 @@
 import io.kubernetes.client.openapi.apis.CoreV1Api;
 import io.kubernetes.client.openapi.models.V1ConfigMap;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1Pod;
+import io.kubernetes.client.openapi.models.V1PodList;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.Mockito;
 
+import java.time.OffsetDateTime;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.mockito.Mockito.*;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.isNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class KubernetesManagerTest {
 
@@ -38,7 +54,7 @@ public class KubernetesManagerTest {
     @Before
     public void setUp() {
         coreV1Api = mock(CoreV1Api.class);
-        kubernetesManager = new KubernetesManager(coreV1Api);
+        kubernetesManager = new KubernetesManager(coreV1Api, "localPodName", 3);
 
         MockInjector.setInstance(KubernetesManager.class, kubernetesManager);
         MockInjector.setInstance(CoreV1Api.class, coreV1Api);
@@ -58,13 +74,13 @@ public void testCreateConfigMapSuccess() throws Exception {
                 .metadata(new V1ObjectMeta().name(name).namespace(namespace))
                 .data(data);
 
-        when(coreV1Api.createNamespacedConfigMap(eq(namespace), eq(configMap), isNull(), isNull(), isNull(),isNull())).thenReturn(configMap);
+        when(coreV1Api.createNamespacedConfigMap(eq(namespace), eq(configMap), isNull(), isNull(), isNull(), isNull())).thenReturn(configMap);
 
         // act
         String result = kubernetesManager.createConfigMap(namespace, name, data);
 
         // assert
-        verify(coreV1Api, times(1)).createNamespacedConfigMap(eq(namespace), any(V1ConfigMap.class), isNull(), isNull(), isNull(),isNull());
+        verify(coreV1Api, times(1)).createNamespacedConfigMap(eq(namespace), any(V1ConfigMap.class), isNull(), isNull(), isNull(), isNull());
         assert name.equals(result);
     }
 
@@ -82,7 +98,7 @@ public void testCreateConfigMapNullData() throws Exception {
         String result = kubernetesManager.createConfigMap(namespace, name, data);
 
         // assert
-        verify(coreV1Api, times(1)).createNamespacedConfigMap(eq(namespace), any(V1ConfigMap.class), isNull(), isNull(), isNull(),isNull());
+        verify(coreV1Api, times(1)).createNamespacedConfigMap(eq(namespace), any(V1ConfigMap.class), isNull(), isNull(), isNull(), isNull());
         assert name.equals(result);
     }
 
@@ -135,20 +151,40 @@ public void testUpdateConfigMapSuccess() throws Exception {
         // arrange
         String namespace = "default";
         String name = "testConfigMap";
-        Map<String, String> data = new HashMap<>();
-        data.put("key", "value");
+
+        V1Pod pod = new V1Pod()
+                .metadata(
+                        new V1ObjectMeta()
+                                .name("localPodName")
+                                .creationTimestamp(OffsetDateTime.now())
+                                .labels(Collections.singletonMap("app", "app")));
+        V1PodList v1PodList = new V1PodList().addItemsItem(new V1Pod().metadata(pod.getMetadata()));
+
+        Map<String, String> existData = new HashMap<>();
+        existData.put("key", "value");
         V1ConfigMap configMap = new V1ConfigMap();
         configMap.metadata(new V1ObjectMeta().name(name).namespace(namespace));
-        configMap.data(data);
+        configMap.data(existData);
 
+        when(coreV1Api.readNamespacedPod("localPodName", namespace, null)).thenReturn(pod);
+        when(coreV1Api.listNamespacedPod(namespace, null, null,
+                null, null, "app=app",
+                null, null, null
+                , null, null)).thenReturn(v1PodList);
         when(coreV1Api.readNamespacedConfigMap(name, namespace, null)).thenReturn(configMap);
         when(coreV1Api.replaceNamespacedConfigMap(name, namespace, configMap, null, null, null, null)).thenReturn(configMap);
 
         // act
-        Boolean success = kubernetesManager.updateConfigMap(namespace, name, data);
+        HashMap<String, String> updateData = new HashMap<>(existData);
+        updateData.put("newKey","newValue");
+        boolean success = kubernetesManager.updateConfigMap(namespace, name, updateData);
 
         // assert
         assertTrue(success);
+        Mockito.verify(coreV1Api, Mockito.times(1)).listNamespacedPod(namespace, null, null,
+                null, null, "app=app",
+                null, null, null
+                , null, null);
     }
 
     /**

apollo-core/src/main/java/com/ctrip/framework/apollo/core/ApolloClientSystemConsts.java

@@ -83,6 +83,16 @@ public class ApolloClientSystemConsts {
    */
   public static final String APOLLO_CACHE_KUBERNETES_NAMESPACE_ENVIRONMENT_VARIABLES = "APOLLO_CACHE_KUBERNETES_NAMESPACE";
 
+  /**
+   * max number of pods that can write the configmap cache in Kubernetes
+   */
+  public static final String APOLLO_CACHE_KUBERNETES_MAX_WRITE_PODS = "apollo.cache.kubernetes.max-write-pods";
+
+  /**
+   * max number of pods that can write the configmap cache in Kubernetes environment variables
+   */
+  public static final String APOLLO_CACHE_KUBERNETES_MAX_WRITE_PODS_ENVIRONMENT_VARIABLES = "APOLLO_CACHE_KUBERNETES_MAX_WRITE_PODS";
+
   /**
    * apollo client access key
    */