diff --git a/dubbo-dependencies-bom/pom.xml b/dubbo-dependencies-bom/pom.xml index 089a617797..0bb37e2663 100644 --- a/dubbo-dependencies-bom/pom.xml +++ b/dubbo-dependencies-bom/pom.xml @@ -143,7 +143,7 @@ 8.5.78 0.5.3 2.1.2 - 1.47.0 + 1.41.0 0.8.1 1.2.1 diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java index dbb30e7586..ce84aab722 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/PilotExchanger.java @@ -26,85 +26,84 @@ import org.apache.dubbo.registry.xds.util.protocol.message.Endpoint; import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; +import org.apache.dubbo.rpc.model.ApplicationModel; import java.util.Collections; +import java.util.HashSet; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Consumer; public class PilotExchanger { - private final XdsChannel xdsChannel; + protected final XdsChannel xdsChannel; - private final RdsProtocol rdsProtocol; + protected final LdsProtocol ldsProtocol; - private final EdsProtocol edsProtocol; + protected final RdsProtocol rdsProtocol; - private ListenerResult listenerResult; + protected final EdsProtocol edsProtocol; - private RouteResult routeResult; + protected Map listenerResult; - private final AtomicLong observeRouteRequest = new AtomicLong(-1); + protected Map routeResult; - private final Map domainObserveRequest = new ConcurrentHashMap<>(); + private final AtomicBoolean isRdsObserve = new AtomicBoolean(false); + private final HashSet domainObserveRequest = new HashSet<>(); private final Map>>> domainObserveConsumer = new ConcurrentHashMap<>(); - private PilotExchanger(URL url) { + protected PilotExchanger(URL url) { xdsChannel = new XdsChannel(url); - int pollingPoolSize = url.getParameter("pollingPoolSize", 10); int pollingTimeout = url.getParameter("pollingTimeout", 10); - LdsProtocol ldsProtocol = new LdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout); - this.rdsProtocol = new RdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout); - this.edsProtocol = new EdsProtocol(xdsChannel, NodeBuilder.build(), pollingPoolSize, pollingTimeout); + 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.listenerResult = ldsProtocol.getListeners(); - this.routeResult = rdsProtocol.getResource(listenerResult.getRouteConfigNames()); + this.routeResult = rdsProtocol.getResource(listenerResult.values().iterator().next().getRouteConfigNames()); // Observer RDS update - if (CollectionUtils.isNotEmpty(listenerResult.getRouteConfigNames())) { - this.observeRouteRequest.set(createRouteObserve()); + if (CollectionUtils.isNotEmpty(listenerResult.values().iterator().next().getRouteConfigNames())) { + createRouteObserve(); + isRdsObserve.set(true); } + // Observe LDS updated ldsProtocol.observeListeners((newListener) -> { // update local cache if (!newListener.equals(listenerResult)) { this.listenerResult = newListener; // update RDS observation - synchronized (observeRouteRequest) { - if (observeRouteRequest.get() == -1) { - this.observeRouteRequest.set(createRouteObserve()); - } else { - rdsProtocol.updateObserve(observeRouteRequest.get(), newListener.getRouteConfigNames()); - } + if (isRdsObserve.get()) { + createRouteObserve(); } } }); } - private long createRouteObserve() { - return rdsProtocol.observeResource(listenerResult.getRouteConfigNames(), (newResult) -> { + private void createRouteObserve() { + rdsProtocol.observeResource(listenerResult.values().iterator().next().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); + newResult.values().forEach(o -> { + Set newRoute = o.searchDomain(domain); + for (Map.Entry entry: routeResult.entrySet()) { + if (!entry.getValue().searchDomain(domain).equals(newRoute)) { + // routers in observed domain has been updated +// Long domainRequest = domainObserveRequest.get(domain); + // router list is empty when observeEndpoints() called and domainRequest has not been created yet + // create new observation + doObserveEndpoints(domain); + } } - } + }); }); - // update local cache routeResult = newResult; - }); + }, false); } public static PilotExchanger initialize(URL url) { @@ -116,17 +115,25 @@ public class PilotExchanger { } public Set getServices() { - return routeResult.getDomains(); + Set domains = new HashSet<>(); + for (Map.Entry entry: routeResult.entrySet()) { + domains.addAll(entry.getValue().getDomains()); + } + return domains; } 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(); + Set endpoints = new HashSet<>(); + for (Map.Entry entry: routeResult.entrySet()) { + Set cluster = entry.getValue().searchDomain(domain); + if (CollectionUtils.isNotEmpty(cluster)) { + Map endpointResultList = edsProtocol.getResource(cluster); + endpointResultList.forEach((k, v) -> endpoints.addAll(v.getEndpoints())); + } else { + return Collections.emptySet(); + } } + return endpoints; } public void observeEndpoints(String domain, Consumer> consumer) { @@ -139,24 +146,29 @@ public class PilotExchanger { v.add(consumer); return v; }); - if (!domainObserveRequest.containsKey(domain)) { + if (!domainObserveRequest.contains(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 = + for (Map.Entry entry: routeResult.entrySet()) { + Set router = entry.getValue().searchDomain(domain); + // if router is empty, do nothing + // observation will be created when RDS updates + if (CollectionUtils.isNotEmpty(router)) { edsProtocol.observeResource( router, - endpointResult -> - // notify consumers - domainObserveConsumer.get(domain).forEach( - consumer1 -> consumer1.accept(endpointResult.getEndpoints()))); - domainObserveRequest.put(domain, endpointRequest); + (endpointResultMap) -> { + endpointResultMap.forEach((k, v) -> { + // notify consumers + domainObserveConsumer.get(domain).forEach( + consumer1 -> consumer1.accept(v.getEndpoints())); + }); + }, false); + domainObserveRequest.add(domain); + } } + } } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java index e58eed0172..d4e069a364 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java @@ -49,14 +49,25 @@ public class XdsChannel { 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; - protected XdsChannel(URL url) { + 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))){ diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java index fdb98c3410..6fcd422635 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java @@ -16,76 +16,77 @@ */ package org.apache.dubbo.registry.xds.util.protocol; -import org.apache.dubbo.common.logger.ErrorTypeAwareLogger; -import org.apache.dubbo.common.logger.LoggerFactory; -import org.apache.dubbo.common.utils.NamedThreadFactory; -import org.apache.dubbo.registry.xds.util.XdsChannel; 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 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.XdsChannel; +import org.apache.dubbo.rpc.model.ApplicationModel; +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.ExecutionException; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Consumer; +import java.util.stream.Collectors; import static org.apache.dubbo.common.constants.LoggerCodeConstants.REGISTRY_ERROR_REQUEST_XDS; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_REQUEST; +import static org.apache.dubbo.common.constants.LoggerCodeConstants.INTERNAL_INTERRUPTED; public abstract class AbstractProtocol> implements XdsProtocol { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(AbstractProtocol.class); - protected final XdsChannel xdsChannel; + protected XdsChannel xdsChannel; protected final Node node; - /** - * Store Request Parameter ( resourceNames ) - * K - requestId, V - resourceNames - */ - protected final Map> requestParam = new ConcurrentHashMap<>(); + private final int checkInterval; - /** - * Store ADS Request Observer ( StreamObserver in Streaming Request ) - * K - requestId, V - StreamObserver - */ - private final Map> requestObserverMap = new ConcurrentHashMap<>(); + protected final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); - /** - * Store Delta-ADS Request Observer ( StreamObserver in Streaming Request ) - * K - requestId, V - StreamObserver - */ - private final Map> observeScheduledMap = new ConcurrentHashMap<>(); + protected final ReentrantReadWriteLock.ReadLock readLock = lock.readLock(); - /** - * Store CompletableFuture for Request ( used to fetch async result in ResponseObserver ) - * K - requestId, V - CompletableFuture - */ - private final Map> streamResult = new ConcurrentHashMap<>(); + protected final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock(); - private final ScheduledExecutorService pollingExecutor; + protected Set observeResourcesName; - private final int pollingTimeout; + public static final String emptyResourceName = "emptyResourcesName"; + private final ReentrantLock resourceLock = new ReentrantLock(); - protected final static AtomicLong requestId = new AtomicLong(0); + protected Map, List>>> consumerObserveMap = new ConcurrentHashMap<>(); - public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { + public Map, List>>> getConsumerObserveMap() { + return consumerObserveMap; + } + private final ApplicationModel applicationModel; + + public AbstractProtocol(XdsChannel xdsChannel, Node node, int checkInterval, ApplicationModel applicationModel) { this.xdsChannel = xdsChannel; this.node = node; - this.pollingExecutor = new ScheduledThreadPoolExecutor(pollingPoolSize, new NamedThreadFactory("Dubbo-registry-xds")); - this.pollingTimeout = pollingTimeout; + this.checkInterval = checkInterval; + this.applicationModel = applicationModel; } + protected Map resourcesMap = new ConcurrentHashMap<>(); + + protected StreamObserver requestObserver; + /** * Abstract method to obtain Type-URL from sub-class * @@ -93,103 +94,126 @@ public abstract class AbstractProtocol> implements */ 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 T getResource(Set resourceNames) { - long request = requestId.getAndIncrement(); + public Map getResource(Set resourceNames) { resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames; - // 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(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel."); - return null; - } finally { - // close observer - //requestObserver.onCompleted(); - - // remove temp - streamResult.remove(request); - requestObserverMap.remove(request); - requestParam.remove(request); + if (!resourceNames.isEmpty() && isCacheExistResource(resourceNames)) { + return getResourceFromCache(resourceNames); + } else { + return getResourceFromRemote(resourceNames); } } - @Override - public long observeResource(Set resourceNames, Consumer consumer) { - long request = requestId.getAndIncrement(); - resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames; - - // Store Request Parameter, which will be used for ACK - requestParam.put(request, resourceNames); - - // call once for full data - consumer.accept(getResource(resourceNames)); - - // channel reused - StreamObserver requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request)); - requestObserverMap.put(request, requestObserver); - - ScheduledFuture scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> { - try { - // origin request, may changed by updateObserve - Set names = requestParam.get(request); - - // use future to get async result, future complete on StreamObserver onNext - CompletableFuture future = new CompletableFuture<>(); - streamResult.put(request, future); - - // observer reused - StreamObserver observer = requestObserverMap.get(request); - - if (observer == null) { - observer = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request)); - requestObserverMap.put(request, observer); - } - - // send request to control panel - observer.onNext(buildDiscoveryRequest(names)); - - try { - // get result - consumer.accept(future.get()); - } catch (InterruptedException | ExecutionException e) { - logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error occur when request control panel."); - } finally { - // close observer - //requestObserver.onCompleted(); - - // remove temp - streamResult.remove(request); - } - } catch (Throwable t) { - logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "Error when requesting observe data. Type: " + getTypeUrl(), t); - } - }, pollingTimeout, pollingTimeout, TimeUnit.SECONDS); - - observeScheduledMap.put(request, scheduledFuture); - - return request; + private Map getResourceFromCache(Set resourceNames) { + return resourceNames.stream() + .collect(Collectors.toMap(k -> k, this::getCacheResource)); } - @Override - public void updateObserve(long request, Set resourceNames) { - // send difference in resourceNames - requestParam.put(request, resourceNames); + public Map getResourceFromRemote(Set resourceNames) { + try { + resourceLock.lock(); + CompletableFuture> future = new CompletableFuture<>(); + if (requestObserver == null) { + requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver()); + } + observeResourcesName = resourceNames; + Set consumerObserveResourceNames = new HashSet<>(); + if (resourceNames.isEmpty()) { + consumerObserveResourceNames.add(emptyResourceName); + } else { + consumerObserveResourceNames = resourceNames; + } + + Consumer> futureConsumer = future::complete; + try { + writeLock.lock(); + consumerObserveMap.computeIfAbsent(consumerObserveResourceNames, key -> new ArrayList<>()) + .add(futureConsumer); + } finally { + writeLock.unlock(); + } + + Set resourceNamesToObserve = new HashSet<>(resourceNames); + resourceNamesToObserve.addAll(resourcesMap.keySet()); + requestObserver.onNext(buildDiscoveryRequest(resourceNamesToObserve)); + logger.info("Send xDS Observe request to remote. Resource count: " + resourceNamesToObserve.size() + ". Resource Type: " + getTypeUrl()); + + try { + Map result = future.get(); + + try { + writeLock.lock(); + consumerObserveMap.get(consumerObserveResourceNames).removeIf(o -> o.equals(futureConsumer)); + } finally { + writeLock.unlock(); + } + + return result; + } catch (InterruptedException e) { + logger.error(INTERNAL_INTERRUPTED, "", "", "InterruptedException occur when request control panel. error=", e); + Thread.currentThread().interrupt(); + } catch (Exception e) { + logger.error(PROTOCOL_FAILED_REQUEST, "", "", "Error occur when request control panel. error=", e); + } + } finally { + resourceLock.unlock(); + } + return Collections.emptyMap(); + } + + public void observeResource(Set resourceNames, Consumer> consumer, boolean isReConnect) { + resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames; + // call once for full data + if (!isReConnect) { + try { + writeLock.lock(); + Set consumerObserveResourceNames = new HashSet<>(); + if (resourceNames.isEmpty()) { + consumerObserveResourceNames.add(emptyResourceName); + } else { + consumerObserveResourceNames = resourceNames; + } + consumerObserveMap.compute(consumerObserveResourceNames, (k, v) -> { + if (v == null) { + v = new ArrayList<>(); + } + // support multi-consumer + v.add(consumer); + return v; + }); + consumer.accept(getResource(resourceNames)); + } finally { + writeLock.unlock(); + } + } + try { + writeLock.lock(); + this.observeResourcesName = consumerObserveMap.keySet() + .stream().flatMap(Set::stream).collect(Collectors.toSet()); + } finally { + writeLock.unlock(); + } } protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames) { @@ -210,51 +234,98 @@ public abstract class AbstractProtocol> implements .build(); } - protected abstract T decodeDiscoveryResponse(DiscoveryResponse response); + protected abstract Map decodeDiscoveryResponse(DiscoveryResponse response); - private class ResponseObserver implements StreamObserver { - private final long requestId; + public class ResponseObserver implements StreamObserver { + + public ResponseObserver() { - public ResponseObserver(long requestId) { - this.requestId = requestId; } @Override public void onNext(DiscoveryResponse value) { - logger.info("receive notification from xds server, type: " + getTypeUrl() + " requestId: " + requestId); - T result = decodeDiscoveryResponse(value); - StreamObserver observer = requestObserverMap.get(requestId); - if (observer == null) { + Map newResult = decodeDiscoveryResponse(value); + Map oldResource = resourcesMap; + discoveryResponseListener(oldResource, newResult); + resourcesMap = newResult; + requestObserver.onNext(buildDiscoveryRequest(Collections.emptySet(), value)); + } + + public 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; } - observer.onNext(buildDiscoveryRequest(Collections.emptySet(), value)); - returnResult(result); + + 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(); + } } @Override public void onError(Throwable t) { logger.error(REGISTRY_ERROR_REQUEST_XDS, "", "", "xDS Client received error message! detail:", t); - clear(); - returnResult(null); - } - - private void returnResult(T result) { - CompletableFuture future = streamResult.get(requestId); - if (future == null) { - return; + if (consumerObserveMap.size() != 0) { + triggerReConnectTask(); } - future.complete(result); } @Override public void onCompleted() { - logger.info("xDS Client completed, requestId: " + requestId); - clear(); - } - - private void clear() { - requestObserverMap.remove(requestId); + logger.info("xDS Client completed"); } } + private void triggerReConnectTask() { + AtomicBoolean isConnectFail = new AtomicBoolean(false); + ScheduledExecutorService scheduledFuture = applicationModel.getFrameworkModel().getBeanFactory() + .getBean(FrameworkExecutorRepository.class).getSharedScheduledExecutor(); + scheduledFuture.scheduleAtFixedRate(() -> { + xdsChannel = new XdsChannel(xdsChannel.getUrl()); + if (xdsChannel.getChannel() != null) { + Set reConnectResourcesNames; + try { + readLock.lock(); + reConnectResourcesNames = consumerObserveMap.keySet() + .stream() + .flatMap(Set::stream) + .collect(Collectors.toSet()); + } finally { + readLock.unlock(); + } + getResourceFromRemote(reConnectResourcesNames); + if (isConnectFail.get()) { + scheduledFuture.shutdown(); + } + } else { + isConnectFail.set(true); + } + }, checkInterval, checkInterval, TimeUnit.SECONDS); + } } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java index 847ec04f03..ce1a088140 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java @@ -16,6 +16,7 @@ */ package org.apache.dubbo.registry.xds.util.protocol; +import java.util.Map; import java.util.Set; import java.util.function.Consumer; @@ -27,7 +28,7 @@ public interface XdsProtocol { * @param resourceNames specified resource name * @return resources, null if request failed */ - T getResource(Set resourceNames); + Map getResource(Set resourceNames); /** * Add a observer resource with {@link Consumer} @@ -36,13 +37,5 @@ public interface XdsProtocol { * @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); + void observeResource(Set resourceNames, Consumer> consumer, boolean isReConnect); } diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java index b171a5406a..5da12d9a88 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java @@ -23,6 +23,7 @@ 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 org.apache.dubbo.rpc.model.ApplicationModel; import com.google.protobuf.Any; import com.google.protobuf.InvalidProtocolBufferException; @@ -33,6 +34,8 @@ 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 java.util.HashMap; +import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.stream.Collectors; @@ -43,8 +46,8 @@ public class EdsProtocol extends AbstractProtocol private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(EdsProtocol.class); - public EdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { - super(xdsChannel, node, pollingPoolSize, pollingTimeout); + public EdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { + super(xdsChannel, node, pollingTimeout, applicationModel); } @Override @@ -52,24 +55,24 @@ public class EdsProtocol extends AbstractProtocol return "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment"; } + @Override - protected EndpointResult decodeDiscoveryResponse(DiscoveryResponse response) { + protected Map decodeDiscoveryResponse(DiscoveryResponse response) { if (getTypeUrl().equals(response.getTypeUrl())) { - Set set = response.getResourcesList().stream() + return response.getResourcesList().stream() .map(EdsProtocol::unpackClusterLoadAssignment) .filter(Objects::nonNull) - .flatMap((e) -> decodeResourceToEndpoint(e).stream()) - .collect(Collectors.toSet()); - return new EndpointResult(set); + .collect(Collectors.toConcurrentMap(ClusterLoadAssignment::getClusterName, this::decodeResourceToEndpoint)); } - return new EndpointResult(); + return new HashMap<>(); } - private static Set decodeResourceToEndpoint(ClusterLoadAssignment resource) { - return resource.getEndpointsList().stream() - .flatMap((e) -> e.getLbEndpointsList().stream()) + private EndpointResult decodeResourceToEndpoint(ClusterLoadAssignment resource) { + Set endpoints = resource.getEndpointsList().stream() + .flatMap(e -> e.getLbEndpointsList().stream()) .map(EdsProtocol::decodeLbEndpointToEndpoint) .collect(Collectors.toSet()); + return new EndpointResult(endpoints); } private static Endpoint decodeLbEndpointToEndpoint(LbEndpoint lbEndpoint) { diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java index 8566e6e027..e1b9ce6bb7 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java @@ -22,6 +22,7 @@ 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 org.apache.dubbo.rpc.model.ApplicationModel; import com.google.protobuf.Any; import com.google.protobuf.InvalidProtocolBufferException; @@ -33,19 +34,21 @@ import io.envoyproxy.envoy.extensions.filters.network.http_connection_manager.v3 import io.envoyproxy.envoy.service.discovery.v3.DiscoveryResponse; import java.util.Collections; +import java.util.HashMap; +import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Consumer; import java.util.stream.Collectors; 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 LdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { - super(xdsChannel, node, pollingPoolSize, pollingTimeout); + public LdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { + super(xdsChannel, node, pollingTimeout, applicationModel); } @Override @@ -53,30 +56,32 @@ public class LdsProtocol extends AbstractProtocol return "type.googleapis.com/envoy.config.listener.v3.Listener"; } - public ListenerResult getListeners() { + public Map getListeners() { return getResource(null); } - public void observeListeners(Consumer consumer) { - observeResource(Collections.emptySet(), consumer); + public void observeListeners(Consumer> consumer) { + observeResource(Collections.emptySet(), consumer, false); } @Override - protected ListenerResult decodeDiscoveryResponse(DiscoveryResponse response) { + 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()) + .flatMap(e -> decodeResourceToListener(e).stream()) .collect(Collectors.toSet()); - return new ListenerResult(set); + Map listenerDecodeResult = new ConcurrentHashMap<>(); + listenerDecodeResult.put(emptyResourceName, new ListenerResult(set)); + return listenerDecodeResult; } - return new ListenerResult(); + return new HashMap<>(); } private Set decodeResourceToListener(Listener resource) { return resource.getFilterChainsList().stream() - .flatMap((e) -> e.getFiltersList().stream()) + .flatMap(e -> e.getFiltersList().stream()) .map(Filter::getTypedConfig) .map(LdsProtocol::unpackHttpConnectionManager) .filter(Objects::nonNull) diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java index 653b53e648..900fb22efe 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java @@ -22,6 +22,7 @@ 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 org.apache.dubbo.rpc.model.ApplicationModel; import com.google.protobuf.Any; import com.google.protobuf.InvalidProtocolBufferException; @@ -43,8 +44,8 @@ public class RdsProtocol extends AbstractProtocol { private static final ErrorTypeAwareLogger logger = LoggerFactory.getErrorTypeAwareLogger(RdsProtocol.class); - public RdsProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) { - super(xdsChannel, node, pollingPoolSize, pollingTimeout); + public RdsProtocol(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { + super(xdsChannel, node, pollingTimeout, applicationModel); } @Override @@ -53,22 +54,17 @@ public class RdsProtocol extends AbstractProtocol { } @Override - protected RouteResult decodeDiscoveryResponse(DiscoveryResponse response) { + protected Map decodeDiscoveryResponse(DiscoveryResponse response) { if (getTypeUrl().equals(response.getTypeUrl())) { - Map> map = response.getResourcesList().stream() + return 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); + .collect(Collectors.toConcurrentMap(RouteConfiguration::getName, this::decodeResourceToListener)); } - return new RouteResult(); + return new HashMap<>(); } - private static Map> decodeResourceToListener(RouteConfiguration resource) { + private RouteResult decodeResourceToListener(RouteConfiguration resource) { Map> map = new HashMap<>(); resource.getVirtualHostsList() .forEach(virtualHost -> { @@ -80,7 +76,7 @@ public class RdsProtocol extends AbstractProtocol { map.put(domain, cluster); } }); - return map; + return new RouteResult(map); } private static RouteConfiguration unpackRouteConfiguration(Any any) { diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java index 2355b3dffa..d3166c5f17 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java @@ -35,6 +35,11 @@ public class RouteResult { this.domainMap = domainMap; } + + public Map> getDomainMap() { + return domainMap; + } + public boolean isNotEmpty() { return !domainMap.isEmpty(); } diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java new file mode 100644 index 0000000000..56e0afb1ce --- /dev/null +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java @@ -0,0 +1,542 @@ +/* + * 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 io.envoyproxy.envoy.config.core.v3.Node; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; +import io.grpc.stub.StreamObserver; +import org.apache.dubbo.common.URL; +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.message.Endpoint; +import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; +import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; +import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; +import org.apache.dubbo.rpc.model.ApplicationModel; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.stream.Collectors; + +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.times; + +public class DsProtocolTest { + private XdsChannel xdsChannel; + + private LdsProtocolMock ldsProtocolMock; + private RdsProtocolMock rdsProtocolMock; + + private EdsProtocolMock edsProtocolMock; + + private Map listenerResult; + + private Set routeConfigNames; + + private Set cluster; + private Map routeResult; + + private Set endpoints; + private Map endpointResult; + + private Map> domainMap; + + private ApplicationModel applicationModel; + private URL url; + + private Node node; + + public DsProtocolTest() { + } + + @BeforeEach + public void setUp() { + this.url = spy(URL.valueOf("xds://istiod.istio-system.svc:15012?secure=plaintext")); + this.node = mock(Node.class); + + this.applicationModel = ApplicationModel.defaultModel(); + xdsChannel = mock(XdsChannel.class); + when(xdsChannel.getUrl()).thenReturn(url); + + this.ldsProtocolMock = spy(new LdsProtocolMock(xdsChannel, node, 1, applicationModel)); + this.rdsProtocolMock = spy(new RdsProtocolMock(xdsChannel, node, 1, applicationModel)); + this.edsProtocolMock = spy(new EdsProtocolMock(xdsChannel, node, 1, applicationModel)); + + // mock lister result + this.routeConfigNames = new HashSet<>(); + routeConfigNames.add("15014"); + Map listenerResults = new HashMap(); + listenerResults.put(ldsProtocolMock.emptyResourceName,new ListenerResult(routeConfigNames)); + this.listenerResult = spy(listenerResults); + + // mock route result + this.domainMap = new HashMap<>(); + Set domainValue = new HashSet<>(); + domainValue.add("outbound|15014||istiod.istio-system.svc.cluster.local"); + domainMap.put("istiod.istio-system.svc", domainValue); + Map routeResults = new HashMap(); + routeResults.put("15014", new RouteResult(domainMap)); + this.routeResult = routeResults; + + // mock eds result + Set cluster = new HashSet<>(); + cluster.add("dubbo-samples-provider"); + this.cluster = cluster; + Endpoint endpoint = new Endpoint(); + endpoint.setWeight(1); + endpoint.setHealthy(true); + endpoint.setPortValue(50051); + endpoint.setAddress("10.1.1.67"); + this.endpoints = new HashSet<>(); + endpoints.add(endpoint); + Map endpointResults = new HashMap(); + endpointResults.put("dubbo-samples-provider" ,new EndpointResult(endpoints)); + this.endpointResult = endpointResults; +// mockedStatic.close(); + } + + + @Test + void testGetResource() { + StreamObserver requestStreamObserver = mock(StreamObserver.class); + + // mock lds getResource + when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); + MockedConstruction ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { + when(mock.get()).thenReturn(listenerResult); + }); + Map ldsResult = ldsProtocolMock.getResource(null); + + Assertions.assertNotNull(ldsResult); + verify(ldsProtocolMock, times(0)).isCacheExistResource(null); + ldsMocked.close(); + + // mock rds getResource + MockedConstruction rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { + when(mock.get()).thenReturn(null); + }); + + Map rdsResult = rdsProtocolMock.getResource(routeConfigNames); + + Assertions.assertNull(rdsResult); + Assertions.assertFalse(rdsProtocolMock.isCacheExistResource(routeConfigNames)); + + rdsProtocolMock.getResourcesMap().putAll(routeResult); + rdsResult = rdsProtocolMock.getResource(routeConfigNames); + Assertions.assertNotNull(rdsResult); + Assertions.assertTrue(rdsProtocolMock.isCacheExistResource(routeConfigNames)); + Map newRdsResult = routeConfigNames.stream().collect(Collectors.toMap(k -> k, v -> rdsProtocolMock.getCacheResource(v))); + Assertions.assertEquals(newRdsResult, rdsResult); + rdsMocked.close(); + + //mock eds getResource + MockedConstruction edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { + when(mock.get()).thenReturn(null); + }); + + Map edsResult = edsProtocolMock.getResource(cluster); + verify(edsProtocolMock, times(1)).isCacheExistResource(cluster); + Assertions.assertNull(edsResult); + Assertions.assertFalse(edsProtocolMock.isCacheExistResource(cluster)); + edsProtocolMock.getResourcesMap().put("dubbo-samples-provider", new EndpointResult(endpoints)); + edsResult = edsProtocolMock.getResource(cluster); + Assertions.assertNotNull(edsResult); + Assertions.assertTrue(edsProtocolMock.isCacheExistResource(cluster)); + + Map newEdsResult = cluster.stream().collect(Collectors.toMap(k -> k, v -> edsProtocolMock.getCacheResource(v))); + Assertions.assertEquals(newEdsResult, endpointResult); + edsMocked.close(); + } + + @Test + void observeDsReConnect() { + StreamObserver requestStreamObserver = mock(StreamObserver.class); + + // mock lds reconnect + when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); + doNothing().when(requestStreamObserver).onNext(any()); + + Map, List>>> ldsMap = new HashMap<>(); + AtomicBoolean isLdsReConnect = new AtomicBoolean(false); + CountDownLatch ldsCountDownLatch = new CountDownLatch(1); +// // support multi-consumer + Consumer> consumer = (listenerResult) -> { + isLdsReConnect.set(true); + ldsCountDownLatch.countDown(); + }; + ldsMap.compute(new HashSet<>(), (k, v) -> { + if (v == null) { + v = new ArrayList<>(); + } + v.add(consumer); + // support multi-consumer + return v; + }); + Assertions.assertFalse(isLdsReConnect.get()); + ldsProtocolMock.setConsumerObserveMap(ldsMap); + try { + triggerConsumerObserveMapListener(listenerResult, ldsProtocolMock); + ldsProtocolMock.getResponseObserve().onError(new RuntimeException()); + ldsCountDownLatch.await(); + Assertions.assertTrue(isLdsReConnect.get()); + } catch (Exception e) { + Assertions.assertTrue(isLdsReConnect.get()); + } + + //mock rds reconnnect + Map, List>>> rdsMap = new HashMap<>(); + AtomicBoolean isRdsReConnect = new AtomicBoolean(false); + CountDownLatch rdsCountDownLatch = new CountDownLatch(1); + // support multi-consumer + Consumer> rdsConsumer = (routeResult) -> { + isRdsReConnect.set(true); + rdsCountDownLatch.countDown(); + }; + rdsMap.compute(new HashSet<>(), (k, v) -> { + if (v == null) { + v = new ArrayList<>(); + } + v.add(rdsConsumer); + // support multi-consumer + return v; + }); + Assertions.assertFalse(isRdsReConnect.get()); + rdsProtocolMock.setConsumerObserveMap(rdsMap); + try { + triggerConsumerObserveMapListener(routeResult, rdsProtocolMock); + rdsProtocolMock.getResponseObserve().onError(new RuntimeException()); + rdsCountDownLatch.await(); + Assertions.assertTrue(isRdsReConnect.get()); + } catch (Exception e) { + Assertions.assertTrue(isRdsReConnect.get()); + } + + // mock eds + Map, List>>> edsMap = new HashMap<>(); + AtomicBoolean isEdsReConnect = new AtomicBoolean(false); + CountDownLatch edsCountDownLatch = new CountDownLatch(1); + // support multi-consumer + Consumer> edsConsumer = (routeResult) -> { + isEdsReConnect.set(true); + edsCountDownLatch.countDown(); + }; + edsMap.compute(new HashSet<>(), (k, v) -> { + if (v == null) { + v = new ArrayList<>(); + } + v.add(edsConsumer); + // support multi-consumer + return v; + }); + Assertions.assertFalse(isEdsReConnect.get()); + edsProtocolMock.setConsumerObserveMap(edsMap); + try { + triggerConsumerObserveMapListener(endpointResult, edsProtocolMock); + edsProtocolMock.getResponseObserve().onError(new RuntimeException()); + edsCountDownLatch.await(); + Assertions.assertTrue(isEdsReConnect.get()); + } catch (Exception e) { + Assertions.assertTrue(isEdsReConnect.get()); + } + } + + + @Test + public void testMultiConsumer() { + StreamObserver requestStreamObserver = mock(StreamObserver.class); + when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver); + doNothing().when(requestStreamObserver).onNext(any()); + MockedConstruction ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { + when(mock.get()).thenReturn(listenerResult); + }); + + Map, List>>> ldsMap = new HashMap<>(); + AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false); + AtomicBoolean ldsIsSecondConsumerInvoke = new AtomicBoolean(false); + CountDownLatch ldsCountDownLatch = new CountDownLatch(2); + + // support repeat consumer + Consumer> ldsFirstConsumer = (listenerResult) -> { + ldsIsFirstConsumerInvoke.set(true); + ldsCountDownLatch.countDown(); + }; + Consumer> ldsSecondConsumer = (listenerResult) -> { + ldsIsSecondConsumerInvoke.set(true); + ldsCountDownLatch.countDown(); + }; + Set ldsResourceNames = new HashSet<>(); + ldsResourceNames.add(ldsProtocolMock.emptyResourceName); + + ldsMap.computeIfAbsent(ldsResourceNames, (key) -> { + List>> consumers = new ArrayList<>(); + consumers.add(ldsFirstConsumer); + consumers.add(ldsSecondConsumer); + return consumers; + }); + Assertions.assertFalse(ldsIsFirstConsumerInvoke.get() || ldsIsSecondConsumerInvoke.get()); + ldsProtocolMock.setConsumerObserveMap(ldsMap); + + Map oldLdsResult = new HashMap<>(); + Map newLdsResult = new HashMap<>(); + + oldLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); + newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); + newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); + try { + ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); + ldsCountDownLatch.await(); + Assertions.assertTrue(ldsIsFirstConsumerInvoke.get() && ldsIsSecondConsumerInvoke.get()); + } catch (Exception e) { + Assertions.assertTrue(ldsIsFirstConsumerInvoke.get() && ldsIsSecondConsumerInvoke.get()); + } finally { + ldsMocked.close(); + } + + // mock rds + MockedConstruction rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { + when(mock.get()).thenReturn(routeResult); + }); + + Map, List>>> rdsMap = new HashMap<>(); + AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false); + AtomicBoolean rdsIsSecondConsumerInvoke = new AtomicBoolean(false); + CountDownLatch rdsCountDownLatch = new CountDownLatch(2); + + // support repeat consumer + Consumer> rdsFirstConsumer = (routeResult) -> { + rdsIsFirstConsumerInvoke.set(true); + rdsCountDownLatch.countDown(); + }; + Consumer> rdsSecondConsumer = (routeResult) -> { + rdsIsSecondConsumerInvoke.set(true); + rdsCountDownLatch.countDown(); + }; + rdsMap.computeIfAbsent(routeConfigNames, (key) -> { + List>> consumers = new ArrayList<>(); + consumers.add(rdsFirstConsumer); + consumers.add(rdsSecondConsumer); + return consumers; + }); + Assertions.assertFalse(rdsIsFirstConsumerInvoke.get() || rdsIsSecondConsumerInvoke.get()); + rdsProtocolMock.setConsumerObserveMap(rdsMap); + + Map oldRdsResult = new HashMap<>(); + Map newRdsResult = new HashMap<>(); + + oldRdsResult.put("15013", new RouteResult(domainMap)); + newRdsResult.put("15014", new RouteResult(domainMap)); + newRdsResult.put("15014", new RouteResult(domainMap)); + + try { + rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); + rdsCountDownLatch.await(); + Assertions.assertTrue(rdsIsFirstConsumerInvoke.get() && rdsIsSecondConsumerInvoke.get()); + } catch (Exception e) { + Assertions.assertTrue(rdsIsSecondConsumerInvoke.get() && rdsIsSecondConsumerInvoke.get()); + } finally { + rdsMocked.close(); + } + + // mock eds + MockedConstruction edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> { + when(mock.get()).thenReturn(endpointResult); + }); + + Map, List>>> edsMap = new HashMap<>(); + AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false); + AtomicBoolean edsIsSecondConsumerInvoke = new AtomicBoolean(false); + CountDownLatch edsCountDownLatch = new CountDownLatch(2); + + // support repeat consumer + Consumer> edsFirstConsumer = (routeResult) -> { + edsIsFirstConsumerInvoke.set(true); + edsCountDownLatch.countDown(); + }; + Consumer> edsSecondConsumer = (routeResult) -> { + edsIsSecondConsumerInvoke.set(true); + edsCountDownLatch.countDown(); + }; + edsMap.computeIfAbsent(cluster, (key) -> { + List>> consumers = new ArrayList<>(); + consumers.add(edsFirstConsumer); + consumers.add(edsSecondConsumer); + return consumers; + }); + Assertions.assertFalse(edsIsFirstConsumerInvoke.get() || edsIsSecondConsumerInvoke.get()); + edsProtocolMock.setConsumerObserveMap(edsMap); + + Map oldEdsResult = new HashMap<>(); + Map newEdsResult = new HashMap<>(); + + oldEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); + newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); + newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); + + try { + edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); + edsCountDownLatch.await(); + Assertions.assertTrue(edsIsFirstConsumerInvoke.get() && edsIsSecondConsumerInvoke.get()); + } catch (Exception e) { + Assertions.assertTrue(edsIsFirstConsumerInvoke.get() && edsIsSecondConsumerInvoke.get()); + } finally { + edsMocked.close(); +// mockedStatic.close(); + } + } + + @Test + void testResponseObserver() { + //mock lds + Map, List>>> ldsMap = new HashMap<>(); + AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false); + + // support repeat consumer + Consumer> ldsFirstConsumer = (listenerResult) -> { + ldsIsFirstConsumerInvoke.set(true); + }; + Set ldsResourceNames = new HashSet<>(); + ldsResourceNames.add(ldsProtocolMock.emptyResourceName); + + ldsMap.computeIfAbsent(ldsResourceNames, (key) -> { + List>> consumers = new ArrayList<>(); + consumers.add(ldsFirstConsumer); + return consumers; + }); + Assertions.assertFalse(ldsIsFirstConsumerInvoke.get()); + ldsProtocolMock.setConsumerObserveMap(ldsMap); + + Map oldLdsResult = new HashMap<>(); + Map newLdsResult = new HashMap<>(); + + oldLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); + newLdsResult.put("emptyResourcesName1", new ListenerResult(routeConfigNames)); + try { + ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); + Assertions.assertFalse(ldsIsFirstConsumerInvoke.get()); + + newLdsResult.put("emptyResourcesName", new ListenerResult(routeConfigNames)); + ldsProtocolMock.getResponseObserve().discoveryResponseListener(oldLdsResult, newLdsResult); + Assertions.assertTrue(ldsIsFirstConsumerInvoke.get()); + + } catch (Exception e) { + Assertions.assertTrue(ldsIsFirstConsumerInvoke.get()); + } + + + // mock rds + Map, List>>> rdsMap = new HashMap<>(); + AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false); + + // support repeat consumer + Consumer> rdsFirstConsumer = (routeResult) -> { + rdsIsFirstConsumerInvoke.set(true); + }; + rdsMap.computeIfAbsent(routeConfigNames, (key) -> { + List>> consumers = new ArrayList<>(); + consumers.add(rdsFirstConsumer); + return consumers; + }); + rdsProtocolMock.setConsumerObserveMap(rdsMap); + + Map oldRdsResult = new HashMap<>(); + Map newRdsResult = new HashMap<>(); + + oldRdsResult.put("15013", new RouteResult(domainMap)); + newRdsResult.put("15013", new RouteResult(domainMap)); + + try { + Assertions.assertFalse(rdsIsFirstConsumerInvoke.get()); + rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); + newRdsResult.put("15014", new RouteResult(domainMap)); + rdsProtocolMock.getResponseObserve().discoveryResponseListener(oldRdsResult, newRdsResult); + Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); + + Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); + } catch (Exception e) { + Assertions.assertTrue(rdsIsFirstConsumerInvoke.get()); + } + + // mock eds + Map, List>>> edsMap = new HashMap<>(); + AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false); + + // support repeat consumer + Consumer> edsFirstConsumer = (routeResult) -> { + edsIsFirstConsumerInvoke.set(true); + }; + edsMap.computeIfAbsent(cluster, (key) -> { + List>> consumers = new ArrayList<>(); + consumers.add(edsFirstConsumer); + return consumers; + }); + Assertions.assertFalse(edsIsFirstConsumerInvoke.get()); + edsProtocolMock.setConsumerObserveMap(edsMap); + + Map oldEdsResult = new HashMap<>(); + Map newEdsResult = new HashMap<>(); + + oldEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); + newEdsResult.put("dubbo-samples-provider2", new EndpointResult(endpoints)); + + try { + edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); + Assertions.assertFalse(edsIsFirstConsumerInvoke.get()); + newEdsResult.put("dubbo-samples-provider", new EndpointResult(endpoints)); + edsProtocolMock.getResponseObserve().discoveryResponseListener(oldEdsResult, newEdsResult); + Assertions.assertTrue(edsIsFirstConsumerInvoke.get()); + } catch (Exception e) { + Assertions.assertTrue(edsIsFirstConsumerInvoke.get()); + } + } + + private void triggerConsumerObserveMapListener(Map resultMap, AbstractProtocol protocol) { + CompletableFuture.runAsync(() -> { + while (true) { + Map, List>>> map = protocol.getConsumerObserveMap(); + if (map != null && map.size() > 1) { + map.forEach((k, v) -> v.forEach(o -> o.accept(resultMap))); + break; + } + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + break; + } + } + }); + }; +} diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java new file mode 100644 index 0000000000..3d6e67b258 --- /dev/null +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.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.protocol.impl; + +import io.envoyproxy.envoy.config.core.v3.Node; +import org.apache.dubbo.registry.xds.util.XdsChannel; +import org.apache.dubbo.registry.xds.util.protocol.message.EndpointResult; +import org.apache.dubbo.rpc.model.ApplicationModel; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Consumer; + +public class EdsProtocolMock extends EdsProtocol{ + + public EdsProtocolMock(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { + super(xdsChannel, node, pollingTimeout, applicationModel); + } + + public Map getResourcesMap() { + return resourcesMap; + } + + public void setResourcesMap(Map resourcesMap) { + this.resourcesMap = resourcesMap; + } + + public void setConsumerObserveMap(Map, List>>> consumerObserveMap) { + this.consumerObserveMap = consumerObserveMap; + } + + public ResponseObserverMock getResponseObserve() { + return new ResponseObserverMock(); + } + + public void setObserveResourcesName(Set observeResourcesName) { + this.observeResourcesName = observeResourcesName; + } + class ResponseObserverMock extends ResponseObserver { + + } +} diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java new file mode 100644 index 0000000000..6644a6a26e --- /dev/null +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java @@ -0,0 +1,84 @@ +/* + * 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 io.envoyproxy.envoy.config.core.v3.Node; +import io.envoyproxy.envoy.service.discovery.v3.DiscoveryRequest; +import io.grpc.stub.StreamObserver; +import org.apache.dubbo.registry.xds.util.XdsChannel; +import org.apache.dubbo.registry.xds.util.protocol.message.ListenerResult; +import org.apache.dubbo.rpc.model.ApplicationModel; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Consumer; + +public class LdsProtocolMock extends LdsProtocol{ + + public LdsProtocolMock(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { + super(xdsChannel, node, pollingTimeout, applicationModel); + } + + public Map getResourcesMap() { + return resourcesMap; + } + + public void setResourcesMap(Map resourcesMap) { + this.resourcesMap = resourcesMap; + } + + public StreamObserver getRequestObserver() { + return requestObserver; + } + + public void setRequestObserver(StreamObserver requestObserver) { + this.requestObserver = requestObserver; + } + + public ResponseObserverMock getResponseObserve() { + return new ResponseObserverMock(); + } + + protected DiscoveryRequest buildDiscoveryRequest(Set resourceNames) { + return DiscoveryRequest.newBuilder() + .setNode(node) + .setTypeUrl(getTypeUrl()) + .addAllResourceNames(resourceNames) + .build(); + } + + public Set getObserveResourcesName() { + return observeResourcesName; + } + + public void setObserveResourcesName(Set observeResourcesName) { + this.observeResourcesName = observeResourcesName; + } + + public Map, List>>> getConsumerObserveMap() { + return consumerObserveMap; + } + + public void setConsumerObserveMap(Map, List>>> consumerObserveMap) { + this.consumerObserveMap = consumerObserveMap; + } + class ResponseObserverMock extends ResponseObserver { + + } +} diff --git a/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java new file mode 100644 index 0000000000..8179b2234a --- /dev/null +++ b/dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.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.impl; + +import io.envoyproxy.envoy.config.core.v3.Node; +import org.apache.dubbo.registry.xds.util.XdsChannel; +import org.apache.dubbo.registry.xds.util.protocol.message.RouteResult; +import org.apache.dubbo.rpc.model.ApplicationModel; + +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Consumer; + +public class RdsProtocolMock extends RdsProtocol{ + + public RdsProtocolMock(XdsChannel xdsChannel, Node node, int pollingTimeout, ApplicationModel applicationModel) { + super(xdsChannel, node, pollingTimeout, applicationModel); + } + + public Map getResourcesMap() { + return resourcesMap; + } + + public void setResourcesMap(Map resourcesMap) { + this.resourcesMap = resourcesMap; + } + + public Set getObserveResourcesName() { + return observeResourcesName; + } + + public ResponseObserverMock getResponseObserve() { + return new ResponseObserverMock(); + } + + public void setConsumerObserveMap(Map, List>>> consumerObserveMap) { + this.consumerObserveMap = consumerObserveMap; + } + public void setObserveResourcesName(Set observeResourcesName) { + this.observeResourcesName = observeResourcesName; + } + + class ResponseObserverMock extends ResponseObserver { + + } +}