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

Collecting registration and consumption status inside process (#7714)

This commit is contained in:
ken.lj 2021-05-10 13:56:04 +08:00 committed by GitHub
parent ca0b982866
commit ea0506aa65
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 191 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-common/pom.xml
View File

@ -63,6 +63,10 @@
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>

79
dubbo-common/src/main/java/org/apache/dubbo/common/status/reporter/FrameworkStatusReporter.java Normal file
View File

@ -0,0 +1,79 @@
/*
* 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.common.status.reporter;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.rpc.model.ApplicationModel;
import com.google.gson.Gson;
import java.util.HashMap;
import java.util.Set;
@SPI
public interface FrameworkStatusReporter {
static final Gson gson = new Gson();
Logger logger = LoggerFactory.getLogger(FrameworkStatusReporter.class);
String REGISTRATION_STATUS = "registration";
String ADDRESS_CONSUMPTION_STATUS = "consumption";
void report(String type, Object obj);
static void reportRegistrationStatus(Object obj) {
doReport(REGISTRATION_STATUS, obj);
}
static void reportConsumptionStatus(Object obj) {
doReport(ADDRESS_CONSUMPTION_STATUS, obj);
}
static void doReport(String type, Object obj) {
// TODO, report asynchronously
try {
Set<FrameworkStatusReporter> reporters = ExtensionLoader.getExtensionLoader(FrameworkStatusReporter.class).getSupportedExtensionInstances();
if (CollectionUtils.isNotEmpty(reporters)) {
FrameworkStatusReporter reporter = reporters.iterator().next();
reporter.report(type, obj);
}
} catch (Exception e) {
logger.info("Report " + type + " status failed because of " + e.getMessage());
}
}
static String createRegistrationReport(String status) {
return "{\"application\":\"" +
ApplicationModel.getName() +
"\",\"status\":\"" +
status +
"\"}";
}
static String createConsumptionReport(String interfaceName, String version, String group, String status) {
HashMap<String, String> migrationStatus = new HashMap<>();
migrationStatus.put("type", "consumption");
migrationStatus.put("application", ApplicationModel.getName());
migrationStatus.put("service", interfaceName);
migrationStatus.put("version", version);
migrationStatus.put("group", group);
migrationStatus.put("status", status);
return gson.toJson(migrationStatus);
}
}

10
dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/utils/ConfigValidationUtils.java
View File

@ -24,6 +24,7 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.serialize.Serialization;
import org.apache.dubbo.common.status.StatusChecker;
import org.apache.dubbo.common.status.reporter.FrameworkStatusReporter;
import org.apache.dubbo.common.threadpool.ThreadPool;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.ConfigUtils;
@ -98,6 +99,7 @@ import static org.apache.dubbo.common.constants.RegistryConstants.REGISTRY_TYPE_
import static org.apache.dubbo.common.constants.RegistryConstants.SERVICE_REGISTRY_PROTOCOL;
import static org.apache.dubbo.common.constants.RemotingConstants.BACKUP_KEY;
import static org.apache.dubbo.common.extension.ExtensionLoader.getExtensionLoader;
import static org.apache.dubbo.common.status.reporter.FrameworkStatusReporter.createRegistrationReport;
import static org.apache.dubbo.config.Constants.ARCHITECTURE;
import static org.apache.dubbo.config.Constants.CONTEXTPATH_KEY;
import static org.apache.dubbo.config.Constants.DUBBO_IP_TO_REGISTRY;
@ -216,8 +218,9 @@ public class ConfigValidationUtils {
registryList.forEach(registryURL -> {
if (provider) {
// for registries enabled service discovery, automatically register interface compatible addresses.
String registerMode;
if (SERVICE_REGISTRY_PROTOCOL.equals(registryURL.getProtocol())) {
String registerMode = registryURL.getParameter(REGISTER_MODE_KEY, ConfigurationUtils.getDynamicGlobalConfiguration().getString(DUBBO_REGISTER_MODE_DEFAULT_KEY, DEFAULT_REGISTER_MODE_INSTANCE));
registerMode = registryURL.getParameter(REGISTER_MODE_KEY, ConfigurationUtils.getDynamicGlobalConfiguration().getString(DUBBO_REGISTER_MODE_DEFAULT_KEY, DEFAULT_REGISTER_MODE_INSTANCE));
if (!isValidRegisterMode(registerMode)) {
registerMode = DEFAULT_REGISTER_MODE_INSTANCE;
}
@ -231,7 +234,7 @@ public class ConfigValidationUtils {
result.add(interfaceCompatibleRegistryURL);
}
} else {
String registerMode = registryURL.getParameter(REGISTER_MODE_KEY, ConfigurationUtils.getDynamicGlobalConfiguration().getString(DUBBO_REGISTER_MODE_DEFAULT_KEY, DEFAULT_REGISTER_MODE_INTERFACE));
registerMode = registryURL.getParameter(REGISTER_MODE_KEY, ConfigurationUtils.getDynamicGlobalConfiguration().getString(DUBBO_REGISTER_MODE_DEFAULT_KEY, DEFAULT_REGISTER_MODE_INTERFACE));
if (!isValidRegisterMode(registerMode)) {
registerMode = DEFAULT_REGISTER_MODE_INTERFACE;
}
@ -248,10 +251,13 @@ public class ConfigValidationUtils {
result.add(registryURL);
}
}
FrameworkStatusReporter.reportRegistrationStatus(createRegistrationReport(registerMode));
} else {
result.add(registryURL);
}
});
return result;
}

2
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/DynamicConfigurationServiceNameMapping.java
View File

@ -40,8 +40,6 @@ import static org.apache.dubbo.rpc.model.ApplicationModel.getName;
*/
public class DynamicConfigurationServiceNameMapping implements ServiceNameMapping {
public static String DEFAULT_MAPPING_GROUP = "mapping";
private static final List<String> IGNORED_SERVICE_INTERFACES = asList(MetadataService.class.getName());
private final Logger logger = LoggerFactory.getLogger(getClass());

2
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/ServiceNameMapping.java
View File

@ -27,7 +27,6 @@ import static org.apache.dubbo.common.constants.CommonConstants.DUBBO;
import static org.apache.dubbo.common.constants.CommonConstants.PROTOCOL_KEY;
import static org.apache.dubbo.common.extension.ExtensionLoader.getExtensionLoader;
import static org.apache.dubbo.common.utils.StringUtils.SLASH;
import static org.apache.dubbo.metadata.DynamicConfigurationServiceNameMapping.DEFAULT_MAPPING_GROUP;
/**
* The interface for Dubbo service name Mapping
@ -36,6 +35,7 @@ import static org.apache.dubbo.metadata.DynamicConfigurationServiceNameMapping.D
*/
@SPI("config")
public interface ServiceNameMapping {
String DEFAULT_MAPPING_GROUP = "mapping";
/**
* Map the specified Dubbo service interface, group, version and protocol to current Dubbo service name

3
dubbo-metadata/dubbo-metadata-api/src/main/java/org/apache/dubbo/metadata/WritableMetadataService.java
View File

@ -21,6 +21,7 @@ import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.rpc.model.ApplicationModel;
import java.util.Map;
import java.util.Set;
import static org.apache.dubbo.common.extension.ExtensionLoader.getExtensionLoader;
@ -93,6 +94,8 @@ public interface WritableMetadataService extends MetadataService {
Set<String> removeCachedMapping(String serviceKey);
Map<String, Set<String>> getCachedMapping();
MetadataInfo getDefaultMetadataInfo();
/**

9
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/AbstractServiceDiscoveryFactory.java
View File

@ -18,6 +18,9 @@ package org.apache.dubbo.registry.client;
import org.apache.dubbo.common.URL;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -30,7 +33,11 @@ import java.util.concurrent.ConcurrentMap;
*/
public abstract class AbstractServiceDiscoveryFactory implements ServiceDiscoveryFactory {
private final ConcurrentMap<String, ServiceDiscovery> discoveries = new ConcurrentHashMap<>();
private static final ConcurrentMap<String, ServiceDiscovery> discoveries = new ConcurrentHashMap<>();
public static List<ServiceDiscovery> getAllServiceDiscoveries() {
return Collections.unmodifiableList(new LinkedList<>(discoveries.values()));
}
@Override
public ServiceDiscovery getServiceDiscovery(URL registryURL) {

10
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/DefaultServiceInstance.java
View File

@ -40,6 +40,8 @@ public class DefaultServiceInstance implements ServiceInstance {
private static final long serialVersionUID = 1149677083747278100L;
private String rawAddress;
private String serviceName;
private String host;
@ -87,6 +89,10 @@ public class DefaultServiceInstance implements ServiceInstance {
this.healthy = true;
}
public void setRawAddress(String rawAddress) {
this.rawAddress = rawAddress;
}
public DefaultServiceInstance(String serviceName) {
this.serviceName = serviceName;
}
@ -249,6 +255,10 @@ public class DefaultServiceInstance implements ServiceInstance {
@Override
public String toString() {
return rawAddress == null ? toFullString() : rawAddress;
}
public String toFullString() {
return "DefaultServiceInstance{" +
", serviceName='" + serviceName + '\'' +
", host='" + host + '\'' +

2
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/ServiceDiscoveryRegistryDirectory.java
View File

@ -103,7 +103,7 @@ public class ServiceDiscoveryRegistryDirectory<T> extends DynamicDirectory<T> im
}
/**
* This implementation wants to make sure all application names related to serviceListener received address notification.
* This implementation makes sure all application names related to serviceListener received address notification.
*
* FIXME, make sure deprecated "interface-application" mapping item be cleared in time.
*/

5
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/metadata/store/InMemoryWritableMetadataService.java
View File

@ -316,6 +316,11 @@ public class InMemoryWritableMetadataService implements WritableMetadataService
return serviceToAppsMapping.remove(serviceKey);
}
@Override
public Map<String, Set<String>> getCachedMapping() {
return serviceToAppsMapping;
}
@Override
public void setMetadataServiceURL(URL url) {
this.metadataServiceURL = url;

38
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/DefaultMigrationAddressComparator.java
View File

@ -27,31 +27,44 @@ import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class DefaultMigrationAddressComparator implements MigrationAddressComparator {
private static final Logger logger = LoggerFactory.getLogger(DefaultMigrationAddressComparator.class);
private static final String MIGRATION_THRESHOLD = "dubbo.application.migration.threshold";
private static final String DEFAULT_THRESHOLD_STRING = "0.8";
private static final float DEFAULT_THREAD = 0.8f;
private static final String DEFAULT_THRESHOLD_STRING = "1.0";
private static final float DEFAULT_THREAD = 1.0f;
public static final String OLD_ADDRESS_SIZE = "OLD_ADDRESS_SIZE";
public static final String NEW_ADDRESS_SIZE = "NEW_ADDRESS_SIZE";
private static final WritableMetadataService localMetadataService = WritableMetadataService.getDefaultExtension();
private Map<String, Map<String, Integer>> serviceMigrationData = new ConcurrentHashMap<>();
@Override
public <T> boolean shouldMigrate(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker, MigrationRule rule) {
Map<String, Integer> migrationData = serviceMigrationData.computeIfAbsent(invoker.getUrl().getDisplayServiceKey(), _k -> new ConcurrentHashMap<>());
if (!serviceDiscoveryInvoker.hasProxyInvokers()) {
migrationData.put(OLD_ADDRESS_SIZE, getAddressSize(invoker));
migrationData.put(NEW_ADDRESS_SIZE, -1);
logger.info("No instance address available, stop compare.");
return false;
}
if (!invoker.hasProxyInvokers()) {
migrationData.put(OLD_ADDRESS_SIZE, -1);
migrationData.put(NEW_ADDRESS_SIZE, getAddressSize(serviceDiscoveryInvoker));
logger.info("No interface address available, stop compare.");
return true;
}
List<Invoker<T>> invokers1 = serviceDiscoveryInvoker.getDirectory().getAllInvokers();
List<Invoker<T>> invokers2 = invoker.getDirectory().getAllInvokers();
int newAddressSize = getAddressSize(serviceDiscoveryInvoker);
int oldAddressSize = getAddressSize(invoker);
int newAddressSize = CollectionUtils.isNotEmpty(invokers1) ? invokers1.size() : 0;
int oldAddressSize = CollectionUtils.isNotEmpty(invokers2) ? invokers2.size() : 0;
migrationData.put(OLD_ADDRESS_SIZE, oldAddressSize);
migrationData.put(NEW_ADDRESS_SIZE, newAddressSize);
String rawThreshold = null;
String serviceKey = invoker.getUrl().getDisplayServiceKey();
@ -82,4 +95,17 @@ public class DefaultMigrationAddressComparator implements MigrationAddressCompar
}
return false;
}
private <T> int getAddressSize(ClusterInvoker<T> invoker) {
if (invoker == null) {
return -1;
}
List<Invoker<T>> invokers = invoker.getDirectory().getAllInvokers();
return CollectionUtils.isNotEmpty(invokers) ? invokers.size() : 0;
}
public Map<String, Integer> getAddressSize(String displayServiceKey) {
return serviceMigrationData.get(displayServiceKey);
}
}

3
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationAddressComparator.java
View File

@ -20,7 +20,10 @@ import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
import java.util.Map;
@SPI
public interface MigrationAddressComparator {
<T> boolean shouldMigrate(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker, MigrationRule rule);
Map<String, Integer> getAddressSize(String displayServiceKey);
}

55
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationInvoker.java
View File

@ -20,6 +20,7 @@ import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.status.reporter.FrameworkStatusReporter;
import org.apache.dubbo.common.utils.CollectionUtils;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.Registry;
@ -40,6 +41,7 @@ import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import static org.apache.dubbo.common.status.reporter.FrameworkStatusReporter.createConsumptionReport;
import static org.apache.dubbo.rpc.cluster.Constants.REFER_KEY;
public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
@ -59,7 +61,6 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
private volatile MigrationRule rule;
private volatile boolean migrated;
private static final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
public MigrationInvoker(RegistryProtocol registryProtocol,
@ -87,6 +88,11 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
this.type = type;
this.url = url;
this.consumerUrl = consumerUrl;
ConsumerModel consumerModel = ApplicationModel.getConsumerModel(consumerUrl.getServiceKey());
if (consumerModel != null) {
consumerModel.getServiceMetadata().addAttribute("currentClusterInvoker", this);
}
}
public ClusterInvoker<T> getInvoker() {
@ -139,9 +145,16 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
@Override
public void fallbackToInterfaceInvoker() {
migrated = false;
refreshInterfaceInvoker();
setListener(invoker, () -> {
this.destroyServiceDiscoveryInvoker();
if (!migrated) {
migrated = true;
this.destroyServiceDiscoveryInvoker();
FrameworkStatusReporter.reportConsumptionStatus(
createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "interface")
);
}
});
}
@ -151,6 +164,8 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
fallbackToInterfaceInvoker();
return;
}
migrated = false;
if (!forceMigrate) {
refreshServiceDiscoveryInvoker();
refreshInterfaceInvoker();
@ -163,7 +178,13 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
} else {
refreshServiceDiscoveryInvoker();
setListener(serviceDiscoveryInvoker, () -> {
this.destroyInterfaceInvoker();
if (!migrated) {
migrated = true;
this.destroyInterfaceInvoker();
FrameworkStatusReporter.reportConsumptionStatus(
createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app")
);
}
});
}
}
@ -228,6 +249,10 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
if (serviceDiscoveryInvoker != null) {
serviceDiscoveryInvoker.destroy();
}
ConsumerModel consumerModel = ApplicationModel.getConsumerModel(consumerUrl.getServiceKey());
if (consumerModel != null) {
consumerModel.getServiceMetadata().getAttributeMap().remove("currentClusterInvoker");
}
}
@Override
@ -329,6 +354,9 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
if (invoker.getDirectory().isNotificationReceived()) {
destroyInterfaceInvoker();
migrated = true;
FrameworkStatusReporter.reportConsumptionStatus(
createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app_app")
);
}
});
} else {
@ -337,6 +365,9 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
if (serviceDiscoveryInvoker.getDirectory().isNotificationReceived()) {
destroyServiceDiscoveryInvoker();
migrated = true;
FrameworkStatusReporter.reportConsumptionStatus(
createConsumptionReport(consumerUrl.getServiceInterface(), consumerUrl.getVersion(), consumerUrl.getGroup(), "app_interface")
);
}
});
}
@ -353,9 +384,6 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
serviceDiscoveryInvoker.destroy();
serviceDiscoveryInvoker = null;
}
updateConsumerModel(currentAvailableInvoker, serviceDiscoveryInvoker);
migrated = true;
}
// protected synchronized void discardServiceDiscoveryInvokerAddress(ClusterInvoker<T> serviceDiscoveryInvoker) {
@ -405,9 +433,6 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
invoker.destroy();
invoker = null;
}
updateConsumerModel(currentAvailableInvoker, invoker);
migrated = true;
}
//
@ -447,16 +472,4 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
public boolean checkInvokerAvailable(ClusterInvoker<T> invoker) {
return invoker != null && !invoker.isDestroyed() && invoker.isAvailable();
}
private void updateConsumerModel(ClusterInvoker<?> workingInvoker, ClusterInvoker<?> backInvoker) {
ConsumerModel consumerModel = ApplicationModel.getConsumerModel(consumerUrl.getServiceKey());
if (consumerModel != null) {
if (workingInvoker != null) {
consumerModel.getServiceMetadata().addAttribute("currentClusterInvoker", workingInvoker);
}
if (backInvoker != null) {
consumerModel.getServiceMetadata().addAttribute("backupClusterInvoker", backInvoker);
}
}
}
}

6
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/integration/RegistryDirectory.java
View File

@ -236,10 +236,10 @@ public class RegistryDirectory<T> extends DynamicDirectory<T> implements NotifyL
} catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
// notify invokers refreshed
this.invokersChanged();
// notify invokers refreshed
this.invokersChanged();
}
}
private List<Invoker<T>> toMergeInvokerList(List<Invoker<T>> invokers) {