You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/07/11 18:25:21 UTC
[GitHub] [flink-kubernetes-operator] gyfora opened a new pull request, #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
gyfora opened a new pull request, #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311
## What is the purpose of the change
Introduce histogram metrics for tracking how long resource lifecycle state transitions take between the following states:
- CREATED
- SUSPENDED
- UPGRADING
- DEPLOYED
- STABLE
- ROLLING_BACK
- ROLLED_BACK
- FAILED
New metrics:
```
FlinkDeployment.Lifecycle.Transition.ResumeTimeSeconds: count=0, min=0, max=0, mean=NaN, ...
FlinkDeployment.Lifecycle.Transition.SuspendTimeSeconds: count=1, min=2, max=2, mean=2.0, ...
FlinkDeployment.Lifecycle.Transition.UpgradeTimeSeconds: count=1, min=33, max=33, mean=33.0, ...
FlinkDeployment.Lifecycle.Transition.StabilizationTimeSeconds: count=1, min=29, max=29, mean=29.0, ...
FlinkDeployment.Lifecycle.Transition.RollbackTimeSeconds: count=0, min=0, max=0, mean=NaN, ...
FlinkDeployment.Lifecycle.Transition.SubmissionTimeSeconds: count=1, min=1, max=1, mean=1.0, ...
FlinkDeployment.Lifecycle.State.STATE_NAME.Count: 0
```
## Brief change log
- Introduce ResourceLifecycleState derived from the resource status
- Add mechanism to track ResourceLifecycleState transitions
- Create histogram metrics for select transitions
- Add count metrics for each state
- Add tests
## Verifying this change
New unit tests + manually verified on minikube
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changes to the `CustomResourceDescriptors`: no
- Core observer or reconciler logic that is regularly executed: no
## Documentation
- Does this pull request introduce a new feature? yes
- If yes, how is the feature documented? **[TODO]**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#issuecomment-1180789758
> > What is `FlinkDeployment.Lifecycle.State.STATE_NAME.Count: 0` ?
>
> @morhidi `STATE_NAME` is a placeholder in this example. In practice it's gonna be one of CREATED, UPGRADING, DEPLOYED etc.
> > What is `FlinkDeployment.Lifecycle.State.STATE_NAME.Count: 0` ?
>
> @morhidi `STATE_NAME` is a placeholder in this example. In practice it's gonna be one of CREATED, UPGRADING, DEPLOYED etc.
nit: you can update the description to FlinkDeployment.Lifecycle.State.<STATE_NAME>.Count = ... to make it obvious
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r918274673
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final int histogramWindowSize = 1000;
Review Comment:
Don't you think it worth introducing a global window size config for the histograms across the operator?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r919678114
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ private static final String TRANSITION_FIRST_DEPLOYMENT = "FirstDeployment";
+ private static final String TRANSITION_RESUME = "Resume";
+ private static final String TRANSITION_UPGRADE = "Upgrade";
+ private static final String TRANSITION_SUSPEND = "Suspend";
+ private static final String TRANSITION_SUBMISSION = "Submission";
+ private static final String TRANSITION_STABILIZATION = "Stabilization";
+ private static final String TRANSITION_ROLLBACK = "Rollback";
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final FlinkConfigManager configManager;
+ private final Clock clock;
+ private final KubernetesOperatorMetricGroup operatorMetricGroup;
+
+ private Map<String, Tuple2<Histogram, Map<String, Histogram>>> transitionMetrics;
+ private Function<MetricGroup, MetricGroup> metricGroupFunction;
+
+ public LifecycleMetrics(
+ FlinkConfigManager configManager,
+ Clock clock,
+ KubernetesOperatorMetricGroup operatorMetricGroup) {
+ this.configManager = configManager;
+ this.clock = clock;
+ this.operatorMetricGroup = operatorMetricGroup;
+ }
+
+ public void onUpdate(CR cr) {
+ getLifecycleMetricTracker(cr).onUpdate(cr.getStatus().getLifecycleState(), clock.instant());
+ }
+
+ public void onRemove(CR cr) {
+ lifecycleTrackers.remove(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()));
+ }
+
+ private ResourceLifecycleMetricTracker getLifecycleMetricTracker(CR cr) {
+ init(cr);
+ createNamespaceStateCountIfMissing(cr.getMetadata().getNamespace());
+ return lifecycleTrackers.computeIfAbsent(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()),
+ k -> {
+ var initialState = cr.getStatus().getLifecycleState();
+ var time =
+ initialState == CREATED
+ ? Instant.parse(cr.getMetadata().getCreationTimestamp())
+ : clock.instant();
+ return new ResourceLifecycleMetricTracker(
+ initialState, time, getTransitionHistograms(cr));
+ });
+ }
+
+ private void createNamespaceStateCountIfMissing(String namespace) {
+ if (!namespaces.add(namespace)) {
+ return;
+ }
+
+ MetricGroup lifecycleGroup =
+ metricGroupFunction.apply(
+ operatorMetricGroup.createResourceNamespaceGroup(
+ configManager.getDefaultConfig(), namespace));
+ for (ResourceLifecycleState state : ResourceLifecycleState.values()) {
+ lifecycleGroup
+ .addGroup("State")
+ .addGroup(state.name())
+ .gauge(
+ "Count",
+ () ->
+ lifecycleTrackers.values().stream()
+ .map(ResourceLifecycleMetricTracker::getCurrentState)
+ .filter(s -> s == state)
+ .count());
+ }
+ }
+
+ private synchronized void init(CR cr) {
+ if (transitionMetrics != null) {
+ return;
+ }
+ this.metricGroupFunction =
+ mg -> mg.addGroup(cr.getClass().getSimpleName()).addGroup("Lifecycle");
+
+ this.transitionMetrics = new ConcurrentHashMap<>();
+ TRACKED_TRANSITIONS.forEach(
+ t ->
+ transitionMetrics.computeIfAbsent(
+ t.metricName,
+ name ->
+ Tuple2.of(
+ createTransitionHistogram(
+ name, operatorMetricGroup),
+ new ConcurrentHashMap<>())));
+ }
+
+ private Map<String, List<Histogram>> getTransitionHistograms(CR cr) {
+ var histos = new HashMap<String, List<Histogram>>();
+ transitionMetrics.forEach(
+ (metricName, t) -> {
+ histos.put(
+ metricName,
+ List.of(
+ t.f0,
+ t.f1.computeIfAbsent(
+ cr.getMetadata().getNamespace(),
+ ns ->
+ createTransitionHistogram(
+ metricName,
+ operatorMetricGroup
+ .createResourceNamespaceGroup(
+ configManager
+ .getDefaultConfig(),
+ ns)))));
+ });
+ return histos;
+ }
+
+ private Histogram createTransitionHistogram(String metricName, MetricGroup group) {
+ return metricGroupFunction
+ .apply(group)
+ .addGroup("Transition")
+ .addGroup(metricName)
+ .histogram(
+ "TimeSeconds",
+ new DescriptiveStatisticsHistogram(
+ configManager
+ .getOperatorConfiguration()
+ .getMetricsHistogramSampleSize()));
+ }
+
+ private static List<Transition> getTrackedTransitions() {
+ return List.of(
+ new Transition(CREATED, DEPLOYED, false, TRANSITION_FIRST_DEPLOYMENT),
+ new Transition(SUSPENDED, STABLE, true, TRANSITION_RESUME),
+ new Transition(STABLE, STABLE, true, TRANSITION_UPGRADE),
+ new Transition(DEPLOYED, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, STABLE, false, TRANSITION_STABILIZATION),
+ new Transition(DEPLOYED, ROLLED_BACK, false, TRANSITION_ROLLBACK),
+ new Transition(UPGRADING, DEPLOYED, true, TRANSITION_SUBMISSION),
+ new Transition(ROLLING_BACK, ROLLED_BACK, true, TRANSITION_SUBMISSION));
+ }
+
+ /**
+ * Pojo for encapsulating state transitions and whether we should measure time from the
+ * beginning of from or since the last update.
+ */
+ @ToString
+ @RequiredArgsConstructor
+ protected static class Transition {
Review Comment:
There are 12 instances of Transition currently with not very well defined names. Also they are treated in the same way, so I feel making them an enum is not a real good fit here.
Do you have a concrete suggestion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora commented on PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#issuecomment-1180808724
cc @tweise @wangyang0918
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r918282383
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final int histogramWindowSize = 1000;
Review Comment:
yes, that would be nice I agree
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r919692265
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ private static final String TRANSITION_FIRST_DEPLOYMENT = "FirstDeployment";
+ private static final String TRANSITION_RESUME = "Resume";
+ private static final String TRANSITION_UPGRADE = "Upgrade";
+ private static final String TRANSITION_SUSPEND = "Suspend";
+ private static final String TRANSITION_SUBMISSION = "Submission";
+ private static final String TRANSITION_STABILIZATION = "Stabilization";
+ private static final String TRANSITION_ROLLBACK = "Rollback";
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final FlinkConfigManager configManager;
+ private final Clock clock;
+ private final KubernetesOperatorMetricGroup operatorMetricGroup;
+
+ private Map<String, Tuple2<Histogram, Map<String, Histogram>>> transitionMetrics;
+ private Function<MetricGroup, MetricGroup> metricGroupFunction;
+
+ public LifecycleMetrics(
+ FlinkConfigManager configManager,
+ Clock clock,
+ KubernetesOperatorMetricGroup operatorMetricGroup) {
+ this.configManager = configManager;
+ this.clock = clock;
+ this.operatorMetricGroup = operatorMetricGroup;
+ }
+
+ public void onUpdate(CR cr) {
+ getLifecycleMetricTracker(cr).onUpdate(cr.getStatus().getLifecycleState(), clock.instant());
+ }
+
+ public void onRemove(CR cr) {
+ lifecycleTrackers.remove(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()));
+ }
+
+ private ResourceLifecycleMetricTracker getLifecycleMetricTracker(CR cr) {
+ init(cr);
+ createNamespaceStateCountIfMissing(cr.getMetadata().getNamespace());
+ return lifecycleTrackers.computeIfAbsent(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()),
+ k -> {
+ var initialState = cr.getStatus().getLifecycleState();
+ var time =
+ initialState == CREATED
+ ? Instant.parse(cr.getMetadata().getCreationTimestamp())
+ : clock.instant();
+ return new ResourceLifecycleMetricTracker(
+ initialState, time, getTransitionHistograms(cr));
+ });
+ }
+
+ private void createNamespaceStateCountIfMissing(String namespace) {
+ if (!namespaces.add(namespace)) {
+ return;
+ }
+
+ MetricGroup lifecycleGroup =
+ metricGroupFunction.apply(
+ operatorMetricGroup.createResourceNamespaceGroup(
+ configManager.getDefaultConfig(), namespace));
+ for (ResourceLifecycleState state : ResourceLifecycleState.values()) {
+ lifecycleGroup
+ .addGroup("State")
+ .addGroup(state.name())
+ .gauge(
+ "Count",
+ () ->
+ lifecycleTrackers.values().stream()
+ .map(ResourceLifecycleMetricTracker::getCurrentState)
+ .filter(s -> s == state)
+ .count());
+ }
+ }
+
+ private synchronized void init(CR cr) {
+ if (transitionMetrics != null) {
+ return;
+ }
+ this.metricGroupFunction =
+ mg -> mg.addGroup(cr.getClass().getSimpleName()).addGroup("Lifecycle");
+
+ this.transitionMetrics = new ConcurrentHashMap<>();
+ TRACKED_TRANSITIONS.forEach(
+ t ->
+ transitionMetrics.computeIfAbsent(
+ t.metricName,
+ name ->
+ Tuple2.of(
+ createTransitionHistogram(
+ name, operatorMetricGroup),
+ new ConcurrentHashMap<>())));
+ }
+
+ private Map<String, List<Histogram>> getTransitionHistograms(CR cr) {
+ var histos = new HashMap<String, List<Histogram>>();
+ transitionMetrics.forEach(
+ (metricName, t) -> {
+ histos.put(
+ metricName,
+ List.of(
+ t.f0,
+ t.f1.computeIfAbsent(
+ cr.getMetadata().getNamespace(),
+ ns ->
+ createTransitionHistogram(
+ metricName,
+ operatorMetricGroup
+ .createResourceNamespaceGroup(
+ configManager
+ .getDefaultConfig(),
+ ns)))));
+ });
+ return histos;
+ }
+
+ private Histogram createTransitionHistogram(String metricName, MetricGroup group) {
+ return metricGroupFunction
+ .apply(group)
+ .addGroup("Transition")
+ .addGroup(metricName)
+ .histogram(
+ "TimeSeconds",
+ new DescriptiveStatisticsHistogram(
+ configManager
+ .getOperatorConfiguration()
+ .getMetricsHistogramSampleSize()));
+ }
+
+ private static List<Transition> getTrackedTransitions() {
+ return List.of(
+ new Transition(CREATED, DEPLOYED, false, TRANSITION_FIRST_DEPLOYMENT),
+ new Transition(SUSPENDED, STABLE, true, TRANSITION_RESUME),
+ new Transition(STABLE, STABLE, true, TRANSITION_UPGRADE),
+ new Transition(DEPLOYED, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, STABLE, false, TRANSITION_STABILIZATION),
+ new Transition(DEPLOYED, ROLLED_BACK, false, TRANSITION_ROLLBACK),
+ new Transition(UPGRADING, DEPLOYED, true, TRANSITION_SUBMISSION),
+ new Transition(ROLLING_BACK, ROLLED_BACK, true, TRANSITION_SUBMISSION));
+ }
+
+ /**
+ * Pojo for encapsulating state transitions and whether we should measure time from the
+ * beginning of from or since the last update.
+ */
+ @ToString
+ @RequiredArgsConstructor
+ protected static class Transition {
Review Comment:
but I might be missing something
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r918282222
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricsTest.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.crd.spec.JobState;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.FAILED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for resource lifecycle metrics. */
+public class ResourceLifecycleMetricsTest {
+
+ @Test
+ public void lifecycleStateTest() {
+ var application = TestUtils.buildApplicationCluster();
+ assertEquals(CREATED, application.getStatus().getLifecycleState());
+
+ ReconciliationUtils.updateStatusBeforeDeploymentAttempt(application, new Configuration());
+ assertEquals(UPGRADING, application.getStatus().getLifecycleState());
+
+ ReconciliationUtils.updateStatusForDeployedSpec(application, new Configuration());
+ assertEquals(DEPLOYED, application.getStatus().getLifecycleState());
+
+ application.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
+ assertEquals(STABLE, application.getStatus().getLifecycleState());
+
+ application.getStatus().setError("errr");
+ assertEquals(STABLE, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.FAILED.name());
+ assertEquals(FAILED, application.getStatus().getLifecycleState());
+
+ application.getStatus().setError("");
+
+ application
+ .getStatus()
+ .getReconciliationStatus()
+ .setState(ReconciliationState.ROLLING_BACK);
+ assertEquals(ROLLING_BACK, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.RECONCILING.name());
+ application.getStatus().getReconciliationStatus().setState(ReconciliationState.ROLLED_BACK);
+ assertEquals(ROLLED_BACK, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.FAILED.name());
+ assertEquals(FAILED, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.RUNNING.name());
+ application.getSpec().getJob().setState(JobState.SUSPENDED);
+ ReconciliationUtils.updateStatusForDeployedSpec(application, new Configuration());
+ assertEquals(SUSPENDED, application.getStatus().getLifecycleState());
+ }
+
+ @Test
+ public void testLifecycleTracker() {
+ var histos = initHistos();
+ var lifecycleTracker =
+ new ResourceLifecycleMetricTracker(CREATED, Instant.ofEpochMilli(1000), histos);
+
+ long ts = 1000;
+ lifecycleTracker.onUpdate(UPGRADING, Instant.ofEpochMilli(ts += 1000));
Review Comment:
we could add a variable for `Instant.ofEpochMilli(ts += 1000)` here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#issuecomment-1180785668
I love this PR @gyfora added some minor comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] tweise commented on pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
tweise commented on PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#issuecomment-1182028303
This is great and maybe also mention (docs!) that you are not only tracking transitions between adjacent states and how this can be used to track the upgrade latency directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r918277417
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final int histogramWindowSize = 1000;
+ private final Configuration configuration;
+ private final Clock clock;
+ private final KubernetesOperatorMetricGroup operatorMetricGroup;
+
+ private Map<String, Tuple2<Histogram, Map<String, Histogram>>> transitionMetrics;
+ private Function<MetricGroup, MetricGroup> metricGroupFunction;
+
+ public LifecycleMetrics(
+ Configuration configuration,
+ Clock clock,
+ KubernetesOperatorMetricGroup operatorMetricGroup) {
+ this.configuration = configuration;
+ this.clock = clock;
+ this.operatorMetricGroup = operatorMetricGroup;
+ }
+
+ public void onUpdate(CR cr) {
+ getLifecycleMetricTracker(cr).onUpdate(cr.getStatus().getLifecycleState(), clock.instant());
+ }
+
+ public void onRemove(CR cr) {
+ lifecycleTrackers.remove(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()));
+ }
+
+ private ResourceLifecycleMetricTracker getLifecycleMetricTracker(CR cr) {
+ init(cr);
+ createNamespaceStateCountIfMissing(cr.getMetadata().getNamespace());
+ return lifecycleTrackers.computeIfAbsent(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()),
+ k -> {
+ var initialState = cr.getStatus().getLifecycleState();
+ var time =
+ initialState == CREATED
+ ? Instant.parse(cr.getMetadata().getCreationTimestamp())
+ : clock.instant();
+ return new ResourceLifecycleMetricTracker(
+ initialState, time, getTransitionHistograms(cr));
+ });
+ }
+
+ private void createNamespaceStateCountIfMissing(String namespace) {
+ if (!namespaces.add(namespace)) {
+ return;
+ }
+
+ MetricGroup lifecycleGroup =
+ metricGroupFunction.apply(
+ operatorMetricGroup.createResourceNamespaceGroup(configuration, namespace));
+ for (ResourceLifecycleState state : ResourceLifecycleState.values()) {
+ lifecycleGroup
+ .addGroup("State")
+ .addGroup(state.name())
+ .gauge(
+ "Count",
+ () ->
+ lifecycleTrackers.values().stream()
+ .map(ResourceLifecycleMetricTracker::getCurrentState)
+ .filter(s -> s == state)
+ .count());
+ }
+ }
+
+ private synchronized void init(CR cr) {
+ if (transitionMetrics != null) {
+ return;
+ }
+ this.metricGroupFunction =
Review Comment:
We should follow this naming convention in other metrics too, i'll add this to my open PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r918272226
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java:
##########
@@ -18,31 +18,37 @@
package org.apache.flink.kubernetes.operator.metrics;
import org.apache.flink.configuration.Configuration;
+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.metrics.lifecycle.LifecycleMetrics;
-import io.fabric8.kubernetes.client.CustomResource;
-
+import java.time.Clock;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/** Metric manager for Operator managed custom resources. */
-public class MetricManager<CR extends CustomResource<?, ?>> {
+public class MetricManager<CR extends AbstractFlinkResource<?, ?>> {
Review Comment:
Shall we call it `ResourceMetricManager`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r918283173
##########
flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/ResourceLifecycleMetricsTest.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.TestUtils;
+import org.apache.flink.kubernetes.operator.crd.spec.JobState;
+import org.apache.flink.kubernetes.operator.crd.status.ReconciliationState;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.FAILED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for resource lifecycle metrics. */
+public class ResourceLifecycleMetricsTest {
+
+ @Test
+ public void lifecycleStateTest() {
+ var application = TestUtils.buildApplicationCluster();
+ assertEquals(CREATED, application.getStatus().getLifecycleState());
+
+ ReconciliationUtils.updateStatusBeforeDeploymentAttempt(application, new Configuration());
+ assertEquals(UPGRADING, application.getStatus().getLifecycleState());
+
+ ReconciliationUtils.updateStatusForDeployedSpec(application, new Configuration());
+ assertEquals(DEPLOYED, application.getStatus().getLifecycleState());
+
+ application.getStatus().getReconciliationStatus().markReconciledSpecAsStable();
+ assertEquals(STABLE, application.getStatus().getLifecycleState());
+
+ application.getStatus().setError("errr");
+ assertEquals(STABLE, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.FAILED.name());
+ assertEquals(FAILED, application.getStatus().getLifecycleState());
+
+ application.getStatus().setError("");
+
+ application
+ .getStatus()
+ .getReconciliationStatus()
+ .setState(ReconciliationState.ROLLING_BACK);
+ assertEquals(ROLLING_BACK, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.RECONCILING.name());
+ application.getStatus().getReconciliationStatus().setState(ReconciliationState.ROLLED_BACK);
+ assertEquals(ROLLED_BACK, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.FAILED.name());
+ assertEquals(FAILED, application.getStatus().getLifecycleState());
+
+ application.getStatus().getJobStatus().setState(JobStatus.RUNNING.name());
+ application.getSpec().getJob().setState(JobState.SUSPENDED);
+ ReconciliationUtils.updateStatusForDeployedSpec(application, new Configuration());
+ assertEquals(SUSPENDED, application.getStatus().getLifecycleState());
+ }
+
+ @Test
+ public void testLifecycleTracker() {
+ var histos = initHistos();
+ var lifecycleTracker =
+ new ResourceLifecycleMetricTracker(CREATED, Instant.ofEpochMilli(1000), histos);
+
+ long ts = 1000;
+ lifecycleTracker.onUpdate(UPGRADING, Instant.ofEpochMilli(ts += 1000));
Review Comment:
The ts is always increasing this way, I think we cannot easily simplify this. Also it's pretty simple already :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r918282131
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/MetricManager.java:
##########
@@ -18,31 +18,37 @@
package org.apache.flink.kubernetes.operator.metrics;
import org.apache.flink.configuration.Configuration;
+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.metrics.lifecycle.LifecycleMetrics;
-import io.fabric8.kubernetes.client.CustomResource;
-
+import java.time.Clock;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/** Metric manager for Operator managed custom resources. */
-public class MetricManager<CR extends CustomResource<?, ?>> {
+public class MetricManager<CR extends AbstractFlinkResource<?, ?>> {
Review Comment:
I think the name is fine for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r919814787
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ private static final String TRANSITION_FIRST_DEPLOYMENT = "FirstDeployment";
+ private static final String TRANSITION_RESUME = "Resume";
+ private static final String TRANSITION_UPGRADE = "Upgrade";
+ private static final String TRANSITION_SUSPEND = "Suspend";
+ private static final String TRANSITION_SUBMISSION = "Submission";
+ private static final String TRANSITION_STABILIZATION = "Stabilization";
+ private static final String TRANSITION_ROLLBACK = "Rollback";
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final FlinkConfigManager configManager;
+ private final Clock clock;
+ private final KubernetesOperatorMetricGroup operatorMetricGroup;
+
+ private Map<String, Tuple2<Histogram, Map<String, Histogram>>> transitionMetrics;
+ private Function<MetricGroup, MetricGroup> metricGroupFunction;
+
+ public LifecycleMetrics(
+ FlinkConfigManager configManager,
+ Clock clock,
+ KubernetesOperatorMetricGroup operatorMetricGroup) {
+ this.configManager = configManager;
+ this.clock = clock;
+ this.operatorMetricGroup = operatorMetricGroup;
+ }
+
+ public void onUpdate(CR cr) {
+ getLifecycleMetricTracker(cr).onUpdate(cr.getStatus().getLifecycleState(), clock.instant());
+ }
+
+ public void onRemove(CR cr) {
+ lifecycleTrackers.remove(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()));
+ }
+
+ private ResourceLifecycleMetricTracker getLifecycleMetricTracker(CR cr) {
+ init(cr);
+ createNamespaceStateCountIfMissing(cr.getMetadata().getNamespace());
+ return lifecycleTrackers.computeIfAbsent(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()),
+ k -> {
+ var initialState = cr.getStatus().getLifecycleState();
+ var time =
+ initialState == CREATED
+ ? Instant.parse(cr.getMetadata().getCreationTimestamp())
+ : clock.instant();
+ return new ResourceLifecycleMetricTracker(
+ initialState, time, getTransitionHistograms(cr));
+ });
+ }
+
+ private void createNamespaceStateCountIfMissing(String namespace) {
+ if (!namespaces.add(namespace)) {
+ return;
+ }
+
+ MetricGroup lifecycleGroup =
+ metricGroupFunction.apply(
+ operatorMetricGroup.createResourceNamespaceGroup(
+ configManager.getDefaultConfig(), namespace));
+ for (ResourceLifecycleState state : ResourceLifecycleState.values()) {
+ lifecycleGroup
+ .addGroup("State")
+ .addGroup(state.name())
+ .gauge(
+ "Count",
+ () ->
+ lifecycleTrackers.values().stream()
+ .map(ResourceLifecycleMetricTracker::getCurrentState)
+ .filter(s -> s == state)
+ .count());
+ }
+ }
+
+ private synchronized void init(CR cr) {
+ if (transitionMetrics != null) {
+ return;
+ }
+ this.metricGroupFunction =
+ mg -> mg.addGroup(cr.getClass().getSimpleName()).addGroup("Lifecycle");
+
+ this.transitionMetrics = new ConcurrentHashMap<>();
+ TRACKED_TRANSITIONS.forEach(
+ t ->
+ transitionMetrics.computeIfAbsent(
+ t.metricName,
+ name ->
+ Tuple2.of(
+ createTransitionHistogram(
+ name, operatorMetricGroup),
+ new ConcurrentHashMap<>())));
+ }
+
+ private Map<String, List<Histogram>> getTransitionHistograms(CR cr) {
+ var histos = new HashMap<String, List<Histogram>>();
+ transitionMetrics.forEach(
+ (metricName, t) -> {
+ histos.put(
+ metricName,
+ List.of(
+ t.f0,
+ t.f1.computeIfAbsent(
+ cr.getMetadata().getNamespace(),
+ ns ->
+ createTransitionHistogram(
+ metricName,
+ operatorMetricGroup
+ .createResourceNamespaceGroup(
+ configManager
+ .getDefaultConfig(),
+ ns)))));
+ });
+ return histos;
+ }
+
+ private Histogram createTransitionHistogram(String metricName, MetricGroup group) {
+ return metricGroupFunction
+ .apply(group)
+ .addGroup("Transition")
+ .addGroup(metricName)
+ .histogram(
+ "TimeSeconds",
+ new DescriptiveStatisticsHistogram(
+ configManager
+ .getOperatorConfiguration()
+ .getMetricsHistogramSampleSize()));
+ }
+
+ private static List<Transition> getTrackedTransitions() {
+ return List.of(
+ new Transition(CREATED, DEPLOYED, false, TRANSITION_FIRST_DEPLOYMENT),
+ new Transition(SUSPENDED, STABLE, true, TRANSITION_RESUME),
+ new Transition(STABLE, STABLE, true, TRANSITION_UPGRADE),
+ new Transition(DEPLOYED, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, STABLE, false, TRANSITION_STABILIZATION),
+ new Transition(DEPLOYED, ROLLED_BACK, false, TRANSITION_ROLLBACK),
+ new Transition(UPGRADING, DEPLOYED, true, TRANSITION_SUBMISSION),
+ new Transition(ROLLING_BACK, ROLLED_BACK, true, TRANSITION_SUBMISSION));
+ }
+
+ /**
+ * Pojo for encapsulating state transitions and whether we should measure time from the
+ * beginning of from or since the last update.
+ */
+ @ToString
+ @RequiredArgsConstructor
+ protected static class Transition {
Review Comment:
Enums only make sense in cases where the different values have different meanings. (and that actually means you have logic built around them)
Transition objects are handled uniformly, you can add / remove new ones later it won't change anything.
I think enum is not a good fit here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora merged pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora merged PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#issuecomment-1180771363
What is `FlinkDeployment.Lifecycle.State.STATE_NAME.Count: 0` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] gyfora commented on pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
gyfora commented on PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#issuecomment-1180786814
> What is `FlinkDeployment.Lifecycle.State.STATE_NAME.Count: 0` ?
@morhidi `STATE_NAME` is a placeholder in this example. In practice it's gonna be one of CREATED, UPGRADING, DEPLOYED etc.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r919663834
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ private static final String TRANSITION_FIRST_DEPLOYMENT = "FirstDeployment";
+ private static final String TRANSITION_RESUME = "Resume";
+ private static final String TRANSITION_UPGRADE = "Upgrade";
+ private static final String TRANSITION_SUSPEND = "Suspend";
+ private static final String TRANSITION_SUBMISSION = "Submission";
+ private static final String TRANSITION_STABILIZATION = "Stabilization";
+ private static final String TRANSITION_ROLLBACK = "Rollback";
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final FlinkConfigManager configManager;
+ private final Clock clock;
+ private final KubernetesOperatorMetricGroup operatorMetricGroup;
+
+ private Map<String, Tuple2<Histogram, Map<String, Histogram>>> transitionMetrics;
+ private Function<MetricGroup, MetricGroup> metricGroupFunction;
+
+ public LifecycleMetrics(
+ FlinkConfigManager configManager,
+ Clock clock,
+ KubernetesOperatorMetricGroup operatorMetricGroup) {
+ this.configManager = configManager;
+ this.clock = clock;
+ this.operatorMetricGroup = operatorMetricGroup;
+ }
+
+ public void onUpdate(CR cr) {
+ getLifecycleMetricTracker(cr).onUpdate(cr.getStatus().getLifecycleState(), clock.instant());
+ }
+
+ public void onRemove(CR cr) {
+ lifecycleTrackers.remove(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()));
+ }
+
+ private ResourceLifecycleMetricTracker getLifecycleMetricTracker(CR cr) {
+ init(cr);
+ createNamespaceStateCountIfMissing(cr.getMetadata().getNamespace());
+ return lifecycleTrackers.computeIfAbsent(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()),
+ k -> {
+ var initialState = cr.getStatus().getLifecycleState();
+ var time =
+ initialState == CREATED
+ ? Instant.parse(cr.getMetadata().getCreationTimestamp())
+ : clock.instant();
+ return new ResourceLifecycleMetricTracker(
+ initialState, time, getTransitionHistograms(cr));
+ });
+ }
+
+ private void createNamespaceStateCountIfMissing(String namespace) {
+ if (!namespaces.add(namespace)) {
+ return;
+ }
+
+ MetricGroup lifecycleGroup =
+ metricGroupFunction.apply(
+ operatorMetricGroup.createResourceNamespaceGroup(
+ configManager.getDefaultConfig(), namespace));
+ for (ResourceLifecycleState state : ResourceLifecycleState.values()) {
+ lifecycleGroup
+ .addGroup("State")
+ .addGroup(state.name())
+ .gauge(
+ "Count",
+ () ->
+ lifecycleTrackers.values().stream()
+ .map(ResourceLifecycleMetricTracker::getCurrentState)
+ .filter(s -> s == state)
+ .count());
+ }
+ }
+
+ private synchronized void init(CR cr) {
+ if (transitionMetrics != null) {
+ return;
+ }
+ this.metricGroupFunction =
+ mg -> mg.addGroup(cr.getClass().getSimpleName()).addGroup("Lifecycle");
+
+ this.transitionMetrics = new ConcurrentHashMap<>();
+ TRACKED_TRANSITIONS.forEach(
+ t ->
+ transitionMetrics.computeIfAbsent(
+ t.metricName,
+ name ->
+ Tuple2.of(
+ createTransitionHistogram(
+ name, operatorMetricGroup),
+ new ConcurrentHashMap<>())));
+ }
+
+ private Map<String, List<Histogram>> getTransitionHistograms(CR cr) {
+ var histos = new HashMap<String, List<Histogram>>();
+ transitionMetrics.forEach(
+ (metricName, t) -> {
+ histos.put(
+ metricName,
+ List.of(
+ t.f0,
+ t.f1.computeIfAbsent(
+ cr.getMetadata().getNamespace(),
+ ns ->
+ createTransitionHistogram(
+ metricName,
+ operatorMetricGroup
+ .createResourceNamespaceGroup(
+ configManager
+ .getDefaultConfig(),
+ ns)))));
+ });
+ return histos;
+ }
+
+ private Histogram createTransitionHistogram(String metricName, MetricGroup group) {
+ return metricGroupFunction
+ .apply(group)
+ .addGroup("Transition")
+ .addGroup(metricName)
+ .histogram(
+ "TimeSeconds",
+ new DescriptiveStatisticsHistogram(
+ configManager
+ .getOperatorConfiguration()
+ .getMetricsHistogramSampleSize()));
+ }
+
+ private static List<Transition> getTrackedTransitions() {
+ return List.of(
+ new Transition(CREATED, DEPLOYED, false, TRANSITION_FIRST_DEPLOYMENT),
+ new Transition(SUSPENDED, STABLE, true, TRANSITION_RESUME),
+ new Transition(STABLE, STABLE, true, TRANSITION_UPGRADE),
+ new Transition(DEPLOYED, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, STABLE, false, TRANSITION_STABILIZATION),
+ new Transition(DEPLOYED, ROLLED_BACK, false, TRANSITION_ROLLBACK),
+ new Transition(UPGRADING, DEPLOYED, true, TRANSITION_SUBMISSION),
+ new Transition(ROLLING_BACK, ROLLED_BACK, true, TRANSITION_SUBMISSION));
+ }
+
+ /**
+ * Pojo for encapsulating state transitions and whether we should measure time from the
+ * beginning of from or since the last update.
+ */
+ @ToString
+ @RequiredArgsConstructor
+ protected static class Transition {
Review Comment:
Have you considered defining `Transition` as enum?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-kubernetes-operator] morhidi commented on a diff in pull request #311: [FLINK-28479] Add metrics for resource lifecycle state transitions
Posted by GitBox <gi...@apache.org>.
morhidi commented on code in PR #311:
URL: https://github.com/apache/flink-kubernetes-operator/pull/311#discussion_r919691516
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/lifecycle/LifecycleMetrics.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.lifecycle;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.crd.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.metrics.KubernetesOperatorMetricGroup;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;
+
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Function;
+
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.CREATED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.DEPLOYED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLED_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.ROLLING_BACK;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.STABLE;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.SUSPENDED;
+import static org.apache.flink.kubernetes.operator.metrics.lifecycle.ResourceLifecycleState.UPGRADING;
+
+/**
+ * Utility for tracking resource lifecycle metrics globally and per namespace.
+ *
+ * @param <CR> Flink resource type.
+ */
+public class LifecycleMetrics<CR extends AbstractFlinkResource<?, ?>> {
+
+ private static final String TRANSITION_FIRST_DEPLOYMENT = "FirstDeployment";
+ private static final String TRANSITION_RESUME = "Resume";
+ private static final String TRANSITION_UPGRADE = "Upgrade";
+ private static final String TRANSITION_SUSPEND = "Suspend";
+ private static final String TRANSITION_SUBMISSION = "Submission";
+ private static final String TRANSITION_STABILIZATION = "Stabilization";
+ private static final String TRANSITION_ROLLBACK = "Rollback";
+
+ public static final List<Transition> TRACKED_TRANSITIONS = getTrackedTransitions();
+
+ private final Map<Tuple2<String, String>, ResourceLifecycleMetricTracker> lifecycleTrackers =
+ new ConcurrentHashMap<>();
+ private final Set<String> namespaces = Collections.newSetFromMap(new ConcurrentHashMap<>());
+
+ private final FlinkConfigManager configManager;
+ private final Clock clock;
+ private final KubernetesOperatorMetricGroup operatorMetricGroup;
+
+ private Map<String, Tuple2<Histogram, Map<String, Histogram>>> transitionMetrics;
+ private Function<MetricGroup, MetricGroup> metricGroupFunction;
+
+ public LifecycleMetrics(
+ FlinkConfigManager configManager,
+ Clock clock,
+ KubernetesOperatorMetricGroup operatorMetricGroup) {
+ this.configManager = configManager;
+ this.clock = clock;
+ this.operatorMetricGroup = operatorMetricGroup;
+ }
+
+ public void onUpdate(CR cr) {
+ getLifecycleMetricTracker(cr).onUpdate(cr.getStatus().getLifecycleState(), clock.instant());
+ }
+
+ public void onRemove(CR cr) {
+ lifecycleTrackers.remove(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()));
+ }
+
+ private ResourceLifecycleMetricTracker getLifecycleMetricTracker(CR cr) {
+ init(cr);
+ createNamespaceStateCountIfMissing(cr.getMetadata().getNamespace());
+ return lifecycleTrackers.computeIfAbsent(
+ Tuple2.of(cr.getMetadata().getNamespace(), cr.getMetadata().getName()),
+ k -> {
+ var initialState = cr.getStatus().getLifecycleState();
+ var time =
+ initialState == CREATED
+ ? Instant.parse(cr.getMetadata().getCreationTimestamp())
+ : clock.instant();
+ return new ResourceLifecycleMetricTracker(
+ initialState, time, getTransitionHistograms(cr));
+ });
+ }
+
+ private void createNamespaceStateCountIfMissing(String namespace) {
+ if (!namespaces.add(namespace)) {
+ return;
+ }
+
+ MetricGroup lifecycleGroup =
+ metricGroupFunction.apply(
+ operatorMetricGroup.createResourceNamespaceGroup(
+ configManager.getDefaultConfig(), namespace));
+ for (ResourceLifecycleState state : ResourceLifecycleState.values()) {
+ lifecycleGroup
+ .addGroup("State")
+ .addGroup(state.name())
+ .gauge(
+ "Count",
+ () ->
+ lifecycleTrackers.values().stream()
+ .map(ResourceLifecycleMetricTracker::getCurrentState)
+ .filter(s -> s == state)
+ .count());
+ }
+ }
+
+ private synchronized void init(CR cr) {
+ if (transitionMetrics != null) {
+ return;
+ }
+ this.metricGroupFunction =
+ mg -> mg.addGroup(cr.getClass().getSimpleName()).addGroup("Lifecycle");
+
+ this.transitionMetrics = new ConcurrentHashMap<>();
+ TRACKED_TRANSITIONS.forEach(
+ t ->
+ transitionMetrics.computeIfAbsent(
+ t.metricName,
+ name ->
+ Tuple2.of(
+ createTransitionHistogram(
+ name, operatorMetricGroup),
+ new ConcurrentHashMap<>())));
+ }
+
+ private Map<String, List<Histogram>> getTransitionHistograms(CR cr) {
+ var histos = new HashMap<String, List<Histogram>>();
+ transitionMetrics.forEach(
+ (metricName, t) -> {
+ histos.put(
+ metricName,
+ List.of(
+ t.f0,
+ t.f1.computeIfAbsent(
+ cr.getMetadata().getNamespace(),
+ ns ->
+ createTransitionHistogram(
+ metricName,
+ operatorMetricGroup
+ .createResourceNamespaceGroup(
+ configManager
+ .getDefaultConfig(),
+ ns)))));
+ });
+ return histos;
+ }
+
+ private Histogram createTransitionHistogram(String metricName, MetricGroup group) {
+ return metricGroupFunction
+ .apply(group)
+ .addGroup("Transition")
+ .addGroup(metricName)
+ .histogram(
+ "TimeSeconds",
+ new DescriptiveStatisticsHistogram(
+ configManager
+ .getOperatorConfiguration()
+ .getMetricsHistogramSampleSize()));
+ }
+
+ private static List<Transition> getTrackedTransitions() {
+ return List.of(
+ new Transition(CREATED, DEPLOYED, false, TRANSITION_FIRST_DEPLOYMENT),
+ new Transition(SUSPENDED, STABLE, true, TRANSITION_RESUME),
+ new Transition(STABLE, STABLE, true, TRANSITION_UPGRADE),
+ new Transition(DEPLOYED, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, UPGRADING, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(STABLE, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(ROLLED_BACK, SUSPENDED, true, TRANSITION_SUSPEND),
+ new Transition(DEPLOYED, STABLE, false, TRANSITION_STABILIZATION),
+ new Transition(DEPLOYED, ROLLED_BACK, false, TRANSITION_ROLLBACK),
+ new Transition(UPGRADING, DEPLOYED, true, TRANSITION_SUBMISSION),
+ new Transition(ROLLING_BACK, ROLLED_BACK, true, TRANSITION_SUBMISSION));
+ }
+
+ /**
+ * Pojo for encapsulating state transitions and whether we should measure time from the
+ * beginning of from or since the last update.
+ */
+ @ToString
+ @RequiredArgsConstructor
+ protected static class Transition {
Review Comment:
We always measure the latency between two states, so it could be STABLE_STABLE, SUPENDED_STABLE, etc. The last param shows which metric they belong to, not really which transition.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org