Merge pull request #74 from reactivegroup/feature/metrics

upgrade log throwable

Author: kevinten10

capa-spi-aws-config/pom.xml

@@ -23,7 +23,7 @@
     <parent>
         <artifactId>capa-aws-parent</artifactId>
         <groupId>group.rxcloud</groupId>
-        <version>1.10.12.2.RELEASE</version>
+        <version>1.11.13.2.RELEASE</version>
     </parent>
 
     <artifactId>capa-spi-aws-config</artifactId>
@@ -39,6 +39,21 @@
             <artifactId>appconfig</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>sts</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.netty</groupId>
+                    <artifactId>netty-codec-http2</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>software.amazon.awssdk</groupId>
+                    <artifactId>netty-nio-client</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <dependency>
             <groupId>com.google.guava</groupId>
             <artifactId>guava</artifactId>

capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStore.java

@@ -128,27 +128,20 @@ public void close() {
      */
     @Override
     protected <T> Mono<List<ConfigurationItem<T>>> doGet(String appId, String group, String label, List<String> keys, Map<String, String> metadata, TypeRef<T> type) {
-        List<ConfigurationItem<T>> items = new ArrayList<>();
         if (CollectionUtils.isNullOrEmpty(keys)) {
-            LOGGER.warn("[Capa.Config] keys is null or empty,appId:{}", appId);
+            LOGGER.warn("[[type=Capa.Config]] keys is null or empty,appId:{}", appId);
             return Mono.error(new CapaException(CapaErrorContext.PARAMETER_ERROR, "keys is null or empty"));
         }
 
         String applicationName = String.format(APPCONFIG_NAME_FORMAT, appId, CapaFoundation.getEnv(FoundationType.TRIP));
         String configurationName = keys.get(0);
 
-        GetConfigurationRequest request = GetConfigurationRequest.builder()
-                .application(applicationName)
-                .clientId(UUID.randomUUID().toString())
-                .configuration(configurationName)
-                .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getConfigAwsAppConfigEnv())
-                .build();
+        //init config and create subscribe polling
+        initAndSubscribe(applicationName, configurationName, group, label, metadata, type);
 
-        return Mono.fromCallable(() -> {
-            GetConfigurationResponse response = appConfigAsyncClient.getConfiguration(request).get(REQUEST_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);
-            return convertToConfigurationList(response, configurationName, type);
-        })
-                .doOnError(e -> LOGGER.error("[Capa.Config.getConfiguration] error occurs when getconfiguration, request:{}", request, e));
+        //get current config from local
+        ConfigurationItem<T> configurationItem = (ConfigurationItem<T>) getConfiguration(applicationName, configurationName).getConfigurationItem();
+        return Mono.just(Lists.newArrayList(configurationItem));
     }
 
     @Override
@@ -157,7 +150,7 @@ protected <T> Flux<SubscribeResp<T>> doSubscribe(String appId, String group, Str
         String applicationName = String.format(APPCONFIG_NAME_FORMAT, appId, CapaFoundation.getEnv(FoundationType.TRIP));
         String configurationName = keys.get(0);
 
-        initSubscribe(applicationName, configurationName, group, label, metadata, type);
+        initAndSubscribe(applicationName, configurationName, group, label, metadata, type);
         return doSub(applicationName, configurationName, group, label, metadata, type, appId);
     }
 
@@ -175,7 +168,7 @@ protected <T> Flux<SubscribeResp<T>> doSubscribe(String appId, String group, Str
     private synchronized <T> Configuration<T> initConfig(String applicationName, String configurationName, String group, String label, Map<String, String> metadata, TypeRef<T> type) {
         // double check whether has been initialized
         if (isInitialized(applicationName, configurationName)) {
-            LOGGER.info("[Capa.Config.initConfig] config has been initialized before,applicationName:{},configurationName:{}", applicationName, configurationName);
+            LOGGER.info("[[type=Capa.Config.initConfig]] config has been initialized before,applicationName:{},configurationName:{}", applicationName, configurationName);
             return Configuration.EMPTY;
         }
         String version = getCurVersion(applicationName, configurationName);
@@ -188,28 +181,28 @@ private synchronized <T> Configuration<T> initConfig(String applicationName, Str
                 .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getConfigAwsAppConfigEnv())
                 .build();
 
-        LOGGER.debug("[Capa.Config.initConfig] call getconfiguration in init process,request:{}", request);
+        LOGGER.debug("[[type=Capa.Config.initConfig]] call getconfiguration in init process,request:{}", request);
 
         return Mono.fromCallable(() -> {
             GetConfigurationResponse response = appConfigAsyncClient.getConfiguration(request).get(REQUEST_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);
-            LOGGER.debug("[Capa.Config.initConfig] call getconfiguration in init process,response:{}", response);
+            LOGGER.debug("[[type=Capa.Config.initConfig]] call getconfiguration in init process,response:{}", response);
             return response;
         })
-                .doOnError(e -> LOGGER.error("[initConfig] error occurs when getconfiguration in init process, request:{}", request, e))
+                .doOnError(e -> LOGGER.error("[[type=Capa.Config.initConfig]] error occurs when getconfiguration in init process, request:{}", request, e))
                 .map(resp -> initConfigurationItem(applicationName, configurationName, type, resp.content(), resp.configurationVersion()))
                 .block();
     }
 
-    private <T> void initSubscribe(String applicationName, String configurationName, String group, String label, Map<String, String> metadata, TypeRef<T> type) {
+    private <T> void initAndSubscribe(String applicationName, String configurationName, String group, String label, Map<String, String> metadata, TypeRef<T> type) {
         if (!isInitialized(applicationName, configurationName)) {
             initConfig(applicationName, configurationName, group, label, metadata, type);
         }
         if (!isSubscribed(applicationName, configurationName)) {
-            createSubscribe(applicationName, configurationName, type);
+            createSubscribePolling(applicationName, configurationName, type);
         }
     }
 
-    private synchronized <T> void createSubscribe(String applicationName, String configurationName, TypeRef<T> type) {
+    private synchronized <T> void createSubscribePolling(String applicationName, String configurationName, TypeRef<T> type) {
         if (isSubscribed(applicationName, configurationName)) {
             return;
         }
@@ -229,17 +222,17 @@ private synchronized <T> void createSubscribe(String applicationName, String con
                                 .clientConfigurationVersion(version)
                                 .environment(AwsCapaConfigurationProperties.AppConfigProperties.Settings.getConfigAwsAppConfigEnv())
                                 .build();
-                        LOGGER.debug("[Capa.Config.subscribePolling] subscribe polling task start,request:{}", request);
+                        LOGGER.debug("[[type=Capa.Config.subscribePolling]] subscribe polling task start,request:{}", request);
 
                         GetConfigurationResponse resp = null;
                         try {
                             resp = appConfigAsyncClient.getConfiguration(request).get(REQUEST_TIMEOUT_IN_SECONDS, TimeUnit.SECONDS);
                         } catch (InterruptedException | ExecutionException | TimeoutException e) {
                             //catch error,log error and not trigger listeners
-                            LOGGER.error("[Capa.Config.subscribePolling] error occurs when getConfiguration in polling process,configurationName:{},version:{}", request.configuration(), request.clientConfigurationVersion(), e);
+                            LOGGER.error("[[type=Capa.Config.subscribePolling]] error occurs when getConfiguration in polling process,configurationName:{},version:{}", request.configuration(), request.clientConfigurationVersion(), e);
                         }
 
-                        LOGGER.debug("[Capa.Config.subscribePolling] subscribe polling task end,response:{}", resp);
+                        LOGGER.debug("[[type=Capa.Config.subscribePolling]] subscribe polling task end,response:{}", resp);
 
                         if (resp != null && !Objects.equals(resp.configurationVersion(), version)) {
                             fluxSink.next(resp);
@@ -255,7 +248,7 @@ private synchronized <T> void createSubscribe(String applicationName, String con
                 })
                 .filter(resp -> resp != Configuration.EMPTY)
                 .subscribe(resp -> {
-                    LOGGER.info("[Capa.Config.triggerListener] receive changes and trigger listeners,response:{}", resp);
+                    LOGGER.info("[[type=Capa.Config.triggerListener]] receive changes and trigger listeners,response:{}", resp);
                     resp.triggers(resp.getConfigurationItem());
                 });
     }
@@ -268,7 +261,7 @@ private <T> Flux<SubscribeResp<T>> doSub(String applicationName, String configur
         return Flux
                 .create(fluxSink -> {
                     configuration.addListener(configurationItem -> {
-                        LOGGER.info("[Capa.Config.listenerOnChange] listener onChanges, configurationItem:{}", configurationItem);
+                        LOGGER.info("[[type=Capa.Config.listenerOnChange]] listener onChanges, configurationItem key:{}", configurationItem.getKey());
                         fluxSink.next(configurationItem);
                     });
                 })
@@ -330,7 +323,7 @@ private <T> Configuration<T> updateConfigurationItem(String applicationName, Str
             configuration.setConfigurationItem(configurationItem);
 
             configMap.put(configurationName, configuration);
-            LOGGER.info("[Capa.Config.updateConfig] update config,key configurationName:{},value configuration:{}", configurationName, configuration);
+            LOGGER.info("[[type=Capa.Config.updateConfig]] update config,key configurationName:{},value configuration:{}", configurationName, configuration);
             return configuration;
         }
     }
@@ -355,11 +348,11 @@ private <T> Configuration<T> initConfigurationItem(String applicationName, Strin
 
         configMap.put(configurationName, configuration);
 
-        LOGGER.info("[Capa.Config.initConfig] process initConfigurationItem,put key configurationName:{},value configuration:{}", configurationName, configuration);
+        LOGGER.info("[[type=Capa.Config.initConfig]] process initConfigurationItem,put key configurationName:{},value configuration:{}", configurationName, configuration);
 
         if (initApplication) {
             versionMap.put(applicationName, configMap);
-            LOGGER.info("[Capa.Config.initConfig] process initConfigurationItem,put key applicationName:{}", applicationName);
+            LOGGER.info("[[type=Capa.Config.initConfig]] process initConfigurationItem,put key applicationName:{}", applicationName);
         }
         return configuration;
     }

capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/entity/Configuration.java

@@ -36,9 +36,8 @@ public class Configuration<T> {
     public String toString() {
         return "Configuration{" +
                 "clientConfigurationVersion='" + clientConfigurationVersion + '\'' +
-                ", configurationItem=" + configurationItem +
-                ", lock=" + lock +
-                ", listeners=" + listeners +
+                ", configurationItem=" + configurationItem.getContent().toString() +
+                ", listenersCount=" + listeners.size() +
                 ", initialized=" + initialized +
                 ", subscribed=" + subscribed +
                 '}';

capa-spi-aws-config/src/main/java/group/rxcloud/capa/spi/aws/config/serializer/PropertiesSerializer.java

@@ -23,7 +23,7 @@
 
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.LinkedHashMap;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Properties;
 
@@ -47,7 +47,7 @@ private Map<String, String> parsePropertiesToMap(InputStream inputStream) {
         } catch (IOException e) {
             LOGGER.error("properties load error", e);
         }
-        Map<String, String> map = new LinkedHashMap<>(properties.size());
+        Map<String, String> map = new HashMap<>(properties.size());
         for (String key : properties.stringPropertyNames()) {
             map.put(key, properties.getProperty(key));
         }

capa-spi-aws-config/src/test/java/group/rxcloud/capa/spi/aws/config/AwsCapaConfigStoreIgnoreTest.java

@@ -52,6 +52,21 @@ void testDoGet(){
         Mono<List<ConfigurationItem<User>>> mono2 = ins.doGet("100012345", "", "", Lists.newArrayList("test1.json"), new HashMap<>(), TypeRef.get(User.class));
         System.out.println(mono2.block().get(0).getContent().getAge());
 
+        Flux<SubscribeResp<User>> flux1 = ins.doSubscribe("100012345", "", "", Lists.newArrayList("test1.json"), new HashMap<>(), TypeRef.get(User.class));
+        flux1.subscribe(resp -> {
+            System.out.println("1:"+resp.getItems().get(0).getContent().getAge());
+        });
+
+        long start = System.currentTimeMillis();
+        while(System.currentTimeMillis()-start<20000){
+
+        }
+        //change config
+        Mono<List<ConfigurationItem<User>>> mono3 = ins.doGet("100012345", "", "", Lists.newArrayList("test1.json"), new HashMap<>(), TypeRef.get(User.class));
+        System.out.println(mono3.block().get(0).getContent().getAge());
+
+        System.out.println("");
+
 //        Mono<List<ConfigurationItem<User>>> mono = ins.doGet("100012345", "", "", Lists.newArrayList("test1.json"), new HashMap<>(), TypeRef.get(User.class));
 //        System.out.println(mono.block().get(0).getContent().getAge());
 //

capa-spi-aws-infrastructure/pom.xml

@@ -23,7 +23,7 @@
     <parent>
         <artifactId>capa-aws-parent</artifactId>
         <groupId>group.rxcloud</groupId>
-        <version>1.10.12.2.RELEASE</version>
+        <version>1.11.13.2.RELEASE</version>
     </parent>
 
     <artifactId>capa-spi-aws-infrastructure</artifactId>

capa-spi-aws-log/pom.xml

@@ -23,7 +23,7 @@
     <parent>
         <artifactId>capa-aws-parent</artifactId>
         <groupId>group.rxcloud</groupId>
-        <version>1.10.12.2.RELEASE</version>
+        <version>1.11.13.2.RELEASE</version>
     </parent>
 
     <artifactId>capa-spi-aws-log</artifactId>
@@ -56,6 +56,21 @@
             <artifactId>cloudwatch</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>sts</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>io.netty</groupId>
+                    <artifactId>netty-codec-http2</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>software.amazon.awssdk</groupId>
+                    <artifactId>netty-nio-client</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <!-- fixme: can we only use jackson -->
         <dependency>
             <groupId>com.google.code.gson</groupId>

capa-spi-aws-log/src/main/java/group/rxcloud/capa/spi/aws/log/appender/CapaAwsLog4jAppender.java

@@ -17,8 +17,8 @@
 package group.rxcloud.capa.spi.aws.log.appender;
 
 import group.rxcloud.capa.infrastructure.hook.Mixer;
-import group.rxcloud.capa.infrastructure.hook.TelemetryHooks;
 import group.rxcloud.capa.spi.aws.log.enums.CapaLogLevel;
+import group.rxcloud.capa.spi.aws.log.manager.CustomLogManager;
 import group.rxcloud.capa.spi.aws.log.manager.LogAppendManager;
 import group.rxcloud.capa.spi.aws.log.manager.LogManager;
 import group.rxcloud.capa.spi.log.CapaLog4jAppenderSpi;
@@ -33,43 +33,57 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 public class CapaAwsLog4jAppender extends CapaLog4jAppenderSpi {
+
     /**
      * The error type name of the log4j appender.
      */
     protected static final String LOG_LOG4J_APPENDER_ERROR_TYPE = "Log4jAppendLogsError";
+
     /**
      * Number of counts each time.
      */
     protected static final Integer COUNTER_NUM = 1;
-    /**
-     * The instance of the {@link TelemetryHooks}.
-     */
-    private static final Optional<TelemetryHooks> TELEMETRY_HOOKS;
+
     /**
      * The namespace for logging error.
      * TODO Set variables to common variables
      */
     private static final String LOG_ERROR_NAMESPACE = "CloudWatchLogs";
+
     /**
      * The metric name for logging error.
      * TODO Set variables to common variables
      */
     private static final String LOG_ERROR_METRIC_NAME = "LogError";
+
+    private static final AtomicBoolean METRIC_INIT = new AtomicBoolean(false);
+
     /**
      * Init an instance of {@link LongCounter}.
      */
     protected static Optional<LongCounter> LONG_COUNTER = Optional.empty();
 
     static {
         PluginManager.addPackage("group.rxcloud.capa.spi.aws.log.appender");
-        TELEMETRY_HOOKS = Mixer.telemetryHooksNullable();
-        TELEMETRY_HOOKS.ifPresent(telemetryHooks -> {
-            Meter meter = telemetryHooks.buildMeter(LOG_ERROR_NAMESPACE).block();
-            LongCounter longCounter = meter.counterBuilder(LOG_ERROR_METRIC_NAME).build();
-            LONG_COUNTER = Optional.ofNullable(longCounter);
-        });
+    }
+
+    static Optional<LongCounter> getCounterOpt() {
+        if (METRIC_INIT.get()) {
+            return LONG_COUNTER;
+        }
+        synchronized (METRIC_INIT) {
+            if (METRIC_INIT.compareAndSet(false, true)) {
+                Mixer.telemetryHooksNullable().ifPresent(telemetryHooks -> {
+                    Meter meter = telemetryHooks.buildMeter(LOG_ERROR_NAMESPACE).block();
+                    LongCounter longCounter = meter.counterBuilder(LOG_ERROR_METRIC_NAME).build();
+                    LONG_COUNTER = Optional.ofNullable(longCounter);
+                });
+            }
+        }
+        return LONG_COUNTER;
     }
 
     @Override
@@ -81,20 +95,23 @@ public void appendLog(LogEvent event) {
                 return;
             }
             Optional<CapaLogLevel> capaLogLevel = CapaLogLevel.toCapaLogLevel(event.getLevel().name());
-            if(capaLogLevel.isPresent() && LogManager.logsCanOutput(capaLogLevel.get())){
+            if (capaLogLevel.isPresent() && LogManager.logsCanOutput(capaLogLevel.get())) {
                 String message = event.getMessage().getFormattedMessage();
                 ReadOnlyStringMap contextData = event.getContextData();
                 Map<String, String> MDCTags = contextData == null ? new HashMap<>() : contextData.toMap();
-                LogAppendManager.appendLogs(message, MDCTags, event.getLevel().name());
+                LogAppendManager.appendLogs(message, MDCTags, event.getLoggerName(), event.getThreadName(),
+                        event.getLevel().name(), event.getTimeMillis(), event.getThrown());
             }
         } catch (Exception e) {
             try {
+                CustomLogManager.error("CapaAwsLog4jAppender appender log error.", e);
                 //Enhance function without affecting function
-                LONG_COUNTER.ifPresent(longCounter -> {
-                    longCounter.bind(Attributes.of(AttributeKey.stringKey(LOG_LOG4J_APPENDER_ERROR_TYPE), LOG_LOG4J_APPENDER_ERROR_TYPE))
-                            .add(COUNTER_NUM);
+                getCounterOpt().ifPresent(longCounter -> {
+                    longCounter.bind(Attributes
+                            .of(AttributeKey.stringKey(LOG_LOG4J_APPENDER_ERROR_TYPE), e.getClass().getName()))
+                               .add(COUNTER_NUM);
                 });
-            } finally {
+            } catch (Throwable ex) {
             }
 
         }