You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/07/08 12:14:58 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-28445] Support operator configuration from multiple configmaps
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 56381ec [FLINK-28445] Support operator configuration from multiple configmaps
56381ec is described below
commit 56381ecf4515ff12d72fe6b1a245160580aacb72
Author: Matyas Orhidi <ma...@apple.com>
AuthorDate: Thu Jul 7 13:30:39 2022 +0200
[FLINK-28445] Support operator configuration from multiple configmaps
---
.../flink-conf.yaml} | 10 ++----
.../{ => conf-override}/kustomization.yaml | 8 ++++-
examples/kustomize/{ => conf-override}/patch.yaml | 42 +++++++++-------------
examples/kustomize/{ => conf-override}/render | 0
.../kustomize/{ => sidecar}/kustomization.yaml | 0
examples/kustomize/{ => sidecar}/patch.yaml | 0
examples/kustomize/{ => sidecar}/render | 0
examples/kustomize/{ => sidecar}/values.yaml | 0
.../operator/config/FlinkConfigManager.java | 21 +++++++++--
.../config/FlinkOperatorConfiguration.java | 3 +-
.../flink/kubernetes/operator/utils/EnvUtils.java | 34 ++++++------------
.../operator/config/FlinkConfigManagerTest.java | 22 ++++++++++++
.../operator/admission/FlinkOperatorWebhook.java | 8 ++---
.../conf/log4j-operator.properties | 2 +-
14 files changed, 82 insertions(+), 68 deletions(-)
diff --git a/examples/kustomize/kustomization.yaml b/examples/kustomize/conf-override/flink-conf.yaml
similarity index 89%
copy from examples/kustomize/kustomization.yaml
copy to examples/kustomize/conf-override/flink-conf.yaml
index 82409f5..9802bc7 100644
--- a/examples/kustomize/kustomization.yaml
+++ b/examples/kustomize/conf-override/flink-conf.yaml
@@ -15,11 +15,5 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-
-resources:
- - template.yaml
-patches:
- - path: patch.yaml
- target:
- kind: Deployment
- name: ".*"
+kubernetes.operator.dynamic.namespaces.enabled: true
+kubernetes.operator.watched.namespaces: default,flink
diff --git a/examples/kustomize/kustomization.yaml b/examples/kustomize/conf-override/kustomization.yaml
similarity index 86%
copy from examples/kustomize/kustomization.yaml
copy to examples/kustomize/conf-override/kustomization.yaml
index 82409f5..03473fe 100644
--- a/examples/kustomize/kustomization.yaml
+++ b/examples/kustomize/conf-override/kustomization.yaml
@@ -15,9 +15,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
-
resources:
- template.yaml
+configMapGenerator:
+ - name: flink-operator-config-override
+ behavior: create
+ files:
+ - flink-conf.yaml
+generatorOptions:
+ disableNameSuffixHash: true
patches:
- path: patch.yaml
target:
diff --git a/examples/kustomize/patch.yaml b/examples/kustomize/conf-override/patch.yaml
similarity index 58%
copy from examples/kustomize/patch.yaml
copy to examples/kustomize/conf-override/patch.yaml
index ae5dbec..8aced08 100644
--- a/examples/kustomize/patch.yaml
+++ b/examples/kustomize/conf-override/patch.yaml
@@ -25,33 +25,23 @@ spec:
spec:
containers:
- name: flink-kubernetes-operator
+ env:
+ - name: CONF_OVERRIDE_DIR
+ value: /opt/flink/conf-override
volumeMounts:
- - name: flink-log
- mountPath: /opt/flink/log
- resources:
- requests:
- memory: "2.5Gi"
- cpu: "1000m"
- limits:
- memory: "2.5Gi"
- cpu: "2000m"
+ - name: flink-operator-config-override-volume
+ mountPath: /opt/flink/conf-override
- name: flink-webhook
+ env:
+ - name: CONF_OVERRIDE_DIR
+ value: /opt/flink/conf-override
volumeMounts:
- - name: flink-log
- mountPath: /opt/flink/log
- resources:
- requests:
- memory: "0.5Gi"
- cpu: "200m"
- limits:
- memory: "0.5Gi"
- cpu: "500m"
- - name: fluentbit
- image: fluent/fluent-bit:1.9.1-debug
- command: [ 'sh','-c','/fluent-bit/bin/fluent-bit -i tail -p refresh_interval=10 -p path=/opt/flink/log/*.log -p multiline.parser=java -o stdout' ]
- volumeMounts:
- - name: flink-log
- mountPath: /opt/flink/log
+ - name: flink-operator-config-override-volume
+ mountPath: /opt/flink/conf-override
volumes:
- - name: flink-log
- emptyDir: { }
+ - name: flink-operator-config-override-volume
+ configMap:
+ name: flink-operator-config-override
+ items:
+ - key: flink-conf.yaml
+ path: flink-conf.yaml
diff --git a/examples/kustomize/render b/examples/kustomize/conf-override/render
similarity index 100%
copy from examples/kustomize/render
copy to examples/kustomize/conf-override/render
diff --git a/examples/kustomize/kustomization.yaml b/examples/kustomize/sidecar/kustomization.yaml
similarity index 100%
rename from examples/kustomize/kustomization.yaml
rename to examples/kustomize/sidecar/kustomization.yaml
diff --git a/examples/kustomize/patch.yaml b/examples/kustomize/sidecar/patch.yaml
similarity index 100%
rename from examples/kustomize/patch.yaml
rename to examples/kustomize/sidecar/patch.yaml
diff --git a/examples/kustomize/render b/examples/kustomize/sidecar/render
similarity index 100%
rename from examples/kustomize/render
rename to examples/kustomize/sidecar/render
diff --git a/examples/kustomize/values.yaml b/examples/kustomize/sidecar/values.yaml
similarity index 100%
rename from examples/kustomize/values.yaml
rename to examples/kustomize/sidecar/values.yaml
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
index e7fbbfc..caccf10 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java
@@ -25,6 +25,7 @@ import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkSessionJobSpec;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.utils.EnvUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.shaded.guava30.com.google.common.cache.Cache;
@@ -72,7 +73,7 @@ public class FlinkConfigManager {
}
public FlinkConfigManager(Consumer<Set<String>> namespaceListener) {
- this(GlobalConfiguration.loadConfiguration(), namespaceListener);
+ this(loadGlobalConfiguration(), namespaceListener);
}
public FlinkConfigManager(
@@ -211,11 +212,27 @@ public class FlinkConfigManager {
return cache;
}
+ private static Configuration loadGlobalConfiguration() {
+ return loadGlobalConfiguration(EnvUtils.get(EnvUtils.ENV_CONF_OVERRIDE_DIR));
+ }
+
+ @VisibleForTesting
+ protected static Configuration loadGlobalConfiguration(Optional<String> confOverrideDir) {
+ if (confOverrideDir.isPresent()) {
+ Configuration configOverrides =
+ GlobalConfiguration.loadConfiguration(confOverrideDir.get());
+ LOG.debug("Loading default configuration with overrides from " + confOverrideDir.get());
+ return GlobalConfiguration.loadConfiguration(configOverrides);
+ }
+ LOG.debug("Loading default configuration");
+ return GlobalConfiguration.loadConfiguration();
+ }
+
private class ConfigUpdater implements Runnable {
public void run() {
try {
LOG.debug("Checking for config update changes...");
- updateDefaultConfig(GlobalConfiguration.loadConfiguration());
+ updateDefaultConfig(loadGlobalConfiguration());
} catch (Exception e) {
LOG.error("Error while updating operator configuration", e);
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
index 68c4966..52fbb85 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkOperatorConfiguration.java
@@ -91,11 +91,10 @@ public class FlinkOperatorConfiguration {
.OPERATOR_SAVEPOINT_HISTORY_MAX_AGE_THRESHOLD);
String flinkServiceHostOverride = null;
- if (EnvUtils.get("KUBERNETES_SERVICE_HOST") == null) {
+ if (EnvUtils.get(EnvUtils.ENV_KUBERNETES_SERVICE_HOST).isEmpty()) {
// not running in k8s, simplify local development
flinkServiceHostOverride = "localhost";
}
-
var watchedNamespaces =
new HashSet<>(
Arrays.asList(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java
index ed2c86f..2103695 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EnvUtils.java
@@ -30,12 +30,10 @@ import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeParseException;
-import java.util.Arrays;
-import java.util.HashSet;
+import java.util.NoSuchElementException;
+import java.util.Optional;
import java.util.Properties;
-import java.util.Set;
-import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET;
import static org.apache.flink.runtime.util.EnvironmentInformation.UNKNOWN;
import static org.apache.flink.runtime.util.EnvironmentInformation.UNKNOWN_COMMIT_ID_ABBREV;
import static org.apache.flink.runtime.util.EnvironmentInformation.getJvmStartupOptionsArray;
@@ -48,11 +46,12 @@ public class EnvUtils {
private static final Logger LOG = LoggerFactory.getLogger(EnvUtils.class);
+ public static final String ENV_KUBERNETES_SERVICE_HOST = "KUBERNETES_SERVICE_HOST";
public static final String ENV_WEBHOOK_KEYSTORE_FILE = "WEBHOOK_KEYSTORE_FILE";
public static final String ENV_WEBHOOK_KEYSTORE_PASSWORD = "WEBHOOK_KEYSTORE_PASSWORD";
public static final String ENV_WEBHOOK_KEYSTORE_TYPE = "WEBHOOK_KEYSTORE_TYPE";
public static final String ENV_WEBHOOK_SERVER_PORT = "WEBHOOK_SERVER_PORT";
- public static final String ENV_WATCHED_NAMESPACES = "FLINK_OPERATOR_WATCH_NAMESPACES";
+ public static final String ENV_CONF_OVERRIDE_DIR = "CONF_OVERRIDE_DIR";
public static final String ENV_HOSTNAME = "HOSTNAME";
public static final String ENV_OPERATOR_NAME = "OPERATOR_NAME";
public static final String ENV_OPERATOR_NAMESPACE = "OPERATOR_NAMESPACE";
@@ -70,8 +69,8 @@ public class EnvUtils {
* @param key the target key
* @return the value value provided by environments.
*/
- public static String get(String key) {
- return System.getenv().get(key);
+ public static Optional<String> get(String key) {
+ return Optional.ofNullable(StringUtils.getIfBlank(System.getenv().get(key), () -> null));
}
/**
@@ -92,11 +91,10 @@ public class EnvUtils {
* @return the value provided by environments.
*/
public static String getRequired(String key) {
- String value = System.getenv().get(key);
- if (StringUtils.isEmpty(value)) {
- throw new RuntimeException("Environments: " + key + " cannot be empty");
- }
- return value;
+ return get(key).orElseThrow(
+ () ->
+ new NoSuchElementException(
+ "Environments: " + key + " cannot be empty"));
}
/**
@@ -205,16 +203,4 @@ public class EnvUtils {
}
return value;
}
-
- private static final String NAMESPACES_SPLITTER_KEY = ",";
-
- public static Set<String> getWatchedNamespaces() {
- String watchedNamespaces = get(EnvUtils.ENV_WATCHED_NAMESPACES);
-
- if (StringUtils.isEmpty(watchedNamespaces)) {
- return DEFAULT_NAMESPACES_SET;
- } else {
- return new HashSet<>(Arrays.asList(watchedNamespaces.split(NAMESPACES_SPLITTER_KEY)));
- }
- }
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
index 6677151..879dfcc 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManagerTest.java
@@ -21,6 +21,7 @@ import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.DeploymentOptionsInternal;
+import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
@@ -31,11 +32,18 @@ import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.utils.Constants;
import io.fabric8.kubernetes.api.model.Pod;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
import java.time.Duration;
+import java.util.Arrays;
import java.util.Map;
+import java.util.Optional;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -163,4 +171,18 @@ public class FlinkConfigManagerTest {
.getObserveConfig(deployment)
.get(KubernetesOperatorConfigOptions.OPERATOR_RECONCILE_INTERVAL));
}
+
+ @Test
+ public void testConfigOverrides(@TempDir Path confOverrideDir) throws IOException {
+
+ assertEquals(
+ 0, FlinkConfigManager.loadGlobalConfiguration(Optional.empty()).keySet().size());
+
+ Files.write(
+ confOverrideDir.resolve(GlobalConfiguration.FLINK_CONF_FILENAME),
+ Arrays.asList("foo: 1", "bar: 2"));
+ var conf =
+ FlinkConfigManager.loadGlobalConfiguration(Optional.of(confOverrideDir.toString()));
+ Assertions.assertEquals(Map.of("foo", "1", "bar", "2"), conf.toMap());
+ }
}
diff --git a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
index 890d6f7..401eacb 100644
--- a/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
+++ b/flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/FlinkOperatorWebhook.java
@@ -39,7 +39,6 @@ import org.apache.flink.shaded.netty4.io.netty.handler.ssl.SupportedCipherSuiteF
import org.apache.flink.shaded.netty4.io.netty.handler.stream.ChunkedWriteHandler;
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
-import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -129,9 +128,9 @@ public class FlinkOperatorWebhook {
}
private static SslContext createSslContext() throws Exception {
- String keystorePath = EnvUtils.get(EnvUtils.ENV_WEBHOOK_KEYSTORE_FILE);
+ var keystorePathOpt = EnvUtils.get(EnvUtils.ENV_WEBHOOK_KEYSTORE_FILE);
- if (StringUtils.isEmpty(keystorePath)) {
+ if (keystorePathOpt.isEmpty()) {
LOG.info(
"No keystore path is defined in "
+ EnvUtils.ENV_WEBHOOK_KEYSTORE_FILE
@@ -142,7 +141,8 @@ public class FlinkOperatorWebhook {
String keystorePassword = EnvUtils.getRequired(EnvUtils.ENV_WEBHOOK_KEYSTORE_PASSWORD);
String keystoreType = EnvUtils.getRequired(EnvUtils.ENV_WEBHOOK_KEYSTORE_TYPE);
KeyStore keyStore = KeyStore.getInstance(keystoreType);
- try (InputStream keyStoreFile = Files.newInputStream(new File(keystorePath).toPath())) {
+ try (InputStream keyStoreFile =
+ Files.newInputStream(new File(keystorePathOpt.get()).toPath())) {
keyStore.load(keyStoreFile, keystorePassword.toCharArray());
}
final KeyManagerFactory kmf =
diff --git a/helm/flink-kubernetes-operator/conf/log4j-operator.properties b/helm/flink-kubernetes-operator/conf/log4j-operator.properties
index 2f6963b..f6ccff7 100644
--- a/helm/flink-kubernetes-operator/conf/log4j-operator.properties
+++ b/helm/flink-kubernetes-operator/conf/log4j-operator.properties
@@ -27,4 +27,4 @@ appender.console.layout.pattern = %style{%d}{yellow} %style{%-30c{1.}}{cyan} %hi
# Do not log config loading
logger.conf.name = org.apache.flink.configuration.GlobalConfiguration
-logger.conf.level = WARN
\ No newline at end of file
+logger.conf.level = WARN