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/04/20 09:15:06 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27023] Unify flink and operator configuration
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 805fef5 [FLINK-27023] Unify flink and operator configuration
805fef5 is described below
commit 805fef5849479b2777ad09258e42a474a32ce058
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Tue Apr 19 12:47:21 2022 +0200
[FLINK-27023] Unify flink and operator configuration
---
docs/content/docs/development/guide.md | 12 ++--
docs/content/docs/operations/configuration.md | 42 +++++++++++---
docs/content/docs/operations/helm.md | 4 +-
docs/content/docs/operations/metrics-logging.md | 37 ++++++------
examples/kustomize/values.yaml | 4 +-
.../flink/kubernetes/operator/FlinkOperator.java | 34 ++++-------
.../kubernetes/operator/config/DefaultConfig.java | 41 --------------
.../config/FlinkOperatorConfiguration.java | 23 +++++---
...s.java => KubernetesOperatorConfigOptions.java} | 18 +++---
.../operator/metrics/OperatorMetricUtils.java | 40 +++++++++++--
.../operator/reconciler/ReconciliationUtils.java | 6 +-
.../flink/kubernetes/operator/utils/EnvUtils.java | 2 -
.../kubernetes/operator/utils/FlinkUtils.java | 16 ------
.../operator/validation/DefaultValidator.java | 4 +-
.../kubernetes/operator/FlinkOperatorTest.java | 12 ++--
.../flink/kubernetes/operator/TestUtils.java | 13 ++---
.../operator/controller/RollbackTest.java | 8 ++-
.../operator/metrics/OperatorMetricUtilsTest.java | 48 ++++++++++++++++
.../sessionjob/SessionJobObserverTest.java | 15 ++---
.../{flink-default-config => }/flink-conf.yaml | 14 +++++
.../conf/flink-operator-config/flink-conf.yaml | 29 ----------
.../log4j-console.properties | 0
...log4j2.properties => log4j-operator.properties} | 0
.../templates/flink-operator.yaml | 66 +++++++---------------
helm/flink-kubernetes-operator/values.yaml | 28 ++++-----
25 files changed, 253 insertions(+), 263 deletions(-)
diff --git a/docs/content/docs/development/guide.md b/docs/content/docs/development/guide.md
index d819425..b5b579d 100644
--- a/docs/content/docs/development/guide.md
+++ b/docs/content/docs/development/guide.md
@@ -31,7 +31,7 @@ We gathered a set of best practices here to aid development.
## Local environment setup
We recommend you install [Docker Desktop](https://www.docker.com/products/docker-desktop), [minikube](https://minikube.sigs.k8s.io/docs/start/)
-and [helm](https://helm.sh/docs/intro/quickstart/) on your local machine. For the setup please refer to our
+and [helm](https://helm.sh/docs/intro/quickstart/) on your local machine. For the setup please refer to our
[quickstart]({{< ref "docs/try-flink-kubernetes-operator/quick-start" >}}#prerequisites).
### Building docker images
@@ -58,7 +58,7 @@ When you want to reset your environment to the defaults you can do the following
eval $(minikube docker-env --unset)
```
-The most useful insight when it comes to minikube that it is just a docker container on your local machine and you can
+The most useful insight when it comes to minikube that it is just a docker container on your local machine and you can
ssh to it with the following command in case you needed to hack something there (like adding a hostpath mount or modifying docker images).
```bash
@@ -101,7 +101,7 @@ helm uninstall flink-kubernetes-operator
### Generating and Upgrading the CRD
By default, the CRD is generated by the [Fabric8 CRDGenerator](https://github.com/fabric8io/kubernetes-client/blob/master/doc/CRD-generator.md), when building from source.
-When installing flink-kubernetes-operator for the first time, the CRD will be applied to the kubernetes cluster automatically. But it will not be removed or upgraded when re-installing the flink-kubernetes-operator, as described in the relevant helm [documentation](https://helm.sh/docs/chart_best_practices/custom_resource_definitions/).
+When installing flink-kubernetes-operator for the first time, the CRD will be applied to the kubernetes cluster automatically. But it will not be removed or upgraded when re-installing the flink-kubernetes-operator, as described in the relevant helm [documentation](https://helm.sh/docs/chart_best_practices/custom_resource_definitions/).
So if the CRD is changed, you have to delete the CRD resource manually, and re-install the flink-kubernetes-operator.
```bash
@@ -110,10 +110,10 @@ kubectl delete crd flinkdeployments.flink.apache.org
### Mounts
-The operator supports to specify the volume mounts. The default mounts to hostPath can be activated by the following command. You can change the default mounts in the `helm/flink-operator/values.yaml`
+The operator supports to specify the volume mounts. The default mounts to hostPath can be activated by the following command. You can change the default mounts in the `helm/flink-kubernetes-operator/values.yaml`
```bash
-helm install flink-operator helm/flink-operator --set operatorVolumeMounts.create=true --set operatorVolumes.create=true
+helm install flink-operator helm/flink-operator --set operatorVolumeMounts.create=true --set operatorVolumes.create=true
```
@@ -125,4 +125,4 @@ Workflows are custom automated processes that you can set up in your repository
Considering the cost of running the builds, the stability, and the maintainability, flink-kubernetes-operator chose GitHub Actions and build the whole CI/CD solution on it.
All the unit tests, integration tests, and the end-to-end tests will be triggered for each PR.
-Note: Please make sure the CI passed before merging.
\ No newline at end of file
+Note: Please make sure the CI passed before merging.
diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md
index 96f22b9..fb495cc 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -26,15 +26,39 @@ under the License.
# Configuration
-## Operator
+## Specifying Operator Configuration
+
+The operator allows users to specify default configuration that will be shared by the Flink operator itself and the Flink deployments.
+
+These configuration files are mounted externally via ConfigMaps. The [Configuration files](https://github.com/apache/flink-kubernetes-operator/tree/main/helm/flink-kubernetes-operator/conf) with default values are shipped in the Helm chart. It is recommended to review and adjust them if needed in the `values.yaml` file before deploying the Operator in production environments.
+
+To append to the default configuration, simply define the `flink-conf.yaml` key in the `defaultConfiguration` section of the Helm `values.yaml` file:
+
+```
+defaultConfiguration:
+ create: true
+ # Set append to false to replace configuration files
+ append: true
+ flink-conf.yaml: |+
+ # Flink Config Overrides
+ kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
+ kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE
+
+ kubernetes.operator.reconciler.reschedule.interval: 15 s
+ kubernetes.operator.observer.progress-check.interval: 5 s
+```
+
+To learn more about metrics and logging configuration please refer to the dedicated [docs page]({{< ref "docs/operations/metrics-logging" >}}).
+
+## Operator Configuration Reference
| Key | Default | Type | Description |
| ---- | ------- | ---- | ----------- |
-| operator.reconciler.reschedule.interval | 60s | Duration | The interval for the controller to reschedule the reconcile process. |
-| operator.observer.rest-ready.delay | 10s | Duration | Final delay before deployment is marked ready after port becomes accessible. |
-| operator.reconciler.max.parallelism | 5 | Integer | The maximum number of threads running the reconciliation loop. Use -1 for infinite. |
-| operator.observer.progress-check.interval | 10s | Duration | The interval for observing status for in-progress operations such as deployment and savepoints. |
-| operator.observer.savepoint.trigger.grace-period | 10s | Duration | The interval before a savepoint trigger attempt is marked as unsuccessful. |
-| operator.observer.flink.client.timeout | 10s | Duration | The timeout for the observer to wait the flink rest client to return. |
-| operator.reconciler.flink.cancel.job.timeout | 1min | Duration | The timeout for the reconciler to wait for flink to cancel job. |
-| ooperator.reconciler.flink.cluster.shutdown.timeout | 60s | Duration | The timeout for the reconciler to wait for flink to shutdown cluster. |
+| kubernetes.operator.reconciler.reschedule.interval | 60s | Duration | The interval for the controller to reschedule the reconcile process. |
+| kubernetes.operator.observer.rest-ready.delay | 10s | Duration | Final delay before deployment is marked ready after port becomes accessible. |
+| kubernetes.operator.reconciler.max.parallelism | 5 | Integer | The maximum number of threads running the reconciliation loop. Use -1 for infinite. |
+| kubernetes.operator.observer.progress-check.interval | 10s | Duration | The interval for observing status for in-progress operations such as deployment and savepoints. |
+| kubernetes.operator.observer.savepoint.trigger.grace-period | 10s | Duration | The interval before a savepoint trigger attempt is marked as unsuccessful. |
+| kubernetes.operator.observer.flink.client.timeout | 10s | Duration | The timeout for the observer to wait the flink rest client to return. |
+| kubernetes.operator.reconciler.flink.cancel.job.timeout | 1min | Duration | The timeout for the reconciler to wait for flink to cancel job. |
+| kubernetes.operator.reconciler.flink.cluster.shutdown.timeout | 60s | Duration | The timeout for the reconciler to wait for flink to shutdown cluster. |
diff --git a/docs/content/docs/operations/helm.md b/docs/content/docs/operations/helm.md
index c0901e0..f2602e3 100644
--- a/docs/content/docs/operations/helm.md
+++ b/docs/content/docs/operations/helm.md
@@ -121,9 +121,9 @@ The example demonstrates that we can still use a `values.yaml` file to override
```yaml
-operatorConfiguration:
+defaultConfiguration:
...
- log4j2.properties: |+
+ log4j-operator.properties: |+
rootLogger.appenderRef.file.ref = LogFile
appender.file.name = LogFile
appender.file.type = File
diff --git a/docs/content/docs/operations/metrics-logging.md b/docs/content/docs/operations/metrics-logging.md
index 3fb31bd..b341bba 100644
--- a/docs/content/docs/operations/metrics-logging.md
+++ b/docs/content/docs/operations/metrics-logging.md
@@ -28,27 +28,31 @@ under the License.
The Flink Kubernetes Operator (Operator) extends the [Flink Metric System](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/metrics/) that allows gathering and exposing metrics to centralized monitoring solutions. The well known [Metric Reporters](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/metric_reporters) are shipped in the operator image and are ready to use.
+In order to specify metrics configuration for the operator, simply prefix them with `kubernetes.operator.`. This logic ensures that we can easily separate Flink job and operator metrics configuration.
+
+Let's look at a few examples.
+
## Slf4j
The default metrics reporter in the operator is Slf4j. It does not require any external monitoring systems, and it is enabled in the `values.yaml` file by default, mainly for demonstrating purposes.
```yaml
-operatorConfiguration:
+defaultConfiguration:
create: true
append: true
flink-conf.yaml: |+
- metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
- metrics.reporter.slf4j.interval: 5 MINUTE
+ kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
+ kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE
```
To use a more robust production grade monitoring solution the configuration needs to be changed.
## How to Enable Prometheus (Example)
The following example shows how to enable the Prometheus metric reporter:
```yaml
-operatorConfiguration:
+defaultConfiguration:
create: true
append: true
flink-conf.yaml: |+
- metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
- metrics.reporter.prom.port: 9999
+ kubernetes.operator.metrics.reporter.prom.class: org.apache.flink.metrics.prometheus.PrometheusReporter
+ kubernetes.operator.metrics.reporter.prom.port: 9999
```
Some metric reporters, including the Prometheus, needs a port to be exposed on the container. This can be achieved be defining a value for the otherwise empty `metrics.port` variable.
Either in the `values.yaml` file:
@@ -95,13 +99,18 @@ Once the custom resource is created in the Kubernetes environment the operator m
# Logging
The Operator controls the logging behaviour for Flink applications and the Operator itself using configuration files mounted externally via ConfigMaps. [Configuration files](https://github.com/apache/flink-kubernetes-operator/tree/main/helm/flink-kubernetes-operator/conf) with default values are shipped in the Helm chart. It is recommended to review and adjust them if needed in the `values.yaml` file before deploying the Operator in production environments.
-To append/override the default log configuration properties for the Operator use:
+To append/override the default log configuration properties for the operator and Flink deployments define the `log4j-operator.properties` and `log4j-console.properties` keys respectively:
+
```yaml
-operatorConfiguration:
+defaultConfiguration:
create: true
append: true
- log4j2.properties: |+
- rootLogger.level = DEBUG
+ log4j-operator.properties: |+
+ # Flink Operator Logging Overrides
+ # rootLogger.level = DEBUG
+ log4j-console.properties: |+
+ # Flink Deployment Logging Overrides
+ # rootLogger.level = DEBUG
```
{{< hint info >}}
@@ -111,12 +120,4 @@ We rely on the MDC provided by the operator-sdk to access this information and u
See the [Java Operator SDK docs](https://javaoperatorsdk.io/docs/features#contextual-info-for-logging-with-mdc) for more detail.
{{< /hint >}}
-To append/override the default log configuration properties for Flink applications use:
-```yaml
-flinkDefaultConfiguration:
- create: true
- append: true
- log4j-console.properties: |+
- rootLogger.level = DEBUG
-```
To learn more about accessing the job logs or changing the log level dynamically check the corresponding [section](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/resource-providers/native_kubernetes/#logging) of the core documentation.
diff --git a/examples/kustomize/values.yaml b/examples/kustomize/values.yaml
index 9c72532..ca83dc8 100644
--- a/examples/kustomize/values.yaml
+++ b/examples/kustomize/values.yaml
@@ -16,8 +16,8 @@
# limitations under the License.
################################################################################
-operatorConfiguration:
- log4j2.properties: |+
+defaultConfiguration:
+ log4j-operator.properties: |+
rootLogger.appenderRef.file.ref = LogFile
appender.file.name = LogFile
appender.file.type = File
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
index be11809..8724918 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/FlinkOperator.java
@@ -18,7 +18,8 @@
package org.apache.flink.kubernetes.operator;
import org.apache.flink.annotation.VisibleForTesting;
-import org.apache.flink.kubernetes.operator.config.DefaultConfig;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
import org.apache.flink.kubernetes.operator.controller.FlinkControllerConfig;
import org.apache.flink.kubernetes.operator.controller.FlinkDeploymentController;
@@ -34,7 +35,6 @@ import org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFact
import org.apache.flink.kubernetes.operator.reconciler.sessionjob.FlinkSessionJobReconciler;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.EnvUtils;
-import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
@@ -60,36 +60,30 @@ public class FlinkOperator {
private final KubernetesClient client;
private final FlinkService flinkService;
private final ConfigurationService configurationService;
- private final DefaultConfig defaultConfig;
+ private final Configuration defaultConfig;
private final Set<FlinkResourceValidator> validators;
public FlinkOperator() {
- this(FlinkUtils.loadDefaultConfig());
+ this(GlobalConfiguration.loadConfiguration());
}
- public FlinkOperator(DefaultConfig defaultConfig) {
- OperatorMetricUtils.initOperatorMetrics(defaultConfig.getOperatorConfig());
+ public FlinkOperator(Configuration defaultConfig) {
+ OperatorMetricUtils.initOperatorMetrics(defaultConfig);
this.defaultConfig = defaultConfig;
this.client = new DefaultKubernetesClient();
- this.operatorConfiguration =
- FlinkOperatorConfiguration.fromConfiguration(defaultConfig.getOperatorConfig());
+ this.operatorConfiguration = FlinkOperatorConfiguration.fromConfiguration(defaultConfig);
this.configurationService = getConfigurationService(operatorConfiguration);
this.operator = new Operator(client, configurationService);
this.flinkService = new FlinkService(client, operatorConfiguration);
- this.validators = ValidatorUtils.discoverValidators(defaultConfig.getFlinkConfig());
+ this.validators = ValidatorUtils.discoverValidators(defaultConfig);
}
private void registerDeploymentController() {
ReconcilerFactory reconcilerFactory =
- new ReconcilerFactory(
- client,
- flinkService,
- operatorConfiguration,
- defaultConfig.getFlinkConfig());
+ new ReconcilerFactory(client, flinkService, operatorConfiguration, defaultConfig);
ObserverFactory observerFactory =
- new ObserverFactory(
- flinkService, operatorConfiguration, defaultConfig.getFlinkConfig());
+ new ObserverFactory(flinkService, operatorConfiguration, defaultConfig);
FlinkDeploymentController controller =
new FlinkDeploymentController(
@@ -110,13 +104,9 @@ public class FlinkOperator {
private void registerSessionJobController() {
Reconciler<FlinkSessionJob> reconciler =
new FlinkSessionJobReconciler(
- client,
- flinkService,
- operatorConfiguration,
- defaultConfig.getFlinkConfig());
+ client, flinkService, operatorConfiguration, defaultConfig);
Observer<FlinkSessionJob> observer =
- new SessionJobObserver(
- operatorConfiguration, flinkService, defaultConfig.getFlinkConfig());
+ new SessionJobObserver(operatorConfiguration, flinkService, defaultConfig);
FlinkSessionJobController controller =
new FlinkSessionJobController(
operatorConfiguration, client, validators, reconciler, observer);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/DefaultConfig.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/DefaultConfig.java
deleted file mode 100644
index 36e6eb1..0000000
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/DefaultConfig.java
+++ /dev/null
@@ -1,41 +0,0 @@
-/*
- * 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.flink.kubernetes.operator.config;
-
-import org.apache.flink.configuration.Configuration;
-
-/** The container for the operator component default config. */
-public class DefaultConfig {
-
- private final Configuration operatorConfig;
- private final Configuration flinkConfig;
-
- public DefaultConfig(Configuration operatorConfig, Configuration flinkConfig) {
- this.operatorConfig = operatorConfig;
- this.flinkConfig = flinkConfig;
- }
-
- public Configuration getOperatorConfig() {
- return operatorConfig;
- }
-
- public Configuration getFlinkConfig() {
- return flinkConfig;
- }
-}
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 63ea937..fece307 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
@@ -50,32 +50,39 @@ public class FlinkOperatorConfiguration {
public static FlinkOperatorConfiguration fromConfiguration(
Configuration operatorConfig, Set<String> watchedNamespaces) {
Duration reconcileInterval =
- operatorConfig.get(OperatorConfigOptions.OPERATOR_RECONCILER_RESCHEDULE_INTERVAL);
+ operatorConfig.get(
+ KubernetesOperatorConfigOptions.OPERATOR_RECONCILER_RESCHEDULE_INTERVAL);
int reconcilerMaxParallelism =
operatorConfig.getInteger(
- OperatorConfigOptions.OPERATOR_RECONCILER_MAX_PARALLELISM);
+ KubernetesOperatorConfigOptions.OPERATOR_RECONCILER_MAX_PARALLELISM);
Duration restApiReadyDelay =
- operatorConfig.get(OperatorConfigOptions.OPERATOR_OBSERVER_REST_READY_DELAY);
+ operatorConfig.get(
+ KubernetesOperatorConfigOptions.OPERATOR_OBSERVER_REST_READY_DELAY);
Duration progressCheckInterval =
- operatorConfig.get(OperatorConfigOptions.OPERATOR_OBSERVER_PROGRESS_CHECK_INTERVAL);
+ operatorConfig.get(
+ KubernetesOperatorConfigOptions.OPERATOR_OBSERVER_PROGRESS_CHECK_INTERVAL);
Duration savepointTriggerGracePeriod =
operatorConfig.get(
- OperatorConfigOptions.OPERATOR_OBSERVER_SAVEPOINT_TRIGGER_GRACE_PERIOD);
+ KubernetesOperatorConfigOptions
+ .OPERATOR_OBSERVER_SAVEPOINT_TRIGGER_GRACE_PERIOD);
Duration flinkClientTimeout =
- operatorConfig.get(OperatorConfigOptions.OPERATOR_OBSERVER_FLINK_CLIENT_TIMEOUT);
+ operatorConfig.get(
+ KubernetesOperatorConfigOptions.OPERATOR_OBSERVER_FLINK_CLIENT_TIMEOUT);
Duration flinkCancelJobTimeout =
operatorConfig.get(
- OperatorConfigOptions.OPERATOR_RECONCILER_FLINK_CANCEL_JOB_TIMEOUT);
+ KubernetesOperatorConfigOptions
+ .OPERATOR_RECONCILER_FLINK_CANCEL_JOB_TIMEOUT);
Duration flinkShutdownClusterTimeout =
operatorConfig.get(
- OperatorConfigOptions.OPERATOR_RECONCILER_FLINK_CLUSTER_SHUTDOWN_TIMEOUT);
+ KubernetesOperatorConfigOptions
+ .OPERATOR_RECONCILER_FLINK_CLUSTER_SHUTDOWN_TIMEOUT);
String flinkServiceHostOverride = null;
if (EnvUtils.get("KUBERNETES_SERVICE_HOST") == null) {
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
similarity index 86%
rename from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
rename to flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index 4145efc..e0b4dd5 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/OperatorConfigOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -26,59 +26,59 @@ import io.javaoperatorsdk.operator.api.config.ConfigurationService;
import java.time.Duration;
/** This class holds configuration constants used by flink operator. */
-public class OperatorConfigOptions {
+public class KubernetesOperatorConfigOptions {
public static final ConfigOption<Duration> OPERATOR_RECONCILER_RESCHEDULE_INTERVAL =
- ConfigOptions.key("operator.reconciler.reschedule.interval")
+ ConfigOptions.key("kubernetes.operator.reconciler.reschedule.interval")
.durationType()
.defaultValue(Duration.ofSeconds(60))
.withDescription(
"The interval for the controller to reschedule the reconcile process.");
public static final ConfigOption<Duration> OPERATOR_OBSERVER_REST_READY_DELAY =
- ConfigOptions.key("operator.observer.rest-ready.delay")
+ ConfigOptions.key("kubernetes.operator.observer.rest-ready.delay")
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
"Final delay before deployment is marked ready after port becomes accessible.");
public static final ConfigOption<Integer> OPERATOR_RECONCILER_MAX_PARALLELISM =
- ConfigOptions.key("operator.reconciler.max.parallelism")
+ ConfigOptions.key("kubernetes.operator.reconciler.max.parallelism")
.intType()
.defaultValue(ConfigurationService.DEFAULT_RECONCILIATION_THREADS_NUMBER)
.withDescription(
"The maximum number of threads running the reconciliation loop. Use -1 for infinite.");
public static final ConfigOption<Duration> OPERATOR_OBSERVER_PROGRESS_CHECK_INTERVAL =
- ConfigOptions.key("operator.observer.progress-check.interval")
+ ConfigOptions.key("kubernetes.operator.observer.progress-check.interval")
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
"The interval for observing status for in-progress operations such as deployment and savepoints.");
public static final ConfigOption<Duration> OPERATOR_OBSERVER_SAVEPOINT_TRIGGER_GRACE_PERIOD =
- ConfigOptions.key("operator.observer.savepoint.trigger.grace-period")
+ ConfigOptions.key("kubernetes.operator.observer.savepoint.trigger.grace-period")
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
"The interval before a savepoint trigger attempt is marked as unsuccessful.");
public static final ConfigOption<Duration> OPERATOR_OBSERVER_FLINK_CLIENT_TIMEOUT =
- ConfigOptions.key("operator.observer.flink.client.timeout")
+ ConfigOptions.key("kubernetes.operator.observer.flink.client.timeout")
.durationType()
.defaultValue(Duration.ofSeconds(10))
.withDescription(
"The timeout for the observer to wait the flink rest client to return.");
public static final ConfigOption<Duration> OPERATOR_RECONCILER_FLINK_CANCEL_JOB_TIMEOUT =
- ConfigOptions.key("operator.reconciler.flink.cancel.job.timeout")
+ ConfigOptions.key("kubernetes.operator.reconciler.flink.cancel.job.timeout")
.durationType()
.defaultValue(Duration.ofMinutes(1))
.withDescription(
"The timeout for the reconciler to wait for flink to cancel job.");
public static final ConfigOption<Duration> OPERATOR_RECONCILER_FLINK_CLUSTER_SHUTDOWN_TIMEOUT =
- ConfigOptions.key("operator.reconciler.flink.cluster.shutdown.timeout")
+ ConfigOptions.key("kubernetes.operator.reconciler.flink.cluster.shutdown.timeout")
.durationType()
.defaultValue(Duration.ofSeconds(60))
.withDescription(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
index a63db24..467776d 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtils.java
@@ -17,6 +17,7 @@
package org.apache.flink.kubernetes.operator.metrics;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.plugin.PluginManager;
import org.apache.flink.core.plugin.PluginUtils;
@@ -28,16 +29,29 @@ import org.apache.flink.runtime.metrics.MetricRegistryImpl;
import org.apache.flink.runtime.metrics.ReporterSetup;
import org.apache.flink.runtime.metrics.util.MetricUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
/** Utility class for flink based operator metrics. */
public class OperatorMetricUtils {
- public static void initOperatorMetrics(Configuration operatorConfig) {
- PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(operatorConfig);
- MetricRegistry metricRegistry = createMetricRegistry(operatorConfig, pluginManager);
+ private static final Logger LOG = LoggerFactory.getLogger(OperatorMetricUtils.class);
+
+ private static final String OPERATOR_METRICS_PREFIX = "kubernetes.operator.metrics.";
+ private static final String METRICS_PREFIX = "metrics.";
+
+ public static void initOperatorMetrics(Configuration defaultConfig) {
+ Configuration metricConfig = createMetricConfig(defaultConfig);
+ LOG.info("Initializing operator metrics using conf: {}", metricConfig);
+ PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(metricConfig);
+ MetricRegistry metricRegistry = createMetricRegistry(metricConfig, pluginManager);
KubernetesOperatorMetricGroup operatorMetricGroup =
KubernetesOperatorMetricGroup.create(
metricRegistry,
- operatorConfig,
+ metricConfig,
EnvUtils.getOrDefault(EnvUtils.ENV_OPERATOR_NAMESPACE, "default"),
EnvUtils.getOrDefault(
EnvUtils.ENV_OPERATOR_NAME, "flink-kubernetes-operator"),
@@ -46,6 +60,24 @@ public class OperatorMetricUtils {
MetricUtils.instantiateStatusMetrics(statusGroup);
}
+ @VisibleForTesting
+ protected static Configuration createMetricConfig(Configuration defaultConfig) {
+ Map<String, String> metricConf = new HashMap<>();
+ defaultConfig
+ .toMap()
+ .forEach(
+ (key, value) -> {
+ if (key.startsWith(OPERATOR_METRICS_PREFIX)) {
+ metricConf.put(
+ key.replaceFirst(OPERATOR_METRICS_PREFIX, METRICS_PREFIX),
+ value);
+ } else if (!key.startsWith(METRICS_PREFIX)) {
+ metricConf.put(key, value);
+ }
+ });
+ return Configuration.fromMap(metricConf);
+ }
+
private static MetricRegistryImpl createMetricRegistry(
Configuration configuration, PluginManager pluginManager) {
return new MetricRegistryImpl(
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
index abf453c..8f604c1 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/ReconciliationUtils.java
@@ -19,7 +19,7 @@ package org.apache.flink.kubernetes.operator.reconciler;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
-import org.apache.flink.kubernetes.operator.config.OperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.CrdConstants;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
@@ -262,7 +262,7 @@ public class ReconciliationUtils {
return true;
}
- if (!configuration.get(OperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED)
+ if (!configuration.get(KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED)
|| reconciliationStatus.getState() == ReconciliationState.ROLLED_BACK
|| reconciliationStatus.isLastReconciledSpecStable()) {
return false;
@@ -277,7 +277,7 @@ public class ReconciliationUtils {
}
Duration readinessTimeout =
- configuration.get(OperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT);
+ configuration.get(KubernetesOperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT);
return Instant.now()
.minus(readinessTimeout)
.isAfter(Instant.ofEpochMilli(reconciliationStatus.getReconciliationTimestamp()));
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 a000582..b82bb43 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
@@ -44,8 +44,6 @@ public class EnvUtils {
private static final Logger LOG = LoggerFactory.getLogger(EnvUtils.class);
- public static final String ENV_FLINK_OPERATOR_CONF_DIR = "FLINK_OPERATOR_CONF_DIR";
- public static final String ENV_FLINK_CONF_DIR = "FLINK_CONF_DIR";
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";
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
index 1b9edb9..dc4ba50 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkUtils.java
@@ -18,12 +18,10 @@
package org.apache.flink.kubernetes.operator.utils;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory;
import org.apache.flink.kubernetes.kubeclient.decorators.ExternalServiceDecorator;
-import org.apache.flink.kubernetes.operator.config.DefaultConfig;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
import org.apache.flink.kubernetes.utils.Constants;
@@ -53,14 +51,6 @@ public class FlinkUtils {
private static final Logger LOG = LoggerFactory.getLogger(FlinkUtils.class);
private static final ObjectMapper MAPPER = new ObjectMapper();
- public static DefaultConfig loadDefaultConfig() {
- Configuration operatorConfig =
- FlinkUtils.loadConfiguration(EnvUtils.get(EnvUtils.ENV_FLINK_OPERATOR_CONF_DIR));
- Configuration flinkConfig =
- FlinkUtils.loadConfiguration(EnvUtils.get(EnvUtils.ENV_FLINK_CONF_DIR));
- return new DefaultConfig(operatorConfig, flinkConfig);
- }
-
public static Configuration getEffectiveConfig(
FlinkDeployment flinkApp, Configuration flinkConfig) {
return getEffectiveConfig(flinkApp.getMetadata(), flinkApp.getSpec(), flinkConfig);
@@ -78,12 +68,6 @@ public class FlinkUtils {
}
}
- public static Configuration loadConfiguration(String confDir) {
- return confDir != null
- ? GlobalConfiguration.loadConfiguration(confDir)
- : new Configuration();
- }
-
public static Pod mergePodTemplates(Pod toPod, Pod fromPod) {
if (fromPod == null) {
return toPod;
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
index e7fd623..2dfe5ff 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/validation/DefaultValidator.java
@@ -19,6 +19,7 @@ package org.apache.flink.kubernetes.operator.validation;
import org.apache.flink.configuration.CheckpointingOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.GlobalConfiguration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
@@ -35,7 +36,6 @@ import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
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.kubernetes.operator.utils.IngressUtils;
import org.apache.flink.kubernetes.utils.Constants;
@@ -59,7 +59,7 @@ public class DefaultValidator implements FlinkResourceValidator {
private final Configuration defaultFlinkConf;
public DefaultValidator() {
- this(FlinkUtils.loadConfiguration(EnvUtils.get(EnvUtils.ENV_FLINK_CONF_DIR)));
+ this(GlobalConfiguration.loadConfiguration());
}
public DefaultValidator(Configuration defaultFlinkConf) {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
index 281a357..7e5147a 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/FlinkOperatorTest.java
@@ -18,7 +18,7 @@
package org.apache.flink.kubernetes.operator;
import org.apache.flink.configuration.Configuration;
-import org.apache.flink.kubernetes.operator.config.DefaultConfig;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import io.javaoperatorsdk.operator.api.config.ConfigurationService;
import org.junit.jupiter.api.Assertions;
@@ -62,12 +62,12 @@ public class FlinkOperatorTest {
private ExecutorService getExecutorForParallelismConfig(Optional<Integer> parallelism) {
var operatorConfig = new Configuration();
parallelism.ifPresent(
- p -> operatorConfig.setInteger("operator.reconciler.max.parallelism", p));
+ p ->
+ operatorConfig.setInteger(
+ KubernetesOperatorConfigOptions.OPERATOR_RECONCILER_MAX_PARALLELISM,
+ p));
- var flinkConfig = new Configuration();
- var config = new DefaultConfig(operatorConfig, flinkConfig);
-
- FlinkOperator flinkOperator = new FlinkOperator(config);
+ FlinkOperator flinkOperator = new FlinkOperator(operatorConfig);
return flinkOperator.getOperator().getConfigurationService().getExecutorService();
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
index b3734dd..044d063 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestUtils.java
@@ -17,6 +17,7 @@
package org.apache.flink.kubernetes.operator;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.HighAvailabilityOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory;
@@ -40,7 +41,6 @@ import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatu
import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
import org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
-import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
import io.fabric8.kubernetes.api.model.Container;
@@ -344,22 +344,19 @@ public class TestUtils {
FlinkOperatorConfiguration operatorConfiguration,
KubernetesClient kubernetesClient,
TestingFlinkService flinkService) {
- var defaultConfig = FlinkUtils.loadDefaultConfig();
+ var defaultConfig = new Configuration();
var controller =
new FlinkDeploymentController(
operatorConfiguration,
kubernetesClient,
- ValidatorUtils.discoverValidators(defaultConfig.getFlinkConfig()),
+ ValidatorUtils.discoverValidators(defaultConfig),
new ReconcilerFactory(
kubernetesClient,
flinkService,
operatorConfiguration,
- defaultConfig.getFlinkConfig()),
- new ObserverFactory(
- flinkService,
- operatorConfiguration,
- defaultConfig.getFlinkConfig()));
+ defaultConfig),
+ new ObserverFactory(flinkService, operatorConfiguration, defaultConfig));
controller.setControllerConfig(
new FlinkControllerConfig(controller, Collections.emptySet()));
return controller;
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
index cd15a50..2a5e320 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/RollbackTest.java
@@ -22,7 +22,7 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
-import org.apache.flink.kubernetes.operator.config.OperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.spec.JobState;
import org.apache.flink.kubernetes.operator.crd.spec.UpgradeMode;
@@ -225,8 +225,10 @@ public class RollbackTest {
throws Exception {
var flinkConfiguration = deployment.getSpec().getFlinkConfiguration();
- flinkConfiguration.put(OperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED.key(), "true");
- flinkConfiguration.put(OperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT.key(), "100");
+ flinkConfiguration.put(
+ KubernetesOperatorConfigOptions.DEPLOYMENT_ROLLBACK_ENABLED.key(), "true");
+ flinkConfiguration.put(
+ KubernetesOperatorConfigOptions.DEPLOYMENT_READINESS_TIMEOUT.key(), "100");
testController.reconcile(deployment, TestUtils.createEmptyContext());
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtilsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtilsTest.java
new file mode 100644
index 0000000..15a6148
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorMetricUtilsTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.flink.kubernetes.operator.metrics;
+
+import org.apache.flink.configuration.Configuration;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** {@link OperatorMetricUtils} tests. */
+public class OperatorMetricUtilsTest {
+
+ @Test
+ public void testMetricConfig() {
+ Configuration configuration =
+ Configuration.fromMap(
+ Map.of(
+ "k",
+ "v",
+ "metrics.flink.metric.conf",
+ "fmc",
+ "kubernetes.operator.metrics.op.metric.conf",
+ "omc"));
+
+ Configuration expectedMetricConf =
+ Configuration.fromMap(Map.of("k", "v", "metrics.op.metric.conf", "omc"));
+
+ assertEquals(expectedMetricConf, OperatorMetricUtils.createMetricConfig(configuration));
+ }
+}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
index a9b9da2..1bcb8bb 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/observer/sessionjob/SessionJobObserverTest.java
@@ -23,7 +23,6 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
-import org.apache.flink.kubernetes.operator.config.DefaultConfig;
import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
import org.apache.flink.kubernetes.operator.reconciler.sessionjob.FlinkSessionJobReconciler;
@@ -39,8 +38,7 @@ public class SessionJobObserverTest {
private final FlinkOperatorConfiguration operatorConfiguration =
FlinkOperatorConfiguration.fromConfiguration(new Configuration());
- private final DefaultConfig defaultConfig =
- new DefaultConfig(new Configuration(), new Configuration());
+ private final Configuration defaultConfig = new Configuration();
@Test
public void testBasicObserve() throws Exception {
@@ -50,8 +48,7 @@ public class SessionJobObserverTest {
new FlinkSessionJobReconciler(
null, flinkService, operatorConfiguration, new Configuration());
final var observer =
- new SessionJobObserver(
- operatorConfiguration, flinkService, defaultConfig.getFlinkConfig());
+ new SessionJobObserver(operatorConfiguration, flinkService, defaultConfig);
final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
// observe the brand new job, nothing to do.
@@ -114,8 +111,7 @@ public class SessionJobObserverTest {
new FlinkSessionJobReconciler(
null, flinkService, operatorConfiguration, new Configuration());
final var observer =
- new SessionJobObserver(
- operatorConfiguration, flinkService, defaultConfig.getFlinkConfig());
+ new SessionJobObserver(operatorConfiguration, flinkService, defaultConfig);
final var readyContext =
TestUtils.createContextWithReadyFlinkDeployment(
Map.of(RestOptions.PORT.key(), "8088"));
@@ -143,8 +139,7 @@ public class SessionJobObserverTest {
new FlinkSessionJobReconciler(
null, flinkService, operatorConfiguration, new Configuration());
final var observer =
- new SessionJobObserver(
- operatorConfiguration, flinkService, defaultConfig.getFlinkConfig());
+ new SessionJobObserver(operatorConfiguration, flinkService, defaultConfig);
final var readyContext = TestUtils.createContextWithReadyFlinkDeployment();
// submit job
@@ -162,7 +157,7 @@ public class SessionJobObserverTest {
Assertions.assertNull(savepointInfo.getTriggerId());
Assertions.assertNull(savepointInfo.getTriggerTimestamp());
- flinkService.triggerSavepoint(jobID, savepointInfo, defaultConfig.getFlinkConfig());
+ flinkService.triggerSavepoint(jobID, savepointInfo, defaultConfig);
Assertions.assertEquals("trigger_0", savepointInfo.getTriggerId());
observer.observe(sessionJob, readyContext);
diff --git a/helm/flink-kubernetes-operator/conf/flink-default-config/flink-conf.yaml b/helm/flink-kubernetes-operator/conf/flink-conf.yaml
similarity index 62%
rename from helm/flink-kubernetes-operator/conf/flink-default-config/flink-conf.yaml
rename to helm/flink-kubernetes-operator/conf/flink-conf.yaml
index 485f99a..fcbea15 100644
--- a/helm/flink-kubernetes-operator/conf/flink-default-config/flink-conf.yaml
+++ b/helm/flink-kubernetes-operator/conf/flink-conf.yaml
@@ -16,6 +16,7 @@
# limitations under the License.
################################################################################
+# Flink job/cluster related configs
taskmanager.numberOfTaskSlots: 2
blob.server.port: 6124
jobmanager.rpc.port: 6123
@@ -24,3 +25,16 @@ queryable-state.proxy.ports: 6125
jobmanager.memory.process.size: 1600m
taskmanager.memory.process.size: 1728m
parallelism.default: 2
+
+# Flink operator related configs
+# kubernetes.operator.reconciler.reschedule.interval: 60 s
+# kubernetes.operator.reconciler.max.parallelism: 5
+# kubernetes.operator.reconciler.flink.cancel.job.timeout: 1 min
+# kubernetes.operator.reconciler.flink.cluster.shutdown.timeout: 60 s
+# kubernetes.operator.observer.rest-ready.delay: 10 s
+# kubernetes.operator.observer.progress-check.interval: 10 s
+# kubernetes.operator.observer.savepoint.trigger.grace-period: 10 s
+# kubernetes.operator.observer.flink.client.timeout: 10 s
+
+# kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
+# kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE
diff --git a/helm/flink-kubernetes-operator/conf/flink-operator-config/flink-conf.yaml b/helm/flink-kubernetes-operator/conf/flink-operator-config/flink-conf.yaml
deleted file mode 100644
index a7254ff..0000000
--- a/helm/flink-kubernetes-operator/conf/flink-operator-config/flink-conf.yaml
+++ /dev/null
@@ -1,29 +0,0 @@
-################################################################################
-# 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.
-################################################################################
-
-# operator.reconciler.reschedule.interval: 60 s
-# operator.reconciler.max.parallelism: 5
-# operator.reconciler.flink.cancel.job.timeout: 1min
-# operator.reconciler.flink.cluster.shutdown.timeout: 60s
-# operator.observer.rest-ready.delay: 10 s
-# operator.observer.progress-check.interval: 10 s
-# operator.observer.savepoint.trigger.grace-period: 10 s
-# operator.observer.flink.client.timeout: 10 s
-
-# metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
-# metrics.reporter.slf4j.interval: 5 MINUTE
diff --git a/helm/flink-kubernetes-operator/conf/flink-default-config/log4j-console.properties b/helm/flink-kubernetes-operator/conf/log4j-console.properties
similarity index 100%
rename from helm/flink-kubernetes-operator/conf/flink-default-config/log4j-console.properties
rename to helm/flink-kubernetes-operator/conf/log4j-console.properties
diff --git a/helm/flink-kubernetes-operator/conf/flink-operator-config/log4j2.properties b/helm/flink-kubernetes-operator/conf/log4j-operator.properties
similarity index 100%
rename from helm/flink-kubernetes-operator/conf/flink-operator-config/log4j2.properties
rename to helm/flink-kubernetes-operator/conf/log4j-operator.properties
diff --git a/helm/flink-kubernetes-operator/templates/flink-operator.yaml b/helm/flink-kubernetes-operator/templates/flink-operator.yaml
index 09a8025..8bdae31 100644
--- a/helm/flink-kubernetes-operator/templates/flink-operator.yaml
+++ b/helm/flink-kubernetes-operator/templates/flink-operator.yaml
@@ -63,18 +63,14 @@ spec:
value: /opt/flink/conf
- name: FLINK_PLUGINS_DIR
value: /opt/flink/plugins
- - name: FLINK_OPERATOR_CONF_DIR
- value: /opt/flink-operator/conf
- name: LOG_CONFIG
- value: -Dlog4j.configurationFile=/opt/flink-operator/conf/log4j2.properties
+ value: -Dlog4j.configurationFile=/opt/flink/conf/log4j-operator.properties
- name: JVM_ARGS
value: {{ .Values.jvmArgs.operator }}
- name: FLINK_OPERATOR_WATCH_NAMESPACES
value: {{ join "," .Values.watchNamespaces }}
volumeMounts:
- name: flink-operator-config-volume
- mountPath: /opt/flink-operator/conf
- - name: flink-default-config-volume
mountPath: /opt/flink/conf
{{- if .Values.operatorVolumeMounts.create }}
{{- toYaml .Values.operatorVolumeMounts.data | nindent 12 }}
@@ -103,7 +99,7 @@ spec:
- name: WEBHOOK_SERVER_PORT
value: "9443"
- name: LOG_CONFIG
- value: -Dlog4j.configurationFile=/opt/flink-operator/conf/log4j2.properties
+ value: -Dlog4j.configurationFile=/opt/flink/conf/log4j-operator.properties
- name: JVM_ARGS
value: {{ .Values.jvmArgs.webhook }}
volumeMounts:
@@ -111,7 +107,7 @@ spec:
mountPath: "/certs"
readOnly: true
- name: flink-operator-config-volume
- mountPath: /opt/flink-operator/conf
+ mountPath: /opt/flink/conf
{{- end }}
volumes:
- name: flink-operator-config-volume
@@ -120,14 +116,8 @@ spec:
items:
- key: flink-conf.yaml
path: flink-conf.yaml
- - key: log4j2.properties
- path: log4j2.properties
- - name: flink-default-config-volume
- configMap:
- name: flink-default-config
- items:
- - key: flink-conf.yaml
- path: flink-conf.yaml
+ - key: log4j-operator.properties
+ path: log4j-operator.properties
- key: log4j-console.properties
path: log4j-console.properties
{{- if .Values.operatorVolumes.create }}
@@ -142,7 +132,7 @@ spec:
path: keystore.p12
{{- end }}
---
-{{- if .Values.operatorConfiguration.create }}
+{{- if .Values.defaultConfiguration.create }}
apiVersion: v1
kind: ConfigMap
metadata:
@@ -152,42 +142,24 @@ metadata:
{{- include "flink-operator.labels" . | nindent 4 }}
data:
flink-conf.yaml: |+
-{{- if .Values.operatorConfiguration.append }}
- {{- $.Files.Get "conf/flink-operator-config/flink-conf.yaml" | nindent 4 -}}
+{{- if .Values.defaultConfiguration.append }}
+ {{- $.Files.Get "conf/flink-conf.yaml" | nindent 4 -}}
{{- end }}
-{{- if index (.Values.operatorConfiguration) "flink-conf.yaml" }}
- {{- index (.Values.operatorConfiguration) "flink-conf.yaml" | nindent 4 -}}
+{{- if index (.Values.defaultConfiguration) "flink-conf.yaml" }}
+ {{- index (.Values.defaultConfiguration) "flink-conf.yaml" | nindent 4 -}}
{{- end }}
- log4j2.properties: |+
-{{- if .Values.operatorConfiguration.append }}
- {{- $.Files.Get "conf/flink-operator-config/log4j2.properties" | nindent 4 -}}
-{{- end }}
-{{- if index (.Values.operatorConfiguration) "log4j2.properties" }}
- {{- index (.Values.operatorConfiguration) "log4j2.properties" | nindent 4 -}}
-{{- end }}
-{{- end }}
----
-{{- if .Values.flinkDefaultConfiguration.create }}
-apiVersion: v1
-kind: ConfigMap
-metadata:
- name: flink-default-config
- namespace: {{ .Release.Namespace }}
- labels:
- {{- include "flink-operator.labels" . | nindent 4 }}
-data:
- flink-conf.yaml: |+
-{{- if .Values.flinkDefaultConfiguration.append }}
- {{- $.Files.Get "conf/flink-default-config/flink-conf.yaml" | nindent 4 -}}
+ log4j-operator.properties: |+
+{{- if .Values.defaultConfiguration.append }}
+ {{- $.Files.Get "conf/log4j-operator.properties" | nindent 4 -}}
{{- end }}
-{{- if index (.Values.flinkDefaultConfiguration) "flink-conf.yaml" }}
- {{- index (.Values.flinkDefaultConfiguration) "flink-conf.yaml" | nindent 4 -}}
+{{- if index (.Values.defaultConfiguration) "log4j-operator.properties" }}
+ {{- index (.Values.defaultConfiguration) "log4j-operator.properties" | nindent 4 -}}
{{- end }}
log4j-console.properties: |+
-{{- if .Values.flinkDefaultConfiguration.append }}
- {{- $.Files.Get "conf/flink-default-config/log4j-console.properties" | nindent 4 -}}
+{{- if .Values.defaultConfiguration.append }}
+ {{- $.Files.Get "conf/log4j-console.properties" | nindent 4 -}}
{{- end }}
-{{- if index (.Values.flinkDefaultConfiguration) "log4j-console.properties" }}
- {{- index (.Values.flinkDefaultConfiguration) "log4j-console.properties" | nindent 4 -}}
+{{- if index (.Values.defaultConfiguration) "log4j-console.properties" }}
+ {{- index (.Values.defaultConfiguration) "log4j-console.properties" | nindent 4 -}}
{{- end }}
{{- end }}
diff --git a/helm/flink-kubernetes-operator/values.yaml b/helm/flink-kubernetes-operator/values.yaml
index 6b3d1a4..32e2e20 100644
--- a/helm/flink-kubernetes-operator/values.yaml
+++ b/helm/flink-kubernetes-operator/values.yaml
@@ -66,27 +66,23 @@ webhook:
# name: jks-password-secret
# key: password-key
-operatorConfiguration:
+defaultConfiguration:
create: true
# Set append to false to replace configuration files
append: true
flink-conf.yaml: |+
- metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
- metrics.reporter.slf4j.interval: 5 MINUTE
+ # Flink Config Overrides
+ kubernetes.operator.metrics.reporter.slf4j.factory.class: org.apache.flink.metrics.slf4j.Slf4jReporterFactory
+ kubernetes.operator.metrics.reporter.slf4j.interval: 5 MINUTE
- operator.reconciler.reschedule.interval: 15 s
- operator.observer.progress-check.interval: 5 s
- # log4j2.properties: |+
- # rootLogger.level = DEBUG
-
-flinkDefaultConfiguration:
- create: true
- # Set append to false to replace configuration files
- append: true
- # flink-conf.yaml: |+
- # ...
- # log4j-console.properties: |+
- # ...
+ kubernetes.operator.reconciler.reschedule.interval: 15 s
+ kubernetes.operator.observer.progress-check.interval: 5 s
+ log4j-operator.properties: |+
+ # Flink Operator Logging Overrides
+ # rootLogger.level = DEBUG
+ log4j-console.properties: |+
+ # Flink Deployment Logging Overrides
+ # rootLogger.level = DEBUG
# (Optional) Exposes metrics port on the container if defined
metrics: