This is an automated email from the ASF dual-hosted git repository.
albumenj pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/dubbo.git
The following commit(s) were added to refs/heads/3.0 by this push:
new c125b99 Add Kubernetes Mesh Rule Support (#8350)
c125b99 is described below
commit c125b99189650b2f3ae148fb19c970435bba3563
Author: Albumen Kevin <[email protected]>
AuthorDate: Wed Aug 4 18:46:39 2021 +0800
Add Kubernetes Mesh Rule Support (#8350)
* Add Kubernetes Mesh Rule Support
* pretty comment
* fix ut
---
.../router/mesh/route/MeshAppRuleListener.java | 13 +-
.../cluster/router/mesh/route/MeshEnvListener.java | 38 ++++
.../cluster/router/mesh/route/MeshRuleManager.java | 33 ++--
.../org/apache/dubbo/common/utils/PojoUtils.java | 96 ++++++++++
.../dubbo/config/utils/ConfigValidationUtils.java | 2 +-
.../apache/dubbo/config/AbstractConfigTest.java | 4 +-
.../kubernetes/KubernetesMeshEnvListener.java | 197 +++++++++++++++++++++
.../kubernetes/KubernetesServiceDiscovery.java | 2 +
.../dubbo/registry/kubernetes/MeshConstant.java | 43 +++++
...o.rpc.cluster.router.mesh.route.MeshEnvListener | 1 +
10 files changed, 412 insertions(+), 17 deletions(-)
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListener.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListener.java
index 838c7ef..000a385 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListener.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListener.java
@@ -21,12 +21,15 @@ import
org.apache.dubbo.common.config.configcenter.ConfigChangedEvent;
import org.apache.dubbo.common.config.configcenter.ConfigurationListener;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.utils.PojoUtils;
import org.apache.dubbo.rpc.cluster.router.mesh.rule.VsDestinationGroup;
import
org.apache.dubbo.rpc.cluster.router.mesh.rule.destination.DestinationRule;
import
org.apache.dubbo.rpc.cluster.router.mesh.rule.virtualservice.VirtualServiceRule;
import
org.apache.dubbo.rpc.cluster.router.mesh.util.VsDestinationGroupRuleDispatcher;
import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+import org.yaml.snakeyaml.representer.Representer;
import java.text.MessageFormat;
import java.util.Map;
@@ -56,18 +59,20 @@ public class MeshAppRuleListener implements
ConfigurationListener {
VsDestinationGroup vsDestinationGroup = new VsDestinationGroup();
vsDestinationGroup.setAppName(appName);
- Yaml yaml = new Yaml();
- Yaml yaml2 = new Yaml();
+ Representer representer = new Representer();
+ representer.getPropertyUtils().setSkipMissingProperties(true);
+
+ Yaml yaml = new Yaml(new SafeConstructor());
Iterable<Object> objectIterable = yaml.loadAll(configInfo);
for (Object result : objectIterable) {
Map resultMap = (Map) result;
if ("DestinationRule".equals(resultMap.get("kind"))) {
- DestinationRule destinationRule =
yaml2.loadAs(yaml2.dump(result), DestinationRule.class);
+ DestinationRule destinationRule =
PojoUtils.mapToPojo(resultMap, DestinationRule.class);
vsDestinationGroup.getDestinationRuleList().add(destinationRule);
} else if ("VirtualService".equals(resultMap.get("kind"))) {
- VirtualServiceRule virtualServiceRule =
yaml2.loadAs(yaml2.dump(result), VirtualServiceRule.class);
+ VirtualServiceRule virtualServiceRule =
PojoUtils.mapToPojo(resultMap, VirtualServiceRule.class);
vsDestinationGroup.getVirtualServiceRuleList().add(virtualServiceRule);
}
}
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshEnvListener.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshEnvListener.java
new file mode 100644
index 0000000..4b92457
--- /dev/null
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshEnvListener.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.rpc.cluster.router.mesh.route;
+
+import org.apache.dubbo.common.extension.SPI;
+
+/**
+ * Mesh Rule Listener
+ * Such as Kubernetes, Service Mesh (xDS) environment support define rule in
env
+ */
+@SPI
+public interface MeshEnvListener {
+ /**
+ * @return whether current environment support listen
+ */
+ default boolean isEnable() {
+ return false;
+ }
+
+ void onSubscribe(String appName, MeshAppRuleListener listener);
+
+ void onUnSubscribe(String appName);
+}
diff --git
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManager.java
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManager.java
index 3be0972..87fdede 100644
---
a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManager.java
+++
b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManager.java
@@ -18,11 +18,13 @@
package org.apache.dubbo.rpc.cluster.router.mesh.route;
import org.apache.dubbo.common.config.configcenter.DynamicConfiguration;
+import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Collection;
+import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -39,23 +41,34 @@ public final class MeshRuleManager {
MeshAppRuleListener meshAppRuleListener = new MeshAppRuleListener(app);
String appRuleDataId = app + MESH_RULE_DATA_ID_SUFFIX;
DynamicConfiguration configuration =
ApplicationModel.getEnvironment().getDynamicConfiguration()
- .orElse(null);
+ .orElse(null);
- if (configuration == null) {
- logger.warn("Doesn't support DynamicConfiguration!");
+ Set<MeshEnvListener> envListeners =
ExtensionLoader.getExtensionLoader(MeshEnvListener.class).getSupportedExtensionInstances();
+
+ if (configuration == null &&
envListeners.stream().noneMatch(MeshEnvListener::isEnable)) {
+ logger.warn("Doesn't support Configuration!");
return;
}
- try {
- String rawConfig = configuration.getConfig(appRuleDataId,
DynamicConfiguration.DEFAULT_GROUP, 5000L);
- if (rawConfig != null) {
- meshAppRuleListener.receiveConfigInfo(rawConfig);
+ if(configuration != null) {
+ try {
+ String rawConfig = configuration.getConfig(appRuleDataId,
DynamicConfiguration.DEFAULT_GROUP, 5000L);
+ if (rawConfig != null) {
+ meshAppRuleListener.receiveConfigInfo(rawConfig);
+ }
+ } catch (Throwable throwable) {
+ logger.error("get MeshRuleManager app rule failed.",
throwable);
+ }
+
+ configuration.addListener(appRuleDataId,
DynamicConfiguration.DEFAULT_GROUP, meshAppRuleListener);
+ }
+
+ for (MeshEnvListener envListener : envListeners) {
+ if(envListener.isEnable()) {
+ envListener.onSubscribe(app, meshAppRuleListener);
}
- } catch (Throwable throwable) {
- logger.error("get MeshRuleManager app rule failed.", throwable);
}
- configuration.addListener(appRuleDataId,
DynamicConfiguration.DEFAULT_GROUP, meshAppRuleListener);
APP_RULE_LISTENERS.put(app, meshAppRuleListener);
}
diff --git
a/dubbo-common/src/main/java/org/apache/dubbo/common/utils/PojoUtils.java
b/dubbo-common/src/main/java/org/apache/dubbo/common/utils/PojoUtils.java
index 1a8330a..889ada1 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/utils/PojoUtils.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/utils/PojoUtils.java
@@ -23,6 +23,7 @@ import org.apache.dubbo.common.logger.LoggerFactory;
import java.lang.reflect.Array;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
+import java.lang.reflect.GenericArrayType;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
@@ -30,6 +31,8 @@ import java.lang.reflect.Modifier;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Proxy;
import java.lang.reflect.Type;
+import java.lang.reflect.TypeVariable;
+import java.lang.reflect.WildcardType;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -41,6 +44,7 @@ import java.util.IdentityHashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Properties;
import java.util.TreeMap;
import java.util.WeakHashMap;
@@ -50,6 +54,8 @@ import java.util.concurrent.ConcurrentSkipListMap;
import java.util.function.Consumer;
import java.util.function.Supplier;
+import static org.apache.dubbo.common.utils.ClassUtils.isAssignableFrom;
+
/**
* PojoUtils. Travel object deeply, and convert complex type to simple type.
* <p/>
@@ -68,6 +74,8 @@ public class PojoUtils {
private static final ConcurrentMap<String, Method> NAME_METHODS_CACHE =
new ConcurrentHashMap<String, Method>();
private static final ConcurrentMap<Class<?>, ConcurrentMap<String, Field>>
CLASS_FIELD_CACHE = new ConcurrentHashMap<Class<?>, ConcurrentMap<String,
Field>>();
private static final boolean GENERIC_WITH_CLZ =
Boolean.parseBoolean(ConfigUtils.getProperty(CommonConstants.GENERIC_WITH_CLZ_KEY,
"true"));
+ private static final List<Class<?>> CLASS_CAN_BE_STRING =
Arrays.asList(Byte.class, Short.class, Integer.class,
+ Long.class, Float.class, Double.class, Boolean.class, Character.class);
public static Object[] generalize(Object[] objs) {
Object[] dests = new Object[objs.length];
@@ -686,4 +694,92 @@ public class PojoUtils {
}
}
+ /**
+ * convert map to a specific class instance
+ *
+ * @param map map wait for convert
+ * @param cls the specified class
+ * @param <T> the type of {@code cls}
+ * @return class instance declare in param {@code cls}
+ * @throws ReflectiveOperationException if the instance creation is failed
+ * @since 2.7.10
+ */
+ public static <T> T mapToPojo(Map<String, Object> map, Class<T> cls)
throws ReflectiveOperationException {
+ T instance = cls.getDeclaredConstructor().newInstance();
+ Map<String, Field> beanPropertyFields =
ReflectUtils.getBeanPropertyFields(cls);
+ for (Map.Entry<String, Field> entry : beanPropertyFields.entrySet()) {
+ String name = entry.getKey();
+ Field field = entry.getValue();
+ Object mapObject = map.get(name);
+ if (mapObject == null) {
+ continue;
+ }
+
+ Type type = field.getGenericType();
+ Object fieldObject = getFieldObject(mapObject, type);
+ field.set(instance, fieldObject);
+ }
+
+ return instance;
+ }
+
+ private static Object getFieldObject(Object mapObject, Type fieldType)
throws ReflectiveOperationException {
+ if (fieldType instanceof Class<?>) {
+ return convertClassType(mapObject, (Class<?>) fieldType);
+ } else if (fieldType instanceof ParameterizedType) {
+ return convertParameterizedType(mapObject, (ParameterizedType)
fieldType);
+ } else if (fieldType instanceof GenericArrayType || fieldType
instanceof TypeVariable<?> || fieldType instanceof WildcardType) {
+ // ignore these type currently
+ return null;
+ } else {
+ throw new IllegalArgumentException("Unrecognized Type: " +
fieldType.toString());
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Object convertClassType(Object mapObject, Class<?> type)
throws ReflectiveOperationException {
+ if (type.isPrimitive() || isAssignableFrom(type,
mapObject.getClass())) {
+ return mapObject;
+ } else if (Objects.equals(type, String.class) &&
CLASS_CAN_BE_STRING.contains(mapObject.getClass())) {
+ // auto convert specified type to string
+ return mapObject.toString();
+ } else if (mapObject instanceof Map) {
+ return mapToPojo((Map<String, Object>) mapObject, type);
+ } else {
+ // type didn't match and mapObject is not another Map struct.
+ // we just ignore this situation.
+ return null;
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Object convertParameterizedType(Object mapObject,
ParameterizedType type) throws ReflectiveOperationException {
+ Type rawType = type.getRawType();
+ if (!isAssignableFrom((Class<?>) rawType, mapObject.getClass())) {
+ return null;
+ }
+
+ Type[] actualTypeArguments = type.getActualTypeArguments();
+ if (isAssignableFrom(Map.class, (Class<?>) rawType)) {
+ Map<Object, Object> map = (Map<Object, Object>)
mapObject.getClass().getDeclaredConstructor().newInstance();
+ for (Map.Entry<Object, Object> entry : ((Map<Object, Object>)
mapObject).entrySet()) {
+ Object key = getFieldObject(entry.getKey(),
actualTypeArguments[0]);
+ Object value = getFieldObject(entry.getValue(),
actualTypeArguments[1]);
+ map.put(key, value);
+ }
+
+ return map;
+ } else if (isAssignableFrom(Collection.class, (Class<?>) rawType)) {
+ Collection<Object> collection = (Collection<Object>)
mapObject.getClass().getDeclaredConstructor().newInstance();
+ for (Object m : (Iterable<?>) mapObject) {
+ Object ele = getFieldObject(m, actualTypeArguments[0]);
+ collection.add(ele);
+ }
+
+ return collection;
+ } else {
+ // ignore other type currently
+ return null;
+ }
+ }
}
diff --git
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/ConfigValidationUtils.java
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/ConfigValidationUtils.java
index 20c8e1a..03211ca 100644
---
a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/ConfigValidationUtils.java
+++
b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/ConfigValidationUtils.java
@@ -702,7 +702,7 @@ public class ConfigValidationUtils {
return;
}
if (value.length() > maxlength) {
- throw new IllegalStateException("Invalid " + property + "=\"" +
value + "\" is longer than " + maxlength);
+ logger.error("Invalid " + property + "=\"" + value + "\" is longer
than " + maxlength);
}
if (pattern != null) {
Matcher matcher = pattern.matcher(value);
diff --git
a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractConfigTest.java
b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractConfigTest.java
index 450ba35..9718452 100644
---
a/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractConfigTest.java
+++
b/dubbo-config/dubbo-config-api/src/test/java/org/apache/dubbo/config/AbstractConfigTest.java
@@ -209,7 +209,7 @@ public class AbstractConfigTest {
@Test
public void checkLength() throws Exception {
- Assertions.assertThrows(IllegalStateException.class, () -> {
+ Assertions.assertDoesNotThrow(() -> {
StringBuilder builder = new StringBuilder();
for (int i = 0; i <= 200; i++) {
builder.append('a');
@@ -220,7 +220,7 @@ public class AbstractConfigTest {
@Test
public void checkPathLength() throws Exception {
- Assertions.assertThrows(IllegalStateException.class, () -> {
+ Assertions.assertDoesNotThrow(() -> {
StringBuilder builder = new StringBuilder();
for (int i = 0; i <= 200; i++) {
builder.append('a');
diff --git
a/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListener.java
b/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListener.java
new file mode 100644
index 0000000..1a0c1fa
--- /dev/null
+++
b/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListener.java
@@ -0,0 +1,197 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.registry.kubernetes;
+
+import org.apache.dubbo.common.logger.Logger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.rpc.cluster.router.mesh.route.MeshAppRuleListener;
+import org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListener;
+
+import com.google.gson.Gson;
+import io.fabric8.kubernetes.api.model.ListOptionsBuilder;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.Watch;
+import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
+import org.yaml.snakeyaml.Yaml;
+import org.yaml.snakeyaml.constructor.SafeConstructor;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class KubernetesMeshEnvListener implements MeshEnvListener {
+ public static final Logger logger =
LoggerFactory.getLogger(KubernetesMeshEnvListener.class);
+ private volatile static boolean usingApiServer = false;
+ private volatile static KubernetesClient kubernetesClient;
+ private volatile static String namespace;
+
+ private final Map<String, MeshAppRuleListener> appRuleListenerMap = new
ConcurrentHashMap<>();
+
+ private final Map<String, Watch> vsAppWatch = new ConcurrentHashMap<>();
+ private final Map<String, Watch> drAppWatch = new ConcurrentHashMap<>();
+
+ private final Map<String, String> vsAppCache = new ConcurrentHashMap<>();
+ private final Map<String, String> drAppCache = new ConcurrentHashMap<>();
+
+ public static void injectKubernetesEnv(KubernetesClient client, String
configuredNamespace) {
+ usingApiServer = true;
+ kubernetesClient = client;
+ namespace = configuredNamespace;
+ }
+
+ @Override
+ public boolean isEnable() {
+ return usingApiServer;
+ }
+
+ @Override
+ public void onSubscribe(String appName, MeshAppRuleListener listener) {
+ appRuleListenerMap.put(appName, listener);
+ logger.info("Subscribe Mesh Rule in Kubernetes. AppName: " + appName);
+
+ // subscribe VisualService
+ subscribeVs(appName);
+
+ // subscribe DestinationRule
+ subscribeDr(appName);
+
+ // notify for start
+ notifyOnce(appName);
+ }
+
+ private void subscribeVs(String appName) {
+ if (vsAppWatch.containsKey(appName)) {
+ return;
+ }
+
+ try {
+ Watch watch = kubernetesClient
+ .customResource(
+ MeshConstant.getVsDefinition())
+ .watch(namespace, appName, null, new
ListOptionsBuilder().build(), new Watcher<String>() {
+ @Override
+ public void eventReceived(Action action, String resource) {
+ logger.info("Received VS Rule notification. AppName: "
+ appName + " Action:" + action + " Resource:" + resource);
+
+ if (action == Action.ADDED || action ==
Action.MODIFIED) {
+ Map drRuleMap = new Gson().fromJson(resource,
Map.class);
+ String vsRule = new Yaml(new
SafeConstructor()).dump(drRuleMap);
+ vsAppCache.put(appName, vsRule);
+ if (drAppCache.containsKey(appName)) {
+ notifyListener(vsRule, appName,
drAppCache.get(appName));
+ }
+ } else {
+
appRuleListenerMap.get(appName).receiveConfigInfo("");
+ }
+ }
+
+ @Override
+ public void onClose(WatcherException cause) {
+ // ignore
+ }
+ });
+ vsAppWatch.put(appName, watch);
+ try {
+ Map<String, Object> vsRule = kubernetesClient
+ .customResource(
+ MeshConstant.getVsDefinition())
+ .get(namespace, appName);
+ vsAppCache.put(appName, new Yaml(new
SafeConstructor()).dump(vsRule));
+ } catch (Throwable ignore) {
+
+ }
+ } catch (IOException e) {
+ logger.error("Error occurred when listen kubernetes crd.", e);
+ }
+ }
+
+ private void notifyListener(String vsRule, String appName, String drRule) {
+ String rule = vsRule + "\n---\n" + drRule;
+ logger.info("Notify App Rule Listener. AppName: " + appName + " Rule:"
+ rule);
+
+ appRuleListenerMap.get(appName).receiveConfigInfo(rule);
+ }
+
+ private void subscribeDr(String appName) {
+ if (drAppWatch.containsKey(appName)) {
+ return;
+ }
+
+ try {
+ Watch watch = kubernetesClient
+ .customResource(
+ MeshConstant.getDrDefinition())
+ .watch(namespace, appName, null, new
ListOptionsBuilder().build(), new Watcher<String>() {
+ @Override
+ public void eventReceived(Action action, String resource) {
+ logger.info("Received VS Rule notification. AppName: "
+ appName + " Action:" + action + " Resource:" + resource);
+
+ if (action == Action.ADDED || action ==
Action.MODIFIED) {
+ Map drRuleMap = new Gson().fromJson(resource,
Map.class);
+ String drRule = new Yaml(new
SafeConstructor()).dump(drRuleMap);
+
+ drAppCache.put(appName, drRule);
+ if (vsAppCache.containsKey(appName)) {
+ notifyListener(vsAppCache.get(appName),
appName, drRule);
+ }
+ } else {
+
appRuleListenerMap.get(appName).receiveConfigInfo("");
+ }
+ }
+
+ @Override
+ public void onClose(WatcherException cause) {
+ // ignore
+ }
+ });
+ drAppWatch.put(appName, watch);
+ try {
+ Map<String, Object> drRule = kubernetesClient
+ .customResource(
+ MeshConstant.getDrDefinition())
+ .get(namespace, appName);
+ drAppCache.put(appName, new Yaml(new
SafeConstructor()).dump(drRule));
+ } catch (Throwable ignore) {
+
+ }
+ } catch (IOException e) {
+ logger.error("Error occurred when listen kubernetes crd.", e);
+ }
+ }
+
+ private void notifyOnce(String appName) {
+ if (vsAppCache.containsKey(appName) &&
drAppCache.containsKey(appName)) {
+ notifyListener(vsAppCache.get(appName), appName,
drAppCache.get(appName));
+ }
+ }
+
+ @Override
+ public void onUnSubscribe(String appName) {
+ appRuleListenerMap.remove(appName);
+
+ if (vsAppWatch.containsKey(appName)) {
+ vsAppWatch.remove(appName).close();
+ }
+ vsAppCache.remove(appName);
+
+ if (drAppWatch.containsKey(appName)) {
+ drAppWatch.remove(appName).close();
+ }
+ drAppCache.remove(appName);
+ }
+}
diff --git
a/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java
b/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java
index 43262c7..0f61a57 100644
---
a/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java
+++
b/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java
@@ -96,6 +96,8 @@ public class KubernetesServiceDiscovery extends
AbstractServiceDiscovery {
" Master URL: " + config.getMasterUrl() +
" Hostname: " + currentHostname;
logger.error(message);
+ } else {
+ KubernetesMeshEnvListener.injectKubernetesEnv(kubernetesClient,
namespace);
}
}
diff --git
a/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/MeshConstant.java
b/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/MeshConstant.java
new file mode 100644
index 0000000..813bdd8
--- /dev/null
+++
b/dubbo-registry/dubbo-registry-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/MeshConstant.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.dubbo.registry.kubernetes;
+
+import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
+
+public class MeshConstant {
+ public static CustomResourceDefinitionContext getVsDefinition() {
+ // TODO cache
+ return new CustomResourceDefinitionContext.Builder()
+ .withGroup("service.dubbo.apache.org")
+ .withVersion("v1alpha1")
+ .withScope("Namespaced")
+ .withName("virtualservices.service.dubbo.apache.org")
+ .withPlural("virtualservices")
+ .withKind("VirtualService").build();
+ }
+
+ public static CustomResourceDefinitionContext getDrDefinition() {
+ // TODO cache
+ return new CustomResourceDefinitionContext.Builder()
+ .withGroup("service.dubbo.apache.org")
+ .withVersion("v1alpha1")
+ .withScope("Namespaced")
+ .withName("destinationrules.service.dubbo.apache.org")
+ .withPlural("destinationrules")
+ .withKind("DestinationRule").build();
+ }
+}
diff --git
a/dubbo-registry/dubbo-registry-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListener
b/dubbo-registry/dubbo-registry-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListener
new file mode 100644
index 0000000..b44c869
--- /dev/null
+++
b/dubbo-registry/dubbo-registry-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListener
@@ -0,0 +1 @@
+kubernetes=org.apache.dubbo.registry.kubernetes.KubernetesMeshEnvListener
