[CELEBORN-1052] Introduce dynamic ConfigService at SystemLevel and TenantLevel

common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala

@@ -366,6 +366,9 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
     }
   }
 
+  def dynamicConfigStoreBackend: String = get(DYNAMIC_CONFIG_STORE_BACKEND)
+  def dynamicConfigRefreshTime: Long = get(DYNAMIC_CONFIG_REFRESH_TIME)
+
   // //////////////////////////////////////////////////////
   //                      Network                        //
   // //////////////////////////////////////////////////////
@@ -4053,4 +4056,21 @@ object CelebornConf extends Logging {
       .doc("Kerberos keytab file path for HDFS storage connection.")
       .stringConf
       .createOptional
+
+  val DYNAMIC_CONFIG_STORE_BACKEND: ConfigEntry[String] =
+    buildConf("celeborn.dynamicConfig.store.backend")
+      .categories("master", "worker")
+      .doc("Store backend for dynamic config, NONE means disabling dynamic config store")
+      .version("0.4.0")
+      .stringConf
+      .checkValues(Set("FS", "NONE"))
+      .createWithDefault("NONE")
+
+  val DYNAMIC_CONFIG_REFRESH_TIME: ConfigEntry[Long] =
+    buildConf("celeborn.dynamicConfig.refresh.time")
+      .categories("master", "worker")
+      .version("0.4.0")
+      .doc("The time interval for refreshing the corresponding dynamic config periodically")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .createWithDefaultString("120s")
 }

conf/dynamicConfig.yaml.template

@@ -0,0 +1,27 @@
+#
+# 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.
+#
+-  level: SYSTEM
+   config:
+     celeborn.worker.directMemoryRatioToPauseReceive: 0.75
+
+
+-  tenantId: tenant_id
+   level: TENANT
+   config:
+
+
+

docs/configuration/master.md

@@ -19,6 +19,8 @@ license: |
 <!--begin-include-->
 | Key | Default | Description | Since |
 | --- | ------- | ----------- | ----- |
+| celeborn.dynamicConfig.refresh.time | 120s | The time interval for refreshing the corresponding dynamic config periodically | 0.4.0 | 
+| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic config, NONE means disabling dynamic config store | 0.4.0 | 
 | celeborn.master.estimatedPartitionSize.initialSize | 64mb | Initial partition size for estimation, it will change according to runtime stats. | 0.3.0 | 
 | celeborn.master.estimatedPartitionSize.update.initialDelay | 5min | Initial delay time before start updating partition size for estimation. | 0.3.0 | 
 | celeborn.master.estimatedPartitionSize.update.interval | 10min | Interval of updating partition size for estimation. | 0.3.0 | 

docs/configuration/worker.md

@@ -19,6 +19,8 @@ license: |
 <!--begin-include-->
 | Key | Default | Description | Since |
 | --- | ------- | ----------- | ----- |
+| celeborn.dynamicConfig.refresh.time | 120s | The time interval for refreshing the corresponding dynamic config periodically | 0.4.0 | 
+| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic config, NONE means disabling dynamic config store | 0.4.0 | 
 | celeborn.master.endpoints | &lt;localhost&gt;:9097 | Endpoints of master nodes for celeborn client to connect, allowed pattern is: `<host1>:<port1>[,<host2>:<port2>]*`, e.g. `clb1:9097,clb2:9098,clb3:9099`. If the port is omitted, 9097 will be used. | 0.2.0 | 
 | celeborn.master.estimatedPartitionSize.minSize | 8mb | Ignore partition size smaller than this configuration of partition size for estimation. | 0.3.0 | 
 | celeborn.shuffle.chunk.size | 8m | Max chunk size of reducer's merged shuffle data. For example, if a reducer's shuffle data is 128M and the data will need 16 fetch chunk requests to fetch. | 0.2.0 | 

service/src/main/scala/org/apache/celeborn/server/common/service/config/ConfigLevel.java

@@ -0,0 +1,23 @@
+/*
+ * 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.celeborn.server.common.service.config;
+
+public enum ConfigLevel {
+  SYSTEM,
+  TENANT,
+}

service/src/main/scala/org/apache/celeborn/server/common/service/config/ConfigService.java

@@ -0,0 +1,39 @@
+/*
+ * 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.celeborn.server.common.service.config;
+
+public interface ConfigService {
+
+  SystemConfig getSystemConfig();
+
+  TenantConfig getRawTenantConfig(String tenantId);
+
+  default DynamicConfig getTenantConfig(String tenantId) {
+    TenantConfig tenantConfig = getRawTenantConfig(tenantId);
+    if (tenantConfig == null || tenantConfig.getConfigs().isEmpty()) {
+      return getSystemConfig();
+    } else {
+      return tenantConfig;
+    }
+  }
+
+  void refreshAllCache();
+
+  void shutdown();
+
+}

service/src/main/scala/org/apache/celeborn/server/common/service/config/DynamicConfig.java

@@ -0,0 +1,117 @@
+/*
+ * 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.celeborn.server.common.service.config;
+
+import org.apache.celeborn.common.internal.config.ConfigEntry;
+import org.apache.celeborn.common.util.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Dynamic configuration is a type of configuration that can be changed at runtime as needed. It can be used at system level/tenant level.
+ * When applying dynamic configuration, the priority order is as follows: tenant level overrides system level,
+ * which in turn overrides static configuration(CelebornConf). This means that if a configuration is defined at the tenant level,
+ * it will be used instead of the system level or static configuration(CelebornConf). If the tenant-level configuration is missing,
+ * the system-level configuration will be used. If the system-level configuration is also missing, CelebornConf
+ * will be used as the default value.
+ */
+public abstract class DynamicConfig {
+  private static final Logger LOG = LoggerFactory.getLogger(DynamicConfig.class);
+  protected Map<String, String> configs = new HashMap<>();
+
+  public abstract DynamicConfig getParentLevelConfig();
+
+  public <T> T getWithDefaultValue(
+      String configKey, T defaultValue, Class<T> finalType, ConfigType configType) {
+    String configValue = configs.get(configKey);
+    T formatValue = configValue != null ? formatValue(configKey, configValue, finalType, configType) : null;
+    if (formatValue == null) {
+      return defaultValue;
+    } else {
+      return formatValue;
+    }
+  }
+
+  public <T> T getValue(String configKey, ConfigEntry<Object> configEntry, Class<T> finalType, ConfigType configType) {
+    String configValue = configs.get(configKey);
+    T formatValue = configValue != null ? formatValue(configKey, configValue, finalType, configType) : null;
+    if (formatValue == null) {
+      DynamicConfig parentLevelConfig = getParentLevelConfig();
+      return parentLevelConfig != null? parentLevelConfig.getValue(configKey, configEntry, finalType, configType): null;
+    } else {
+      return formatValue;
+    }
+  }
+
+  public <T> T formatValue(String configKey, String configValue, Class<T> finalType, ConfigType configType) {
+    try {
+      if (configValue != null) {
+        if (ConfigType.BYTES == configType) {
+          return convert(finalType, String.valueOf(Utils.byteStringAsBytes(configValue)));
+        } else if (ConfigType.TIME_MS == configType) {
+          return convert(finalType, String.valueOf(Utils.timeStringAsMs(configValue)));
+        } else {
+          return convert(finalType, configValue);
+        }
+      }
+    } catch (Exception e) {
+      LOG.warn("Config {} value format is not valid, refer to parent if exist", configKey, e);
+    }
+    return null;
+  }
+
+  public Map<String, String> getConfigs() {
+    return configs;
+  }
+
+  @Override
+  public String toString() {
+    final StringBuilder sb = new StringBuilder("DynamicConfig{");
+    sb.append("configs=").append(configs);
+    sb.append('}');
+    return sb.toString();
+  }
+
+  public enum ConfigType {
+    BYTES,
+    STRING,
+    TIME_MS,
+  }
+
+  public static <T> T convert(Class<T> clazz, String value) {
+    if (Boolean.TYPE == clazz) {
+      return (T) Boolean.valueOf(value);
+    } else if (Byte.TYPE == clazz) {
+      return (T) Byte.valueOf(value);
+    } else if (Short.TYPE == clazz) {
+      return (T) Short.valueOf(value);
+    } else if (Integer.TYPE == clazz) {
+      return (T) Integer.valueOf(value);
+    } else if (Long.TYPE == clazz) {
+      return (T) Long.valueOf(value);
+    } else if (Float.TYPE == clazz) {
+      return (T) Float.valueOf(value);
+    } else if (Double.TYPE == clazz) {
+      return (T) Double.valueOf(value);
+    }
+    return (T) value;
+  }
+}

service/src/main/scala/org/apache/celeborn/server/common/service/config/DynamicConfigServiceFactory.java

@@ -0,0 +1,32 @@
+/*
+ * 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.celeborn.server.common.service.config;
+
+import org.apache.celeborn.common.CelebornConf;
+
+public class DynamicConfigServiceFactory {
+
+  public static ConfigService getConfigService(CelebornConf celebornConf) {
+    String configStoreBackend = celebornConf.dynamicConfigStoreBackend();
+    if ("FS".equals(configStoreBackend)) {
+      return new FsConfigServiceImpl(celebornConf);
+    }
+
+    return null;
+  }
+}