diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml index 8d84b38d8b..f409ae8892 100644 --- a/dubbo-dependencies-bom/pom.xml +++ b/dubbo-dependencies-bom/pom.xml @@ -125,6 +125,7 @@ 1.20 3.8.1 1.5.9 + 0.1.23 2.0 3.0.19.Final @@ -533,6 +534,11 @@ commons-lang3 ${commons_lang3_version} + + io.envoyproxy.controlplane + api + ${envoy_api_version} + diff --git a/dubbo-distribution/dubbo-all/pom.xml b/dubbo-distribution/dubbo-all/pom.xml index e8e4406211..6b009b43e4 100644 --- a/dubbo-distribution/dubbo-all/pom.xml +++ b/dubbo-distribution/dubbo-all/pom.xml @@ -171,6 +171,13 @@ compile true + + org.apache.dubbo + dubbo-registry-xds + ${project.version} + compile + true + org.apache.dubbo dubbo-monitor-api @@ -374,6 +381,7 @@ org.apache.dubbo:dubbo-registry-multiple org.apache.dubbo:dubbo-registry-kubernetes org.apache.dubbo:dubbo-registry-dns + org.apache.dubbo:dubbo-registry-xds org.apache.dubbo:dubbo-monitor-api org.apache.dubbo:dubbo-monitor-default org.apache.dubbo:dubbo-container-api diff --git a/dubbo-distribution/dubbo-bom/pom.xml b/dubbo-distribution/dubbo-bom/pom.xml index d3c1b3305e..5826d1047f 100644 --- a/dubbo-distribution/dubbo-bom/pom.xml +++ b/dubbo-distribution/dubbo-bom/pom.xml @@ -133,6 +133,11 @@ dubbo-registry-nacos ${project.version} + + org.apache.dubbo + dubbo-registry-xds + ${project.version} + org.apache.dubbo dubbo-monitor-api diff --git a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/SelfHostMetaServiceDiscovery.java b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/SelfHostMetaServiceDiscovery.java index 25ab77e5bf..1034420473 100644 --- a/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/SelfHostMetaServiceDiscovery.java +++ b/dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/SelfHostMetaServiceDiscovery.java @@ -240,7 +240,8 @@ public abstract class SelfHostMetaServiceDiscovery implements ServiceDiscovery { serviceInstanceRevisionMap.put(serviceName, serviceInstanceRevision)); if (logger.isDebugEnabled()) { - logger.debug("Poll DNS data. Service Instance changed: " + changed + " Service Name: " + serviceName); + logger.debug("Service changed event received (possibly because of DNS polling). " + + "Service Instance changed: " + changed + " Service Name: " + serviceName); } if (changed) { diff --git a/dubbo-registry/dubbo-registry-xds/pom.xml b/dubbo-registry/dubbo-registry-xds/pom.xml new file mode 100644 index 0000000000..effda5aa93 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/pom.xml @@ -0,0 +1,68 @@ + + + + + org.apache.dubbo + dubbo-registry + ${revision} + + 4.0.0 + + dubbo-registry-xds + ${project.artifactId} + The Xds registry module of Dubbo project + + + + + org.apache.dubbo + dubbo-registry-api + ${project.version} + + + + org.apache.dubbo + dubbo-common + ${project.version} + + + + io.grpc + grpc-protobuf + + + + io.grpc + grpc-stub + + + + io.envoyproxy.controlplane + api + + + + com.google.protobuf + protobuf-java-util + + + + + \ No newline at end of file diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistry.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistry.java new file mode 100644 index 0000000000..a4bd51e6b1 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistry.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.registry.xds; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.registry.NotifyListener; +import org.apache.dubbo.registry.support.FailbackRegistry; + +/** + * Empty implements for xDS
+ * xDS only support `Service Discovery` mode register
+ * Used to compat past version like 2.6.x, 2.7.x with interface level register
+ * {@link XdsServiceDiscovery} is the real implementation of xDS + */ +public class XdsRegistry extends FailbackRegistry { + public XdsRegistry(URL url) { + super(url); + } + + @Override + public boolean isAvailable() { + return true; + } + + @Override + public void doRegister(URL url) { + + } + + @Override + public void doUnregister(URL url) { + + } + + @Override + public void doSubscribe(URL url, NotifyListener listener) { + + } + + @Override + public void doUnsubscribe(URL url, NotifyListener listener) { + + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistryFactory.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsRegistryFactory.java new file mode 100644 index 0000000000..8fa130b197 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/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.registry.xds; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.registry.Registry; +import org.apache.dubbo.registry.support.AbstractRegistryFactory; + +public class XdsRegistryFactory extends AbstractRegistryFactory { + + @Override + protected String createRegistryCacheKey(URL url) { + return url.toFullString(); + } + + @Override + protected Registry createRegistry(URL url) { + return new XdsRegistry(url); + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java new file mode 100644 index 0000000000..0651e3ede4 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.registry.client.DefaultServiceInstance; +import org.apache.dubbo.registry.client.SelfHostMetaServiceDiscovery; +import org.apache.dubbo.registry.client.ServiceInstance; +import org.apache.dubbo.registry.client.event.listener.ServiceInstancesChangedListener; +import org.apache.dubbo.registry.xds.util.PilotExchanger; +import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; + +import java.util.Collection; +import java.util.Comparator; +import java.util.LinkedList; +import java.util.List; +import java.util.Set; + +public class XdsServiceDiscovery extends SelfHostMetaServiceDiscovery { + private PilotExchanger exchanger; + private URL registryURL; + + @Override + public void doInitialize(URL registryURL) throws Exception { + exchanger = PilotExchanger.initialize(registryURL); + } + + @Override + public void doDestroy() throws Exception { + exchanger.destroy(); + } + + @Override + public Set getServices() { + return exchanger.getServices(); + } + + @Override + public List getInstances(String serviceName) throws NullPointerException { + Set endpoints = exchanger.getEndpoints(serviceName); + return changedToInstances(serviceName, endpoints); + } + + @Override + public void addServiceInstancesChangedListener(ServiceInstancesChangedListener listener) throws NullPointerException, IllegalArgumentException { + listener.getServiceNames().forEach(serviceName -> { + exchanger.observeEndpoints(serviceName, (endpoints -> { + notifyListener(serviceName, listener, changedToInstances(serviceName, endpoints)); + })); + }); + } + + private List changedToInstances(String serviceName, Collection endpoints) { + List instances = new LinkedList<>(); + endpoints.forEach(endpoint -> { + DefaultServiceInstance serviceInstance = new DefaultServiceInstance(serviceName, endpoint.getAddress(), endpoint.getPortValue()); + // fill metadata by SelfHostMetaServiceDiscovery, will be fetched by RPC request + fillServiceInstance(serviceInstance); + instances.add(serviceInstance); + }); + instances.sort(Comparator.comparingInt(ServiceInstance::hashCode)); + return instances; + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.java new file mode 100644 index 0000000000..ef77ccbbc9 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscoveryFactory.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.registry.xds; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.registry.client.AbstractServiceDiscoveryFactory; +import org.apache.dubbo.registry.client.ServiceDiscovery; + +public class XdsServiceDiscoveryFactory extends AbstractServiceDiscoveryFactory { + @Override + protected ServiceDiscovery createDiscovery(URL registryURL) { + return new XdsServiceDiscovery(); + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/NodeBuilder.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/NodeBuilder.java new file mode 100644 index 0000000000..fad0dbf880 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/NodeBuilder.java @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util; + +import io.envoyproxy.envoy.config.core.v3.Node; + +public class NodeBuilder { + public static Node build() { + // TODO: fetch data from environment + return Node.newBuilder() + .setId("sidecar~127.0.0.1~ratings-v1-7dc98c7588-lwvqd.default~default.svc.cluster.local") + .setCluster("ratings.default") + .build(); + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java new file mode 100644 index 0000000000..3d49bc1a9f --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java @@ -0,0 +1,146 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util; + +import org.apache.dubbo.common.URL; +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.common.utils.ConcurrentHashSet; +import org.apache.dubbo.registry.xds.util.protocol.impl.EdsProtocol; +import org.apache.dubbo.registry.xds.util.protocol.impl.LdsProtocol; +import org.apache.dubbo.registry.xds.util.protocol.impl.RdsProtocol; +import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; +import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; +import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; +import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; + +import java.util.Collections; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Consumer; + +public class PilotExchanger { + + private final XdsChannel xdsChannel; + + private final RdsProtocol rdsProtocol; + + private final EdsProtocol edsProtocol; + + private ListenerResult listenerResult; + + private RouteResult routeResult; + + private final long observeRouteRequest; + + private final Map domainObserveRequest = new ConcurrentHashMap<>(); + + private final Map>>> domainObserveConsumer = new ConcurrentHashMap<>(); + + private PilotExchanger(URL url) { + xdsChannel = new XdsChannel(url); + LdsProtocol ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build()); + this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build()); + this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build()); + + this.listenerResult = ldsProtocol.getListeners(); + this.routeResult = rdsProtocol.getResource(listenerResult.getRouteConfigNames()); + + // Observer RDS update + this.observeRouteRequest = rdsProtocol.observeResource(listenerResult.getRouteConfigNames(), (newResult) -> { + // check if observed domain update ( will update endpoint observation ) + domainObserveConsumer.forEach((domain, consumer) -> { + Set newRoute = newResult.searchDomain(domain); + if (!routeResult.searchDomain(domain).equals(newRoute)) { + // routers in observed domain has been updated + Long domainRequest = domainObserveRequest.get(domain); + if (domainRequest == null) { + // router list is empty when observeEndpoints() called and domainRequest has not been created yet + // create new observation + doObserveEndpoints(domain); + } else { + // update observation by domainRequest + edsProtocol.updateObserve(domainRequest, newRoute); + } + } + }); + // update local cache + routeResult = newResult; + }); + + // Observe LDS updated + ldsProtocol.observeListeners((newListener) -> { + // update local cache + this.listenerResult = newListener; + // update RDS observation + rdsProtocol.updateObserve(observeRouteRequest, newListener.getRouteConfigNames()); + }); + } + + public static PilotExchanger initialize(URL url) { + return new PilotExchanger(url); + } + + public void destroy() { + xdsChannel.destroy(); + } + + public Set getServices() { + return routeResult.getDomains(); + } + + public Set getEndpoints(String domain) { + Set cluster = routeResult.searchDomain(domain); + if (CollectionUtils.isNotEmpty(cluster)) { + EndpointResult endpoint = edsProtocol.getResource(cluster); + return endpoint.getEndpoints(); + } else { + return Collections.emptySet(); + } + } + + public void observeEndpoints(String domain, Consumer> consumer) { + // store Consumer + domainObserveConsumer.compute(domain, (k, v) -> { + if (v == null) { + v = new ConcurrentHashSet<>(); + } + // support multi-consumer + v.add(consumer); + return v; + }); + if (!domainObserveRequest.containsKey(domain)) { + doObserveEndpoints(domain); + } + } + + private void doObserveEndpoints(String domain) { + Set router = routeResult.searchDomain(domain); + // if router is empty, do nothing + // observation will be created when RDS updates + if (CollectionUtils.isNotEmpty(router)) { + long endpointRequest = + edsProtocol.observeResource( + router, + endpointResult -> + // notify consumers + domainObserveConsumer.get(domain).forEach( + consumer1 -> consumer1.accept(endpointResult.getEndpoints()))); + domainObserveRequest.put(domain, endpointRequest); + } + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java new file mode 100644 index 0000000000..5336378909 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.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.registry.xds.util; + +import org.apache.dubbo.common.URL; + +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.ManagedChannelBuilder; +import io.grpc.stub.StreamObserver; + +public class XdsChannel { + private final ManagedChannel channel; + + protected XdsChannel(URL url) { + channel = ManagedChannelBuilder.forAddress(url.getHost(), url.getPort()) + .usePlaintext() + .build(); + } + + 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-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java new file mode 100644 index 0000000000..13ac5be347 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java @@ -0,0 +1,262 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol; + +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.util.XdsChannel; + +import io.envoyproxy.envoy.config.core.v3.Node; +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.stub.StreamObserver; + +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; + +public abstract class AbstractProtocol> implements XdsProtocol{ + + private static final Logger logger = LoggerFactory.getLogger(AbstractProtocol.class); + + protected final XdsChannel xdsChannel; + + protected final Node node; + + /** + * Store Request Parameter ( resourceNames ) + * K - requestId, V - resourceNames + */ + protected final Map> requestParam = new ConcurrentHashMap<>(); + + /** + * Store ADS Request Observer ( StreamObserver in Streaming Request ) + * K - requestId, V - StreamObserver + */ + private final Map> requestObserverMap = new ConcurrentHashMap<>(); + + /** + * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request ) + * K - requestId, V - StreamObserver + */ + private final Map> deltaRequestObserverMap = new ConcurrentHashMap<>(); + + /** + * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver ) + * K - requestId, V - CompletableFuture + */ + private final Map> streamResult = new ConcurrentHashMap<>(); + + /** + * Store consumers for Observers ( will consume message produced by Delta-ADS ) + * K - requestId, V - Consumer + */ + private final Map> consumers = new ConcurrentHashMap<>(); + + protected final AtomicLong requestId = new AtomicLong(0); + + public AbstractProtocol(XdsChannel xdsChannel, Node node) { + this.xdsChannel = xdsChannel; + this.node = node; + } + + /** + * Abstract method to obtain Type-URL from sub-class + * + * @return Type-URL of xDS + */ + public abstract String getTypeUrl(); + + @Override + public T getResource(Set resourceNames) { + long request = requestId.getAndIncrement(); + + // Store Request Parameter, which will be used for ACK + requestParam.put(request, resourceNames); + + // create observer + StreamObserver requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request)); + + // use future to get async result + CompletableFuture future = new CompletableFuture<>(); + requestObserverMap.put(request, requestObserver); + streamResult.put(request, future); + + // send request to control panel + requestObserver.onNext(buildDiscoveryRequest(resourceNames)); + + try { + // get result + return future.get(); + } catch (InterruptedException | ExecutionException e) { + logger.error("Error occur when request control panel."); + return null; + } finally { + // close observer + requestObserver.onCompleted(); + + // remove temp + streamResult.remove(request); + requestObserverMap.remove(request); + requestParam.remove(request); + } + } + + @Override + public long observeResource(Set resourceNames, Consumer consumer) { + long request = requestId.getAndIncrement(); + + // Store Request Parameter, which will be used for ACK + requestParam.put(request, resourceNames); + + // call once for full data + consumer.accept(getResource(resourceNames)); + + consumers.put(request, consumer); + deltaRequestObserverMap.compute(request, (k, v) -> { + // create Delta-ADS observer + v= xdsChannel.observeDeltaDiscoveryRequest(new DeltaResponseObserver(request)); + + // send observe request + v.onNext(buildDeltaDiscoveryRequest(resourceNames)); + return v; + }); + return request; + } + + @Override + public void updateObserve(long request, Set resourceNames) { + // send difference in resourceNames + deltaRequestObserverMap.get(request).onNext(buildDeltaDiscoveryRequest(request, resourceNames)); + } + + protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames) { + return DiscoveryRequest.newBuilder() + .setNode(node) + .setTypeUrl(getTypeUrl()) + .addAllResourceNames(resourceNames) + .build(); + } + + protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames, DiscoveryResponse response) { + // for ACK + return DiscoveryRequest.newBuilder() + .setNode(node) + .setTypeUrl(getTypeUrl()) + .addAllResourceNames(resourceNames) + .setVersionInfo(response.getVersionInfo()) + .setResponseNonce(response.getNonce()) + .build(); + } + + protected DeltaDiscoveryRequest buildDeltaDiscoveryRequest(Set resourceNames) { + return DeltaDiscoveryRequest.newBuilder() + .setNode(node) + .setTypeUrl(getTypeUrl()) + .addAllResourceNamesSubscribe(resourceNames) + .build(); + } + + protected DeltaDiscoveryRequest buildDeltaDiscoveryRequest(long request, Set resourceNames) { + // compare with previous + Set previous = requestParam.get(request); + Set unsubscribe = new HashSet(previous) {{ + removeAll(resourceNames); + }}; + requestParam.put(request, resourceNames); + return DeltaDiscoveryRequest.newBuilder() + .setNode(node) + .setTypeUrl(getTypeUrl()) + .addAllResourceNamesUnsubscribe(unsubscribe) + .addAllResourceNamesSubscribe(resourceNames) + .build(); + } + + private DeltaDiscoveryRequest buildDeltaDiscoveryRequest(Set resourceNames, DeltaDiscoveryResponse response) { + // for ACK + return DeltaDiscoveryRequest.newBuilder() + .setNode(node) + .setTypeUrl(getTypeUrl()) + .addAllResourceNamesSubscribe(resourceNames) + .setResponseNonce(response.getNonce()) + .build(); + } + + protected abstract T decodeDiscoveryResponse(DiscoveryResponse response); + + protected abstract S decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, S previous); + + private class ResponseObserver implements StreamObserver { + private final long requestId; + + public ResponseObserver(long requestId) { + this.requestId = requestId; + } + + @Override + public void onNext(DiscoveryResponse value) { + T result = decodeDiscoveryResponse(value); + requestObserverMap.get(requestId).onNext(buildDiscoveryRequest(requestParam.get(requestId), value)); + streamResult.get(requestId).complete(result); + } + + @Override + public void onError(Throwable t) { + logger.error("xDS Client received error message! detail:", t); + } + + @Override + public void onCompleted() { + // ignore + } + } + + private class DeltaResponseObserver implements StreamObserver { + private S delta = null; + private final long requestId; + + public DeltaResponseObserver(long requestId) { + this.requestId = requestId; + } + + @Override + public void onNext(DeltaDiscoveryResponse value) { + delta = decodeDeltaDiscoveryResponse(value, delta); + T routes = delta.getResource(); + consumers.get(requestId).accept(routes); + deltaRequestObserverMap.get(requestId).onNext(buildDeltaDiscoveryRequest(requestParam.get(requestId), value)); + } + + @Override + public void onError(Throwable t) { + logger.error("xDS Client received error message! detail:", t); + } + + @Override + public void onCompleted() { + // ignore + } + } + +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/DeltaResource.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/DeltaResource.java new file mode 100644 index 0000000000..bcae39cee1 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/DeltaResource.java @@ -0,0 +1,32 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol; + +/** + * A interface for resources in xDS, which can be updated by ADS delta stream + *
+ * This interface is design to unify the way of fetching data in delta stream + * in {@link org.apache.dubbo.registry.xds.util.PilotExchanger} + */ +public interface DeltaResource { + /** + * Get resource from delta stream + * + * @return the newest resource from stream + */ + T getResource(); +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java new file mode 100644 index 0000000000..25ec3fdab7 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol; + +import java.util.Set; +import java.util.function.Consumer; + +public interface XdsProtocol { + /** + * Gets all {@link T resource} by the specified resource name. + * For LDS, the {@param resourceNames} is ignored + * + * @param resourceNames specified resource name + * @return resources, null if request failed + */ + T getResource(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)} + */ + long observeResource(Set resourceNames, Consumer consumer); + + /** + * Update observed resource list in {@link XdsProtocol#observeResource(Set, Consumer)} + * + * @param request requestId returned by {@link XdsProtocol#observeResource(Set, Consumer)} + * @param resourceNames new resource name list to observe + */ + void updateObserve(long request, Set resourceNames); +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaEndpoint.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaEndpoint.java new file mode 100644 index 0000000000..16823aa567 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaEndpoint.java @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol.delta; + +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.registry.xds.util.protocol.DeltaResource; +import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; +import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +public class DeltaEndpoint implements DeltaResource { + private final Map> data = new ConcurrentHashMap<>(); + + public void addResource(String resourceName, Set endpoints) { + data.put(resourceName, endpoints); + } + + public void removeResource(Collection resourceName) { + if (CollectionUtils.isNotEmpty(resourceName)) { + resourceName.forEach(data::remove); + } + } + + @Override + public EndpointResult getResource() { + Set set = data.values().stream() + .flatMap(Set::stream) + .collect(Collectors.toSet()); + return new EndpointResult(set); + } + +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaListener.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaListener.java new file mode 100644 index 0000000000..8b66217276 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaListener.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.registry.xds.util.protocol.delta; + +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.registry.xds.util.protocol.DeltaResource; +import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; + +public class DeltaListener implements DeltaResource { + private final Map> data = new ConcurrentHashMap<>(); + + public void addResource(String resourceName, Set listeners) { + data.put(resourceName, listeners); + } + + public void removeResource(Collection resourceName) { + if (CollectionUtils.isNotEmpty(resourceName)) { + resourceName.forEach(data::remove); + } + } + + @Override + public ListenerResult getResource() { + Set set = data.values().stream() + .flatMap(Set::stream) + .collect(Collectors.toSet()); + return new ListenerResult(set); + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaRoute.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaRoute.java new file mode 100644 index 0000000000..71fdb479b3 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/delta/DeltaRoute.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.registry.xds.util.protocol.delta; + +import org.apache.dubbo.common.utils.CollectionUtils; +import org.apache.dubbo.registry.xds.util.protocol.DeltaResource; +import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; + +import java.util.Collection; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +public class DeltaRoute implements DeltaResource { + private final Map>> data = new ConcurrentHashMap<>(); + + public void addResource(String resourceName, Map> route) { + data.put(resourceName, route); + } + + public void removeResource(Collection resourceName) { + if (CollectionUtils.isNotEmpty(resourceName)) { + resourceName.forEach(data::remove); + } + } + + @Override + public RouteResult getResource() { + Map> result = new ConcurrentHashMap<>(); + data.values().forEach(result::putAll); + return new RouteResult(result); + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java new file mode 100644 index 0000000000..7476c00aab --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java @@ -0,0 +1,114 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol.impl; + +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.util.XdsChannel; +import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol; +import org.apache.dubbo.registry.xds.util.protocol.delta.DeltaEndpoint; +import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; +import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; + +import com.google.protobuf.Any; +import com.google.protobuf.InvalidProtocolBufferException; +import io.envoyproxy.envoy.config.core.v3.HealthStatus; +import io.envoyproxy.envoy.config.core.v3.Node; +import io.envoyproxy.envoy.config.core.v3.SocketAddress; +import io.envoyproxy.envoy.config.endpoint.v3.ClusterLoadAssignment; +import io.envoyproxy.envoy.config.endpoint.v3.LbEndpoint; +import io.envoyproxy.envoy.service.discovery.v3.DeltaDiscoveryResponse; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; +import io.envoyproxy.envoy.service.discovery.v3.Resource; + +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +public class EdsProtocol extends AbstractProtocol { + + private static final Logger logger = LoggerFactory.getLogger(LdsProtocol.class); + + public EdsProtocol(XdsChannel xdsChannel, Node node) { + super(xdsChannel, node); + } + + @Override + public String getTypeUrl() { + return "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"; + } + + @Override + protected EndpointResult decodeDiscoveryResponse(DiscoveryResponse response) { + if (getTypeUrl().equals(response.getTypeUrl())) { + Set set = response.getResourcesList().stream() + .map(EdsProtocol::unpackClusterLoadAssignment) + .filter(Objects::nonNull) + .flatMap((e) -> decodeResourceToEndpoint(e).stream()) + .collect(Collectors.toSet()); + return new EndpointResult(set); + } + return new EndpointResult(); + } + + @Override + protected DeltaEndpoint decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, DeltaEndpoint previous) { + DeltaEndpoint deltaEndpoint = previous; + if (deltaEndpoint == null) { + deltaEndpoint = new DeltaEndpoint(); + } + if (getTypeUrl().equals(response.getTypeUrl())) { + deltaEndpoint.removeResource(response.getRemovedResourcesList()); + for (Resource resource : response.getResourcesList()) { + ClusterLoadAssignment unpackedResource = unpackClusterLoadAssignment(resource.getResource()); + if (unpackedResource == null) { + continue; + } + deltaEndpoint.addResource(resource.getName(), decodeResourceToEndpoint(unpackedResource)); + } + } + return previous; + } + + private static Set decodeResourceToEndpoint(ClusterLoadAssignment resource) { + return resource.getEndpointsList().stream() + .flatMap((e) -> e.getLbEndpointsList().stream()) + .map(EdsProtocol::decodeLbEndpointToEndpoint) + .collect(Collectors.toSet()); + } + + private static Endpoint decodeLbEndpointToEndpoint(LbEndpoint lbEndpoint) { + Endpoint endpoint = new Endpoint(); + SocketAddress address = lbEndpoint.getEndpoint().getAddress().getSocketAddress(); + endpoint.setAddress(address.getAddress()); + endpoint.setPortValue(address.getPortValue()); + boolean healthy = HealthStatus.HEALTHY.equals(lbEndpoint.getHealthStatus()) || + HealthStatus.UNKNOWN.equals(lbEndpoint.getHealthStatus()); + endpoint.setHealthy(healthy); + endpoint.setWeight(lbEndpoint.getLoadBalancingWeight().getValue()); + return endpoint; + } + + private static ClusterLoadAssignment unpackClusterLoadAssignment(Any any) { + try { + return any.unpack(ClusterLoadAssignment.class); + } catch (InvalidProtocolBufferException e) { + logger.error("Error occur when decode xDS response.", e); + return null; + } + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java new file mode 100644 index 0000000000..3116efa850 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java @@ -0,0 +1,123 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol.impl; + +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.util.XdsChannel; +import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol; +import org.apache.dubbo.registry.xds.util.protocol.delta.DeltaListener; +import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; + +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.DeltaDiscoveryResponse; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; +import io.envoyproxy.envoy.service.discovery.v3.Resource; + +import java.util.Objects; +import java.util.Set; +import java.util.function.Consumer; +import java.util.stream.Collectors; + +public class LdsProtocol extends AbstractProtocol { + + private static final Logger logger = LoggerFactory.getLogger(LdsProtocol.class); + + public LdsProtocol(XdsChannel xdsChannel, Node node) { + super(xdsChannel, node); + } + + @Override + public String getTypeUrl() { + return "type.googleapis.com/envoy.config.listener.v3.Listener"; + } + + public ListenerResult getListeners() { + return getResource(null); + } + + public void observeListeners(Consumer consumer) { + observeResource(null,consumer); + } + + @Override + protected ListenerResult 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()); + return new ListenerResult(set); + } + return new ListenerResult(); + } + + @Override + protected DeltaListener decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, DeltaListener previous) { + DeltaListener deltaListener = previous; + if (deltaListener == null) { + deltaListener = new DeltaListener(); + } + if (getTypeUrl().equals(response.getTypeUrl())) { + deltaListener.removeResource(response.getRemovedResourcesList()); + for (Resource resource : response.getResourcesList()) { + Listener unpackedResource = unpackListener(resource.getResource()); + if (unpackedResource == null) { + continue; + } + deltaListener.addResource(resource.getName(), decodeResourceToListener(unpackedResource)); + } + } + return deltaListener; + } + + 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("Error occur when decode xDS response.", e); + return null; + } + } + + private static HttpConnectionManager unpackHttpConnectionManager(Any any) { + try { + return any.unpack(HttpConnectionManager.class); + } catch (InvalidProtocolBufferException e) { + logger.error("Error occur when decode xDS response.", e); + return null; + } + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java new file mode 100644 index 0000000000..a85b2eaf22 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol.impl; + +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; +import org.apache.dubbo.registry.xds.util.XdsChannel; +import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol; +import org.apache.dubbo.registry.xds.util.protocol.delta.DeltaRoute; +import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; + +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.service.discovery.v3.DeltaDiscoveryResponse; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; +import io.envoyproxy.envoy.service.discovery.v3.Resource; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.stream.Collectors; + +public class RdsProtocol extends AbstractProtocol { + + private static final Logger logger = LoggerFactory.getLogger(AbstractProtocol.class); + + public RdsProtocol(XdsChannel xdsChannel, Node node) { + super(xdsChannel, node); + } + + @Override + public String getTypeUrl() { + return "type.googleapis.com/envoy.config.route.v3.RouteConfiguration"; + } + + @Override + protected RouteResult decodeDiscoveryResponse(DiscoveryResponse response) { + if (getTypeUrl().equals(response.getTypeUrl())) { + Map> map = response.getResourcesList().stream() + .map(RdsProtocol::unpackRouteConfiguration) + .filter(Objects::nonNull) + .map(RdsProtocol::decodeResourceToListener) + .reduce((a, b) -> { + a.putAll(b); + return a; + }).orElse(new HashMap<>()); + return new RouteResult(map); + } + return new RouteResult(); + } + + @Override + protected DeltaRoute decodeDeltaDiscoveryResponse(DeltaDiscoveryResponse response, DeltaRoute previous) { + DeltaRoute deltaRoute = previous; + if (deltaRoute == null) { + deltaRoute = new DeltaRoute(); + } + if (getTypeUrl().equals(response.getTypeUrl())) { + deltaRoute.removeResource(response.getRemovedResourcesList()); + for (Resource resource : response.getResourcesList()) { + RouteConfiguration unpackedResource = unpackRouteConfiguration(resource.getResource()); + if (unpackedResource == null) { + continue; + } + deltaRoute.addResource(resource.getName(), decodeResourceToListener(unpackedResource)); + } + } + return deltaRoute; + } + + private static Map> decodeResourceToListener(RouteConfiguration resource) { + Map> map = new HashMap<>(); + resource.getVirtualHostsList() + .forEach(virtualHost -> { + Set cluster = virtualHost.getRoutesList().stream() + .map(Route::getRoute) + .map(RouteAction::getCluster) + .collect(Collectors.toSet()); + for (String domain : virtualHost.getDomainsList()) { + map.put(domain, cluster); + } + }); + return map; + } + + private static RouteConfiguration unpackRouteConfiguration(Any any) { + try { + return any.unpack(RouteConfiguration.class); + } catch (InvalidProtocolBufferException e) { + logger.error("Error occur when decode xDS response.", e); + return null; + } + } + + +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java new file mode 100644 index 0000000000..b966bda9ba --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/Endpoint.java @@ -0,0 +1,88 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol.message; + +import java.util.Objects; + +public class Endpoint { + private String address; + private int portValue; + private boolean healthy; + private int weight; + + 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; + } + + @Override + public String toString() { + return "Endpoint{" + + "address='" + address + '\'' + + ", portValue='" + portValue + '\'' + + ", healthy=" + healthy + + ", weight=" + weight + + '}'; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Endpoint endpoint = (Endpoint) o; + return healthy == endpoint.healthy && + weight == endpoint.weight && + Objects.equals(address, endpoint.address) && + Objects.equals(portValue, endpoint.portValue); + } + + @Override + public int hashCode() { + return Objects.hash(address, portValue, healthy, weight); + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/EndpointResult.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/EndpointResult.java new file mode 100644 index 0000000000..fc75eeeb57 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/EndpointResult.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol.message; + +import org.apache.dubbo.common.utils.ConcurrentHashSet; + +import java.util.Objects; +import java.util.Set; + +public class EndpointResult { + private Set endpoints; + + public EndpointResult() { + this.endpoints = new ConcurrentHashSet<>(); + } + + public EndpointResult(Set endpoints) { + this.endpoints = endpoints; + } + + public Set getEndpoints() { + return endpoints; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + EndpointResult that = (EndpointResult) o; + return Objects.equals(endpoints, that.endpoints); + } + + @Override + public int hashCode() { + return Objects.hash(endpoints); + } + + @Override + public String toString() { + return "EndpointResult{" + + "endpoints=" + endpoints + + '}'; + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/ListenerResult.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/ListenerResult.java new file mode 100644 index 0000000000..f6d6a9c32a --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/ListenerResult.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.registry.xds.util.protocol.message; + +import org.apache.dubbo.common.utils.ConcurrentHashSet; + +import java.util.Objects; +import java.util.Set; + +public class ListenerResult { + private Set routeConfigNames; + + public ListenerResult() { + this.routeConfigNames = new ConcurrentHashSet<>(); + } + + public ListenerResult(Set routeConfigNames) { + this.routeConfigNames = routeConfigNames; + } + + public Set getRouteConfigNames() { + return routeConfigNames; + } + + public void setRouteConfigNames(Set routeConfigNames) { + this.routeConfigNames = routeConfigNames; + } + + public void mergeRouteConfigNames(Set names) { + this.routeConfigNames.addAll(names); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + ListenerResult listenerResult = (ListenerResult) o; + return Objects.equals(routeConfigNames, listenerResult.routeConfigNames); + } + + @Override + public int hashCode() { + return Objects.hash(routeConfigNames); + } + + @Override + public String toString() { + return "ListenerResult{" + + "routeConfigNames=" + routeConfigNames + + '}'; + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java new file mode 100644 index 0000000000..3aac915f46 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java @@ -0,0 +1,73 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.dubbo.registry.xds.util.protocol.message; + +import org.apache.dubbo.common.utils.ConcurrentHashSet; + +import java.util.Collections; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +public class RouteResult { + private final Map> domainMap; + + public RouteResult() { + this.domainMap = new ConcurrentHashMap<>(); + } + + public RouteResult(Map> domainMap) { + this.domainMap = domainMap; + } + + public boolean isNotEmpty() { + return !domainMap.isEmpty(); + } + + public Set searchDomain(String domain) { + return domainMap.getOrDefault(domain, new ConcurrentHashSet<>()); + } + + public Set getDomains() { + return Collections.unmodifiableSet(domainMap.keySet()); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + RouteResult that = (RouteResult) o; + return Objects.equals(domainMap, that.domainMap); + } + + @Override + public int hashCode() { + return Objects.hash(domainMap); + } + + @Override + public String toString() { + return "RouteResult{" + + "domainMap=" + domainMap + + '}'; + } +} diff --git a/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory new file mode 100644 index 0000000000..453d3d30b7 --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.RegistryFactory @@ -0,0 +1 @@ +xds=org.apache.dubbo.registry.xds.XdsRegistryFactory \ No newline at end of file diff --git a/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscovery b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscovery new file mode 100644 index 0000000000..6fe7bd61db --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscovery @@ -0,0 +1 @@ +xds=org.apache.dubbo.registry.xds.XdsServiceDiscovery \ No newline at end of file diff --git a/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory new file mode 100644 index 0000000000..0335fa16eb --- /dev/null +++ b/dubbo-registry/dubbo-registry-xds/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.registry.client.ServiceDiscoveryFactory @@ -0,0 +1 @@ +xds=org.apache.dubbo.registry.xds.XdsServiceDiscoveryFactory \ No newline at end of file diff --git a/dubbo-registry/pom.xml b/dubbo-registry/pom.xml index db3ee35bf2..a652eb894c 100644 --- a/dubbo-registry/pom.xml +++ b/dubbo-registry/pom.xml @@ -37,5 +37,6 @@ dubbo-registry-multiple dubbo-registry-kubernetes dubbo-registry-dns + dubbo-registry-xds