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:40:46 UTC
[flink-kubernetes-operator] branch main updated (56381ec -> f93da5c)
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
from 56381ec [FLINK-28445] Support operator configuration from multiple configmaps
new 1b3acc2 [FLINK-28297] Improve operator metric groups
new 308f8ad [FLINK-27914] Integrate JOSDK metrics with Flink Metrics reporter
new f93da5c [docs] Break up documentation into sections
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/content/docs/operations/configuration.md | 30 +++--
docs/content/docs/operations/metrics-logging.md | 3 +-
.../shortcodes/generated/dynamic_section.html | 72 +++++++++++
.../kubernetes_operator_metric_configuration.html | 36 ++++++
.../generated/system_advanced_section.html | 60 +++++++++
.../shortcodes/generated/system_section.html | 78 +++++++++++
.../configuration/ConfigOptionsDocGenerator.java | 4 +-
.../flink/kubernetes/operator/FlinkOperator.java | 21 ++-
.../operator/config/FlinkConfigManager.java | 2 +-
.../config/FlinkOperatorConfiguration.java | 8 +-
.../config/KubernetesOperatorConfigOptions.java | 34 +++++
.../metrics/KubernetesOperatorMetricGroup.java | 24 ++--
.../metrics/KubernetesOperatorMetricOptions.java | 26 +++-
...oup.java => KubernetesResourceMetricGroup.java} | 47 ++-----
...=> KubernetesResourceNamespaceMetricGroup.java} | 55 ++++----
...=> KubernetesResourceNamespaceScopeFormat.java} | 23 ++--
...mat.java => KubernetesResourceScopeFormat.java} | 29 +++--
.../kubernetes/operator/metrics/MetricManager.java | 19 +--
.../operator/metrics/OperatorJosdkMetrics.java | 143 +++++++++++++++++++++
.../operator/metrics/OperatorMetricUtils.java | 2 +-
.../flink/kubernetes/operator/TestUtils.java | 28 ++++
.../kubernetes/operator/TestingStatusRecorder.java | 5 +-
.../TestingFlinkDeploymentController.java | 5 +-
.../listener/FlinkResourceListenerTest.java | 5 +-
.../metrics/KubernetesOperatorMetricGroupTest.java | 48 ++++++-
.../operator/metrics/OperatorJosdkMetricsTest.java | 119 +++++++++++++++++
.../operator/utils/StatusRecorderTest.java | 73 +++++++----
.../operator/admission/FlinkOperatorWebhook.java | 2 +-
28 files changed, 832 insertions(+), 169 deletions(-)
create mode 100644 docs/layouts/shortcodes/generated/dynamic_section.html
create mode 100644 docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html
create mode 100644 docs/layouts/shortcodes/generated/system_advanced_section.html
create mode 100644 docs/layouts/shortcodes/generated/system_section.html
copy flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/{KubernetesOperatorMetricGroup.java => KubernetesResourceMetricGroup.java} (51%)
copy flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/{KubernetesOperatorMetricGroup.java => KubernetesResourceNamespaceMetricGroup.java} (56%)
copy flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/{KubernetesOperatorScopeFormat.java => KubernetesResourceNamespaceScopeFormat.java} (63%)
copy flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/{KubernetesOperatorScopeFormat.java => KubernetesResourceScopeFormat.java} (56%)
create mode 100644 flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
create mode 100644 flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java
[flink-kubernetes-operator] 02/03: [FLINK-27914] Integrate JOSDK metrics with Flink Metrics reporter
Posted by gy...@apache.org.
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
commit 308f8add74165febaa2f80c61a2a9d242d8b0629
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Thu Jul 7 14:13:38 2022 +0200
[FLINK-27914] Integrate JOSDK metrics with Flink Metrics reporter
---
docs/content/docs/operations/metrics-logging.md | 3 +-
.../kubernetes_operator_config_configuration.html | 6 +
.../flink/kubernetes/operator/FlinkOperator.java | 9 +-
.../config/FlinkOperatorConfiguration.java | 5 +
.../config/KubernetesOperatorConfigOptions.java | 7 +
.../metrics/KubernetesOperatorMetricOptions.java | 2 +-
.../operator/metrics/OperatorJosdkMetrics.java | 143 +++++++++++++++++++++
.../operator/metrics/OperatorJosdkMetricsTest.java | 119 +++++++++++++++++
8 files changed, 289 insertions(+), 5 deletions(-)
diff --git a/docs/content/docs/operations/metrics-logging.md b/docs/content/docs/operations/metrics-logging.md
index 0e34dcc..13aa01d 100644
--- a/docs/content/docs/operations/metrics-logging.md
+++ b/docs/content/docs/operations/metrics-logging.md
@@ -28,7 +28,6 @@ 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.
-
## Deployment Metrics
The Operator gathers aggregates metrics about managed resources.
@@ -39,7 +38,7 @@ The Operator gathers aggregates metrics about managed resources.
| Namespace | FlinkSessionJob.Count | Number of managed FlinkSessionJob instances per namespace | Gauge |
## System Metrics
-The Operator gathers metrics about the JVM process and exposes it similarly to core Flink [System metrics](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/metrics/#system-metrics). The list of metrics are not repeated in this document.
+The Operator gathers metrics about the JVM process and exposes it similarly to core Flink [System metrics](https://nightlies.apache.org/flink/flink-docs-master/docs/ops/metrics/#system-metrics). The list of metrics are not repeated in this document.
## Metric Reporters
diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index cfc1ba9..5c34d95 100644
--- a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++ b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -80,6 +80,12 @@
<td>Boolean</td>
<td>Enables last-state fallback for savepoint upgrade mode. When the job is not running thus savepoint cannot be triggered but HA metadata is available for last state restore the operator can initiate the upgrade process when the flag is enabled.</td>
</tr>
+ <tr>
+ <td><h5>kubernetes.operator.josdk.metrics.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Enable forwarding of Java Operator SDK metrics to the Flink metric registry.</td>
+ </tr>
<tr>
<td><h5>kubernetes.operator.observer.progress-check.interval</h5></td>
<td style="word-wrap: break-word;">10 s</td>
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 419702d..13bbe89 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
@@ -30,6 +30,7 @@ import org.apache.flink.kubernetes.operator.listener.FlinkResourceListener;
import org.apache.flink.kubernetes.operator.listener.ListenerUtils;
import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
+import org.apache.flink.kubernetes.operator.metrics.OperatorJosdkMetrics;
import org.apache.flink.kubernetes.operator.metrics.OperatorMetricUtils;
import org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
import org.apache.flink.kubernetes.operator.observer.sessionjob.SessionJobObserver;
@@ -79,12 +80,12 @@ public class FlinkOperator {
conf != null
? new FlinkConfigManager(conf) // For testing only
: new FlinkConfigManager(this::handleNamespaceChanges);
+ this.metricGroup =
+ OperatorMetricUtils.initOperatorMetrics(configManager.getDefaultConfig());
this.operator = new Operator(client, this::overrideOperatorConfigs);
this.flinkService = new FlinkService(client, configManager);
this.validators = ValidatorUtils.discoverValidators(configManager);
this.listeners = ListenerUtils.discoverListeners(configManager);
- this.metricGroup =
- OperatorMetricUtils.initOperatorMetrics(configManager.getDefaultConfig());
PluginManager pluginManager =
PluginUtils.createPluginManagerFromRootFolder(configManager.getDefaultConfig());
FileSystem.initialize(configManager.getDefaultConfig(), pluginManager);
@@ -109,6 +110,10 @@ public class FlinkOperator {
LOG.info("Configuring operator with {} reconciliation threads.", parallelism);
overrider.withConcurrentReconciliationThreads(parallelism);
}
+ if (configManager.getOperatorConfiguration().getJosdkMetricsEnabled()) {
+ overrider.withMetrics(
+ new OperatorJosdkMetrics(metricGroup, configManager.getDefaultConfig()));
+ }
}
private void registerDeploymentController() {
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 52fbb85..d295501 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
@@ -43,6 +43,7 @@ public class FlinkOperatorConfiguration {
String flinkServiceHostOverride;
Set<String> watchedNamespaces;
Boolean dynamicNamespacesEnabled;
+ Boolean josdkMetricsEnabled;
Duration flinkCancelJobTimeout;
Duration flinkShutdownClusterTimeout;
String artifactsBaseDir;
@@ -108,6 +109,9 @@ public class FlinkOperatorConfiguration {
operatorConfig.get(
KubernetesOperatorConfigOptions.OPERATOR_DYNAMIC_NAMESPACES_ENABLED);
+ boolean josdkMetricsEnabled =
+ operatorConfig.get(KubernetesOperatorConfigOptions.OPERATOR_JOSDK_METRICS_ENABLED);
+
RetryConfiguration retryConfiguration = new FlinkOperatorRetryConfiguration(operatorConfig);
return new FlinkOperatorConfiguration(
@@ -119,6 +123,7 @@ public class FlinkOperatorConfiguration {
flinkServiceHostOverride,
watchedNamespaces,
dynamicNamespacesEnabled,
+ josdkMetricsEnabled,
flinkCancelJobTimeout,
flinkShutdownClusterTimeout,
artifactsBaseDir,
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index aa6031c..b01e845 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -210,6 +210,13 @@ public class KubernetesOperatorConfigOptions {
.defaultValue(false)
.withDescription("Enables dynamic change of watched/monitored namespaces.");
+ public static final ConfigOption<Boolean> OPERATOR_JOSDK_METRICS_ENABLED =
+ ConfigOptions.key("kubernetes.operator.josdk.metrics.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Enable forwarding of Java Operator SDK metrics to the Flink metric registry.");
+
public static final ConfigOption<Duration> OPERATOR_RETRY_INITIAL_INTERVAL =
ConfigOptions.key("kubernetes.operator.retry.initial.interval")
.durationType()
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
index 30b02f9..20e0f94 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
@@ -31,7 +31,7 @@ public class KubernetesOperatorMetricOptions {
public static final ConfigOption<String> SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCENS =
ConfigOptions.key("metrics.scope.k8soperator.resourcens")
- .defaultValue("<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>.")
+ .defaultValue("<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>")
.withDescription(
"Defines the scope format string that is applied to all metrics scoped to the kubernetes operator resource namespace.");
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
new file mode 100644
index 0000000..c35ff53
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetrics.java
@@ -0,0 +1,143 @@
+/*
+ * 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.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+
+import io.javaoperatorsdk.operator.api.monitoring.Metrics;
+import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import io.javaoperatorsdk.operator.processing.event.Event;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Implementation of {@link Metrics} to monitor and forward JOSDK metrics to {@link MetricRegistry}.
+ */
+public class OperatorJosdkMetrics implements Metrics {
+
+ private static final String OPERATOR_SDK_GROUP = "JOSDK";
+ private static final String RECONCILIATION = "Reconciliation";
+ private static final String RESOURCE = "Resource";
+ private static final String EVENT = "Event";
+
+ private final KubernetesOperatorMetricGroup operatorMetricGroup;
+ private final Configuration conf;
+
+ private final Map<ResourceID, KubernetesResourceNamespaceMetricGroup> resourceNsMetricGroups =
+ new ConcurrentHashMap<>();
+ private final Map<ResourceID, KubernetesResourceMetricGroup> resourceMetricGroups =
+ new ConcurrentHashMap<>();
+
+ private final Map<String, Counter> counters = new ConcurrentHashMap<>();
+
+ public OperatorJosdkMetrics(
+ KubernetesOperatorMetricGroup operatorMetricGroup, Configuration conf) {
+ this.operatorMetricGroup = operatorMetricGroup;
+ this.conf = conf;
+ }
+
+ @Override
+ public void receivedEvent(Event event) {
+ if (event instanceof ResourceEvent) {
+ var action = ((ResourceEvent) event).getAction();
+ counter(
+ getResourceMg(event.getRelatedCustomResourceID()),
+ Collections.emptyList(),
+ RESOURCE,
+ EVENT)
+ .inc();
+ counter(
+ getResourceMg(event.getRelatedCustomResourceID()),
+ Collections.emptyList(),
+ RESOURCE,
+ EVENT,
+ action.name())
+ .inc();
+ }
+ }
+
+ @Override
+ public void cleanupDoneFor(ResourceID resourceID) {
+ counter(getResourceMg(resourceID), Collections.emptyList(), RECONCILIATION, "cleanup")
+ .inc();
+ }
+
+ @Override
+ public void reconcileCustomResource(ResourceID resourceID, RetryInfo retryInfoNullable) {
+ counter(getResourceMg(resourceID), Collections.emptyList(), RECONCILIATION).inc();
+
+ if (retryInfoNullable != null) {
+ counter(getResourceMg(resourceID), Collections.emptyList(), RECONCILIATION, "retries")
+ .inc();
+ }
+ }
+
+ @Override
+ public void finishedReconciliation(ResourceID resourceID) {
+ counter(getResourceMg(resourceID), Collections.emptyList(), RECONCILIATION, "finished")
+ .inc();
+ }
+
+ @Override
+ public void failedReconciliation(ResourceID resourceID, Exception exception) {
+ counter(getResourceMg(resourceID), Collections.emptyList(), RECONCILIATION, "failed").inc();
+ }
+
+ @Override
+ public <T extends Map<?, ?>> T monitorSizeOf(T map, String name) {
+ operatorMetricGroup.addGroup(name).gauge("size", map::size);
+ return map;
+ }
+
+ private Counter counter(
+ MetricGroup parent, List<Tuple2<String, String>> additionalTags, String... names) {
+ MetricGroup group = parent.addGroup(OPERATOR_SDK_GROUP);
+ for (String name : names) {
+ group = group.addGroup(name);
+ }
+ for (Tuple2<String, String> tag : additionalTags) {
+ group = group.addGroup(tag.f0, tag.f1);
+ }
+ var finalGroup = group;
+ return counters.computeIfAbsent(
+ String.join(".", group.getScopeComponents()), s -> finalGroup.counter("Count"));
+ }
+
+ private KubernetesResourceNamespaceMetricGroup getResourceNsMg(ResourceID resourceID) {
+ return resourceNsMetricGroups.computeIfAbsent(
+ resourceID,
+ rid ->
+ operatorMetricGroup.createResourceNamespaceGroup(
+ conf, rid.getNamespace().orElse("default")));
+ }
+
+ private KubernetesResourceMetricGroup getResourceMg(ResourceID resourceID) {
+ return resourceMetricGroups.computeIfAbsent(
+ resourceID,
+ rid -> getResourceNsMg(rid).createResourceNamespaceGroup(conf, rid.getName()));
+ }
+}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java
new file mode 100644
index 0000000..0a719d7
--- /dev/null
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/OperatorJosdkMetricsTest.java
@@ -0,0 +1,119 @@
+/*
+ * 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.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
+
+import io.javaoperatorsdk.operator.api.reconciler.RetryInfo;
+import io.javaoperatorsdk.operator.processing.event.ResourceID;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceAction;
+import io.javaoperatorsdk.operator.processing.event.source.controller.ResourceEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** {@link OperatorJosdkMetrics} tests. */
+public class OperatorJosdkMetricsTest {
+
+ private final ResourceID resourceId = new ResourceID("testns", "testname");
+ private final String resourcePrefix =
+ "testhost.k8soperator.flink-operator-test.testopname.resource.testname.testns.JOSDK.";
+ private final String systemPrefix =
+ "testhost.k8soperator.flink-operator-test.testopname.system.";
+
+ private Map<String, Metric> metrics = new HashMap<>();
+ private OperatorJosdkMetrics operatorMetrics;
+
+ @BeforeEach
+ public void setup() {
+ TestingMetricRegistry registry =
+ TestingMetricRegistry.builder()
+ .setDelimiter(".".charAt(0))
+ .setRegisterConsumer(
+ (metric, name, group) -> {
+ metrics.put(group.getMetricIdentifier(name), metric);
+ })
+ .build();
+ operatorMetrics =
+ new OperatorJosdkMetrics(
+ TestUtils.createTestMetricGroup(registry, new Configuration()),
+ new Configuration());
+ }
+
+ @Test
+ public void testMetrics() {
+ operatorMetrics.failedReconciliation(resourceId, null);
+ assertEquals(1, metrics.size());
+ assertEquals(1, getCount("Reconciliation.failed"));
+ operatorMetrics.failedReconciliation(resourceId, null);
+ operatorMetrics.failedReconciliation(resourceId, null);
+ assertEquals(1, metrics.size());
+ assertEquals(3, getCount("Reconciliation.failed"));
+
+ operatorMetrics.reconcileCustomResource(resourceId, null);
+ assertEquals(2, metrics.size());
+ assertEquals(1, getCount("Reconciliation"));
+
+ operatorMetrics.reconcileCustomResource(
+ resourceId,
+ new RetryInfo() {
+ @Override
+ public int getAttemptCount() {
+ return 0;
+ }
+
+ @Override
+ public boolean isLastAttempt() {
+ return false;
+ }
+ });
+ assertEquals(3, metrics.size());
+ assertEquals(2, getCount("Reconciliation"));
+ assertEquals(1, getCount("Reconciliation.retries"));
+
+ operatorMetrics.receivedEvent(new ResourceEvent(ResourceAction.ADDED, resourceId, null));
+ assertEquals(5, metrics.size());
+ assertEquals(1, getCount("Resource.Event"));
+ assertEquals(1, getCount("Resource.Event.ADDED"));
+
+ operatorMetrics.cleanupDoneFor(resourceId);
+ assertEquals(6, metrics.size());
+ assertEquals(1, getCount("Reconciliation.cleanup"));
+
+ operatorMetrics.finishedReconciliation(resourceId);
+ assertEquals(7, metrics.size());
+ assertEquals(1, getCount("Reconciliation.finished"));
+
+ operatorMetrics.monitorSizeOf(Map.of("a", "b", "c", "d"), "mymap");
+ assertEquals(8, metrics.size());
+ assertEquals(2, ((Gauge<Integer>) metrics.get(systemPrefix + "mymap.size")).getValue());
+ }
+
+ private long getCount(String name) {
+ return ((Counter) metrics.get(resourcePrefix + name + ".Count")).getCount();
+ }
+}
[flink-kubernetes-operator] 01/03: [FLINK-28297] Improve operator metric groups
Posted by gy...@apache.org.
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
commit 1b3acc23868f9ad09ea73e7556a7b31176496783
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Thu Jul 7 10:20:26 2022 +0200
[FLINK-28297] Improve operator metric groups
---
.../flink/kubernetes/operator/FlinkOperator.java | 12 ++--
.../metrics/KubernetesOperatorMetricGroup.java | 24 ++++---
.../metrics/KubernetesOperatorMetricOptions.java | 18 +++++-
...oup.java => KubernetesResourceMetricGroup.java} | 47 ++++----------
...=> KubernetesResourceNamespaceMetricGroup.java} | 55 +++++++---------
.../KubernetesResourceNamespaceScopeFormat.java | 46 ++++++++++++++
.../metrics/KubernetesResourceScopeFormat.java | 52 +++++++++++++++
.../kubernetes/operator/metrics/MetricManager.java | 19 +++---
.../operator/metrics/OperatorMetricUtils.java | 2 +-
.../flink/kubernetes/operator/TestUtils.java | 28 +++++++++
.../kubernetes/operator/TestingStatusRecorder.java | 5 +-
.../TestingFlinkDeploymentController.java | 5 +-
.../listener/FlinkResourceListenerTest.java | 5 +-
.../metrics/KubernetesOperatorMetricGroupTest.java | 48 +++++++++++++-
.../operator/utils/StatusRecorderTest.java | 73 ++++++++++++++--------
15 files changed, 309 insertions(+), 130 deletions(-)
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 590af2c..419702d 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
@@ -28,6 +28,7 @@ import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus;
import org.apache.flink.kubernetes.operator.listener.FlinkResourceListener;
import org.apache.flink.kubernetes.operator.listener.ListenerUtils;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.metrics.OperatorMetricUtils;
import org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
@@ -40,7 +41,6 @@ import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
-import org.apache.flink.metrics.MetricGroup;
import io.fabric8.kubernetes.client.DefaultKubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClient;
@@ -70,7 +70,7 @@ public class FlinkOperator {
private final FlinkConfigManager configManager;
private final Set<FlinkResourceValidator> validators;
private final Set<RegisteredController> registeredControllers = new HashSet<>();
- private final MetricGroup metricGroup;
+ private final KubernetesOperatorMetricGroup metricGroup;
private final Collection<FlinkResourceListener> listeners;
public FlinkOperator(@Nullable Configuration conf) {
@@ -114,7 +114,9 @@ public class FlinkOperator {
private void registerDeploymentController() {
var statusRecorder =
StatusRecorder.<FlinkDeploymentStatus>create(
- client, new MetricManager<>(metricGroup), listeners);
+ client,
+ new MetricManager<>(metricGroup, configManager.getDefaultConfig()),
+ listeners);
var eventRecorder = EventRecorder.create(client, listeners);
var reconcilerFactory =
new ReconcilerFactory(
@@ -137,7 +139,9 @@ public class FlinkOperator {
var eventRecorder = EventRecorder.create(client, listeners);
var statusRecorder =
StatusRecorder.<FlinkSessionJobStatus>create(
- client, new MetricManager<>(metricGroup), listeners);
+ client,
+ new MetricManager<>(metricGroup, configManager.getDefaultConfig()),
+ listeners);
var reconciler =
new SessionJobReconciler(
client, flinkService, configManager, eventRecorder, statusRecorder);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
index 3b1c088..a410d6f 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
@@ -26,14 +26,12 @@ import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import java.util.Map;
-/** Flink based operator metric group. */
-public class KubernetesOperatorMetricGroup
- extends AbstractMetricGroup<KubernetesOperatorMetricGroup> {
+/** Base metric group for Flink Operator System metrics. */
+public class KubernetesOperatorMetricGroup extends AbstractMetricGroup<AbstractMetricGroup<?>> {
- private static final String GROUP_NAME = "k8soperator";
- private final String namespace;
- private final String name;
- private final String hostname;
+ protected final String namespace;
+ protected final String name;
+ protected final String hostname;
private KubernetesOperatorMetricGroup(
MetricRegistry registry,
@@ -47,6 +45,16 @@ public class KubernetesOperatorMetricGroup
this.hostname = hostname;
}
+ public KubernetesResourceNamespaceMetricGroup createResourceNamespaceGroup(
+ Configuration config, String resourceNs) {
+ return new KubernetesResourceNamespaceMetricGroup(
+ registry,
+ this,
+ KubernetesResourceNamespaceScopeFormat.fromConfig(config)
+ .formatScope(namespace, name, hostname, resourceNs),
+ resourceNs);
+ }
+
public static KubernetesOperatorMetricGroup create(
MetricRegistry metricRegistry,
Configuration configuration,
@@ -71,7 +79,7 @@ public class KubernetesOperatorMetricGroup
@Override
protected final String getGroupName(CharacterFilter filter) {
- return GROUP_NAME;
+ return "k8soperator";
}
@Override
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
index 42e1d40..30b02f9 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
@@ -23,8 +23,22 @@ import org.apache.flink.configuration.ConfigOptions;
/** Configuration options for metrics. */
public class KubernetesOperatorMetricOptions {
public static final ConfigOption<String> SCOPE_NAMING_KUBERNETES_OPERATOR =
- ConfigOptions.key("metrics.scope.k8soperator")
- .defaultValue("<host>.k8soperator.<namespace>.<name>")
+ ConfigOptions.key("metrics.scope.k8soperator.system")
+ .defaultValue("<host>.k8soperator.<namespace>.<name>.system")
+ .withDeprecatedKeys("metrics.scope.k8soperator")
.withDescription(
"Defines the scope format string that is applied to all metrics scoped to the kubernetes operator.");
+
+ public static final ConfigOption<String> SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCENS =
+ ConfigOptions.key("metrics.scope.k8soperator.resourcens")
+ .defaultValue("<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>.")
+ .withDescription(
+ "Defines the scope format string that is applied to all metrics scoped to the kubernetes operator resource namespace.");
+
+ public static final ConfigOption<String> SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCE =
+ ConfigOptions.key("metrics.scope.k8soperator.resource")
+ .defaultValue(
+ "<host>.k8soperator.<namespace>.<name>.resource.<resourcens>.<resourcename>")
+ .withDescription(
+ "Defines the scope format string that is applied to all metrics scoped to the kubernetes operator resource.");
}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceMetricGroup.java
similarity index 51%
copy from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
copy to flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceMetricGroup.java
index 3b1c088..b76a594 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceMetricGroup.java
@@ -17,61 +17,36 @@
package org.apache.flink.kubernetes.operator.metrics;
-import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.CharacterFilter;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
-import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import java.util.Map;
-/** Flink based operator metric group. */
-public class KubernetesOperatorMetricGroup
- extends AbstractMetricGroup<KubernetesOperatorMetricGroup> {
+/** Base metric group for Flink Operator Resource level metrics. */
+public class KubernetesResourceMetricGroup
+ extends AbstractMetricGroup<KubernetesResourceNamespaceMetricGroup> {
- private static final String GROUP_NAME = "k8soperator";
- private final String namespace;
- private final String name;
- private final String hostname;
+ private final String resourceName;
- private KubernetesOperatorMetricGroup(
+ protected KubernetesResourceMetricGroup(
MetricRegistry registry,
+ KubernetesResourceNamespaceMetricGroup parent,
String[] scope,
- String namespace,
- String name,
- String hostname) {
- super(registry, scope, null);
- this.namespace = namespace;
- this.name = name;
- this.hostname = hostname;
- }
-
- public static KubernetesOperatorMetricGroup create(
- MetricRegistry metricRegistry,
- Configuration configuration,
- String namespace,
- String name,
- String hostname) {
- return new KubernetesOperatorMetricGroup(
- metricRegistry,
- KubernetesOperatorScopeFormat.fromConfig(configuration)
- .formatScope(namespace, name, hostname),
- namespace,
- name,
- hostname);
+ String resourceName) {
+ super(registry, scope, parent);
+ this.resourceName = resourceName;
}
@Override
protected final void putVariables(Map<String, String> variables) {
- variables.put(KubernetesOperatorScopeFormat.NAMESPACE, namespace);
- variables.put(KubernetesOperatorScopeFormat.NAME, name);
- variables.put(ScopeFormat.SCOPE_HOST, hostname);
+ variables.put(KubernetesResourceScopeFormat.RESOURCE, resourceName);
}
@Override
protected final String getGroupName(CharacterFilter filter) {
- return GROUP_NAME;
+ return "resource";
}
@Override
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceMetricGroup.java
similarity index 56%
copy from flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
copy to flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceMetricGroup.java
index 3b1c088..bd6f8e2 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroup.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceMetricGroup.java
@@ -22,56 +22,47 @@ import org.apache.flink.metrics.CharacterFilter;
import org.apache.flink.runtime.metrics.MetricRegistry;
import org.apache.flink.runtime.metrics.dump.QueryScopeInfo;
import org.apache.flink.runtime.metrics.groups.AbstractMetricGroup;
-import org.apache.flink.runtime.metrics.scope.ScopeFormat;
import java.util.Map;
-/** Flink based operator metric group. */
-public class KubernetesOperatorMetricGroup
+/** Base metric group for Flink Operator Resource namespace level metrics. */
+public class KubernetesResourceNamespaceMetricGroup
extends AbstractMetricGroup<KubernetesOperatorMetricGroup> {
- private static final String GROUP_NAME = "k8soperator";
- private final String namespace;
- private final String name;
- private final String hostname;
+ private final String resourceNs;
- private KubernetesOperatorMetricGroup(
+ protected KubernetesResourceNamespaceMetricGroup(
MetricRegistry registry,
+ KubernetesOperatorMetricGroup parent,
String[] scope,
- String namespace,
- String name,
- String hostname) {
- super(registry, scope, null);
- this.namespace = namespace;
- this.name = name;
- this.hostname = hostname;
+ String resourceNs) {
+ super(registry, scope, parent);
+ this.resourceNs = resourceNs;
}
- public static KubernetesOperatorMetricGroup create(
- MetricRegistry metricRegistry,
- Configuration configuration,
- String namespace,
- String name,
- String hostname) {
- return new KubernetesOperatorMetricGroup(
- metricRegistry,
- KubernetesOperatorScopeFormat.fromConfig(configuration)
- .formatScope(namespace, name, hostname),
- namespace,
- name,
- hostname);
+ public KubernetesResourceMetricGroup createResourceNamespaceGroup(
+ Configuration config, String resourceName) {
+ return new KubernetesResourceMetricGroup(
+ registry,
+ this,
+ KubernetesResourceScopeFormat.fromConfig(config)
+ .formatScope(
+ parent.namespace,
+ parent.name,
+ parent.hostname,
+ resourceNs,
+ resourceName),
+ resourceName);
}
@Override
protected final void putVariables(Map<String, String> variables) {
- variables.put(KubernetesOperatorScopeFormat.NAMESPACE, namespace);
- variables.put(KubernetesOperatorScopeFormat.NAME, name);
- variables.put(ScopeFormat.SCOPE_HOST, hostname);
+ variables.put(KubernetesResourceNamespaceScopeFormat.RESOURCE_NS, resourceNs);
}
@Override
protected final String getGroupName(CharacterFilter filter) {
- return GROUP_NAME;
+ return "namespace";
}
@Override
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceScopeFormat.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceScopeFormat.java
new file mode 100644
index 0000000..17b6ffe
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceNamespaceScopeFormat.java
@@ -0,0 +1,46 @@
+/*
+ * 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.apache.flink.runtime.metrics.scope.ScopeFormat;
+
+import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCENS;
+import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAME;
+import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAMESPACE;
+
+/** Format for metrics. * */
+public class KubernetesResourceNamespaceScopeFormat extends ScopeFormat {
+
+ public static final String RESOURCE_NS = asVariable("resourcens");
+
+ public KubernetesResourceNamespaceScopeFormat(String format) {
+ super(format, null, new String[] {NAMESPACE, NAME, SCOPE_HOST, RESOURCE_NS});
+ }
+
+ public String[] formatScope(String namespace, String name, String hostname, String resourceNs) {
+ final String[] template = copyTemplate();
+ final String[] values = {namespace, name, hostname, resourceNs};
+ return bindVariables(template, values);
+ }
+
+ public static KubernetesResourceNamespaceScopeFormat fromConfig(Configuration config) {
+ String format = config.getString(SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCENS);
+ return new KubernetesResourceNamespaceScopeFormat(format);
+ }
+}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceScopeFormat.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceScopeFormat.java
new file mode 100644
index 0000000..7f05ab6
--- /dev/null
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesResourceScopeFormat.java
@@ -0,0 +1,52 @@
+/*
+ * 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.apache.flink.runtime.metrics.scope.ScopeFormat;
+
+import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions.SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCE;
+import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAME;
+import static org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorScopeFormat.NAMESPACE;
+import static org.apache.flink.kubernetes.operator.metrics.KubernetesResourceNamespaceScopeFormat.RESOURCE_NS;
+
+/** Format for metrics. * */
+public class KubernetesResourceScopeFormat extends ScopeFormat {
+
+ public static final String RESOURCE = asVariable("resourcename");
+
+ public KubernetesResourceScopeFormat(String format) {
+ super(format, null, new String[] {NAMESPACE, NAME, SCOPE_HOST, RESOURCE_NS, RESOURCE});
+ }
+
+ public String[] formatScope(
+ String namespace,
+ String name,
+ String hostname,
+ String resourceNs,
+ String resourceName) {
+ final String[] template = copyTemplate();
+ final String[] values = {namespace, name, hostname, resourceNs, resourceName};
+ return bindVariables(template, values);
+ }
+
+ public static KubernetesResourceScopeFormat fromConfig(Configuration config) {
+ String format = config.getString(SCOPE_NAMING_KUBERNETES_OPERATOR_RESOURCE);
+ return new KubernetesResourceScopeFormat(format);
+ }
+}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java
index 8cd9c3d..18b96b4 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java
@@ -17,9 +17,9 @@
package org.apache.flink.kubernetes.operator.metrics;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
-import org.apache.flink.metrics.MetricGroup;
import io.fabric8.kubernetes.client.CustomResource;
@@ -28,12 +28,13 @@ import java.util.concurrent.ConcurrentHashMap;
/** Metric manager for Operator managed custom resources. */
public class MetricManager<CR extends CustomResource<?, ?>> {
- public static final String NS_SCOPE_KEY = "resourcens";
- private final MetricGroup metricGroup;
+ private final KubernetesOperatorMetricGroup opMetricGroup;
+ private final Configuration conf;
private final Map<String, CustomResourceMetrics> metrics = new ConcurrentHashMap<>();
- public MetricManager(MetricGroup metricGroup) {
- this.metricGroup = metricGroup;
+ public MetricManager(KubernetesOperatorMetricGroup opMetricGroup, Configuration conf) {
+ this.opMetricGroup = opMetricGroup;
+ this.conf = conf;
}
public void onUpdate(CR cr) {
@@ -50,12 +51,12 @@ public class MetricManager<CR extends CustomResource<?, ?>> {
}
private CustomResourceMetrics getCustomResourceMetricsImpl(CR cr) {
+ var namespaceMg =
+ opMetricGroup.createResourceNamespaceGroup(conf, cr.getMetadata().getNamespace());
if (cr instanceof FlinkDeployment) {
- return new FlinkDeploymentMetrics(
- metricGroup.addGroup(NS_SCOPE_KEY, cr.getMetadata().getNamespace()));
+ return new FlinkDeploymentMetrics(namespaceMg);
} else if (cr instanceof FlinkSessionJob) {
- return new FlinkSessionJobMetrics(
- metricGroup.addGroup(NS_SCOPE_KEY, cr.getMetadata().getNamespace()));
+ return new FlinkSessionJobMetrics(namespaceMg);
} else {
throw new IllegalArgumentException("Unknown CustomResource");
}
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 0e53caf..6040d39 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
@@ -43,7 +43,7 @@ public class OperatorMetricUtils {
private static final String OPERATOR_METRICS_PREFIX = "kubernetes.operator.metrics.";
private static final String METRICS_PREFIX = "metrics.";
- public static MetricGroup initOperatorMetrics(Configuration defaultConfig) {
+ public static KubernetesOperatorMetricGroup initOperatorMetrics(Configuration defaultConfig) {
Configuration metricConfig = createMetricConfig(defaultConfig);
LOG.info("Initializing operator metrics using conf: {}", metricConfig);
PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(metricConfig);
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 f8b6290..038ffd7 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
@@ -18,9 +18,11 @@
package org.apache.flink.kubernetes.operator;
import org.apache.flink.configuration.CheckpointingOptions;
+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;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.FlinkSessionJob;
import org.apache.flink.kubernetes.operator.crd.spec.FlinkDeploymentSpec;
@@ -36,6 +38,10 @@ import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.FlinkSessionJobStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.kubernetes.operator.metrics.MetricManager;
+import org.apache.flink.runtime.metrics.MetricRegistry;
+import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
import io.fabric8.kubernetes.api.model.Container;
import io.fabric8.kubernetes.api.model.ContainerStatus;
@@ -421,6 +427,28 @@ public class TestUtils {
}
}
+ public static <T extends AbstractFlinkResource<?, ?>> MetricManager<T> createTestMetricManager(
+ Configuration conf) {
+ return createTestMetricManager(
+ TestingMetricRegistry.builder()
+ .setDelimiter(".".charAt(0))
+ .setRegisterConsumer((metric, name, group) -> {})
+ .build(),
+ conf);
+ }
+
+ public static <T extends AbstractFlinkResource<?, ?>> MetricManager<T> createTestMetricManager(
+ MetricRegistry metricRegistry, Configuration conf) {
+
+ return new MetricManager<>(createTestMetricGroup(metricRegistry, conf), conf);
+ }
+
+ public static KubernetesOperatorMetricGroup createTestMetricGroup(
+ MetricRegistry metricRegistry, Configuration conf) {
+ return KubernetesOperatorMetricGroup.create(
+ metricRegistry, conf, TEST_NAMESPACE, "testopname", "testhost");
+ }
+
/** Testing ResponseProvider. */
public static class ValidatingResponseProvider<T> implements ResponseProvider<Object> {
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java
index 8306761..5e9cfcc 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingStatusRecorder.java
@@ -18,11 +18,10 @@
package org.apache.flink.kubernetes.operator;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.crd.status.CommonStatus;
-import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
-import org.apache.flink.metrics.testutils.MetricListener;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -30,7 +29,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
public class TestingStatusRecorder<STATUS extends CommonStatus<?>> extends StatusRecorder<STATUS> {
public TestingStatusRecorder() {
- super(null, new MetricManager<>(new MetricListener().getMetricGroup()), (r, s) -> {});
+ super(null, TestUtils.createTestMetricManager(new Configuration()), (r, s) -> {});
}
@Override
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
index f78b769..68a29f7 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/controller/TestingFlinkDeploymentController.java
@@ -17,19 +17,18 @@
package org.apache.flink.kubernetes.operator.controller;
+import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.TestingFlinkService;
import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
-import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.observer.deployment.ObserverFactory;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.reconciler.deployment.ReconcilerFactory;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
import org.apache.flink.kubernetes.operator.utils.ValidatorUtils;
-import org.apache.flink.metrics.testutils.MetricListener;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
@@ -67,7 +66,7 @@ public class TestingFlinkDeploymentController
statusRecorder =
new StatusRecorder<>(
kubernetesClient,
- new MetricManager<>(new MetricListener().getMetricGroup()),
+ TestUtils.createTestMetricManager(configManager.getDefaultConfig()),
statusUpdateCounter);
flinkDeploymentController =
new FlinkDeploymentController(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java
index 10a25da..e68c910 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/listener/FlinkResourceListenerTest.java
@@ -17,14 +17,13 @@
package org.apache.flink.kubernetes.operator.listener;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
-import org.apache.flink.kubernetes.operator.metrics.MetricManager;
import org.apache.flink.kubernetes.operator.utils.EventRecorder;
import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
-import org.apache.flink.metrics.testutils.MetricListener;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
@@ -56,7 +55,7 @@ public class FlinkResourceListenerTest {
var statusRecorder =
StatusRecorder.<FlinkDeploymentStatus>create(
kubernetesClient,
- new MetricManager<>(new MetricListener().getMetricGroup()),
+ TestUtils.createTestMetricManager(new Configuration()),
listeners);
var eventRecorder = EventRecorder.create(kubernetesClient, listeners);
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
index 2e1bfc3..b02f112 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricGroupTest.java
@@ -44,10 +44,12 @@ public class KubernetesOperatorMetricGroupTest {
"flink-kubernetes-operator",
"localhost");
assertArrayEquals(
- new String[] {"localhost", "k8soperator", "default", "flink-kubernetes-operator"},
+ new String[] {
+ "localhost", "k8soperator", "default", "flink-kubernetes-operator", "system"
+ },
group.getScopeComponents());
assertEquals(
- "localhost.k8soperator.default.flink-kubernetes-operator.test",
+ "localhost.k8soperator.default.flink-kubernetes-operator.system.test",
group.getMetricIdentifier("test"));
assertEquals(
@@ -96,6 +98,48 @@ public class KubernetesOperatorMetricGroupTest {
registry.shutdown().get();
}
+ @Test
+ public void testSubGroupVariables() throws Exception {
+ var configuration = new Configuration();
+ var registry = new MetricRegistryImpl(fromConfiguration(configuration));
+ var operatorMetricGroup =
+ KubernetesOperatorMetricGroup.create(
+ registry,
+ configuration,
+ "default",
+ "flink-kubernetes-operator",
+ "localhost");
+
+ var namespaceGroup = operatorMetricGroup.createResourceNamespaceGroup(configuration, "rns");
+ var resourceGroup = namespaceGroup.createResourceNamespaceGroup(configuration, "rn");
+
+ assertEquals(
+ ImmutableMap.of(
+ "<host>",
+ "localhost",
+ "<namespace>",
+ "default",
+ "<name>",
+ "flink-kubernetes-operator",
+ "<resourcens>",
+ "rns"),
+ namespaceGroup.getAllVariables());
+ assertEquals(
+ ImmutableMap.of(
+ "<host>",
+ "localhost",
+ "<namespace>",
+ "default",
+ "<name>",
+ "flink-kubernetes-operator",
+ "<resourcens>",
+ "rns",
+ "<resourcename>",
+ "rn"),
+ resourceGroup.getAllVariables());
+ registry.shutdown().get();
+ }
+
private static MetricRegistryConfiguration fromConfiguration(Configuration configuration) {
return MetricRegistryConfiguration.fromConfiguration(configuration, Long.MAX_VALUE);
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
index e3cf2eb..52a2f45 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/utils/StatusRecorderTest.java
@@ -17,21 +17,25 @@
package org.apache.flink.kubernetes.operator.utils;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.kubernetes.operator.TestUtils;
import org.apache.flink.kubernetes.operator.crd.FlinkDeployment;
import org.apache.flink.kubernetes.operator.crd.status.FlinkDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.JobManagerDeploymentStatus;
import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
import org.apache.flink.kubernetes.operator.metrics.MetricManager;
-import org.apache.flink.metrics.testutils.MetricListener;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Metric;
+import org.apache.flink.runtime.metrics.util.TestingMetricRegistry;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.junit.jupiter.api.Test;
+import java.util.HashMap;
+
import static org.apache.flink.kubernetes.operator.metrics.FlinkDeploymentMetrics.METRIC_GROUP_NAME;
-import static org.apache.flink.kubernetes.operator.metrics.MetricManager.NS_SCOPE_KEY;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -47,7 +51,7 @@ public class StatusRecorderTest {
var helper =
new StatusRecorder<FlinkDeploymentStatus>(
kubernetesClient,
- new MetricManager<>(new MetricListener().getMetricGroup()),
+ TestUtils.createTestMetricManager(new Configuration()),
(e, s) -> {});
var deployment = TestUtils.buildApplicationCluster();
kubernetesClient.resource(deployment).createOrReplace();
@@ -70,49 +74,64 @@ public class StatusRecorderTest {
@Test
public void testFlinkDeploymentMetrics() throws InterruptedException {
- var metricListener = new MetricListener();
+ var metrics = new HashMap<String, Metric>();
+ TestingMetricRegistry registry =
+ TestingMetricRegistry.builder()
+ .setDelimiter(".".charAt(0))
+ .setRegisterConsumer(
+ (metric, name, group) -> {
+ metrics.put(group.getMetricIdentifier(name), metric);
+ })
+ .build();
+
+ var metricManager =
+ (MetricManager) TestUtils.createTestMetricManager(registry, new Configuration());
var helper =
new StatusRecorder<FlinkDeploymentStatus>(
- kubernetesClient,
- new MetricManager<>(metricListener.getMetricGroup()),
- (e, s) -> {});
+ kubernetesClient, metricManager, (e, s) -> {});
var deployment = TestUtils.buildApplicationCluster();
kubernetesClient.resource(deployment).createOrReplace();
helper.updateStatusFromCache(deployment);
- assertEquals(1, metricListener.getGauge(totalIdentifier(deployment)).get().getValue());
- assertEquals(1, metricListener.getGauge(perStatusIdentifier(deployment)).get().getValue());
+ assertEquals(1, ((Gauge) metrics.get(totalIdentifier(deployment))).getValue());
+ assertEquals(1, ((Gauge) metrics.get(perStatusIdentifier(deployment))).getValue());
for (JobManagerDeploymentStatus status : JobManagerDeploymentStatus.values()) {
deployment.getStatus().setJobManagerDeploymentStatus(status);
helper.patchAndCacheStatus(deployment);
- assertEquals(1, metricListener.getGauge(totalIdentifier(deployment)).get().getValue());
- assertEquals(
- 1, metricListener.getGauge(perStatusIdentifier(deployment)).get().getValue());
+ assertEquals(1, ((Gauge) metrics.get(totalIdentifier(deployment))).getValue());
+ assertEquals(1, ((Gauge) metrics.get(perStatusIdentifier(deployment))).getValue());
}
helper.removeCachedStatus(deployment);
- assertEquals(0, metricListener.getGauge(totalIdentifier(deployment)).get().getValue());
+ assertEquals(0, ((Gauge) metrics.get(totalIdentifier(deployment))).getValue());
for (JobManagerDeploymentStatus status : JobManagerDeploymentStatus.values()) {
- assertEquals(
- 0, metricListener.getGauge(perStatusIdentifier(deployment)).get().getValue());
+ assertEquals(0, ((Gauge) metrics.get(perStatusIdentifier(deployment))).getValue());
}
}
- private String[] totalIdentifier(FlinkDeployment deployment) {
- return new String[] {
- NS_SCOPE_KEY, deployment.getMetadata().getNamespace(), METRIC_GROUP_NAME, "Count"
- };
+ private String totalIdentifier(FlinkDeployment deployment) {
+ String baseScope = "testhost.k8soperator.flink-operator-test.testopname.";
+ String[] metricScope =
+ new String[] {
+ "namespace", deployment.getMetadata().getNamespace(), METRIC_GROUP_NAME, "Count"
+ };
+ return baseScope + String.join(".", metricScope);
}
- private String[] perStatusIdentifier(FlinkDeployment deployment) {
- return new String[] {
- NS_SCOPE_KEY,
- deployment.getMetadata().getNamespace(),
- METRIC_GROUP_NAME,
- deployment.getStatus().getJobManagerDeploymentStatus().name(),
- "Count"
- };
+ private String perStatusIdentifier(FlinkDeployment deployment) {
+
+ String baseScope = "testhost.k8soperator.flink-operator-test.testopname.";
+ String[] metricScope =
+ new String[] {
+ "namespace",
+ deployment.getMetadata().getNamespace(),
+ METRIC_GROUP_NAME,
+ deployment.getStatus().getJobManagerDeploymentStatus().name(),
+ "Count"
+ };
+
+ return baseScope + String.join(".", metricScope);
}
}
[flink-kubernetes-operator] 03/03: [docs] Break up documentation into sections
Posted by gy...@apache.org.
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
commit f93da5c30dd958309ba492b6e1358afa53a287d1
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Fri Jul 8 08:25:49 2022 +0200
[docs] Break up documentation into sections
---
docs/content/docs/operations/configuration.md | 30 ++++++---
.../shortcodes/generated/dynamic_section.html | 72 ++++++++++++++++++++
.../kubernetes_operator_config_configuration.html | 6 --
.../kubernetes_operator_metric_configuration.html | 36 ++++++++++
.../generated/system_advanced_section.html | 60 +++++++++++++++++
.../shortcodes/generated/system_section.html | 78 ++++++++++++++++++++++
.../configuration/ConfigOptionsDocGenerator.java | 4 +-
.../flink/kubernetes/operator/FlinkOperator.java | 2 +-
.../operator/config/FlinkConfigManager.java | 2 +-
.../config/FlinkOperatorConfiguration.java | 7 +-
.../config/KubernetesOperatorConfigOptions.java | 41 ++++++++++--
.../metrics/KubernetesOperatorMetricOptions.java | 8 +++
.../operator/admission/FlinkOperatorWebhook.java | 2 +-
13 files changed, 319 insertions(+), 29 deletions(-)
diff --git a/docs/content/docs/operations/configuration.md b/docs/content/docs/operations/configuration.md
index 89e0edb..460120c 100644
--- a/docs/content/docs/operations/configuration.md
+++ b/docs/content/docs/operations/configuration.md
@@ -52,7 +52,7 @@ To learn more about metrics and logging configuration please refer to the dedica
## Dynamic Operator Configuration
-The Kubernetes operator supports dynamic config changes through the operator ConfigMaps. Dynamic operator configuration is enabled by default, and can be disabled by setting `kubernetes.operator.dynamic.config.enabled` to false. Time interval for checking dynamic config changes is specified by `kubernetes.operator.dynamic.config.check.interval` of which default value is 5 minutes.
+The Kubernetes operator supports dynamic config changes through the operator ConfigMaps. Dynamic operator configuration is enabled by default, and can be disabled by setting `kubernetes.operator.dynamic.config.enabled` to false. Time interval for checking dynamic config changes is specified by `kubernetes.operator.dynamic.config.check.interval` of which default value is 5 minutes.
Verify whether dynamic operator configuration updates is enabled via the `deploy/flink-kubernetes-operator` log has:
@@ -70,14 +70,26 @@ Verify whether the config value of `kubernetes.operator.reconcile.interval` is u
## Operator Configuration Reference
-{{< generated/kubernetes_operator_config_configuration >}}
+### System Configuration
-## Job Specific Configuration Reference
+General operator system configuration. Cannot be overridden on a per-resource basis.
-Job specific configuration can be configured under `spec.flinkConfiguration` and it will override flink configurations defined in `flink-conf.yaml`.
+{{< generated/system_section >}}
-- For application clusters, `spec.flinkConfiguration` will be located in `FlinkDeployment` CustomResource.
-- For session clusters, configuring `spec.flinkConfiguration` in parent `FlinkDeployment` will be applied to all session jobs within the session cluster.
- - You can configure some additional job specific supplemental configuration through `spec.flinkConfiguration` in `FlinkSessionJob` CustomResource.
- Those session job level configurations will override the parent session cluster's Flink configuration. Please note only the following configurations are considered to be valid configurations.
- - `kubernetes.operator.user.artifacts.http.header`
+### Resource/User Configuration
+
+These options can be configured on both an operator and a per-resource level. When set under `spec.flinkConfiguration` for the Flink resources it will override the default value provided in the operator default configuration (`flink-conf.yaml`).
+
+{{< generated/dynamic_section >}}
+
+### System Metrics Configuration
+
+Operator system metrics configuration. Cannot be overridden on a per-resource basis.
+
+{{< generated/kubernetes_operator_metric_configuration >}}
+
+### Advanced System Configuration
+
+Advanced operator system configuration. Cannot be overridden on a per-resource basis.
+
+{{< generated/system_advanced_section >}}
diff --git a/docs/layouts/shortcodes/generated/dynamic_section.html b/docs/layouts/shortcodes/generated/dynamic_section.html
new file mode 100644
index 0000000..8d8fb62
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/dynamic_section.html
@@ -0,0 +1,72 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>kubernetes.operator.deployment.readiness.timeout</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>The timeout for deployments to become ready/stable before being rolled back if rollback is enabled.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.deployment.rollback.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to enable rolling back failed deployment upgrades.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.jm-deployment-recovery.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to enable recovery of missing/deleted jobmanager deployments.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.job.upgrade.ignore-pending-savepoint</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to ignore pending savepoint during job upgrade.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.job.upgrade.last-state-fallback.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Enables last-state fallback for savepoint upgrade mode. When the job is not running thus savepoint cannot be triggered but HA metadata is available for last state restore the operator can initiate the upgrade process when the flag is enabled.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.periodic.savepoint.interval</h5></td>
+ <td style="word-wrap: break-word;">0 ms</td>
+ <td>Duration</td>
+ <td>Interval at which periodic savepoints will be triggered. The triggering schedule is not guaranteed, savepoints will be triggered as part of the regular reconcile loop.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.savepoint.history.max.age</h5></td>
+ <td style="word-wrap: break-word;">86400000 ms</td>
+ <td>Duration</td>
+ <td>Maximum age for savepoint history entries to retain. Due to lazy clean-up, the most recent savepoint may live longer than the max age.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.savepoint.history.max.count</h5></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>Maximum number of savepoint history entries to retain.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.savepoint.trigger.grace-period</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>The interval before a savepoint trigger attempt is marked as unsuccessful.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.user.artifacts.http.header</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Map</td>
+ <td>Custom HTTP header for HttpArtifactFetcher. The header will be applied when getting the session job artifacts. Expected format: headerKey1:headerValue1,headerKey2:headerValue2.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
index 5c34d95..cfc1ba9 100644
--- a/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
+++ b/docs/layouts/shortcodes/generated/kubernetes_operator_config_configuration.html
@@ -80,12 +80,6 @@
<td>Boolean</td>
<td>Enables last-state fallback for savepoint upgrade mode. When the job is not running thus savepoint cannot be triggered but HA metadata is available for last state restore the operator can initiate the upgrade process when the flag is enabled.</td>
</tr>
- <tr>
- <td><h5>kubernetes.operator.josdk.metrics.enabled</h5></td>
- <td style="word-wrap: break-word;">true</td>
- <td>Boolean</td>
- <td>Enable forwarding of Java Operator SDK metrics to the Flink metric registry.</td>
- </tr>
<tr>
<td><h5>kubernetes.operator.observer.progress-check.interval</h5></td>
<td style="word-wrap: break-word;">10 s</td>
diff --git a/docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html
new file mode 100644
index 0000000..37ee840
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/kubernetes_operator_metric_configuration.html
@@ -0,0 +1,36 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>kubernetes.operator.josdk.metrics.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Enable forwarding of Java Operator SDK metrics to the Flink metric registry.</td>
+ </tr>
+ <tr>
+ <td><h5>metrics.scope.k8soperator.resource</h5></td>
+ <td style="word-wrap: break-word;">"<host>.k8soperator.<namespace>.<name>.resource.<resourcens>.<resourcename>"</td>
+ <td>String</td>
+ <td>Defines the scope format string that is applied to all metrics scoped to the kubernetes operator resource.</td>
+ </tr>
+ <tr>
+ <td><h5>metrics.scope.k8soperator.resourcens</h5></td>
+ <td style="word-wrap: break-word;">"<host>.k8soperator.<namespace>.<name>.namespace.<resourcens>"</td>
+ <td>String</td>
+ <td>Defines the scope format string that is applied to all metrics scoped to the kubernetes operator resource namespace.</td>
+ </tr>
+ <tr>
+ <td><h5>metrics.scope.k8soperator.system</h5></td>
+ <td style="word-wrap: break-word;">"<host>.k8soperator.<namespace>.<name>.system"</td>
+ <td>String</td>
+ <td>Defines the scope format string that is applied to all metrics scoped to the kubernetes operator.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/system_advanced_section.html b/docs/layouts/shortcodes/generated/system_advanced_section.html
new file mode 100644
index 0000000..72d525d
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/system_advanced_section.html
@@ -0,0 +1,60 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>kubernetes.operator.config.cache.size</h5></td>
+ <td style="word-wrap: break-word;">1000</td>
+ <td>Integer</td>
+ <td>Max config cache size.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.config.cache.timeout</h5></td>
+ <td style="word-wrap: break-word;">10 min</td>
+ <td>Duration</td>
+ <td>Expiration time for cached configs.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.dynamic.config.check.interval</h5></td>
+ <td style="word-wrap: break-word;">5 min</td>
+ <td>Duration</td>
+ <td>Time interval for checking config changes.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.dynamic.config.enabled</h5></td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to enable on-the-fly config changes through the operator configmap.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.observer.progress-check.interval</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The interval for observing status for in-progress operations such as deployment and savepoints.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.observer.rest-ready.delay</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>Final delay before deployment is marked ready after port becomes accessible.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.savepoint.history.max.age.threshold</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Duration</td>
+ <td>Maximum age threshold for savepoint history entries to retain.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.savepoint.history.max.count.threshold</h5></td>
+ <td style="word-wrap: break-word;">(none)</td>
+ <td>Integer</td>
+ <td>Maximum number threshold of savepoint history entries to retain.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/docs/layouts/shortcodes/generated/system_section.html b/docs/layouts/shortcodes/generated/system_section.html
new file mode 100644
index 0000000..887326b
--- /dev/null
+++ b/docs/layouts/shortcodes/generated/system_section.html
@@ -0,0 +1,78 @@
+<table class="configuration table table-bordered">
+ <thead>
+ <tr>
+ <th class="text-left" style="width: 20%">Key</th>
+ <th class="text-left" style="width: 15%">Default</th>
+ <th class="text-left" style="width: 10%">Type</th>
+ <th class="text-left" style="width: 55%">Description</th>
+ </tr>
+ </thead>
+ <tbody>
+ <tr>
+ <td><h5>kubernetes.operator.dynamic.namespaces.enabled</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Enables dynamic change of watched/monitored namespaces.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.flink.client.cancel.timeout</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>The timeout for the reconciler to wait for flink to cancel job.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.flink.client.timeout</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
+ <td>Duration</td>
+ <td>The timeout for the observer to wait the flink rest client to return.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.reconcile.interval</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>The interval for the controller to reschedule the reconcile process.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.reconcile.parallelism</h5></td>
+ <td style="word-wrap: break-word;">5</td>
+ <td>Integer</td>
+ <td>The maximum number of threads running the reconciliation loop. Use -1 for infinite.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.resource.cleanup.timeout</h5></td>
+ <td style="word-wrap: break-word;">1 min</td>
+ <td>Duration</td>
+ <td>The timeout for the resource clean up to wait for flink to shutdown cluster.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.retry.initial.interval</h5></td>
+ <td style="word-wrap: break-word;">5 s</td>
+ <td>Duration</td>
+ <td>Initial interval of automatic reconcile retries on recoverable errors.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.retry.interval.multiplier</h5></td>
+ <td style="word-wrap: break-word;">2.0</td>
+ <td>Double</td>
+ <td>Interval multiplier of automatic reconcile retries on recoverable errors.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.retry.max.attempts</h5></td>
+ <td style="word-wrap: break-word;">10</td>
+ <td>Integer</td>
+ <td>Max attempts of automatic reconcile retries on recoverable errors.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.user.artifacts.base.dir</h5></td>
+ <td style="word-wrap: break-word;">"/opt/flink/artifacts"</td>
+ <td>String</td>
+ <td>The base dir to put the session job artifacts.</td>
+ </tr>
+ <tr>
+ <td><h5>kubernetes.operator.watched.namespaces</h5></td>
+ <td style="word-wrap: break-word;">"JOSDK_ALL_NAMESPACES"</td>
+ <td>String</td>
+ <td>Comma separated list of namespaces the operator monitors for custom resources.</td>
+ </tr>
+ </tbody>
+</table>
diff --git a/flink-kubernetes-docs/src/main/java/org/apache/flink/kubernetes/operator/docs/configuration/ConfigOptionsDocGenerator.java b/flink-kubernetes-docs/src/main/java/org/apache/flink/kubernetes/operator/docs/configuration/ConfigOptionsDocGenerator.java
index ba182d3..43c81b4 100644
--- a/flink-kubernetes-docs/src/main/java/org/apache/flink/kubernetes/operator/docs/configuration/ConfigOptionsDocGenerator.java
+++ b/flink-kubernetes-docs/src/main/java/org/apache/flink/kubernetes/operator/docs/configuration/ConfigOptionsDocGenerator.java
@@ -72,7 +72,9 @@ public class ConfigOptionsDocGenerator {
static final OptionsClassLocation[] LOCATIONS =
new OptionsClassLocation[] {
new OptionsClassLocation(
- "flink-kubernetes-operator", "org.apache.flink.kubernetes.operator.config")
+ "flink-kubernetes-operator", "org.apache.flink.kubernetes.operator.config"),
+ new OptionsClassLocation(
+ "flink-kubernetes-operator", "org.apache.flink.kubernetes.operator.metrics")
};
static final String DEFAULT_PATH_PREFIX = "src/main/java";
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 13bbe89..3324478 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
@@ -110,7 +110,7 @@ public class FlinkOperator {
LOG.info("Configuring operator with {} reconciliation threads.", parallelism);
overrider.withConcurrentReconciliationThreads(parallelism);
}
- if (configManager.getOperatorConfiguration().getJosdkMetricsEnabled()) {
+ if (configManager.getOperatorConfiguration().isJosdkMetricsEnabled()) {
overrider.withMetrics(
new OperatorJosdkMetrics(metricGroup, configManager.getDefaultConfig()));
}
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 caccf10..c70dbae 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
@@ -132,7 +132,7 @@ public class FlinkConfigManager {
.orElse(Set.of());
this.operatorConfiguration = FlinkOperatorConfiguration.fromConfiguration(newConf);
var newNs = this.operatorConfiguration.getWatchedNamespaces();
- if (this.operatorConfiguration.getDynamicNamespacesEnabled() && !oldNs.equals(newNs)) {
+ if (this.operatorConfiguration.isDynamicNamespacesEnabled() && !oldNs.equals(newNs)) {
this.namespaceListener.accept(operatorConfiguration.getWatchedNamespaces());
}
this.defaultConfig = newConf.clone();
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 d295501..8983b15 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
@@ -19,6 +19,7 @@
package org.apache.flink.kubernetes.operator.config;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricOptions;
import org.apache.flink.kubernetes.operator.utils.EnvUtils;
import io.javaoperatorsdk.operator.api.config.RetryConfiguration;
@@ -42,8 +43,8 @@ public class FlinkOperatorConfiguration {
Duration flinkClientTimeout;
String flinkServiceHostOverride;
Set<String> watchedNamespaces;
- Boolean dynamicNamespacesEnabled;
- Boolean josdkMetricsEnabled;
+ boolean dynamicNamespacesEnabled;
+ boolean josdkMetricsEnabled;
Duration flinkCancelJobTimeout;
Duration flinkShutdownClusterTimeout;
String artifactsBaseDir;
@@ -110,7 +111,7 @@ public class FlinkOperatorConfiguration {
KubernetesOperatorConfigOptions.OPERATOR_DYNAMIC_NAMESPACES_ENABLED);
boolean josdkMetricsEnabled =
- operatorConfig.get(KubernetesOperatorConfigOptions.OPERATOR_JOSDK_METRICS_ENABLED);
+ operatorConfig.get(KubernetesOperatorMetricOptions.OPERATOR_JOSDK_METRICS_ENABLED);
RetryConfiguration retryConfiguration = new FlinkOperatorRetryConfiguration(operatorConfig);
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
index b01e845..d76aaef 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java
@@ -18,6 +18,7 @@
package org.apache.flink.kubernetes.operator.config;
+import org.apache.flink.annotation.docs.Documentation;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
@@ -30,6 +31,11 @@ import java.util.Map;
/** This class holds configuration constants used by flink operator. */
public class KubernetesOperatorConfigOptions {
+ public static final String SECTION_SYSTEM = "system";
+ public static final String SECTION_ADVANCED = "system_advanced";
+ public static final String SECTION_DYNAMIC = "dynamic";
+
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_RECONCILE_INTERVAL =
ConfigOptions.key("kubernetes.operator.reconcile.interval")
.durationType()
@@ -38,6 +44,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The interval for the controller to reschedule the reconcile process.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_OBSERVER_REST_READY_DELAY =
ConfigOptions.key("kubernetes.operator.observer.rest-ready.delay")
.durationType()
@@ -45,6 +52,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Final delay before deployment is marked ready after port becomes accessible.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Integer> OPERATOR_RECONCILE_PARALLELISM =
ConfigOptions.key("kubernetes.operator.reconcile.parallelism")
.intType()
@@ -53,6 +61,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The maximum number of threads running the reconciliation loop. Use -1 for infinite.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_OBSERVER_PROGRESS_CHECK_INTERVAL =
ConfigOptions.key("kubernetes.operator.observer.progress-check.interval")
.durationType()
@@ -60,6 +69,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The interval for observing status for in-progress operations such as deployment and savepoints.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration> OPERATOR_SAVEPOINT_TRIGGER_GRACE_PERIOD =
ConfigOptions.key("kubernetes.operator.savepoint.trigger.grace-period")
.durationType()
@@ -69,6 +79,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The interval before a savepoint trigger attempt is marked as unsuccessful.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_FLINK_CLIENT_TIMEOUT =
ConfigOptions.key("kubernetes.operator.flink.client.timeout")
.durationType()
@@ -77,6 +88,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The timeout for the observer to wait the flink rest client to return.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_FLINK_CLIENT_CANCEL_TIMEOUT =
ConfigOptions.key("kubernetes.operator.flink.client.cancel.timeout")
.durationType()
@@ -85,6 +97,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The timeout for the reconciler to wait for flink to cancel job.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_RESOURCE_CLEANUP_TIMEOUT =
ConfigOptions.key("kubernetes.operator.resource.cleanup.timeout")
.durationType()
@@ -94,12 +107,14 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"The timeout for the resource clean up to wait for flink to shutdown cluster.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean> DEPLOYMENT_ROLLBACK_ENABLED =
ConfigOptions.key("kubernetes.operator.deployment.rollback.enabled")
.booleanType()
.defaultValue(false)
.withDescription("Whether to enable rolling back failed deployment upgrades.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration> DEPLOYMENT_READINESS_TIMEOUT =
ConfigOptions.key("kubernetes.operator.deployment.readiness.timeout")
.durationType()
@@ -108,18 +123,21 @@ public class KubernetesOperatorConfigOptions {
"The timeout for deployments to become ready/stable "
+ "before being rolled back if rollback is enabled.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<String> OPERATOR_USER_ARTIFACTS_BASE_DIR =
ConfigOptions.key("kubernetes.operator.user.artifacts.base.dir")
.stringType()
.defaultValue("/opt/flink/artifacts")
.withDescription("The base dir to put the session job artifacts.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean> JOB_UPGRADE_IGNORE_PENDING_SAVEPOINT =
ConfigOptions.key("kubernetes.operator.job.upgrade.ignore-pending-savepoint")
.booleanType()
.defaultValue(false)
.withDescription("Whether to ignore pending savepoint during job upgrade.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Boolean> OPERATOR_DYNAMIC_CONFIG_ENABLED =
ConfigOptions.key("kubernetes.operator.dynamic.config.enabled")
.booleanType()
@@ -127,24 +145,28 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Whether to enable on-the-fly config changes through the operator configmap.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_DYNAMIC_CONFIG_CHECK_INTERVAL =
ConfigOptions.key("kubernetes.operator.dynamic.config.check.interval")
.durationType()
.defaultValue(Duration.ofMinutes(5))
.withDescription("Time interval for checking config changes.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_CONFIG_CACHE_TIMEOUT =
ConfigOptions.key("kubernetes.operator.config.cache.timeout")
.durationType()
.defaultValue(Duration.ofMinutes(10))
.withDescription("Expiration time for cached configs.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Integer> OPERATOR_CONFIG_CACHE_SIZE =
ConfigOptions.key("kubernetes.operator.config.cache.size")
.intType()
.defaultValue(1000)
.withDescription("Max config cache size.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean> OPERATOR_JM_DEPLOYMENT_RECOVERY_ENABLED =
ConfigOptions.key("kubernetes.operator.jm-deployment-recovery.enabled")
.booleanType()
@@ -154,12 +176,14 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Whether to enable recovery of missing/deleted jobmanager deployments.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Integer> OPERATOR_SAVEPOINT_HISTORY_MAX_COUNT =
ConfigOptions.key("kubernetes.operator.savepoint.history.max.count")
.intType()
.defaultValue(10)
.withDescription("Maximum number of savepoint history entries to retain.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Integer> OPERATOR_SAVEPOINT_HISTORY_MAX_COUNT_THRESHOLD =
ConfigOptions.key(OPERATOR_SAVEPOINT_HISTORY_MAX_COUNT.key() + ".threshold")
.intType()
@@ -167,6 +191,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Maximum number threshold of savepoint history entries to retain.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration> OPERATOR_SAVEPOINT_HISTORY_MAX_AGE =
ConfigOptions.key("kubernetes.operator.savepoint.history.max.age")
.durationType()
@@ -174,6 +199,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Maximum age for savepoint history entries to retain. Due to lazy clean-up, the most recent savepoint may live longer than the max age.");
+ @Documentation.Section(SECTION_ADVANCED)
public static final ConfigOption<Duration> OPERATOR_SAVEPOINT_HISTORY_MAX_AGE_THRESHOLD =
ConfigOptions.key(OPERATOR_SAVEPOINT_HISTORY_MAX_AGE.key() + ".threshold")
.durationType()
@@ -181,6 +207,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Maximum age threshold for savepoint history entries to retain.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Map<String, String>> JAR_ARTIFACT_HTTP_HEADER =
ConfigOptions.key("kubernetes.operator.user.artifacts.http.header")
.mapType()
@@ -189,6 +216,7 @@ public class KubernetesOperatorConfigOptions {
"Custom HTTP header for HttpArtifactFetcher. The header will be applied when getting the session job artifacts. "
+ "Expected format: headerKey1:headerValue1,headerKey2:headerValue2.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Duration> PERIODIC_SAVEPOINT_INTERVAL =
ConfigOptions.key("kubernetes.operator.periodic.savepoint.interval")
.durationType()
@@ -197,6 +225,7 @@ public class KubernetesOperatorConfigOptions {
"Interval at which periodic savepoints will be triggered. "
+ "The triggering schedule is not guaranteed, savepoints will be triggered as part of the regular reconcile loop.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<String> OPERATOR_WATCHED_NAMESPACES =
ConfigOptions.key("kubernetes.operator.watched.namespaces")
.stringType()
@@ -204,19 +233,14 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Comma separated list of namespaces the operator monitors for custom resources.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Boolean> OPERATOR_DYNAMIC_NAMESPACES_ENABLED =
ConfigOptions.key("kubernetes.operator.dynamic.namespaces.enabled")
.booleanType()
.defaultValue(false)
.withDescription("Enables dynamic change of watched/monitored namespaces.");
- public static final ConfigOption<Boolean> OPERATOR_JOSDK_METRICS_ENABLED =
- ConfigOptions.key("kubernetes.operator.josdk.metrics.enabled")
- .booleanType()
- .defaultValue(true)
- .withDescription(
- "Enable forwarding of Java Operator SDK metrics to the Flink metric registry.");
-
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Duration> OPERATOR_RETRY_INITIAL_INTERVAL =
ConfigOptions.key("kubernetes.operator.retry.initial.interval")
.durationType()
@@ -224,6 +248,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Initial interval of automatic reconcile retries on recoverable errors.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Double> OPERATOR_RETRY_INTERVAL_MULTIPLIER =
ConfigOptions.key("kubernetes.operator.retry.interval.multiplier")
.doubleType()
@@ -231,6 +256,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Interval multiplier of automatic reconcile retries on recoverable errors.");
+ @Documentation.Section(SECTION_SYSTEM)
public static final ConfigOption<Integer> OPERATOR_RETRY_MAX_ATTEMPTS =
ConfigOptions.key("kubernetes.operator.retry.max.attempts")
.intType()
@@ -238,6 +264,7 @@ public class KubernetesOperatorConfigOptions {
.withDescription(
"Max attempts of automatic reconcile retries on recoverable errors.");
+ @Documentation.Section(SECTION_DYNAMIC)
public static final ConfigOption<Boolean> OPERATOR_JOB_UPGRADE_LAST_STATE_FALLBACK_ENABLED =
ConfigOptions.key("kubernetes.operator.job.upgrade.last-state-fallback.enabled")
.booleanType()
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
index 20e0f94..924fc4c 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/KubernetesOperatorMetricOptions.java
@@ -22,6 +22,14 @@ import org.apache.flink.configuration.ConfigOptions;
/** Configuration options for metrics. */
public class KubernetesOperatorMetricOptions {
+
+ public static final ConfigOption<Boolean> OPERATOR_JOSDK_METRICS_ENABLED =
+ ConfigOptions.key("kubernetes.operator.josdk.metrics.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Enable forwarding of Java Operator SDK metrics to the Flink metric registry.");
+
public static final ConfigOption<String> SCOPE_NAMING_KUBERNETES_OPERATOR =
ConfigOptions.key("metrics.scope.k8soperator.system")
.defaultValue("<host>.k8soperator.<namespace>.<name>.system")
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 401eacb..df2f9e3 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
@@ -62,7 +62,7 @@ public class FlinkOperatorWebhook {
EnvUtils.logEnvironmentInfo(LOG, "Flink Kubernetes Webhook", args);
var informerManager = new InformerManager(new DefaultKubernetesClient());
var configManager = new FlinkConfigManager(informerManager::setNamespaces);
- if (!configManager.getOperatorConfiguration().getDynamicNamespacesEnabled()) {
+ if (!configManager.getOperatorConfiguration().isDynamicNamespacesEnabled()) {
informerManager.setNamespaces(
configManager.getOperatorConfiguration().getWatchedNamespaces());
}