natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

【xds】init xds (#13747)

This commit is contained in:
hjyp 2024-03-22 17:14:48 +08:00 committed by GitHub
parent b338e20190
commit 9926a93223
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
59 changed files with 3601 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.artifacts
View File

@ -74,6 +74,7 @@ dubbo-registry-multicast
dubbo-registry-multiple
dubbo-registry-nacos
dubbo-registry-zookeeper
dubbo-xds
dubbo-remoting
dubbo-remoting-api
dubbo-remoting-http

5
dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java
View File

@ -21,6 +21,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.RpcException;
import java.util.List;
@ -98,4 +99,8 @@ public interface Directory<T> extends Node {
default boolean isNotificationReceived() {
return false;
}
default Protocol getProtocol() {
return null;
}
}

5
dubbo-config/dubbo-config-api/pom.xml
View File

@ -251,5 +251,10 @@
<artifactId>resteasy-jackson-provider</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-xds</artifactId>
<version>${project.parent.version}</version>
</dependency>
</dependencies>
</project>

11
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java
View File

@ -53,6 +53,7 @@ import org.apache.dubbo.rpc.protocol.injvm.InjvmProtocol;
import org.apache.dubbo.rpc.service.GenericService;
import org.apache.dubbo.rpc.stub.StubSuppliers;
import org.apache.dubbo.rpc.support.ProtocolUtils;
import org.apache.dubbo.xds.PilotExchanger;
import java.beans.Transient;
import java.util.ArrayList;
@ -655,6 +656,16 @@ public class ReferenceConfig<T> extends ReferenceConfigBase<T> {
private void createInvoker() {
if (urls.size() == 1) {
URL curUrl = urls.get(0);
if (curUrl.getParameter("registry", "null").startsWith("xds")) {
// TODO: The PilotExchanger requests xds resources asynchronously,
// and the xdsDirectory call filter chain may have an exception with invoker null,
// which needs to be synchronized later.
// move to deployer
curUrl = curUrl.addParameter("xds", true);
PilotExchanger.initialize(curUrl);
}
invoker = protocolSPI.refer(interfaceClass, curUrl);
// registry url, mesh-enable and unloadClusterRelated is true, not need Cluster.
if (!UrlUtils.isRegistry(curUrl) && !curUrl.getParameter(UNLOAD_CLUSTER_RELATED, false)) {

6
dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml
View File

@ -37,6 +37,12 @@
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-xds</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-registry-zookeeper</artifactId>

1
dubbo-distribution/dubbo-all-shaded/pom.xml
View File

@ -296,7 +296,6 @@
<scope>compile</scope>
<optional>true</optional>
</dependency>
<!-- remoting -->
<dependency>
<groupId>org.apache.dubbo</groupId>

12
dubbo-distribution/dubbo-all/pom.xml
View File

@ -334,6 +334,13 @@
<scope>compile</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-xds</artifactId>
<version>${project.version}</version>
<scope>compile</scope>
<optional>true</optional>
</dependency>
<!-- remoting -->
<dependency>
@ -552,6 +559,7 @@
<include>org.apache.dubbo:dubbo-registry-multiple</include>
<include>org.apache.dubbo:dubbo-registry-nacos</include>
<include>org.apache.dubbo:dubbo-registry-zookeeper</include>
<include>org.apache.dubbo:dubbo-xds</include>
<include>org.apache.dubbo:dubbo-remoting-api</include>
<include>org.apache.dubbo:dubbo-remoting-http</include>
<include>org.apache.dubbo:dubbo-remoting-http12</include>
@ -1020,6 +1028,10 @@
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/dubbo/internal/org.apache.dubbo.registry.integration.ServiceURLCustomizer</resource>
</transformer>
<transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
<resource>META-INF/dubbo/internal/org.apache.dubbo.xds.bootstrap.XdsCertificateSigner</resource>
</transformer>
</transformers>
<filters>
<filter>

5
dubbo-distribution/dubbo-bom/pom.xml
View File

@ -377,6 +377,11 @@
<artifactId>dubbo-registry-zookeeper</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-xds</artifactId>
<version>${project.version}</version>
</dependency>
<!-- remoting -->
<dependency>

7
dubbo-metadata/dubbo-metadata-api/pom.xml
View File

@ -37,13 +37,6 @@
<version>${project.parent.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-cluster</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metrics-api</artifactId>

11
dubbo-registry/dubbo-registry-api/pom.xml
View File

@ -38,12 +38,6 @@
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-cluster</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metadata-api</artifactId>
@ -102,5 +96,10 @@
<artifactId>log4j-slf4j-impl</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-cluster</artifactId>
<version>${project.parent.version}</version>
</dependency>
</dependencies>
</project>

4
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java
View File

@ -169,6 +169,10 @@ public abstract class DynamicDirectory<T> extends AbstractDirectory<T> implement
this.protocol = protocol;
}
public Protocol getProtocol() {
return this.protocol;
}
public void setRegistry(Registry registry) {
this.registry = registry;
}

5
dubbo-test/dubbo-dependencies-all/pom.xml
View File

@ -293,6 +293,11 @@
<artifactId>dubbo-registry-zookeeper</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-xds</artifactId>
<version>${project.version}</version>
</dependency>
<!-- remoting -->
<dependency>

146
dubbo-xds/pom.xml Normal file
View File

@ -0,0 +1,146 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-parent</artifactId>
<version>${revision}</version>
<relativePath>../pom.xml</relativePath>
</parent>
<artifactId>dubbo-xds</artifactId>
<packaging>jar</packaging>
<name>${project.artifactId}</name>
<description>The xds module of dubbo project</description>
<properties>
<skip_maven_deploy>false</skip_maven_deploy>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-injvm</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-remoting-netty4</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-test-check</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metrics-registry</artifactId>
<version>${project.parent.version}</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-metrics-default</artifactId>
<version>${project.parent.version}</version>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-tracing-integration-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
<artifactId>log4j-slf4j-impl</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-api</artifactId>
<version>1.61.0</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-protobuf</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-stub</artifactId>
</dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-netty-shaded</artifactId>
</dependency>
<dependency>
<groupId>io.envoyproxy.controlplane</groupId>
<artifactId>api</artifactId>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java-util</artifactId>
</dependency>
<dependency>
<groupId>org.bouncycastle</groupId>
<artifactId>bcpkix-jdk15on</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-cluster</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-registry-api</artifactId>
<version>${project.parent.version}</version>
</dependency>
<dependency>
<groupId>org.apache.dubbo</groupId>
<artifactId>dubbo-rpc-dubbo</artifactId>
<version>${project.parent.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

144
dubbo-xds/src/main/java/org/apache/dubbo/xds/AdsObserver.java Normal file
View File

@ -0,0 +1,144 @@
/*
* 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.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.common.threadpool.manager.FrameworkExecutorRepository;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.xds.protocol.AbstractProtocol;
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<String, XdsListener> listeners = new ConcurrentHashMap<>();
protected StreamObserver<DiscoveryRequest> requestObserver;
private final Map<String, DiscoveryRequest> 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 <T> void addListener(AbstractProtocol<T> 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<DiscoveryResponse> {
private AdsObserver adsObserver;
public ResponseObserver(AdsObserver adsObserver) {
this.adsObserver = adsObserver;
}
@Override
public void onNext(DiscoveryResponse discoveryResponse) {
System.out.println("Receive message from server");
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();
}
public void destroy() {
this.xdsChannel.destroy();
}
}

43
dubbo-xds/src/main/java/org/apache/dubbo/xds/NodeBuilder.java Normal file
View File

@ -0,0 +1,43 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.xds;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.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();
}
}

161
dubbo-xds/src/main/java/org/apache/dubbo/xds/PilotExchanger.java Normal file
View File

@ -0,0 +1,161 @@
/*
* 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.xds;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.ConcurrentHashSet;
import org.apache.dubbo.xds.directory.XdsDirectory;
import org.apache.dubbo.xds.protocol.impl.CdsProtocol;
import org.apache.dubbo.xds.protocol.impl.EdsProtocol;
import org.apache.dubbo.xds.protocol.impl.LdsProtocol;
import org.apache.dubbo.xds.protocol.impl.RdsProtocol;
import org.apache.dubbo.xds.resource.XdsCluster;
import org.apache.dubbo.xds.resource.XdsRouteConfiguration;
import org.apache.dubbo.xds.resource.XdsVirtualHost;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
public class PilotExchanger {
protected final AdsObserver adsObserver;
protected final LdsProtocol ldsProtocol;
protected final RdsProtocol rdsProtocol;
protected final EdsProtocol edsProtocol;
protected final CdsProtocol cdsProtocol;
private final Set<String> domainObserveRequest = new ConcurrentHashSet<String>();
private static PilotExchanger GLOBAL_PILOT_EXCHANGER = null;
private static final Map<String, XdsVirtualHost> xdsVirtualHostMap = new ConcurrentHashMap<>();
private static final Map<String, XdsCluster> xdsClusterMap = new ConcurrentHashMap<>();
private final Map<String, Set<XdsDirectory>> rdsListeners = new ConcurrentHashMap<>();
private final Map<String, Set<XdsDirectory>> cdsListeners = new ConcurrentHashMap<>();
protected PilotExchanger(URL url) {
int pollingTimeout = url.getParameter("pollingTimeout", 10);
adsObserver = new AdsObserver(url, NodeBuilder.build());
// rds resources callback
Consumer<List<XdsRouteConfiguration>> rdsCallback = (xdsRouteConfigurations) -> {
xdsRouteConfigurations.forEach(xdsRouteConfiguration -> {
xdsRouteConfiguration.getVirtualHosts().forEach((serviceName, xdsVirtualHost) -> {
this.xdsVirtualHostMap.put(serviceName, xdsVirtualHost);
// when resource update, notify subscribers
if (rdsListeners.containsKey(serviceName)) {
for (XdsDirectory listener : rdsListeners.get(serviceName)) {
listener.onRdsChange(serviceName, xdsVirtualHost);
}
}
});
});
};
// eds resources callback
Consumer<List<XdsCluster>> edsCallback = (xdsClusters) -> {
xdsClusters.forEach(xdsCluster -> {
this.xdsClusterMap.put(xdsCluster.getName(), xdsCluster);
if (cdsListeners.containsKey(xdsCluster.getName())) {
for (XdsDirectory listener : cdsListeners.get(xdsCluster.getName())) {
listener.onEdsChange(xdsCluster.getName(), xdsCluster);
}
}
});
};
this.rdsProtocol = new RdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout, rdsCallback);
this.edsProtocol = new EdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout, edsCallback);
this.ldsProtocol = new LdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
this.cdsProtocol = new CdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
// lds resources callback，listen to all rds resources in the callback function
Consumer<Set<String>> ldsCallback = rdsProtocol::subscribeResource;
ldsProtocol.setUpdateCallback(ldsCallback);
ldsProtocol.subscribeListeners();
// cds resources callback，listen to all cds resources in the callback function
Consumer<Set<String>> cdsCallback = edsProtocol::subscribeResource;
cdsProtocol.setUpdateCallback(cdsCallback);
cdsProtocol.subscribeClusters();
}
public static Map<String, XdsVirtualHost> getXdsVirtualHostMap() {
return xdsVirtualHostMap;
}
public static Map<String, XdsCluster> getXdsClusterMap() {
return xdsClusterMap;
}
public void subscribeRds(String applicationName, XdsDirectory listener) {
rdsListeners.computeIfAbsent(applicationName, key -> new ConcurrentHashSet<>());
rdsListeners.get(applicationName).add(listener);
if (xdsVirtualHostMap.containsKey(applicationName)) {
listener.onRdsChange(applicationName, this.xdsVirtualHostMap.get(applicationName));
}
}
public void unSubscribeRds(String applicationName, XdsDirectory listener) {
rdsListeners.get(applicationName).remove(listener);
}
public void subscribeCds(String clusterName, XdsDirectory listener) {
cdsListeners.computeIfAbsent(clusterName, key -> new ConcurrentHashSet<>());
cdsListeners.get(clusterName).add(listener);
if (xdsClusterMap.containsKey(clusterName)) {
listener.onEdsChange(clusterName, xdsClusterMap.get(clusterName));
}
}
public void unSubscribeCds(String clusterName, XdsDirectory listener) {
cdsListeners.get(clusterName).remove(listener);
}
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() {
this.adsObserver.destroy();
}
}

142
dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsChannel.java Normal file
View File

@ -0,0 +1,142 @@
/*
* 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.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.common.url.component.URLAddress;
import org.apache.dubbo.xds.bootstrap.Bootstrapper;
import org.apache.dubbo.xds.bootstrap.BootstrapperImpl;
import org.apache.dubbo.xds.bootstrap.XdsCertificateSigner;
import java.io.ByteArrayInputStream;
import java.nio.charset.StandardCharsets;
import io.envoyproxy.envoy.service.discovery.v3.AggregatedDiscoveryServiceGrpc;
import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryResponse;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import io.grpc.ManagedChannel;
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.channel.epoll.EpollDomainSocketChannel;
import io.grpc.netty.shaded.io.netty.channel.epoll.EpollEventLoopGroup;
import io.grpc.netty.shaded.io.netty.channel.unix.DomainSocketAddress;
import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
import io.grpc.netty.shaded.io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.grpc.stub.StreamObserver;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_CREATE_CHANNEL_XDS;
public class XdsChannel {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(XdsChannel.class);
private static final String USE_AGENT = "use-agent";
private URL url;
private static final String SECURE = "secure";
private static final String PLAINTEXT = "plaintext";
private final ManagedChannel channel;
public URL getUrl() {
return url;
}
public ManagedChannel getChannel() {
return channel;
}
public XdsChannel(URL url) {
ManagedChannel managedChannel = null;
this.url = url;
try {
if (!url.getParameter(USE_AGENT, false)) {
if (PLAINTEXT.equals(url.getParameter(SECURE))) {
managedChannel = NettyChannelBuilder.forAddress(url.getHost(), url.getPort())
.usePlaintext()
.build();
} else {
XdsCertificateSigner signer = url.getOrDefaultApplicationModel()
.getExtensionLoader(XdsCertificateSigner.class)
.getExtension(url.getParameter("signer", "istio"));
XdsCertificateSigner.CertPair certPair = signer.GenerateCert(url);
SslContext context = GrpcSslContexts.forClient()
.trustManager(InsecureTrustManagerFactory.INSTANCE)
.keyManager(
new ByteArrayInputStream(
certPair.getPublicKey().getBytes(StandardCharsets.UTF_8)),
new ByteArrayInputStream(
certPair.getPrivateKey().getBytes(StandardCharsets.UTF_8)))
.build();
managedChannel = NettyChannelBuilder.forAddress(url.getHost(), url.getPort())
.sslContext(context)
.build();
}
} else {
BootstrapperImpl bootstrapper = new BootstrapperImpl();
Bootstrapper.BootstrapInfo bootstrapInfo = bootstrapper.bootstrap();
URLAddress address =
URLAddress.parse(bootstrapInfo.servers().get(0).target(), null, false);
EpollEventLoopGroup elg = new EpollEventLoopGroup();
managedChannel = NettyChannelBuilder.forAddress(new DomainSocketAddress("/" + address.getPath()))
.eventLoopGroup(elg)
.channelType(EpollDomainSocketChannel.class)
.usePlaintext()
.build();
}
} catch (Exception e) {
logger.error(
REGISTRY_ERROR_CREATE_CHANNEL_XDS,
"",
"",
"Error occurred when creating gRPC channel to control panel.",
e);
}
channel = managedChannel;
}
public StreamObserver<DeltaDiscoveryRequest> observeDeltaDiscoveryRequest(
StreamObserver<DeltaDiscoveryResponse> observer) {
return AggregatedDiscoveryServiceGrpc.newStub(channel).deltaAggregatedResources(observer);
}
public StreamObserver<DiscoveryRequest> createDeltaDiscoveryRequest(StreamObserver<DiscoveryResponse> observer) {
return AggregatedDiscoveryServiceGrpc.newStub(channel).streamAggregatedResources(observer);
}
public StreamObserver<io.envoyproxy.envoy.api.v2.DeltaDiscoveryRequest> observeDeltaDiscoveryRequestV2(
StreamObserver<io.envoyproxy.envoy.api.v2.DeltaDiscoveryResponse> observer) {
return io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.newStub(channel)
.deltaAggregatedResources(observer);
}
public StreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryRequest> createDeltaDiscoveryRequestV2(
StreamObserver<io.envoyproxy.envoy.api.v2.DiscoveryResponse> observer) {
return io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.newStub(channel)
.streamAggregatedResources(observer);
}
public void destroy() {
channel.shutdown();
}
}

28
dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsInitializationException.java Normal file
View File

@ -0,0 +1,28 @@
/*
* 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.xds;
public final class XdsInitializationException extends Exception {
public XdsInitializationException(String message) {
super(message);
}
public XdsInitializationException(String message, Throwable cause) {
super(message, cause);
}
}

23
dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsListener.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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.xds;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
public interface XdsListener {
void process(DiscoveryResponse discoveryResponse);
}

131
dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapInfoImpl.java Normal file
View File

@ -0,0 +1,131 @@
/*
* 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.xds.bootstrap;
import javax.annotation.Nullable;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import io.envoyproxy.envoy.config.core.v3.Node;
public final class BootstrapInfoImpl extends Bootstrapper.BootstrapInfo {
private final List<Bootstrapper.ServerInfo> servers;
private final String serverListenerResourceNameTemplate;
private final Map<String, Bootstrapper.CertificateProviderInfo> certProviders;
private final Node node;
BootstrapInfoImpl(
List<Bootstrapper.ServerInfo> servers,
String serverListenerResourceNameTemplate,
Map<String, Bootstrapper.CertificateProviderInfo> certProviders,
Node node) {
this.servers = servers;
this.serverListenerResourceNameTemplate = serverListenerResourceNameTemplate;
this.certProviders = certProviders;
this.node = node;
}
@Override
public List<Bootstrapper.ServerInfo> servers() {
return servers;
}
public Map<String, Bootstrapper.CertificateProviderInfo> certProviders() {
return certProviders;
}
@Override
public Node node() {
return node;
}
@Override
public String serverListenerResourceNameTemplate() {
return serverListenerResourceNameTemplate;
}
@Override
public String toString() {
return "BootstrapInfo{"
+ "servers=" + servers + ", "
+ "serverListenerResourceNameTemplate=" + serverListenerResourceNameTemplate + ", "
+ "node=" + node + ", "
+ "}";
}
public static final class Builder extends Bootstrapper.BootstrapInfo.Builder {
private List<Bootstrapper.ServerInfo> servers;
private Node node;
private Map<String, Bootstrapper.CertificateProviderInfo> certProviders;
private String serverListenerResourceNameTemplate;
Builder() {}
@Override
Bootstrapper.BootstrapInfo.Builder servers(List<Bootstrapper.ServerInfo> servers) {
this.servers = new LinkedList<>(servers);
return this;
}
@Override
Bootstrapper.BootstrapInfo.Builder node(Node node) {
if (node == null) {
throw new NullPointerException("Null node");
}
this.node = node;
return this;
}
@Override
Bootstrapper.BootstrapInfo.Builder certProviders(
@Nullable Map<String, Bootstrapper.CertificateProviderInfo> certProviders) {
this.certProviders = certProviders;
return this;
}
@Override
Bootstrapper.BootstrapInfo.Builder serverListenerResourceNameTemplate(
@Nullable String serverListenerResourceNameTemplate) {
this.serverListenerResourceNameTemplate = serverListenerResourceNameTemplate;
return this;
}
@Override
Bootstrapper.BootstrapInfo build() {
if (this.servers == null || this.node == null) {
StringBuilder missing = new StringBuilder();
if (this.servers == null) {
missing.append(" servers");
}
if (this.node == null) {
missing.append(" node");
}
throw new IllegalStateException("Missing required properties:" + missing);
}
return new BootstrapInfoImpl(
this.servers, this.serverListenerResourceNameTemplate, this.certProviders, this.node);
}
}
}

75
dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/Bootstrapper.java Normal file
View File

@ -0,0 +1,75 @@
/*
* 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.xds.bootstrap;
import org.apache.dubbo.xds.XdsInitializationException;
import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.grpc.ChannelCredentials;
public abstract class Bootstrapper {
public abstract BootstrapInfo bootstrap() throws XdsInitializationException;
BootstrapInfo bootstrap(Map<String, ?> rawData) throws XdsInitializationException {
throw new UnsupportedOperationException();
}
public abstract static class ServerInfo {
public abstract String target();
abstract ChannelCredentials channelCredentials();
abstract boolean useProtocolV3();
abstract boolean ignoreResourceDeletion();
}
public abstract static class CertificateProviderInfo {
public abstract String pluginName();
public abstract Map<String, ?> config();
}
public abstract static class BootstrapInfo {
public abstract List<ServerInfo> servers();
public abstract Map<String, CertificateProviderInfo> certProviders();
public abstract Node node();
public abstract String serverListenerResourceNameTemplate();
abstract static class Builder {
abstract Builder servers(List<ServerInfo> servers);
abstract Builder node(Node node);
abstract Builder certProviders(@Nullable Map<String, CertificateProviderInfo> certProviders);
abstract Builder serverListenerResourceNameTemplate(@Nullable String serverListenerResourceNameTemplate);
abstract BootstrapInfo build();
}
}
}

179
dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapperImpl.java Normal file
View File

@ -0,0 +1,179 @@
/*
* 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.xds.bootstrap;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.xds.XdsInitializationException;
import javax.annotation.Nullable;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.grpc.ChannelCredentials;
import io.grpc.internal.JsonParser;
import io.grpc.internal.JsonUtil;
public class BootstrapperImpl extends Bootstrapper {
static final String BOOTSTRAP_PATH_SYS_ENV_VAR = "GRPC_XDS_BOOTSTRAP";
static String bootstrapPathFromEnvVar = System.getenv(BOOTSTRAP_PATH_SYS_ENV_VAR);
private static final Logger logger = LoggerFactory.getLogger(BootstrapperImpl.class);
private FileReader reader = LocalFileReader.INSTANCE;
private static final String SERVER_FEATURE_XDS_V3 = "xds_v3";
private static final String SERVER_FEATURE_IGNORE_RESOURCE_DELETION = "ignore_resource_deletion";
public BootstrapInfo bootstrap() throws XdsInitializationException {
String filePath = bootstrapPathFromEnvVar;
String fileContent = null;
if (filePath != null) {
try {
fileContent = reader.readFile(filePath);
} catch (IOException e) {
throw new XdsInitializationException("Fail to read bootstrap file", e);
}
}
if (fileContent == null) throw new XdsInitializationException("Cannot find bootstrap configuration");
Map<String, ?> rawBootstrap;
try {
rawBootstrap = (Map<String, ?>) JsonParser.parse(fileContent);
} catch (IOException e) {
throw new XdsInitializationException("Failed to parse JSON", e);
}
return bootstrap(rawBootstrap);
}
@Override
BootstrapInfo bootstrap(Map<String, ?> rawData) throws XdsInitializationException {
BootstrapInfo.Builder builder = new BootstrapInfoImpl.Builder();
List<?> rawServerConfigs = JsonUtil.getList(rawData, "xds_servers");
if (rawServerConfigs == null) {
throw new XdsInitializationException("Invalid bootstrap: 'xds_servers' does not exist.");
}
List<ServerInfo> servers = parseServerInfos(rawServerConfigs);
builder.servers(servers);
Node.Builder nodeBuilder = Node.newBuilder();
Map<String, ?> rawNode = JsonUtil.getObject(rawData, "node");
if (rawNode != null) {
String id = JsonUtil.getString(rawNode, "id");
if (id != null) {
nodeBuilder.setId(id);
}
String cluster = JsonUtil.getString(rawNode, "cluster");
if (cluster != null) {
nodeBuilder.setCluster(cluster);
}
Map<String, ?> metadata = JsonUtil.getObject(rawNode, "metadata");
Map<String, ?> rawLocality = JsonUtil.getObject(rawNode, "locality");
}
builder.node(nodeBuilder.build());
Map<String, ?> certProvidersBlob = JsonUtil.getObject(rawData, "certificate_providers");
if (certProvidersBlob != null) {
Map<String, CertificateProviderInfo> certProviders = new HashMap<>(certProvidersBlob.size());
for (String name : certProvidersBlob.keySet()) {
Map<String, ?> valueMap = JsonUtil.getObject(certProvidersBlob, name);
String pluginName = checkForNull(JsonUtil.getString(valueMap, "plugin_name"), "plugin_name");
Map<String, ?> config = checkForNull(JsonUtil.getObject(valueMap, "config"), "config");
CertificateProviderInfoImpl certificateProviderInfo =
new CertificateProviderInfoImpl(pluginName, config);
certProviders.put(name, certificateProviderInfo);
}
builder.certProviders(certProviders);
}
return builder.build();
}
private static List<ServerInfo> parseServerInfos(List<?> rawServerConfigs) throws XdsInitializationException {
List<ServerInfo> servers = new LinkedList<>();
List<Map<String, ?>> serverConfigList = JsonUtil.checkObjectList(rawServerConfigs);
for (Map<String, ?> serverConfig : serverConfigList) {
String serverUri = JsonUtil.getString(serverConfig, "server_uri");
if (serverUri == null) {
throw new XdsInitializationException("Invalid bootstrap: missing 'server_uri'");
}
List<?> rawChannelCredsList = JsonUtil.getList(serverConfig, "channel_creds");
if (rawChannelCredsList == null || rawChannelCredsList.isEmpty()) {
throw new XdsInitializationException(
"Invalid bootstrap: server " + serverUri + " 'channel_creds' required");
}
ChannelCredentials channelCredentials =
parseChannelCredentials(JsonUtil.checkObjectList(rawChannelCredsList), serverUri);
// if (channelCredentials == null) {
// throw new XdsInitializationException(
// "Server " + serverUri + ": no supported channel credentials found");
// }
boolean useProtocolV3 = false;
boolean ignoreResourceDeletion = false;
List<String> serverFeatures = JsonUtil.getListOfStrings(serverConfig, "server_features");
if (serverFeatures != null) {
useProtocolV3 = serverFeatures.contains(SERVER_FEATURE_XDS_V3);
ignoreResourceDeletion = serverFeatures.contains(SERVER_FEATURE_IGNORE_RESOURCE_DELETION);
}
servers.add(new ServerInfoImpl(serverUri, channelCredentials, useProtocolV3, ignoreResourceDeletion));
}
return servers;
}
void setFileReader(FileReader reader) {
this.reader = reader;
}
/**
* Reads the content of the file with the given path in the file system.
*/
interface FileReader {
String readFile(String path) throws IOException;
}
private enum LocalFileReader implements FileReader {
INSTANCE;
@Override
public String readFile(String path) throws IOException {
return new String(Files.readAllBytes(Paths.get(path)), StandardCharsets.UTF_8);
}
}
private static <T> T checkForNull(T value, String fieldName) throws XdsInitializationException {
if (value == null) {
throw new XdsInitializationException("Invalid bootstrap: '" + fieldName + "' does not exist.");
}
return value;
}
@Nullable
private static ChannelCredentials parseChannelCredentials(List<Map<String, ?>> jsonList, String serverUri)
throws XdsInitializationException {
return null;
}
}

45
dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/CertificateProviderInfoImpl.java Normal file
View File

@ -0,0 +1,45 @@
/*
* 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.xds.bootstrap;
import java.util.Map;
final class CertificateProviderInfoImpl extends Bootstrapper.CertificateProviderInfo {
private final String pluginName;
private final Map<String, ?> config;
CertificateProviderInfoImpl(String pluginName, Map<String, ?> config) {
this.pluginName = pluginName;
this.config = config;
}
@Override
public String pluginName() {
return pluginName;
}
@Override
public Map<String, ?> config() {
return config;
}
@Override
public String toString() {
return "CertificateProviderInfo{" + "pluginName=" + pluginName + ", " + "config=" + config + "}";
}
}

71
dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/ServerInfoImpl.java Normal file
View File

@ -0,0 +1,71 @@
/*
* 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.xds.bootstrap;
import io.grpc.ChannelCredentials;
final class ServerInfoImpl extends Bootstrapper.ServerInfo {
private final String target;
private final ChannelCredentials channelCredentials;
private final boolean useProtocolV3;
private final boolean ignoreResourceDeletion;
ServerInfoImpl(
String target,
ChannelCredentials channelCredentials,
boolean useProtocolV3,
boolean ignoreResourceDeletion) {
this.target = target;
this.channelCredentials = channelCredentials;
this.useProtocolV3 = useProtocolV3;
this.ignoreResourceDeletion = ignoreResourceDeletion;
}
@Override
public String target() {
return target;
}
@Override
ChannelCredentials channelCredentials() {
return channelCredentials;
}
@Override
boolean useProtocolV3() {
return useProtocolV3;
}
@Override
boolean ignoreResourceDeletion() {
return ignoreResourceDeletion;
}
@Override
public String toString() {
return "ServerInfo{"
+ "target=" + target + ", "
+ "channelCredentials=" + channelCredentials + ", "
+ "useProtocolV3=" + useProtocolV3 + ", "
+ "ignoreResourceDeletion=" + ignoreResourceDeletion
+ "}";
}
}

58
dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/XdsCertificateSigner.java Normal file
View File

@ -0,0 +1,58 @@
/*
* 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.xds.bootstrap;
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;
}
}
}

34
dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsCluster.java Normal file
View File

@ -0,0 +1,34 @@
/*
* 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.xds.cluster;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
import org.apache.dubbo.rpc.cluster.support.wrapper.AbstractCluster;
import org.apache.dubbo.xds.directory.XdsDirectory;
public class XdsCluster extends AbstractCluster {
public static final String NAME = "xds";
@Override
protected <T> AbstractClusterInvoker<T> doJoin(Directory<T> directory) throws RpcException {
XdsDirectory<T> xdsDirectory = new XdsDirectory<>(directory);
return new XdsClusterInvoker<>(xdsDirectory);
}
}

60
dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsClusterInvoker.java Normal file
View File

@ -0,0 +1,60 @@
/*
* 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.xds.cluster;
import org.apache.dubbo.common.Version;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.LoadBalance;
import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
import org.apache.dubbo.rpc.support.RpcUtils;
import java.util.List;
public class XdsClusterInvoker<T> extends AbstractClusterInvoker<T> {
public XdsClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance)
throws RpcException {
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
try {
return invokeWithContext(invoker, invocation);
} catch (Throwable e) {
if (e instanceof RpcException && ((RpcException) e).isBiz()) { // biz exception.
throw (RpcException) e;
}
throw new RpcException(
e instanceof RpcException ? ((RpcException) e).getCode() : 0,
"Xds invoke providers " + invoker.getUrl() + " "
+ loadbalance.getClass().getSimpleName()
+ " for service " + getInterface().getName()
+ " method " + RpcUtils.getMethodName(invocation) + " on consumer "
+ NetUtils.getLocalHost()
+ " use dubbo version " + Version.getVersion()
+ ", but no luck to perform the invocation. Last error is: " + e.getMessage(),
e.getCause() != null ? e.getCause() : e);
}
}
}

181
dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java Normal file
View File

@ -0,0 +1,181 @@
/*
* 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.xds.directory;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.SingleRouterChain;
import org.apache.dubbo.rpc.cluster.directory.AbstractDirectory;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.xds.PilotExchanger;
import org.apache.dubbo.xds.resource.XdsCluster;
import org.apache.dubbo.xds.resource.XdsClusterWeight;
import org.apache.dubbo.xds.resource.XdsEndpoint;
import org.apache.dubbo.xds.resource.XdsRoute;
import org.apache.dubbo.xds.resource.XdsRouteAction;
import org.apache.dubbo.xds.resource.XdsVirtualHost;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
public class XdsDirectory<T> extends AbstractDirectory<T> {
private final URL url;
private final Class<T> serviceType;
private final String[] applicationNames;
private final String protocolName;
PilotExchanger pilotExchanger = PilotExchanger.getInstance();
private Protocol protocol;
private final Map<String, XdsVirtualHost> xdsVirtualHostMap = new ConcurrentHashMap<>();
private final Map<String, XdsCluster<T>> xdsClusterMap = new ConcurrentHashMap<>();
public XdsDirectory(Directory<T> directory) {
super(directory.getConsumerUrl(), true);
this.serviceType = directory.getInterface();
this.url = directory.getConsumerUrl();
this.applicationNames = url.getParameter("provided-by").split(",");
this.protocolName = url.getParameter("protocol", "dubbo");
this.protocol = directory.getProtocol();
super.routerChain = directory.getRouterChain();
// subscribe resource
for (String applicationName : applicationNames) {
pilotExchanger.subscribeRds(applicationName, this);
}
}
public Map<String, XdsVirtualHost> getXdsVirtualHostMap() {
return xdsVirtualHostMap;
}
public Map<String, XdsCluster<T>> getXdsClusterMap() {
return xdsClusterMap;
}
public Protocol getProtocol() {
return protocol;
}
public void setProtocol(Protocol protocol) {
this.protocol = protocol;
}
@Override
public Class<T> getInterface() {
return serviceType;
}
public List<Invoker<T>> doList(
SingleRouterChain<T> singleRouterChain, BitList<Invoker<T>> invokers, Invocation invocation) {
List<Invoker<T>> result = singleRouterChain.route(this.getConsumerUrl(), invokers, invocation);
return (List) (result == null ? BitList.emptyList() : result);
}
@Override
public List<Invoker<T>> getAllInvokers() {
return super.getInvokers();
}
public void onRdsChange(String applicationName, XdsVirtualHost xdsVirtualHost) {
Set<String> oldCluster = getAllCluster();
xdsVirtualHostMap.put(applicationName, xdsVirtualHost);
Set<String> newCluster = getAllCluster();
changeClusterSubscribe(oldCluster, newCluster);
}
private Set<String> getAllCluster() {
if (CollectionUtils.isEmptyMap(xdsVirtualHostMap)) {
return new HashSet<>();
}
Set<String> clusters = new HashSet<>();
xdsVirtualHostMap.forEach((applicationName, xdsVirtualHost) -> {
for (XdsRoute xdsRoute : xdsVirtualHost.getRoutes()) {
XdsRouteAction action = xdsRoute.getRouteAction();
if (action.getCluster() != null) {
clusters.add(action.getCluster());
} else if (CollectionUtils.isNotEmpty(action.getClusterWeights())) {
for (XdsClusterWeight weightedCluster : action.getClusterWeights()) {
clusters.add(weightedCluster.getName());
}
}
}
});
return clusters;
}
private void changeClusterSubscribe(Set<String> oldCluster, Set<String> newCluster) {
Set<String> removeSubscribe = new HashSet<>(oldCluster);
Set<String> addSubscribe = new HashSet<>(newCluster);
removeSubscribe.removeAll(newCluster);
addSubscribe.removeAll(oldCluster);
// remove subscribe cluster
for (String cluster : removeSubscribe) {
pilotExchanger.unSubscribeCds(cluster, this);
xdsClusterMap.remove(cluster);
// TODO: delete invokers which belong unsubscribed cluster
}
// add subscribe cluster
for (String cluster : addSubscribe) {
pilotExchanger.subscribeCds(cluster, this);
}
}
public void onEdsChange(String clusterName, XdsCluster<T> xdsCluster) {
xdsClusterMap.put(clusterName, xdsCluster);
String lbPolicy = xdsCluster.getLbPolicy();
List<XdsEndpoint> xdsEndpoints = xdsCluster.getXdsEndpoints();
BitList<Invoker<T>> invokers = new BitList<>(Collections.emptyList());
xdsEndpoints.forEach(e -> {
String ip = e.getAddress();
int port = e.getPortValue();
URL url = new URL(this.protocolName, ip, port);
// set cluster name
url = url.addParameter("clusterID", clusterName);
// set load balance policy
url = url.addParameter("loadbalance", lbPolicy);
// cluster to invoker
Invoker<T> invoker = this.protocol.refer(this.serviceType, url);
invokers.add(invoker);
});
// TODO: Consider cases where some clients are not available
super.getInvokers().addAll(invokers);
// super.setInvokers(invokers);
xdsCluster.setInvokers(invokers);
}
@Override
public boolean isAvailable() {
return false;
}
}

109
dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioConstant.java Normal file
View File

@ -0,0 +1,109 @@
/*
* 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.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 = "localhost: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 = "";
public static final String KUBERNETES_CA_PATH = "E:/k8s/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";
}

194
dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioEnv.java Normal file
View File

@ -0,0 +1,194 @@
/*
* 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.xds.istio;
import org.apache.dubbo.common.constants.LoggerCodeConstants;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
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.xds.istio.IstioConstant.NS;
import static org.apache.dubbo.xds.istio.IstioConstant.SA;
import static org.apache.dubbo.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);
}
}
// TODO：Subsequent implementation in the security section
return "null";
}
public String getCsrHost() {
// spiffe://<trust_domain>/ns/<namespace>/sa/<service_account>
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;
}
}

22
dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/XdsEnv.java Normal file
View File

@ -0,0 +1,22 @@
/*
* 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.xds.istio;
public interface XdsEnv {
String getCluster();
}

212
dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/AbstractProtocol.java Normal file
View File

@ -0,0 +1,212 @@
/*
* 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.xds.protocol;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.xds.AdsObserver;
import org.apache.dubbo.xds.XdsListener;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
public abstract class AbstractProtocol<T> implements XdsProtocol, XdsListener {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class);
protected AdsObserver adsObserver;
protected final Node node;
private final int checkInterval;
protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
protected final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
protected final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
protected Set<String> observeResourcesName;
public static final String emptyResourceName = "emptyResourcesName";
private final ReentrantLock resourceLock = new ReentrantLock();
protected Map<Set<String>, List<Consumer<Map<String, T>>>> consumerObserveMap = new ConcurrentHashMap<>();
public Map<Set<String>, List<Consumer<Map<String, T>>>> getConsumerObserveMap() {
return consumerObserveMap;
}
protected Map<String, T> resourcesMap = new ConcurrentHashMap<>();
public AbstractProtocol(AdsObserver adsObserver, Node node, int checkInterval) {
this.adsObserver = adsObserver;
this.node = node;
this.checkInterval = checkInterval;
adsObserver.addListener(this);
}
/**
* Abstract method to obtain Type-URL from sub-class
*
* @return Type-URL of xDS
*/
public abstract String getTypeUrl();
public boolean isCacheExistResource(Set<String> resourceNames) {
for (String resourceName : resourceNames) {
if ("".equals(resourceName)) {
continue;
}
if (!resourcesMap.containsKey(resourceName)) {
return false;
}
}
return true;
}
public T getCacheResource(String resourceName) {
if (resourceName == null || resourceName.length() == 0) {
return null;
}
return resourcesMap.get(resourceName);
}
@Override
public void subscribeResource(Set<String> resourceNames) {
resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
if (!resourceNames.isEmpty() && isCacheExistResource(resourceNames)) {
getResourceFromCache(resourceNames);
} else {
getResourceFromRemote(resourceNames);
}
}
private Map<String, T> getResourceFromCache(Set<String> resourceNames) {
return resourceNames.stream()
.filter(o -> !StringUtils.isEmpty(o))
.collect(Collectors.toMap(k -> k, this::getCacheResource));
}
public Map<String, T> getResourceFromRemote(Set<String> resourceNames) {
try {
resourceLock.lock();
CompletableFuture<Map<String, T>> future = new CompletableFuture<>();
observeResourcesName = resourceNames;
Set<String> consumerObserveResourceNames = new HashSet<>();
if (resourceNames.isEmpty()) {
consumerObserveResourceNames.add(emptyResourceName);
} else {
consumerObserveResourceNames = resourceNames;
}
Consumer<Map<String, T>> futureConsumer = future::complete;
try {
writeLock.lock();
ConcurrentHashMapUtils.computeIfAbsent(
(ConcurrentHashMap<Set<String>, List<Consumer<Map<String, T>>>>) consumerObserveMap,
consumerObserveResourceNames,
key -> new ArrayList<>())
.add(futureConsumer);
} finally {
writeLock.unlock();
}
Set<String> resourceNamesToObserve = new HashSet<>(resourceNames);
resourceNamesToObserve.addAll(resourcesMap.keySet());
adsObserver.request(buildDiscoveryRequest(resourceNamesToObserve));
logger.info("Send xDS Observe request to remote. Resource count: " + resourceNamesToObserve.size()
+ ". Resource Type: " + getTypeUrl());
} finally {
resourceLock.unlock();
}
return Collections.emptyMap();
}
protected DiscoveryRequest buildDiscoveryRequest(Set<String> resourceNames) {
return DiscoveryRequest.newBuilder()
.setNode(node)
.setTypeUrl(getTypeUrl())
.addAllResourceNames(resourceNames)
.build();
}
protected abstract Map<String, T> decodeDiscoveryResponse(DiscoveryResponse response);
@Override
public final void process(DiscoveryResponse discoveryResponse) {
Map<String, T> newResult = decodeDiscoveryResponse(discoveryResponse);
Map<String, T> oldResource = resourcesMap;
// discoveryResponseListener(oldResource, newResult);
resourcesMap = newResult;
}
private void discoveryResponseListener(Map<String, T> oldResult, Map<String, T> newResult) {
Set<String> changedResourceNames = new HashSet<>();
oldResult.forEach((key, origin) -> {
if (!Objects.equals(origin, newResult.get(key))) {
changedResourceNames.add(key);
}
});
newResult.forEach((key, origin) -> {
if (!Objects.equals(origin, oldResult.get(key))) {
changedResourceNames.add(key);
}
});
if (changedResourceNames.isEmpty()) {
return;
}
logger.info("Receive resource update notification from xds server. Change resource count: "
+ changedResourceNames.stream() + ". Type: " + getTypeUrl());
// call once for full data
try {
readLock.lock();
for (Map.Entry<Set<String>, List<Consumer<Map<String, T>>>> entry : consumerObserveMap.entrySet()) {
if (entry.getKey().stream().noneMatch(changedResourceNames::contains)) {
// none update
continue;
}
Map<String, T> dsResultMap =
entry.getKey().stream().collect(Collectors.toMap(k -> k, v -> newResult.get(v)));
entry.getValue().forEach(o -> o.accept(dsResultMap));
}
} finally {
readLock.unlock();
}
}
}

39
dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/XdsProtocol.java Normal file
View File

@ -0,0 +1,39 @@
/*
* 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.xds.protocol;
import java.util.Set;
public interface XdsProtocol {
/**
* Gets all resources by the specified resource name.
* For LDS, the {@param resourceNames} is ignored
*
* @param resourceNames specified resource name
* @return resources, null if request failed
*/
void subscribeResource(Set<String> resourceNames);
/**
* Add a observer resource with {@link Consumer}
*
* @param resourceNames specified resource name
* @param consumer resource notifier, will be called when resource updated
* @return requestId, used when resourceNames update with {@link XdsProtocol#updateObserve(long, Set)}
*/
// void observeResource(Set<String> resourceNames, Consumer<Map<String, T>> consumer, boolean isReConnect);
}

98
dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/CdsProtocol.java Normal file
View File

@ -0,0 +1,98 @@
/*
* 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.xds.protocol.impl;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.xds.AdsObserver;
import org.apache.dubbo.xds.protocol.AbstractProtocol;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import io.envoyproxy.envoy.config.cluster.v3.Cluster;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS;
public class CdsProtocol extends AbstractProtocol<String> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(CdsProtocol.class);
public void setUpdateCallback(Consumer<Set<String>> updateCallback) {
this.updateCallback = updateCallback;
}
private Consumer<Set<String>> updateCallback;
public CdsProtocol(AdsObserver adsObserver, Node node, int checkInterval) {
super(adsObserver, node, checkInterval);
}
@Override
public String getTypeUrl() {
return "type.googleapis.com/envoy.config.cluster.v3.Cluster";
}
public void subscribeClusters() {
subscribeResource(null);
}
@Override
protected Map<String, String> decodeDiscoveryResponse(DiscoveryResponse response) {
if (getTypeUrl().equals(response.getTypeUrl())) {
Set<String> set = response.getResourcesList().stream()
.map(CdsProtocol::unpackCluster)
.filter(Objects::nonNull)
.map(Cluster::getName)
.collect(Collectors.toSet());
updateCallback.accept(set);
// Map<String, ListenerResult> listenerDecodeResult = new ConcurrentHashMap<>();
// listenerDecodeResult.put(emptyResourceName, new ListenerResult(set));
// return listenerDecodeResult;
}
return new HashMap<>();
}
private static Cluster unpackCluster(Any any) {
try {
return any.unpack(Cluster.class);
} catch (InvalidProtocolBufferException e) {
logger.error(REGISTRY_ERROR_RESPONSE_XDS, "", "", "Error occur when decode xDS response.", e);
return null;
}
}
private static HttpConnectionManager unpackHttpConnectionManager(Any any) {
try {
if (!any.is(HttpConnectionManager.class)) {
return null;
}
return any.unpack(HttpConnectionManager.class);
} catch (InvalidProtocolBufferException e) {
logger.error(REGISTRY_ERROR_RESPONSE_XDS, "", "", "Error occur when decode xDS response.", e);
return null;
}
}
}

131
dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/EdsProtocol.java Normal file
View File

@ -0,0 +1,131 @@
/*
* 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.xds.protocol.impl;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.xds.AdsObserver;
import org.apache.dubbo.xds.protocol.AbstractProtocol;
import org.apache.dubbo.xds.resource.XdsCluster;
import org.apache.dubbo.xds.resource.XdsEndpoint;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import io.envoyproxy.envoy.config.cluster.v3.Cluster;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment;
import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS;
public class EdsProtocol extends AbstractProtocol<String> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(EdsProtocol.class);
public void setUpdateCallback(Consumer<List<XdsCluster>> updateCallback) {
this.updateCallback = updateCallback;
}
private Consumer<List<XdsCluster>> updateCallback;
public EdsProtocol(
AdsObserver adsObserver, Node node, int checkInterval, Consumer<List<XdsCluster>> updateCallback) {
super(adsObserver, node, checkInterval);
this.updateCallback = updateCallback;
}
@Override
public String getTypeUrl() {
return "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment";
}
@Override
protected Map<String, String> decodeDiscoveryResponse(DiscoveryResponse response) {
List<XdsCluster> clusters = parse(response);
updateCallback.accept(clusters);
// if (getTypeUrl().equals(response.getTypeUrl())) {
// return response.getResourcesList().stream()
// .map(EdsProtocol::unpackClusterLoadAssignment)
// .filter(Objects::nonNull)
// .collect(Collectors.toConcurrentMap(
// ClusterLoadAssignment::getClusterName, this::decodeResourceToEndpoint));
// }
return new HashMap<>();
}
public List<XdsCluster> parse(DiscoveryResponse response) {
if (!getTypeUrl().equals(response.getTypeUrl())) {
return null;
}
return response.getResourcesList().stream()
.map(EdsProtocol::unpackClusterLoadAssignment)
.filter(Objects::nonNull)
.map(this::parseCluster)
.collect(Collectors.toList());
}
public XdsCluster parseCluster(ClusterLoadAssignment cluster) {
XdsCluster xdsCluster = new XdsCluster();
xdsCluster.setName(cluster.getClusterName());
List<XdsEndpoint> xdsEndpoints = cluster.getEndpointsList().stream()
.flatMap(e -> e.getLbEndpointsList().stream())
.map(LbEndpoint::getEndpoint)
.map(this::parseEndpoint)
.collect(Collectors.toList());
xdsCluster.setXdsEndpoints(xdsEndpoints);
return xdsCluster;
}
public XdsEndpoint parseEndpoint(io.envoyproxy.envoy.config.endpoint.v3.Endpoint endpoint) {
XdsEndpoint xdsEndpoint = new XdsEndpoint();
xdsEndpoint.setAddress(endpoint.getAddress().getSocketAddress().getAddress());
xdsEndpoint.setPortValue(endpoint.getAddress().getSocketAddress().getPortValue());
return xdsEndpoint;
}
private static ClusterLoadAssignment unpackClusterLoadAssignment(Any any) {
try {
return any.unpack(ClusterLoadAssignment.class);
} catch (InvalidProtocolBufferException e) {
logger.error(REGISTRY_ERROR_RESPONSE_XDS, "", "", "Error occur when decode xDS response.", e);
return null;
}
}
private static Cluster unpackCluster(Any any) {
try {
return any.unpack(Cluster.class);
} catch (InvalidProtocolBufferException e) {
logger.error(REGISTRY_ERROR_RESPONSE_XDS, "", "", "Error occur when decode xDS response.", e);
return null;
}
}
}

112
dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/LdsProtocol.java Normal file
View File

@ -0,0 +1,112 @@
/*
* 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.xds.protocol.impl;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.xds.AdsObserver;
import org.apache.dubbo.xds.protocol.AbstractProtocol;
import org.apache.dubbo.xds.resource.XdsVirtualHost;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.config.listener.v3.Filter;
import io.envoyproxy.envoy.config.listener.v3.Listener;
import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager;
import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3.Rds;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS;
public class LdsProtocol extends AbstractProtocol<XdsVirtualHost> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(LdsProtocol.class);
public void setUpdateCallback(Consumer<Set<String>> updateCallback) {
this.updateCallback = updateCallback;
}
private Consumer<Set<String>> updateCallback;
public LdsProtocol(AdsObserver adsObserver, Node node, int checkInterval) {
super(adsObserver, node, checkInterval);
}
@Override
public String getTypeUrl() {
return "type.googleapis.com/envoy.config.listener.v3.Listener";
}
public void subscribeListeners() {
subscribeResource(null);
}
@Override
protected Map<String, XdsVirtualHost> decodeDiscoveryResponse(DiscoveryResponse response) {
if (getTypeUrl().equals(response.getTypeUrl())) {
Set<String> set = response.getResourcesList().stream()
.map(LdsProtocol::unpackListener)
.filter(Objects::nonNull)
.flatMap(e -> decodeResourceToListener(e).stream())
.collect(Collectors.toSet());
updateCallback.accept(set);
// Map<String, ListenerResult> listenerDecodeResult = new ConcurrentHashMap<>();
// listenerDecodeResult.put(emptyResourceName, new ListenerResult(set));
return null;
}
return new HashMap<>();
}
private Set<String> decodeResourceToListener(Listener resource) {
return resource.getFilterChainsList().stream()
.flatMap(e -> e.getFiltersList().stream())
.map(Filter::getTypedConfig)
.map(LdsProtocol::unpackHttpConnectionManager)
.filter(Objects::nonNull)
.map(HttpConnectionManager::getRds)
.map(Rds::getRouteConfigName)
.collect(Collectors.toSet());
}
private static Listener unpackListener(Any any) {
try {
return any.unpack(Listener.class);
} catch (InvalidProtocolBufferException e) {
logger.error(REGISTRY_ERROR_RESPONSE_XDS, "", "", "Error occur when decode xDS response.", e);
return null;
}
}
private static HttpConnectionManager unpackHttpConnectionManager(Any any) {
try {
if (!any.is(HttpConnectionManager.class)) {
return null;
}
return any.unpack(HttpConnectionManager.class);
} catch (InvalidProtocolBufferException e) {
logger.error(REGISTRY_ERROR_RESPONSE_XDS, "", "", "Error occur when decode xDS response.", e);
return null;
}
}
}

175
dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/RdsProtocol.java Normal file
View File

@ -0,0 +1,175 @@
/*
* 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.xds.protocol.impl;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.xds.AdsObserver;
import org.apache.dubbo.xds.protocol.AbstractProtocol;
import org.apache.dubbo.xds.resource.XdsRoute;
import org.apache.dubbo.xds.resource.XdsRouteAction;
import org.apache.dubbo.xds.resource.XdsRouteConfiguration;
import org.apache.dubbo.xds.resource.XdsRouteMatch;
import org.apache.dubbo.xds.resource.XdsVirtualHost;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import com.google.protobuf.Any;
import com.google.protobuf.InvalidProtocolBufferException;
import io.envoyproxy.envoy.config.core.v3.Node;
import io.envoyproxy.envoy.config.route.v3.Route;
import io.envoyproxy.envoy.config.route.v3.RouteAction;
import io.envoyproxy.envoy.config.route.v3.RouteConfiguration;
import io.envoyproxy.envoy.config.route.v3.RouteMatch;
import io.envoyproxy.envoy.config.route.v3.VirtualHost;
import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_RESPONSE_XDS;
public class RdsProtocol extends AbstractProtocol<String> {
private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(RdsProtocol.class);
protected Consumer<List<XdsRouteConfiguration>> updateCallback;
public RdsProtocol(
AdsObserver adsObserver,
Node node,
int checkInterval,
Consumer<List<XdsRouteConfiguration>> updateCallback) {
super(adsObserver, node, checkInterval);
this.updateCallback = updateCallback;
}
@Override
public String getTypeUrl() {
return "type.googleapis.com/envoy.config.route.v3.RouteConfiguration";
}
@Override
protected Map<String, String> decodeDiscoveryResponse(DiscoveryResponse response) {
List<XdsRouteConfiguration> xdsRouteConfigurations = parse(response);
System.out.println(xdsRouteConfigurations);
updateCallback.accept(xdsRouteConfigurations);
// if (getTypeUrl().equals(response.getTypeUrl())) {
// return response.getResourcesList().stream()
// .map(RdsProtocol::unpackRouteConfiguration)
// .filter(Objects::nonNull)
// .collect(Collectors.toConcurrentMap(RouteConfiguration::getName,
// this::decodeResourceToListener));
// }
return new HashMap<>();
}
public List<XdsRouteConfiguration> parse(DiscoveryResponse response) {
if (!getTypeUrl().equals(response.getTypeUrl())) {
return null;
}
return response.getResourcesList().stream()
.map(RdsProtocol::unpackRouteConfiguration)
.filter(Objects::nonNull)
.map(this::parseRouteConfiguration)
.collect(Collectors.toList());
}
public XdsRouteConfiguration parseRouteConfiguration(RouteConfiguration routeConfiguration) {
XdsRouteConfiguration xdsRouteConfiguration = new XdsRouteConfiguration();
xdsRouteConfiguration.setName(routeConfiguration.getName());
List<XdsVirtualHost> xdsVirtualHosts = routeConfiguration.getVirtualHostsList().stream()
.map(this::parseVirtualHost)
.collect(Collectors.toList());
Map<String, XdsVirtualHost> xdsVirtualHostMap = new HashMap<>();
xdsVirtualHosts.forEach(xdsVirtualHost -> {
String domain = xdsVirtualHost.getDomains().get(0).split("\\.")[0];
xdsVirtualHostMap.put(domain, xdsVirtualHost);
// for (String domain : xdsVirtualHost.getDomains()) {
// xdsVirtualHostMap.put(domain, xdsVirtualHost);
// }
});
xdsRouteConfiguration.setVirtualHosts(xdsVirtualHostMap);
return xdsRouteConfiguration;
}
public XdsVirtualHost parseVirtualHost(VirtualHost virtualHost) {
XdsVirtualHost xdsVirtualHost = new XdsVirtualHost();
List<String> domains = virtualHost.getDomainsList();
List<XdsRoute> xdsRoutes =
virtualHost.getRoutesList().stream().map(this::parseRoute).collect(Collectors.toList());
xdsVirtualHost.setName(virtualHost.getName());
xdsVirtualHost.setRoutes(xdsRoutes);
xdsVirtualHost.setDomains(domains);
return xdsVirtualHost;
}
public XdsRoute parseRoute(Route route) {
XdsRoute xdsRoute = new XdsRoute();
XdsRouteMatch xdsRouteMatch = parseRouteMatch(route.getMatch());
XdsRouteAction xdsRouteAction = parseRouteAction(route.getRoute());
xdsRoute.setRouteMatch(xdsRouteMatch);
xdsRoute.setRouteAction(xdsRouteAction);
return xdsRoute;
}
public XdsRouteMatch parseRouteMatch(RouteMatch routeMatch) {
XdsRouteMatch xdsRouteMatch = new XdsRouteMatch();
String prefix = routeMatch.getPrefix();
String path = routeMatch.getPath();
xdsRouteMatch.setPrefix(prefix);
xdsRouteMatch.setPath(path);
return xdsRouteMatch;
}
public XdsRouteAction parseRouteAction(RouteAction routeAction) {
XdsRouteAction xdsRouteAction = new XdsRouteAction();
String cluster = routeAction.getCluster();
if (cluster.equals("")) {
System.out.println("parse weight clusters");
}
xdsRouteAction.setCluster(cluster);
return xdsRouteAction;
}
private static RouteConfiguration unpackRouteConfiguration(Any any) {
try {
return any.unpack(RouteConfiguration.class);
} catch (InvalidProtocolBufferException e) {
logger.error(REGISTRY_ERROR_RESPONSE_XDS, "", "", "Error occur when decode xDS response.", e);
return null;
}
}
}

50
dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistry.java Normal file
View File

@ -0,0 +1,50 @@
/*
* 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.xds.registry;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.support.FailbackRegistry;
/**
* Empty implements for xDS <br/>
* xDS only support `Service Discovery` mode register <br/>
* Used to compat past version like 2.6.x, 2.7.x with interface level register <br/>
* {@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) {}
}

34
dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistryFactory.java Normal file
View File

@ -0,0 +1,34 @@
/*
* 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.xds.registry;
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);
}
}

56
dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscovery.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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.xds.registry;
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.ReflectionBasedServiceDiscovery;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.xds.PilotExchanger;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_INITIALIZE_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);
doInitialize(registryURL);
}
public void doInitialize(URL registryURL) {
try {
exchanger = PilotExchanger.initialize(registryURL);
} catch (Throwable t) {
logger.error(REGISTRY_ERROR_INITIALIZE_XDS, "", "", t.getMessage(), t);
}
}
public void doDestroy() {
try {
if (exchanger == null) {
return;
}
exchanger.destroy();
} catch (Throwable t) {
logger.error(REGISTRY_ERROR_INITIALIZE_XDS, "", "", t.getMessage(), t);
}
}
}

47
dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscoveryFactory.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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.xds.registry;
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.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 XdsServiceDiscovery 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;
}
}

64
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsCluster.java Normal file
View File

@ -0,0 +1,64 @@
/*
* 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.xds.resource;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import java.util.List;
public class XdsCluster<T> {
private String name;
private String lbPolicy;
private List<XdsEndpoint> xdsEndpoints;
public BitList<Invoker<T>> getInvokers() {
return invokers;
}
public void setInvokers(BitList<Invoker<T>> invokers) {
this.invokers = invokers;
}
private BitList<Invoker<T>> invokers;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getLbPolicy() {
return lbPolicy;
}
public void setLbPolicy(String lbPolicy) {
this.lbPolicy = lbPolicy;
}
public List<XdsEndpoint> getXdsEndpoints() {
return xdsEndpoints;
}
public void setXdsEndpoints(List<XdsEndpoint> xdsEndpoints) {
this.xdsEndpoints = xdsEndpoints;
}
}

37
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterWeight.java Normal file
View File

@ -0,0 +1,37 @@
/*
* 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.xds.resource;
public class XdsClusterWeight {
private final String name;
private final int weight;
public XdsClusterWeight(String name, int weight) {
this.name = name;
this.weight = weight;
}
public String getName() {
return name;
}
public int getWeight() {
return weight;
}
}

66
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpoint.java Normal file
View File

@ -0,0 +1,66 @@
/*
* 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.xds.resource;
public class XdsEndpoint {
private String clusterName;
private String address;
private int portValue;
private boolean healthy;
private int weight;
public String getClusterName() {
return clusterName;
}
public void setClusterName(String clusterName) {
this.clusterName = clusterName;
}
public String getAddress() {
return address;
}
public void setAddress(String address) {
this.address = address;
}
public int getPortValue() {
return portValue;
}
public void setPortValue(int portValue) {
this.portValue = portValue;
}
public boolean isHealthy() {
return healthy;
}
public void setHealthy(boolean healthy) {
this.healthy = healthy;
}
public int getWeight() {
return weight;
}
public void setWeight(int weight) {
this.weight = weight;
}
}

49
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRoute.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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.xds.resource;
public class XdsRoute {
private String name;
private XdsRouteMatch routeMatch;
private XdsRouteAction routeAction;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public XdsRouteMatch getRouteMatch() {
return routeMatch;
}
public void setRouteMatch(XdsRouteMatch routeMatch) {
this.routeMatch = routeMatch;
}
public XdsRouteAction getRouteAction() {
return routeAction;
}
public void setRouteAction(XdsRouteAction routeAction) {
this.routeAction = routeAction;
}
}

41
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteAction.java Normal file
View File

@ -0,0 +1,41 @@
/*
* 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.xds.resource;
import java.util.List;
public class XdsRouteAction {
private String cluster;
private List<XdsClusterWeight> clusterWeights;
public String getCluster() {
return cluster;
}
public void setCluster(String cluster) {
this.cluster = cluster;
}
public List<XdsClusterWeight> getClusterWeights() {
return clusterWeights;
}
public void setClusterWeights(List<XdsClusterWeight> clusterWeights) {
this.clusterWeights = clusterWeights;
}
}

41
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteConfiguration.java Normal file
View File

@ -0,0 +1,41 @@
/*
* 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.xds.resource;
import java.util.Map;
public class XdsRouteConfiguration {
private String name;
private Map<String, XdsVirtualHost> virtualHosts;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Map<String, XdsVirtualHost> getVirtualHosts() {
return virtualHosts;
}
public void setVirtualHosts(Map<String, XdsVirtualHost> virtualHosts) {
this.virtualHosts = virtualHosts;
}
}

70
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteMatch.java Normal file
View File

@ -0,0 +1,70 @@
/*
* 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.xds.resource;
public class XdsRouteMatch {
private String prefix;
private String path;
public String getRegex() {
return regex;
}
public void setRegex(String regex) {
this.regex = regex;
}
private String regex;
public boolean isCaseSensitive() {
return caseSensitive;
}
public void setCaseSensitive(boolean caseSensitive) {
this.caseSensitive = caseSensitive;
}
private boolean caseSensitive;
public String getPrefix() {
return prefix;
}
public void setPrefix(String prefix) {
this.prefix = prefix;
}
public String getPath() {
return path;
}
public void setPath(String path) {
this.path = path;
}
public boolean isMatch(String input) {
if (getPath() != null && !getPath().equals("")) {
return isCaseSensitive() ? getPath().equals(input) : getPath().equalsIgnoreCase(input);
} else if (getPrefix() != null) {
return isCaseSensitive()
? input.startsWith(getPrefix())
: input.toLowerCase().startsWith(getPrefix());
}
return input.matches(getRegex());
}
}

52
dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsVirtualHost.java Normal file
View File

@ -0,0 +1,52 @@
/*
* 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.xds.resource;
import java.util.List;
public class XdsVirtualHost {
private String name;
private List<String> domains;
private List<XdsRoute> routes;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public List<String> getDomains() {
return domains;
}
public void setDomains(List<String> domains) {
this.domains = domains;
}
public List<XdsRoute> getRoutes() {
return routes;
}
public void setRoutes(List<XdsRoute> routes) {
this.routes = routes;
}
}

121
dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java Normal file
View File

@ -0,0 +1,121 @@
/*
* 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.xds.router;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcException;
import org.apache.dubbo.rpc.cluster.router.RouterSnapshotNode;
import org.apache.dubbo.rpc.cluster.router.state.AbstractStateRouter;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.support.RpcUtils;
import org.apache.dubbo.xds.PilotExchanger;
import org.apache.dubbo.xds.resource.XdsCluster;
import org.apache.dubbo.xds.resource.XdsClusterWeight;
import org.apache.dubbo.xds.resource.XdsRoute;
import org.apache.dubbo.xds.resource.XdsVirtualHost;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
public class XdsRouter<T> extends AbstractStateRouter<T> {
private final PilotExchanger pilotExchanger = PilotExchanger.getInstance();
private Map<String, XdsVirtualHost> xdsVirtualHostMap = new ConcurrentHashMap<>();
private Map<String, XdsCluster> xdsClusterMap = new ConcurrentHashMap<>();
public XdsRouter(URL url) {
super(url);
}
@Override
protected BitList<Invoker<T>> doRoute(
BitList<Invoker<T>> invokers,
URL url,
Invocation invocation,
boolean needToPrintMessage,
Holder<RouterSnapshotNode<T>> routerSnapshotNodeHolder,
Holder<String> messageHolder)
throws RpcException {
// return all invokers directly if xds is not used
// TODO：need to consider where to set ‘xds’ param
if (!url.getParameter("xds", false)) {
return invokers;
}
// 1. match cluster
String matchedCluster = matchCluster(invocation);
// 2. match invokers
BitList<Invoker<T>> matchedInvokers = matchInvoker(matchedCluster, invokers);
return matchedInvokers;
}
private String matchCluster(Invocation invocation) {
String cluster = null;
String serviceName = invocation.getInvoker().getUrl().getParameter("providedBy");
XdsVirtualHost xdsVirtualHost = pilotExchanger.getXdsVirtualHostMap().get(serviceName);
// match route
for (XdsRoute xdsRoute : xdsVirtualHost.getRoutes()) {
// match path
String path = "/" + invocation.getInvoker().getUrl().getPath() + "/" + RpcUtils.getMethodName(invocation);
if (xdsRoute.getRouteMatch().isMatch(path)) {
cluster = xdsRoute.getRouteAction().getCluster();
// if weighted cluster
if (cluster == null) {
cluster = computeWeightCluster(xdsRoute.getRouteAction().getClusterWeights());
}
}
}
return cluster;
}
private String computeWeightCluster(List<XdsClusterWeight> weightedClusters) {
int totalWeight = Math.max(
weightedClusters.stream().mapToInt(XdsClusterWeight::getWeight).sum(), 1);
int target = ThreadLocalRandom.current().nextInt(1, totalWeight + 1);
for (XdsClusterWeight xdsClusterWeight : weightedClusters) {
int weight = xdsClusterWeight.getWeight();
target -= weight;
if (target <= 0) {
return xdsClusterWeight.getName();
}
}
return null;
}
private BitList<Invoker<T>> matchInvoker(String clusterName, BitList<Invoker<T>> invokers) {
List<Invoker<T>> filterInvokers = invokers.stream()
.filter(inv -> inv.getUrl().getParameter("clusterID").equals(clusterName))
.collect(Collectors.toList());
return new BitList<>(filterInvokers);
}
}

31
dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouterFactory.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.xds.router;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
import org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
@Activate(order = 100)
public class XdsRouterFactory implements StateRouterFactory {
@Override
public <T> StateRouter<T> getRouter(Class<T> interfaceClass, URL url) {
return new XdsRouter<>(url);
}
}

1
dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory Normal file
View File

@ -0,0 +1 @@
xds=org.apache.dubbo.xds.registry.XdsRegistryFactory

1
dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory Normal file
View File

@ -0,0 +1 @@
xds=org.apache.dubbo.xds.registry.XdsServiceDiscoveryFactory

1
dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster Normal file
View File

@ -0,0 +1 @@
xds=org.apache.dubbo.xds.cluster.XdsCluster

1
dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory Normal file
View File

@ -0,0 +1 @@
xds=org.apache.dubbo.xds.router.XdsRouterFactory

23
dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoService.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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.xds;
public interface DemoService {
default String sayHello() {
return null;
}
}

24
dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoServiceImpl.java Normal file
View File

@ -0,0 +1,24 @@
/*
* 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.xds;
public class DemoServiceImpl implements DemoService {
@Override
public String sayHello() {
return "hello";
}
}

118
dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoTest.java Normal file
View File

@ -0,0 +1,118 @@
/*
* 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.xds;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.Protocol;
import org.apache.dubbo.rpc.ProxyFactory;
import org.apache.dubbo.rpc.cluster.Directory;
import org.apache.dubbo.rpc.cluster.RouterChain;
import org.apache.dubbo.rpc.cluster.SingleRouterChain;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.xds.directory.XdsDirectory;
import org.apache.dubbo.xds.resource.XdsCluster;
import org.apache.dubbo.xds.resource.XdsVirtualHost;
import org.apache.dubbo.xds.router.XdsRouter;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.mockito.Mockito;
public class DemoTest {
private Protocol protocol =
ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
private ProxyFactory proxy =
ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
// @Test
public void testXdsRouterInitial() throws InterruptedException {
URL url = URL.valueOf("xds://localhost:15010/?secure=plaintext");
PilotExchanger.initialize(url);
Thread.sleep(7000);
Directory directory = Mockito.mock(Directory.class);
Mockito.when(directory.getConsumerUrl())
.thenReturn(URL.valueOf("dubbo://0.0.0.0:15010/DemoService?provided-by=dubbo-samples-xds-provider"));
Mockito.when(directory.getInterface()).thenReturn(DemoService.class);
// doReturn(DemoService.class).when(directory.getInterface());
Mockito.when(directory.getProtocol()).thenReturn(protocol);
SingleRouterChain singleRouterChain =
new SingleRouterChain<>(Collections.emptyList(), Arrays.asList(new XdsRouter<>(url)), false, null);
RouterChain routerChain = new RouterChain<>(new SingleRouterChain[] {singleRouterChain, singleRouterChain});
// doReturn(routerChain).when(directory.getRouterChain());
Mockito.when(directory.getRouterChain()).thenReturn(routerChain);
XdsDirectory<?> xdsDirectory = new XdsDirectory<>(directory);
Invocation invocation = Mockito.mock(Invocation.class);
Invoker invoker = Mockito.mock(Invoker.class);
URL url1 = URL.valueOf("consumer://0.0.0.0:15010/DemoService?providedBy=dubbo-samples-xds-provider&xds=true");
Mockito.when(invoker.getUrl()).thenReturn(url1);
// doReturn(invoker).when(invocation.getInvoker());
Mockito.when(invocation.getInvoker()).thenReturn(invoker);
while (true) {
Map<String, XdsVirtualHost> xdsVirtualHostMap = xdsDirectory.getXdsVirtualHostMap();
Map<String, ? extends XdsCluster<?>> xdsClusterMap = xdsDirectory.getXdsClusterMap();
if (!xdsVirtualHostMap.isEmpty() && !xdsClusterMap.isEmpty()) {
// xdsRouterDemo.route(invokers, url, invocation, false, null);
xdsDirectory.list(invocation);
break;
}
Thread.yield();
}
}
private Invoker<Object> createInvoker(String app, String address) {
URL url = URL.valueOf("dubbo://" + address + "/DemoInterface?"
+ (StringUtils.isEmpty(app) ? "" : "remote.application=" + app));
Invoker invoker = Mockito.mock(Invoker.class);
Mockito.when(invoker.getUrl()).thenReturn(url);
return invoker;
}
@AfterAll
public static void after() {
// ProtocolUtils.closeAll();
ApplicationModel.defaultModel()
.getDefaultModule()
.getServiceRepository()
.unregisterService(DemoService.class);
}
@BeforeAll
public static void setup() {
ApplicationModel.defaultModel()
.getDefaultModule()
.getServiceRepository()
.registerService(DemoService.class);
}
}

1
pom.xml
View File

@ -95,6 +95,7 @@
<module>dubbo-spring-boot</module>
<module>dubbo-test</module>
<module>dubbo-maven-plugin</module>
<module>dubbo-xds</module>
</modules>
<scm>