diff --git a/.artifacts b/.artifacts index cd6334693d..ef7a4e32ba 100644 --- a/.artifacts +++ b/.artifacts @@ -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 diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java index 7ac843b870..c2ebd347d9 100644 --- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java +++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/Directory.java @@ -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 extends Node { default boolean isNotificationReceived() { return false; } + + default Protocol getProtocol() { + return null; + } } diff --git a/dubbo-config/dubbo-config-api/pom.xml b/dubbo-config/dubbo-config-api/pom.xml index 12490c4d9e..4fdaa61842 100644 --- a/dubbo-config/dubbo-config-api/pom.xml +++ b/dubbo-config/dubbo-config-api/pom.xml @@ -251,5 +251,10 @@ resteasy-jackson-provider test + + org.apache.dubbo + dubbo-xds + ${project.parent.version} + diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java index cca118bbfa..9cdf9bde49 100644 --- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java +++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/ReferenceConfig.java @@ -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 extends ReferenceConfigBase { 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)) { diff --git a/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml b/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml index 024baf4152..c0891cdce2 100644 --- a/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml +++ b/dubbo-demo/dubbo-demo-spring-boot/dubbo-demo-spring-boot-consumer/pom.xml @@ -37,6 +37,12 @@ ${project.parent.version} + + org.apache.dubbo + dubbo-xds + ${project.version} + + org.apache.dubbo dubbo-registry-zookeeper diff --git a/dubbo-distribution/dubbo-all-shaded/pom.xml b/dubbo-distribution/dubbo-all-shaded/pom.xml index 8f60eb20b5..dd9f1bb98e 100644 --- a/dubbo-distribution/dubbo-all-shaded/pom.xml +++ b/dubbo-distribution/dubbo-all-shaded/pom.xml @@ -296,7 +296,6 @@ compile true - org.apache.dubbo diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml index 19fd7dbbfc..e27da32179 100644 --- a/dubbo-distribution/dubbo-all/pom.xml +++ b/dubbo-distribution/dubbo-all/pom.xml @@ -334,6 +334,13 @@ compile true + + org.apache.dubbo + dubbo-xds + ${project.version} + compile + true + @@ -552,6 +559,7 @@ org.apache.dubbo:dubbo-registry-multiple org.apache.dubbo:dubbo-registry-nacos org.apache.dubbo:dubbo-registry-zookeeper + org.apache.dubbo:dubbo-xds org.apache.dubbo:dubbo-remoting-api org.apache.dubbo:dubbo-remoting-http org.apache.dubbo:dubbo-remoting-http12 @@ -1020,6 +1028,10 @@ META-INF/dubbo/internal/org.apache.dubbo.registry.integration.ServiceURLCustomizer + + + META-INF/dubbo/internal/org.apache.dubbo.xds.bootstrap.XdsCertificateSigner + diff --git a/dubbo-distribution/dubbo-bom/pom.xml b/dubbo-distribution/dubbo-bom/pom.xml index 7e535f3421..2c6dbc6654 100644 --- a/dubbo-distribution/dubbo-bom/pom.xml +++ b/dubbo-distribution/dubbo-bom/pom.xml @@ -377,6 +377,11 @@ dubbo-registry-zookeeper ${project.version} + + org.apache.dubbo + dubbo-xds + ${project.version} + diff --git a/dubbo-metadata/dubbo-metadata-api/pom.xml b/dubbo-metadata/dubbo-metadata-api/pom.xml index 8fc6af86ea..81fb55e5d3 100644 --- a/dubbo-metadata/dubbo-metadata-api/pom.xml +++ b/dubbo-metadata/dubbo-metadata-api/pom.xml @@ -37,13 +37,6 @@ ${project.parent.version} true - - - org.apache.dubbo - dubbo-cluster - ${project.parent.version} - - org.apache.dubbo dubbo-metrics-api diff --git a/dubbo-registry/dubbo-registry-api/pom.xml b/dubbo-registry/dubbo-registry-api/pom.xml index 3d40787236..625461277d 100644 --- a/dubbo-registry/dubbo-registry-api/pom.xml +++ b/dubbo-registry/dubbo-registry-api/pom.xml @@ -38,12 +38,6 @@ ${project.parent.version} - - org.apache.dubbo - dubbo-cluster - ${project.parent.version} - - org.apache.dubbo dubbo-metadata-api @@ -102,5 +96,10 @@ log4j-slf4j-impl test + + org.apache.dubbo + dubbo-cluster + ${project.parent.version} + diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java index be8c2e74c2..a0a54e1df8 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/DynamicDirectory.java @@ -169,6 +169,10 @@ public abstract class DynamicDirectory extends AbstractDirectory implement this.protocol = protocol; } + public Protocol getProtocol() { + return this.protocol; + } + public void setRegistry(Registry registry) { this.registry = registry; } diff --git a/dubbo-test/dubbo-dependencies-all/pom.xml b/dubbo-test/dubbo-dependencies-all/pom.xml index 8bfc74434b..c8288c7b93 100644 --- a/dubbo-test/dubbo-dependencies-all/pom.xml +++ b/dubbo-test/dubbo-dependencies-all/pom.xml @@ -293,6 +293,11 @@ dubbo-registry-zookeeper ${project.version} + + org.apache.dubbo + dubbo-xds + ${project.version} + diff --git a/dubbo-xds/pom.xml b/dubbo-xds/pom.xml new file mode 100644 index 0000000000..5fe64b4d68 --- /dev/null +++ b/dubbo-xds/pom.xml @@ -0,0 +1,146 @@ + + + + 4.0.0 + + org.apache.dubbo + dubbo-parent + ${revision} + ../pom.xml + + dubbo-xds + jar + ${project.artifactId} + The xds module of dubbo project + + false + + + + org.apache.dubbo + dubbo-rpc-api + ${project.parent.version} + + + org.apache.dubbo + dubbo-rpc-injvm + ${project.parent.version} + test + + + org.apache.dubbo + dubbo-remoting-netty4 + ${project.parent.version} + test + + + org.apache.curator + curator-framework + test + + + org.apache.zookeeper + zookeeper + test + + + org.apache.dubbo + dubbo-test-check + ${project.parent.version} + test + + + org.apache.dubbo + dubbo-metrics-registry + ${project.parent.version} + compile + + + org.apache.dubbo + dubbo-metrics-default + ${project.parent.version} + true + + + io.micrometer + micrometer-tracing-integration-test + test + + + org.apache.logging.log4j + log4j-slf4j-impl + test + + + io.grpc + grpc-api + 1.61.0 + compile + + + io.grpc + grpc-protobuf + + + + io.grpc + grpc-stub + + + + io.grpc + grpc-netty-shaded + + + + io.envoyproxy.controlplane + api + + + + com.google.protobuf + protobuf-java + + + + com.google.protobuf + protobuf-java-util + + + org.bouncycastle + bcpkix-jdk15on + + + org.apache.dubbo + dubbo-cluster + ${project.parent.version} + + + org.apache.dubbo + dubbo-registry-api + ${project.parent.version} + + + org.apache.dubbo + dubbo-rpc-dubbo + ${project.parent.version} + test + + + + diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/AdsObserver.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/AdsObserver.java new file mode 100644 index 0000000000..d756fd047c --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/AdsObserver.java @@ -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 listeners = new ConcurrentHashMap<>(); + + protected StreamObserver requestObserver; + + private final Map observedResources = new ConcurrentHashMap<>(); + + public AdsObserver(URL url, Node node) { + this.url = url; + this.node = node; + this.xdsChannel = new XdsChannel(url); + this.applicationModel = url.getOrDefaultApplicationModel(); + } + + public void addListener(AbstractProtocol protocol) { + listeners.put(protocol.getTypeUrl(), protocol); + } + + public void request(DiscoveryRequest discoveryRequest) { + if (requestObserver == null) { + requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(this)); + } + requestObserver.onNext(discoveryRequest); + observedResources.put(discoveryRequest.getTypeUrl(), discoveryRequest); + } + + private static class ResponseObserver implements StreamObserver { + private AdsObserver adsObserver; + + public ResponseObserver(AdsObserver adsObserver) { + this.adsObserver = adsObserver; + } + + @Override + public void onNext(DiscoveryResponse discoveryResponse) { + 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(); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/NodeBuilder.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/NodeBuilder.java new file mode 100644 index 0000000000..7dfe2bc27e --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/NodeBuilder.java @@ -0,0 +1,43 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.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(); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/PilotExchanger.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/PilotExchanger.java new file mode 100644 index 0000000000..76942fae9b --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/PilotExchanger.java @@ -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 domainObserveRequest = new ConcurrentHashSet(); + + private static PilotExchanger GLOBAL_PILOT_EXCHANGER = null; + + private static final Map xdsVirtualHostMap = new ConcurrentHashMap<>(); + + private static final Map xdsClusterMap = new ConcurrentHashMap<>(); + + private final Map> rdsListeners = new ConcurrentHashMap<>(); + + private final Map> cdsListeners = new ConcurrentHashMap<>(); + + protected PilotExchanger(URL url) { + int pollingTimeout = url.getParameter("pollingTimeout", 10); + adsObserver = new AdsObserver(url, NodeBuilder.build()); + + // rds resources callback + Consumer> 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> 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> ldsCallback = rdsProtocol::subscribeResource; + ldsProtocol.setUpdateCallback(ldsCallback); + ldsProtocol.subscribeListeners(); + + // cds resources callback，listen to all cds resources in the callback function + Consumer> cdsCallback = edsProtocol::subscribeResource; + cdsProtocol.setUpdateCallback(cdsCallback); + cdsProtocol.subscribeClusters(); + } + + public static Map getXdsVirtualHostMap() { + return xdsVirtualHostMap; + } + + public static Map 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(); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsChannel.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsChannel.java new file mode 100644 index 0000000000..4c47e94aaa --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsChannel.java @@ -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 observeDeltaDiscoveryRequest( + StreamObserver observer) { + return AggregatedDiscoveryServiceGrpc.newStub(channel).deltaAggregatedResources(observer); + } + + public StreamObserver createDeltaDiscoveryRequest(StreamObserver observer) { + return AggregatedDiscoveryServiceGrpc.newStub(channel).streamAggregatedResources(observer); + } + + public StreamObserver observeDeltaDiscoveryRequestV2( + StreamObserver observer) { + return io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.newStub(channel) + .deltaAggregatedResources(observer); + } + + public StreamObserver createDeltaDiscoveryRequestV2( + StreamObserver observer) { + return io.envoyproxy.envoy.service.discovery.v2.AggregatedDiscoveryServiceGrpc.newStub(channel) + .streamAggregatedResources(observer); + } + + public void destroy() { + channel.shutdown(); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsInitializationException.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsInitializationException.java new file mode 100644 index 0000000000..a57def0b5d --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsInitializationException.java @@ -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); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsListener.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsListener.java new file mode 100644 index 0000000000..fcd8d65b84 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/XdsListener.java @@ -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); +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapInfoImpl.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapInfoImpl.java new file mode 100644 index 0000000000..8f31ce304d --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapInfoImpl.java @@ -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 servers; + + private final String serverListenerResourceNameTemplate; + + private final Map certProviders; + + private final Node node; + + BootstrapInfoImpl( + List servers, + String serverListenerResourceNameTemplate, + Map certProviders, + Node node) { + this.servers = servers; + this.serverListenerResourceNameTemplate = serverListenerResourceNameTemplate; + this.certProviders = certProviders; + this.node = node; + } + + @Override + public List servers() { + return servers; + } + + public Map 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 servers; + private Node node; + + private Map certProviders; + + private String serverListenerResourceNameTemplate; + + Builder() {} + + @Override + Bootstrapper.BootstrapInfo.Builder servers(List 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 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); + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/Bootstrapper.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/Bootstrapper.java new file mode 100644 index 0000000000..5a5e63cc2d --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/Bootstrapper.java @@ -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 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 config(); + } + + public abstract static class BootstrapInfo { + public abstract List servers(); + + public abstract Map certProviders(); + + public abstract Node node(); + + public abstract String serverListenerResourceNameTemplate(); + + abstract static class Builder { + + abstract Builder servers(List servers); + + abstract Builder node(Node node); + + abstract Builder certProviders(@Nullable Map certProviders); + + abstract Builder serverListenerResourceNameTemplate(@Nullable String serverListenerResourceNameTemplate); + + abstract BootstrapInfo build(); + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapperImpl.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapperImpl.java new file mode 100644 index 0000000000..a77dd2d019 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/BootstrapperImpl.java @@ -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 rawBootstrap; + try { + rawBootstrap = (Map) JsonParser.parse(fileContent); + } catch (IOException e) { + throw new XdsInitializationException("Failed to parse JSON", e); + } + return bootstrap(rawBootstrap); + } + + @Override + BootstrapInfo bootstrap(Map 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 servers = parseServerInfos(rawServerConfigs); + builder.servers(servers); + + Node.Builder nodeBuilder = Node.newBuilder(); + Map 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 metadata = JsonUtil.getObject(rawNode, "metadata"); + Map rawLocality = JsonUtil.getObject(rawNode, "locality"); + } + builder.node(nodeBuilder.build()); + + Map certProvidersBlob = JsonUtil.getObject(rawData, "certificate_providers"); + if (certProvidersBlob != null) { + Map certProviders = new HashMap<>(certProvidersBlob.size()); + for (String name : certProvidersBlob.keySet()) { + Map valueMap = JsonUtil.getObject(certProvidersBlob, name); + String pluginName = checkForNull(JsonUtil.getString(valueMap, "plugin_name"), "plugin_name"); + Map 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 parseServerInfos(List rawServerConfigs) throws XdsInitializationException { + List servers = new LinkedList<>(); + List> serverConfigList = JsonUtil.checkObjectList(rawServerConfigs); + for (Map 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 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 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> jsonList, String serverUri) + throws XdsInitializationException { + return null; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/CertificateProviderInfoImpl.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/CertificateProviderInfoImpl.java new file mode 100644 index 0000000000..4eb5520b39 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/CertificateProviderInfoImpl.java @@ -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 config; + + CertificateProviderInfoImpl(String pluginName, Map config) { + this.pluginName = pluginName; + this.config = config; + } + + @Override + public String pluginName() { + return pluginName; + } + + @Override + public Map config() { + return config; + } + + @Override + public String toString() { + return "CertificateProviderInfo{" + "pluginName=" + pluginName + ", " + "config=" + config + "}"; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/ServerInfoImpl.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/ServerInfoImpl.java new file mode 100644 index 0000000000..a3be13996c --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/ServerInfoImpl.java @@ -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 + + "}"; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/XdsCertificateSigner.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/XdsCertificateSigner.java new file mode 100644 index 0000000000..3122095dda --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/bootstrap/XdsCertificateSigner.java @@ -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; + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsCluster.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsCluster.java new file mode 100644 index 0000000000..5933c0614f --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsCluster.java @@ -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 AbstractClusterInvoker doJoin(Directory directory) throws RpcException { + XdsDirectory xdsDirectory = new XdsDirectory<>(directory); + return new XdsClusterInvoker<>(xdsDirectory); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsClusterInvoker.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsClusterInvoker.java new file mode 100644 index 0000000000..c66b6c034a --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/cluster/XdsClusterInvoker.java @@ -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 extends AbstractClusterInvoker { + + public XdsClusterInvoker(Directory directory) { + super(directory); + } + + @Override + protected Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) + throws RpcException { + Invoker 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); + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java new file mode 100644 index 0000000000..5c03161ced --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/directory/XdsDirectory.java @@ -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 extends AbstractDirectory { + + private final URL url; + + private final Class serviceType; + + private final String[] applicationNames; + + private final String protocolName; + + PilotExchanger pilotExchanger = PilotExchanger.getInstance(); + + private Protocol protocol; + + private final Map xdsVirtualHostMap = new ConcurrentHashMap<>(); + + private final Map> xdsClusterMap = new ConcurrentHashMap<>(); + + public XdsDirectory(Directory 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 getXdsVirtualHostMap() { + return xdsVirtualHostMap; + } + + public Map> getXdsClusterMap() { + return xdsClusterMap; + } + + public Protocol getProtocol() { + return protocol; + } + + public void setProtocol(Protocol protocol) { + this.protocol = protocol; + } + + @Override + public Class getInterface() { + return serviceType; + } + + public List> doList( + SingleRouterChain singleRouterChain, BitList> invokers, Invocation invocation) { + List> result = singleRouterChain.route(this.getConsumerUrl(), invokers, invocation); + return (List) (result == null ? BitList.emptyList() : result); + } + + @Override + public List> getAllInvokers() { + return super.getInvokers(); + } + + public void onRdsChange(String applicationName, XdsVirtualHost xdsVirtualHost) { + Set oldCluster = getAllCluster(); + xdsVirtualHostMap.put(applicationName, xdsVirtualHost); + Set newCluster = getAllCluster(); + changeClusterSubscribe(oldCluster, newCluster); + } + + private Set getAllCluster() { + if (CollectionUtils.isEmptyMap(xdsVirtualHostMap)) { + return new HashSet<>(); + } + Set 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 oldCluster, Set newCluster) { + Set removeSubscribe = new HashSet<>(oldCluster); + Set 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 xdsCluster) { + xdsClusterMap.put(clusterName, xdsCluster); + String lbPolicy = xdsCluster.getLbPolicy(); + List xdsEndpoints = xdsCluster.getXdsEndpoints(); + BitList> 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 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; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioConstant.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioConstant.java new file mode 100644 index 0000000000..d25d1e3abd --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioConstant.java @@ -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"; +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioEnv.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioEnv.java new file mode 100644 index 0000000000..fae9187aeb --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/IstioEnv.java @@ -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:///ns//sa/ + return SPIFFE + trustDomain + NS + workloadNameSpace + SA + getServiceAccount(); + } + + public String getTrustDomain() { + return trustDomain; + } + + public String getWorkloadNameSpace() { + return workloadNameSpace; + } + + @Override + public String getCluster() { + return null; + } + + public int getRasKeySize() { + return rasKeySize; + } + + public boolean isECCFirst() { + return IstioConstant.DEFAULT_ECC_SIG_ALG.equals(eccSigAlg); + } + + public int getSecretTTL() { + return secretTTL; + } + + public float getSecretGracePeriodRatio() { + return secretGracePeriodRatio; + } + + public String getIstioMetaClusterId() { + return istioMetaClusterId; + } + + public String getCaCert() { + File caFile; + if (IstioConstant.ISTIO_PILOT_CERT_PROVIDER.equals(pilotCertProvider)) { + caFile = new File(IstioConstant.ISTIO_CA_PATH); + } else { + return null; + } + if (caFile.canRead()) { + try { + return FileUtils.readFileToString(caFile, StandardCharsets.UTF_8); + } catch (IOException e) { + logger.error( + LoggerCodeConstants.REGISTRY_ISTIO_EXCEPTION, "File Read Failed", "", "read ca file error", e); + } + } + return null; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/XdsEnv.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/XdsEnv.java new file mode 100644 index 0000000000..21d430cb45 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/istio/XdsEnv.java @@ -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(); +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/AbstractProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/AbstractProtocol.java new file mode 100644 index 0000000000..43a8e36c96 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/AbstractProtocol.java @@ -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 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 observeResourcesName; + + public static final String emptyResourceName = "emptyResourcesName"; + private final ReentrantLock resourceLock = new ReentrantLock(); + + protected Map, List>>> consumerObserveMap = new ConcurrentHashMap<>(); + + public Map, List>>> getConsumerObserveMap() { + return consumerObserveMap; + } + + protected Map 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 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 resourceNames) { + resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames; + + if (!resourceNames.isEmpty() && isCacheExistResource(resourceNames)) { + getResourceFromCache(resourceNames); + } else { + getResourceFromRemote(resourceNames); + } + } + + private Map getResourceFromCache(Set resourceNames) { + return resourceNames.stream() + .filter(o -> !StringUtils.isEmpty(o)) + .collect(Collectors.toMap(k -> k, this::getCacheResource)); + } + + public Map getResourceFromRemote(Set resourceNames) { + try { + resourceLock.lock(); + CompletableFuture> future = new CompletableFuture<>(); + observeResourcesName = resourceNames; + Set consumerObserveResourceNames = new HashSet<>(); + if (resourceNames.isEmpty()) { + consumerObserveResourceNames.add(emptyResourceName); + } else { + consumerObserveResourceNames = resourceNames; + } + + Consumer> futureConsumer = future::complete; + try { + writeLock.lock(); + ConcurrentHashMapUtils.computeIfAbsent( + (ConcurrentHashMap, List>>>) consumerObserveMap, + consumerObserveResourceNames, + key -> new ArrayList<>()) + .add(futureConsumer); + } finally { + writeLock.unlock(); + } + + Set 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 resourceNames) { + return DiscoveryRequest.newBuilder() + .setNode(node) + .setTypeUrl(getTypeUrl()) + .addAllResourceNames(resourceNames) + .build(); + } + + protected abstract Map decodeDiscoveryResponse(DiscoveryResponse response); + + @Override + public final void process(DiscoveryResponse discoveryResponse) { + Map newResult = decodeDiscoveryResponse(discoveryResponse); + Map oldResource = resourcesMap; + // discoveryResponseListener(oldResource, newResult); + resourcesMap = newResult; + } + + private void discoveryResponseListener(Map oldResult, Map newResult) { + Set 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, List>>> entry : consumerObserveMap.entrySet()) { + if (entry.getKey().stream().noneMatch(changedResourceNames::contains)) { + // none update + continue; + } + + Map dsResultMap = + entry.getKey().stream().collect(Collectors.toMap(k -> k, v -> newResult.get(v))); + entry.getValue().forEach(o -> o.accept(dsResultMap)); + } + } finally { + readLock.unlock(); + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/XdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/XdsProtocol.java new file mode 100644 index 0000000000..838988ad51 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/XdsProtocol.java @@ -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 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 resourceNames, Consumer> consumer, boolean isReConnect); +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/CdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/CdsProtocol.java new file mode 100644 index 0000000000..037f1c7fc8 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/CdsProtocol.java @@ -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 { + private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(CdsProtocol.class); + + public void setUpdateCallback(Consumer> updateCallback) { + this.updateCallback = updateCallback; + } + + private Consumer> 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 decodeDiscoveryResponse(DiscoveryResponse response) { + if (getTypeUrl().equals(response.getTypeUrl())) { + Set set = response.getResourcesList().stream() + .map(CdsProtocol::unpackCluster) + .filter(Objects::nonNull) + .map(Cluster::getName) + .collect(Collectors.toSet()); + updateCallback.accept(set); + // Map 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; + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/EdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/EdsProtocol.java new file mode 100644 index 0000000000..65b1586882 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/EdsProtocol.java @@ -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 { + + private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(EdsProtocol.class); + + public void setUpdateCallback(Consumer> updateCallback) { + this.updateCallback = updateCallback; + } + + private Consumer> updateCallback; + + public EdsProtocol( + AdsObserver adsObserver, Node node, int checkInterval, Consumer> updateCallback) { + super(adsObserver, node, checkInterval); + this.updateCallback = updateCallback; + } + + @Override + public String getTypeUrl() { + return "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"; + } + + @Override + protected Map decodeDiscoveryResponse(DiscoveryResponse response) { + List 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 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 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; + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/LdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/LdsProtocol.java new file mode 100644 index 0000000000..583b35f107 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/LdsProtocol.java @@ -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 { + private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(LdsProtocol.class); + + public void setUpdateCallback(Consumer> updateCallback) { + this.updateCallback = updateCallback; + } + + private Consumer> 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 decodeDiscoveryResponse(DiscoveryResponse response) { + if (getTypeUrl().equals(response.getTypeUrl())) { + Set set = response.getResourcesList().stream() + .map(LdsProtocol::unpackListener) + .filter(Objects::nonNull) + .flatMap(e -> decodeResourceToListener(e).stream()) + .collect(Collectors.toSet()); + updateCallback.accept(set); + // Map listenerDecodeResult = new ConcurrentHashMap<>(); + // listenerDecodeResult.put(emptyResourceName, new ListenerResult(set)); + return null; + } + return new HashMap<>(); + } + + private Set 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; + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/RdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/RdsProtocol.java new file mode 100644 index 0000000000..de077fd7c2 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/protocol/impl/RdsProtocol.java @@ -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 { + + private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(RdsProtocol.class); + + protected Consumer> updateCallback; + + public RdsProtocol( + AdsObserver adsObserver, + Node node, + int checkInterval, + Consumer> updateCallback) { + super(adsObserver, node, checkInterval); + this.updateCallback = updateCallback; + } + + @Override + public String getTypeUrl() { + return "type.googleapis.com/envoy.config.route.v3.RouteConfiguration"; + } + + @Override + protected Map decodeDiscoveryResponse(DiscoveryResponse response) { + List 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 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 xdsVirtualHosts = routeConfiguration.getVirtualHostsList().stream() + .map(this::parseVirtualHost) + .collect(Collectors.toList()); + + Map 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 domains = virtualHost.getDomainsList(); + + List 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; + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistry.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistry.java new file mode 100644 index 0000000000..2eb7ccad10 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistry.java @@ -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
+ * xDS only support `Service Discovery` mode register
+ * Used to compat past version like 2.6.x, 2.7.x with interface level register
+ * {@link XdsServiceDiscovery} is the real implementation of xDS + */ +public class XdsRegistry extends FailbackRegistry { + public XdsRegistry(URL url) { + super(url); + } + + @Override + public boolean isAvailable() { + return true; + } + + @Override + public void doRegister(URL url) {} + + @Override + public void doUnregister(URL url) {} + + @Override + public void doSubscribe(URL url, NotifyListener listener) {} + + @Override + public void doUnsubscribe(URL url, NotifyListener listener) {} +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistryFactory.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistryFactory.java new file mode 100644 index 0000000000..65a5ce00f3 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsRegistryFactory.java @@ -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); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscovery.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscovery.java new file mode 100644 index 0000000000..febf221e3b --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscovery.java @@ -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); + } + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscoveryFactory.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscoveryFactory.java new file mode 100644 index 0000000000..0769329ac3 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/registry/XdsServiceDiscoveryFactory.java @@ -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; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsCluster.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsCluster.java new file mode 100644 index 0000000000..173c9cb382 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsCluster.java @@ -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 { + private String name; + + private String lbPolicy; + + private List xdsEndpoints; + + public BitList> getInvokers() { + return invokers; + } + + public void setInvokers(BitList> invokers) { + this.invokers = invokers; + } + + private BitList> 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 getXdsEndpoints() { + return xdsEndpoints; + } + + public void setXdsEndpoints(List xdsEndpoints) { + this.xdsEndpoints = xdsEndpoints; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterWeight.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterWeight.java new file mode 100644 index 0000000000..1595929544 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsClusterWeight.java @@ -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; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpoint.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpoint.java new file mode 100644 index 0000000000..bc79be342e --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsEndpoint.java @@ -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; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRoute.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRoute.java new file mode 100644 index 0000000000..33000307a1 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRoute.java @@ -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; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteAction.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteAction.java new file mode 100644 index 0000000000..101163fd8e --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteAction.java @@ -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 clusterWeights; + + public String getCluster() { + return cluster; + } + + public void setCluster(String cluster) { + this.cluster = cluster; + } + + public List getClusterWeights() { + return clusterWeights; + } + + public void setClusterWeights(List clusterWeights) { + this.clusterWeights = clusterWeights; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteConfiguration.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteConfiguration.java new file mode 100644 index 0000000000..e32ad90372 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteConfiguration.java @@ -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 virtualHosts; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public Map getVirtualHosts() { + return virtualHosts; + } + + public void setVirtualHosts(Map virtualHosts) { + this.virtualHosts = virtualHosts; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteMatch.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteMatch.java new file mode 100644 index 0000000000..57e0b3363f --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsRouteMatch.java @@ -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()); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsVirtualHost.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsVirtualHost.java new file mode 100644 index 0000000000..6575741fce --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/resource/XdsVirtualHost.java @@ -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 domains; + + private List routes; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public List getDomains() { + return domains; + } + + public void setDomains(List domains) { + this.domains = domains; + } + + public List getRoutes() { + return routes; + } + + public void setRoutes(List routes) { + this.routes = routes; + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java new file mode 100644 index 0000000000..f28281b4b5 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouter.java @@ -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 extends AbstractStateRouter { + + private final PilotExchanger pilotExchanger = PilotExchanger.getInstance(); + + private Map xdsVirtualHostMap = new ConcurrentHashMap<>(); + + private Map xdsClusterMap = new ConcurrentHashMap<>(); + + public XdsRouter(URL url) { + super(url); + } + + @Override + protected BitList> doRoute( + BitList> invokers, + URL url, + Invocation invocation, + boolean needToPrintMessage, + Holder> routerSnapshotNodeHolder, + Holder 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> 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 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> matchInvoker(String clusterName, BitList> invokers) { + + List> filterInvokers = invokers.stream() + .filter(inv -> inv.getUrl().getParameter("clusterID").equals(clusterName)) + .collect(Collectors.toList()); + return new BitList<>(filterInvokers); + } +} diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouterFactory.java b/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouterFactory.java new file mode 100644 index 0000000000..3c0f510136 --- /dev/null +++ b/dubbo-xds/src/main/java/org/apache/dubbo/xds/router/XdsRouterFactory.java @@ -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 StateRouter getRouter(Class interfaceClass, URL url) { + return new XdsRouter<>(url); + } +} diff --git a/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory new file mode 100644 index 0000000000..0df432b5c2 --- /dev/null +++ b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory @@ -0,0 +1 @@ +xds=org.apache.dubbo.xds.registry.XdsRegistryFactory diff --git a/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory new file mode 100644 index 0000000000..5c44c79696 --- /dev/null +++ b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory @@ -0,0 +1 @@ +xds=org.apache.dubbo.xds.registry.XdsServiceDiscoveryFactory diff --git a/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster new file mode 100644 index 0000000000..63fc1e2140 --- /dev/null +++ b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster @@ -0,0 +1 @@ +xds=org.apache.dubbo.xds.cluster.XdsCluster diff --git a/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory new file mode 100644 index 0000000000..c6153b11c8 --- /dev/null +++ b/dubbo-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory @@ -0,0 +1 @@ +xds=org.apache.dubbo.xds.router.XdsRouterFactory diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoService.java b/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoService.java new file mode 100644 index 0000000000..7ae9ade8fc --- /dev/null +++ b/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoService.java @@ -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; + } +} diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoServiceImpl.java b/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoServiceImpl.java new file mode 100644 index 0000000000..7e5a735696 --- /dev/null +++ b/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoServiceImpl.java @@ -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"; + } +} diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoTest.java b/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoTest.java new file mode 100644 index 0000000000..fb27c37112 --- /dev/null +++ b/dubbo-xds/src/test/java/org/apache/dubbo/xds/DemoTest.java @@ -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 xdsVirtualHostMap = xdsDirectory.getXdsVirtualHostMap(); + Map> xdsClusterMap = xdsDirectory.getXdsClusterMap(); + if (!xdsVirtualHostMap.isEmpty() && !xdsClusterMap.isEmpty()) { + // xdsRouterDemo.route(invokers, url, invocation, false, null); + xdsDirectory.list(invocation); + break; + } + Thread.yield(); + } + } + + private Invoker 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); + } +} diff --git a/pom.xml b/pom.xml index 4dafbf47e6..8aaca36cf3 100644 --- a/pom.xml +++ b/pom.xml @@ -95,6 +95,7 @@ dubbo-spring-boot dubbo-test dubbo-maven-plugin + dubbo-xds