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

[3.0] enhance migration rule: support interface configuration (#6883)

This commit is contained in:
ken.lj 2020-11-06 14:31:16 +08:00 committed by GitHub
parent 23f242293f
commit 26b916951d
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 146 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -20,6 +20,8 @@ import org.apache.dubbo.common.config.ConfigurationUtils;
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.common.utils.StringUtils;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
@ -32,7 +34,7 @@ public class DefaultMigrationAddressComparator implements MigrationAddressCompar
private static final float DEFAULT_THREAD = 0.8f;
@Override
public <T> boolean shouldMigrate(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker) {
public <T> boolean shouldMigrate(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker, MigrationRule rule) {
if (!serviceDiscoveryInvoker.isAvailable()) {
logger.info("No instance address available, will not migrate.");
return false;
@ -48,7 +50,8 @@ public class DefaultMigrationAddressComparator implements MigrationAddressCompar
int newAddressSize = CollectionUtils.isNotEmpty(invokers1) ? invokers1.size() : 0;
int oldAddressSize = CollectionUtils.isNotEmpty(invokers2) ? invokers2.size() : 0;
String rawThreshold = ConfigurationUtils.getDynamicProperty(MIGRATION_THRESHOLD, DEFAULT_THRESHOLD_STRING);
String rawThreshold = rule.getThreshold(invoker.getUrl().getServiceKey());
rawThreshold = StringUtils.isNotEmpty(rawThreshold) ? rawThreshold : ConfigurationUtils.getDynamicProperty(MIGRATION_THRESHOLD, DEFAULT_THRESHOLD_STRING);
float threshold;
try {
threshold = Float.parseFloat(rawThreshold);

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

@ -17,9 +17,10 @@
package org.apache.dubbo.registry.client.migration;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
@SPI
public interface MigrationAddressComparator {
<T> boolean shouldMigrate(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker);
<T> boolean shouldMigrate(ClusterInvoker<T> serviceDiscoveryInvoker, ClusterInvoker<T> invoker, MigrationRule rule);
}

5
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationClusterInvoker.java
View File

@ -17,6 +17,7 @@
package org.apache.dubbo.registry.client.migration;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
import org.apache.dubbo.registry.client.migration.model.MigrationStep;
import org.apache.dubbo.rpc.cluster.ClusterInvoker;
@ -33,6 +34,10 @@ public interface MigrationClusterInvoker<T> extends ClusterInvoker<T> {
void setMigrationStep(MigrationStep step);
MigrationRule getMigrationRule();
void setMigrationRule(MigrationRule rule);
boolean invokersChanged();
void fallbackToInterfaceInvoker();

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

@ -22,6 +22,7 @@ import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.registry.Registry;
import org.apache.dubbo.registry.client.migration.model.MigrationRule;
import org.apache.dubbo.registry.client.migration.model.MigrationStep;
import org.apache.dubbo.registry.integration.DynamicDirectory;
import org.apache.dubbo.registry.integration.RegistryProtocol;
@ -52,6 +53,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
private volatile ClusterInvoker<T> serviceDiscoveryInvoker;
private volatile ClusterInvoker<T> currentAvailableInvoker;
private volatile MigrationStep step;
private volatile MigrationRule rule;
public MigrationInvoker(RegistryProtocol registryProtocol,
Cluster cluster,
@ -248,6 +250,16 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
this.step = step;
}
@Override
public MigrationRule getMigrationRule() {
return rule;
}
@Override
public void setMigrationRule(MigrationRule rule) {
this.rule = rule;
}
@Override
public boolean invokersChanged() {
return invokersChanged;
@ -262,7 +274,7 @@ public class MigrationInvoker<T> implements MigrationClusterInvoker<T> {
}
Set<MigrationAddressComparator> detectors = ExtensionLoader.getExtensionLoader(MigrationAddressComparator.class).getSupportedExtensionInstances();
if (detectors != null && detectors.stream().allMatch(migrationDetector -> migrationDetector.shouldMigrate(serviceDiscoveryInvoker, invoker))) {
if (detectors != null && detectors.stream().allMatch(migrationDetector -> migrationDetector.shouldMigrate(serviceDiscoveryInvoker, invoker, rule))) {
discardInterfaceInvokerAddress(invoker);
} else {
discardServiceDiscoveryInvokerAddress(serviceDiscoveryInvoker);

18
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/MigrationRuleHandler.java
View File

@ -16,6 +16,7 @@
*/
package org.apache.dubbo.registry.client.migration;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.config.ConfigurationUtils;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.logger.Logger;
@ -33,9 +34,12 @@ public class MigrationRuleHandler<T> {
private MigrationClusterInvoker<T> migrationInvoker;
private MigrationStep currentStep;
private MigrationRule rule;
private URL consumerURL;
public MigrationRuleHandler(MigrationClusterInvoker<T> invoker) {
public MigrationRuleHandler(MigrationClusterInvoker<T> invoker, URL url) {
this.migrationInvoker = invoker;
this.consumerURL = url;
}
public void doMigrate(String rawRule) {
@ -48,8 +52,13 @@ public class MigrationRuleHandler<T> {
} else if (INIT.equals(rawRule)) {
step = Enum.valueOf(MigrationStep.class, ConfigurationUtils.getDynamicProperty(DUBBO_SERVICEDISCOVERY_MIGRATION, step.name()));
} else {
MigrationRule rule = MigrationRule.parse(rawRule);
step = rule.getStep();
try {
rule = MigrationRule.parse(rawRule);
setMigrationRule(rule);
step = rule.getStep(consumerURL.getServiceKey());
} catch (Exception e) {
logger.error("Parse migration rule error, will use default step " + step, e);
}
}
if (currentStep == null || currentStep != step) {
@ -70,6 +79,9 @@ public class MigrationRuleHandler<T> {
public void setCurrentStep(MigrationStep currentStep) {
this.currentStep = currentStep;
}
public void setMigrationRule(MigrationRule rule) {
this.migrationInvoker.setMigrationStep(currentStep);
}
}

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

@ -85,7 +85,7 @@ public class MigrationRuleListener implements RegistryProtocolListener, Configur
public synchronized void onRefer(RegistryProtocol registryProtocol, ClusterInvoker<?> invoker, URL url) {
MigrationInvoker<?> migrationInvoker = (MigrationInvoker<?>) invoker;
MigrationRuleHandler<?> migrationListener = new MigrationRuleHandler<>(migrationInvoker);
MigrationRuleHandler<?> migrationListener = new MigrationRuleHandler<>(migrationInvoker, url);
listeners.add(migrationListener);
migrationListener.doMigrate(rawRule);

47
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/model/InterfaceMigrationRule.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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.client.migration.model;
public class InterfaceMigrationRule {
private String serviceKey;
private MigrationStep step;
private String threshold;
public String getServiceKey() {
return serviceKey;
}
public void setServiceKey(String serviceKey) {
this.serviceKey = serviceKey;
}
public MigrationStep getStep() {
return step;
}
public void setStep(MigrationStep step) {
this.step = step;
}
public String getThreshold() {
return threshold;
}
public void setThreshold(String threshold) {
this.threshold = threshold;
}
}

58
dubbo-registry/dubbo-registry-api/src/main/java/org/apache/dubbo/registry/client/migration/model/MigrationRule.java
View File

@ -19,16 +19,31 @@ package org.apache.dubbo.registry.client.migration.model;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.Constructor;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* # key = demo-consumer.migration
* # group = DUBBO_SERVICEDISCOVERY_MIGRATION
* # content
* key: demo-consumer
* step: APPLICATION_FIRST
* threshold: 1.0
* interfaces:
* - serviceKey: DemoService:1.0.0
* threshold: 1.0
* step: APPLICATION_FIRST
* - serviceKey: GreetingService:1.0.0
* step: FORCE_APPLICATION
*/
public class MigrationRule {
private String key;
private MigrationStep step;
private String threshold;
private List<InterfaceMigrationRule> interfaces;
private transient Map<String, InterfaceMigrationRule> interfaceRules;
public String getKey() {
return key;
@ -38,14 +53,57 @@ public class MigrationRule {
this.key = key;
}
public MigrationStep getStep(String serviceKey) {
InterfaceMigrationRule rule = interfaceRules.get(serviceKey);
if (rule != null) {
return rule.getStep() == null ? step : rule.getStep();
}
return step;
}
public InterfaceMigrationRule getInterfaceRule(String serviceKey) {
return interfaceRules.get(serviceKey);
}
public MigrationStep getStep() {
return step;
}
public String getThreshold(String serviceKey) {
InterfaceMigrationRule rule = interfaceRules.get(serviceKey);
if (rule != null) {
return rule.getThreshold() == null ? threshold : rule.getThreshold();
}
return threshold;
}
public String getThreshold() {
return threshold;
}
public void setThreshold(String threshold) {
this.threshold = threshold;
}
public void setStep(MigrationStep step) {
this.step = step;
}
public List<InterfaceMigrationRule> getInterfaces() {
return interfaces;
}
public void setInterfaces(List<InterfaceMigrationRule> interfaces) {
this.interfaces = interfaces;
if (interfaces != null) {
this.interfaceRules = new HashMap<>();
interfaces.forEach(rule -> {
interfaceRules.put(rule.getServiceKey(), rule);
});
}
}
public static MigrationRule parse(String rawRule) {
Constructor constructor = new Constructor(MigrationRule.class);
Yaml yaml = new Yaml(constructor);