Make xds router run (#11319)

Author: AlbumenJ

dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java

@@ -16,6 +16,12 @@
  */
 package org.apache.dubbo.registry.xds;
 
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
 import org.apache.dubbo.common.logger.LoggerFactory;
@@ -28,12 +34,6 @@
 import org.apache.dubbo.rpc.model.ApplicationModel;
 import org.apache.dubbo.rpc.model.ScopeModelUtil;
 
-import java.util.Collection;
-import java.util.Comparator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
-
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_INITIALIZE_XDS;
 import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_PARSING_XDS;
 
@@ -88,6 +88,7 @@ private List<ServiceInstance> changedToInstances(String serviceName, Collection<
             try {
                 DefaultServiceInstance serviceInstance = new DefaultServiceInstance(serviceName, endpoint.getAddress(), endpoint.getPortValue(), ScopeModelUtil.getApplicationModel(getUrl().getScopeModel()));
                 // fill metadata by SelfHostMetaServiceDiscovery, will be fetched by RPC request
+                serviceInstance.putExtendParam("clusterName", endpoint.getClusterName());
                 fillServiceInstance(serviceInstance);
                 instances.add(serviceInstance);
             } catch (Throwable t) {

dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/AdsObserver.java

@@ -0,0 +1,130 @@
+/*
+ * 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 java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
+import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol;
+import org.apache.dubbo.registry.xds.util.protocol.DeltaResource;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
+import io.envoyproxy.envoy.config.core.v3.Node;
+import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest;
+import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
+import io.grpc.stub.StreamObserver;
+
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS;
+
+public class AdsObserver {
+    private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AdsObserver.class);
+    private final ApplicationModel applicationModel;
+    private final URL url;
+    private final Node node;
+    private volatile XdsChannel xdsChannel;
+
+    private final Map<String, XdsListener> listeners = new ConcurrentHashMap<>();
+
+    protected StreamObserver<DiscoveryRequest> requestObserver;
+
+    private final Map<String, DiscoveryRequest> observedResources = new ConcurrentHashMap<>();
+
+    public AdsObserver(URL url, Node node) {
+        this.url = url;
+        this.node = node;
+        this.xdsChannel = new XdsChannel(url);
+        this.applicationModel = url.getOrDefaultApplicationModel();
+    }
+
+    public <T, S extends DeltaResource<T>> void addListener(AbstractProtocol<T, S> protocol) {
+        listeners.put(protocol.getTypeUrl(), protocol);
+    }
+
+    public void request(DiscoveryRequest discoveryRequest) {
+        if (requestObserver == null) {
+            requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(this));
+        }
+        requestObserver.onNext(discoveryRequest);
+        observedResources.put(discoveryRequest.getTypeUrl(), discoveryRequest);
+    }
+
+    private static class ResponseObserver implements StreamObserver<DiscoveryResponse> {
+        private AdsObserver adsObserver;
+
+        public ResponseObserver(AdsObserver adsObserver) {
+            this.adsObserver = adsObserver;
+        }
+
+        @Override
+        public void onNext(DiscoveryResponse discoveryResponse) {
+            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();
+    }
+}

dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java

@@ -18,13 +18,18 @@
 
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
+import java.util.stream.Collectors;
 
 import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository;
 import org.apache.dubbo.common.utils.CollectionUtils;
 import org.apache.dubbo.common.utils.ConcurrentHashSet;
 import org.apache.dubbo.registry.xds.util.protocol.AbstractProtocol;
@@ -38,8 +43,6 @@
 import org.apache.dubbo.rpc.cluster.router.xds.RdsVirtualHostListener;
 import org.apache.dubbo.rpc.model.ApplicationModel;
 
-import io.envoyproxy.envoy.config.route.v3.VirtualHost;
-
 public class PilotExchanger {
 
     protected final XdsChannel xdsChannel;
@@ -61,15 +64,18 @@ public class PilotExchanger {
 
     private final Map<String, Consumer<RdsVirtualHostListener>> rdsObserveConsumer = new ConcurrentHashMap<>();
 
-    private static  PilotExchanger GLOBAL_PILOT_EXCHANGER = null;
+    private static PilotExchanger GLOBAL_PILOT_EXCHANGER = null;
+
+    private final ApplicationModel applicationModel;
 
     protected PilotExchanger(URL url) {
         xdsChannel = new XdsChannel(url);
         int pollingTimeout = url.getParameter("pollingTimeout", 10);
-        ApplicationModel applicationModel = url.getOrDefaultApplicationModel();
-        this.ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout, applicationModel);
-        this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout, applicationModel);
-        this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build(), pollingTimeout, applicationModel);
+        this.applicationModel = url.getOrDefaultApplicationModel();
+        AdsObserver adsObserver = new AdsObserver(url, NodeBuilder.build());
+        this.ldsProtocol = new LdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
+        this.rdsProtocol = new RdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
+        this.edsProtocol = new EdsProtocol(adsObserver, NodeBuilder.build(), pollingTimeout);
 
         this.listenerResult = ldsProtocol.getListeners();
         this.routeResult = rdsProtocol.getResource(listenerResult.values().iterator().next().getRouteConfigNames());
@@ -96,26 +102,31 @@ protected PilotExchanger(URL url) {
     private void createRouteObserve() {
         rdsProtocol.observeResource(listenerResult.values().iterator().next().getRouteConfigNames(), (newResult) -> {
             // check if observed domain update ( will update endpoint observation )
+            List<String> domainsToUpdate = new LinkedList<>();
             domainObserveConsumer.forEach((domain, consumer) -> {
                 newResult.values().forEach(o -> {
                     Set<String> newRoute = o.searchDomain(domain);
-                    for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
+                    for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
                         if (!entry.getValue().searchDomain(domain).equals(newRoute)) {
                             // routers in observed domain has been updated
 //                    Long domainRequest = domainObserveRequest.get(domain);
                             // router list is empty when observeEndpoints() called and domainRequest has not been created yet
                             // create new observation
-                            doObserveEndpoints(domain);
+                            domainsToUpdate.add(domain);
+//                            doObserveEndpoints(domain);
                         }
                     }
                 });
             });
             routeResult = newResult;
+            ExecutorService executorService = applicationModel.getFrameworkModel().getBeanFactory()
+                .getBean(FrameworkExecutorRepository.class).getSharedExecutor();
+            executorService.submit(() -> domainsToUpdate.forEach(this::doObserveEndpoints));
         }, false);
     }
 
     public static PilotExchanger initialize(URL url) {
-        synchronized (PilotExchanger.class){
+        synchronized (PilotExchanger.class) {
             if (GLOBAL_PILOT_EXCHANGER != null) {
                 return GLOBAL_PILOT_EXCHANGER;
             }
@@ -140,15 +151,15 @@ public void destroy() {
 
     public Set<String> getServices() {
         Set<String> domains = new HashSet<>();
-        for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
+        for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
             domains.addAll(entry.getValue().getDomains());
         }
         return domains;
     }
 
     public Set<Endpoint> getEndpoints(String domain) {
         Set<Endpoint> endpoints = new HashSet<>();
-        for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
+        for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
             Set<String> cluster = entry.getValue().searchDomain(domain);
             if (CollectionUtils.isNotEmpty(cluster)) {
                 Map<String, EndpointResult> endpointResultList = edsProtocol.getResource(cluster);
@@ -176,19 +187,21 @@ public void observeEndpoints(String domain, Consumer<Set<Endpoint>> consumer) {
     }
 
     private void doObserveEndpoints(String domain) {
-        for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
+        for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
             Set<String> router = entry.getValue().searchDomain(domain);
             // if router is empty, do nothing
             // observation will be created when RDS updates
             if (CollectionUtils.isNotEmpty(router)) {
                 edsProtocol.observeResource(
                     router,
                     (endpointResultMap) -> {
-                        endpointResultMap.forEach((k, v) -> {
-                            // notify consumers
-                            domainObserveConsumer.get(domain).forEach(
-                                consumer1 -> consumer1.accept(v.getEndpoints()));
-                        });
+                        Set<Endpoint> endpoints = endpointResultMap.values().stream()
+                            .map(EndpointResult::getEndpoints)
+                            .flatMap(Set::stream)
+                            .collect(Collectors.toSet());
+                        for (Consumer<Set<Endpoint>> consumer : domainObserveConsumer.get(domain)) {
+                            consumer.accept(endpoints);
+                        }
                     }, false);
                 domainObserveRequest.add(domain);
             }
@@ -201,18 +214,28 @@ public void unObserveEndpoints(String domain, Consumer<Set<Endpoint>> consumer)
         domainObserveRequest.remove(domain);
     }
 
-    public VirtualHost getVirtualHost(String domain) {
-        for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
-            if (entry.getValue().searchVirtualHost(domain) != null) {
-                return entry.getValue().searchVirtualHost(domain);
-            }
-        }
-        return null;
+    public void observeEds(Set<String> clusterNames, Consumer<Map<String, EndpointResult>> consumer) {
+        edsProtocol.observeResource(clusterNames, consumer, false);
     }
 
-    public void unObserveRds(String domain) {
-        for (Map.Entry<String, RouteResult> entry : routeResult.entrySet()) {
-            entry.getValue().removeVirtualHost(domain);
-        }
+    public void unObserveEds(Set<String> clusterNames, Consumer<Map<String, EndpointResult>> consumer) {
+        edsProtocol.unobserveResource(clusterNames, consumer);
+    }
+
+    public void observeRds(Set<String> clusterNames, Consumer<Map<String, RouteResult>> consumer) {
+        rdsProtocol.observeResource(clusterNames, consumer, false);
+    }
+
+    public void unObserveRds(Set<String> clusterNames, Consumer<Map<String, RouteResult>> consumer) {
+        rdsProtocol.unobserveResource(clusterNames, consumer);
     }
+
+    public void observeLds(Consumer<Map<String, ListenerResult>> consumer) {
+        ldsProtocol.observeResource(Collections.singleton(AbstractProtocol.emptyResourceName), consumer, false);
+    }
+
+    public void unObserveLds(Consumer<Map<String, ListenerResult>> consumer) {
+        ldsProtocol.unobserveResource(Collections.singleton(AbstractProtocol.emptyResourceName), consumer);
+    }
+
 }

dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/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.registry.xds.util;
+
+import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse;
+
+public interface XdsListener {
+    void process(DiscoveryResponse discoveryResponse);
+}