From 414608245329fbafd00a68dd1d53a0cd97ad3b6f Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Tue, 26 Sep 2023 18:55:08 +0800 Subject: [PATCH 1/6] Skip refresh registry config for internal service (#13134) --- .../bootstrap/builders/InternalServiceConfigBuilder.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/InternalServiceConfigBuilder.java b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/InternalServiceConfigBuilder.java index 5a2dbdbaf6..af37236120 100644 --- a/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/InternalServiceConfigBuilder.java +++ b/dubbo-config/dubbo-config-api/src/main/java/org/apache/dubbo/config/bootstrap/builders/InternalServiceConfigBuilder.java @@ -262,8 +262,11 @@ public class InternalServiceConfigBuilder { serviceConfig.setScopeModel(applicationModel.getInternalModule()); serviceConfig.setApplication(applicationConfig); - RegistryConfig registryConfig = new RegistryConfig("N/A"); + RegistryConfig registryConfig = new RegistryConfig(); + registryConfig.refresh(); + registryConfig.setNeedRefresh(false); registryConfig.setId(this.registryId); + registryConfig.setAddress("N/A"); registryConfig.setScopeModel(this.applicationModel); serviceConfig.setRegistry(registryConfig); From 26174f05486cb8b5f449bd4f2debf9837d1b1f17 Mon Sep 17 00:00:00 2001 From: songxiaosheng Date: Tue, 26 Sep 2023 20:32:54 +0800 Subject: [PATCH 2/6] :bug: destory xds exchanger null (#13086) --- .../org/apache/dubbo/registry/xds/XdsServiceDiscovery.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java index 91d6a72a73..b40b68650b 100644 --- a/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java +++ b/dubbo-xds/src/main/java/org/apache/dubbo/registry/xds/XdsServiceDiscovery.java @@ -59,6 +59,9 @@ public class XdsServiceDiscovery extends ReflectionBasedServiceDiscovery { @Override public void doDestroy() { try { + if (exchanger == null) { + return; + } exchanger.destroy(); } catch (Throwable t) { logger.error(REGISTRY_ERROR_INITIALIZE_XDS, "", "", t.getMessage(), t); From 0d88f6283033e2389cdd4c044da0d67a6f6c6ed1 Mon Sep 17 00:00:00 2001 From: KitwahSin <38373423+KitwahSin@users.noreply.github.com> Date: Wed, 27 Sep 2023 15:31:25 +0800 Subject: [PATCH 3/6] =?UTF-8?q?fix=EF=BC=9Aonnext=20callback=20method=20wo?= =?UTF-8?q?uld=20be=20invalid=20when=20the=20provider=20method=20with=20tr?= =?UTF-8?q?iple=20protocol=20is=20overloaded.=20(#13057)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * For the same method, rpcType.BI_STREAM and rpcType.SERVER_STREAM cannot exist at the same time. * remove unused imported item. * add exception test. * add licence comment. --------- Co-authored-by: Albumen Kevin --- .../model/ReflectionServiceDescriptor.java | 13 ++++++- .../ReflectionServiceDescriptorTest.java | 11 ++++++ .../dubbo/rpc/support/DemoService1.java | 25 ++++++++++++++ .../dubbo/rpc/support/DemoService1Impl.java | 34 +++++++++++++++++++ 4 files changed, 82 insertions(+), 1 deletion(-) create mode 100644 dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1.java create mode 100644 dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1Impl.java diff --git a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptor.java b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptor.java index 629d6c0385..489f7e4c6c 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptor.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptor.java @@ -75,7 +75,18 @@ public class ReflectionServiceDescriptor implements ServiceDescriptor { methods.forEach((methodName, methodList) -> { Map descMap = descToMethods.computeIfAbsent(methodName, k -> new HashMap<>()); - methodList.forEach(methodModel -> descMap.put(methodModel.getParamDesc(), methodModel)); + // not support BI_STREAM and SERVER_STREAM at the same time, for example, + // void foo(Request, StreamObserver) ---> SERVER_STREAM + // StreamObserver foo(StreamObserver) ---> BI_STREAM + long streamMethodCount = methodList.stream() + .peek(methodModel -> descMap.put(methodModel.getParamDesc(), methodModel)) + .map(MethodDescriptor::getRpcType) + .filter(rpcType -> rpcType == MethodDescriptor.RpcType.SERVER_STREAM + || rpcType == MethodDescriptor.RpcType.BI_STREAM) + .count(); + if (streamMethodCount > 1L) + throw new IllegalStateException("Stream method could not be overloaded.There are " + streamMethodCount + +" stream method signatures. method(" + methodName + ")"); }); } diff --git a/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptorTest.java b/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptorTest.java index 3361312e1f..b7011526be 100644 --- a/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptorTest.java +++ b/dubbo-common/src/test/java/org/apache/dubbo/rpc/model/ReflectionServiceDescriptorTest.java @@ -21,6 +21,7 @@ import org.apache.dubbo.common.utils.ReflectUtils; import org.apache.dubbo.metadata.definition.TypeDefinitionBuilder; import org.apache.dubbo.rpc.support.DemoService; +import org.apache.dubbo.rpc.support.DemoService1; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.Mockito; @@ -42,6 +43,16 @@ class ReflectionServiceDescriptorTest { Assertions.assertEquals(1, service2.getMethods("sayHello2").size()); } + @Test + void testStreamRpcTypeException() { + try { + new ReflectionServiceDescriptor(DemoService1.class); + } catch (IllegalStateException e) { + Assertions.assertTrue(e.getMessage() + .contains("Stream method could not be overloaded.")); + } + } + @Test void getFullServiceDefinition() { TypeDefinitionBuilder.initBuilders(new FrameworkModel()); diff --git a/dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1.java b/dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1.java new file mode 100644 index 0000000000..1251c7cce9 --- /dev/null +++ b/dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1.java @@ -0,0 +1,25 @@ +/* + * 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.support; + +import org.apache.dubbo.common.stream.StreamObserver; + +public interface DemoService1 { + StreamObserver sayHello(StreamObserver request); + + void sayHello(String msg, StreamObserver request); +} diff --git a/dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1Impl.java b/dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1Impl.java new file mode 100644 index 0000000000..7b5ec906fb --- /dev/null +++ b/dubbo-common/src/test/java/org/apache/dubbo/rpc/support/DemoService1Impl.java @@ -0,0 +1,34 @@ +/* + * 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.support; + +import org.apache.dubbo.common.stream.StreamObserver; + +public class DemoService1Impl implements DemoService1{ + @Override + public StreamObserver sayHello(StreamObserver request) { + request.onNext("BI_STREAM"); + return request; + } + + @Override + public void sayHello(String msg, StreamObserver request) { + request.onNext(msg); + request.onNext("SERVER_STREAM"); + request.onCompleted(); + } +} From 34c5bbf7137a496b1fd1f5adb2f290e999be91ff Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Thu, 28 Sep 2023 11:09:06 +0800 Subject: [PATCH 4/6] Support customize collector sync period (#13140) --- .../dubbo/common/constants/MetricsConstants.java | 2 ++ .../java/org/apache/dubbo/config/MetricsConfig.java | 13 +++++++++++++ .../metrics/report/AbstractMetricsReporter.java | 7 +++++-- 3 files changed, 20 insertions(+), 2 deletions(-) diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java index 4c4437d427..e891fe9ad3 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java @@ -51,6 +51,8 @@ public interface MetricsConstants { String ENABLE_COLLECTOR_SYNC_KEY = "enable.collector.sync"; + String COLLECTOR_SYNC_PERIOD_KEY = "collector.sync.period"; + String AGGREGATION_COLLECTOR_KEY = "aggregation"; String AGGREGATION_ENABLED_KEY = "aggregation.enabled"; diff --git a/dubbo-common/src/main/java/org/apache/dubbo/config/MetricsConfig.java b/dubbo-common/src/main/java/org/apache/dubbo/config/MetricsConfig.java index 98d05209a5..d1cf0c1d43 100644 --- a/dubbo-common/src/main/java/org/apache/dubbo/config/MetricsConfig.java +++ b/dubbo-common/src/main/java/org/apache/dubbo/config/MetricsConfig.java @@ -71,6 +71,11 @@ public class MetricsConfig extends AbstractConfig { */ private Boolean enableCollectorSync; + /** + * Collector sync period. + */ + private Integer collectorSyncPeriod; + /** * @deprecated After metrics config is refactored. * This parameter should no longer use and will be deleted in the future. @@ -248,6 +253,14 @@ public class MetricsConfig extends AbstractConfig { this.enableCollectorSync = enableCollectorSync; } + public Integer getCollectorSyncPeriod() { + return collectorSyncPeriod; + } + + public void setCollectorSyncPeriod(Integer collectorSyncPeriod) { + this.collectorSyncPeriod = collectorSyncPeriod; + } + public Boolean getUseGlobalRegistry() { return useGlobalRegistry; } diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java index 13c8cecdc4..70033771c5 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java @@ -53,6 +53,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_METRICS_COLLECTOR_EXCEPTION; +import static org.apache.dubbo.common.constants.MetricsConstants.COLLECTOR_SYNC_PERIOD_KEY; import static org.apache.dubbo.common.constants.MetricsConstants.ENABLE_COLLECTOR_SYNC_KEY; import static org.apache.dubbo.common.constants.MetricsConstants.ENABLE_JVM_METRICS_KEY; @@ -77,7 +78,7 @@ public abstract class AbstractMetricsReporter implements MetricsReporter { private ScheduledExecutorService collectorSyncJobExecutor = null; private static final int DEFAULT_SCHEDULE_INITIAL_DELAY = 5; - private static final int DEFAULT_SCHEDULE_PERIOD = 3; + private static final int DEFAULT_SCHEDULE_PERIOD = 60; protected AbstractMetricsReporter(URL url, ApplicationModel applicationModel) { this.url = url; @@ -143,9 +144,11 @@ public abstract class AbstractMetricsReporter implements MetricsReporter { private void scheduleMetricsCollectorSyncJob() { boolean enableCollectorSync = url.getParameter(ENABLE_COLLECTOR_SYNC_KEY, true); if (enableCollectorSync) { + int collectSyncPeriod = url.getParameter(COLLECTOR_SYNC_PERIOD_KEY, DEFAULT_SCHEDULE_PERIOD); + NamedThreadFactory threadFactory = new NamedThreadFactory("metrics-collector-sync-job", true); collectorSyncJobExecutor = Executors.newScheduledThreadPool(1, threadFactory); - collectorSyncJobExecutor.scheduleWithFixedDelay(this::refreshData, DEFAULT_SCHEDULE_INITIAL_DELAY, DEFAULT_SCHEDULE_PERIOD, TimeUnit.SECONDS); + collectorSyncJobExecutor.scheduleWithFixedDelay(this::refreshData, DEFAULT_SCHEDULE_INITIAL_DELAY, collectSyncPeriod, TimeUnit.SECONDS); } } From 9dc5bf1944d34cffc19d510d46123a20c9f3ecab Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Fri, 29 Sep 2023 09:44:41 +0800 Subject: [PATCH 5/6] Enhance spring security class check (#13136) * Enhance spring security class check * Catch exception --- ...thenticationExceptionTranslatorFilter.java | 6 ++++- ...textHolderAuthenticationPrepareFilter.java | 13 +++++++++-- ...extHolderAuthenticationResolverFilter.java | 10 ++++++-- .../model/SecurityScopeModelInitializer.java | 23 +++++++++++++++---- .../spring/security/utils/SecurityNames.java | 2 ++ 5 files changed, 44 insertions(+), 10 deletions(-) diff --git a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/AuthenticationExceptionTranslatorFilter.java b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/AuthenticationExceptionTranslatorFilter.java index 00dadabf59..e3a246d2bc 100644 --- a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/AuthenticationExceptionTranslatorFilter.java +++ b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/AuthenticationExceptionTranslatorFilter.java @@ -30,10 +30,14 @@ import org.springframework.security.core.AuthenticationException; import static org.apache.dubbo.rpc.RpcException.AUTHORIZATION_EXCEPTION; import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME; -@Activate(group = CommonConstants.PROVIDER, order = Integer.MAX_VALUE, onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME}) +@Activate(group = CommonConstants.PROVIDER, order = Integer.MAX_VALUE, onClass = { + SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME, + JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME}) public class AuthenticationExceptionTranslatorFilter implements Filter, Filter.Listener { diff --git a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationPrepareFilter.java b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationPrepareFilter.java index 6617f4be85..c71ef982f5 100644 --- a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationPrepareFilter.java +++ b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationPrepareFilter.java @@ -18,6 +18,8 @@ package org.apache.dubbo.spring.security.filter; import org.apache.dubbo.common.constants.CommonConstants; import org.apache.dubbo.common.extension.Activate; +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.rpc.Invocation; import org.apache.dubbo.rpc.Invoker; @@ -33,12 +35,17 @@ import org.springframework.security.core.context.SecurityContext; import org.springframework.security.core.context.SecurityContextHolder; import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME; -@Activate(group = CommonConstants.CONSUMER, order = -10000, onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME}) +@Activate(group = CommonConstants.CONSUMER, order = -10000, onClass = { + SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME, + JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME}) public class ContextHolderAuthenticationPrepareFilter implements ClusterFilter { + private final Logger logger = LoggerFactory.getLogger(getClass()); private final ObjectMapperCodec mapper; public ContextHolderAuthenticationPrepareFilter(ApplicationModel applicationModel) { @@ -47,7 +54,9 @@ public class ContextHolderAuthenticationPrepareFilter implements ClusterFilter { @Override public Result invoke(Invoker invoker, Invocation invocation) throws RpcException { - setSecurityContext(invocation); + if (this.mapper != null) { + setSecurityContext(invocation); + } return invoker.invoke(invocation); } diff --git a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationResolverFilter.java b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationResolverFilter.java index 9f2567185a..ff8a4fbea7 100644 --- a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationResolverFilter.java +++ b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/filter/ContextHolderAuthenticationResolverFilter.java @@ -32,10 +32,14 @@ import org.springframework.security.core.Authentication; import org.springframework.security.core.context.SecurityContextHolder; import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME; -@Activate(group = CommonConstants.PROVIDER, order = -10000, onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME}) +@Activate(group = CommonConstants.PROVIDER, order = -10000, onClass = { + SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME, + JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME}) public class ContextHolderAuthenticationResolverFilter implements Filter { private final ObjectMapperCodec mapper; @@ -46,7 +50,9 @@ public class ContextHolderAuthenticationResolverFilter implements Filter { @Override public Result invoke(Invoker invoker, Invocation invocation) throws RpcException { - getSecurityContext(invocation); + if (this.mapper != null) { + getSecurityContext(invocation); + } return invoker.invoke(invocation); } diff --git a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/model/SecurityScopeModelInitializer.java b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/model/SecurityScopeModelInitializer.java index 286adedb1d..d87a90cd83 100644 --- a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/model/SecurityScopeModelInitializer.java +++ b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/model/SecurityScopeModelInitializer.java @@ -19,6 +19,8 @@ package org.apache.dubbo.spring.security.model; import org.apache.dubbo.common.beans.factory.ScopeBeanFactory; import org.apache.dubbo.common.extension.Activate; +import org.apache.dubbo.common.logger.Logger; +import org.apache.dubbo.common.logger.LoggerFactory; import org.apache.dubbo.rpc.model.ApplicationModel; import org.apache.dubbo.rpc.model.FrameworkModel; import org.apache.dubbo.rpc.model.ModuleModel; @@ -29,22 +31,33 @@ import org.apache.dubbo.spring.security.jackson.ObjectMapperCodecCustomer; import java.util.Set; import static org.apache.dubbo.spring.security.utils.SecurityNames.CORE_JACKSON_2_MODULE_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.JAVA_TIME_MODULE_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.OBJECT_MAPPER_CLASS_NAME; import static org.apache.dubbo.spring.security.utils.SecurityNames.SECURITY_CONTEXT_HOLDER_CLASS_NAME; +import static org.apache.dubbo.spring.security.utils.SecurityNames.SIMPLE_MODULE_CLASS_NAME; -@Activate(onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME}) +@Activate(onClass = {SECURITY_CONTEXT_HOLDER_CLASS_NAME, CORE_JACKSON_2_MODULE_CLASS_NAME, OBJECT_MAPPER_CLASS_NAME, + JAVA_TIME_MODULE_CLASS_NAME, SIMPLE_MODULE_CLASS_NAME}) public class SecurityScopeModelInitializer implements ScopeModelInitializer { + private final Logger logger = LoggerFactory.getLogger(getClass()); + @Override public void initializeFrameworkModel(FrameworkModel frameworkModel) { ScopeBeanFactory beanFactory = frameworkModel.getBeanFactory(); - ObjectMapperCodec objectMapperCodec = beanFactory.getOrRegisterBean(ObjectMapperCodec.class); + try { + ObjectMapperCodec objectMapperCodec = new ObjectMapperCodec(); - Set objectMapperCodecCustomerList = frameworkModel.getExtensionLoader(ObjectMapperCodecCustomer.class).getSupportedExtensionInstances(); + Set objectMapperCodecCustomerList = frameworkModel.getExtensionLoader(ObjectMapperCodecCustomer.class).getSupportedExtensionInstances(); - for (ObjectMapperCodecCustomer objectMapperCodecCustomer : objectMapperCodecCustomerList) { - objectMapperCodecCustomer.customize(objectMapperCodec); + for (ObjectMapperCodecCustomer objectMapperCodecCustomer : objectMapperCodecCustomerList) { + objectMapperCodecCustomer.customize(objectMapperCodec); + } + + beanFactory.registerBean(objectMapperCodec); + } catch (Throwable t) { + logger.info("Failed to initialize ObjectMapperCodecCustomer and spring security related features are disabled.", t); } } diff --git a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/utils/SecurityNames.java b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/utils/SecurityNames.java index 97e167095b..f40adf4822 100644 --- a/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/utils/SecurityNames.java +++ b/dubbo-plugin/dubbo-spring-security/src/main/java/org/apache/dubbo/spring/security/utils/SecurityNames.java @@ -24,6 +24,8 @@ final public class SecurityNames { public static final String SECURITY_CONTEXT_HOLDER_CLASS_NAME = "org.springframework.security.core.context.SecurityContextHolder"; public static final String CORE_JACKSON_2_MODULE_CLASS_NAME = "org.springframework.security.jackson2.CoreJackson2Module"; public static final String OBJECT_MAPPER_CLASS_NAME = "com.fasterxml.jackson.databind.ObjectMapper"; + public static final String JAVA_TIME_MODULE_CLASS_NAME = "com.fasterxml.jackson.datatype.jsr310.JavaTimeModule"; + public static final String SIMPLE_MODULE_CLASS_NAME = "com.fasterxml.jackson.databind.module.SimpleModule"; private SecurityNames() {} From 70600eb632776a72604a61aaa2d7a008a7f2c1bb Mon Sep 17 00:00:00 2001 From: Albumen Kevin Date: Sat, 30 Sep 2023 10:06:21 +0800 Subject: [PATCH 6/6] Add metrics version compare (#13104) --- .../metrics/collector/MetricsCollector.java | 8 +++ .../data/ApplicationStatComposite.java | 14 +++- .../dubbo/metrics/data/BaseStatComposite.java | 10 +++ .../metrics/data/MethodStatComposite.java | 17 ++++- .../dubbo/metrics/data/RtStatComposite.java | 18 ++++- .../metrics/data/ServiceStatComposite.java | 29 +++++++- .../dubbo/metrics/report/MetricsExport.java | 7 ++ .../dubbo/metrics/report/MetricsReporter.java | 4 +- .../ConfigCenterMetricsCollector.java | 15 +++- .../collector/AggregateMetricsCollector.java | 66 ++++++++++++------ .../collector/DefaultMetricsCollector.java | 20 ++++++ .../collector/HistogramMetricsCollector.java | 9 ++- .../collector/sample/MetricsSampler.java | 9 +++ .../sample/ThreadPoolMetricsSampler.java | 25 ++++--- .../ThreadRejectMetricsCountSampler.java | 15 +++- .../report/AbstractMetricsReporter.java | 68 ++++++++++++------- .../report/nop/NopMetricsReporter.java | 2 +- .../collector/MetadataMetricsCollector.java | 4 ++ .../NopPrometheusMetricsReporter.java | 2 +- .../prometheus/PrometheusMetricsReporter.java | 11 +-- .../PrometheusMetricsReporterCmd.java | 2 +- .../PrometheusMetricsReporterTest.java | 7 +- .../PrometheusMetricsThreadPoolTest.java | 5 +- .../collector/RegistryMetricsCollector.java | 7 ++ .../collector/RegistryStatComposite.java | 22 +++++- .../impl/DefaultMetricsReporterCmd.java | 2 +- 26 files changed, 316 insertions(+), 82 deletions(-) diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/MetricsCollector.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/MetricsCollector.java index f8b676cd3b..19bd58bba9 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/MetricsCollector.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/MetricsCollector.java @@ -43,6 +43,14 @@ public interface MetricsCollector extends MetricsLif */ List collect(); + /** + * Check if samples have been changed. + * Note that this method will reset the changed flag to false using CAS. + * + * @return true if samples have been changed + */ + boolean calSamplesChanged(); + default void initMetrics(MetricsEvent event) {}; } diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ApplicationStatComposite.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ApplicationStatComposite.java index 47a432e480..d0bfe24bb2 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ApplicationStatComposite.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ApplicationStatComposite.java @@ -30,6 +30,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; /** @@ -46,11 +47,16 @@ public class ApplicationStatComposite extends AbstractMetricsExport { private final Map applicationNumStats = new ConcurrentHashMap<>(); + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); + public void init(List appKeys) { if (CollectionUtils.isEmpty(appKeys)) { return; } - appKeys.forEach(appKey -> applicationNumStats.put(appKey, new AtomicLong(0L))); + appKeys.forEach(appKey -> { + applicationNumStats.put(appKey, new AtomicLong(0L)); + }); + samplesChanged.set(true); } public void incrementSize(MetricsKey metricsKey, int size) { @@ -78,4 +84,10 @@ public class ApplicationStatComposite extends AbstractMetricsExport { return applicationNumStats; } + + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/BaseStatComposite.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/BaseStatComposite.java index 6dde0939bf..f3fcd2d997 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/BaseStatComposite.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/BaseStatComposite.java @@ -133,4 +133,14 @@ public abstract class BaseStatComposite implements MetricsExport { public RtStatComposite getRtStatComposite() { return rtStatComposite; } + + @Override + public boolean calSamplesChanged() { + // Should ensure that all the composite's samplesChanged have been compareAndSet, and cannot flip the `or` logic + boolean changed = applicationStatComposite.calSamplesChanged(); + changed = rtStatComposite.calSamplesChanged() || changed; + changed = serviceStatComposite.calSamplesChanged() || changed; + changed = methodStatComposite.calSamplesChanged() || changed; + return changed; + } } diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/MethodStatComposite.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/MethodStatComposite.java index 06cd8ade6e..1e26b7e25c 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/MethodStatComposite.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/MethodStatComposite.java @@ -33,6 +33,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; /** @@ -43,6 +44,8 @@ import java.util.concurrent.atomic.AtomicLong; public class MethodStatComposite extends AbstractMetricsExport { private boolean serviceLevel; + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); + public MethodStatComposite(ApplicationModel applicationModel) { super(applicationModel); this.serviceLevel = MethodMetric.isServiceLevel(getApplicationModel()); @@ -54,7 +57,10 @@ public class MethodStatComposite extends AbstractMetricsExport { if (CollectionUtils.isEmpty(metricsKeyWrappers)) { return; } - metricsKeyWrappers.forEach(appKey -> methodNumStats.put(appKey, new ConcurrentHashMap<>())); + metricsKeyWrappers.forEach(appKey -> { + methodNumStats.put(appKey, new ConcurrentHashMap<>()); + }); + samplesChanged.set(true); } public void initMethodKey(MetricsKeyWrapper wrapper, Invocation invocation) { @@ -63,6 +69,7 @@ public class MethodStatComposite extends AbstractMetricsExport { } methodNumStats.get(wrapper).computeIfAbsent(new MethodMetric(getApplicationModel(), invocation, serviceLevel), k -> new AtomicLong(0L)); + samplesChanged.set(true); } public void incrementMethodKey(MetricsKeyWrapper wrapper, MethodMetric methodMetric, int size) { @@ -71,7 +78,8 @@ public class MethodStatComposite extends AbstractMetricsExport { } AtomicLong stat = methodNumStats.get(wrapper).get(methodMetric); if (stat == null) { - methodNumStats.get(wrapper).putIfAbsent(methodMetric, new AtomicLong(0L)); + methodNumStats.get(wrapper).computeIfAbsent(methodMetric, (k)-> new AtomicLong(0L)); + samplesChanged.set(true); stat = methodNumStats.get(wrapper).get(methodMetric); } stat.getAndAdd(size); @@ -97,4 +105,9 @@ public class MethodStatComposite extends AbstractMetricsExport { return list; } + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/RtStatComposite.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/RtStatComposite.java index d335c0dfba..878349542b 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/RtStatComposite.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/RtStatComposite.java @@ -38,6 +38,7 @@ import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.LongAccumulator; import java.util.function.BiConsumer; @@ -52,6 +53,8 @@ import java.util.stream.Collectors; public class RtStatComposite extends AbstractMetricsExport { private boolean serviceLevel; + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); + public RtStatComposite(ApplicationModel applicationModel) { super(applicationModel); this.serviceLevel = MethodMetric.isServiceLevel(getApplicationModel()); @@ -66,10 +69,10 @@ public class RtStatComposite extends AbstractMetricsExport { for (MetricsPlaceValue placeValue : placeValues) { List> containers = initStats(placeValue); for (LongContainer container : containers) { - rtStats.computeIfAbsent(container.getMetricsKeyWrapper().getType(), k -> new ArrayList<>()) - .add(container); + rtStats.computeIfAbsent(container.getMetricsKeyWrapper().getType(), k -> new ArrayList<>()).add(container); } } + samplesChanged.set(true); } private List> initStats(MetricsPlaceValue placeValue) { @@ -95,6 +98,7 @@ public class RtStatComposite extends AbstractMetricsExport { Number current = (Number) container.get(key); if (current == null) { container.putIfAbsent(key, container.getInitFunc().apply(key)); + samplesChanged.set(true); current = (Number) container.get(key); } container.getConsumerFunc().accept(responseTime, current); @@ -114,6 +118,7 @@ public class RtStatComposite extends AbstractMetricsExport { if (actions == null) { actions = calServiceRtActions(invocation, registryOpType); cache.putIfAbsent(registryOpType, actions); + samplesChanged.set(true); actions = cache.get(registryOpType); } } else { @@ -134,6 +139,7 @@ public class RtStatComposite extends AbstractMetricsExport { Number current = (Number) container.get(key); if (current == null) { container.putIfAbsent(key, container.getInitFunc().apply(key)); + samplesChanged.set(true); current = (Number) container.get(key); } actions.add(new Action(container.getConsumerFunc(), current)); @@ -155,6 +161,7 @@ public class RtStatComposite extends AbstractMetricsExport { if (actions == null) { actions = calMethodRtActions(invocation, registryOpType); cache.putIfAbsent(registryOpType, actions); + samplesChanged.set(true); actions = cache.get(registryOpType); } } else { @@ -174,6 +181,7 @@ public class RtStatComposite extends AbstractMetricsExport { Number current = (Number) container.get(key); if (current == null) { container.putIfAbsent(key, container.getInitFunc().apply(key)); + samplesChanged.set(true); current = (Number) container.get(key); } actions.add(new Action(container.getConsumerFunc(), current)); @@ -218,4 +226,10 @@ public class RtStatComposite extends AbstractMetricsExport { consumerFunc.accept(responseTime, initValue); } } + + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ServiceStatComposite.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ServiceStatComposite.java index afedd71641..1cf214293e 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ServiceStatComposite.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/data/ServiceStatComposite.java @@ -30,6 +30,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; /** @@ -39,6 +40,8 @@ import java.util.concurrent.atomic.AtomicLong; */ public class ServiceStatComposite extends AbstractMetricsExport { + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); + public ServiceStatComposite(ApplicationModel applicationModel) { super(applicationModel); } @@ -49,7 +52,10 @@ public class ServiceStatComposite extends AbstractMetricsExport { if (CollectionUtils.isEmpty(metricsKeyWrappers)) { return; } - metricsKeyWrappers.forEach(appKey -> serviceWrapperNumStats.put(appKey, new ConcurrentHashMap<>())); + metricsKeyWrappers.forEach(appKey -> { + serviceWrapperNumStats.put(appKey, new ConcurrentHashMap<>()); + }); + samplesChanged.set(true); } public void incrementServiceKey(MetricsKeyWrapper wrapper, String serviceKey, int size) { @@ -64,7 +70,13 @@ public class ServiceStatComposite extends AbstractMetricsExport { if (extra != null) { serviceKeyMetric.setExtraInfo(extra); } - serviceWrapperNumStats.get(wrapper).computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L)).getAndAdd(size); + Map map = serviceWrapperNumStats.get(wrapper); + AtomicLong metrics = map.get(serviceKeyMetric); + if (metrics == null) { + metrics = map.computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L)); + samplesChanged.set(true); + } + metrics.getAndAdd(size); // MetricsSupport.fillZero(serviceWrapperNumStats); } @@ -80,7 +92,13 @@ public class ServiceStatComposite extends AbstractMetricsExport { if (extra != null) { serviceKeyMetric.setExtraInfo(extra); } - serviceWrapperNumStats.get(wrapper).computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L)).set(num); + Map stats = serviceWrapperNumStats.get(wrapper); + AtomicLong metrics = stats.get(serviceKeyMetric); + if (metrics == null) { + metrics = stats.computeIfAbsent(serviceKeyMetric, k -> new AtomicLong(0L)); + samplesChanged.set(true); + } + metrics.set(num); } @Override @@ -95,4 +113,9 @@ public class ServiceStatComposite extends AbstractMetricsExport { return list; } + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsExport.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsExport.java index d2b7c9eca5..3b1b9f3459 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsExport.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsExport.java @@ -33,4 +33,11 @@ public interface MetricsExport { */ List export(MetricsCategory category); + /** + * Check if samples have been changed. + * Note that this method will reset the changed flag to false using CAS. + * + * @return true if samples have been changed + */ + boolean calSamplesChanged(); } diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsReporter.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsReporter.java index 14bdc70217..50770e249f 100644 --- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsReporter.java +++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/report/MetricsReporter.java @@ -28,10 +28,10 @@ public interface MetricsReporter { */ void init(); - void refreshData(); + void resetIfSamplesChanged(); String getResponse(); - + default String getResponseWithName(String metricsName) { return null; } diff --git a/dubbo-metrics/dubbo-metrics-config-center/src/main/java/org/apache/dubbo/metrics/config/collector/ConfigCenterMetricsCollector.java b/dubbo-metrics/dubbo-metrics-config-center/src/main/java/org/apache/dubbo/metrics/config/collector/ConfigCenterMetricsCollector.java index d0dd1a7d7a..a86f97695e 100644 --- a/dubbo-metrics/dubbo-metrics-config-center/src/main/java/org/apache/dubbo/metrics/config/collector/ConfigCenterMetricsCollector.java +++ b/dubbo-metrics/dubbo-metrics-config-center/src/main/java/org/apache/dubbo/metrics/config/collector/ConfigCenterMetricsCollector.java @@ -34,6 +34,7 @@ import java.util.List; import java.util.Map; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import static org.apache.dubbo.metrics.model.MetricsCategory.CONFIGCENTER; @@ -47,6 +48,7 @@ public class ConfigCenterMetricsCollector extends CombMetricsCollector updatedMetrics = new ConcurrentHashMap<>(); @@ -76,7 +78,12 @@ public class ConfigCenterMetricsCollector extends CombMetricsCollector new AtomicLong(0L)).addAndGet(size); + AtomicLong metrics = updatedMetrics.get(metric); + if (metrics == null) { + metrics = updatedMetrics.computeIfAbsent(metric, k -> new AtomicLong(0L)); + samplesChanged.set(true); + } + metrics.addAndGet(size); } @@ -91,5 +98,9 @@ public class ConfigCenterMetricsCollector extends CombMetricsCollector private boolean enableRtPxx; private boolean enableRt; private boolean enableRequest; + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); private final ConcurrentMap rtAgr = new ConcurrentHashMap<>(); @@ -90,9 +92,9 @@ public class AggregateMetricsCollector implements MetricsCollector AggregationConfig aggregation = optional.get().getAggregation(); this.bucketNum = Optional.ofNullable(aggregation.getBucketNum()).orElse(DEFAULT_BUCKET_NUM); this.timeWindowSeconds = Optional.ofNullable(aggregation.getTimeWindowSeconds()) - .orElse(DEFAULT_TIME_WINDOW_SECONDS); + .orElse(DEFAULT_TIME_WINDOW_SECONDS); this.qpsTimeWindowMillSeconds = Optional.ofNullable(aggregation.getQpsTimeWindowMillSeconds()) - .orElse(DEFAULT_QPS_TIME_WINDOW_MILL_SECONDS); + .orElse(DEFAULT_QPS_TIME_WINDOW_MILL_SECONDS); this.enableQps = Optional.ofNullable(aggregation.getEnableQps()).orElse(true); this.enableRtPxx = Optional.ofNullable(aggregation.getEnableRtPxx()).orElse(true); this.enableRt = Optional.ofNullable(aggregation.getEnableRt()).orElse(true); @@ -127,8 +129,12 @@ public class AggregateMetricsCollector implements MetricsCollector public void onEvent(RequestEvent event) { if (enableQps) { MethodMetric metric = calcWindowCounter(event, MetricsKey.METRIC_REQUESTS); - TimeWindowCounter qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric, - methodMetric -> new TimeWindowCounter(bucketNum, TimeUnit.MILLISECONDS.toSeconds(qpsTimeWindowMillSeconds))); + TimeWindowCounter qpsCounter = qps.get(metric); + if (qpsCounter == null) { + qpsCounter = ConcurrentHashMapUtils.computeIfAbsent(qps, metric, + methodMetric -> new TimeWindowCounter(bucketNum, TimeUnit.MILLISECONDS.toSeconds(qpsTimeWindowMillSeconds))); + samplesChanged.set(true); + } qpsCounter.increment(); } } @@ -163,14 +169,22 @@ public class AggregateMetricsCollector implements MetricsCollector MethodMetric metric = new MethodMetric(applicationModel, event.getAttachmentValue(MetricsConstants.INVOCATION), serviceLevel); long responseTime = event.getTimePair().calc(); if (enableRt) { - TimeWindowQuantile quantile = ConcurrentHashMapUtils.computeIfAbsent(rt, metric, - k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds)); + TimeWindowQuantile quantile = rt.get(metric); + if (quantile == null) { + quantile = ConcurrentHashMapUtils.computeIfAbsent(rt, metric, + k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds)); + samplesChanged.set(true); + } quantile.add(responseTime); } if (enableRtPxx) { - TimeWindowAggregator timeWindowAggregator = ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric, - methodMetric -> new TimeWindowAggregator(bucketNum, timeWindowSeconds)); + TimeWindowAggregator timeWindowAggregator = rtAgr.get(metric); + if (timeWindowAggregator == null) { + timeWindowAggregator = ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric, + methodMetric -> new TimeWindowAggregator(bucketNum, timeWindowSeconds)); + samplesChanged.set(true); + } timeWindowAggregator.add(responseTime); } } @@ -183,8 +197,12 @@ public class AggregateMetricsCollector implements MetricsCollector ConcurrentMap counter = methodTypeCounter.computeIfAbsent(metricsKeyWrapper, k -> new ConcurrentHashMap<>()); - TimeWindowCounter windowCounter = ConcurrentHashMapUtils.computeIfAbsent(counter, metric, - methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds)); + TimeWindowCounter windowCounter = counter.get(metric); + if (windowCounter == null) { + windowCounter = ConcurrentHashMapUtils.computeIfAbsent(counter, metric, + methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds)); + samplesChanged.set(true); + } windowCounter.increment(); return metric; } @@ -225,13 +243,13 @@ public class AggregateMetricsCollector implements MetricsCollector ConcurrentHashMap windowCounter = methodTypeCounter.get(metricsKeyWrapper); if (windowCounter != null) { windowCounter.forEach((k, v) -> list.add(new GaugeMetricSample<>(metricsKey.getNameByType(k.getSide()), - metricsKey.getDescription(), k.getTags(), REQUESTS, v, TimeWindowCounter::get))); + metricsKey.getDescription(), k.getTags(), REQUESTS, v, TimeWindowCounter::get))); } } private void collectQPS(List list) { qps.forEach((k, v) -> list.add(new GaugeMetricSample<>(MetricsKey.METRIC_QPS.getNameByType(k.getSide()), - MetricsKey.METRIC_QPS.getDescription(), k.getTags(), QPS, v, value -> { + MetricsKey.METRIC_QPS.getDescription(), k.getTags(), QPS, v, value -> { double total = value.get(); long millSeconds = value.bucketLivedMillSeconds(); return total / millSeconds * 1000; @@ -241,24 +259,24 @@ public class AggregateMetricsCollector implements MetricsCollector private void collectRT(List list) { rt.forEach((k, v) -> { list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P99.getNameByType(k.getSide()), - MetricsKey.METRIC_RT_P99.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.99))); + MetricsKey.METRIC_RT_P99.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.99))); list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P95.getNameByType(k.getSide()), - MetricsKey.METRIC_RT_P95.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.95))); + MetricsKey.METRIC_RT_P95.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.95))); list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P90.getNameByType(k.getSide()), - MetricsKey.METRIC_RT_P90.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.90))); + MetricsKey.METRIC_RT_P90.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.90))); list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_P50.getNameByType(k.getSide()), - MetricsKey.METRIC_RT_P50.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.50))); + MetricsKey.METRIC_RT_P50.getDescription(), k.getTags(), RT, v, value -> value.quantile(0.50))); }); rtAgr.forEach((k, v) -> { list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_MIN_AGG.getNameByType(k.getSide()), - MetricsKey.METRIC_RT_MIN_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMin())); + MetricsKey.METRIC_RT_MIN_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMin())); list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_MAX_AGG.getNameByType(k.getSide()), - MetricsKey.METRIC_RT_MAX_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMax())); + MetricsKey.METRIC_RT_MAX_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getMax())); list.add(new GaugeMetricSample<>(MetricsKey.METRIC_RT_AVG_AGG.getNameByType(k.getSide()), - MetricsKey.METRIC_RT_AVG_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getAvg())); + MetricsKey.METRIC_RT_AVG_AGG.getDescription(), k.getTags(), RT, v, value -> v.get().getAvg())); }); } @@ -288,14 +306,17 @@ public class AggregateMetricsCollector implements MetricsCollector public void initQpsMetric(MethodMetric metric) { ConcurrentHashMapUtils.computeIfAbsent(qps, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds)); + samplesChanged.set(true); } public void initRtMetric(MethodMetric metric) { ConcurrentHashMapUtils.computeIfAbsent(rt, metric, k -> new TimeWindowQuantile(DEFAULT_COMPRESSION, bucketNum, timeWindowSeconds)); + samplesChanged.set(true); } public void initRtAgrMetric(MethodMetric metric) { ConcurrentHashMapUtils.computeIfAbsent(rtAgr, metric, k -> new TimeWindowAggregator(bucketNum, timeWindowSeconds)); + samplesChanged.set(true); } public void initWindowCounter(MetricsEvent event, MetricsKey targetKey) { @@ -307,6 +328,13 @@ public class AggregateMetricsCollector implements MetricsCollector ConcurrentMap counter = methodTypeCounter.computeIfAbsent(metricsKeyWrapper, k -> new ConcurrentHashMap<>()); ConcurrentHashMapUtils.computeIfAbsent(counter, metric, methodMetric -> new TimeWindowCounter(bucketNum, timeWindowSeconds)); + samplesChanged.set(true); } + + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/DefaultMetricsCollector.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/DefaultMetricsCollector.java index 9d04633260..f1fd60d91a 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/DefaultMetricsCollector.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/DefaultMetricsCollector.java @@ -72,6 +72,7 @@ public class DefaultMetricsCollector extends CombMetricsCollector private final AtomicBoolean initialized = new AtomicBoolean(); + private final AtomicBoolean samplesChanged = new AtomicBoolean(); public DefaultMetricsCollector(ApplicationModel applicationModel) { super(new BaseStatComposite(applicationModel) { @@ -91,11 +92,13 @@ public class DefaultMetricsCollector extends CombMetricsCollector super.setEventMulticaster(new DefaultSubDispatcher(this)); samplers.add(applicationSampler); samplers.add(threadPoolSampler); + samplesChanged.set(true); this.applicationModel = applicationModel; } public void addSampler(MetricsSampler sampler) { samplers.add(sampler); + samplesChanged.set(true); } public void setApplicationName(String applicationName) { @@ -203,5 +206,22 @@ public class DefaultMetricsCollector extends CombMetricsCollector MetricsCountSampleConfigurer sampleConfigure) { sampleConfigure.configureMetrics(configure -> new ApplicationMetric(applicationModel)); } + + @Override + public boolean calSamplesChanged() { + return false; + } }; + + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + boolean changed = samplesChanged.compareAndSet(true, false); + // Should ensure that all the sampler's samplesChanged have been compareAndSet, and cannot flip the `or` logic + changed = stats.calSamplesChanged() || changed; + for (MetricsSampler sampler : samplers) { + changed = sampler.calSamplesChanged() || changed; + } + return changed; + } } diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java index 932c285d8e..a7169ba12a 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/HistogramMetricsCollector.java @@ -17,7 +17,6 @@ package org.apache.dubbo.metrics.collector; -import io.micrometer.core.instrument.Timer; import org.apache.dubbo.common.utils.ConcurrentHashMapUtils; import org.apache.dubbo.config.MetricsConfig; import org.apache.dubbo.config.context.ConfigManager; @@ -33,6 +32,8 @@ import org.apache.dubbo.metrics.register.HistogramMetricRegister; import org.apache.dubbo.metrics.sample.HistogramMetricSample; import org.apache.dubbo.rpc.model.ApplicationModel; +import io.micrometer.core.instrument.Timer; + import java.util.ArrayList; import java.util.List; import java.util.concurrent.ConcurrentHashMap; @@ -110,4 +111,10 @@ public class HistogramMetricsCollector extends AbstractMetricsListener collect() { return new ArrayList<>(); } + + @Override + public boolean calSamplesChanged() { + // Histogram is directly register micrometer + return false; + } } diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MetricsSampler.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MetricsSampler.java index 55111af991..52578d627d 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MetricsSampler.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/MetricsSampler.java @@ -18,9 +18,18 @@ package org.apache.dubbo.metrics.collector.sample; import org.apache.dubbo.metrics.model.sample.MetricSample; + import java.util.List; public interface MetricsSampler { List sample(); + + /** + * Check if samples have been changed. + * Note that this method will reset the changed flag to false using CAS. + * + * @return true if samples have been changed + */ + boolean calSamplesChanged(); } diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSampler.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSampler.java index ec4a74b98d..9ccb55f8db 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSampler.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadPoolMetricsSampler.java @@ -23,13 +23,12 @@ import org.apache.dubbo.common.threadpool.manager.FrameworkExecutorRepository; import org.apache.dubbo.common.threadpool.support.AbortPolicyWithReport; import org.apache.dubbo.common.utils.ConcurrentHashMapUtils; import org.apache.dubbo.metrics.collector.DefaultMetricsCollector; -import org.apache.dubbo.metrics.model.key.MetricsKey; import org.apache.dubbo.metrics.model.ThreadPoolMetric; +import org.apache.dubbo.metrics.model.key.MetricsKey; import org.apache.dubbo.metrics.model.sample.GaugeMetricSample; import org.apache.dubbo.metrics.model.sample.MetricSample; import org.apache.dubbo.rpc.model.ApplicationModel; - import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -38,10 +37,10 @@ import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.dubbo.common.constants.CommonConstants.CONSUMER_SHARED_EXECUTOR_SERVICE_COMPONENT_KEY; import static org.apache.dubbo.common.constants.CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY; - import static org.apache.dubbo.common.constants.LoggerCodeConstants.COMMON_METRICS_COLLECTOR_EXCEPTION; import static org.apache.dubbo.config.Constants.CLIENT_THREAD_POOL_NAME; import static org.apache.dubbo.config.Constants.SERVER_THREAD_POOL_NAME; @@ -56,6 +55,8 @@ public class ThreadPoolMetricsSampler implements MetricsSampler { private DataStore dataStore; private final Map sampleThreadPoolExecutor = new ConcurrentHashMap<>(); private final ConcurrentHashMap threadPoolMetricMap = new ConcurrentHashMap<>(); + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); + public ThreadPoolMetricsSampler(DefaultMetricsCollector collector) { this.collector = collector; } @@ -63,7 +64,10 @@ public class ThreadPoolMetricsSampler implements MetricsSampler { public void addExecutors(String name, ExecutorService executorService) { Optional.ofNullable(executorService).filter(Objects::nonNull).filter(e -> e instanceof ThreadPoolExecutor) .map(e -> (ThreadPoolExecutor) e) - .ifPresent(threadPoolExecutor -> sampleThreadPoolExecutor.put(name, threadPoolExecutor)); + .ifPresent(threadPoolExecutor -> { + sampleThreadPoolExecutor.put(name, threadPoolExecutor); + samplesChanged.set(true); + }); } @Override @@ -113,7 +117,7 @@ public class ThreadPoolMetricsSampler implements MetricsSampler { for (Map.Entry entry : executors.entrySet()) { ExecutorService executor = (ExecutorService) entry.getValue(); if (executor instanceof ThreadPoolExecutor) { - this.addExecutors( SERVER_THREAD_POOL_NAME + "-" + entry.getKey(), executor); + this.addExecutors(SERVER_THREAD_POOL_NAME + "-" + entry.getKey(), executor); } } executors = dataStore.get(CONSUMER_SHARED_EXECUTOR_SERVICE_COMPONENT_KEY); @@ -125,9 +129,9 @@ public class ThreadPoolMetricsSampler implements MetricsSampler { } ThreadRejectMetricsCountSampler threadRejectMetricsCountSampler = new ThreadRejectMetricsCountSampler(collector); - this.sampleThreadPoolExecutor.entrySet().stream().filter(entry->entry.getKey().startsWith(SERVER_THREAD_POOL_NAME)).forEach(entry->{ - if(entry.getValue().getRejectedExecutionHandler() instanceof AbortPolicyWithReport) { - MetricThreadPoolExhaustedListener metricThreadPoolExhaustedListener=new MetricThreadPoolExhaustedListener(entry.getKey(),threadRejectMetricsCountSampler); + this.sampleThreadPoolExecutor.entrySet().stream().filter(entry -> entry.getKey().startsWith(SERVER_THREAD_POOL_NAME)).forEach(entry -> { + if (entry.getValue().getRejectedExecutionHandler() instanceof AbortPolicyWithReport) { + MetricThreadPoolExhaustedListener metricThreadPoolExhaustedListener = new MetricThreadPoolExhaustedListener(entry.getKey(), threadRejectMetricsCountSampler); ((AbortPolicyWithReport) entry.getValue().getRejectedExecutionHandler()).addThreadPoolExhaustedEventListener(metricThreadPoolExhaustedListener); } }); @@ -137,4 +141,9 @@ public class ThreadPoolMetricsSampler implements MetricsSampler { } } + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadRejectMetricsCountSampler.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadRejectMetricsCountSampler.java index 160f3c31fc..2c0cbcd518 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadRejectMetricsCountSampler.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/collector/sample/ThreadRejectMetricsCountSampler.java @@ -16,19 +16,23 @@ */ package org.apache.dubbo.metrics.collector.sample; + import org.apache.dubbo.common.utils.ConcurrentHashSet; import org.apache.dubbo.metrics.collector.DefaultMetricsCollector; import org.apache.dubbo.metrics.model.Metric; import org.apache.dubbo.metrics.model.MetricsCategory; -import org.apache.dubbo.metrics.model.key.MetricsKey; import org.apache.dubbo.metrics.model.ThreadPoolRejectMetric; +import org.apache.dubbo.metrics.model.key.MetricsKey; import org.apache.dubbo.metrics.model.sample.GaugeMetricSample; import org.apache.dubbo.metrics.model.sample.MetricSample; + import java.util.ArrayList; import java.util.List; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.function.ToDoubleFunction; + import static org.apache.dubbo.metrics.model.MetricsCategory.THREAD_POOL; public class ThreadRejectMetricsCountSampler extends SimpleMetricsCountSampler { @@ -36,6 +40,8 @@ public class ThreadRejectMetricsCountSampler extends SimpleMetricsCountSampler metricNames = new ConcurrentHashSet<>(); + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); + public ThreadRejectMetricsCountSampler(DefaultMetricsCollector collector) { this.collector = collector; this.collector.addSampler(this); @@ -44,6 +50,7 @@ public class ThreadRejectMetricsCountSampler extends SimpleMetricsCountSampler sampleConfigure) { sampleConfigure.configureMetrics(configure -> new ThreadPoolRejectMetric(collector.getApplicationName(),configure.getSource())); } + + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java index 70033771c5..ac0700221f 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/AbstractMetricsReporter.java @@ -148,38 +148,22 @@ public abstract class AbstractMetricsReporter implements MetricsReporter { NamedThreadFactory threadFactory = new NamedThreadFactory("metrics-collector-sync-job", true); collectorSyncJobExecutor = Executors.newScheduledThreadPool(1, threadFactory); - collectorSyncJobExecutor.scheduleWithFixedDelay(this::refreshData, DEFAULT_SCHEDULE_INITIAL_DELAY, collectSyncPeriod, TimeUnit.SECONDS); + collectorSyncJobExecutor.scheduleWithFixedDelay(this::resetIfSamplesChanged, DEFAULT_SCHEDULE_INITIAL_DELAY, collectSyncPeriod, TimeUnit.SECONDS); } } - @SuppressWarnings({"unchecked", "rawtypes"}) - public void refreshData() { + @SuppressWarnings({"unchecked"}) + public void resetIfSamplesChanged() { collectors.forEach(collector -> { + if (!collector.calSamplesChanged()) { + // Metrics has not been changed since last time, no need to reload + return; + } + // Collect all the samples and register them to the micrometer registry List samples = collector.collect(); for (MetricSample sample : samples) { try { - switch (sample.getType()) { - case GAUGE: - GaugeMetricSample gaugeSample = (GaugeMetricSample) sample; - List tags = getTags(gaugeSample); - - Gauge.builder(gaugeSample.getName(), gaugeSample.getValue(), gaugeSample.getApply()) - .description(gaugeSample.getDescription()).tags(tags).register(compositeRegistry); - break; - case COUNTER: - CounterMetricSample counterMetricSample = (CounterMetricSample) sample; - FunctionCounter.builder(counterMetricSample.getName(), counterMetricSample.getValue(), - Number::doubleValue).description(counterMetricSample.getDescription()) - .tags(getTags(counterMetricSample)) - .register(compositeRegistry); - case TIMER: - case LONG_TASK_TIMER: - case DISTRIBUTION_SUMMARY: - // TODO - break; - default: - break; - } + registerSample(sample); } catch (Exception e) { logger.error(COMMON_METRICS_COLLECTOR_EXCEPTION, "", "", "error occurred when synchronize metrics collector.", e); } @@ -187,6 +171,40 @@ public abstract class AbstractMetricsReporter implements MetricsReporter { }); } + @SuppressWarnings({"rawtypes"}) + private void registerSample(MetricSample sample) { + switch (sample.getType()) { + case GAUGE: + registerGaugeSample((GaugeMetricSample) sample); + break; + case COUNTER: + registerCounterSample((CounterMetricSample) sample); + case TIMER: + case LONG_TASK_TIMER: + case DISTRIBUTION_SUMMARY: + // TODO + break; + default: + break; + } + } + + @SuppressWarnings({"rawtypes"}) + private void registerCounterSample(CounterMetricSample sample) { + FunctionCounter.builder(sample.getName(), sample.getValue(), Number::doubleValue) + .description(sample.getDescription()) + .tags(getTags(sample)) + .register(compositeRegistry); + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private void registerGaugeSample(GaugeMetricSample sample) { + Gauge.builder(sample.getName(), sample.getValue(), sample.getApply()) + .description(sample.getDescription()) + .tags(getTags(sample)) + .register(compositeRegistry); + } + private static List getTags(MetricSample gaugeSample) { List tags = new ArrayList<>(); gaugeSample.getTags().forEach((k, v) -> { diff --git a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/nop/NopMetricsReporter.java b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/nop/NopMetricsReporter.java index 0d3a078378..f326f5467c 100644 --- a/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/nop/NopMetricsReporter.java +++ b/dubbo-metrics/dubbo-metrics-default/src/main/java/org/apache/dubbo/metrics/report/nop/NopMetricsReporter.java @@ -35,7 +35,7 @@ public class NopMetricsReporter implements MetricsReporter { } @Override - public void refreshData() { + public void resetIfSamplesChanged() { } diff --git a/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/collector/MetadataMetricsCollector.java b/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/collector/MetadataMetricsCollector.java index f3127f1ab4..82b5e689d3 100644 --- a/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/collector/MetadataMetricsCollector.java +++ b/dubbo-metrics/dubbo-metrics-metadata/src/main/java/org/apache/dubbo/metrics/metadata/collector/MetadataMetricsCollector.java @@ -99,4 +99,8 @@ public class MetadataMetricsCollector extends CombMetricsCollector { - reporter.refreshData(); + reporter.resetIfSamplesChanged(); String response = reporter.getPrometheusRegistry().scrape(); httpExchange.sendResponseHeaders(200, response.getBytes().length); try (OutputStream os = httpExchange.getResponseBody()) { diff --git a/dubbo-metrics/dubbo-metrics-prometheus/src/test/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsThreadPoolTest.java b/dubbo-metrics/dubbo-metrics-prometheus/src/test/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsThreadPoolTest.java index 107c09ea97..4b30009cf1 100644 --- a/dubbo-metrics/dubbo-metrics-prometheus/src/test/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsThreadPoolTest.java +++ b/dubbo-metrics/dubbo-metrics-prometheus/src/test/java/org/apache/dubbo/metrics/prometheus/PrometheusMetricsThreadPoolTest.java @@ -16,7 +16,6 @@ */ package org.apache.dubbo.metrics.prometheus; -import com.sun.net.httpserver.HttpServer; import org.apache.dubbo.config.ApplicationConfig; import org.apache.dubbo.config.MetricsConfig; import org.apache.dubbo.config.nested.PrometheusConfig; @@ -25,6 +24,8 @@ import org.apache.dubbo.metrics.collector.sample.ThreadRejectMetricsCountSampler import org.apache.dubbo.metrics.model.sample.GaugeMetricSample; import org.apache.dubbo.metrics.model.sample.MetricSample; import org.apache.dubbo.rpc.model.ApplicationModel; + +import com.sun.net.httpserver.HttpServer; import org.apache.http.client.methods.CloseableHttpResponse; import org.apache.http.client.methods.HttpGet; import org.apache.http.impl.client.CloseableHttpClient; @@ -122,7 +123,7 @@ public class PrometheusMetricsThreadPoolTest { try { HttpServer prometheusExporterHttpServer = HttpServer.create(new InetSocketAddress(port), 0); prometheusExporterHttpServer.createContext("/metrics", httpExchange -> { - reporter.refreshData(); + reporter.resetIfSamplesChanged(); String response = reporter.getPrometheusRegistry().scrape(); httpExchange.sendResponseHeaders(200, response.getBytes().length); try (OutputStream os = httpExchange.getResponseBody()) { diff --git a/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/collector/RegistryMetricsCollector.java b/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/collector/RegistryMetricsCollector.java index b510ed541b..6559a49ed2 100644 --- a/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/collector/RegistryMetricsCollector.java +++ b/dubbo-metrics/dubbo-metrics-registry/src/main/java/org/apache/dubbo/metrics/registry/collector/RegistryMetricsCollector.java @@ -149,4 +149,11 @@ public class RegistryMetricsCollector extends CombMetricsCollector> appStats = new ConcurrentHashMap<>(); + private final AtomicBoolean samplesChanged = new AtomicBoolean(true); + public RegistryStatComposite(ApplicationModel applicationModel) { super(applicationModel); init(RegistryMetricsConstants.REGISTER_LEVEL_KEYS); @@ -50,7 +53,10 @@ public class RegistryStatComposite extends AbstractMetricsExport { if (CollectionUtils.isEmpty(appKeys)) { return; } - appKeys.forEach(appKey -> appStats.put(appKey, new ConcurrentHashMap<>())); + appKeys.forEach(appKey -> { + appStats.put(appKey, new ConcurrentHashMap<>()); + }); + samplesChanged.set(true); } @Override @@ -71,11 +77,23 @@ public class RegistryStatComposite extends AbstractMetricsExport { } ApplicationMetric applicationMetric = new ApplicationMetric(getApplicationModel()); applicationMetric.setExtraInfo(Collections.singletonMap(RegistryConstants.REGISTRY_CLUSTER_KEY.toLowerCase(), name)); - appStats.get(metricsKey).computeIfAbsent(applicationMetric, k -> new AtomicLong(0L)).getAndAdd(SELF_INCREMENT_SIZE); + Map stats = appStats.get(metricsKey); + AtomicLong metrics = stats.get(applicationMetric); + if (metrics == null) { + metrics = stats.computeIfAbsent(applicationMetric, k -> new AtomicLong(0L)); + samplesChanged.set(true); + } + metrics.getAndAdd(SELF_INCREMENT_SIZE); MetricsSupport.fillZero(appStats); } public Map> getAppStats() { return appStats; } + + @Override + public boolean calSamplesChanged() { + // CAS to get and reset the flag in an atomic operation + return samplesChanged.compareAndSet(true, false); + } } diff --git a/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/DefaultMetricsReporterCmd.java b/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/DefaultMetricsReporterCmd.java index de8ec4b904..ceb4d5756b 100644 --- a/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/DefaultMetricsReporterCmd.java +++ b/dubbo-plugin/dubbo-qos/src/main/java/org/apache/dubbo/qos/command/impl/DefaultMetricsReporterCmd.java @@ -102,7 +102,7 @@ public class DefaultMetricsReporterCmd implements BaseCommand { String response = "DefaultMetricsReporter not init"; MetricsReporter metricsReporter = applicationModel.getBeanFactory().getBean(DefaultMetricsReporter.class); if (metricsReporter != null) { - metricsReporter.refreshData(); + metricsReporter.resetIfSamplesChanged(); response = metricsReporter.getResponseWithName(metricsName); } return response;