diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml
index 9577847696..99187e7f11 100644
--- a/dubbo-distribution/dubbo-all/pom.xml
+++ b/dubbo-distribution/dubbo-all/pom.xml
@@ -129,15 +129,6 @@
true
-
-
- org.apache.dubbo
- dubbo-kubernetes
- ${project.version}
- compile
- true
-
-
org.apache.dubbo
@@ -429,15 +420,6 @@
true
-
-
- org.apache.dubbo
- dubbo-xds
- ${project.version}
- compile
- true
-
-
org.springframework
@@ -557,8 +539,6 @@
org.apache.dubbo:dubbo-serialization-hessian2
org.apache.dubbo:dubbo-serialization-fastjson2
org.apache.dubbo:dubbo-serialization-jdk
- org.apache.dubbo:dubbo-kubernetes
- org.apache.dubbo:dubbo-xds
@@ -753,9 +733,6 @@
META-INF/dubbo/internal/org.apache.dubbo.registry.integration.RegistryProtocolListener
-
- META-INF/dubbo/internal/org.apache.dubbo.registry.xds.XdsCertificateSigner
-
META-INF/dubbo/internal/org.apache.dubbo.rpc.protocol.rest.filter.RestResponseInterceptor
diff --git a/dubbo-distribution/dubbo-bom/pom.xml b/dubbo-distribution/dubbo-bom/pom.xml
index 28d4302cec..dacced4646 100644
--- a/dubbo-distribution/dubbo-bom/pom.xml
+++ b/dubbo-distribution/dubbo-bom/pom.xml
@@ -175,13 +175,6 @@
${project.version}
-
-
- org.apache.dubbo
- dubbo-kubernetes
- ${project.version}
-
-
org.apache.dubbo
@@ -595,12 +588,6 @@
${project.version}
-
-
- org.apache.dubbo
- dubbo-xds
- ${project.version}
-
diff --git a/dubbo-distribution/dubbo-core-spi/pom.xml b/dubbo-distribution/dubbo-core-spi/pom.xml
index 88939753e9..212ea97fca 100644
--- a/dubbo-distribution/dubbo-core-spi/pom.xml
+++ b/dubbo-distribution/dubbo-core-spi/pom.xml
@@ -515,12 +515,6 @@
META-INF/dubbo/internal/org.apache.dubbo.registry.integration.RegistryProtocolListener
-
-
- META-INF/dubbo/internal/org.apache.dubbo.registry.xds.XdsCertificateSigner
-
-
diff --git a/dubbo-kubernetes/pom.xml b/dubbo-kubernetes/pom.xml
deleted file mode 100644
index 8f5a0e1acb..0000000000
--- a/dubbo-kubernetes/pom.xml
+++ /dev/null
@@ -1,73 +0,0 @@
-
-
-
- 4.0.0
-
- org.apache.dubbo
- dubbo-parent
- ${revision}
- ../pom.xml
-
-
- dubbo-kubernetes
- ${project.artifactId}
- The Kubernetes Integration
-
- false
-
-
-
-
- org.apache.dubbo
- dubbo-registry-api
- ${project.parent.version}
-
-
- org.apache.dubbo
- dubbo-common
- ${project.parent.version}
-
-
- org.apache.dubbo
- dubbo-metadata-api
- ${project.parent.version}
-
-
- io.fabric8
- kubernetes-client
-
-
- io.fabric8
- kubernetes-server-mock
- test
-
-
- org.mockito
- mockito-junit-jupiter
- 3.12.4
- test
-
-
- org.junit.jupiter
- junit-jupiter-api
-
-
-
-
-
-
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListener.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListener.java
deleted file mode 100644
index 43e79862fe..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListener.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * 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.ErrorTypeAwareLogger;
-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 java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import io.fabric8.kubernetes.api.model.GenericKubernetesResource;
-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.LoaderOptions;
-import org.yaml.snakeyaml.Yaml;
-import org.yaml.snakeyaml.constructor.SafeConstructor;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_LISTEN_KUBERNETES;
-
-public class KubernetesMeshEnvListener implements MeshEnvListener {
- public static final ErrorTypeAwareLogger logger =
- LoggerFactory.getErrorTypeAwareLogger(KubernetesMeshEnvListener.class);
- private static volatile boolean usingApiServer = false;
- private static volatile KubernetesClient kubernetesClient;
- private static volatile String namespace;
-
- private final Map appRuleListenerMap = new ConcurrentHashMap<>();
-
- private final Map vsAppWatch = new ConcurrentHashMap<>();
- private final Map drAppWatch = new ConcurrentHashMap<>();
-
- private final Map vsAppCache = new ConcurrentHashMap<>();
- private final Map 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
- .genericKubernetesResources(MeshConstant.getVsDefinition())
- .inNamespace(namespace)
- .withName(appName)
- .watch(new Watcher() {
- @Override
- public void eventReceived(Action action, GenericKubernetesResource resource) {
- if (logger.isInfoEnabled()) {
- logger.info("Received VS Rule notification. AppName: " + appName + " Action:" + action
- + " Resource:" + resource);
- }
-
- if (action == Action.ADDED || action == Action.MODIFIED) {
- String vsRule = new Yaml(new SafeConstructor(new LoaderOptions())).dump(resource);
- 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 {
- GenericKubernetesResource vsRule = kubernetesClient
- .genericKubernetesResources(MeshConstant.getVsDefinition())
- .inNamespace(namespace)
- .withName(appName)
- .get();
- vsAppCache.put(appName, new Yaml(new SafeConstructor(new LoaderOptions())).dump(vsRule));
- } catch (Throwable ignore) {
-
- }
- } catch (Exception e) {
- logger.error(REGISTRY_ERROR_LISTEN_KUBERNETES, "", "", "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
- .genericKubernetesResources(MeshConstant.getDrDefinition())
- .inNamespace(namespace)
- .withName(appName)
- .watch(new Watcher() {
- @Override
- public void eventReceived(Action action, GenericKubernetesResource resource) {
- if (logger.isInfoEnabled()) {
- logger.info("Received VS Rule notification. AppName: " + appName + " Action:" + action
- + " Resource:" + resource);
- }
-
- if (action == Action.ADDED || action == Action.MODIFIED) {
- String drRule = new Yaml(new SafeConstructor(new LoaderOptions())).dump(resource);
-
- 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 {
- GenericKubernetesResource drRule = kubernetesClient
- .genericKubernetesResources(MeshConstant.getDrDefinition())
- .inNamespace(namespace)
- .withName(appName)
- .get();
- drAppCache.put(appName, new Yaml(new SafeConstructor(new LoaderOptions())).dump(drRule));
- } catch (Throwable ignore) {
-
- }
- } catch (Exception e) {
- logger.error(REGISTRY_ERROR_LISTEN_KUBERNETES, "", "", "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-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListenerFactory.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListenerFactory.java
deleted file mode 100644
index 9d1c6d06cf..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesMeshEnvListenerFactory.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.MeshEnvListener;
-import org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListenerFactory;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-
-public class KubernetesMeshEnvListenerFactory implements MeshEnvListenerFactory {
- public static final Logger logger = LoggerFactory.getLogger(KubernetesMeshEnvListenerFactory.class);
- private final AtomicBoolean initialized = new AtomicBoolean(false);
- private MeshEnvListener listener = null;
-
- @Override
- public MeshEnvListener getListener() {
- try {
- if (initialized.compareAndSet(false, true)) {
- listener = new NopKubernetesMeshEnvListener();
- }
- } catch (Throwable t) {
- logger.info("Current Env not support Kubernetes.");
- }
- return listener;
- }
-}
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesRegistry.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesRegistry.java
deleted file mode 100644
index 2d51c1bfa2..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesRegistry.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.URL;
-import org.apache.dubbo.registry.NotifyListener;
-import org.apache.dubbo.registry.support.FailbackRegistry;
-
-/**
- * Empty implements for Kubernetes
- * Kubernetes only support `Service Discovery` mode register
- * Used to compat past version like 2.6.x, 2.7.x with interface level register
- * {@link KubernetesServiceDiscovery} is the real implementation of Kubernetes
- */
-public class KubernetesRegistry extends FailbackRegistry {
- public KubernetesRegistry(URL url) {
- super(url);
- }
-
- @Override
- public boolean isAvailable() {
- return true;
- }
-
- @Override
- public void doRegister(URL url) {}
-
- @Override
- public void doUnregister(URL url) {}
-
- @Override
- public void doSubscribe(URL url, NotifyListener listener) {}
-
- @Override
- public void doUnsubscribe(URL url, NotifyListener listener) {}
-}
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesRegistryFactory.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesRegistryFactory.java
deleted file mode 100644
index fe0e0477d7..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesRegistryFactory.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.URL;
-import org.apache.dubbo.registry.Registry;
-import org.apache.dubbo.registry.support.AbstractRegistryFactory;
-
-public class KubernetesRegistryFactory extends AbstractRegistryFactory {
-
- @Override
- protected String createRegistryCacheKey(URL url) {
- return url.toFullString();
- }
-
- @Override
- protected Registry createRegistry(URL url) {
- return new KubernetesRegistry(url);
- }
-}
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java
deleted file mode 100644
index 7792abdd2b..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscovery.java
+++ /dev/null
@@ -1,451 +0,0 @@
-/*
- * 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.URL;
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.JsonUtils;
-import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.registry.client.AbstractServiceDiscovery;
-import org.apache.dubbo.registry.client.DefaultServiceInstance;
-import org.apache.dubbo.registry.client.ServiceInstance;
-import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
-import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
-import org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst;
-import org.apache.dubbo.registry.kubernetes.util.KubernetesConfigUtils;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.rpc.model.ScopeModelUtil;
-
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Collectors;
-
-import io.fabric8.kubernetes.api.model.EndpointAddress;
-import io.fabric8.kubernetes.api.model.EndpointPort;
-import io.fabric8.kubernetes.api.model.EndpointSubset;
-import io.fabric8.kubernetes.api.model.Endpoints;
-import io.fabric8.kubernetes.api.model.Pod;
-import io.fabric8.kubernetes.api.model.PodBuilder;
-import io.fabric8.kubernetes.api.model.Service;
-import io.fabric8.kubernetes.client.Config;
-import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.KubernetesClientBuilder;
-import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
-import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_UNABLE_ACCESS_KUBERNETES;
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_UNABLE_FIND_SERVICE_KUBERNETES;
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_UNABLE_MATCH_KUBERNETES;
-
-public class KubernetesServiceDiscovery extends AbstractServiceDiscovery {
- private final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(getClass());
-
- private KubernetesClient kubernetesClient;
-
- private String currentHostname;
-
- private final URL registryURL;
-
- private final String namespace;
-
- private final boolean enableRegister;
-
- public static final String KUBERNETES_PROPERTIES_KEY = "io.dubbo/metadata";
-
- private static final ConcurrentHashMap SERVICE_UPDATE_TIME = new ConcurrentHashMap<>(64);
-
- private static final ConcurrentHashMap> SERVICE_INFORMER =
- new ConcurrentHashMap<>(64);
-
- private static final ConcurrentHashMap> PODS_INFORMER =
- new ConcurrentHashMap<>(64);
-
- private static final ConcurrentHashMap> ENDPOINTS_INFORMER =
- new ConcurrentHashMap<>(64);
-
- public KubernetesServiceDiscovery(ApplicationModel applicationModel, URL registryURL) {
- super(applicationModel, registryURL);
- Config config = KubernetesConfigUtils.createKubernetesConfig(registryURL);
- this.kubernetesClient = new KubernetesClientBuilder().withConfig(config).build();
- this.currentHostname = System.getenv("HOSTNAME");
- this.registryURL = registryURL;
- this.namespace = config.getNamespace();
- this.enableRegister = registryURL.getParameter(KubernetesClientConst.ENABLE_REGISTER, true);
-
- boolean availableAccess;
- try {
- availableAccess = kubernetesClient.pods().withName(currentHostname).get() != null;
- } catch (Throwable e) {
- availableAccess = false;
- }
- if (!availableAccess) {
- String message = "Unable to access api server. " + "Please check your url config."
- + " Master URL: "
- + config.getMasterUrl() + " Hostname: "
- + currentHostname;
- logger.error(REGISTRY_UNABLE_ACCESS_KUBERNETES, "", "", message);
- } else {
- KubernetesMeshEnvListener.injectKubernetesEnv(kubernetesClient, namespace);
- }
- }
-
- @Override
- public void doDestroy() {
- SERVICE_INFORMER.forEach((k, v) -> v.close());
- SERVICE_INFORMER.clear();
-
- PODS_INFORMER.forEach((k, v) -> v.close());
- PODS_INFORMER.clear();
-
- ENDPOINTS_INFORMER.forEach((k, v) -> v.close());
- ENDPOINTS_INFORMER.clear();
-
- kubernetesClient.close();
- }
-
- @Override
- public void doRegister(ServiceInstance serviceInstance) throws RuntimeException {
- if (enableRegister) {
- kubernetesClient
- .pods()
- .inNamespace(namespace)
- .withName(currentHostname)
- .edit(pod -> new PodBuilder(pod)
- .editOrNewMetadata()
- .addToAnnotations(
- KUBERNETES_PROPERTIES_KEY, JsonUtils.toJson(serviceInstance.getMetadata()))
- .endMetadata()
- .build());
- if (logger.isInfoEnabled()) {
- logger.info("Write Current Service Instance Metadata to Kubernetes pod. " + "Current pod name: "
- + currentHostname);
- }
- }
- }
-
- /**
- * Comparing to {@link AbstractServiceDiscovery#doUpdate(ServiceInstance, ServiceInstance)}, unregister() is unnecessary here.
- */
- @Override
- public void doUpdate(ServiceInstance oldServiceInstance, ServiceInstance newServiceInstance)
- throws RuntimeException {
- reportMetadata(newServiceInstance.getServiceMetadata());
- this.doRegister(newServiceInstance);
- }
-
- @Override
- public void doUnregister(ServiceInstance serviceInstance) throws RuntimeException {
- if (enableRegister) {
- kubernetesClient
- .pods()
- .inNamespace(namespace)
- .withName(currentHostname)
- .edit(pod -> new PodBuilder(pod)
- .editOrNewMetadata()
- .removeFromAnnotations(KUBERNETES_PROPERTIES_KEY)
- .endMetadata()
- .build());
- if (logger.isInfoEnabled()) {
- logger.info(
- "Remove Current Service Instance from Kubernetes pod. Current pod name: " + currentHostname);
- }
- }
- }
-
- @Override
- public Set getServices() {
- return kubernetesClient.services().inNamespace(namespace).list().getItems().stream()
- .map(service -> service.getMetadata().getName())
- .collect(Collectors.toSet());
- }
-
- @Override
- public List getInstances(String serviceName) throws NullPointerException {
- Endpoints endpoints = null;
- SharedIndexInformer endInformer = ENDPOINTS_INFORMER.get(serviceName);
- if (endInformer != null) {
- // get endpoints directly from informer local store
- List endpointsList = endInformer.getStore().list();
- if (endpointsList.size() > 0) {
- endpoints = endpointsList.get(0);
- }
- }
- if (endpoints == null) {
- endpoints = kubernetesClient
- .endpoints()
- .inNamespace(namespace)
- .withName(serviceName)
- .get();
- }
-
- return toServiceInstance(endpoints, serviceName);
- }
-
- @Override
- public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener)
- throws NullPointerException, IllegalArgumentException {
- listener.getServiceNames().forEach(serviceName -> {
- SERVICE_UPDATE_TIME.put(serviceName, new AtomicLong(0L));
-
- // Watch Service Endpoint Modification
- watchEndpoints(listener, serviceName);
-
- // Watch Pods Modification, happens when ServiceInstance updated
- watchPods(listener, serviceName);
-
- // Watch Service Modification, happens when Service Selector updated, used to update pods watcher
- watchService(listener, serviceName);
- });
- }
-
- private void watchEndpoints(ServiceInstancesChangedListener listener, String serviceName) {
- SharedIndexInformer endInformer = kubernetesClient
- .endpoints()
- .inNamespace(namespace)
- .withName(serviceName)
- .inform(new ResourceEventHandler() {
- @Override
- public void onAdd(Endpoints endpoints) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Endpoint Event. Event type: added. Current pod name: "
- + currentHostname + ". Endpoints is: " + endpoints);
- }
- notifyServiceChanged(serviceName, listener, toServiceInstance(endpoints, serviceName));
- }
-
- @Override
- public void onUpdate(Endpoints oldEndpoints, Endpoints newEndpoints) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Endpoint Event. Event type: updated. Current pod name: "
- + currentHostname + ". The new Endpoints is: " + newEndpoints);
- }
- notifyServiceChanged(serviceName, listener, toServiceInstance(newEndpoints, serviceName));
- }
-
- @Override
- public void onDelete(Endpoints endpoints, boolean deletedFinalStateUnknown) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Endpoint Event. Event type: deleted. Current pod name: "
- + currentHostname + ". Endpoints is: " + endpoints);
- }
- notifyServiceChanged(serviceName, listener, toServiceInstance(endpoints, serviceName));
- }
- });
-
- ENDPOINTS_INFORMER.put(serviceName, endInformer);
- }
-
- private void watchPods(ServiceInstancesChangedListener listener, String serviceName) {
- Map serviceSelector = getServiceSelector(serviceName);
- if (serviceSelector == null) {
- return;
- }
-
- SharedIndexInformer podInformer = kubernetesClient
- .pods()
- .inNamespace(namespace)
- .withLabels(serviceSelector)
- .inform(new ResourceEventHandler() {
- @Override
- public void onAdd(Pod pod) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Pods Event. Event type: added. Current pod name: " + currentHostname
- + ". Pod is: " + pod);
- }
- }
-
- @Override
- public void onUpdate(Pod oldPod, Pod newPod) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Pods Event. Event type: updated. Current pod name: "
- + currentHostname + ". new Pod is: " + newPod);
- }
-
- notifyServiceChanged(serviceName, listener, getInstances(serviceName));
- }
-
- @Override
- public void onDelete(Pod pod, boolean deletedFinalStateUnknown) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Pods Event. Event type: deleted. Current pod name: "
- + currentHostname + ". Pod is: " + pod);
- }
- }
- });
-
- PODS_INFORMER.put(serviceName, podInformer);
- }
-
- private void watchService(ServiceInstancesChangedListener listener, String serviceName) {
- SharedIndexInformer serviceInformer = kubernetesClient
- .services()
- .inNamespace(namespace)
- .withName(serviceName)
- .inform(new ResourceEventHandler() {
- @Override
- public void onAdd(Service service) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Service Added Event. " + "Current pod name: " + currentHostname);
- }
- }
-
- @Override
- public void onUpdate(Service oldService, Service newService) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Service Update Event. Update Pods Watcher. Current pod name: "
- + currentHostname + ". The new Service is: " + newService);
- }
- if (PODS_INFORMER.containsKey(serviceName)) {
- PODS_INFORMER.get(serviceName).close();
- PODS_INFORMER.remove(serviceName);
- }
- watchPods(listener, serviceName);
- }
-
- @Override
- public void onDelete(Service service, boolean deletedFinalStateUnknown) {
- if (logger.isDebugEnabled()) {
- logger.debug("Received Service Delete Event. " + "Current pod name: " + currentHostname);
- }
- }
- });
-
- SERVICE_INFORMER.put(serviceName, serviceInformer);
- }
-
- private void notifyServiceChanged(
- String serviceName, ServiceInstancesChangedListener listener, List serviceInstanceList) {
- long receivedTime = System.nanoTime();
-
- ServiceInstancesChangedEvent event;
-
- event = new ServiceInstancesChangedEvent(serviceName, serviceInstanceList);
-
- AtomicLong updateTime = SERVICE_UPDATE_TIME.get(serviceName);
- long lastUpdateTime = updateTime.get();
-
- if (lastUpdateTime <= receivedTime) {
- if (updateTime.compareAndSet(lastUpdateTime, receivedTime)) {
- listener.onEvent(event);
- return;
- }
- }
-
- if (logger.isInfoEnabled()) {
- logger.info("Discard Service Instance Data. "
- + "Possible Cause: Newer message has been processed or Failed to update time record by CAS. "
- + "Current Data received time: "
- + receivedTime + ". " + "Newer Data received time: "
- + lastUpdateTime + ".");
- }
- }
-
- @Override
- public URL getUrl() {
- return registryURL;
- }
-
- private Map getServiceSelector(String serviceName) {
- Service service = kubernetesClient
- .services()
- .inNamespace(namespace)
- .withName(serviceName)
- .get();
- if (service == null) {
- return null;
- }
- return service.getSpec().getSelector();
- }
-
- private List toServiceInstance(Endpoints endpoints, String serviceName) {
- Map serviceSelector = getServiceSelector(serviceName);
- if (serviceSelector == null) {
- return new LinkedList<>();
- }
- Map pods =
- kubernetesClient.pods().inNamespace(namespace).withLabels(serviceSelector).list().getItems().stream()
- .collect(Collectors.toMap(pod -> pod.getMetadata().getName(), pod -> pod));
-
- List instances = new LinkedList<>();
- Set instancePorts = new HashSet<>();
-
- for (EndpointSubset endpointSubset : endpoints.getSubsets()) {
- instancePorts.addAll(endpointSubset.getPorts().stream()
- .map(EndpointPort::getPort)
- .collect(Collectors.toSet()));
- }
-
- for (EndpointSubset endpointSubset : endpoints.getSubsets()) {
- for (EndpointAddress address : endpointSubset.getAddresses()) {
- Pod pod = pods.get(address.getTargetRef().getName());
- String ip = address.getIp();
- if (pod == null) {
- logger.warn(
- REGISTRY_UNABLE_MATCH_KUBERNETES,
- "",
- "",
- "Unable to match Kubernetes Endpoint address with Pod. " + "EndpointAddress Hostname: "
- + address.getTargetRef().getName());
- continue;
- }
- instancePorts.forEach(port -> {
- ServiceInstance serviceInstance = new DefaultServiceInstance(
- serviceName, ip, port, ScopeModelUtil.getApplicationModel(getUrl().getScopeModel()));
-
- String properties = pod.getMetadata().getAnnotations().get(KUBERNETES_PROPERTIES_KEY);
- if (StringUtils.isNotEmpty(properties)) {
- serviceInstance.getMetadata().putAll(JsonUtils.toJavaObject(properties, Map.class));
- instances.add(serviceInstance);
- } else {
- logger.warn(
- REGISTRY_UNABLE_FIND_SERVICE_KUBERNETES,
- "",
- "",
- "Unable to find Service Instance metadata in Pod Annotations. "
- + "Possibly cause: provider has not been initialized successfully. "
- + "EndpointAddress Hostname: "
- + address.getTargetRef().getName());
- }
- });
- }
- }
-
- return instances;
- }
-
- /**
- * UT used only
- */
- @Deprecated
- public void setCurrentHostname(String currentHostname) {
- this.currentHostname = currentHostname;
- }
-
- /**
- * UT used only
- */
- @Deprecated
- public void setKubernetesClient(KubernetesClient kubernetesClient) {
- this.kubernetesClient = kubernetesClient;
- }
-}
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryFactory.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryFactory.java
deleted file mode 100644
index 7d11dfaaa8..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryFactory.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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.URL;
-import org.apache.dubbo.registry.client.AbstractServiceDiscoveryFactory;
-import org.apache.dubbo.registry.client.ServiceDiscovery;
-
-public class KubernetesServiceDiscoveryFactory extends AbstractServiceDiscoveryFactory {
- @Override
- protected ServiceDiscovery createDiscovery(URL registryURL) {
- return new KubernetesServiceDiscovery(applicationModel, registryURL);
- }
-}
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/MeshConstant.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/MeshConstant.java
deleted file mode 100644
index e8b8f8407f..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/MeshConstant.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/NopKubernetesMeshEnvListener.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/NopKubernetesMeshEnvListener.java
deleted file mode 100644
index 818b8df64d..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/NopKubernetesMeshEnvListener.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.rpc.cluster.router.mesh.route.MeshAppRuleListener;
-import org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListener;
-
-public class NopKubernetesMeshEnvListener implements MeshEnvListener {
-
- @Override
- public boolean isEnable() {
- return false;
- }
-
- @Override
- public void onSubscribe(String appName, MeshAppRuleListener listener) {}
-
- @Override
- public void onUnSubscribe(String appName) {}
-}
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/util/KubernetesClientConst.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/util/KubernetesClientConst.java
deleted file mode 100644
index b7ace53894..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/util/KubernetesClientConst.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.util;
-
-public class KubernetesClientConst {
- public static final String DEFAULT_MASTER_PLACEHOLDER = "DEFAULT_MASTER_HOST";
- public static final String DEFAULT_MASTER_URL = "https://kubernetes.default.svc";
-
- public static final String ENABLE_REGISTER = "enableRegister";
-
- public static final String TRUST_CERTS = "trustCerts";
-
- public static final String USE_HTTPS = "useHttps";
-
- public static final String HTTP2_DISABLE = "http2Disable";
-
- public static final String NAMESPACE = "namespace";
-
- public static final String API_VERSION = "apiVersion";
-
- public static final String CA_CERT_FILE = "caCertFile";
-
- public static final String CA_CERT_DATA = "caCertData";
-
- public static final String CLIENT_CERT_FILE = "clientCertFile";
-
- public static final String CLIENT_CERT_DATA = "clientCertData";
-
- public static final String CLIENT_KEY_FILE = "clientKeyFile";
-
- public static final String CLIENT_KEY_DATA = "clientKeyData";
-
- public static final String CLIENT_KEY_ALGO = "clientKeyAlgo";
-
- public static final String CLIENT_KEY_PASSPHRASE = "clientKeyPassphrase";
-
- public static final String OAUTH_TOKEN = "oauthToken";
-
- public static final String USERNAME = "username";
-
- public static final String PASSWORD = "password";
-
- public static final String WATCH_RECONNECT_INTERVAL = "watchReconnectInterval";
-
- public static final String WATCH_RECONNECT_LIMIT = "watchReconnectLimit";
-
- public static final String CONNECTION_TIMEOUT = "connectionTimeout";
-
- public static final String REQUEST_TIMEOUT = "requestTimeout";
-
- public static final String ROLLING_TIMEOUT = "rollingTimeout";
-
- public static final String LOGGING_INTERVAL = "loggingInterval";
-
- public static final String HTTP_PROXY = "httpProxy";
-
- public static final String HTTPS_PROXY = "httpsProxy";
-
- public static final String PROXY_USERNAME = "proxyUsername";
-
- public static final String PROXY_PASSWORD = "proxyPassword";
-
- public static final String NO_PROXY = "noProxy";
-}
diff --git a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/util/KubernetesConfigUtils.java b/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/util/KubernetesConfigUtils.java
deleted file mode 100644
index 332bb6e733..0000000000
--- a/dubbo-kubernetes/src/main/java/org/apache/dubbo/registry/kubernetes/util/KubernetesConfigUtils.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * 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.util;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.utils.StringUtils;
-
-import java.util.Base64;
-
-import io.fabric8.kubernetes.client.Config;
-import io.fabric8.kubernetes.client.ConfigBuilder;
-
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.API_VERSION;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CA_CERT_DATA;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CA_CERT_FILE;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CLIENT_CERT_DATA;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CLIENT_CERT_FILE;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CLIENT_KEY_ALGO;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CLIENT_KEY_DATA;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CLIENT_KEY_FILE;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CLIENT_KEY_PASSPHRASE;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.CONNECTION_TIMEOUT;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.DEFAULT_MASTER_PLACEHOLDER;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.DEFAULT_MASTER_URL;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.HTTP2_DISABLE;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.HTTPS_PROXY;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.HTTP_PROXY;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.LOGGING_INTERVAL;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.NAMESPACE;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.NO_PROXY;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.OAUTH_TOKEN;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.PASSWORD;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.PROXY_PASSWORD;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.PROXY_USERNAME;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.REQUEST_TIMEOUT;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.TRUST_CERTS;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.USERNAME;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.USE_HTTPS;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.WATCH_RECONNECT_INTERVAL;
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.WATCH_RECONNECT_LIMIT;
-
-public class KubernetesConfigUtils {
-
- public static Config createKubernetesConfig(URL url) {
- // Init default config
- Config base = Config.autoConfigure(null);
-
- // replace config with parameters if presents
- return new ConfigBuilder(base) //
- .withMasterUrl(buildMasterUrl(url)) //
- .withApiVersion(url.getParameter(API_VERSION, base.getApiVersion())) //
- .withNamespace(url.getParameter(NAMESPACE, base.getNamespace())) //
- .withUsername(url.getParameter(USERNAME, base.getUsername())) //
- .withPassword(url.getParameter(PASSWORD, base.getPassword())) //
- .withOauthToken(url.getParameter(OAUTH_TOKEN, base.getOauthToken())) //
- .withCaCertFile(url.getParameter(CA_CERT_FILE, base.getCaCertFile())) //
- .withCaCertData(url.getParameter(CA_CERT_DATA, decodeBase64(base.getCaCertData()))) //
- .withClientKeyFile(url.getParameter(CLIENT_KEY_FILE, base.getClientKeyFile())) //
- .withClientKeyData(url.getParameter(CLIENT_KEY_DATA, decodeBase64(base.getClientKeyData()))) //
- .withClientCertFile(url.getParameter(CLIENT_CERT_FILE, base.getClientCertFile())) //
- .withClientCertData(url.getParameter(CLIENT_CERT_DATA, decodeBase64(base.getClientCertData()))) //
- .withClientKeyAlgo(url.getParameter(CLIENT_KEY_ALGO, base.getClientKeyAlgo())) //
- .withClientKeyPassphrase(url.getParameter(CLIENT_KEY_PASSPHRASE, base.getClientKeyPassphrase())) //
- .withConnectionTimeout(url.getParameter(CONNECTION_TIMEOUT, base.getConnectionTimeout())) //
- .withRequestTimeout(url.getParameter(REQUEST_TIMEOUT, base.getRequestTimeout())) //
- .withWatchReconnectInterval(
- url.getParameter(WATCH_RECONNECT_INTERVAL, base.getWatchReconnectInterval())) //
- .withWatchReconnectLimit(url.getParameter(WATCH_RECONNECT_LIMIT, base.getWatchReconnectLimit())) //
- .withLoggingInterval(url.getParameter(LOGGING_INTERVAL, base.getLoggingInterval())) //
- .withTrustCerts(url.getParameter(TRUST_CERTS, base.isTrustCerts())) //
- .withHttp2Disable(url.getParameter(HTTP2_DISABLE, base.isHttp2Disable())) //
- .withHttpProxy(url.getParameter(HTTP_PROXY, base.getHttpProxy())) //
- .withHttpsProxy(url.getParameter(HTTPS_PROXY, base.getHttpsProxy())) //
- .withProxyUsername(url.getParameter(PROXY_USERNAME, base.getProxyUsername())) //
- .withProxyPassword(url.getParameter(PROXY_PASSWORD, base.getProxyPassword())) //
- .withNoProxy(url.getParameter(NO_PROXY, base.getNoProxy())) //
- .build();
- }
-
- private static String buildMasterUrl(URL url) {
- if (DEFAULT_MASTER_PLACEHOLDER.equalsIgnoreCase(url.getHost())) {
- return DEFAULT_MASTER_URL;
- }
- return (url.getParameter(USE_HTTPS, true) ? "https://" : "http://") + url.getHost() + ":" + url.getPort();
- }
-
- private static String decodeBase64(String str) {
- return StringUtils.isNotEmpty(str) ? new String(Base64.getDecoder().decode(str)) : null;
- }
-}
diff --git a/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory b/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory
deleted file mode 100644
index 94177d81ea..0000000000
--- a/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory
+++ /dev/null
@@ -1 +0,0 @@
-kubernetes=org.apache.dubbo.registry.kubernetes.KubernetesRegistryFactory
\ No newline at end of file
diff --git a/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory b/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory
deleted file mode 100644
index 4301ab8b4b..0000000000
--- a/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory
+++ /dev/null
@@ -1 +0,0 @@
-kubernetes=org.apache.dubbo.registry.kubernetes.KubernetesServiceDiscoveryFactory
\ No newline at end of file
diff --git a/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListenerFactory b/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListenerFactory
deleted file mode 100644
index 4dfae84b80..0000000000
--- a/dubbo-kubernetes/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.mesh.route.MeshEnvListenerFactory
+++ /dev/null
@@ -1 +0,0 @@
-kubernetes=org.apache.dubbo.registry.kubernetes.KubernetesMeshEnvListenerFactory
diff --git a/dubbo-kubernetes/src/test/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryTest.java b/dubbo-kubernetes/src/test/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryTest.java
deleted file mode 100644
index 6f47be6015..0000000000
--- a/dubbo-kubernetes/src/test/java/org/apache/dubbo/registry/kubernetes/KubernetesServiceDiscoveryTest.java
+++ /dev/null
@@ -1,289 +0,0 @@
-/*
- * 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.URL;
-import org.apache.dubbo.config.ApplicationConfig;
-import org.apache.dubbo.registry.client.DefaultServiceInstance;
-import org.apache.dubbo.registry.client.ServiceInstance;
-import org.apache.dubbo.registry.client.event.ServiceInstancesChangedEvent;
-import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
-import org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.rpc.model.ScopeModelUtil;
-
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-
-import io.fabric8.kubernetes.api.model.Endpoints;
-import io.fabric8.kubernetes.api.model.EndpointsBuilder;
-import io.fabric8.kubernetes.api.model.Pod;
-import io.fabric8.kubernetes.api.model.PodBuilder;
-import io.fabric8.kubernetes.api.model.Service;
-import io.fabric8.kubernetes.api.model.ServiceBuilder;
-import io.fabric8.kubernetes.client.Config;
-import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
-import io.fabric8.kubernetes.client.server.mock.KubernetesServer;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.ArgumentCaptor;
-import org.mockito.Mockito;
-import org.mockito.junit.jupiter.MockitoExtension;
-
-import static org.apache.dubbo.registry.kubernetes.util.KubernetesClientConst.NAMESPACE;
-import static org.awaitility.Awaitility.await;
-
-@ExtendWith({MockitoExtension.class})
-class KubernetesServiceDiscoveryTest {
-
- private static final String SERVICE_NAME = "TestService";
-
- private static final String POD_NAME = "TestServer";
-
- public KubernetesServer mockServer = new KubernetesServer(false, true);
-
- private NamespacedKubernetesClient mockClient;
-
- private ServiceInstancesChangedListener mockListener = Mockito.mock(ServiceInstancesChangedListener.class);
-
- private URL serverUrl;
-
- private Map selector;
-
- private KubernetesServiceDiscovery serviceDiscovery;
-
- @BeforeEach
- public void setUp() {
- mockServer.before();
- mockClient = mockServer.getClient().inNamespace("dubbo-demo");
-
- ApplicationModel applicationModel = ApplicationModel.defaultModel();
- applicationModel.getApplicationConfigManager().setApplication(new ApplicationConfig());
-
- serverUrl = URL.valueOf(mockClient.getConfiguration().getMasterUrl())
- .setProtocol("kubernetes")
- .addParameter(NAMESPACE, "dubbo-demo")
- .addParameter(KubernetesClientConst.USE_HTTPS, "false")
- .addParameter(KubernetesClientConst.HTTP2_DISABLE, "true");
- serverUrl.setScopeModel(applicationModel);
-
- this.serviceDiscovery = new KubernetesServiceDiscovery(applicationModel, serverUrl);
-
- System.setProperty(Config.KUBERNETES_AUTH_TRYKUBECONFIG_SYSTEM_PROPERTY, "false");
- System.setProperty(Config.KUBERNETES_AUTH_TRYSERVICEACCOUNT_SYSTEM_PROPERTY, "false");
-
- selector = new HashMap<>(4);
- selector.put("l", "v");
- Pod pod = new PodBuilder()
- .withNewMetadata()
- .withName(POD_NAME)
- .withLabels(selector)
- .endMetadata()
- .build();
-
- Service service = new ServiceBuilder()
- .withNewMetadata()
- .withName(SERVICE_NAME)
- .endMetadata()
- .withNewSpec()
- .withSelector(selector)
- .endSpec()
- .build();
-
- Endpoints endPoints = new EndpointsBuilder()
- .withNewMetadata()
- .withName(SERVICE_NAME)
- .endMetadata()
- .addNewSubset()
- .addNewAddress()
- .withIp("ip1")
- .withNewTargetRef()
- .withUid("uid1")
- .withName(POD_NAME)
- .endTargetRef()
- .endAddress()
- .addNewPort("Test", "Test", 12345, "TCP")
- .endSubset()
- .build();
-
- mockClient.pods().resource(pod).create();
- mockClient.services().resource(service).create();
- mockClient.endpoints().resource(endPoints).create();
- }
-
- @AfterEach
- public void destroy() throws Exception {
- serviceDiscovery.destroy();
- mockClient.close();
- mockServer.after();
- }
-
- @Test
- void testEndpointsUpdate() {
- serviceDiscovery.setCurrentHostname(POD_NAME);
- serviceDiscovery.setKubernetesClient(mockClient);
-
- ServiceInstance serviceInstance = new DefaultServiceInstance(
- SERVICE_NAME,
- "Test",
- 12345,
- ScopeModelUtil.getApplicationModel(serviceDiscovery.getUrl().getScopeModel()));
-
- serviceDiscovery.doRegister(serviceInstance);
-
- HashSet serviceList = new HashSet<>(4);
- serviceList.add(SERVICE_NAME);
- Mockito.when(mockListener.getServiceNames()).thenReturn(serviceList);
- Mockito.doNothing().when(mockListener).onEvent(Mockito.any());
-
- serviceDiscovery.addServiceInstancesChangedListener(mockListener);
- mockClient.endpoints().withName(SERVICE_NAME).edit(endpoints -> new EndpointsBuilder(endpoints)
- .editFirstSubset()
- .addNewAddress()
- .withIp("ip2")
- .withNewTargetRef()
- .withUid("uid2")
- .withName(POD_NAME)
- .endTargetRef()
- .endAddress()
- .endSubset()
- .build());
-
- await().until(() -> {
- ArgumentCaptor captor =
- ArgumentCaptor.forClass(ServiceInstancesChangedEvent.class);
- Mockito.verify(mockListener, Mockito.atLeast(0)).onEvent(captor.capture());
- return captor.getValue().getServiceInstances().size() == 2;
- });
- ArgumentCaptor eventArgumentCaptor =
- ArgumentCaptor.forClass(ServiceInstancesChangedEvent.class);
- Mockito.verify(mockListener, Mockito.times(2)).onEvent(eventArgumentCaptor.capture());
- Assertions.assertEquals(
- 2, eventArgumentCaptor.getValue().getServiceInstances().size());
-
- serviceDiscovery.doUnregister(serviceInstance);
- }
-
- @Test
- void testPodsUpdate() throws Exception {
- serviceDiscovery.setCurrentHostname(POD_NAME);
- serviceDiscovery.setKubernetesClient(mockClient);
-
- ServiceInstance serviceInstance = new DefaultServiceInstance(
- SERVICE_NAME,
- "Test",
- 12345,
- ScopeModelUtil.getApplicationModel(serviceDiscovery.getUrl().getScopeModel()));
-
- serviceDiscovery.doRegister(serviceInstance);
-
- HashSet serviceList = new HashSet<>(4);
- serviceList.add(SERVICE_NAME);
- Mockito.when(mockListener.getServiceNames()).thenReturn(serviceList);
- Mockito.doNothing().when(mockListener).onEvent(Mockito.any());
-
- serviceDiscovery.addServiceInstancesChangedListener(mockListener);
-
- serviceInstance = new DefaultServiceInstance(
- SERVICE_NAME,
- "Test12345",
- 12345,
- ScopeModelUtil.getApplicationModel(serviceDiscovery.getUrl().getScopeModel()));
- serviceDiscovery.doUpdate(serviceInstance, serviceInstance);
-
- await().until(() -> {
- ArgumentCaptor captor =
- ArgumentCaptor.forClass(ServiceInstancesChangedEvent.class);
- Mockito.verify(mockListener, Mockito.atLeast(0)).onEvent(captor.capture());
- return captor.getValue().getServiceInstances().size() == 1;
- });
- ArgumentCaptor eventArgumentCaptor =
- ArgumentCaptor.forClass(ServiceInstancesChangedEvent.class);
- Mockito.verify(mockListener, Mockito.times(1)).onEvent(eventArgumentCaptor.capture());
- Assertions.assertEquals(
- 1, eventArgumentCaptor.getValue().getServiceInstances().size());
-
- serviceDiscovery.doUnregister(serviceInstance);
- }
-
- @Test
- void testServiceUpdate() {
- serviceDiscovery.setCurrentHostname(POD_NAME);
- serviceDiscovery.setKubernetesClient(mockClient);
-
- ServiceInstance serviceInstance = new DefaultServiceInstance(
- SERVICE_NAME,
- "Test",
- 12345,
- ScopeModelUtil.getApplicationModel(serviceDiscovery.getUrl().getScopeModel()));
-
- serviceDiscovery.doRegister(serviceInstance);
-
- HashSet serviceList = new HashSet<>(4);
- serviceList.add(SERVICE_NAME);
- Mockito.when(mockListener.getServiceNames()).thenReturn(serviceList);
- Mockito.doNothing().when(mockListener).onEvent(Mockito.any());
-
- serviceDiscovery.addServiceInstancesChangedListener(mockListener);
-
- selector.put("app", "test");
- mockClient.services().withName(SERVICE_NAME).edit(service -> new ServiceBuilder(service)
- .editSpec()
- .addToSelector(selector)
- .endSpec()
- .build());
-
- await().until(() -> {
- ArgumentCaptor captor =
- ArgumentCaptor.forClass(ServiceInstancesChangedEvent.class);
- Mockito.verify(mockListener, Mockito.atLeast(0)).onEvent(captor.capture());
- return captor.getValue().getServiceInstances().size() == 1;
- });
- ArgumentCaptor eventArgumentCaptor =
- ArgumentCaptor.forClass(ServiceInstancesChangedEvent.class);
- Mockito.verify(mockListener, Mockito.times(1)).onEvent(eventArgumentCaptor.capture());
- Assertions.assertEquals(
- 1, eventArgumentCaptor.getValue().getServiceInstances().size());
-
- serviceDiscovery.doUnregister(serviceInstance);
- }
-
- @Test
- void testGetInstance() {
- serviceDiscovery.setCurrentHostname(POD_NAME);
- serviceDiscovery.setKubernetesClient(mockClient);
-
- ServiceInstance serviceInstance = new DefaultServiceInstance(
- SERVICE_NAME,
- "Test",
- 12345,
- ScopeModelUtil.getApplicationModel(serviceDiscovery.getUrl().getScopeModel()));
-
- serviceDiscovery.doRegister(serviceInstance);
-
- serviceDiscovery.doUpdate(serviceInstance, serviceInstance);
-
- Assertions.assertEquals(1, serviceDiscovery.getServices().size());
- Assertions.assertEquals(1, serviceDiscovery.getInstances(SERVICE_NAME).size());
-
- serviceDiscovery.doUnregister(serviceInstance);
- }
-}
diff --git a/dubbo-kubernetes/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker b/dubbo-kubernetes/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
deleted file mode 100644
index ca6ee9cea8..0000000000
--- a/dubbo-kubernetes/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker
+++ /dev/null
@@ -1 +0,0 @@
-mock-maker-inline
\ No newline at end of file
diff --git a/dubbo-test/dubbo-dependencies-all/pom.xml b/dubbo-test/dubbo-dependencies-all/pom.xml
index b39ac92a5c..b0632440c5 100644
--- a/dubbo-test/dubbo-dependencies-all/pom.xml
+++ b/dubbo-test/dubbo-dependencies-all/pom.xml
@@ -120,13 +120,6 @@
${project.version}
-
-
- org.apache.dubbo
- dubbo-kubernetes
- ${project.version}
-
-
org.apache.dubbo
@@ -471,11 +464,5 @@
${project.version}
-
-
- org.apache.dubbo
- dubbo-xds
- ${project.version}
-
diff --git a/dubbo-xds/pom.xml b/dubbo-xds/pom.xml
deleted file mode 100644
index 9e9ef216d2..0000000000
--- a/dubbo-xds/pom.xml
+++ /dev/null
@@ -1,123 +0,0 @@
-
-
-
- 4.0.0
-
- org.apache.dubbo
- dubbo-parent
- ${revision}
- ../pom.xml
-
-
- dubbo-xds
- ${project.artifactId}
- The xDS Integration
-
- false
-
-
-
-
-
- org.apache.dubbo
- dubbo-registry-api
- ${project.version}
-
-
-
- org.apache.dubbo
- dubbo-common
- ${project.version}
-
-
-
- io.grpc
- grpc-protobuf
-
-
-
- io.grpc
- grpc-stub
-
-
-
- io.grpc
- grpc-netty-shaded
-
-
-
- io.envoyproxy.controlplane
- api
-
-
-
- com.google.protobuf
- protobuf-java
-
-
-
- com.google.protobuf
- protobuf-java-util
-
-
-
- org.bouncycastle
- bcprov-jdk15on
-
-
- org.bouncycastle
- bcpkix-jdk15on
-
-
- org.bouncycastle
- bcprov-ext-jdk15on
-
-
-
-
-
-
-
- org.xolstice.maven.plugins
- protobuf-maven-plugin
- ${maven_protobuf_plugin_version}
-
- com.google.protobuf:protoc:${protobuf-java_version}:exe:${os.detected.classifier}
- grpc-java
- io.grpc:protoc-gen-grpc-java:${grpc_version}:exe:${os.detected.classifier}
-
-
-
-
- compile
- compile-custom
-
-
-
-
-
-
-
- kr.motd.maven
- os-maven-plugin
- ${maven_os_plugin_version}
-
-
-
-
-
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java
deleted file mode 100644
index 1ef1a8b31c..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsCertificateSigner.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.xds;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.extension.Adaptive;
-import org.apache.dubbo.common.extension.SPI;
-
-@SPI
-public interface XdsCertificateSigner {
-
- @Adaptive(value = "signer")
- CertPair GenerateCert(URL url);
-
- class CertPair {
- private final String privateKey;
- private final String publicKey;
- private final long createTime;
- private final long expireTime;
-
- public CertPair(String privateKey, String publicKey, long createTime, long expireTime) {
- this.privateKey = privateKey;
- this.publicKey = publicKey;
- this.createTime = createTime;
- this.expireTime = expireTime;
- }
-
- public String getPrivateKey() {
- return privateKey;
- }
-
- public String getPublicKey() {
- return publicKey;
- }
-
- public long getCreateTime() {
- return createTime;
- }
-
- public boolean isExpire() {
- return System.currentTimeMillis() < expireTime;
- }
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java
deleted file mode 100644
index c09ea80877..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsEnv.java
+++ /dev/null
@@ -1,22 +0,0 @@
-/*
- * 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.xds;
-
-public interface XdsEnv {
-
- String getCluster();
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsInitializationException.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsInitializationException.java
deleted file mode 100644
index 9a9b9a35ca..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsInitializationException.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/*
- * 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.xds;
-
-public final class XdsInitializationException extends Exception {
-
- public XdsInitializationException(String message) {
- super(message);
- }
-
- public XdsInitializationException(String message, Throwable cause) {
- super(message, cause);
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistry.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistry.java
deleted file mode 100644
index 30e005ffe7..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistry.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * 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.xds;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.registry.NotifyListener;
-import org.apache.dubbo.registry.support.FailbackRegistry;
-
-/**
- * Empty implements for xDS
- * xDS only support `Service Discovery` mode register
- * Used to compat past version like 2.6.x, 2.7.x with interface level register
- * {@link XdsServiceDiscovery} is the real implementation of xDS
- */
-public class XdsRegistry extends FailbackRegistry {
- public XdsRegistry(URL url) {
- super(url);
- }
-
- @Override
- public boolean isAvailable() {
- return true;
- }
-
- @Override
- public void doRegister(URL url) {}
-
- @Override
- public void doUnregister(URL url) {}
-
- @Override
- public void doSubscribe(URL url, NotifyListener listener) {}
-
- @Override
- public void doUnsubscribe(URL url, NotifyListener listener) {}
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistryFactory.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistryFactory.java
deleted file mode 100644
index 8fa130b197..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistryFactory.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.xds;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.registry.Registry;
-import org.apache.dubbo.registry.support.AbstractRegistryFactory;
-
-public class XdsRegistryFactory extends AbstractRegistryFactory {
-
- @Override
- protected String createRegistryCacheKey(URL url) {
- return url.toFullString();
- }
-
- @Override
- protected Registry createRegistry(URL url) {
- return new XdsRegistry(url);
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java
deleted file mode 100644
index 3385ff97a0..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java
+++ /dev/null
@@ -1,117 +0,0 @@
-/*
- * 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.xds;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.registry.client.DefaultServiceInstance;
-import org.apache.dubbo.registry.client.ReflectionBasedServiceDiscovery;
-import org.apache.dubbo.registry.client.ServiceInstance;
-import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener;
-import org.apache.dubbo.registry.xds.util.PilotExchanger;
-import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-import org.apache.dubbo.rpc.model.ScopeModelUtil;
-
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_INITIALIZE_XDS;
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_PARSING_XDS;
-
-public class XdsServiceDiscovery extends ReflectionBasedServiceDiscovery {
-
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(XdsServiceDiscovery.class);
-
- private PilotExchanger exchanger;
-
- public XdsServiceDiscovery(ApplicationModel applicationModel, URL registryURL) {
- super(applicationModel, registryURL);
- }
-
- @Override
- public void doInitialize(URL registryURL) {
- try {
- exchanger = PilotExchanger.initialize(registryURL);
- } catch (Throwable t) {
- logger.error(REGISTRY_ERROR_INITIALIZE_XDS, "", "", t.getMessage(), t);
- }
- }
-
- @Override
- public void doDestroy() {
- try {
- if (exchanger == null) {
- return;
- }
- exchanger.destroy();
- } catch (Throwable t) {
- logger.error(REGISTRY_ERROR_INITIALIZE_XDS, "", "", t.getMessage(), t);
- }
- }
-
- @Override
- public Set getServices() {
- return exchanger.getServices();
- }
-
- @Override
- public List getInstances(String serviceName) throws NullPointerException {
- Set endpoints = exchanger.getEndpoints(serviceName);
- return changedToInstances(serviceName, endpoints);
- }
-
- @Override
- public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener)
- throws NullPointerException, IllegalArgumentException {
- listener.getServiceNames()
- .forEach(serviceName -> exchanger.observeEndpoints(
- serviceName,
- (endpoints ->
- notifyListener(serviceName, listener, changedToInstances(serviceName, endpoints)))));
- }
-
- private List changedToInstances(String serviceName, Collection endpoints) {
- List instances = new LinkedList<>();
- endpoints.forEach(endpoint -> {
- try {
- DefaultServiceInstance serviceInstance = new DefaultServiceInstance(
- serviceName,
- endpoint.getAddress(),
- endpoint.getPortValue(),
- ScopeModelUtil.getApplicationModel(getUrl().getScopeModel()));
- // fill metadata by SelfHostMetaServiceDiscovery, will be fetched by RPC request
- serviceInstance.putExtendParam("clusterName", endpoint.getClusterName());
- fillServiceInstance(serviceInstance);
- instances.add(serviceInstance);
- } catch (Throwable t) {
- logger.error(
- REGISTRY_ERROR_PARSING_XDS,
- "",
- "",
- "Error occurred when parsing endpoints. Endpoints List:" + endpoints,
- t);
- }
- });
- instances.sort(Comparator.comparingInt(ServiceInstance::hashCode));
- return instances;
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java
deleted file mode 100644
index 0f763c59c9..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * 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.xds;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.registry.client.AbstractServiceDiscoveryFactory;
-import org.apache.dubbo.registry.client.ServiceDiscovery;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_INITIALIZE_XDS;
-
-public class XdsServiceDiscoveryFactory extends AbstractServiceDiscoveryFactory {
-
- private static final ErrorTypeAwareLogger logger =
- LoggerFactory.getErrorTypeAwareLogger(XdsServiceDiscoveryFactory.class);
-
- @Override
- protected ServiceDiscovery createDiscovery(URL registryURL) {
- XdsServiceDiscovery xdsServiceDiscovery = new XdsServiceDiscovery(ApplicationModel.defaultModel(), registryURL);
- try {
- xdsServiceDiscovery.doInitialize(registryURL);
- } catch (Exception e) {
- logger.error(
- REGISTRY_ERROR_INITIALIZE_XDS,
- "",
- "",
- "Error occurred when initialize xDS service discovery impl.",
- e);
- }
- return xdsServiceDiscovery;
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java
deleted file mode 100644
index db54ea5628..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioCitadelCertificateSigner.java
+++ /dev/null
@@ -1,294 +0,0 @@
-/*
- * 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.xds.istio;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.registry.xds.XdsCertificateSigner;
-import org.apache.dubbo.rpc.RpcException;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.StringWriter;
-import java.nio.charset.StandardCharsets;
-import java.security.InvalidAlgorithmParameterException;
-import java.security.KeyPair;
-import java.security.KeyPairGenerator;
-import java.security.NoSuchAlgorithmException;
-import java.security.PrivateKey;
-import java.security.PublicKey;
-import java.security.SecureRandom;
-import java.security.spec.ECGenParameterSpec;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import io.grpc.ManagedChannel;
-import io.grpc.Metadata;
-import io.grpc.netty.shaded.io.grpc.netty.GrpcSslContexts;
-import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
-import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import io.grpc.stub.MetadataUtils;
-import io.grpc.stub.StreamObserver;
-import istio.v1.auth.IstioCertificateRequest;
-import istio.v1.auth.IstioCertificateResponse;
-import istio.v1.auth.IstioCertificateServiceGrpc;
-import org.bouncycastle.asn1.pkcs.PKCSObjectIdentifiers;
-import org.bouncycastle.asn1.x500.X500Name;
-import org.bouncycastle.asn1.x509.Extension;
-import org.bouncycastle.asn1.x509.ExtensionsGenerator;
-import org.bouncycastle.asn1.x509.GeneralName;
-import org.bouncycastle.asn1.x509.GeneralNames;
-import org.bouncycastle.openssl.jcajce.JcaPEMWriter;
-import org.bouncycastle.operator.ContentSigner;
-import org.bouncycastle.operator.OperatorCreationException;
-import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;
-import org.bouncycastle.pkcs.PKCS10CertificationRequest;
-import org.bouncycastle.pkcs.jcajce.JcaPKCS10CertificationRequestBuilder;
-import org.bouncycastle.util.io.pem.PemObject;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_GENERATE_CERT_ISTIO;
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_FAILED_GENERATE_KEY_ISTIO;
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_RECEIVE_ERROR_MSG_ISTIO;
-
-public class IstioCitadelCertificateSigner implements XdsCertificateSigner {
-
- private static final ErrorTypeAwareLogger logger =
- LoggerFactory.getErrorTypeAwareLogger(IstioCitadelCertificateSigner.class);
-
- private final org.apache.dubbo.registry.xds.istio.IstioEnv istioEnv;
-
- private CertPair certPair;
-
- public IstioCitadelCertificateSigner() {
- // watch cert, Refresh every 30s
- ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(1);
- scheduledThreadPool.scheduleAtFixedRate(new GenerateCertTask(), 0, 30, TimeUnit.SECONDS);
- istioEnv = IstioEnv.getInstance();
- }
-
- @Override
- public CertPair GenerateCert(URL url) {
-
- if (certPair != null && !certPair.isExpire()) {
- return certPair;
- }
- return doGenerateCert();
- }
-
- private class GenerateCertTask implements Runnable {
- @Override
- public void run() {
- doGenerateCert();
- }
- }
-
- private CertPair doGenerateCert() {
- synchronized (this) {
- if (certPair == null || certPair.isExpire()) {
- try {
- certPair = createCert();
- } catch (IOException e) {
- logger.error(REGISTRY_FAILED_GENERATE_CERT_ISTIO, "", "", "Generate Cert from Istio failed.", e);
- throw new RpcException("Generate Cert from Istio failed.", e);
- }
- }
- }
- return certPair;
- }
-
- public CertPair createCert() throws IOException {
- PublicKey publicKey = null;
- PrivateKey privateKey = null;
- ContentSigner signer = null;
-
- if (istioEnv.isECCFirst()) {
- try {
- ECGenParameterSpec ecSpec = new ECGenParameterSpec("secp256r1");
- KeyPairGenerator g = KeyPairGenerator.getInstance("EC");
- g.initialize(ecSpec, new SecureRandom());
- KeyPair keypair = g.generateKeyPair();
- publicKey = keypair.getPublic();
- privateKey = keypair.getPrivate();
- signer = new JcaContentSignerBuilder("SHA256withECDSA").build(keypair.getPrivate());
- } catch (NoSuchAlgorithmException | InvalidAlgorithmParameterException | OperatorCreationException e) {
- logger.error(
- REGISTRY_FAILED_GENERATE_KEY_ISTIO,
- "",
- "",
- "Generate Key with secp256r1 algorithm failed. Please check if your system support. "
- + "Will attempt to generate with RSA2048.",
- e);
- }
- }
-
- if (publicKey == null) {
- try {
- KeyPairGenerator kpGenerator = KeyPairGenerator.getInstance("RSA");
- kpGenerator.initialize(istioEnv.getRasKeySize());
- KeyPair keypair = kpGenerator.generateKeyPair();
- publicKey = keypair.getPublic();
- privateKey = keypair.getPrivate();
- signer = new JcaContentSignerBuilder("SHA256WithRSA").build(keypair.getPrivate());
- } catch (NoSuchAlgorithmException | OperatorCreationException e) {
- logger.error(
- REGISTRY_FAILED_GENERATE_KEY_ISTIO,
- "",
- "",
- "Generate Key with SHA256WithRSA algorithm failed. Please check if your system support.",
- e);
- throw new RpcException(e);
- }
- }
-
- String csr = generateCsr(publicKey, signer);
- String caCert = istioEnv.getCaCert();
- ManagedChannel channel;
- if (StringUtils.isNotEmpty(caCert)) {
- channel = NettyChannelBuilder.forTarget(istioEnv.getCaAddr())
- .sslContext(GrpcSslContexts.forClient()
- .trustManager(new ByteArrayInputStream(caCert.getBytes(StandardCharsets.UTF_8)))
- .build())
- .build();
- } else {
- channel = NettyChannelBuilder.forTarget(istioEnv.getCaAddr())
- .sslContext(GrpcSslContexts.forClient()
- .trustManager(InsecureTrustManagerFactory.INSTANCE)
- .build())
- .build();
- }
-
- Metadata header = new Metadata();
- Metadata.Key key = Metadata.Key.of("authorization", Metadata.ASCII_STRING_MARSHALLER);
- header.put(key, "Bearer " + istioEnv.getServiceAccount());
-
- key = Metadata.Key.of("ClusterID", Metadata.ASCII_STRING_MARSHALLER);
- header.put(key, istioEnv.getIstioMetaClusterId());
-
- IstioCertificateServiceGrpc.IstioCertificateServiceStub stub = IstioCertificateServiceGrpc.newStub(channel);
-
- stub = stub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(header));
-
- CountDownLatch countDownLatch = new CountDownLatch(1);
- StringBuffer publicKeyBuilder = new StringBuffer();
- AtomicBoolean failed = new AtomicBoolean(false);
- stub.createCertificate(
- generateRequest(csr), generateResponseObserver(countDownLatch, publicKeyBuilder, failed));
-
- long expireTime =
- System.currentTimeMillis() + (long) (istioEnv.getSecretTTL() * istioEnv.getSecretGracePeriodRatio());
-
- try {
- countDownLatch.await();
- } catch (InterruptedException e) {
- throw new RpcException("Generate Cert Failed. Wait for cert failed.", e);
- }
-
- if (failed.get()) {
- throw new RpcException("Generate Cert Failed. Send csr request failed. Please check log above.");
- }
-
- String privateKeyPem = generatePrivatePemKey(privateKey);
- CertPair certPair =
- new CertPair(privateKeyPem, publicKeyBuilder.toString(), System.currentTimeMillis(), expireTime);
-
- channel.shutdown();
- return certPair;
- }
-
- private IstioCertificateRequest generateRequest(String csr) {
- return IstioCertificateRequest.newBuilder()
- .setCsr(csr)
- .setValidityDuration(istioEnv.getSecretTTL())
- .build();
- }
-
- private StreamObserver generateResponseObserver(
- CountDownLatch countDownLatch, StringBuffer publicKeyBuilder, AtomicBoolean failed) {
- return new StreamObserver() {
- @Override
- public void onNext(IstioCertificateResponse istioCertificateResponse) {
- for (int i = 0; i < istioCertificateResponse.getCertChainCount(); i++) {
- publicKeyBuilder.append(
- istioCertificateResponse.getCertChainBytes(i).toStringUtf8());
- }
- if (logger.isDebugEnabled()) {
- logger.debug("Receive Cert chain from Istio Citadel. \n" + publicKeyBuilder);
- }
- countDownLatch.countDown();
- }
-
- @Override
- public void onError(Throwable throwable) {
- failed.set(true);
- logger.error(
- REGISTRY_RECEIVE_ERROR_MSG_ISTIO,
- "",
- "",
- "Receive error message from Istio Citadel grpc stub.",
- throwable);
- countDownLatch.countDown();
- }
-
- @Override
- public void onCompleted() {
- countDownLatch.countDown();
- }
- };
- }
-
- private String generatePrivatePemKey(PrivateKey privateKey) throws IOException {
- String key = generatePemKey("RSA PRIVATE KEY", privateKey.getEncoded());
- if (logger.isDebugEnabled()) {
- logger.debug("Generated Private Key. \n" + key);
- }
- return key;
- }
-
- private String generatePemKey(String type, byte[] content) throws IOException {
- PemObject pemObject = new PemObject(type, content);
- StringWriter str = new StringWriter();
- JcaPEMWriter jcaPEMWriter = new JcaPEMWriter(str);
- jcaPEMWriter.writeObject(pemObject);
- jcaPEMWriter.close();
- str.close();
- return str.toString();
- }
-
- private String generateCsr(PublicKey publicKey, ContentSigner signer) throws IOException {
- GeneralNames subjectAltNames = new GeneralNames(new GeneralName[] {new GeneralName(6, istioEnv.getCsrHost())});
-
- ExtensionsGenerator extGen = new ExtensionsGenerator();
- extGen.addExtension(Extension.subjectAlternativeName, true, subjectAltNames);
-
- PKCS10CertificationRequest request = new JcaPKCS10CertificationRequestBuilder(
- new X500Name("O=" + istioEnv.getTrustDomain()), publicKey)
- .addAttribute(PKCSObjectIdentifiers.pkcs_9_at_extensionRequest, extGen.generate())
- .build(signer);
-
- String csr = generatePemKey("CERTIFICATE REQUEST", request.getEncoded());
-
- if (logger.isDebugEnabled()) {
- logger.debug("CSR Request to Istio Citadel. \n" + csr);
- }
- return csr;
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java
deleted file mode 100644
index 018fa3a0fa..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioConstant.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*
- * 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.xds.istio;
-
-public class IstioConstant {
- /**
- * Address of the spiffe certificate provider. Defaults to discoveryAddress
- */
- public static final String CA_ADDR_KEY = "CA_ADDR";
-
- /**
- * CA and xDS services
- */
- public static final String DEFAULT_CA_ADDR = "istiod.istio-system.svc:15012";
-
- /**
- * The trust domain for spiffe certificates
- */
- public static final String TRUST_DOMAIN_KEY = "TRUST_DOMAIN";
-
- /**
- * The trust domain for spiffe certificates default value
- */
- public static final String DEFAULT_TRUST_DOMAIN = "cluster.local";
-
- public static final String WORKLOAD_NAMESPACE_KEY = "WORKLOAD_NAMESPACE";
-
- public static final String DEFAULT_WORKLOAD_NAMESPACE = "default";
-
- /**
- * k8s jwt token
- */
- public static final String KUBERNETES_SA_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token";
-
- public static final String KUBERNETES_CA_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt";
-
- public static final String ISTIO_SA_PATH = "/var/run/secrets/tokens/istio-token";
-
- public static final String ISTIO_CA_PATH = "/var/run/secrets/istio/root-cert.pem";
-
- public static final String KUBERNETES_NAMESPACE_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/namespace";
-
- public static final String RSA_KEY_SIZE_KEY = "RSA_KEY_SIZE";
-
- public static final String DEFAULT_RSA_KEY_SIZE = "2048";
-
- /**
- * The type of ECC signature algorithm to use when generating private keys
- */
- public static final String ECC_SIG_ALG_KEY = "ECC_SIGNATURE_ALGORITHM";
-
- public static final String DEFAULT_ECC_SIG_ALG = "ECDSA";
-
- /**
- * The cert lifetime requested by istio agent
- */
- public static final String SECRET_TTL_KEY = "SECRET_TTL";
-
- /**
- * The cert lifetime default value 24h0m0s
- */
- public static final String DEFAULT_SECRET_TTL = "86400"; // 24 * 60 * 60
-
- /**
- * The grace period ratio for the cert rotation
- */
- public static final String SECRET_GRACE_PERIOD_RATIO_KEY = "SECRET_GRACE_PERIOD_RATIO";
-
- /**
- * The grace period ratio for the cert rotation, by default 0.5
- */
- public static final String DEFAULT_SECRET_GRACE_PERIOD_RATIO = "0.5";
-
- public static final String ISTIO_META_CLUSTER_ID_KEY = "ISTIO_META_CLUSTER_ID";
-
- public static final String PILOT_CERT_PROVIDER_KEY = "PILOT_CERT_PROVIDER";
-
- public static final String ISTIO_PILOT_CERT_PROVIDER = "istiod";
-
- public static final String DEFAULT_ISTIO_META_CLUSTER_ID = "Kubernetes";
-
- public static final String SPIFFE = "spiffe://";
-
- public static final String NS = "/ns/";
-
- public static final String SA = "/sa/";
-
- public static final String JWT_POLICY = "JWT_POLICY";
-
- public static final String DEFAULT_JWT_POLICY = "first-party-jwt";
-
- public static final String FIRST_PARTY_JWT = "first-party-jwt";
-
- public static final String THIRD_PARTY_JWT = "third-party-jwt";
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java
deleted file mode 100644
index e508e7cc29..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/istio/IstioEnv.java
+++ /dev/null
@@ -1,195 +0,0 @@
-/*
- * 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.xds.istio;
-
-import org.apache.dubbo.common.constants.LoggerCodeConstants;
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.registry.xds.XdsEnv;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.charset.StandardCharsets;
-import java.util.Optional;
-
-import org.apache.commons.io.FileUtils;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_READ_FILE_ISTIO;
-import static org.apache.dubbo.registry.xds.istio.IstioConstant.NS;
-import static org.apache.dubbo.registry.xds.istio.IstioConstant.SA;
-import static org.apache.dubbo.registry.xds.istio.IstioConstant.SPIFFE;
-
-public class IstioEnv implements XdsEnv {
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(IstioEnv.class);
-
- private static final IstioEnv INSTANCE = new IstioEnv();
-
- private String podName;
-
- private String caAddr;
-
- private String jwtPolicy;
-
- private String trustDomain;
-
- private String workloadNameSpace;
-
- private int rasKeySize;
-
- private String eccSigAlg;
-
- private int secretTTL;
-
- private float secretGracePeriodRatio;
-
- private String istioMetaClusterId;
-
- private String pilotCertProvider;
-
- private IstioEnv() {
- jwtPolicy =
- Optional.ofNullable(System.getenv(IstioConstant.JWT_POLICY)).orElse(IstioConstant.DEFAULT_JWT_POLICY);
- podName = Optional.ofNullable(System.getenv("POD_NAME")).orElse(System.getenv("HOSTNAME"));
- trustDomain = Optional.ofNullable(System.getenv(IstioConstant.TRUST_DOMAIN_KEY))
- .orElse(IstioConstant.DEFAULT_TRUST_DOMAIN);
- workloadNameSpace = Optional.ofNullable(System.getenv(IstioConstant.WORKLOAD_NAMESPACE_KEY))
- .orElseGet(() -> {
- File namespaceFile = new File(IstioConstant.KUBERNETES_NAMESPACE_PATH);
- if (namespaceFile.canRead()) {
- try {
- return FileUtils.readFileToString(namespaceFile, StandardCharsets.UTF_8);
- } catch (IOException e) {
- logger.error(REGISTRY_ERROR_READ_FILE_ISTIO, "", "", "read namespace file error", e);
- }
- }
- return IstioConstant.DEFAULT_WORKLOAD_NAMESPACE;
- });
- caAddr = Optional.ofNullable(System.getenv(IstioConstant.CA_ADDR_KEY)).orElse(IstioConstant.DEFAULT_CA_ADDR);
- rasKeySize = Integer.parseInt(Optional.ofNullable(System.getenv(IstioConstant.RSA_KEY_SIZE_KEY))
- .orElse(IstioConstant.DEFAULT_RSA_KEY_SIZE));
- eccSigAlg = Optional.ofNullable(System.getenv(IstioConstant.ECC_SIG_ALG_KEY))
- .orElse(IstioConstant.DEFAULT_ECC_SIG_ALG);
- secretTTL = Integer.parseInt(Optional.ofNullable(System.getenv(IstioConstant.SECRET_TTL_KEY))
- .orElse(IstioConstant.DEFAULT_SECRET_TTL));
- secretGracePeriodRatio =
- Float.parseFloat(Optional.ofNullable(System.getenv(IstioConstant.SECRET_GRACE_PERIOD_RATIO_KEY))
- .orElse(IstioConstant.DEFAULT_SECRET_GRACE_PERIOD_RATIO));
- istioMetaClusterId = Optional.ofNullable(System.getenv(IstioConstant.ISTIO_META_CLUSTER_ID_KEY))
- .orElse(IstioConstant.DEFAULT_ISTIO_META_CLUSTER_ID);
- pilotCertProvider = Optional.ofNullable(System.getenv(IstioConstant.PILOT_CERT_PROVIDER_KEY))
- .orElse("");
-
- if (getServiceAccount() == null) {
- throw new UnsupportedOperationException("Unable to found kubernetes service account token file. "
- + "Please check if work in Kubernetes and mount service account token file correctly.");
- }
- }
-
- public static IstioEnv getInstance() {
- return INSTANCE;
- }
-
- public String getPodName() {
- return podName;
- }
-
- public String getCaAddr() {
- return caAddr;
- }
-
- public String getServiceAccount() {
- File saFile;
- switch (jwtPolicy) {
- case IstioConstant.FIRST_PARTY_JWT:
- saFile = new File(IstioConstant.KUBERNETES_SA_PATH);
- break;
- case IstioConstant.THIRD_PARTY_JWT:
- default:
- saFile = new File(IstioConstant.ISTIO_SA_PATH);
- }
- if (saFile.canRead()) {
- try {
- return FileUtils.readFileToString(saFile, StandardCharsets.UTF_8);
- } catch (IOException e) {
- logger.error(
- LoggerCodeConstants.REGISTRY_ISTIO_EXCEPTION,
- "File Read Failed",
- "",
- "Unable to read token file.",
- e);
- }
- }
-
- return null;
- }
-
- public String getCsrHost() {
- // spiffe:///ns//sa/
- return SPIFFE + trustDomain + NS + workloadNameSpace + SA + getServiceAccount();
- }
-
- public String getTrustDomain() {
- return trustDomain;
- }
-
- public String getWorkloadNameSpace() {
- return workloadNameSpace;
- }
-
- @Override
- public String getCluster() {
- return null;
- }
-
- public int getRasKeySize() {
- return rasKeySize;
- }
-
- public boolean isECCFirst() {
- return IstioConstant.DEFAULT_ECC_SIG_ALG.equals(eccSigAlg);
- }
-
- public int getSecretTTL() {
- return secretTTL;
- }
-
- public float getSecretGracePeriodRatio() {
- return secretGracePeriodRatio;
- }
-
- public String getIstioMetaClusterId() {
- return istioMetaClusterId;
- }
-
- public String getCaCert() {
- File caFile;
- if (IstioConstant.ISTIO_PILOT_CERT_PROVIDER.equals(pilotCertProvider)) {
- caFile = new File(IstioConstant.ISTIO_CA_PATH);
- } else {
- return null;
- }
- if (caFile.canRead()) {
- try {
- return FileUtils.readFileToString(caFile, StandardCharsets.UTF_8);
- } catch (IOException e) {
- logger.error(
- LoggerCodeConstants.REGISTRY_ISTIO_EXCEPTION, "File Read Failed", "", "read ca file error", e);
- }
- }
- return null;
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java
deleted file mode 100644
index 5e1734806d..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java
+++ /dev/null
@@ -1,140 +0,0 @@
-/*
- * 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.xds.util;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
-import org.apache.dubbo.common.logger.LoggerFactory;
-import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
-import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol;
-import org.apache.dubbo.registry.xds.util.protocol.DeltaResource;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import io.envoyproxy.envoy.config.core.v3.Node;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
-import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
-import io.grpc.stub.StreamObserver;
-
-import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS;
-
-public class AdsObserver {
- private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AdsObserver.class);
- private final ApplicationModel applicationModel;
- private final URL url;
- private final Node node;
- private volatile XdsChannel xdsChannel;
-
- private final Map listeners = new ConcurrentHashMap<>();
-
- protected StreamObserver requestObserver;
-
- private final Map observedResources = new ConcurrentHashMap<>();
-
- public AdsObserver(URL url, Node node) {
- this.url = url;
- this.node = node;
- this.xdsChannel = new XdsChannel(url);
- this.applicationModel = url.getOrDefaultApplicationModel();
- }
-
- public > void addListener(AbstractProtocol protocol) {
- listeners.put(protocol.getTypeUrl(), protocol);
- }
-
- public void request(DiscoveryRequest discoveryRequest) {
- if (requestObserver == null) {
- requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(this));
- }
- requestObserver.onNext(discoveryRequest);
- observedResources.put(discoveryRequest.getTypeUrl(), discoveryRequest);
- }
-
- private static class ResponseObserver implements StreamObserver {
- private AdsObserver adsObserver;
-
- public ResponseObserver(AdsObserver adsObserver) {
- this.adsObserver = adsObserver;
- }
-
- @Override
- public void onNext(DiscoveryResponse discoveryResponse) {
- XdsListener xdsListener = adsObserver.listeners.get(discoveryResponse.getTypeUrl());
- xdsListener.process(discoveryResponse);
- adsObserver.requestObserver.onNext(buildAck(discoveryResponse));
- }
-
- protected DiscoveryRequest buildAck(DiscoveryResponse response) {
- // for ACK
- return DiscoveryRequest.newBuilder()
- .setNode(adsObserver.node)
- .setTypeUrl(response.getTypeUrl())
- .setVersionInfo(response.getVersionInfo())
- .setResponseNonce(response.getNonce())
- .addAllResourceNames(adsObserver
- .observedResources
- .get(response.getTypeUrl())
- .getResourceNamesList())
- .build();
- }
-
- @Override
- public void onError(Throwable throwable) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "xDS Client received error message! detail:", throwable);
- adsObserver.triggerReConnectTask();
- }
-
- @Override
- public void onCompleted() {
- logger.info("xDS Client completed");
- adsObserver.triggerReConnectTask();
- }
- }
-
- private void triggerReConnectTask() {
- ScheduledExecutorService scheduledFuture = applicationModel
- .getFrameworkModel()
- .getBeanFactory()
- .getBean(FrameworkExecutorRepository.class)
- .getSharedScheduledExecutor();
- scheduledFuture.schedule(this::recover, 3, TimeUnit.SECONDS);
- }
-
- private void recover() {
- try {
- xdsChannel = new XdsChannel(url);
- if (xdsChannel.getChannel() != null) {
- requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(this));
- observedResources.values().forEach(requestObserver::onNext);
- return;
- } else {
- logger.error(
- REGISTRY_ERROR_REQUEST_XDS,
- "",
- "",
- "Recover failed for xDS connection. Will retry. Create channel failed.");
- }
- } catch (Exception e) {
- logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Recover failed for xDS connection. Will retry.", e);
- }
- triggerReConnectTask();
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/NodeBuilder.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/NodeBuilder.java
deleted file mode 100644
index eaa88f3904..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/NodeBuilder.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.xds.util;
-
-import org.apache.dubbo.common.utils.NetUtils;
-import org.apache.dubbo.registry.xds.istio.IstioEnv;
-
-import io.envoyproxy.envoy.config.core.v3.Node;
-
-public class NodeBuilder {
-
- private static final String SVC_CLUSTER_LOCAL = ".svc.cluster.local";
-
- public static Node build() {
- // String podName = System.getenv("metadata.name");
- // String podNamespace = System.getenv("metadata.namespace");
-
- String podName = IstioEnv.getInstance().getPodName();
- String podNamespace = IstioEnv.getInstance().getWorkloadNameSpace();
- String svcName = IstioEnv.getInstance().getIstioMetaClusterId();
-
- // id -> sidecar~ip~{POD_NAME}~{NAMESPACE_NAME}.svc.cluster.local
- // cluster -> {SVC_NAME}
- return Node.newBuilder()
- .setId("sidecar~" + NetUtils.getLocalHost() + "~" + podName + "~" + podNamespace + SVC_CLUSTER_LOCAL)
- .setCluster(svcName)
- .build();
- }
-}
diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
deleted file mode 100644
index 87fa81083a..0000000000
--- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java
+++ /dev/null
@@ -1,250 +0,0 @@
-/*
- * 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.xds.util;
-
-import org.apache.dubbo.common.URL;
-import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
-import org.apache.dubbo.common.utils.CollectionUtils;
-import org.apache.dubbo.common.utils.ConcurrentHashSet;
-import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol;
-import org.apache.dubbo.registry.xds.util.protocol.impl.EdsProtocol;
-import org.apache.dubbo.registry.xds.util.protocol.impl.LdsProtocol;
-import org.apache.dubbo.registry.xds.util.protocol.impl.RdsProtocol;
-import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint;
-import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult;
-import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult;
-import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult;
-import org.apache.dubbo.rpc.cluster.router.xds.RdsVirtualHostListener;
-import org.apache.dubbo.rpc.model.ApplicationModel;
-
-import java.util.Collections;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
-import java.util.stream.Collectors;
-
-public class PilotExchanger {
-
- protected final XdsChannel xdsChannel;
-
- protected final LdsProtocol ldsProtocol;
-
- protected final RdsProtocol rdsProtocol;
-
- protected final EdsProtocol edsProtocol;
-
- protected Map listenerResult;
-
- protected Map routeResult;
-
- private final AtomicBoolean isRdsObserve = new AtomicBoolean(false);
- private final Set domainObserveRequest = new ConcurrentHashSet();
-
- private final Map>>> domainObserveConsumer = new ConcurrentHashMap<>();
-
- private final Map> rdsObserveConsumer = new ConcurrentHashMap<>();
-
- private static PilotExchanger GLOBAL_PILOT_EXCHANGER = null;
-
- private final ApplicationModel applicationModel;
-
- protected PilotExchanger(URL url) {
- xdsChannel = new XdsChannel(url);
- int pollingTimeout = url.getParameter("pollingTimeout", 10);
- this.applicationModel = url.getOrDefaultApplicationModel();
- AdsObserver adsObserver = new AdsObserver(url, NodeBuilder.build());
- this.ldsProtocol = new LdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
- this.rdsProtocol = new RdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
- this.edsProtocol = new EdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
-
- this.listenerResult = ldsProtocol.getListeners();
- this.routeResult = rdsProtocol.getResource(
- listenerResult.values().iterator().next().getRouteConfigNames());
- Set ldsResourcesName = new HashSet<>();
- ldsResourcesName.add(AbstractProtocol.emptyResourceName);
- // Observer RDS update
- if (CollectionUtils.isNotEmpty(listenerResult.values().iterator().next().getRouteConfigNames())) {
- createRouteObserve();
- isRdsObserve.set(true);
- }
- // Observe LDS updated
- ldsProtocol.observeResource(
- ldsResourcesName,
- (newListener) -> {
- // update local cache
- if (!newListener.equals(listenerResult)) {
- this.listenerResult = newListener;
- // update RDS observation
- if (isRdsObserve.get()) {
- createRouteObserve();
- }
- }
- },
- false);
- }
-
- private void createRouteObserve() {
- rdsProtocol.observeResource(
- listenerResult.values().iterator().next().getRouteConfigNames(),
- (newResult) -> {
- // check if observed domain update ( will update endpoint observation )
- List domainsToUpdate = new LinkedList<>();
- domainObserveConsumer.forEach((domain, consumer) -> {
- newResult.values().forEach(o -> {
- Set newRoute = o.searchDomain(domain);
- for (Map.Entry entry : routeResult.entrySet()) {
- if (!entry.getValue().searchDomain(domain).equals(newRoute)) {
- // routers in observed domain has been updated
- // Long domainRequest = domainObserveRequest.get(domain);
- // router list is empty when observeEndpoints() called and domainRequest has not
- // been created yet
- // create new observation
- domainsToUpdate.add(domain);
- // doObserveEndpoints(domain);
- }
- }
- });
- });
- routeResult = newResult;
- ExecutorService executorService = applicationModel
- .getFrameworkModel()
- .getBeanFactory()
- .getBean(FrameworkExecutorRepository.class)
- .getSharedExecutor();
- executorService.submit(() -> domainsToUpdate.forEach(this::doObserveEndpoints));
- },
- false);
- }
-
- public static PilotExchanger initialize(URL url) {
- synchronized (PilotExchanger.class) {
- if (GLOBAL_PILOT_EXCHANGER != null) {
- return GLOBAL_PILOT_EXCHANGER;
- }
- return (GLOBAL_PILOT_EXCHANGER = new PilotExchanger(url));
- }
- }
-
- public static PilotExchanger getInstance() {
- synchronized (PilotExchanger.class) {
- return GLOBAL_PILOT_EXCHANGER;
- }
- }
-
- public static boolean isEnabled() {
- return GLOBAL_PILOT_EXCHANGER != null;
- }
-
- public void destroy() {
- xdsChannel.destroy();
- }
-
- public Set getServices() {
- Set domains = new HashSet<>();
- for (Map.Entry entry : routeResult.entrySet()) {
- domains.addAll(entry.getValue().getDomains());
- }
- return domains;
- }
-
- public Set getEndpoints(String domain) {
- Set endpoints = new HashSet<>();
- for (Map.Entry entry : routeResult.entrySet()) {
- Set cluster = entry.getValue().searchDomain(domain);
- if (CollectionUtils.isNotEmpty(cluster)) {
- Map endpointResultList = edsProtocol.getResource(cluster);
- endpointResultList.forEach((k, v) -> endpoints.addAll(v.getEndpoints()));
- } else {
- return Collections.emptySet();
- }
- }
- return endpoints;
- }
-
- public void observeEndpoints(String domain, Consumer> consumer) {
- // store Consumer
- domainObserveConsumer.compute(domain, (k, v) -> {
- if (v == null) {
- v = new ConcurrentHashSet<>();
- }
- // support multi-consumer
- v.add(consumer);
- return v;
- });
- if (!domainObserveRequest.contains(domain)) {
- doObserveEndpoints(domain);
- }
- }
-
- private void doObserveEndpoints(String domain) {
- for (Map.Entry entry : routeResult.entrySet()) {
- Set router = entry.getValue().searchDomain(domain);
- // if router is empty, do nothing
- // observation will be created when RDS updates
- if (CollectionUtils.isNotEmpty(router)) {
- edsProtocol.observeResource(
- router,
- (endpointResultMap) -> {
- Set endpoints = endpointResultMap.values().stream()
- .map(EndpointResult::getEndpoints)
- .flatMap(Set::stream)
- .collect(Collectors.toSet());
- for (Consumer> consumer : domainObserveConsumer.get(domain)) {
- consumer.accept(endpoints);
- }
- },
- false);
- domainObserveRequest.add(domain);
- }
- }
- }
-
- public void unObserveEndpoints(String domain, Consumer> consumer) {
- domainObserveConsumer.get(domain).remove(consumer);
- domainObserveRequest.remove(domain);
- }
-
- public void observeEds(Set clusterNames, Consumer