Skip to content

Commit

Permalink
Merge pull request #1951 from theonefx/endpoints
Browse files Browse the repository at this point in the history
add Endpoints and dubbo issue fix
  • Loading branch information
theonefx authored Jan 31, 2021
2 parents ddac4ef + 0471b45 commit 68c25bc
Show file tree
Hide file tree
Showing 10 changed files with 222 additions and 34 deletions.
2 changes: 1 addition & 1 deletion pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@

<properties>
<!-- Project revision -->
<revision>2.2.4.RELEASE</revision>
<revision>2.2.5.RC2</revision>

<!-- Dependency Versions -->
<spring-cloud-commons.version>2.2.5.RELEASE</spring-cloud-commons.version>
Expand Down
2 changes: 1 addition & 1 deletion spring-cloud-alibaba-dependencies/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
<description>Spring Cloud Alibaba Dependencies</description>

<properties>
<revision>2.2.4.RELEASE</revision>
<revision>2.2.5.RC2</revision>
<sentinel.version>1.8.0</sentinel.version>
<seata.version>1.3.0</seata.version>
<nacos.client.version>1.4.1</nacos.client.version>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -74,8 +74,9 @@ public void run(String... args) throws InterruptedException {
while (true) {
mySink.input5().poll(m -> {
String payload = (String) m.getPayload();
if(payload.contains("0")){
throw new IllegalArgumentException("111111111111111111111111111111111111111111");
if (payload.contains("0")) {
throw new IllegalArgumentException(
"111111111111111111111111111111111111111111");
}
System.out.println("pull msg: " + payload);
}, new ParameterizedTypeReference<String>() {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import com.alibaba.cloud.nacos.NacosDiscoveryProperties;
import com.alibaba.cloud.nacos.NacosServiceManager;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.alibaba.nacos.api.naming.pojo.ServiceInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -66,6 +67,11 @@ public Map<String, Object> nacosDiscovery() {

try {
subscribe = namingService.getSubscribeServices();
for (ServiceInfo serviceInfo : subscribe) {
List<Instance> instances = namingService.getAllInstances(
serviceInfo.getName(), serviceInfo.getGroupName());
serviceInfo.setHosts(instances);
}
}
catch (Exception e) {
log.error("get subscribe services from nacos fail,", e);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,11 @@

package com.alibaba.cloud.dubbo.actuate;

import com.alibaba.cloud.dubbo.actuate.endpoint.DubboDiscoveryEndpoint;
import com.alibaba.cloud.dubbo.actuate.endpoint.DubboExportedURLsEndpoint;
import com.alibaba.cloud.dubbo.actuate.endpoint.DubboRestMetadataEndpoint;

import org.springframework.boot.actuate.autoconfigure.endpoint.condition.ConditionalOnAvailableEndpoint;
import org.springframework.boot.actuate.autoconfigure.web.ManagementContextConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
Expand All @@ -34,7 +35,7 @@
@ConditionalOnClass(
name = "org.springframework.boot.actuate.endpoint.annotation.Endpoint")
@PropertySource("classpath:/META-INF/dubbo/default/actuator-endpoints.properties")
@ManagementContextConfiguration
@Configuration(proxyBeanMethods = false)
public class DubboMetadataEndpointAutoConfiguration {

@Bean
Expand All @@ -44,4 +45,19 @@ public DubboRestMetadataEndpoint dubboRestMetadataEndpoint() {
return new DubboRestMetadataEndpoint();
}

@Bean
@ConditionalOnMissingBean
@ConditionalOnAvailableEndpoint
public DubboDiscoveryEndpoint dubboDiscoveryEndpoint() {

return new DubboDiscoveryEndpoint();
}

@Bean
@ConditionalOnMissingBean
@ConditionalOnAvailableEndpoint
public DubboExportedURLsEndpoint dubboServiceMetadataEndpoint() {
return new DubboExportedURLsEndpoint();
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
/*
* Copyright 2013-2018 the original author or authors.
*
* Licensed 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
*
* https://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 com.alibaba.cloud.dubbo.actuate.endpoint;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import com.alibaba.cloud.dubbo.registry.DubboCloudRegistry;
import com.alibaba.cloud.dubbo.registry.SpringCloudRegistryFactory;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.NotifyListener;
import org.apache.dubbo.registry.integration.RegistryDirectory;
import org.apache.dubbo.rpc.Invoker;

import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;

import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SIDE;
import static org.apache.dubbo.common.constants.CommonConstants.SIDE_KEY;
import static org.springframework.http.MediaType.APPLICATION_JSON_VALUE;

/**
* Dubbo Registry Directory Metadata {@link DubboCloudRegistry}.
*
* @author <a href="mailto:[email protected]">Theonefx</a>
*/
@Endpoint(id = "dubboRegistryDirectory")
public class DubboDiscoveryEndpoint {

@ReadOperation(produces = APPLICATION_JSON_VALUE)
public Object get() {
DubboCloudRegistry registry = (DubboCloudRegistry) SpringCloudRegistryFactory
.getRegistries().stream().filter(o -> o instanceof DubboCloudRegistry)
.findFirst().orElse(null);

if (registry == null) {
return Collections.emptyMap();
}

Map<URL, Set<NotifyListener>> subscribeMap = registry.getSubscribed();

Map<String, List<Map<String, Object>>> result = new HashMap<>();
subscribeMap.forEach((url, listeners) -> {
String side = url.getParameter(SIDE_KEY);
if (!CONSUMER_SIDE.equals(side)) {
return;
}

List<Map<String, Object>> pairs = result.computeIfAbsent(url.getServiceKey(),
o -> new ArrayList<>());

Map<String, Object> pair = new HashMap<>();
List<String> invokerServices = new ArrayList<>();
for (NotifyListener listener : listeners) {
if (!(listener instanceof RegistryDirectory)) {
continue;
}
RegistryDirectory<?> directory = (RegistryDirectory<?>) listener;
List<? extends Invoker<?>> invokers = directory.getAllInvokers();
if (invokers == null) {
continue;
}
invokerServices.addAll(invokers.stream().map(Invoker::getUrl)
.map(URL::toServiceString).collect(Collectors.toList()));
}
pair.put("invokers", invokerServices);
pair.put("subscribeUrl", url.toMap());

pairs.add(pair);
});
return result;
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
/*
* Copyright 2013-2018 the original author or authors.
*
* Licensed 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
*
* https://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 com.alibaba.cloud.dubbo.actuate.endpoint;

import com.alibaba.cloud.dubbo.service.DubboMetadataService;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;

import static org.springframework.http.MediaType.APPLICATION_JSON_UTF8_VALUE;

/**
* Dubbo exported URLs.
* {@link org.springframework.boot.actuate.endpoint.annotation.Endpoint}.
*
* @author <a href="mailto:[email protected]">Theonefx</a>
*/
@Endpoint(id = "dubboExportedURLs")
public class DubboExportedURLsEndpoint {

@Autowired
private DubboMetadataService dubboMetadataService;

@ReadOperation(produces = APPLICATION_JSON_UTF8_VALUE)
public Object get() {
return dubboMetadataService.getAllExportedURLs();
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,8 @@
import org.springframework.cloud.client.discovery.DiscoveryClient;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.util.CollectionUtils;

import static java.lang.String.format;
Expand Down Expand Up @@ -184,16 +186,30 @@ private void subscribeURLs(URL url, NotifyListener listener) {
subscribeURLs(url, getServices(url), listener);

// Async subscription
registerServiceInstancesChangedListener(url, event -> {
registerServiceInstancesChangedListener(url,

Set<String> serviceNames = getServices(url);
new ApplicationListener<ServiceInstancesChangedEvent>() {

String serviceName = event.getServiceName();
private final URL url2subscribe = url;

if (serviceNames.contains(serviceName)) {
subscribeURLs(url, serviceNames, listener);
}
});
@Override
@Order
public void onApplicationEvent(ServiceInstancesChangedEvent event) {
Set<String> serviceNames = getServices(url);

String serviceName = event.getServiceName();

if (serviceNames.contains(serviceName)) {
subscribeURLs(url, serviceNames, listener);
}
}

@Override
public String toString() {
return "ServiceInstancesChangedEventListener:"
+ url.getServiceKey();
}
});
}

private void subscribeURLs(URL url, Set<String> serviceNames,
Expand Down Expand Up @@ -375,12 +391,12 @@ private void notifyAllSubscribedURLs(URL url, List<URL> subscribedURLs,
// Add the EMPTY_PROTOCOL URL
subscribedURLs.add(emptyURL(url));

if (isDubboMetadataServiceURL(url)) {
// if meta service change, and serviceInstances is zero, will clean up
// information about this client
String serviceName = url.getParameter(GROUP_KEY);
repository.removeMetadataAndInitializedService(serviceName, url);
}
// if (isDubboMetadataServiceURL(url)) {
// if meta service change, and serviceInstances is zero, will clean up
// information about this client
// String serviceName = url.getParameter(GROUP_KEY);
// repository.removeMetadataAndInitializedService(serviceName, url);
// }
}

if (logger.isDebugEnabled()) {
Expand Down Expand Up @@ -415,7 +431,7 @@ private List<ServiceInstance> doGetServiceInstances(String serviceName) {
}

private String generateId(URL url) {
return url.getServiceKey();
return url.toString();
}

private URL emptyURL(URL url) {
Expand Down Expand Up @@ -450,16 +466,30 @@ private void subscribeDubboMetadataServiceURLs(URL subscribedURL,

// Sync subscription
if (containsProviderCategory(subscribedURL)) {
registerServiceInstancesChangedListener(subscribedURL, event -> {
registerServiceInstancesChangedListener(subscribedURL,
new ApplicationListener<ServiceInstancesChangedEvent>() {

private final URL url2subscribe = subscribedURL;

@Override
@Order(Ordered.LOWEST_PRECEDENCE - 1)
public void onApplicationEvent(
ServiceInstancesChangedEvent event) {
String sourceServiceName = event.getServiceName();
String serviceName = getServiceName(subscribedURL);

String sourceServiceName = event.getServiceName();
String serviceName = getServiceName(subscribedURL);
if (Objects.equals(sourceServiceName, serviceName)) {
subscribeDubboMetadataServiceURLs(subscribedURL, listener,
sourceServiceName);
}
}

if (Objects.equals(sourceServiceName, serviceName)) {
subscribeDubboMetadataServiceURLs(subscribedURL, listener,
sourceServiceName);
}
});
@Override
public String toString() {
return "ServiceInstancesChangedEventListener:"
+ subscribedURL.getServiceKey();
}
});
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -84,8 +84,6 @@ public void removeProxy(String serviceName) {
*/
public DubboMetadataService getProxy(List<ServiceInstance> serviceInstances) {

DubboMetadataService dubboMetadataService = null;

// attempt to get the proxy of DubboMetadataService in maximum times
int attempts = serviceInstances.size();

Expand All @@ -98,15 +96,16 @@ public DubboMetadataService getProxy(List<ServiceInstance> serviceInstances) {
serviceInstance.get());

for (URL dubboMetadataServiceURL : dubboMetadataServiceURLs) {
dubboMetadataService = createProxyIfAbsent(dubboMetadataServiceURL);
DubboMetadataService dubboMetadataService = createProxyIfAbsent(
dubboMetadataServiceURL);
if (dubboMetadataService != null) {
return dubboMetadataService;
}
}
}
}

return dubboMetadataService;
return null;
}

/**
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,7 @@ com.alibaba.cloud.dubbo.autoconfigure.DubboServiceRegistrationAutoConfiguration,
com.alibaba.cloud.dubbo.autoconfigure.DubboServiceRegistrationNonWebApplicationAutoConfiguration,\
com.alibaba.cloud.dubbo.autoconfigure.DubboLoadBalancedRestTemplateAutoConfiguration,\
com.alibaba.cloud.dubbo.autoconfigure.DubboServiceAutoConfiguration,\
com.alibaba.cloud.dubbo.autoconfigure.DubboServiceDiscoveryAutoConfiguration
org.springframework.boot.actuate.autoconfigure.web.ManagementContextConfiguration=\
com.alibaba.cloud.dubbo.autoconfigure.DubboServiceDiscoveryAutoConfiguration,\
com.alibaba.cloud.dubbo.actuate.DubboMetadataEndpointAutoConfiguration
org.springframework.context.ApplicationContextInitializer=\
com.alibaba.cloud.dubbo.context.DubboServiceRegistrationApplicationContextInitializer
Expand Down

0 comments on commit 68c25bc

Please sign in to comment.