From fde6b5a6db87f37c4e86cc5ecda04f5ea889e7e2 Mon Sep 17 00:00:00 2001 From: chenyijiang Date: Thu, 2 Dec 2021 18:49:51 +0800 Subject: [PATCH] Add merged properties config. --- .../component/telemetry/SamplerConfig.java | 82 +++++++--------- .../telemetry/TestMixerProvider.java | 9 +- .../hook/MergedPropertiesConfig.java | 93 +++++++++++++++++++ 3 files changed, 130 insertions(+), 54 deletions(-) create mode 100644 sdk-infrastructure/src/main/java/group/rxcloud/capa/infrastructure/hook/MergedPropertiesConfig.java diff --git a/sdk-component/src/main/java/group/rxcloud/capa/component/telemetry/SamplerConfig.java b/sdk-component/src/main/java/group/rxcloud/capa/component/telemetry/SamplerConfig.java index 94d4bfb..f321db8 100644 --- a/sdk-component/src/main/java/group/rxcloud/capa/component/telemetry/SamplerConfig.java +++ b/sdk-component/src/main/java/group/rxcloud/capa/component/telemetry/SamplerConfig.java @@ -16,18 +16,12 @@ */ package group.rxcloud.capa.component.telemetry; -import com.google.common.collect.Lists; import group.rxcloud.capa.component.telemetry.metrics.CapaMeterProviderBuilder; import group.rxcloud.capa.infrastructure.CapaProperties; -import group.rxcloud.capa.infrastructure.hook.ConfigurationHooks; +import group.rxcloud.capa.infrastructure.hook.MergedPropertiesConfig; import group.rxcloud.capa.infrastructure.hook.Mixer; -import group.rxcloud.cloudruntimes.domain.core.configuration.SubConfigurationResp; -import group.rxcloud.cloudruntimes.utils.TypeRef; -import org.apache.commons.collections.CollectionUtils; -import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import reactor.core.publisher.Flux; import java.io.Serializable; import java.util.function.Supplier; @@ -37,12 +31,6 @@ */ public class SamplerConfig implements Serializable { - private static final long serialVersionUID = -2113523925814197551L; - - public static final transient String FILE_PATH = "capa-component-telemetry-sample.properties"; - - public static final transient String COMMON_FILE_SUFFIX = "telemetry-common"; - /** * Sample all data as default. */ @@ -53,25 +41,44 @@ public class SamplerConfig implements Serializable { public static final transient SamplerConfig CONFIG = new SamplerConfig(); - public static final transient Supplier DEFAULT_SUPPLIER = () -> { - return CONFIG; - }; + private static final long serialVersionUID = -2113523925814197551L; private static final transient Logger log = LoggerFactory.getLogger(CapaMeterProviderBuilder.class); + public static transient Supplier DEFAULT_SUPPLIER = () -> { + return CONFIG; + }; static { Mixer.configurationHooksNullable().ifPresent(hooks -> { + + String fileName = "capa-component-telemetry-sample.properties"; + + String suffix = "telemetry-common"; + try { - subscribeConfiguration(hooks, hooks.defaultConfigurationAppId(), true); - } catch (Throwable throwable) { - log.warn("Fail to load global telemetry config. Dynamic global config is disabled for capa telemetry.", - throwable); - } - try { - subscribeConfiguration(hooks, - CapaProperties.COMPONENT_PROPERTIES_SUPPLIER.apply(COMMON_FILE_SUFFIX).getProperty("appId"), - false); + MergedPropertiesConfig config = new MergedPropertiesConfig(fileName, hooks + .defaultConfigurationAppId(), + CapaProperties.COMPONENT_PROPERTIES_SUPPLIER.apply(suffix).getProperty("appId")); + String metricKey = "metricsEnable"; + String traceKey = "traceEnable"; + SamplerConfig dynamicConfig = new SamplerConfig() { + @Override + public Boolean isMetricsEnable() { + return !config.containsKey(metricKey) || Boolean.TRUE.toString() + .equalsIgnoreCase(config.get(metricKey)); + } + + @Override + public Boolean isTraceEnable() { + return !config.containsKey(traceKey) || Boolean.TRUE.toString() + .equalsIgnoreCase(config.get(traceKey)); + } + }; + + DEFAULT_SUPPLIER = () -> { + return dynamicConfig; + }; } catch (Throwable throwable) { log.warn("Fail to load global telemetry config. Dynamic global config is disabled for capa telemetry.", throwable); @@ -84,31 +91,6 @@ public class SamplerConfig implements Serializable { private Boolean traceEnable; - private static void subscribeConfiguration(ConfigurationHooks configurationHooks, String appId, boolean prior) { - String storeName = configurationHooks.registryStoreNames().get(0); - Flux> configFlux = configurationHooks.subscribeConfiguration( - storeName, - appId, - Lists.newArrayList(FILE_PATH), - null, - StringUtils.EMPTY, - StringUtils.EMPTY, - TypeRef.get(SamplerConfig.class)); - configFlux.subscribe(resp -> { - if (CollectionUtils.isNotEmpty(resp.getItems())) { - SamplerConfig config = resp.getItems().get(0).getContent(); - if (config != null) { - if (config.metricsEnable != null && (prior || CONFIG.metricsEnable == null)) { - CONFIG.metricsEnable = config.metricsEnable; - } - if (config.traceEnable != null && (prior || CONFIG.traceEnable == null)) { - CONFIG.traceEnable = config.traceEnable; - } - } - } - }); - } - public Boolean isMetricsEnable() { return metricsEnable == null ? DEFAULT_CONFIG.metricsEnable : metricsEnable; } diff --git a/sdk-component/src/test/java/group/rxcloud/capa/component/telemetry/TestMixerProvider.java b/sdk-component/src/test/java/group/rxcloud/capa/component/telemetry/TestMixerProvider.java index 2f1ffe5..3518ada 100644 --- a/sdk-component/src/test/java/group/rxcloud/capa/component/telemetry/TestMixerProvider.java +++ b/sdk-component/src/test/java/group/rxcloud/capa/component/telemetry/TestMixerProvider.java @@ -27,6 +27,7 @@ import java.util.List; import java.util.Map; +import java.util.Properties; /** * @author: chenyijiang @@ -34,8 +35,8 @@ */ public class TestMixerProvider implements Mixer.MixerProvider { - SamplerConfig app = new SamplerConfig() {{setMetricsEnable(false);}}; - SamplerConfig global = new SamplerConfig() {{setTraceEnable(false); setMetricsEnable(true);}}; + Properties app = new Properties() {{put("metricsEnable", false);}}; + Properties global = new Properties() {{put("traceEnable", false); put("metricsEnable", true);}}; private ConfigurationHooks configurationHooks = new ConfigurationHooks() { @@ -52,7 +53,7 @@ public String defaultConfigurationAppId() { @Override public Flux> subscribeConfiguration(String storeName, String appId, List keys, Map metadata, String group, String label, TypeRef type) { - if (type.getType() == SamplerConfig.class) { + if (type.getType() == Properties.class) { if ("123".equals(appId)) { return Flux.just(getSubscribeResponse(global)); } @@ -65,7 +66,7 @@ public Flux> subscribeConfiguration(String storeName } }; - private SubConfigurationResp getSubscribeResponse(SamplerConfig samplerConfig) { + private SubConfigurationResp getSubscribeResponse(Properties samplerConfig) { SubConfigurationResp subConfigurationResp = new SubConfigurationResp<>(); ConfigurationItem item = new ConfigurationItem(); item.setContent(samplerConfig); diff --git a/sdk-infrastructure/src/main/java/group/rxcloud/capa/infrastructure/hook/MergedPropertiesConfig.java b/sdk-infrastructure/src/main/java/group/rxcloud/capa/infrastructure/hook/MergedPropertiesConfig.java new file mode 100644 index 0000000..91fd138 --- /dev/null +++ b/sdk-infrastructure/src/main/java/group/rxcloud/capa/infrastructure/hook/MergedPropertiesConfig.java @@ -0,0 +1,93 @@ +/* + * 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 group.rxcloud.capa.infrastructure.hook; + +import group.rxcloud.cloudruntimes.domain.core.configuration.SubConfigurationResp; +import group.rxcloud.cloudruntimes.utils.TypeRef; +import reactor.core.publisher.Flux; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.atomic.AtomicReferenceArray; + +/** + * Config provider to merge multiple properties file which takes the input order as their priority. + */ +public class MergedPropertiesConfig { + + private final String fileName; + + private final AtomicReferenceArray properties; + + private final Object lock = new Object(); + + private volatile Map merged; + + public MergedPropertiesConfig(String fileName, String... appIds) { + this.fileName = fileName; + properties = new AtomicReferenceArray<>(appIds.length); + merged = new HashMap<>(); + Mixer.configurationHooksNullable().ifPresent(hooks -> { + for (int i = 0; i < appIds.length; i++) { + subscribeConfigurationByAppId(hooks, appIds[i], i); + } + }); + } + + public boolean containsKey(String key) { + return merged.containsKey(key); + } + + public String get(String key) { + return merged.get(key); + } + + private void subscribeConfigurationByAppId(ConfigurationHooks configurationHooks, String appId, int index) { + String storeName = configurationHooks.registryStoreNames().get(0); + + Flux> configFlux = configurationHooks.subscribeConfiguration( + storeName, + appId, + Collections.singletonList(fileName), + null, + "", + "", + TypeRef.get(Properties.class)); + + + configFlux.subscribe(resp -> { + if (!resp.getItems().isEmpty()) { + properties.set(index, resp.getItems().get(0).getContent()); + } else { + properties.set(index, null); + } + + synchronized (lock) { + Map merged = new HashMap<>(); + for (int i = 0; i < properties.length(); i++) { + Properties item = properties.get(i); + if (item != null) { + item.forEach((k, v) -> merged.putIfAbsent(String.valueOf(k), String.valueOf(v))); + } + } + this.merged = merged; + } + }); + } +}