map = yaml.load(rawRule);
TagRouterRule rule = TagRouterRule.parseFromMap(map);
rule.setRawRule(rawRule);
diff --git a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvoker.java b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvoker.java
index 02a96e810b..445aaf236b 100644
--- a/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvoker.java
+++ b/dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/support/wrapper/ScopeClusterInvoker.java
@@ -39,7 +39,6 @@ import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.dubbo.rpc.Constants.GENERIC_KEY;
-import static org.apache.dubbo.rpc.Constants.LOCAL_KEY;
import static org.apache.dubbo.rpc.Constants.LOCAL_PROTOCOL;
import static org.apache.dubbo.rpc.Constants.SCOPE_KEY;
import static org.apache.dubbo.rpc.Constants.SCOPE_REMOTE;
@@ -47,24 +46,23 @@ import static org.apache.dubbo.rpc.Constants.SCOPE_LOCAL;
import static org.apache.dubbo.rpc.cluster.Constants.PEER_KEY;
/**
- * A ClusterInvoker that selects between local and remote invokers at runtime.
+ * ScopeClusterInvoker is a cluster invoker which handles the invocation logic of a single service in a specific scope.
+ *
+ * It selects between local and remote invoker at runtime.
+ *
+ * @param the type of service interface
*/
public class ScopeClusterInvoker implements ClusterInvoker, ExporterChangeListener {
-
+ private final Object createLock = new Object();
private Protocol protocolSPI;
private final Directory directory;
private final Invoker invoker;
private final AtomicBoolean isExported;
private volatile Invoker injvmInvoker;
private volatile InjvmExporterListener injvmExporterListener;
-
private boolean peerFlag;
-
private boolean injvmFlag;
- private final Object createLock = new Object();
-
-
public ScopeClusterInvoker(Directory directory, Invoker invoker) {
this.directory = directory;
this.invoker = invoker;
@@ -72,30 +70,6 @@ public class ScopeClusterInvoker implements ClusterInvoker, ExporterChange
init();
}
- private void init() {
- Boolean peer = (Boolean) getUrl().getAttribute(PEER_KEY);
- String isInjvm = getUrl().getParameter(LOCAL_PROTOCOL);
- if (peer != null && peer) {
- peerFlag = true;
- return;
- }
- if (injvmInvoker == null && LOCAL_PROTOCOL.equalsIgnoreCase(getRegistryUrl().getProtocol())) {
- injvmInvoker = invoker;
- isExported.compareAndSet(false, true);
- injvmFlag = true;
- return;
- }
- if (Boolean.TRUE.toString().equalsIgnoreCase(isInjvm) || LOCAL_KEY.equalsIgnoreCase(getUrl().getParameter(SCOPE_KEY))) {
- injvmFlag = true;
- } else if (isInjvm == null) {
- injvmFlag = isNotRemoteOrGeneric();
- }
-
- protocolSPI = getUrl().getApplicationModel().getExtensionLoader(Protocol.class).getAdaptiveExtension();
- injvmExporterListener = getUrl().getOrDefaultFrameworkModel().getBeanFactory().getBean(InjvmExporterListener.class);
- injvmExporterListener.addExporterChangeListener(this, getUrl().getServiceKey());
- }
-
@Override
public URL getUrl() {
return directory.getConsumerUrl();
@@ -135,62 +109,28 @@ public class ScopeClusterInvoker implements ClusterInvoker, ExporterChange
return directory.getInterface();
}
+ /**
+ * Checks if the current ScopeClusterInvoker is exported to the local JVM and invokes the corresponding Invoker.
+ * If it's not exported locally, then it delegates the invocation to the original Invoker.
+ *
+ * @param invocation the invocation to be performed
+ * @return the result of the invocation
+ * @throws RpcException if there was an error during the invocation
+ */
@Override
public Result invoke(Invocation invocation) throws RpcException {
if (peerFlag) {
+ // If it's a point-to-point direct connection, invoke the original Invoker
return invoker.invoke(invocation);
}
if (isInjvmExported()) {
+ // If it's exported to the local JVM, invoke the corresponding Invoker
return injvmInvoker.invoke(invocation);
}
+ // Otherwise, delegate the invocation to the original Invoker
return invoker.invoke(invocation);
}
- private boolean isNotRemoteOrGeneric() {
- return !SCOPE_REMOTE.equalsIgnoreCase(getUrl().getParameter(SCOPE_KEY)) &&
- !getUrl().getParameter(GENERIC_KEY, false);
- }
-
- private boolean isInjvmExported() {
- Boolean localInvoke = RpcContext.getServiceContext().getLocalInvoke();
- boolean isExportedValue = isExported.get();
- boolean local = (localInvoke != null && localInvoke);
- // Determine whether this call is local
- if (isExportedValue && local) {
- return true;
- }
-
- // Determine whether this call is remote
- if (localInvoke != null && !localInvoke) {
- return false;
- }
-
- // When calling locally, determine whether it does not meet the requirements
- if (!isExportedValue && (SCOPE_LOCAL.equalsIgnoreCase(getUrl().getParameter(SCOPE_KEY)) ||
- Boolean.TRUE.toString().equalsIgnoreCase(getUrl().getParameter(LOCAL_PROTOCOL))|| local)) {
- throw new RpcException("Local service has not been exposed yet!");
- }
-
- return isExportedValue && injvmFlag;
- }
-
-
- private void createInjvmInvoker() {
- if (injvmInvoker == null) {
- synchronized (createLock) {
- if (injvmInvoker == null) {
- URL url = new ServiceConfigURL(LOCAL_PROTOCOL, NetUtils.getLocalHost(), getUrl().getPort(), getInterface().getName(), getUrl().getParameters());
- url = url.setScopeModel(getUrl().getScopeModel());
- url = url.setServiceModel(getUrl().getServiceModel());
- Invoker invoker = protocolSPI.refer(getInterface(), url);
- List> invokers = new ArrayList<>();
- invokers.add(invoker);
- injvmInvoker = Cluster.getCluster(url.getScopeModel(), Cluster.DEFAULT, false).join(new StaticDirectory(url, invokers), true);
- }
- }
- }
- }
-
@Override
public void onExporterChangeExport(Exporter exporter) {
if (isExported.get()) {
@@ -212,6 +152,103 @@ public class ScopeClusterInvoker implements ClusterInvoker, ExporterChange
}
}
+ public Invoker getInvoker() {
+ return invoker;
+ }
+
+ /**
+ * Initializes the ScopeClusterInvoker instance.
+ */
+ private void init() {
+ Boolean peer = (Boolean) getUrl().getAttribute(PEER_KEY);
+ String isInjvm = getUrl().getParameter(LOCAL_PROTOCOL);
+ // When the point-to-point direct connection is directly connected,
+ // the initialization is directly ended
+ if (peer != null && peer) {
+ peerFlag = true;
+ return;
+ }
+ // Check if the service has been exported through Injvm protocol
+ if (injvmInvoker == null && LOCAL_PROTOCOL.equalsIgnoreCase(getRegistryUrl().getProtocol())) {
+ injvmInvoker = invoker;
+ isExported.compareAndSet(false, true);
+ injvmFlag = true;
+ return;
+ }
+ // Check if the service has been exported through Injvm protocol or the SCOPE_LOCAL parameter is set
+ if (Boolean.TRUE.toString().equalsIgnoreCase(isInjvm) || SCOPE_LOCAL.equalsIgnoreCase(getUrl().getParameter(SCOPE_KEY))) {
+ injvmFlag = true;
+ } else if (isInjvm == null) {
+ injvmFlag = isNotRemoteOrGeneric();
+ }
+
+ protocolSPI = getUrl().getApplicationModel().getExtensionLoader(Protocol.class).getAdaptiveExtension();
+ injvmExporterListener = getUrl().getOrDefaultFrameworkModel().getBeanFactory().getBean(InjvmExporterListener.class);
+ injvmExporterListener.addExporterChangeListener(this, getUrl().getServiceKey());
+ }
+
+ /**
+ * Check if the service is a generalized call or the SCOPE_REMOTE parameter is set
+ *
+ * @return boolean
+ */
+ private boolean isNotRemoteOrGeneric() {
+ return !SCOPE_REMOTE.equalsIgnoreCase(getUrl().getParameter(SCOPE_KEY)) &&
+ !getUrl().getParameter(GENERIC_KEY, false);
+ }
+
+ /**
+ * Checks whether the current ScopeClusterInvoker is exported to the local JVM and returns a boolean value.
+ *
+ * @return true if the ScopeClusterInvoker is exported to the local JVM, false otherwise
+ * @throws RpcException if there was an error during the invocation
+ */
+ private boolean isInjvmExported() {
+ Boolean localInvoke = RpcContext.getServiceContext().getLocalInvoke();
+ boolean isExportedValue = isExported.get();
+ boolean local = (localInvoke != null && localInvoke);
+ // Determine whether this call is local
+ if (isExportedValue && local) {
+ return true;
+ }
+
+ // Determine whether this call is remote
+ if (localInvoke != null && !localInvoke) {
+ return false;
+ }
+
+ // When calling locally, determine whether it does not meet the requirements
+ if (!isExportedValue && (SCOPE_LOCAL.equalsIgnoreCase(getUrl().getParameter(SCOPE_KEY)) ||
+ Boolean.TRUE.toString().equalsIgnoreCase(getUrl().getParameter(LOCAL_PROTOCOL)) || local)) {
+ // If it's supposed to be exported to the local JVM ,but it's not, throw an exception
+ throw new RpcException("Local service for " + getUrl().getServiceInterface() + " has not been exposed yet!");
+ }
+
+ return isExportedValue && injvmFlag;
+ }
+
+ /**
+ * Creates a new Invoker for the current ScopeClusterInvoker and exports it to the local JVM.
+ */
+ private void createInjvmInvoker() {
+ if (injvmInvoker == null) {
+ synchronized (createLock) {
+ if (injvmInvoker == null) {
+ URL url = new ServiceConfigURL(LOCAL_PROTOCOL, NetUtils.getLocalHost(), getUrl().getPort(), getInterface().getName(), getUrl().getParameters());
+ url = url.setScopeModel(getUrl().getScopeModel());
+ url = url.setServiceModel(getUrl().getServiceModel());
+ Invoker invoker = protocolSPI.refer(getInterface(), url);
+ List> invokers = new ArrayList<>();
+ invokers.add(invoker);
+ injvmInvoker = Cluster.getCluster(url.getScopeModel(), Cluster.DEFAULT, false).join(new StaticDirectory(url, invokers), true);
+ }
+ }
+ }
+ }
+
+ /**
+ * Destroy the existing InjvmInvoker.
+ */
private void destroyInjvmInvoker() {
if (injvmInvoker != null) {
injvmInvoker.destroy();
@@ -219,7 +256,4 @@ public class ScopeClusterInvoker implements ClusterInvoker, ExporterChange
}
}
- public Invoker getInvoker() {
- return invoker;
- }
}
diff --git a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster
index 5ff23dc553..75dbd9f694 100644
--- a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster
+++ b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.Cluster
@@ -8,4 +8,4 @@ forking=org.apache.dubbo.rpc.cluster.support.ForkingCluster
available=org.apache.dubbo.rpc.cluster.support.AvailableCluster
mergeable=org.apache.dubbo.rpc.cluster.support.MergeableCluster
broadcast=org.apache.dubbo.rpc.cluster.support.BroadcastCluster
-zone-aware=org.apache.dubbo.rpc.cluster.support.registry.ZoneAwareCluster
+zone-aware=org.apache.dubbo.rpc.cluster.support.registry.ZoneAwareCluster
\ No newline at end of file
diff --git a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.filter.ClusterFilter b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.filter.ClusterFilter
index b21c3d843f..cd0a2f44e8 100644
--- a/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.filter.ClusterFilter
+++ b/dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.filter.ClusterFilter
@@ -1,3 +1,5 @@
consumercontext=org.apache.dubbo.rpc.cluster.filter.support.ConsumerContextFilter
consumer-classloader=org.apache.dubbo.rpc.cluster.filter.support.ConsumerClassLoaderFilter
router-snapshot=org.apache.dubbo.rpc.cluster.router.RouterSnapshotFilter
+observationsender=org.apache.dubbo.rpc.cluster.filter.support.ObservationSenderFilter
+metricsClusterFilter=org.apache.dubbo.rpc.cluster.filter.support.MetricsClusterFilter
\ No newline at end of file
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/configurator/parser/ConfigParserTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/configurator/parser/ConfigParserTest.java
index a7e50304c9..dd30a0a037 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/configurator/parser/ConfigParserTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/configurator/parser/ConfigParserTest.java
@@ -22,6 +22,7 @@ import org.apache.dubbo.rpc.cluster.configurator.parser.model.ConfiguratorConfig
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.yaml.snakeyaml.LoaderOptions;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.SafeConstructor;
@@ -50,7 +51,7 @@ class ConfigParserTest {
@Test
void snakeYamlBasicTest() throws IOException {
try (InputStream yamlStream = this.getClass().getResourceAsStream("/ServiceNoApp.yml")) {
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Map map = yaml.load(yamlStream);
ConfiguratorConfig config = ConfiguratorConfig.parseFromMap(map);
Assertions.assertNotNull(config);
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
index 62787ccd2b..68ee46afb1 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java
@@ -19,15 +19,15 @@ package org.apache.dubbo.rpc.cluster.directory;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.Holder;
import org.apache.dubbo.common.utils.NetUtils;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.cluster.router.MockInvoker;
import org.apache.dubbo.rpc.cluster.router.condition.ConditionStateRouterFactory;
import org.apache.dubbo.rpc.cluster.router.state.BitList;
import org.apache.dubbo.rpc.cluster.router.state.StateRouter;
-
import org.apache.dubbo.rpc.model.ApplicationModel;
+
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -52,9 +52,9 @@ class StaticDirectoryTest {
List routers = new ArrayList();
routers.add(router);
List> originInvokers = new ArrayList>();
- Invoker invoker1 = new MockInvoker(URL.valueOf("dubbo://10.20.3.3:20880/com.foo.BarService"));
- Invoker invoker2 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"));
- Invoker invoker3 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"));
+ Invoker invoker1 = new MockInvoker(URL.valueOf("dubbo://10.20.3.3:20880/com.foo.BarService"), true);
+ Invoker invoker2 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"), true);
+ Invoker invoker3 = new MockInvoker(URL.valueOf("dubbo://" + NetUtils.getLocalHost() + ":20880/com.foo.BarService"), true);
originInvokers.add(invoker1);
originInvokers.add(invoker2);
originInvokers.add(invoker3);
@@ -62,10 +62,10 @@ class StaticDirectoryTest {
List> filteredInvokers = router.route(invokers.clone(), URL.valueOf("consumer://" + NetUtils.getLocalHost() + "/com.foo.BarService"), new RpcInvocation(), false, new Holder<>());
- ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+ ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
StaticDirectory staticDirectory = new StaticDirectory<>(filteredInvokers);
boolean isAvailable = staticDirectory.isAvailable();
- Assertions.assertTrue(!isAvailable);
+ Assertions.assertTrue(isAvailable);
List> newInvokers = staticDirectory.list(new MockDirInvocation());
Assertions.assertTrue(newInvokers.size() > 0);
staticDirectory.destroy();
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/AbstractObservationFilterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/AbstractObservationFilterTest.java
new file mode 100644
index 0000000000..2f293484b4
--- /dev/null
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/AbstractObservationFilterTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.rpc.cluster.filter;
+
+import io.micrometer.tracing.test.SampleTestRunner;
+import org.apache.dubbo.config.ApplicationConfig;
+import org.apache.dubbo.config.TracingConfig;
+import org.apache.dubbo.rpc.AppResponse;
+import org.apache.dubbo.rpc.BaseFilter;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.junit.jupiter.api.AfterEach;
+
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+abstract class AbstractObservationFilterTest extends SampleTestRunner {
+
+ ApplicationModel applicationModel;
+ RpcInvocation invocation;
+
+ BaseFilter filter;
+
+ Invoker invoker = mock(Invoker.class);
+
+ static final String INTERFACE_NAME = "org.apache.dubbo.MockInterface";
+ static final String METHOD_NAME = "mockMethod";
+ static final String GROUP = "mockGroup";
+ static final String VERSION = "1.0.0";
+
+ @AfterEach
+ public void teardown() {
+ if (applicationModel != null) {
+ applicationModel.destroy();
+ }
+ }
+
+ abstract BaseFilter createFilter(ApplicationModel applicationModel);
+
+ void setupConfig() {
+ ApplicationConfig config = new ApplicationConfig();
+ config.setName("MockObservations");
+
+ applicationModel = ApplicationModel.defaultModel();
+ applicationModel.getApplicationConfigManager().setApplication(config);
+
+ invocation = new RpcInvocation(new MockInvocation());
+ invocation.addInvokedInvoker(invoker);
+
+ applicationModel.getBeanFactory().registerBean(getObservationRegistry());
+ TracingConfig tracingConfig = new TracingConfig();
+ tracingConfig.setEnabled(true);
+ applicationModel.getApplicationConfigManager().setTracing(tracingConfig);
+
+ filter = createFilter(applicationModel);
+
+ given(invoker.invoke(invocation)).willReturn(new AppResponse("success"));
+
+ initParam();
+ }
+
+ private void initParam() {
+ invocation.setTargetServiceUniqueName(GROUP + "/" + INTERFACE_NAME + ":" + VERSION);
+ invocation.setMethodName(METHOD_NAME);
+ invocation.setParameterTypes(new Class[] {String.class});
+ }
+
+}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java
index 7f0bb6ff07..3a173c2ceb 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/DefaultFilterChainBuilderTest.java
@@ -49,7 +49,6 @@ class DefaultFilterChainBuilderTest {
};
Invoker invokerAfterBuild = defaultFilterChainBuilder.buildInvokerChain(invokerWithoutFilter, REFERENCE_FILTER_KEY, CONSUMER);
- Assertions.assertTrue(invokerAfterBuild instanceof AbstractInvoker);
// verify that if LogFilter is configured, LogFilter should exist in the filter chain
URL urlWithFilter = URL.valueOf("injvm://127.0.0.1/DemoService")
@@ -64,8 +63,6 @@ class DefaultFilterChainBuilderTest {
};
invokerAfterBuild = defaultFilterChainBuilder.buildInvokerChain(invokerWithFilter, REFERENCE_FILTER_KEY, CONSUMER);
Assertions.assertTrue(invokerAfterBuild instanceof FilterChainBuilder.CallbackRegistrationInvoker);
- Assertions.assertEquals(1, ((FilterChainBuilder.CallbackRegistrationInvoker) invokerAfterBuild).filters.size());
-
}
@Test
@@ -84,7 +81,7 @@ class DefaultFilterChainBuilderTest {
};
Invoker invokerAfterBuild = defaultFilterChainBuilder.buildInvokerChain(invokerWithoutFilter, REFERENCE_FILTER_KEY, CONSUMER);
- Assertions.assertTrue(invokerAfterBuild instanceof AbstractInvoker);
+// Assertions.assertTrue(invokerAfterBuild instanceof AbstractInvoker);
// verify that if LogFilter is configured, LogFilter should exist in the filter chain
URL urlWithFilter = URL.valueOf("dubbo://127.0.0.1:20880/DemoService")
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/MetricsClusterFilterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/MetricsClusterFilterTest.java
new file mode 100644
index 0000000000..629e3fc6f1
--- /dev/null
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/MetricsClusterFilterTest.java
@@ -0,0 +1,183 @@
+/*
+ * 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.rpc.cluster.filter;
+
+import org.apache.dubbo.common.URL;
+import org.apache.dubbo.common.constants.CommonConstants;
+import org.apache.dubbo.config.ApplicationConfig;
+import org.apache.dubbo.metrics.collector.DefaultMetricsCollector;
+import org.apache.dubbo.metrics.filter.MetricsFilter;
+import org.apache.dubbo.metrics.model.key.MetricsKey;
+import org.apache.dubbo.metrics.model.sample.CounterMetricSample;
+import org.apache.dubbo.metrics.model.sample.MetricSample;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcException;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.cluster.filter.support.MetricsClusterFilter;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+class MetricsClusterFilterTest {
+
+ private ApplicationModel applicationModel;
+ private MetricsFilter filter;
+ private MetricsClusterFilter metricsClusterFilter;
+ private DefaultMetricsCollector collector;
+ private RpcInvocation invocation;
+ private final Invoker invoker = mock(Invoker.class);
+
+ private static final String INTERFACE_NAME = "org.apache.dubbo.MockInterface";
+ private static final String METHOD_NAME = "mockMethod";
+ private static final String GROUP = "mockGroup";
+ private static final String VERSION = "1.0.0";
+ private String side;
+
+ private AtomicBoolean initApplication = new AtomicBoolean(false);
+
+
+ @BeforeEach
+ public void setup() {
+ ApplicationConfig config = new ApplicationConfig();
+ config.setName("MockMetrics");
+ //RpcContext.getContext().setAttachment("MockMetrics","MockMetrics");
+
+ applicationModel = ApplicationModel.defaultModel();
+ applicationModel.getApplicationConfigManager().setApplication(config);
+
+ invocation = new RpcInvocation();
+ filter = new MetricsFilter();
+
+ collector = applicationModel.getBeanFactory().getOrRegisterBean(DefaultMetricsCollector.class);
+ if(!initApplication.get()) {
+ collector.collectApplication(applicationModel);
+ initApplication.set(true);
+ }
+ filter.setApplicationModel(applicationModel);
+ side = CommonConstants.CONSUMER;
+ invocation.setInvoker(new TestMetricsInvoker(side));
+ RpcContext.getServiceContext().setUrl(URL.valueOf("test://test:11/test?accesslog=true&group=dubbo&version=1.1&side=" + side));
+
+ metricsClusterFilter = new MetricsClusterFilter();
+ metricsClusterFilter.setApplicationModel(applicationModel);
+ }
+
+ @AfterEach
+ public void teardown() {
+ applicationModel.destroy();
+ }
+
+ @Test
+ public void testNoProvider(){
+ testClusterFilterError(RpcException.FORBIDDEN_EXCEPTION,
+ MetricsKey.METRIC_REQUESTS_SERVICE_UNAVAILABLE_FAILED.getNameByType(CommonConstants.CONSUMER));
+ }
+
+ private void testClusterFilterError(int errorCode,String name){
+ collector.setCollectEnabled(true);
+ given(invoker.invoke(invocation)).willThrow(new RpcException(errorCode));
+ initParam();
+
+ Long count = 1L;
+
+ for (int i = 0; i < count; i++) {
+ try {
+ metricsClusterFilter.invoke(invoker, invocation);
+ } catch (Exception e) {
+ Assertions.assertTrue(e instanceof RpcException);
+ metricsClusterFilter.onError(e, invoker, invocation);
+ }
+ }
+ Map metricsMap = getMetricsMap();
+ Assertions.assertTrue(metricsMap.containsKey(name));
+
+ MetricSample sample = metricsMap.get(name);
+
+ Assertions.assertSame(((CounterMetricSample) sample).getValue().longValue(), count);
+ teardown();
+ }
+
+
+
+ private void initParam() {
+ invocation.setTargetServiceUniqueName(GROUP + "/" + INTERFACE_NAME + ":" + VERSION);
+ invocation.setMethodName(METHOD_NAME);
+ invocation.setParameterTypes(new Class[]{String.class});
+ }
+
+ private Map getMetricsMap() {
+ List samples = collector.collect();
+ List samples1 = new ArrayList<>();
+ for (MetricSample sample : samples) {
+ if (sample.getName().contains("dubbo.thread.pool")) {
+ continue;
+ }
+ samples1.add(sample);
+ }
+ return samples1.stream().collect(Collectors.toMap(MetricSample::getName, Function.identity()));
+ }
+
+ public class TestMetricsInvoker implements Invoker {
+
+ private String side;
+
+ public TestMetricsInvoker(String side) {
+ this.side = side;
+ }
+
+ @Override
+ public Class getInterface() {
+ return null;
+ }
+
+ @Override
+ public Result invoke(Invocation invocation) throws RpcException {
+ return null;
+ }
+
+ @Override
+ public URL getUrl() {
+ return URL.valueOf("test://test:11/test?accesslog=true&group=dubbo&version=1.1&side="+side);
+ }
+
+ @Override
+ public boolean isAvailable() {
+ return true;
+ }
+
+ @Override
+ public void destroy() {
+
+ }
+ }
+}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/MockInvocation.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/MockInvocation.java
new file mode 100644
index 0000000000..9bd89552e3
--- /dev/null
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/MockInvocation.java
@@ -0,0 +1,168 @@
+/*
+ * 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.rpc.cluster.filter;
+
+import org.apache.dubbo.rpc.AttachmentsAdapter;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.RpcInvocation;
+import org.apache.dubbo.rpc.model.ServiceModel;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.dubbo.common.constants.CommonConstants.DUBBO_VERSION_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.GROUP_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.PATH_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.TIMEOUT_KEY;
+import static org.apache.dubbo.common.constants.CommonConstants.VERSION_KEY;
+import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
+
+/**
+ * MockInvocation.java
+ */
+public class MockInvocation extends RpcInvocation {
+
+ private Map attachments;
+
+ public MockInvocation() {
+ attachments = new HashMap<>();
+ attachments.put(PATH_KEY, "dubbo");
+ attachments.put(GROUP_KEY, "dubbo");
+ attachments.put(VERSION_KEY, "1.0.0");
+ attachments.put(DUBBO_VERSION_KEY, "1.0.0");
+ attachments.put(TOKEN_KEY, "sfag");
+ attachments.put(TIMEOUT_KEY, "1000");
+ }
+
+ @Override
+ public String getTargetServiceUniqueName() {
+ return null;
+ }
+
+ @Override
+ public String getProtocolServiceKey() {
+ return null;
+ }
+
+ public String getMethodName() {
+ return "echo";
+ }
+
+ @Override
+ public String getServiceName() {
+ return "DemoService";
+ }
+
+ public Class[] getParameterTypes() {
+ return new Class[] {String.class};
+ }
+
+ public Object[] getArguments() {
+ return new Object[] {"aa"};
+ }
+
+ public Map getAttachments() {
+ return new AttachmentsAdapter.ObjectToStringMap(attachments);
+ }
+
+ @Override
+ public Map getObjectAttachments() {
+ return attachments;
+ }
+
+ @Override
+ public void setAttachment(String key, String value) {
+ setObjectAttachment(key, value);
+ }
+
+ @Override
+ public void setAttachment(String key, Object value) {
+ setObjectAttachment(key, value);
+ }
+
+ @Override
+ public void setObjectAttachment(String key, Object value) {
+ attachments.put(key, value);
+ }
+
+ @Override
+ public void setAttachmentIfAbsent(String key, String value) {
+ setObjectAttachmentIfAbsent(key, value);
+ }
+
+ @Override
+ public void setAttachmentIfAbsent(String key, Object value) {
+ setObjectAttachmentIfAbsent(key, value);
+ }
+
+ @Override
+ public void setObjectAttachmentIfAbsent(String key, Object value) {
+ attachments.put(key, value);
+ }
+
+ public Invoker getInvoker() {
+ return null;
+ }
+
+ @Override
+ public void setServiceModel(ServiceModel serviceModel) {
+
+ }
+
+ @Override
+ public ServiceModel getServiceModel() {
+ return null;
+ }
+
+ @Override
+ public Object put(Object key, Object value) {
+ return null;
+ }
+
+ @Override
+ public Object get(Object key) {
+ return null;
+ }
+
+ @Override
+ public Map getAttributes() {
+ return null;
+ }
+
+ public String getAttachment(String key) {
+ return (String) getObjectAttachments().get(key);
+ }
+
+ @Override
+ public Object getObjectAttachment(String key) {
+ return attachments.get(key);
+ }
+
+ public String getAttachment(String key, String defaultValue) {
+ return (String) getObjectAttachments().get(key);
+ }
+
+ @Override
+ public Object getObjectAttachment(String key, Object defaultValue) {
+ Object result = attachments.get(key);
+ if (result == null) {
+ return defaultValue;
+ }
+ return result;
+ }
+
+}
diff --git a/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/observation/ObservationSenderFilterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/ObservationSenderFilterTest.java
similarity index 92%
rename from dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/observation/ObservationSenderFilterTest.java
rename to dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/ObservationSenderFilterTest.java
index e25c2412b6..8166203400 100644
--- a/dubbo-metrics/dubbo-metrics-default/src/test/java/org/apache/dubbo/metrics/observation/ObservationSenderFilterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/filter/ObservationSenderFilterTest.java
@@ -15,21 +15,22 @@
* limitations under the License.
*/
-package org.apache.dubbo.metrics.observation;
+package org.apache.dubbo.rpc.cluster.filter;
import io.micrometer.common.KeyValues;
import io.micrometer.core.tck.MeterRegistryAssert;
+import io.micrometer.tracing.test.SampleTestRunner;
import io.micrometer.tracing.test.simple.SpansAssert;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.rpc.RpcContext;
-import org.apache.dubbo.rpc.cluster.filter.ClusterFilter;
+import org.apache.dubbo.rpc.cluster.filter.support.ObservationSenderFilter;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.assertj.core.api.BDDAssertions;
class ObservationSenderFilterTest extends AbstractObservationFilterTest {
@Override
- public SampleTestRunnerConsumer yourCode() {
+ public SampleTestRunner.SampleTestRunnerConsumer yourCode() {
return (buildingBlocks, meterRegistry) -> {
setupConfig();
setupAttachments();
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/LoadBalanceBaseTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/LoadBalanceBaseTest.java
index 4e71762ab0..564680472d 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/LoadBalanceBaseTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/loadbalance/LoadBalanceBaseTest.java
@@ -224,7 +224,7 @@ class LoadBalanceBaseTest {
@Override
public String toString() {
- return JsonUtils.getJson().toJson(this);
+ return JsonUtils.toJson(this);
}
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
index 57b3162695..b025e8f8e9 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java
@@ -18,7 +18,7 @@ package org.apache.dubbo.rpc.cluster.router.file;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;
-import org.apache.dubbo.metrics.event.GlobalMetricsEventMulticaster;
+import org.apache.dubbo.metrics.event.MetricsDispatcher;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
@@ -61,7 +61,7 @@ class FileRouterEngineTest {
@BeforeAll
public static void setUpBeforeClass() throws Exception {
- ApplicationModel.defaultModel().getBeanFactory().registerBean(GlobalMetricsEventMulticaster.class);
+ ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
System.setProperty(ENABLE_CONNECTIVITY_VALIDATION, "false");
}
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListenerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListenerTest.java
index d089d33a76..8619baf581 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListenerTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshAppRuleListenerTest.java
@@ -27,6 +27,7 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
+import org.yaml.snakeyaml.LoaderOptions;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.SafeConstructor;
@@ -132,7 +133,7 @@ class MeshAppRuleListenerTest {
List> rulesReceived = ruleCaptor.getValue();
assertEquals(2, rulesReceived.size());
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule1)));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule2)));
@@ -163,7 +164,7 @@ class MeshAppRuleListenerTest {
List> rulesReceived = ruleCaptor.getValue();
assertEquals(2, rulesReceived.size());
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule1)));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule2)));
@@ -219,7 +220,7 @@ class MeshAppRuleListenerTest {
List> rulesReceived = ruleCaptor.getValue();
assertEquals(2, rulesReceived.size());
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule1)));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule2)));
@@ -258,7 +259,7 @@ class MeshAppRuleListenerTest {
List> rulesReceived = ruleCaptor.getValue();
assertEquals(1, rulesReceived.size());
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule2)));
meshAppRuleListener.receiveConfigInfo(rule1 + "---\n" + rule4);
@@ -282,7 +283,7 @@ class MeshAppRuleListenerTest {
@Override
public void onRuleChange(String appName, List> rules) {
Assertions.assertEquals("demo-route", appName);
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Assertions.assertTrue(rules.contains(yaml.load(rule5)));
Assertions.assertTrue(rules.contains(yaml.load(rule6)));
count.incrementAndGet();
@@ -303,7 +304,7 @@ class MeshAppRuleListenerTest {
@Override
public void onRuleChange(String appName, List> rules) {
Assertions.assertEquals("demo-route", appName);
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Assertions.assertTrue(rules.contains(yaml.load(rule7)));
Assertions.assertTrue(rules.contains(yaml.load(rule8)));
count.incrementAndGet();
@@ -353,7 +354,7 @@ class MeshAppRuleListenerTest {
List> rulesReceived = ruleCaptor.getValue();
assertEquals(2, rulesReceived.size());
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule1)));
Assertions.assertTrue(rulesReceived.contains(yaml.load(rule2)));
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManagerTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManagerTest.java
index 4c07bddc27..ab609ec860 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManagerTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleManagerTest.java
@@ -27,6 +27,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import org.yaml.snakeyaml.LoaderOptions;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.SafeConstructor;
@@ -166,7 +167,7 @@ class MeshRuleManagerTest {
@Override
public void onRuleChange(String appName, List> rules) {
assertEquals("dubbo-demo", appName);
- Yaml yaml = new Yaml(new SafeConstructor());
+ Yaml yaml = new Yaml(new SafeConstructor(new LoaderOptions()));
assertTrue(rules.contains(yaml.load(rule1)));
assertTrue(rules.contains(yaml.load(rule2)));
diff --git a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
index 979218beae..35f2a7e9ae 100644
--- a/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
+++ b/dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/mesh/route/MeshRuleRouterTest.java
@@ -33,6 +33,7 @@ import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import org.yaml.snakeyaml.LoaderOptions;
import org.yaml.snakeyaml.Yaml;
import org.yaml.snakeyaml.constructor.SafeConstructor;
@@ -239,7 +240,7 @@ class MeshRuleRouterTest {
void testRuleChange() {
StandardMeshRuleRouter