Separate the code for indicator monitoring

.artifacts

@@ -59,7 +59,7 @@ dubbo-metrics-prometheus
 dubbo-metrics-registry
 dubbo-metrics-config-center
 dubbo-metrics-netty
-dubbo-metrics-event
+dubbo-metrics-monitor
 dubbo-native
 dubbo-parent
 dubbo-plugin

dubbo-cluster/pom.xml

@@ -75,18 +75,6 @@
       <version>${project.parent.version}</version>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.apache.dubbo</groupId>
-      <artifactId>dubbo-metrics-registry</artifactId>
-      <version>${project.parent.version}</version>
-      <scope>compile</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.apache.dubbo</groupId>
-      <artifactId>dubbo-metrics-default</artifactId>
-      <version>${project.parent.version}</version>
-      <optional>true</optional>
-    </dependency>
     <dependency>
       <groupId>io.micrometer</groupId>
       <artifactId>micrometer-tracing-integration-test</artifactId>

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/AbstractDirectory.java

@@ -21,16 +21,14 @@
 import org.apache.dubbo.common.config.Configuration;
 import org.apache.dubbo.common.config.ConfigurationUtils;
 import org.apache.dubbo.common.constants.LoggerCodeConstants;
+import org.apache.dubbo.common.event.DubboEventBus;
 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.common.utils.ConcurrentHashSet;
 import org.apache.dubbo.common.utils.LockUtils;
 import org.apache.dubbo.common.utils.NetUtils;
 import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.metrics.event.MetricsEventBus;
-import org.apache.dubbo.metrics.model.key.MetricsKey;
-import org.apache.dubbo.metrics.registry.event.RegistryEvent;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.RpcContext;
@@ -46,7 +44,6 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -393,14 +390,13 @@ public void checkConnectivity() {
                                 checkConnectivity();
                             }
                         });
-                        MetricsEventBus.publish(RegistryEvent.refreshDirectoryEvent(
-                                applicationModel, getSummary(), getDirectoryMeta()));
+                        DubboEventBus.publish(
+                                new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
                     },
                     reconnectTaskPeriod,
                     TimeUnit.MILLISECONDS);
         }
-        MetricsEventBus.publish(
-                RegistryEvent.refreshDirectoryEvent(applicationModel, getSummary(), getDirectoryMeta()));
+        DubboEventBus.publish(new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
     }
 
     /**
@@ -416,8 +412,7 @@ public void refreshInvoker() {
                 refreshInvokerInternal();
             }
         });
-        MetricsEventBus.publish(
-                RegistryEvent.refreshDirectoryEvent(applicationModel, getSummary(), getDirectoryMeta()));
+        DubboEventBus.publish(new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
     }
 
     protected Map<String, String> getDirectoryMeta() {
@@ -452,8 +447,7 @@ public void addDisabledInvoker(Invoker<T> invoker) {
                 logger.info("Disable service address: " + invoker.getUrl() + ".");
             }
         });
-        MetricsEventBus.publish(
-                RegistryEvent.refreshDirectoryEvent(applicationModel, getSummary(), getDirectoryMeta()));
+        DubboEventBus.publish(new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
     }
 
     @Override
@@ -468,8 +462,7 @@ public void recoverDisabledInvoker(Invoker<T> invoker) {
                 }
             }
         });
-        MetricsEventBus.publish(
-                RegistryEvent.refreshDirectoryEvent(applicationModel, getSummary(), getDirectoryMeta()));
+        DubboEventBus.publish(new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
     }
 
     protected final void refreshRouter(BitList<Invoker<T>> newlyInvokers, Runnable switchAction) {
@@ -532,8 +525,7 @@ protected void setInvokers(BitList<Invoker<T>> invokers) {
             this.invokersInitialized = true;
         });
 
-        MetricsEventBus.publish(
-                RegistryEvent.refreshDirectoryEvent(applicationModel, getSummary(), getDirectoryMeta()));
+        DubboEventBus.publish(new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
     }
 
     protected void destroyInvokers() {
@@ -550,8 +542,7 @@ private boolean addValidInvoker(Invoker<T> invoker) {
         LockUtils.safeLock(invokerRefreshLock, LockUtils.DEFAULT_TIMEOUT, () -> {
             result.set(this.validInvokers.add(invoker));
         });
-        MetricsEventBus.publish(
-                RegistryEvent.refreshDirectoryEvent(applicationModel, getSummary(), getDirectoryMeta()));
+        DubboEventBus.publish(new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
         return result.get();
     }
 
@@ -560,8 +551,7 @@ private boolean removeValidInvoker(Invoker<T> invoker) {
         LockUtils.safeLock(invokerRefreshLock, LockUtils.DEFAULT_TIMEOUT, () -> {
             result.set(this.validInvokers.remove(invoker));
         });
-        MetricsEventBus.publish(
-                RegistryEvent.refreshDirectoryEvent(applicationModel, getSummary(), getDirectoryMeta()));
+        DubboEventBus.publish(new DirectoryRefreshEvent(applicationModel, getSummary(), getDirectoryMeta()));
         return result.get();
     }
 
@@ -581,14 +571,13 @@ protected String joinValidInvokerAddresses() {
                 .collect(Collectors.joining(","));
     }
 
-    private Map<MetricsKey, Map<String, Integer>> getSummary() {
-        Map<MetricsKey, Map<String, Integer>> summaryMap = new HashMap<>();
-
-        summaryMap.put(MetricsKey.DIRECTORY_METRIC_NUM_VALID, groupByServiceKey(getValidInvokers()));
-        summaryMap.put(MetricsKey.DIRECTORY_METRIC_NUM_DISABLE, groupByServiceKey(getDisabledInvokers()));
-        summaryMap.put(MetricsKey.DIRECTORY_METRIC_NUM_TO_RECONNECT, groupByServiceKey(getInvokersToReconnect()));
-        summaryMap.put(MetricsKey.DIRECTORY_METRIC_NUM_ALL, groupByServiceKey(getInvokers()));
-        return summaryMap;
+    private DirectoryRefreshEvent.Summary getSummary() {
+        DirectoryRefreshEvent.Summary summary = new DirectoryRefreshEvent.Summary();
+        summary.directoryNumValidMap = groupByServiceKey(getValidInvokers());
+        summary.directoryNumDisableMap = groupByServiceKey(getDisabledInvokers());
+        summary.directoryNumToReConnectMap = groupByServiceKey(getInvokersToReconnect());
+        summary.directoryNumAllMap = groupByServiceKey(getInvokers());
+        return summary;
     }
 
     private Map<String, Integer> groupByServiceKey(Collection<Invoker<T>> invokers) {

dubbo-cluster/src/main/java/org/apache/dubbo/rpc/cluster/directory/DirectoryRefreshEvent.java

@@ -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.rpc.cluster.directory;
+
+import org.apache.dubbo.common.event.DubboEvent;
+import org.apache.dubbo.common.utils.TimePair;
+import org.apache.dubbo.rpc.model.ApplicationModel;
+
+import java.util.Map;
+
+public class DirectoryRefreshEvent extends DubboEvent {
+
+    private final TimePair timePair;
+
+    private final Summary summary;
+
+    private final Map<String, String> attachments;
+
+    public DirectoryRefreshEvent(ApplicationModel applicationModel, Summary summary, Map<String, String> attachments) {
+        super(applicationModel);
+        this.summary = summary;
+        this.attachments = attachments;
+        this.timePair = TimePair.start();
+    }
+
+    public Summary getSummary() {
+        return summary;
+    }
+
+    public Map<String, String> getAttachments() {
+        return attachments;
+    }
+
+    public TimePair getTimePair() {
+        return timePair;
+    }
+
+    public static class Summary {
+
+        public Map<String, Integer> directoryNumValidMap;
+
+        public Map<String, Integer> directoryNumDisableMap;
+
+        public Map<String, Integer> directoryNumToReConnectMap;
+
+        public Map<String, Integer> directoryNumAllMap;
+    }
+}

dubbo-cluster/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.rpc.cluster.filter.ClusterFilter

@@ -1,4 +1,3 @@
 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
-metricsConsumerFilter=org.apache.dubbo.rpc.cluster.filter.support.MetricsConsumerFilter
+router-snapshot=org.apache.dubbo.rpc.cluster.router.RouterSnapshotFilter

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/directory/StaticDirectoryTest.java

@@ -19,14 +19,12 @@
 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.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 java.util.ArrayList;
 import java.util.List;
@@ -70,7 +68,6 @@ void testStaticDirectory() {
                 new RpcInvocation(),
                 false,
                 new Holder<>());
-        ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
         StaticDirectory<String> staticDirectory = new StaticDirectory<>(filteredInvokers);
         boolean isAvailable = staticDirectory.isAvailable();
         Assertions.assertTrue(isAvailable);

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/router/file/FileRouterEngineTest.java

@@ -18,7 +18,6 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.extension.ExtensionLoader;
-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;
@@ -31,7 +30,6 @@
 import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
 import org.apache.dubbo.rpc.cluster.router.state.StateRouterFactory;
 import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
-import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import javax.script.ScriptEngineManager;
 
@@ -63,7 +61,6 @@ class FileRouterEngineTest {
 
     @BeforeAll
     public static void setUpBeforeClass() throws Exception {
-        ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
         System.setProperty(ENABLE_CONNECTIVITY_VALIDATION, "false");
     }
 

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/AbstractClusterInvokerTest.java

@@ -20,7 +20,6 @@
 import org.apache.dubbo.common.extension.ExtensionLoader;
 import org.apache.dubbo.common.utils.NetUtils;
 import org.apache.dubbo.common.utils.StringUtils;
-import org.apache.dubbo.metrics.event.MetricsDispatcher;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.Result;
@@ -34,7 +33,6 @@
 import org.apache.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance;
 import org.apache.dubbo.rpc.cluster.loadbalance.RandomLoadBalance;
 import org.apache.dubbo.rpc.cluster.loadbalance.RoundRobinLoadBalance;
-import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -109,7 +107,6 @@ public static void afterClass() {
     @SuppressWarnings({"unchecked"})
     @BeforeEach
     public void setUp() throws Exception {
-        ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
         Map<String, Object> attributes = new HashMap<>();
         attributes.put("application", "abstractClusterInvokerTest");
         url = url.putAttribute(REFER_KEY, attributes);

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/ConnectivityValidationTest.java

@@ -19,7 +19,6 @@
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.constants.CommonConstants;
 import org.apache.dubbo.common.utils.CollectionUtils;
-import org.apache.dubbo.metrics.event.MetricsDispatcher;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.Result;
@@ -28,7 +27,6 @@
 import org.apache.dubbo.rpc.cluster.Directory;
 import org.apache.dubbo.rpc.cluster.LoadBalance;
 import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
-import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import java.util.Collections;
 import java.util.HashSet;
@@ -72,7 +70,6 @@ class ConnectivityValidationTest {
 
     @BeforeEach
     public void setup() {
-        ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
         invoker1 = Mockito.mock(Invoker.class);
         invoker2 = Mockito.mock(Invoker.class);
         invoker3 = Mockito.mock(Invoker.class);

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java

@@ -17,7 +17,6 @@
 package org.apache.dubbo.rpc.cluster.support;
 
 import org.apache.dubbo.common.URL;
-import org.apache.dubbo.metrics.event.MetricsDispatcher;
 import org.apache.dubbo.rpc.AppResponse;
 import org.apache.dubbo.rpc.AsyncRpcResult;
 import org.apache.dubbo.rpc.Invocation;
@@ -30,7 +29,6 @@
 import org.apache.dubbo.rpc.cluster.SingleRouterChain;
 import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
 import org.apache.dubbo.rpc.cluster.router.state.BitList;
-import org.apache.dubbo.rpc.model.ApplicationModel;
 import org.apache.dubbo.rpc.protocol.AbstractInvoker;
 
 import java.util.ArrayList;
@@ -67,7 +65,6 @@ class FailoverClusterInvokerTest {
      */
     @BeforeEach
     public void setUp() throws Exception {
-        ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
         dic = mock(Directory.class);
 
         given(dic.getUrl()).willReturn(url);

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/MockAbstractClusterInvokerTest.java

@@ -17,7 +17,6 @@
 package org.apache.dubbo.rpc.cluster.support;
 
 import org.apache.dubbo.common.URL;
-import org.apache.dubbo.metrics.event.MetricsDispatcher;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.Result;
@@ -26,7 +25,6 @@
 import org.apache.dubbo.rpc.RpcInvocation;
 import org.apache.dubbo.rpc.cluster.LoadBalance;
 import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
-import org.apache.dubbo.rpc.model.ApplicationModel;
 
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -89,7 +87,6 @@ public static void afterClass() {
     @SuppressWarnings({"unchecked"})
     @BeforeEach
     public void setUp() throws Exception {
-        ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
         Map<String, Object> attributes = new HashMap<>();
         attributes.put("application", "abstractClusterInvokerTest");
         url = url.putAttribute(REFER_KEY, attributes);

dubbo-cluster/src/test/java/org/apache/dubbo/rpc/cluster/support/wrapper/MockClusterInvokerTest.java

@@ -18,7 +18,6 @@
 
 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.extension.ExtensionLoader;
-import org.apache.dubbo.metrics.event.MetricsDispatcher;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.Protocol;
@@ -29,7 +28,6 @@
 import org.apache.dubbo.rpc.cluster.LoadBalance;
 import org.apache.dubbo.rpc.cluster.directory.StaticDirectory;
 import org.apache.dubbo.rpc.cluster.support.AbstractClusterInvoker;
-import org.apache.dubbo.rpc.model.ApplicationModel;
 import org.apache.dubbo.rpc.support.MockProtocol;
 
 import java.util.ArrayList;
@@ -51,7 +49,6 @@ class MockClusterInvokerTest {
 
     @BeforeEach
     public void beforeMethod() {
-        ApplicationModel.defaultModel().getBeanFactory().registerBean(MetricsDispatcher.class);
         invokers.clear();
     }