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

Feature/modify xds subscribe (#10885) 

Co-authored-by: Albumen Kevin <jhq0812@gmail.com>
This commit is contained in:
aamingaa 2023-01-06 15:16:53 +08:00 committed by GitHub
parent 32db2636bd
commit 4188b884b2
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 1095 additions and 253 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dubbo-dependencies-bom/pom.xml
View File

@ -143,7 +143,7 @@
<tomcat_embed_version>8.5.78</tomcat_embed_version>
<jetcd_version>0.5.3</jetcd_version>
<nacos_version>2.1.2</nacos_version>
<grpc.version>1.47.0</grpc.version>
<grpc.version>1.41.0</grpc.version>
<grpc_contrib_verdion>0.8.1</grpc_contrib_verdion>
<jprotoc_version>1.2.1</jprotoc_version>
<!-- Log libs -->

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

@ -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<String, ListenerResult> listenerResult;
private final AtomicLong observeRouteRequest = new AtomicLong(-1);
protected Map<String, RouteResult> routeResult;
private final Map<String, Long> domainObserveRequest = new ConcurrentHashMap<>();
private final AtomicBoolean isRdsObserve = new AtomicBoolean(false);
private final HashSet<String> domainObserveRequest = new HashSet<>();
private final Map<String, Set<Consumer<Set<Endpoint>>>> 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<String> 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<String> newRoute = o.searchDomain(domain);
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);
}
}
}
});
});
// update local cache
routeResult = newResult;
});
}, false);
}
public static PilotExchanger initialize(URL url) {
@ -116,17 +115,25 @@ public class PilotExchanger {
}
public Set<String> getServices() {
return routeResult.getDomains();
Set<String> domains = new HashSet<>();
for (Map.Entry<String, RouteResult> entry: routeResult.entrySet()) {
domains.addAll(entry.getValue().getDomains());
}
return domains;
}
public Set<Endpoint> getEndpoints(String domain) {
Set<String> cluster = routeResult.searchDomain(domain);
if (CollectionUtils.isNotEmpty(cluster)) {
EndpointResult endpoint = edsProtocol.getResource(cluster);
return endpoint.getEndpoints();
} else {
return Collections.emptySet();
Set<Endpoint> endpoints = new HashSet<>();
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);
endpointResultList.forEach((k, v) -> endpoints.addAll(v.getEndpoints()));
} else {
return Collections.emptySet();
}
}
return endpoints;
}
public void observeEndpoints(String domain, Consumer<Set<Endpoint>> 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<String> 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<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,
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);
}
}
}
}

13
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/XdsChannel.java
View File

@ -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))){

371
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/AbstractProtocol.java
View File

@ -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<T, S extends DeltaResource<T>> implements XdsProtocol<T> {
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<Long, Set<String>> requestParam = new ConcurrentHashMap<>();
private final int checkInterval;
/**
* Store ADS Request Observer ( StreamObserver in Streaming Request )
* K - requestId, V - StreamObserver
*/
private final Map<Long, StreamObserver<DiscoveryRequest>> 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<Long, ScheduledFuture<?>> 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<Long, CompletableFuture<T>> streamResult = new ConcurrentHashMap<>();
protected final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
private final ScheduledExecutorService pollingExecutor;
protected Set<String> 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<Set<String>, List<Consumer<Map<String, T>>>> consumerObserveMap = new ConcurrentHashMap<>();
public AbstractProtocol(XdsChannel xdsChannel, Node node, int pollingPoolSize, int pollingTimeout) {
public Map<Set<String>, List<Consumer<Map<String, T>>>> 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<String, T> resourcesMap = new ConcurrentHashMap<>();
protected StreamObserver<DiscoveryRequest> requestObserver;
/**
* Abstract method to obtain Type-URL from sub-class
*
@ -93,103 +94,126 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
*/
public abstract String getTypeUrl();
public boolean isCacheExistResource(Set<String> resourceNames) {
for (String resourceName : resourceNames) {
if ("".equals(resourceName)) {
continue;
}
if (!resourcesMap.containsKey(resourceName)) {
return false;
}
}
return true;
}
public T getCacheResource(String resourceName) {
if (resourceName == null || resourceName.length() == 0) {
return null;
}
return resourcesMap.get(resourceName);
}
@Override
public T getResource(Set<String> resourceNames) {
long request = requestId.getAndIncrement();
public Map<String, T> getResource(Set<String> resourceNames) {
resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
// Store Request Parameter, which will be used for ACK
requestParam.put(request, resourceNames);
// create observer
StreamObserver<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
// use future to get async result
CompletableFuture<T> 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<String> resourceNames, Consumer<T> 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<DiscoveryRequest> requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver(request));
requestObserverMap.put(request, requestObserver);
ScheduledFuture<?> scheduledFuture = pollingExecutor.scheduleAtFixedRate(() -> {
try {
// origin request, may changed by updateObserve
Set<String> names = requestParam.get(request);
// use future to get async result, future complete on StreamObserver onNext
CompletableFuture<T> future = new CompletableFuture<>();
streamResult.put(request, future);
// observer reused
StreamObserver<DiscoveryRequest> 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<String, T> getResourceFromCache(Set<String> resourceNames) {
return resourceNames.stream()
.collect(Collectors.toMap(k -> k, this::getCacheResource));
}
@Override
public void updateObserve(long request, Set<String> resourceNames) {
// send difference in resourceNames
requestParam.put(request, resourceNames);
public Map<String, T> getResourceFromRemote(Set<String> resourceNames) {
try {
resourceLock.lock();
CompletableFuture<Map<String, T>> future = new CompletableFuture<>();
if (requestObserver == null) {
requestObserver = xdsChannel.createDeltaDiscoveryRequest(new ResponseObserver());
}
observeResourcesName = resourceNames;
Set<String> consumerObserveResourceNames = new HashSet<>();
if (resourceNames.isEmpty()) {
consumerObserveResourceNames.add(emptyResourceName);
} else {
consumerObserveResourceNames = resourceNames;
}
Consumer<Map<String, T>> futureConsumer = future::complete;
try {
writeLock.lock();
consumerObserveMap.computeIfAbsent(consumerObserveResourceNames, key -> new ArrayList<>())
.add(futureConsumer);
} finally {
writeLock.unlock();
}
Set<String> 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<String, T> 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<String> resourceNames, Consumer<Map<String, T>> consumer, boolean isReConnect) {
resourceNames = resourceNames == null ? Collections.emptySet() : resourceNames;
// call once for full data
if (!isReConnect) {
try {
writeLock.lock();
Set<String> 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<String> resourceNames) {
@ -210,51 +234,98 @@ public abstract class AbstractProtocol<T, S extends DeltaResource<T>> implements
.build();
}
protected abstract T decodeDiscoveryResponse(DiscoveryResponse response);
protected abstract Map<String, T> decodeDiscoveryResponse(DiscoveryResponse response);
private class ResponseObserver implements StreamObserver<DiscoveryResponse> {
private final long requestId;
public class ResponseObserver implements StreamObserver<DiscoveryResponse> {
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<DiscoveryRequest> observer = requestObserverMap.get(requestId);
if (observer == null) {
Map<String, T> newResult = decodeDiscoveryResponse(value);
Map<String, T> oldResource = resourcesMap;
discoveryResponseListener(oldResource, newResult);
resourcesMap = newResult;
requestObserver.onNext(buildDiscoveryRequest(Collections.emptySet(), value));
}
public void discoveryResponseListener(Map<String, T> oldResult, Map<String, T> newResult) {
Set<String> changedResourceNames = new HashSet<>();
oldResult.forEach((key, origin) -> {
if (!Objects.equals(origin, newResult.get(key))) {
changedResourceNames.add(key);
}
});
newResult.forEach((key, origin) -> {
if (!Objects.equals(origin, oldResult.get(key))) {
changedResourceNames.add(key);
}
});
if (changedResourceNames.isEmpty()) {
return;
}
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<Set<String>, List<Consumer<Map<String, T>>>> entry : consumerObserveMap.entrySet()) {
if (entry.getKey().stream().noneMatch(changedResourceNames::contains)) {
// none update
continue;
}
Map<String, T> dsResultMap = entry.getKey()
.stream()
.collect(Collectors.toMap(k -> k, v -> newResult.get(v)));
entry.getValue().forEach(o -> o.accept(dsResultMap));
}
} finally {
readLock.unlock();
}
}
@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<T> 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<String> 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);
}
}

13
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/XdsProtocol.java
View File

@ -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<T> {
* @param resourceNames specified resource name
* @return resources, null if request failed
*/
T getResource(Set<String> resourceNames);
Map<String, T> getResource(Set<String> resourceNames);
/**
* Add a observer resource with {@link Consumer}
@ -36,13 +37,5 @@ public interface XdsProtocol<T> {
* @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<String> resourceNames, Consumer<T> 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<String> resourceNames);
void observeResource(Set<String> resourceNames, Consumer<Map<String, T>> consumer, boolean isReConnect);
}

25
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocol.java
View File

@ -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<EndpointResult, DeltaEndpoint>
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<EndpointResult, DeltaEndpoint>
return "type.googleapis.com/envoy.config.endpoint.v3.ClusterLoadAssignment";
}
@Override
protected EndpointResult decodeDiscoveryResponse(DiscoveryResponse response) {
protected Map<String, EndpointResult> decodeDiscoveryResponse(DiscoveryResponse response) {
if (getTypeUrl().equals(response.getTypeUrl())) {
Set<Endpoint> 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<Endpoint> decodeResourceToEndpoint(ClusterLoadAssignment resource) {
return resource.getEndpointsList().stream()
.flatMap((e) -> e.getLbEndpointsList().stream())
private EndpointResult decodeResourceToEndpoint(ClusterLoadAssignment resource) {
Set<Endpoint> 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) {

27
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocol.java
View File

@ -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<ListenerResult, DeltaListener> {
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<ListenerResult, DeltaListener>
return "type.googleapis.com/envoy.config.listener.v3.Listener";
}
public ListenerResult getListeners() {
public Map<String, ListenerResult> getListeners() {
return getResource(null);
}
public void observeListeners(Consumer<ListenerResult> consumer) {
observeResource(Collections.emptySet(), consumer);
public void observeListeners(Consumer<Map<String, ListenerResult>> consumer) {
observeResource(Collections.emptySet(), consumer, false);
}
@Override
protected ListenerResult decodeDiscoveryResponse(DiscoveryResponse response) {
protected Map<String, ListenerResult> decodeDiscoveryResponse(DiscoveryResponse response) {
if (getTypeUrl().equals(response.getTypeUrl())) {
Set<String> set = response.getResourcesList().stream()
.map(LdsProtocol::unpackListener)
.filter(Objects::nonNull)
.flatMap((e) -> decodeResourceToListener(e).stream())
.flatMap(e -> decodeResourceToListener(e).stream())
.collect(Collectors.toSet());
return new ListenerResult(set);
Map<String, ListenerResult> listenerDecodeResult = new ConcurrentHashMap<>();
listenerDecodeResult.put(emptyResourceName, new ListenerResult(set));
return listenerDecodeResult;
}
return new ListenerResult();
return new HashMap<>();
}
private Set<String> 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)

22
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocol.java
View File

@ -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<RouteResult, DeltaRoute> {
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<RouteResult, DeltaRoute> {
}
@Override
protected RouteResult decodeDiscoveryResponse(DiscoveryResponse response) {
protected Map<String, RouteResult> decodeDiscoveryResponse(DiscoveryResponse response) {
if (getTypeUrl().equals(response.getTypeUrl())) {
Map<String, Set<String>> 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<String, Set<String>> decodeResourceToListener(RouteConfiguration resource) {
private RouteResult decodeResourceToListener(RouteConfiguration resource) {
Map<String, Set<String>> map = new HashMap<>();
resource.getVirtualHostsList()
.forEach(virtualHost -> {
@ -80,7 +76,7 @@ public class RdsProtocol extends AbstractProtocol<RouteResult, DeltaRoute> {
map.put(domain, cluster);
}
});
return map;
return new RouteResult(map);
}
private static RouteConfiguration unpackRouteConfiguration(Any any) {

5
dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/util/protocol/message/RouteResult.java
View File

@ -35,6 +35,11 @@ public class RouteResult {
this.domainMap = domainMap;
}
public Map<String, Set<String>> getDomainMap() {
return domainMap;
}
public boolean isNotEmpty() {
return !domainMap.isEmpty();
}

542
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/DsProtocolTest.java Normal file
View File

@ -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<String, ListenerResult> listenerResult;
private Set<String> routeConfigNames;
private Set<String> cluster;
private Map<String, RouteResult> routeResult;
private Set<Endpoint> endpoints;
private Map<String, EndpointResult> endpointResult;
private Map<String, Set<String>> 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<String, ListenerResult> listenerResults = new HashMap();
listenerResults.put(ldsProtocolMock.emptyResourceName,new ListenerResult(routeConfigNames));
this.listenerResult = spy(listenerResults);
// mock route result
this.domainMap = new HashMap<>();
Set<String> domainValue = new HashSet<>();
domainValue.add("outbound|15014||istiod.istio-system.svc.cluster.local");
domainMap.put("istiod.istio-system.svc", domainValue);
Map<String, RouteResult> routeResults = new HashMap();
routeResults.put("15014", new RouteResult(domainMap));
this.routeResult = routeResults;
// mock eds result
Set<String> 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<String, EndpointResult> endpointResults = new HashMap();
endpointResults.put("dubbo-samples-provider" ,new EndpointResult(endpoints));
this.endpointResult = endpointResults;
// mockedStatic.close();
}
@Test
void testGetResource() {
StreamObserver<DiscoveryRequest> requestStreamObserver = mock(StreamObserver.class);
// mock lds getResource
when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver);
MockedConstruction<CompletableFuture> ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> {
when(mock.get()).thenReturn(listenerResult);
});
Map<String, ListenerResult> ldsResult = ldsProtocolMock.getResource(null);
Assertions.assertNotNull(ldsResult);
verify(ldsProtocolMock, times(0)).isCacheExistResource(null);
ldsMocked.close();
// mock rds getResource
MockedConstruction<CompletableFuture> rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> {
when(mock.get()).thenReturn(null);
});
Map<String, RouteResult> 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<String, RouteResult> newRdsResult = routeConfigNames.stream().collect(Collectors.toMap(k -> k, v -> rdsProtocolMock.getCacheResource(v)));
Assertions.assertEquals(newRdsResult, rdsResult);
rdsMocked.close();
//mock eds getResource
MockedConstruction<CompletableFuture> edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> {
when(mock.get()).thenReturn(null);
});
Map<String, EndpointResult> 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<String, EndpointResult> newEdsResult = cluster.stream().collect(Collectors.toMap(k -> k, v -> edsProtocolMock.getCacheResource(v)));
Assertions.assertEquals(newEdsResult, endpointResult);
edsMocked.close();
}
@Test
void observeDsReConnect() {
StreamObserver<DiscoveryRequest> requestStreamObserver = mock(StreamObserver.class);
// mock lds reconnect
when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver);
doNothing().when(requestStreamObserver).onNext(any());
Map<Set<String>, List<Consumer<Map<String, ListenerResult>>>> ldsMap = new HashMap<>();
AtomicBoolean isLdsReConnect = new AtomicBoolean(false);
CountDownLatch ldsCountDownLatch = new CountDownLatch(1);
// // support multi-consumer
Consumer<Map<String, ListenerResult>> 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<Set<String>, List<Consumer<Map<String, RouteResult>>>> rdsMap = new HashMap<>();
AtomicBoolean isRdsReConnect = new AtomicBoolean(false);
CountDownLatch rdsCountDownLatch = new CountDownLatch(1);
// support multi-consumer
Consumer<Map<String, RouteResult>> 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<Set<String>, List<Consumer<Map<String, EndpointResult>>>> edsMap = new HashMap<>();
AtomicBoolean isEdsReConnect = new AtomicBoolean(false);
CountDownLatch edsCountDownLatch = new CountDownLatch(1);
// support multi-consumer
Consumer<Map<String, EndpointResult>> 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<DiscoveryRequest> requestStreamObserver = mock(StreamObserver.class);
when(xdsChannel.createDeltaDiscoveryRequest(any(StreamObserver.class))).thenReturn(requestStreamObserver);
doNothing().when(requestStreamObserver).onNext(any());
MockedConstruction<CompletableFuture> ldsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> {
when(mock.get()).thenReturn(listenerResult);
});
Map<Set<String>, List<Consumer<Map<String, ListenerResult>>>> ldsMap = new HashMap<>();
AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false);
AtomicBoolean ldsIsSecondConsumerInvoke = new AtomicBoolean(false);
CountDownLatch ldsCountDownLatch = new CountDownLatch(2);
// support repeat consumer
Consumer<Map<String, ListenerResult>> ldsFirstConsumer = (listenerResult) -> {
ldsIsFirstConsumerInvoke.set(true);
ldsCountDownLatch.countDown();
};
Consumer<Map<String, ListenerResult>> ldsSecondConsumer = (listenerResult) -> {
ldsIsSecondConsumerInvoke.set(true);
ldsCountDownLatch.countDown();
};
Set<String> ldsResourceNames = new HashSet<>();
ldsResourceNames.add(ldsProtocolMock.emptyResourceName);
ldsMap.computeIfAbsent(ldsResourceNames, (key) -> {
List<Consumer<Map<String, ListenerResult>>> consumers = new ArrayList<>();
consumers.add(ldsFirstConsumer);
consumers.add(ldsSecondConsumer);
return consumers;
});
Assertions.assertFalse(ldsIsFirstConsumerInvoke.get() || ldsIsSecondConsumerInvoke.get());
ldsProtocolMock.setConsumerObserveMap(ldsMap);
Map<String, ListenerResult> oldLdsResult = new HashMap<>();
Map<String, ListenerResult> 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<CompletableFuture> rdsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> {
when(mock.get()).thenReturn(routeResult);
});
Map<Set<String>, List<Consumer<Map<String, RouteResult>>>> rdsMap = new HashMap<>();
AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false);
AtomicBoolean rdsIsSecondConsumerInvoke = new AtomicBoolean(false);
CountDownLatch rdsCountDownLatch = new CountDownLatch(2);
// support repeat consumer
Consumer<Map<String, RouteResult>> rdsFirstConsumer = (routeResult) -> {
rdsIsFirstConsumerInvoke.set(true);
rdsCountDownLatch.countDown();
};
Consumer<Map<String, RouteResult>> rdsSecondConsumer = (routeResult) -> {
rdsIsSecondConsumerInvoke.set(true);
rdsCountDownLatch.countDown();
};
rdsMap.computeIfAbsent(routeConfigNames, (key) -> {
List<Consumer<Map<String, RouteResult>>> consumers = new ArrayList<>();
consumers.add(rdsFirstConsumer);
consumers.add(rdsSecondConsumer);
return consumers;
});
Assertions.assertFalse(rdsIsFirstConsumerInvoke.get() || rdsIsSecondConsumerInvoke.get());
rdsProtocolMock.setConsumerObserveMap(rdsMap);
Map<String, RouteResult> oldRdsResult = new HashMap<>();
Map<String, RouteResult> 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<CompletableFuture> edsMocked = mockConstruction(CompletableFuture.class, (mock, context) -> {
when(mock.get()).thenReturn(endpointResult);
});
Map<Set<String>, List<Consumer<Map<String, EndpointResult>>>> edsMap = new HashMap<>();
AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false);
AtomicBoolean edsIsSecondConsumerInvoke = new AtomicBoolean(false);
CountDownLatch edsCountDownLatch = new CountDownLatch(2);
// support repeat consumer
Consumer<Map<String, EndpointResult>> edsFirstConsumer = (routeResult) -> {
edsIsFirstConsumerInvoke.set(true);
edsCountDownLatch.countDown();
};
Consumer<Map<String, EndpointResult>> edsSecondConsumer = (routeResult) -> {
edsIsSecondConsumerInvoke.set(true);
edsCountDownLatch.countDown();
};
edsMap.computeIfAbsent(cluster, (key) -> {
List<Consumer<Map<String, EndpointResult>>> consumers = new ArrayList<>();
consumers.add(edsFirstConsumer);
consumers.add(edsSecondConsumer);
return consumers;
});
Assertions.assertFalse(edsIsFirstConsumerInvoke.get() || edsIsSecondConsumerInvoke.get());
edsProtocolMock.setConsumerObserveMap(edsMap);
Map<String, EndpointResult> oldEdsResult = new HashMap<>();
Map<String, EndpointResult> 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<Set<String>, List<Consumer<Map<String, ListenerResult>>>> ldsMap = new HashMap<>();
AtomicBoolean ldsIsFirstConsumerInvoke = new AtomicBoolean(false);
// support repeat consumer
Consumer<Map<String, ListenerResult>> ldsFirstConsumer = (listenerResult) -> {
ldsIsFirstConsumerInvoke.set(true);
};
Set<String> ldsResourceNames = new HashSet<>();
ldsResourceNames.add(ldsProtocolMock.emptyResourceName);
ldsMap.computeIfAbsent(ldsResourceNames, (key) -> {
List<Consumer<Map<String, ListenerResult>>> consumers = new ArrayList<>();
consumers.add(ldsFirstConsumer);
return consumers;
});
Assertions.assertFalse(ldsIsFirstConsumerInvoke.get());
ldsProtocolMock.setConsumerObserveMap(ldsMap);
Map<String, ListenerResult> oldLdsResult = new HashMap<>();
Map<String, ListenerResult> 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<Set<String>, List<Consumer<Map<String, RouteResult>>>> rdsMap = new HashMap<>();
AtomicBoolean rdsIsFirstConsumerInvoke = new AtomicBoolean(false);
// support repeat consumer
Consumer<Map<String, RouteResult>> rdsFirstConsumer = (routeResult) -> {
rdsIsFirstConsumerInvoke.set(true);
};
rdsMap.computeIfAbsent(routeConfigNames, (key) -> {
List<Consumer<Map<String, RouteResult>>> consumers = new ArrayList<>();
consumers.add(rdsFirstConsumer);
return consumers;
});
rdsProtocolMock.setConsumerObserveMap(rdsMap);
Map<String, RouteResult> oldRdsResult = new HashMap<>();
Map<String, RouteResult> 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<Set<String>, List<Consumer<Map<String, EndpointResult>>>> edsMap = new HashMap<>();
AtomicBoolean edsIsFirstConsumerInvoke = new AtomicBoolean(false);
// support repeat consumer
Consumer<Map<String, EndpointResult>> edsFirstConsumer = (routeResult) -> {
edsIsFirstConsumerInvoke.set(true);
};
edsMap.computeIfAbsent(cluster, (key) -> {
List<Consumer<Map<String, EndpointResult>>> consumers = new ArrayList<>();
consumers.add(edsFirstConsumer);
return consumers;
});
Assertions.assertFalse(edsIsFirstConsumerInvoke.get());
edsProtocolMock.setConsumerObserveMap(edsMap);
Map<String, EndpointResult> oldEdsResult = new HashMap<>();
Map<String, EndpointResult> 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 <T> void triggerConsumerObserveMapListener(Map<String, T> resultMap, AbstractProtocol protocol) {
CompletableFuture.runAsync(() -> {
while (true) {
Map<Set<String>, List<Consumer<Map<String, T>>>> 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;
}
}
});
};
}

58
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/EdsProtocolMock.java Normal file
View File

@ -0,0 +1,58 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.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<String, EndpointResult> getResourcesMap() {
return resourcesMap;
}
public void setResourcesMap(Map<String, EndpointResult> resourcesMap) {
this.resourcesMap = resourcesMap;
}
public void setConsumerObserveMap(Map<Set<String>, List<Consumer<Map<String, EndpointResult>>>> consumerObserveMap) {
this.consumerObserveMap = consumerObserveMap;
}
public ResponseObserverMock getResponseObserve() {
return new ResponseObserverMock();
}
public void setObserveResourcesName(Set<String> observeResourcesName) {
this.observeResourcesName = observeResourcesName;
}
class ResponseObserverMock extends ResponseObserver {
}
}

84
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/LdsProtocolMock.java Normal file
View File

@ -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<String, ListenerResult> getResourcesMap() {
return resourcesMap;
}
public void setResourcesMap(Map<String, ListenerResult> resourcesMap) {
this.resourcesMap = resourcesMap;
}
public StreamObserver<DiscoveryRequest> getRequestObserver() {
return requestObserver;
}
public void setRequestObserver(StreamObserver<DiscoveryRequest> requestObserver) {
this.requestObserver = requestObserver;
}
public ResponseObserverMock getResponseObserve() {
return new ResponseObserverMock();
}
protected DiscoveryRequest buildDiscoveryRequest(Set<String> resourceNames) {
return DiscoveryRequest.newBuilder()
.setNode(node)
.setTypeUrl(getTypeUrl())
.addAllResourceNames(resourceNames)
.build();
}
public Set<String> getObserveResourcesName() {
return observeResourcesName;
}
public void setObserveResourcesName(Set<String> observeResourcesName) {
this.observeResourcesName = observeResourcesName;
}
public Map<Set<String>, List<Consumer<Map<String, ListenerResult>>>> getConsumerObserveMap() {
return consumerObserveMap;
}
public void setConsumerObserveMap(Map<Set<String>, List<Consumer<Map<String, ListenerResult>>>> consumerObserveMap) {
this.consumerObserveMap = consumerObserveMap;
}
class ResponseObserverMock extends ResponseObserver {
}
}

62
dubbo-xds/src/test/java/org/apache/dubbo/registry/xds/util/protocol/impl/RdsProtocolMock.java Normal file
View File

@ -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<String, RouteResult> getResourcesMap() {
return resourcesMap;
}
public void setResourcesMap(Map<String, RouteResult> resourcesMap) {
this.resourcesMap = resourcesMap;
}
public Set<String> getObserveResourcesName() {
return observeResourcesName;
}
public ResponseObserverMock getResponseObserve() {
return new ResponseObserverMock();
}
public void setConsumerObserveMap(Map<Set<String>, List<Consumer<Map<String, RouteResult>>>> consumerObserveMap) {
this.consumerObserveMap = consumerObserveMap;
}
public void setObserveResourcesName(Set<String> observeResourcesName) {
this.observeResourcesName = observeResourcesName;
}
class ResponseObserverMock extends ResponseObserver {
}
}