抽取配置服务独立使用

sermant-io · Sep 12, 2024 · 4a4ba0c · 4a4ba0c
1 parent 4028c0b
commit 4a4ba0c
Show file tree

Hide file tree

Showing 15 changed files with 238 additions and 92 deletions.
diff --git a/sermant-plugins/sermant-mq-grayscale/mq-config-common/pom.xml b/sermant-plugins/sermant-mq-grayscale/mq-config-common/pom.xml
@@ -14,7 +14,6 @@
     <properties>
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
-        <package.plugin.type>service</package.plugin.type>
     </properties>
 
     <dependencies>

diff --git a/sermant-plugins/sermant-mq-grayscale/mq-config-service/pom.xml b/sermant-plugins/sermant-mq-grayscale/mq-config-service/pom.xml
@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <artifactId>sermant-mq-grayscale</artifactId>
+        <groupId>io.sermant</groupId>
+        <version>1.0.0</version>
+    </parent>
+
+    <artifactId>mq-config-service</artifactId>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <package.plugin.type>service</package.plugin.type>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.yaml</groupId>
+            <artifactId>snakeyaml</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.sermant</groupId>
+            <artifactId>sermant-agentcore-core</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.sermant</groupId>
+            <artifactId>mq-config-common</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>
diff --git a/...e/mq-config-service/src/main/java/io/sermant/mq/grayscale/listener/MqGrayConfigCache.java b/...e/mq-config-service/src/main/java/io/sermant/mq/grayscale/listener/MqGrayConfigCache.java
@@ -0,0 +1,50 @@
+/*
+ * Copyright (C) 2024-2024 Sermant Authors. All rights reserved.
+ *
+ *   Licensed 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 io.sermant.mq.grayscale.listener;
+
+import io.sermant.mq.grayscale.config.MqGrayscaleConfig;
+
+/**
+ * grayscale config cache
+ *
+ * @author chengyouling
+ * @since 2024-09-12
+ **/
+public class MqGrayConfigCache {
+    private static MqGrayscaleConfig cacheConfig;
+
+    /**
+     * get cache mqGrayscaleConfig
+     */
+    public static MqGrayscaleConfig getCacheConfig() {
+        return cacheConfig;
+    }
+
+    /**
+     * set cache mqGrayscaleConfig
+     */
+    public static void setCacheConfig(MqGrayscaleConfig config) {
+        cacheConfig = config;
+    }
+
+    /**
+     * clear cache mqGrayscaleConfig
+     */
+    public static void clearCacheConfig() {
+        cacheConfig = new MqGrayscaleConfig();
+    }
+}
diff --git a/...tmq/config/RocketMqGrayConfigHandler.java → ...ayscale/listener/MqGrayConfigHandler.java b/...tmq/config/RocketMqGrayConfigHandler.java → ...ayscale/listener/MqGrayConfigHandler.java
@@ -14,14 +14,13 @@
  *   limitations under the License.
  */
 
-package io.sermant.mq.grayscale.rocketmq.config;
+package io.sermant.mq.grayscale.listener;
 
 import io.sermant.core.common.LoggerFactory;
 import io.sermant.core.service.dynamicconfig.common.DynamicConfigEvent;
 import io.sermant.core.service.dynamicconfig.common.DynamicConfigEventType;
 import io.sermant.core.utils.StringUtils;
 import io.sermant.mq.grayscale.config.MqGrayscaleConfig;
-import io.sermant.mq.grayscale.rocketmq.utils.RocketMqGrayscaleConfigUtils;
 
 import org.yaml.snakeyaml.DumperOptions;
 import org.yaml.snakeyaml.Yaml;
@@ -36,7 +35,7 @@
  * @author chengyouling
  * @since 2024-05-27
  **/
-public class RocketMqGrayConfigHandler {
+public class MqGrayConfigHandler {
     private static final Logger LOGGER = LoggerFactory.getLogger();
 
     /**
@@ -49,7 +48,7 @@ public class RocketMqGrayConfigHandler {
     /**
      * construction
      */
-    public RocketMqGrayConfigHandler() {
+    public MqGrayConfigHandler() {
         Representer representer = new Representer(new DumperOptions());
         representer.getPropertyUtils().setSkipMissingProperties(true);
         this.yaml = new Yaml(representer);
@@ -65,15 +64,15 @@ public void handle(DynamicConfigEvent event) {
             return;
         }
         if (event.getEventType() == DynamicConfigEventType.DELETE) {
-            RocketMqGrayscaleConfigUtils.resetGrayscaleConfig();
+            MqGrayConfigCache.clearCacheConfig();
             return;
         }
         if (!StringUtils.isEmpty(event.getContent())) {
             LOGGER.info(String.format(Locale.ROOT, "mqGrayscale [%s] dynamicConfig context: %s",
                     event.getGroup(), event.getContent()));
             MqGrayscaleConfig config = yaml.loadAs(event.getContent(), MqGrayscaleConfig.class);
             if (config != null) {
-                RocketMqGrayscaleConfigUtils.setGrayscaleConfig(config, event.getEventType());
+                MqGrayConfigCache.setCacheConfig(config);
             }
         }
     }

diff --git a/...mq/config/RocketMqGrayConfigListener.java → ...yscale/listener/MqGrayConfigListener.java b/...mq/config/RocketMqGrayConfigListener.java → ...yscale/listener/MqGrayConfigListener.java
@@ -14,7 +14,7 @@
  *   limitations under the License.
  */
 
-package io.sermant.mq.grayscale.rocketmq.config;
+package io.sermant.mq.grayscale.listener;
 
 import io.sermant.core.service.dynamicconfig.common.DynamicConfigEvent;
 import io.sermant.core.service.dynamicconfig.common.DynamicConfigListener;
@@ -25,14 +25,14 @@
  * @author chengyouling
  * @since 2024-05-27
  **/
-public class RocketMqGrayConfigListener implements DynamicConfigListener {
-    private final RocketMqGrayConfigHandler handler;
+public class MqGrayConfigListener implements DynamicConfigListener {
+    private final MqGrayConfigHandler handler;
 
     /**
      * construction
      */
-    public RocketMqGrayConfigListener() {
-        handler = new RocketMqGrayConfigHandler();
+    public MqGrayConfigListener() {
+        handler = new MqGrayConfigHandler();
     }
 
     @Override

diff --git a/...ice/RocketMqGrayDynamicConfigService.java → ...e/service/MqGrayDynamicConfigService.java b/...ice/RocketMqGrayDynamicConfigService.java → ...e/service/MqGrayDynamicConfigService.java
@@ -14,15 +14,15 @@
  *   limitations under the License.
  */
 
-package io.sermant.mq.grayscale.rocketmq.service;
+package io.sermant.mq.grayscale.service;
 
 import io.sermant.core.common.LoggerFactory;
 import io.sermant.core.config.ConfigManager;
 import io.sermant.core.plugin.config.ServiceMeta;
 import io.sermant.core.plugin.service.PluginService;
 import io.sermant.core.plugin.subscribe.CommonGroupConfigSubscriber;
 import io.sermant.core.plugin.subscribe.ConfigSubscriber;
-import io.sermant.mq.grayscale.rocketmq.config.RocketMqGrayConfigListener;
+import io.sermant.mq.grayscale.listener.MqGrayConfigListener;
 
 import java.util.logging.Logger;
 
@@ -32,13 +32,13 @@
  * @author chengyouling
  * @since 2024-05-27
  **/
-public class RocketMqGrayDynamicConfigService implements PluginService {
+public class MqGrayDynamicConfigService implements PluginService {
     private static final Logger LOGGER = LoggerFactory.getLogger();
 
     @Override
     public void start() {
         ConfigSubscriber subscriber = new CommonGroupConfigSubscriber(
-                ConfigManager.getConfig(ServiceMeta.class).getService(), new RocketMqGrayConfigListener(),
+                ConfigManager.getConfig(ServiceMeta.class).getService(), new MqGrayConfigListener(),
                 "mq-grayscale");
         subscriber.subscribe();
         LOGGER.info("Success to subscribe mq-grayscale config");

diff --git a/...sermant.core.plugin.service.PluginService → ...sermant.core.plugin.service.PluginService b/...sermant.core.plugin.service.PluginService → ...sermant.core.plugin.service.PluginService
@@ -14,4 +14,4 @@
 # limitations under the License.
 #
 
-io.sermant.mq.grayscale.rocketmq.service.RocketMqGrayDynamicConfigService
+io.sermant.mq.grayscale.service.MqGrayDynamicConfigService
diff --git a/sermant-plugins/sermant-mq-grayscale/mq-grayscale-rocketmq-plugin/pom.xml b/sermant-plugins/sermant-mq-grayscale/mq-grayscale-rocketmq-plugin/pom.xml
@@ -20,12 +20,13 @@
 
     <dependencies>
         <dependency>
-            <groupId>org.yaml</groupId>
-            <artifactId>snakeyaml</artifactId>
+            <groupId>io.sermant</groupId>
+            <artifactId>mq-config-common</artifactId>
+            <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>io.sermant</groupId>
-            <artifactId>mq-config-common</artifactId>
+            <artifactId>mq-config-service</artifactId>
             <version>${project.version}</version>
         </dependency>
         <dependency>

diff --git a/...c/main/java/io/sermant/mq/grayscale/rocketmq/interceptor/RocketMqAbstractIntercepter.java b/...c/main/java/io/sermant/mq/grayscale/rocketmq/interceptor/RocketMqAbstractIntercepter.java
@@ -37,6 +37,7 @@ public ExecuteContext before(ExecuteContext context) throws Exception {
 
     @Override
     public ExecuteContext after(ExecuteContext context) throws Exception {
+        RocketMqGrayscaleConfigUtils.checkAndUpdateCacheGrayConfig();
         if (!RocketMqGrayscaleConfigUtils.isPluginEnabled()) {
             return context;
         }

diff --git a/.../io/sermant/mq/grayscale/rocketmq/interceptor/RocketMqConsumerConstructorInterceptor.java b/.../io/sermant/mq/grayscale/rocketmq/interceptor/RocketMqConsumerConstructorInterceptor.java
@@ -19,7 +19,11 @@
 import io.sermant.core.plugin.agent.entity.ExecuteContext;
 import io.sermant.core.plugin.agent.interceptor.AbstractInterceptor;
 import io.sermant.core.utils.StringUtils;
+import io.sermant.mq.grayscale.listener.MqGrayConfigCache;
 import io.sermant.mq.grayscale.rocketmq.utils.RocketMqGrayscaleConfigUtils;
+import io.sermant.mq.grayscale.rocketmq.utils.RocketMqSubscriptionDataUtils;
+
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * DefaultMQPushConsumer/DefaultLitePullConsumer/DefaultMQPullConsumer Constructor method interceptor
@@ -29,8 +33,13 @@
  * @since 2024-05-27
  **/
 public class RocketMqConsumerConstructorInterceptor extends AbstractInterceptor {
+    private final AtomicBoolean cacheConfigInit = new AtomicBoolean(false);
+
     @Override
     public ExecuteContext before(ExecuteContext context) throws Exception {
+        if (cacheConfigInit.compareAndSet(false, true)) {
+            RocketMqGrayscaleConfigUtils.createMqGrayConfigs();
+        }
         if (!RocketMqGrayscaleConfigUtils.isPluginEnabled()) {
             return context;
         }

diff --git a/.../mq/grayscale/rocketmq/interceptor/RocketMqPullConsumerSubscriptionUpdateInterceptor.java b/.../mq/grayscale/rocketmq/interceptor/RocketMqPullConsumerSubscriptionUpdateInterceptor.java
@@ -67,6 +67,7 @@ private void buildSql92SubscriptionData(ExecuteContext context, SubscriptionData
         if (StringUtils.isEmpty(RocketMqGrayscaleConfigUtils.getGrayGroupTag())) {
             RocketMqConsumerGroupAutoCheck.setMqClientInstance(subscriptionData.getTopic(), consumerGroup, instance);
         }
+        RocketMqConsumerGroupAutoCheck.queryGrayConsumerGroup();
         String subscribeScope = RocketMqSubscriptionDataUtils.buildSubscribeScope(subscriptionData.getTopic(),
                 consumerGroup, namesrvAddr);
         RocketMqSubscriptionDataUtils.resetsSql92SubscriptionData(subscriptionData, subscribeScope);

diff --git a/...nt/mq/grayscale/rocketmq/interceptor/RocketMqSchedulerRebuildSubscriptionInterceptor.java b/...nt/mq/grayscale/rocketmq/interceptor/RocketMqSchedulerRebuildSubscriptionInterceptor.java
@@ -64,6 +64,8 @@ private void buildSql92SubscriptionData(SubscriptionData subscriptionData, Rebal
             if (StringUtils.isEmpty(RocketMqGrayscaleConfigUtils.getGrayGroupTag())) {
                 RocketMqConsumerGroupAutoCheck.setMqClientInstance(topic, consumerGroup, instance);
             }
+            RocketMqConsumerGroupAutoCheck.queryGrayConsumerGroup();
+            RocketMqConsumerGroupAutoCheck.startSchedulerCheckGroupTask();
             String namesrvAddr = balance.getmQClientFactory().getClientConfig().getNamesrvAddr();
             String subscribeScope = RocketMqSubscriptionDataUtils.buildSubscribeScope(topic, consumerGroup,
                     namesrvAddr);