Skip to content

Commit 45d2cce

Browse files
24kpure24kpure
committed
feat: reduce conflicts when update configmap in k8s #89 
1 parent e64c35b commit 45d2cce

File tree

4 files changed

+128
-14
lines changed

4 files changed

+128
-14
lines changed

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,18 +16,24 @@
1616
*/
1717
package com.ctrip.framework.apollo.kubernetes;
1818

19+
import com.ctrip.framework.apollo.build.ApolloInjector;
1920
import com.ctrip.framework.apollo.core.utils.StringUtils;
2021
import com.ctrip.framework.apollo.exceptions.ApolloConfigException;
22+
import com.ctrip.framework.apollo.util.ConfigUtil;
2123
import io.kubernetes.client.openapi.ApiClient;
2224
import io.kubernetes.client.openapi.ApiException;
2325
import io.kubernetes.client.openapi.apis.CoreV1Api;
2426
import io.kubernetes.client.openapi.models.V1ConfigMap;
27+
import io.kubernetes.client.openapi.models.V1Node;
2528
import io.kubernetes.client.openapi.models.V1ObjectMeta;
29+
import io.kubernetes.client.openapi.models.V1Pod;
30+
import io.kubernetes.client.openapi.models.V1PodList;
2631
import io.kubernetes.client.util.Config;
2732
import org.slf4j.Logger;
2833
import org.slf4j.LoggerFactory;
2934
import org.springframework.stereotype.Service;
3035

36+
import java.util.Comparator;
3137
import java.util.HashMap;
3238
import java.util.Map;
3339
import java.util.Objects;
@@ -39,11 +45,14 @@ public class KubernetesManager {
3945

4046
private ApiClient client;
4147
private CoreV1Api coreV1Api;
48+
private int propertyKubernetesMaxWritePods;
4249

4350
public KubernetesManager() {
4451
try {
4552
client = Config.defaultClient();
4653
coreV1Api = new CoreV1Api(client);
54+
ConfigUtil configUtil = ApolloInjector.getInstance(ConfigUtil.class);
55+
propertyKubernetesMaxWritePods = configUtil.getPropertyKubernetesMaxWritePods();
4756
} catch (Exception e) {
4857
String errorMessage = "Failed to initialize Kubernetes client: " + e.getMessage();
4958
logger.error(errorMessage, e);
@@ -132,6 +141,10 @@ public boolean updateConfigMap(String k8sNamespace, String name, Map<String, Str
132141
return false;
133142
}
134143

144+
if (!isWritePod(k8sNamespace)) {
145+
return true;
146+
}
147+
135148
int maxRetries = 5;
136149
int retryCount = 0;
137150
long waitTime = 100;
@@ -205,4 +218,41 @@ public boolean checkConfigMapExist(String k8sNamespace, String configMapName) {
205218
return false;
206219
}
207220
}
221+
222+
/**
223+
* check pod whether pod can write configmap
224+
*
225+
* @param k8sNamespace config map namespace
226+
* @return true if this pod can write configmap, false otherwise
227+
*/
228+
private boolean isWritePod(String k8sNamespace) {
229+
try {
230+
String localPodName = System.getenv("HOSTNAME");
231+
V1Node localNode = coreV1Api.readNode(localPodName, null);
232+
V1ObjectMeta localMetadata = localNode.getMetadata();
233+
if (localMetadata == null || localMetadata.getLabels() == null) {
234+
return true;
235+
}
236+
String appName = localMetadata.getLabels().get("app");
237+
String labelSelector = "app=" + appName;
238+
239+
V1PodList v1PodList = coreV1Api.listNamespacedPod(k8sNamespace, null, null,
240+
null, null, labelSelector,
241+
null, null, null
242+
, null, null);
243+
244+
return v1PodList.getItems().stream()
245+
.map(V1Pod::getMetadata)
246+
.filter(Objects::nonNull)
247+
//Make each node selects the same write nodes by sorting
248+
.filter(metadata -> metadata.getCreationTimestamp() != null)
249+
.sorted(Comparator.comparing(V1ObjectMeta::getCreationTimestamp))
250+
.map(V1ObjectMeta::getName)
251+
.limit(propertyKubernetesMaxWritePods)
252+
.anyMatch(localPodName::equals);
253+
} catch (Exception e) {
254+
logger.info("Select write nodes error:{}", e.getMessage(), e);
255+
return true;
256+
}
257+
}
208258
}

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

Lines changed: 31 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -72,12 +72,14 @@ public class ConfigUtil {
7272
private boolean propertyFileCacheEnabled = true;
7373
private boolean overrideSystemProperties = true;
7474
private boolean propertyKubernetesCacheEnabled = false;
75+
private int propertyKubernetesMaxWritePods = 3;
7576
private boolean clientMonitorEnabled = false;
7677
private boolean clientMonitorJmxEnabled = false;
7778
private String monitorExternalType = "";
7879
private long monitorExternalExportPeriod = 10;
7980
private int monitorExceptionQueueSize = 25;
8081

82+
8183
public ConfigUtil() {
8284
warnLogRateLimiter = RateLimiter.create(0.017); // 1 warning log output per minute
8385
initRefreshInterval();
@@ -93,6 +95,7 @@ public ConfigUtil() {
9395
initPropertyFileCacheEnabled();
9496
initOverrideSystemProperties();
9597
initPropertyKubernetesCacheEnabled();
98+
initPropertyKubernetesMaxWritePods();
9699
initClientMonitorEnabled();
97100
initClientMonitorJmxEnabled();
98101
initClientMonitorExternalType();
@@ -390,31 +393,44 @@ private String getDeprecatedCustomizedCacheRoot() {
390393
}
391394

392395
public String getK8sNamespace() {
393-
String k8sNamespace = getCacheKubernetesNamespace();
396+
return getK8sConfigProperties(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE,
397+
ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE_ENVIRONMENT_VARIABLES,
398+
ConfigConsts.KUBERNETES_CACHE_CONFIG_MAP_NAMESPACE_DEFAULT);
399+
}
394400

395-
if (!Strings.isNullOrEmpty(k8sNamespace)) {
396-
return k8sNamespace;
401+
private void initPropertyKubernetesMaxWritePods() {
402+
String propertyKubernetesMaxWritePodsStr = getK8sConfigProperties(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_MAX_WRITE_PODS,
403+
ApolloClientSystemConsts.APOLLO_CACHE_MAX_WRITE_PODS_ENVIRONMENT_VARIABLES,
404+
String.valueOf(propertyKubernetesMaxWritePods));
405+
if (!Strings.isNullOrEmpty(propertyKubernetesMaxWritePodsStr)) {
406+
try {
407+
propertyKubernetesMaxWritePods = Integer.parseInt(propertyKubernetesMaxWritePodsStr);
408+
} catch (Throwable ex) {
409+
logger.error("Config for {} is invalid: {}",
410+
ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE, propertyKubernetesMaxWritePodsStr);
411+
}
397412
}
398-
399-
return ConfigConsts.KUBERNETES_CACHE_CONFIG_MAP_NAMESPACE_DEFAULT;
400413
}
401414

402-
private String getCacheKubernetesNamespace() {
415+
private String getK8sConfigProperties(String key, String environmentKey, String defaultValue) {
403416
// 1. Get from System Property
404-
String k8sNamespace = System.getProperty(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE);
417+
String k8sNamespace = System.getProperty(key);
405418
if (Strings.isNullOrEmpty(k8sNamespace)) {
406419
// 2. Get from OS environment variable
407-
k8sNamespace = System.getenv(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE_ENVIRONMENT_VARIABLES);
420+
k8sNamespace = System.getenv(environmentKey);
408421
}
409422
if (Strings.isNullOrEmpty(k8sNamespace)) {
410423
// 3. Get from server.properties
411-
k8sNamespace = Foundation.server().getProperty(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE, null);
424+
k8sNamespace = Foundation.server().getProperty(key, null);
412425
}
413426
if (Strings.isNullOrEmpty(k8sNamespace)) {
414427
// 4. Get from app.properties
415-
k8sNamespace = Foundation.app().getProperty(ApolloClientSystemConsts.APOLLO_CACHE_KUBERNETES_NAMESPACE, null);
428+
k8sNamespace = Foundation.app().getProperty(key, null);
429+
}
430+
if (!Strings.isNullOrEmpty(k8sNamespace)) {
431+
return k8sNamespace;
416432
}
417-
return k8sNamespace;
433+
return defaultValue;
418434
}
419435

420436
public boolean isInLocalMode() {
@@ -524,6 +540,10 @@ public boolean isPropertyKubernetesCacheEnabled() {
524540
return propertyKubernetesCacheEnabled;
525541
}
526542

543+
public int getPropertyKubernetesMaxWritePods() {
544+
return propertyKubernetesMaxWritePods;
545+
}
546+
527547
public boolean isOverrideSystemProperties() {
528548
return overrideSystemProperties;
529549
}

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

Lines changed: 37 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,32 @@
2020
import io.kubernetes.client.openapi.ApiException;
2121
import io.kubernetes.client.openapi.apis.CoreV1Api;
2222
import io.kubernetes.client.openapi.models.V1ConfigMap;
23+
import io.kubernetes.client.openapi.models.V1Node;
2324
import io.kubernetes.client.openapi.models.V1ObjectMeta;
25+
import io.kubernetes.client.openapi.models.V1Pod;
26+
import io.kubernetes.client.openapi.models.V1PodList;
2427
import org.junit.Before;
2528
import org.junit.Test;
29+
import org.mockito.Mockito;
2630

31+
import java.time.OffsetDateTime;
32+
import java.util.Collections;
2733
import java.util.HashMap;
2834
import java.util.Map;
2935

30-
import static org.mockito.Mockito.*;
31-
import static org.junit.Assert.*;
36+
import static org.junit.Assert.assertEquals;
37+
import static org.junit.Assert.assertFalse;
38+
import static org.junit.Assert.assertNull;
39+
import static org.junit.Assert.assertTrue;
40+
import static org.mockito.Mockito.any;
41+
import static org.mockito.Mockito.doReturn;
42+
import static org.mockito.Mockito.doThrow;
43+
import static org.mockito.Mockito.eq;
44+
import static org.mockito.Mockito.isNull;
45+
import static org.mockito.Mockito.mock;
46+
import static org.mockito.Mockito.times;
47+
import static org.mockito.Mockito.verify;
48+
import static org.mockito.Mockito.when;
3249

3350
public class KubernetesManagerTest {
3451

@@ -135,20 +152,37 @@ public void testUpdateConfigMapSuccess() throws Exception {
135152
// arrange
136153
String namespace = "default";
137154
String name = "testConfigMap";
155+
156+
V1Node node = new V1Node()
157+
.metadata(
158+
new V1ObjectMeta()
159+
.creationTimestamp(OffsetDateTime.now())
160+
.labels(Collections.singletonMap("app", "app")));
161+
V1PodList v1PodList = new V1PodList().addItemsItem(new V1Pod().metadata(node.getMetadata()));
162+
138163
Map<String, String> data = new HashMap<>();
139164
data.put("key", "value");
140165
V1ConfigMap configMap = new V1ConfigMap();
141166
configMap.metadata(new V1ObjectMeta().name(name).namespace(namespace));
142167
configMap.data(data);
143168

169+
when(coreV1Api.readNode(null, null)).thenReturn(node);
170+
when(coreV1Api.listNamespacedPod(namespace, null, null,
171+
null, null, "app=app",
172+
null, null, null
173+
, null, null)).thenReturn(v1PodList);
144174
when(coreV1Api.readNamespacedConfigMap(name, namespace, null)).thenReturn(configMap);
145175
when(coreV1Api.replaceNamespacedConfigMap(name, namespace, configMap, null, null, null, null)).thenReturn(configMap);
146176

147177
// act
148-
Boolean success = kubernetesManager.updateConfigMap(namespace, name, data);
178+
boolean success = kubernetesManager.updateConfigMap(namespace, name, data);
149179

150180
// assert
151181
assertTrue(success);
182+
Mockito.verify(coreV1Api, Mockito.times(1)).listNamespacedPod(namespace, null, null,
183+
null, null, "app=app",
184+
null, null, null
185+
, null, null);
152186
}
153187

154188
/**

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,16 @@ public class ApolloClientSystemConsts {
8383
*/
8484
public static final String APOLLO_CACHE_KUBERNETES_NAMESPACE_ENVIRONMENT_VARIABLES = "APOLLO_CACHE_KUBERNETES_NAMESPACE";
8585

86+
/**
87+
* max number of pods that can write the configmap cache in Kubernetes
88+
*/
89+
public static final String APOLLO_CACHE_KUBERNETES_MAX_WRITE_PODS = "apollo.cache.kubernetes.max-write-pods";
90+
91+
/**
92+
* max number of pods that can write the configmap cache in Kubernetes environment variables
93+
*/
94+
public static final String APOLLO_CACHE_MAX_WRITE_PODS_ENVIRONMENT_VARIABLES = "APOLLO_CACHE_MAX_WRITE_PODS_NAMESPACE";
95+
8696
/**
8797
* apollo client access key
8898
*/

0 commit comments

Comments
 (0)