You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "mateczagany (via GitHub)" <gi...@apache.org> on 2024/04/29 11:53:00 UTC
[PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
mateczagany opened a new pull request, #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821
This PR is still in early development, please keep that in mind.
Notes:
- In case of a periodic snapshots, the Operator will create new `FlinkStateSnapshot` CRs, and the snapshot will be taken when that resource is reconciled. Its labels are not final yet.
- In case of upgrade snapshots, the Operator will create a new `FlinkStateSnapshot` CR, marking it with `alreadyExists`, this CR won't be reconciled later.
TODO:
- Re-visit `EventRecorder` class and new method
- Implement initial snapshot reference
- Refactor SnapshotUtils to make its methods mostly re-usable in new logic
- FlinkStateSnapshot metrics
- Unit tests
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688221106
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkStateSnapshotState.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.api.status;
+
+import org.apache.flink.annotation.Experimental;
+
+/** Describes current snapshot state. */
+@Experimental
+public enum FlinkStateSnapshotState {
Review Comment:
Good idea, thank you!
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "anupamaggarwal (via GitHub)" <gi...@apache.org>.
anupamaggarwal commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1691872393
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.SnapshotUtils;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+
+ FlinkStateSnapshotUtils.snapshotSuccessful(
+ resource, resource.getSpec().getSavepoint().getPath(), true);
+ return;
+ }
+
+ if (FlinkStateSnapshotUtils.abandonSnapshotIfJobNotRunning(
+ ctx.getKubernetesClient(),
+ ctx.getResource(),
+ ctx.getSecondaryResource().orElse(null),
+ eventRecorder)) {
+ return;
+ }
+
+ var jobId = ctx.getSecondaryResource().orElseThrow().getStatus().getJobStatus().getJobId();
+
+ Optional<String> triggerIdOpt;
+ try {
+ triggerIdOpt = triggerCheckpointOrSavepoint(resource.getSpec(), ctx, jobId);
+ } catch (Exception e) {
+ LOG.error("Failed to trigger snapshot for resource {}", ctx.getResource(), e);
+ throw new ReconciliationException(e);
+ }
+
+ if (triggerIdOpt.isEmpty()) {
+ LOG.warn("Failed to trigger snapshot {}", resource.getMetadata().getName());
+ return;
+ }
+
+ FlinkStateSnapshotUtils.snapshotInProgress(resource, triggerIdOpt.get());
+ }
+
+ public DeleteControl cleanup(FlinkStateSnapshotContext ctx) throws Exception {
+ var resource = ctx.getResource();
+ var state = resource.getStatus().getState();
+ var resourceName = resource.getMetadata().getName();
+ LOG.info("Cleaning up resource {}...", resourceName);
+
+ if (resource.getSpec().isCheckpoint()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (!resource.getSpec().getSavepoint().getDisposeOnDelete()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (resource.getSpec().getJobReference() == null
+ || resource.getSpec().getJobReference().getName() == null) {
+ return DeleteControl.defaultDelete();
+ }
+
+ switch (state) {
+ case IN_PROGRESS:
+ LOG.info(
+ "Cannot delete resource {} yet as savepoint is still in progress...",
+ resourceName);
+ return DeleteControl.noFinalizerRemoval()
+ .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ case COMPLETED:
+ var flinkDeployment = getFlinkDeployment(ctx);
+ return handleSnapshotCleanup(resource, flinkDeployment, ctx);
Review Comment:
I was trying out the periodic snapshot feature today. Does it make sense to introduce a feature to auto-cleanup the `flinkstatesnapshots.flink.apache.org` CRs and only retain upto some max retention limit (we could still retain the savepoint dir?) IIUC `disposeOnDelete` for periodic savepoints is set to false by default (so it should be relatively safe to do so).
If the user sets the savepoint interval `kubernetes.operator.periodic.savepoint.interval` to a low value, and there are multiple jobs on the cluster it would lead to creation of many CRs. I am not familiar with K8s internals but could this increase the load on the API server?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getValidatedFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (StringUtils.isNotBlank(snapshotRef.getPath())) {
+ return snapshotRef.getPath();
+ }
+
+ if (StringUtils.isBlank(snapshotRef.getName())) {
+ throw new IllegalArgumentException(
+ String.format("Invalid snapshot name: %s", snapshotRef.getName()));
+ }
+
+ var result =
+ snapshotRef.getNamespace() == null
+ ? kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get()
+ : kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .inNamespace(snapshotRef.getNamespace())
+ .withName(snapshotRef.getName())
+ .get();
+
+ if (result == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot find snapshot %s in namespace %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ // We can return the savepoint path if it's marked as completed without waiting for the
+ // reconciler to update its status.
+ if (result.getSpec().isSavepoint() && result.getSpec().getSavepoint().getAlreadyExists()) {
+ var path = result.getSpec().getSavepoint().getPath();
+ if (!StringUtils.isBlank(path)) {
+ return path;
+ }
+ }
+
+ if (COMPLETED != result.getStatus().getState()) {
+ throw new IllegalStateException(
+ String.format(
+ "Snapshot %s/%s is not complete yet.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ var path = result.getStatus().getPath();
+ if (StringUtils.isBlank(path)) {
+ throw new IllegalStateException(
+ String.format(
+ "Snapshot %s/%s path is incorrect: %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName(), path));
+ }
+
+ return path;
+ }
+
+ protected static FlinkStateSnapshot createFlinkStateSnapshot(
+ KubernetesClient kubernetesClient,
+ String name,
+ FlinkStateSnapshotSpec spec,
+ SnapshotTriggerType triggerType) {
+ var metadata = new ObjectMeta();
+ metadata.setName(name);
+ metadata.getLabels().put(CrdConstants.LABEL_SNAPSHOT_TYPE, triggerType.name());
+
+ var snapshot = new FlinkStateSnapshot();
+ snapshot.setSpec(spec);
+ snapshot.setMetadata(metadata);
+
+ return kubernetesClient.resource(snapshot).create();
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param savepointPath savepoint path if any
+ * @param triggerType trigger type
+ * @param savepointFormatType format type
+ * @param disposeOnDelete should dispose of data on deletion
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createSavepointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ @Nullable String savepointPath,
+ SnapshotTriggerType triggerType,
+ SavepointFormatType savepointFormatType,
+ boolean disposeOnDelete) {
+ var savepointSpec =
+ SavepointSpec.builder()
+ .path(savepointPath)
+ .formatType(savepointFormatType)
+ .disposeOnDelete(disposeOnDelete)
+ .alreadyExists(triggerType == SnapshotTriggerType.UPGRADE)
+ .build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .savepoint(savepointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(SAVEPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param checkpointType type of checkpoint
+ * @param triggerType trigger type
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createCheckpointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ CheckpointType checkpointType,
+ SnapshotTriggerType triggerType) {
+ var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .checkpoint(checkpointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(CHECKPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Based on job configuration and operator configuration, decide if {@link FlinkStateSnapshot}
+ * resources should be used or not. Operator configuration will disable the usage of the
+ * corresponding CRD was not installed on this Kubernetes cluster.
+ *
+ * @param operatorConfiguration operator config
+ * @param configuration job config
+ * @return true if snapshot resources should be created
+ */
+ public static boolean isSnapshotResourceEnabled(
+ FlinkOperatorConfiguration operatorConfiguration, Configuration configuration) {
+ return configuration.get(SNAPSHOT_RESOURCE_ENABLED)
+ && operatorConfiguration.isSnapshotResourcesEnabled();
+ }
+
+ /**
+ * Return a generated name for a {@link FlinkStateSnapshot} to be created.
+ *
+ * @param snapshotType type of snapshot
+ * @param triggerType trigger type of snapshot
+ * @param referencedResource referenced resource
+ * @return result name
+ */
+ public static String getFlinkStateSnapshotName(
+ SnapshotType snapshotType,
+ SnapshotTriggerType triggerType,
+ AbstractFlinkResource<?, ?> referencedResource) {
+ return String.format(
+ "%s-%s-%s-%s",
+ referencedResource.getMetadata().getName(),
+ snapshotType.name().toLowerCase(),
+ triggerType.name().toLowerCase(),
+ UUID.randomUUID());
Review Comment:
Hi @mateczagany, wdyt about having a timestamp / epoch value instead of the UUID here? It might help users figure out the latest snapshot CR in case they do a
`kubectl get flinkstatesnapshots.flink.apache.org` (and not have to pass in `--sort-by=.metadata.creationTimestamp` explicitly) ?
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667645251
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -266,19 +304,31 @@ protected void restoreJob(
Optional<String> savepointOpt = Optional.empty();
if (spec.getJob().getUpgradeMode() != UpgradeMode.STATELESS) {
- savepointOpt =
- Optional.ofNullable(
- ctx.getResource()
- .getStatus()
- .getJobStatus()
- .getSavepointInfo()
- .getLastSavepoint())
- .flatMap(s -> Optional.ofNullable(s.getLocation()));
+ if (FlinkStateSnapshotUtils.shouldCreateSnapshotResource(
+ ctx.getOperatorConfig(), deployConfig)) {
+ savepointOpt = getLatestSavepointPathFromFlinkStateSnapshots(ctx);
+ } else {
+ savepointOpt =
+ Optional.ofNullable(
+ ctx.getResource()
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint())
+ .flatMap(s -> Optional.ofNullable(s.getLocation()));
Review Comment:
Actually I think the cleanest / simplest thing to do would be to add a new status field like:
```
snapshotReference / upgradeSnapshotReference
```
This should be updated during SAVEPOINT deployments (when we start from savepoint/checkpoint), after savepoint cancellation or when observing the last state of a terminal job. (basically every time we currently update the lastSavepoint for upgrade purposes
)
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1671974248
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSnapshot(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ reconciler.reconcile(ctx);
+ } catch (Exception e) {
+ FlinkStateSnapshotUtils.snapshotFailed(
+ josdkContext.getClient(), eventRecorder, flinkStateSnapshot, e.getMessage());
+ LOG.error("Failed to reconcile {}", flinkStateSnapshot, e);
Review Comment:
This path threw `ReconciliationException` before commit https://github.com/apache/flink-kubernetes-operator/pull/821/commits/0d2260aa206ea3ccb1b33a5b54e49eb5b1e75205 but instead changing the state to `FAILED` here, then letting `getUpdateControl()` decide if we have reached the backoff limit and possibly setting the state back to `TRIGGER_PENDING` seems like a better solution.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSnapshot(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ reconciler.reconcile(ctx);
+ } catch (Exception e) {
+ FlinkStateSnapshotUtils.snapshotFailed(
+ josdkContext.getClient(), eventRecorder, flinkStateSnapshot, e.getMessage());
+ LOG.error("Failed to reconcile {}", flinkStateSnapshot, e);
Review Comment:
This codepath threw `ReconciliationException` before commit https://github.com/apache/flink-kubernetes-operator/pull/821/commits/0d2260aa206ea3ccb1b33a5b54e49eb5b1e75205 but instead changing the state to `FAILED` here, then letting `getUpdateControl()` decide if we have reached the backoff limit and possibly setting the state back to `TRIGGER_PENDING` seems like a better solution.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667757174
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -266,19 +304,31 @@ protected void restoreJob(
Optional<String> savepointOpt = Optional.empty();
if (spec.getJob().getUpgradeMode() != UpgradeMode.STATELESS) {
- savepointOpt =
- Optional.ofNullable(
- ctx.getResource()
- .getStatus()
- .getJobStatus()
- .getSavepointInfo()
- .getLastSavepoint())
- .flatMap(s -> Optional.ofNullable(s.getLocation()));
+ if (FlinkStateSnapshotUtils.shouldCreateSnapshotResource(
+ ctx.getOperatorConfig(), deployConfig)) {
+ savepointOpt = getLatestSavepointPathFromFlinkStateSnapshots(ctx);
+ } else {
+ savepointOpt =
+ Optional.ofNullable(
+ ctx.getResource()
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint())
+ .flatMap(s -> Optional.ofNullable(s.getLocation()));
Review Comment:
@mateczagany I think the only good solution is to introduce a separate `upgradeSnapshotReference` that is only updated during the upgrade cycle (and during terminal job observe). Any snapshot observed this way is logically speaking always the latest.
A manual / periodic snapshot should never override this, doing that would be surely an error in the logic so this is absolutely not a problem. This way we will actually separate the upgrade snapshot handling from the savepoint taking / management which was kind of confused together until now.
In other words we don't need any of the "3 solutions" :)
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688126716
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ // status might be null here
+ flinkStateSnapshot.setStatus(
+ Objects.requireNonNullElseGet(
+ flinkStateSnapshot.getStatus(), FlinkStateSnapshotStatus::new));
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ observer.observe(ctx);
+
+ if (validateSnapshot(ctx)) {
+ reconciler.reconcile(ctx);
+ }
+
+ notifyListeners(ctx);
+ return getUpdateControl(ctx);
+ }
+
+ @Override
+ public DeleteControl cleanup(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+ DeleteControl deleteControl;
Review Comment:
Thank you, it was a leftover before a refactor
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1678990863
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSnapshot(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ reconciler.reconcile(ctx);
+ } catch (Exception e) {
+ FlinkStateSnapshotUtils.snapshotFailed(
+ josdkContext.getClient(), eventRecorder, flinkStateSnapshot, e.getMessage());
+ LOG.error("Failed to reconcile {}", flinkStateSnapshot, e);
Review Comment:
Errors are now handled by updateErrorStatus since commit 0315d1062dcf0b81e61af77be4a0c12e9330398b
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688463075
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/listener/AuditUtils.java:
##########
@@ -53,6 +62,16 @@ private static String format(@NonNull CommonStatus<?> status) {
: status.getError());
}
+ private static String format(@NonNull FlinkStateSnapshotStatus status) {
+ if (StringUtils.isEmpty(status.getError())) {
+ return String.format(
+ ">>> Status[Snapshot] | Info | %s | %s", status.getState(), status.getPath());
Review Comment:
I have added some changes for this, thank you!
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#issuecomment-2271129992
@anupamaggarwal @ferenc-csaky unless there are any concerns I would like to merge this. Afterwards we can focus on the follow up items and even get a chance to paralelnize this work
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1690995181
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -72,21 +74,26 @@ public class FlinkConfigManager {
private volatile Configuration defaultConfig;
private volatile FlinkOperatorConfiguration defaultOperatorConfiguration;
+ private final boolean snapshotCrdInstalled;
Review Comment:
Added the note, thank you. I am going to also note here so others can see:
This field will be determined in `FlinkOperator` when the operator starts, by checking if `FlinkStateSnapshot` CRD is installed on the Kubernetes cluster.
When generating the default configuration for the operator, `kubernetes.operator.snapshot.resource.enabled` will be set to false in this class if `snapshotCrdInstalled` is false, to allow users to continue using the operator if they upgraded from a previous version but did not install the new `FlinkStateSnapshot` CRD.
This check and field is planned to be removed in the release after this FLIP gets released (so hopefully 1.11 if this PR is released in 1.10).
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1690999244
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,395 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getAndValidateFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (!StringUtils.isBlank(snapshotRef.getPath())) {
+ return snapshotRef.getPath();
+ }
+
+ if (StringUtils.isBlank(snapshotRef.getName())) {
+ throw new IllegalArgumentException(
+ String.format("Invalid snapshot name: %s", snapshotRef.getName()));
+ }
+
+ FlinkStateSnapshot result;
+ if (snapshotRef.getName() != null) {
+ var namespace = snapshotRef.getNamespace();
+ if (namespace == null) {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .inNamespace(namespace)
+ .withName(snapshotRef.getName())
+ .get();
+ }
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ }
Review Comment:
Looks like late-night code, no idea how I left it like this :D thank you
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688087051
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/JobSpec.java:
##########
@@ -90,8 +100,8 @@ public class JobSpec implements Diffable<JobSpec> {
/**
* Nonce used to trigger a full redeployment of the job from the savepoint path specified in
- * initialSavepointPath. In order to trigger redeployment, change the number to a different
- * non-null value. Rollback is not possible after redeployment.
+ * initialSavepointPath or initialSavepointName. In order to trigger redeployment, change the
Review Comment:
Forgot to rename it
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora merged PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1691933742
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.SnapshotUtils;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+
+ FlinkStateSnapshotUtils.snapshotSuccessful(
+ resource, resource.getSpec().getSavepoint().getPath(), true);
+ return;
+ }
+
+ if (FlinkStateSnapshotUtils.abandonSnapshotIfJobNotRunning(
+ ctx.getKubernetesClient(),
+ ctx.getResource(),
+ ctx.getSecondaryResource().orElse(null),
+ eventRecorder)) {
+ return;
+ }
+
+ var jobId = ctx.getSecondaryResource().orElseThrow().getStatus().getJobStatus().getJobId();
+
+ Optional<String> triggerIdOpt;
+ try {
+ triggerIdOpt = triggerCheckpointOrSavepoint(resource.getSpec(), ctx, jobId);
+ } catch (Exception e) {
+ LOG.error("Failed to trigger snapshot for resource {}", ctx.getResource(), e);
+ throw new ReconciliationException(e);
+ }
+
+ if (triggerIdOpt.isEmpty()) {
+ LOG.warn("Failed to trigger snapshot {}", resource.getMetadata().getName());
+ return;
+ }
+
+ FlinkStateSnapshotUtils.snapshotInProgress(resource, triggerIdOpt.get());
+ }
+
+ public DeleteControl cleanup(FlinkStateSnapshotContext ctx) throws Exception {
+ var resource = ctx.getResource();
+ var state = resource.getStatus().getState();
+ var resourceName = resource.getMetadata().getName();
+ LOG.info("Cleaning up resource {}...", resourceName);
+
+ if (resource.getSpec().isCheckpoint()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (!resource.getSpec().getSavepoint().getDisposeOnDelete()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (resource.getSpec().getJobReference() == null
+ || resource.getSpec().getJobReference().getName() == null) {
+ return DeleteControl.defaultDelete();
+ }
+
+ switch (state) {
+ case IN_PROGRESS:
+ LOG.info(
+ "Cannot delete resource {} yet as savepoint is still in progress...",
+ resourceName);
+ return DeleteControl.noFinalizerRemoval()
+ .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ case COMPLETED:
+ var flinkDeployment = getFlinkDeployment(ctx);
+ return handleSnapshotCleanup(resource, flinkDeployment, ctx);
Review Comment:
Definitely! I would not feel comfortable releasing a new operator version that does not have auto-cleanup of FlinkStateSnapshot CRs. For the now-deprecated way of taking snapshots this could be configured using `kubernetes.operator.savepoint.history.max.age` and `kubernetes.operator.savepoint.history.max.count`.
You are right, `disposeOnDelete` is false by default, and it can be controlled by the config `kubernetes.operator.savepoint.dispose-on-delete`.
I am not sure of the exact approach I will take to handle cleanup, but it will be implemented under [FLINK-35493](https://issues.apache.org/jira/browse/FLINK-35493) in another PR. That's a feature I felt we could easily separate to another PR, as this one is already growing quite big :D
With a low periodic interval, there won't be any more load on the API server and etcd than tracking snapshots in the Flink resource CR (as the operator does currently). Periodically cleaning them up might affect the load on API server/etcd depending on what solution we will go with. But that will be done under FLINK-35493.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1684358470
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/JobSpec.java:
##########
@@ -90,8 +100,8 @@ public class JobSpec implements Diffable<JobSpec> {
/**
* Nonce used to trigger a full redeployment of the job from the savepoint path specified in
- * initialSavepointPath. In order to trigger redeployment, change the number to a different
- * non-null value. Rollback is not possible after redeployment.
+ * initialSavepointPath or initialSavepointName. In order to trigger redeployment, change the
Review Comment:
I might lack context here, but three is no `initialSavepointName` field in the whole codebase other than this comment and the generated doc from it.
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.api.spec;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** Spec for checkpoint state snapshots. */
+@Experimental
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CheckpointSpec {
+ /** Type of checkpoint to take. */
+ private CheckpointType checkpointType = CheckpointType.FULL;
Review Comment:
Due to [FLINK-33723](https://issues.apache.org/jira/browse/FLINK-33723), `INCREMENTAL` checkpoint triggering is disabled in Flink 1.19, so I think we may need to handle this on the operator side as well to not cause errors. Versions < 1.19 will work (although the actual checkpoint semantics might be undesired, see the linked jira for deatails).
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1671974248
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,202 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSnapshot(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ reconciler.reconcile(ctx);
+ } catch (Exception e) {
+ FlinkStateSnapshotUtils.snapshotFailed(
+ josdkContext.getClient(), eventRecorder, flinkStateSnapshot, e.getMessage());
+ LOG.error("Failed to reconcile {}", flinkStateSnapshot, e);
Review Comment:
This codepath threw `ReconciliationException` before commit https://github.com/apache/flink-kubernetes-operator/pull/821/commits/71a4137fe362c6ee20735787d40c9530268d0b49 but instead changing the state to `FAILED` here, then letting `getUpdateControl()` decide if we have reached the backoff limit and possibly setting the state back to `TRIGGER_PENDING` seems like a better solution.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1668650953
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -266,19 +304,31 @@ protected void restoreJob(
Optional<String> savepointOpt = Optional.empty();
if (spec.getJob().getUpgradeMode() != UpgradeMode.STATELESS) {
- savepointOpt =
- Optional.ofNullable(
- ctx.getResource()
- .getStatus()
- .getJobStatus()
- .getSavepointInfo()
- .getLastSavepoint())
- .flatMap(s -> Optional.ofNullable(s.getLocation()));
+ if (FlinkStateSnapshotUtils.shouldCreateSnapshotResource(
+ ctx.getOperatorConfig(), deployConfig)) {
+ savepointOpt = getLatestSavepointPathFromFlinkStateSnapshots(ctx);
+ } else {
+ savepointOpt =
+ Optional.ofNullable(
+ ctx.getResource()
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint())
+ .flatMap(s -> Optional.ofNullable(s.getLocation()));
Review Comment:
I have added a commit that should cover these points, I have also resolved related conversations. The commit also adds support for `savepointTriggerNonce` and `checkpointTriggerNonce` with the new snapshot resources. I will also add unit tests for these later.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1689490037
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -72,21 +74,26 @@ public class FlinkConfigManager {
private volatile Configuration defaultConfig;
private volatile FlinkOperatorConfiguration defaultOperatorConfiguration;
+ private final boolean snapshotCrdInstalled;
Review Comment:
As we discussed offline, the whole snapshot CRD install check is planned for 1 release to give more meaningful errors for users the first time they upgrade to an operator version that includes this feature, but will be useless after that. So I think it would make sense to mark this with a TODO to remove it in the 1.11 release.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1689526438
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java:
##########
@@ -83,20 +82,20 @@ void deleteClusterDeployment(
Configuration conf,
boolean deleteHaData);
- void cancelSessionJob(FlinkSessionJob sessionJob, UpgradeMode upgradeMode, Configuration conf)
+ Optional<String> cancelSessionJob(
+ FlinkSessionJob sessionJob, UpgradeMode upgradeMode, Configuration conf)
throws Exception;
- void triggerSavepoint(
+ String triggerSavepoint(
String jobId,
- SnapshotTriggerType triggerType,
- SavepointInfo savepointInfo,
+ org.apache.flink.core.execution.SavepointFormatType savepointFormatType,
+ String savepointDirectory,
Configuration conf)
throws Exception;
- void triggerCheckpoint(
+ String triggerCheckpoint(
String jobId,
- SnapshotTriggerType triggerType,
- CheckpointInfo checkpointInfo,
+ org.apache.flink.core.execution.CheckpointType checkpointFormatType,
Review Comment:
nit: checkpointFormatType -> checkpointType
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotContext.java:
##########
@@ -44,61 +45,56 @@ public class FlinkStateSnapshotContext {
private final Context<FlinkStateSnapshot> josdkContext;
private final FlinkConfigManager configManager;
- private FlinkOperatorConfiguration operatorConfig;
- private Configuration referencedJobObserveConfig;
- private FlinkDeployment referencedJobFlinkDeployment;
+ @Getter(lazy = true)
+ private final FlinkOperatorConfiguration operatorConfig = operatorConfig();
+
+ @Getter(lazy = true)
+ private final Configuration referencedJobObserveConfig = referencedJobObserveConfig();
+
+ @Getter(lazy = true)
+ private final FlinkDeployment referencedJobFlinkDeployment = referencedJobFlinkDeployment();
/**
* @return Operator configuration for this resource.
*/
- public FlinkOperatorConfiguration getOperatorConfig() {
- if (operatorConfig != null) {
- return operatorConfig;
- }
- return operatorConfig =
- configManager.getOperatorConfiguration(
- getResource().getMetadata().getNamespace(), null);
+ public FlinkOperatorConfiguration operatorConfig() {
+ return configManager.getOperatorConfiguration(
+ getResource().getMetadata().getNamespace(), null);
+ }
+
+ public Configuration referencedJobObserveConfig() {
+ return configManager.getObserveConfig(getReferencedJobFlinkDeployment());
+ }
+
+ public FlinkDeployment referencedJobFlinkDeployment() {
Review Comment:
These lazy init methods can be private.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/FlinkConfigManager.java:
##########
@@ -72,21 +74,26 @@ public class FlinkConfigManager {
private volatile Configuration defaultConfig;
private volatile FlinkOperatorConfiguration defaultOperatorConfiguration;
+ private final boolean snapshotCrdInstalled;
Review Comment:
As we discussed offline, the whole snapshot CRD install check is planned for 1 release to give more meaningful errors for users, so I think it would make sense to mark this with a TODO to remove it in the 1.11 release.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,395 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getAndValidateFlinkStateSnapshotPath(
Review Comment:
nit: `getValidatedFlinkStateSnapshotPath` is closer to what actually happens in this function.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,395 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getAndValidateFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (!StringUtils.isBlank(snapshotRef.getPath())) {
+ return snapshotRef.getPath();
+ }
+
+ if (StringUtils.isBlank(snapshotRef.getName())) {
+ throw new IllegalArgumentException(
+ String.format("Invalid snapshot name: %s", snapshotRef.getName()));
+ }
+
+ FlinkStateSnapshot result;
+ if (snapshotRef.getName() != null) {
+ var namespace = snapshotRef.getNamespace();
+ if (namespace == null) {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .inNamespace(namespace)
+ .withName(snapshotRef.getName())
+ .get();
+ }
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ }
Review Comment:
According to this code the `kubernetesClient` call will be the same if `snapshotRef.getName()` is `null` OR `snapshotRef.getNamespace()` is `null`, as L81-85, and L95-99 is the same. Also, we validate the `snapshotRef.getName()` is not `null` at L72, so I do not see why the another null-check is meaningful. So unless i miss something the whole `if` block can be simplified to:
```java
var result =
snapshotRef.getNamespace() == null
? kubernetesClient
.resources(FlinkStateSnapshot.class)
.withName(snapshotRef.getName())
.get()
: kubernetesClient
.resources(FlinkStateSnapshot.class)
.inNamespace(snapshotRef.getNamespace())
.withName(snapshotRef.getName())
.get();
```
##########
flink-kubernetes-webhook/src/main/java/org/apache/flink/kubernetes/operator/admission/mutator/FlinkMutator.java:
##########
@@ -92,4 +96,16 @@ private FlinkDeployment mutateDeployment(HasMetadata resource) {
throw new RuntimeException(e);
}
}
+
+ private FlinkStateSnapshot mutateStateSnapshot(HasMetadata resource) {
+ try {
+ var savepoint = mapper.convertValue(resource, FlinkStateSnapshot.class);
Review Comment:
nit: savepoint -> snapshot.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -525,18 +513,14 @@ public void triggerSavepoint(
.get(timeout, TimeUnit.SECONDS);
LOG.info("Savepoint successfully triggered: " + response.getTriggerId().toHexString());
- savepointInfo.setTrigger(
- response.getTriggerId().toHexString(),
- triggerType,
- SavepointFormatType.valueOf(savepointFormatType.name()));
+ return response.getTriggerId().toHexString();
}
}
@Override
- public void triggerCheckpoint(
+ public String triggerCheckpoint(
String jobId,
- SnapshotTriggerType triggerType,
- org.apache.flink.kubernetes.operator.api.status.CheckpointInfo checkpointInfo,
+ org.apache.flink.core.execution.CheckpointType checkpointFormatType,
Review Comment:
nit: checkpointFormatType -> checkpointType
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,395 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getAndValidateFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (!StringUtils.isBlank(snapshotRef.getPath())) {
Review Comment:
nit: `StringUtils.isNotBlank(...)` to avoid invert.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -487,34 +484,25 @@ public void cancelSessionJob(
}
jobStatus.setState(JobStatus.FINISHED.name());
- savepointOpt.ifPresent(
- location -> {
- Savepoint sp = Savepoint.of(location, SnapshotTriggerType.UPGRADE);
- jobStatus.getSavepointInfo().updateLastSavepoint(sp);
- });
+ return savepointOpt;
}
@Override
- public void triggerSavepoint(
+ public String triggerSavepoint(
String jobId,
- SnapshotTriggerType triggerType,
- org.apache.flink.kubernetes.operator.api.status.SavepointInfo savepointInfo,
+ org.apache.flink.core.execution.SavepointFormatType savepointFormatType,
+ String savepointDirectory,
Configuration conf)
throws Exception {
- LOG.info("Triggering new savepoint");
+ LOG.info("Triggering new savepoint using new method");
Review Comment:
This seems to me like a dev debug log.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,395 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getAndValidateFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (!StringUtils.isBlank(snapshotRef.getPath())) {
+ return snapshotRef.getPath();
+ }
+
+ if (StringUtils.isBlank(snapshotRef.getName())) {
+ throw new IllegalArgumentException(
+ String.format("Invalid snapshot name: %s", snapshotRef.getName()));
+ }
+
+ FlinkStateSnapshot result;
+ if (snapshotRef.getName() != null) {
+ var namespace = snapshotRef.getNamespace();
+ if (namespace == null) {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .inNamespace(namespace)
+ .withName(snapshotRef.getName())
+ .get();
+ }
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ }
+
+ if (result == null) {
+ throw new IllegalArgumentException(
Review Comment:
IMO `IllegalStateException` would be better here, as we validate the name already so the passed argument itself seems to be valid. Same for the following 2 throws in the current function.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,395 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getAndValidateFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (!StringUtils.isBlank(snapshotRef.getPath())) {
+ return snapshotRef.getPath();
+ }
+
+ if (StringUtils.isBlank(snapshotRef.getName())) {
+ throw new IllegalArgumentException(
+ String.format("Invalid snapshot name: %s", snapshotRef.getName()));
+ }
+
+ FlinkStateSnapshot result;
+ if (snapshotRef.getName() != null) {
+ var namespace = snapshotRef.getNamespace();
+ if (namespace == null) {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .inNamespace(namespace)
+ .withName(snapshotRef.getName())
+ .get();
+ }
+ } else {
+ result =
+ kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get();
+ }
+
+ if (result == null) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot find snapshot %s in namespace %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ // We can return the savepoint path if it's marked as completed without waiting for the
+ // reconciler to update its status.
+ if (result.getSpec().isSavepoint() && result.getSpec().getSavepoint().getAlreadyExists()) {
+ var path = result.getSpec().getSavepoint().getPath();
+ if (!StringUtils.isBlank(path)) {
+ return path;
+ }
+ }
+
+ if (COMPLETED != result.getStatus().getState()) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Snapshot %s/%s is not complete yet.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ var path = result.getStatus().getPath();
+ if (StringUtils.isBlank(path)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Snapshot %s/%s path is incorrect: %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName(), path));
+ }
+
+ return path;
+ }
+
+ protected static FlinkStateSnapshot createFlinkStateSnapshot(
+ KubernetesClient kubernetesClient,
+ String name,
+ FlinkStateSnapshotSpec spec,
+ SnapshotTriggerType triggerType) {
+ var metadata = new ObjectMeta();
+ metadata.setName(name);
+ metadata.getLabels().put(CrdConstants.LABEL_SNAPSHOT_TYPE, triggerType.name());
+
+ var snapshot = new FlinkStateSnapshot();
+ snapshot.setSpec(spec);
+ snapshot.setMetadata(metadata);
+
+ return kubernetesClient.resource(snapshot).create();
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param savepointPath savepoint path if any
+ * @param triggerType trigger type
+ * @param savepointFormatType format type
+ * @param disposeOnDelete should dispose of data on deletion
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createSavepointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ @Nullable String savepointPath,
+ SnapshotTriggerType triggerType,
+ SavepointFormatType savepointFormatType,
+ boolean disposeOnDelete) {
+ var savepointSpec =
+ SavepointSpec.builder()
+ .path(savepointPath)
+ .formatType(savepointFormatType)
+ .disposeOnDelete(disposeOnDelete)
+ .alreadyExists(triggerType == SnapshotTriggerType.UPGRADE)
+ .build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .savepoint(savepointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(SAVEPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param checkpointType type of checkpoint
+ * @param triggerType trigger type
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createCheckpointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ CheckpointType checkpointType,
+ SnapshotTriggerType triggerType) {
+ var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .checkpoint(checkpointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(CHECKPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Based on job configuration and operator configuration, decide if {@link FlinkStateSnapshot}
+ * resources should be used or not. Operator configuration will disable the usage of the
+ * corresponding CRD was not installed on this Kubernetes cluster.
+ *
+ * @param operatorConfiguration operator config
+ * @param configuration job config
+ * @return true if snapshot resources should be created
+ */
+ public static boolean isSnapshotResourceEnabled(
+ FlinkOperatorConfiguration operatorConfiguration, Configuration configuration) {
+ return configuration.get(SNAPSHOT_RESOURCE_ENABLED)
+ && operatorConfiguration.isSnapshotResourcesEnabled();
+ }
+
+ /**
+ * Return a generated name for a {@link FlinkStateSnapshot} to be created.
+ *
+ * @param snapshotType type of snapshot
+ * @param triggerType trigger type of snapshot
+ * @param referencedResource referenced resource
+ * @return result name
+ */
+ public static String getFlinkStateSnapshotName(
+ SnapshotType snapshotType,
+ SnapshotTriggerType triggerType,
+ AbstractFlinkResource<?, ?> referencedResource) {
+ return String.format(
+ "%s-%s-%s-%s",
+ referencedResource.getMetadata().getName(),
+ snapshotType.name().toLowerCase(),
+ triggerType.name().toLowerCase(),
+ UUID.randomUUID());
+ }
+
+ /**
+ * For an upgrade savepoint, create a {@link FlinkStateSnapshot} on the Kubernetes cluster and
+ * return its reference if snapshot resources are enabled. In other case return a reference
+ * containing only the path.
+ *
+ * @param conf job configuration
+ * @param operatorConf operator configuration
+ * @param kubernetesClient kubernetes client
+ * @param flinkResource referenced Flink resource
+ * @param savepointFormatType savepoint format type
+ * @param savepointPath path of savepoint
+ * @return reference for snapshot
+ */
+ public static FlinkStateSnapshotReference createReferenceForUpgradeSavepoint(
+ Configuration conf,
+ FlinkOperatorConfiguration operatorConf,
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> flinkResource,
+ SavepointFormatType savepointFormatType,
+ String savepointPath) {
+ if (isSnapshotResourceEnabled(operatorConf, conf)) {
+ var snapshot =
+ createSavepointResource(
+ kubernetesClient,
+ flinkResource,
+ savepointPath,
+ SnapshotTriggerType.UPGRADE,
+ savepointFormatType,
+ conf.get(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_JOB_SAVEPOINT_DISPOSE_ON_DELETE));
+ return FlinkStateSnapshotReference.fromResource(snapshot);
+ } else {
Review Comment:
nit: else can be omitted
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java:
##########
@@ -145,6 +150,125 @@ public static InformerEventSource<Deployment, FlinkDeployment> getDeploymentInfo
return new InformerEventSource<>(configuration, context);
}
+ public static EventSource[] getFlinkStateSnapshotInformerEventSources(
+ EventSourceContext<FlinkStateSnapshot> context) {
+ context.getPrimaryCache()
+ .addIndexer(
+ FLINK_STATE_SNAPSHOT_IDX,
+ savepoint -> {
+ if (savepoint.getSpec().getJobReference() == null
+ || savepoint.getSpec().getJobReference().getName() == null) {
+ return Collections.emptyList();
+ }
+ return List.of(
+ indexKey(
+ savepoint.getSpec().getJobReference().toString(),
+ savepoint.getMetadata().getNamespace()));
+ });
+
+ InformerConfiguration<FlinkSessionJob> configurationFlinkSessionJob =
+ InformerConfiguration.from(FlinkSessionJob.class, context)
+ .withSecondaryToPrimaryMapper(
Review Comment:
IMO this logic can be extracted to a separate method and then we can cast it at call time:
```java
private static SecondaryToPrimaryMapper<? extends AbstractFlinkResource<?, ?>>
getSnapshotPrimaryMapper(EventSourceContext<FlinkStateSnapshot> ctx) {
return flinkResource ->
ctx
.getPrimaryCache()
.byIndex(
FLINK_STATE_SNAPSHOT_IDX,
indexKey(
flinkResource.getMetadata().getName(),
flinkResource.getMetadata().getNamespace()))
.stream()
.map(ResourceID::fromResource)
.collect(Collectors.toSet());
}
```
Since the body is the same for both FlinkDeployment and FlinkSessionJob.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1692766135
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.SnapshotUtils;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+
+ FlinkStateSnapshotUtils.snapshotSuccessful(
+ resource, resource.getSpec().getSavepoint().getPath(), true);
+ return;
+ }
+
+ if (FlinkStateSnapshotUtils.abandonSnapshotIfJobNotRunning(
+ ctx.getKubernetesClient(),
+ ctx.getResource(),
+ ctx.getSecondaryResource().orElse(null),
+ eventRecorder)) {
+ return;
+ }
+
+ var jobId = ctx.getSecondaryResource().orElseThrow().getStatus().getJobStatus().getJobId();
+
+ Optional<String> triggerIdOpt;
+ try {
+ triggerIdOpt = triggerCheckpointOrSavepoint(resource.getSpec(), ctx, jobId);
+ } catch (Exception e) {
+ LOG.error("Failed to trigger snapshot for resource {}", ctx.getResource(), e);
+ throw new ReconciliationException(e);
+ }
+
+ if (triggerIdOpt.isEmpty()) {
+ LOG.warn("Failed to trigger snapshot {}", resource.getMetadata().getName());
+ return;
+ }
+
+ FlinkStateSnapshotUtils.snapshotInProgress(resource, triggerIdOpt.get());
+ }
+
+ public DeleteControl cleanup(FlinkStateSnapshotContext ctx) throws Exception {
+ var resource = ctx.getResource();
+ var state = resource.getStatus().getState();
+ var resourceName = resource.getMetadata().getName();
+ LOG.info("Cleaning up resource {}...", resourceName);
+
+ if (resource.getSpec().isCheckpoint()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (!resource.getSpec().getSavepoint().getDisposeOnDelete()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (resource.getSpec().getJobReference() == null
+ || resource.getSpec().getJobReference().getName() == null) {
+ return DeleteControl.defaultDelete();
+ }
+
+ switch (state) {
+ case IN_PROGRESS:
+ LOG.info(
+ "Cannot delete resource {} yet as savepoint is still in progress...",
+ resourceName);
+ return DeleteControl.noFinalizerRemoval()
+ .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ case COMPLETED:
+ var flinkDeployment = getFlinkDeployment(ctx);
+ return handleSnapshotCleanup(resource, flinkDeployment, ctx);
Review Comment:
I have done some investigation, and I think I will be able to easily implement this in the FlinkDeployment and FlinkSessionJob controller observe phase, just like the operator currently cleans up savepoints. I could just add a new `InformerEventSource` so we would be able to list `FlinkStateSnapshot` resources as secondary resources when we reconcile FlinkDeployment/FlinkSessionJob resources without querying the Kubernetes API server.
Once this PR is merged, I will create a new PR for this functionality.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1666908525
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/CustomResourceDefinitionWatcherImpl.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.crd;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Default implementation for CustomResourceDefinitionWatcher. */
+@RequiredArgsConstructor
+public class CustomResourceDefinitionWatcherImpl implements CustomResourceDefinitionWatcher {
Review Comment:
I have pushed a commit for this, please take a look at it, I am personally not 100% satisfied with either solution. The new solution is a lot more simple, but I had to hardcode the check to `FlinkConfigManager`, and play with the constructors in a way such that the tests won't try to detect if the `FlinkStateSnapshot` CR is installed.
I also kept `FlinkStateSnapshotUtils.shouldCreateSnapshotResource()` that checks both the operator config and Flink resource config to make it easy to opt-out on a per-job basis. This also helped me with the current tests being able to decide which snapshot method to use.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1668364184
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -266,19 +304,31 @@ protected void restoreJob(
Optional<String> savepointOpt = Optional.empty();
if (spec.getJob().getUpgradeMode() != UpgradeMode.STATELESS) {
- savepointOpt =
- Optional.ofNullable(
- ctx.getResource()
- .getStatus()
- .getJobStatus()
- .getSavepointInfo()
- .getLastSavepoint())
- .flatMap(s -> Optional.ofNullable(s.getLocation()));
+ if (FlinkStateSnapshotUtils.shouldCreateSnapshotResource(
+ ctx.getOperatorConfig(), deployConfig)) {
+ savepointOpt = getLatestSavepointPathFromFlinkStateSnapshots(ctx);
+ } else {
+ savepointOpt =
+ Optional.ofNullable(
+ ctx.getResource()
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint())
+ .flatMap(s -> Optional.ofNullable(s.getLocation()));
Review Comment:
Okay, I am definitely happy with this plan, I will push a commit soon:
- the new field will be updated on upgrade, stop-with-savepoint and when the job becomes terminal
- the field won't be set during manual and periodic snapshots (in contrast to how it worked so far)
- savepoint path in `AbstractJobReconciler#restoreJob()` will be set to this field
- `lastSavepoint` will still be used for periodic and manual savepoints if `FlinkStateSnapshot` is disabled. This will be deprecated.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688104509
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotContext.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.JobKind;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Optional;
+
+/** Context for reconciling a snapshot. */
+@Getter
+@RequiredArgsConstructor
+public class FlinkStateSnapshotContext {
+
+ private final FlinkStateSnapshot resource;
+ private final FlinkStateSnapshotStatus originalStatus;
+ private final Context<FlinkStateSnapshot> josdkContext;
+ private final FlinkConfigManager configManager;
+
+ private FlinkOperatorConfiguration operatorConfig;
+ private Configuration referencedJobObserveConfig;
+ private FlinkDeployment referencedJobFlinkDeployment;
+
+ /**
+ * @return Operator configuration for this resource.
+ */
+ public FlinkOperatorConfiguration getOperatorConfig() {
+ if (operatorConfig != null) {
+ return operatorConfig;
+ }
+ return operatorConfig =
+ configManager.getOperatorConfiguration(
+ getResource().getMetadata().getNamespace(), null);
Review Comment:
I have flipped the ifs for now, I will come back to check using Lombok lazy getter
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1687799832
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java:
##########
@@ -426,6 +438,14 @@ public static String operatorConfigKey(String key) {
.withDescription(
"Max allowed checkpoint age for initiating last-state upgrades on running jobs. If a checkpoint is not available within the desired age (and nothing in progress) a savepoint will be triggered.");
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<Boolean> OPERATOR_JOB_SAVEPOINT_DISPOSE_ON_DELETE =
+ operatorConfig("savepoint.dispose-on-delete")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Savepoint data for FlinkStateSnapshot resources created by the operator during upgrades and periodic savepoints will be disposed of automatically when the generated Kubernetes resource is deleted.");
Review Comment:
nit: "...savepoints will be disposed ~of~ automatically when..."
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkStateSnapshotState.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.api.status;
+
+import org.apache.flink.annotation.Experimental;
+
+/** Describes current snapshot state. */
+@Experimental
+public enum FlinkStateSnapshotState {
Review Comment:
Maybe move this inside the `FlinkStateSnapshotStatus`, and then the enum name could be simply `State`. This might improve the overall readability and we can work around the kind of confusing wording here a bit better.
I see that `ReconciliationStatus` and `ReconciliationState` follows the current setup and this just setup the same way, but the wording is straightforward there.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java:
##########
@@ -297,6 +297,18 @@ public static String operatorConfigKey(String key) {
"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<Boolean> SNAPSHOT_RESOURCE_ENABLED =
+ operatorConfig("snapshot.resource.enabled")
+ .booleanType()
+ .defaultValue(true)
+ .withDescription(
+ "Create new FlinkStateSnapshot resources for storing snapshots. "
+ + "Disable if you wish to use the deprecated mode and save snapshot results to "
+ + "FlinkDeployment/FlinkSessionJob status fields. The Operator will fallback to the "
Review Comment:
nit: "The Operator will fallback to ~the~"
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ // status might be null here
+ flinkStateSnapshot.setStatus(
+ Objects.requireNonNullElseGet(
+ flinkStateSnapshot.getStatus(), FlinkStateSnapshotStatus::new));
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ observer.observe(ctx);
+
+ if (validateSnapshot(ctx)) {
+ reconciler.reconcile(ctx);
+ }
+
+ notifyListeners(ctx);
+ return getUpdateControl(ctx);
+ }
+
+ @Override
+ public DeleteControl cleanup(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+ DeleteControl deleteControl;
Review Comment:
This local variable is unnecessary. We could return inside the `try`, as the `catch` has its own return as well.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotContext.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.JobKind;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Optional;
+
+/** Context for reconciling a snapshot. */
+@Getter
+@RequiredArgsConstructor
+public class FlinkStateSnapshotContext {
+
+ private final FlinkStateSnapshot resource;
+ private final FlinkStateSnapshotStatus originalStatus;
+ private final Context<FlinkStateSnapshot> josdkContext;
+ private final FlinkConfigManager configManager;
+
+ private FlinkOperatorConfiguration operatorConfig;
+ private Configuration referencedJobObserveConfig;
+ private FlinkDeployment referencedJobFlinkDeployment;
+
+ /**
+ * @return Operator configuration for this resource.
+ */
+ public FlinkOperatorConfiguration getOperatorConfig() {
+ if (operatorConfig != null) {
+ return operatorConfig;
+ }
+ return operatorConfig =
+ configManager.getOperatorConfiguration(
+ getResource().getMetadata().getNamespace(), null);
+ }
+
+ public Optional<AbstractFlinkResource<?, ?>> getSecondaryResource() {
+ var jobRef = getResource().getSpec().getJobReference();
+ if (jobRef == null) {
+ return Optional.empty();
+ }
+
+ if (JobKind.FLINK_DEPLOYMENT.equals(jobRef.getKind())) {
+ return getJosdkContext().getSecondaryResource(FlinkDeployment.class).map(r -> r);
+ } else if (JobKind.FLINK_SESSION_JOB.equals(jobRef.getKind())) {
+ return getJosdkContext().getSecondaryResource(FlinkSessionJob.class).map(r -> r);
+ } else {
+ return Optional.empty();
+ }
Review Comment:
nit: We can eliminate the first null-check, so the flow will be more straightforward, but getting the `jobKind` is a bit more complex:
```java
var jobKind =
Optional.ofNullable(getResource().getSpec().getJobReference())
.map(JobReference::getKind)
.orElse(null);
if (JobKind.FLINK_DEPLOYMENT.equals(jobKind)) {
return getJosdkContext().getSecondaryResource(FlinkDeployment.class).map(r -> r);
}
if (JobKind.FLINK_SESSION_JOB.equals(jobKind)) {
return getJosdkContext().getSecondaryResource(FlinkSessionJob.class).map(r -> r);
}
return Optional.empty();
```
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/SavepointInfo.java:
##########
@@ -35,7 +35,10 @@
@AllArgsConstructor
@Builder
public class SavepointInfo implements SnapshotInfo {
- /** Last completed savepoint by the operator. */
+ /**
+ * Last completed savepoint by the operator for manual and periodic snapshots. Only used if
Review Comment:
Should not this be marked as `@Deprecated` as well?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotContext.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.JobKind;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Optional;
+
+/** Context for reconciling a snapshot. */
+@Getter
+@RequiredArgsConstructor
+public class FlinkStateSnapshotContext {
+
+ private final FlinkStateSnapshot resource;
+ private final FlinkStateSnapshotStatus originalStatus;
+ private final Context<FlinkStateSnapshot> josdkContext;
+ private final FlinkConfigManager configManager;
+
+ private FlinkOperatorConfiguration operatorConfig;
+ private Configuration referencedJobObserveConfig;
+ private FlinkDeployment referencedJobFlinkDeployment;
+
+ /**
+ * @return Operator configuration for this resource.
+ */
+ public FlinkOperatorConfiguration getOperatorConfig() {
+ if (operatorConfig != null) {
+ return operatorConfig;
+ }
+ return operatorConfig =
+ configManager.getOperatorConfiguration(
+ getResource().getMetadata().getNamespace(), null);
Review Comment:
IMO inverting the current `if` condition creates a more natural and readable flow with 1 `return`, so i would consider that as an improvement.
But Lombok's [lazy getter](https://projectlombok.org/features/GetterLazy) would be even better to use for all currently non-final fields, which could be `final` in that case an we can eliminate the null-checks from the src.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,216 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
Review Comment:
Ctor can be replaced with `@RequiredArgsConstructor`.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java:
##########
@@ -226,7 +229,27 @@ private void setJobIdIfNecessary(
@Override
protected void cancelJob(FlinkResourceContext<FlinkDeployment> ctx, UpgradeMode upgradeMode)
throws Exception {
- ctx.getFlinkService().cancelJob(ctx.getResource(), upgradeMode, ctx.getObserveConfig());
+ var conf = ctx.getObserveConfig() != null ? ctx.getObserveConfig() : new Configuration();
+ var savepointFormatType =
+ conf.get(KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE);
+
+ var savepointOpt = ctx.getFlinkService().cancelJob(ctx.getResource(), upgradeMode, conf);
+ savepointOpt.ifPresent(
Review Comment:
The body of this optional could be moved to `AbstractJobReconciler` as a new protected method, so we only duplicate as minimal logic as we can.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java:
##########
@@ -273,15 +280,19 @@ private void disposeSavepointQuietly(
}
}
- private void observeLatestSavepoint(
+ private void observeLatestCheckpoint(
Review Comment:
Is this always a checkpoint? Shouldn't this be more like `observeLatestSnapshot`? Also, we should update the error log at L297.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,154 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSnapshotState(ctx);
+ }
+ }
+
+ private void observeSnapshotState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ if (StringUtils.isEmpty(triggerId)) {
+ return;
+ }
+
+ LOG.debug("Observing snapshot state for resource {}...", resourceName);
+
+ if (FlinkStateSnapshotUtils.abandonSnapshotIfJobNotRunning(
+ ctx.getKubernetesClient(),
+ ctx.getResource(),
+ ctx.getSecondaryResource().orElse(null),
+ eventRecorder)) {
+ return;
+ }
+
+ var jobId = ctx.getSecondaryResource().orElseThrow().getStatus().getJobStatus().getJobId();
+ var ctxFlinkDeployment =
+ ctxFactory.getResourceContext(
+ ctx.getReferencedJobFlinkDeployment(), ctx.getJosdkContext());
+ var observeConfig = ctx.getReferencedJobObserveConfig();
+
+ if (resource.getSpec().isSavepoint()) {
+ var savepointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchSavepointInfo(triggerId, jobId, observeConfig);
+ handleSavepoint(ctx, savepointInfo);
+ } else {
+ var checkpointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchCheckpointInfo(triggerId, jobId, observeConfig);
+ handleCheckpoint(ctx, checkpointInfo, ctxFlinkDeployment, jobId);
+ }
+ }
+
+ private void handleSavepoint(
+ FlinkStateSnapshotContext ctx, SavepointFetchResult savepointInfo) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+
+ if (savepointInfo.isPending()) {
+ LOG.debug(
+ "Savepoint '{}' with ID {} is pending",
+ resourceName,
+ resource.getStatus().getTriggerId());
+ } else if (savepointInfo.getError() != null) {
+ throw new ReconciliationException(savepointInfo.getError());
+ } else {
+ LOG.info("Savepoint {} successful: {}", resourceName, savepointInfo.getLocation());
+ FlinkStateSnapshotUtils.snapshotSuccessful(
+ resource, savepointInfo.getLocation(), false);
+ }
+ }
+
+ private void handleCheckpoint(
+ FlinkStateSnapshotContext ctx,
+ CheckpointFetchResult checkpointInfo,
+ FlinkResourceContext<FlinkDeployment> ctxFlinkDeployment,
+ String jobId) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+
+ if (checkpointInfo.isPending()) {
+ LOG.debug(
+ "Checkpoint for {} with ID {} is pending",
+ resourceName,
+ resource.getStatus().getTriggerId());
+ return;
+ }
+
+ if (checkpointInfo.getError() != null) {
+ throw new ReconciliationException(checkpointInfo.getError());
+ } else {
Review Comment:
nit: `else` can be omitted, so this fairly big block can spare 1 indentation level.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/listener/AuditUtils.java:
##########
@@ -53,6 +62,16 @@ private static String format(@NonNull CommonStatus<?> status) {
: status.getError());
}
+ private static String format(@NonNull FlinkStateSnapshotStatus status) {
+ if (StringUtils.isEmpty(status.getError())) {
+ return String.format(
+ ">>> Status[Snapshot] | Info | %s | %s", status.getState(), status.getPath());
Review Comment:
I thin we should consider line field length consistency, just as the already existing `format` methods.
Also now that we have different kind of `Status` logs, maybe identify the job status changes as well would make sense. Calling it `Status[Job]` seems appropriate, but we should make sure we define a correct `Status[%-8s]` template or something like that.
And as far as I can see the `Event` logs are also aiming the same field length compared to status, so probably we should try to keep a completely consistent format.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/metrics/FlinkStateSnapshotMetrics.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * 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.api.FlinkStateSnapshot;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/** FlinkSessionJob metrics. */
Review Comment:
`FlinkSessionJob` -> `FlinkStateSnapshot`. Same for the field level variable and the `onUpdate`/`onRemove` param. Copy-paste residue. :)
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "ferenc-csaky (via GitHub)" <gi...@apache.org>.
ferenc-csaky commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1686812531
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.kubernetes.operator.api.spec;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/** Spec for checkpoint state snapshots. */
+@Experimental
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CheckpointSpec {
+ /** Type of checkpoint to take. */
+ private CheckpointType checkpointType = CheckpointType.FULL;
Review Comment:
Due to [FLINK-33723](https://issues.apache.org/jira/browse/FLINK-33723), `INCREMENTAL` checkpoint triggering via REST API is disabled in Flink 1.19, so I think we may need to handle this on the operator side as well to not cause errors. Versions < 1.19 will work (although the actual checkpoint semantics might be undesired, see the linked jira for deatails).
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667707042
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -266,19 +304,31 @@ protected void restoreJob(
Optional<String> savepointOpt = Optional.empty();
if (spec.getJob().getUpgradeMode() != UpgradeMode.STATELESS) {
- savepointOpt =
- Optional.ofNullable(
- ctx.getResource()
- .getStatus()
- .getJobStatus()
- .getSavepointInfo()
- .getLastSavepoint())
- .flatMap(s -> Optional.ofNullable(s.getLocation()));
+ if (FlinkStateSnapshotUtils.shouldCreateSnapshotResource(
+ ctx.getOperatorConfig(), deployConfig)) {
+ savepointOpt = getLatestSavepointPathFromFlinkStateSnapshots(ctx);
+ } else {
+ savepointOpt =
+ Optional.ofNullable(
+ ctx.getResource()
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint())
+ .flatMap(s -> Optional.ofNullable(s.getLocation()));
Review Comment:
Thank you for detailing this problem so well, it makes perfect sense, I have overlooked this part of the operator.
I think I could tackle this problem with a new status field as you have proposed called `upgradeSnapshotReference` in the job status, but this still leaves an important feature out which was covered before: Manual savepoints triggered by creating a new `FlinkStateSnapshot` resource, and periodic savepoints that create new `FlinkStateSnapshot` resources will not update this new field `upgradeSnapshotReference` in FlinkDeployment/FlinkSessionJob statuses.
I see 3 solutions:
- Update FlinkDeployment/FlinkSessionJob status from the snapshot controller, I agree that this is not an optimal pattern, and I would avoid it.
- List all `FlinkStateSnapshot` resources when re-deploying a Flink job and we check `upgradeSnapshotReference`. We could compare the time the snapshots were taken and use the most recent. This might lead to confusion because e.g. starting a suspended job might not use the savepoint we can see in the status.
- From the FlinkDeployment/FlinkSessionJob controllers check the status of `FlinkStateSnapshot` resources with the help of `InformerEventSource`, not sure how viable this is.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1666904033
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSavepoint(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
Review Comment:
StatusRecorder will also notify `FlinkResourceListener` instances, I am not sure if we should give up that functionality for the snapshot resource. But I agree that we should use the JOSDK update mechanism if possible, I will try to refactor it.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1678989460
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSavepoint(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
Review Comment:
Done in commit 0315d1062dcf0b81e61af77be4a0c12e9330398b - I added some logic to decide if the status was modified, and notify listeners and `patchStatus` if it changed.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#issuecomment-2247282685
Thanks for your feedback @anupamaggarwal, really appreciate it!
1. I agree, and this is what I expected to happen. Except it does not, because I forgot to add the FlinkStateSnapshot resource to the `ValidatingWebhookConfiguration` resource in the Helm chart, I will push a commit soon.
2. This PR is already quite huge, so I have decided to do the examples/docs in a separate PR, there is currently a draft for it: https://github.com/apache/flink-kubernetes-operator/pull/854
3. You are right, adding the config there would improve its visibility to the end-users, so I will do that in a future commit, thank you!
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688136504
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java:
##########
@@ -273,15 +280,19 @@ private void disposeSavepointQuietly(
}
}
- private void observeLatestSavepoint(
+ private void observeLatestCheckpoint(
Review Comment:
Yep, this will be the latest checkpoint retrieved via Flink REST API, I have updated the log
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#issuecomment-2212378196
A small addition to the previous comments. The SnapshotObserver currently also observes and updates the lastSavepointInfo for terminal jobs. This is a key mechanism to be able to handle failures during stateful upgrades.
So we have to update that logic as well so that instead of record the lastSavepoint status we use a new shared logic with the snapshot custom resource mechanism. This mechanism is covered by some tests currently, enabling the snapshot CR for those tests would probably help repro the problem. For example:
1. Introduce a failure after executing a stop-with-savepoint operation (before the snapshot CR was created)
2. This is the point where the observer would actually observe the savepoint/checkpoint info from the terminal job and update the status
3. Assert that the upgrade is actually executed from the correct savepoint
I believe with the current implementation the savpoint info would be lost
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667389146
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSavepointState(ctx);
+ }
+ }
+
+ private void observeSavepointState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ if (StringUtils.isEmpty(triggerId)) {
+ return;
+ }
+
+ LOG.debug("Observing savepoint state for resource {}...", resourceName);
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s for savepoint %s was not found",
+ resource.getSpec().getJobReference(),
+ resourceName)));
Review Comment:
Yeah, I agree this should be abandoned as well. I have also added an event that will be produced when the snapshot gets to an abandoned state.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667350815
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java:
##########
@@ -410,6 +430,14 @@ public static String operatorConfigKey(String key) {
.withDescription(
"Max number of reconcile loops triggered within the rate limiter refresh period for each resource. Setting the limit <= 0 disables the limiter.");
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<Boolean> OPERATOR_JOB_UPGRADE_SAVEPOINT_DISPOSE_ON_DELETE =
+ operatorConfig("job.upgrade.savepoint.dispose-on-delete")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Savepoint data created by upgrade savepoints as FlinkStateSnapshot resources will be disposed of automatically when the Kubernetes resource gets deleted.");
+
Review Comment:
Good idea, I have named it `savepoint.dispose-on-delete`.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667350676
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java:
##########
@@ -27,11 +27,13 @@
import io.fabric8.crd.generator.annotation.PrinterColumn;
import lombok.AllArgsConstructor;
import lombok.Data;
+import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import org.apache.commons.lang3.StringUtils;
/** Last observed common status of the Flink deployment/Flink SessionJob. */
+@EqualsAndHashCode
Review Comment:
No clue how it got here, I could safely remove it.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#issuecomment-2265719872
I rebased the branch on top of main, but I also pushed commit 566caf9ba7220e8a53dc520822621db2928eafa4
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "anupamaggarwal (via GitHub)" <gi...@apache.org>.
anupamaggarwal commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1691868012
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getValidatedFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (StringUtils.isNotBlank(snapshotRef.getPath())) {
+ return snapshotRef.getPath();
+ }
+
+ if (StringUtils.isBlank(snapshotRef.getName())) {
+ throw new IllegalArgumentException(
+ String.format("Invalid snapshot name: %s", snapshotRef.getName()));
+ }
+
+ var result =
+ snapshotRef.getNamespace() == null
+ ? kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get()
+ : kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .inNamespace(snapshotRef.getNamespace())
+ .withName(snapshotRef.getName())
+ .get();
+
+ if (result == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot find snapshot %s in namespace %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ // We can return the savepoint path if it's marked as completed without waiting for the
+ // reconciler to update its status.
+ if (result.getSpec().isSavepoint() && result.getSpec().getSavepoint().getAlreadyExists()) {
+ var path = result.getSpec().getSavepoint().getPath();
+ if (!StringUtils.isBlank(path)) {
+ return path;
+ }
+ }
+
+ if (COMPLETED != result.getStatus().getState()) {
+ throw new IllegalStateException(
+ String.format(
+ "Snapshot %s/%s is not complete yet.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ var path = result.getStatus().getPath();
+ if (StringUtils.isBlank(path)) {
+ throw new IllegalStateException(
+ String.format(
+ "Snapshot %s/%s path is incorrect: %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName(), path));
+ }
+
+ return path;
+ }
+
+ protected static FlinkStateSnapshot createFlinkStateSnapshot(
+ KubernetesClient kubernetesClient,
+ String name,
+ FlinkStateSnapshotSpec spec,
+ SnapshotTriggerType triggerType) {
+ var metadata = new ObjectMeta();
+ metadata.setName(name);
+ metadata.getLabels().put(CrdConstants.LABEL_SNAPSHOT_TYPE, triggerType.name());
+
+ var snapshot = new FlinkStateSnapshot();
+ snapshot.setSpec(spec);
+ snapshot.setMetadata(metadata);
+
+ return kubernetesClient.resource(snapshot).create();
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param savepointPath savepoint path if any
+ * @param triggerType trigger type
+ * @param savepointFormatType format type
+ * @param disposeOnDelete should dispose of data on deletion
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createSavepointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ @Nullable String savepointPath,
+ SnapshotTriggerType triggerType,
+ SavepointFormatType savepointFormatType,
+ boolean disposeOnDelete) {
+ var savepointSpec =
+ SavepointSpec.builder()
+ .path(savepointPath)
+ .formatType(savepointFormatType)
+ .disposeOnDelete(disposeOnDelete)
+ .alreadyExists(triggerType == SnapshotTriggerType.UPGRADE)
+ .build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .savepoint(savepointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(SAVEPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param checkpointType type of checkpoint
+ * @param triggerType trigger type
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createCheckpointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ CheckpointType checkpointType,
+ SnapshotTriggerType triggerType) {
+ var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .checkpoint(checkpointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(CHECKPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Based on job configuration and operator configuration, decide if {@link FlinkStateSnapshot}
+ * resources should be used or not. Operator configuration will disable the usage of the
+ * corresponding CRD was not installed on this Kubernetes cluster.
+ *
+ * @param operatorConfiguration operator config
+ * @param configuration job config
+ * @return true if snapshot resources should be created
+ */
+ public static boolean isSnapshotResourceEnabled(
+ FlinkOperatorConfiguration operatorConfiguration, Configuration configuration) {
+ return configuration.get(SNAPSHOT_RESOURCE_ENABLED)
+ && operatorConfiguration.isSnapshotResourcesEnabled();
+ }
+
+ /**
+ * Return a generated name for a {@link FlinkStateSnapshot} to be created.
+ *
+ * @param snapshotType type of snapshot
+ * @param triggerType trigger type of snapshot
+ * @param referencedResource referenced resource
+ * @return result name
+ */
+ public static String getFlinkStateSnapshotName(
+ SnapshotType snapshotType,
+ SnapshotTriggerType triggerType,
+ AbstractFlinkResource<?, ?> referencedResource) {
+ return String.format(
+ "%s-%s-%s-%s",
+ referencedResource.getMetadata().getName(),
+ snapshotType.name().toLowerCase(),
+ triggerType.name().toLowerCase(),
+ UUID.randomUUID());
Review Comment:
Hi @mateczagany, wdyt about having a epoch (any lexicographically increasing) value instead of the UUID here? It might help users figure out the latest snapshot CR in case they do a
`kubectl get flinkstatesnapshots.flink.apache.org` (and not have to pass in `--sort-by=.metadata.creationTimestamp` explicitly) ?
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1691962117
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java:
##########
@@ -0,0 +1,382 @@
+/*
+ * 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.utils;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.CheckpointSpec;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotReference;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.spec.JobReference;
+import org.apache.flink.kubernetes.operator.api.spec.SavepointSpec;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+import org.apache.flink.kubernetes.operator.api.status.SavepointFormatType;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+import org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.SnapshotType;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.lang3.StringUtils;
+
+import javax.annotation.Nullable;
+
+import java.time.Instant;
+import java.util.UUID;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.ABANDONED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.IN_PROGRESS;
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+import static org.apache.flink.kubernetes.operator.config.KubernetesOperatorConfigOptions.SNAPSHOT_RESOURCE_ENABLED;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT;
+import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT;
+
+/** Utilities class for FlinkStateSnapshot resources. */
+public class FlinkStateSnapshotUtils {
+
+ /**
+ * From a snapshot reference, return its snapshot path. If a {@link FlinkStateSnapshot} is
+ * referenced, it will be retrieved from Kubernetes.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param snapshotRef snapshot reference
+ * @return found savepoint path
+ */
+ public static String getValidatedFlinkStateSnapshotPath(
+ KubernetesClient kubernetesClient, FlinkStateSnapshotReference snapshotRef) {
+ if (StringUtils.isNotBlank(snapshotRef.getPath())) {
+ return snapshotRef.getPath();
+ }
+
+ if (StringUtils.isBlank(snapshotRef.getName())) {
+ throw new IllegalArgumentException(
+ String.format("Invalid snapshot name: %s", snapshotRef.getName()));
+ }
+
+ var result =
+ snapshotRef.getNamespace() == null
+ ? kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .withName(snapshotRef.getName())
+ .get()
+ : kubernetesClient
+ .resources(FlinkStateSnapshot.class)
+ .inNamespace(snapshotRef.getNamespace())
+ .withName(snapshotRef.getName())
+ .get();
+
+ if (result == null) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot find snapshot %s in namespace %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ // We can return the savepoint path if it's marked as completed without waiting for the
+ // reconciler to update its status.
+ if (result.getSpec().isSavepoint() && result.getSpec().getSavepoint().getAlreadyExists()) {
+ var path = result.getSpec().getSavepoint().getPath();
+ if (!StringUtils.isBlank(path)) {
+ return path;
+ }
+ }
+
+ if (COMPLETED != result.getStatus().getState()) {
+ throw new IllegalStateException(
+ String.format(
+ "Snapshot %s/%s is not complete yet.",
+ snapshotRef.getNamespace(), snapshotRef.getName()));
+ }
+
+ var path = result.getStatus().getPath();
+ if (StringUtils.isBlank(path)) {
+ throw new IllegalStateException(
+ String.format(
+ "Snapshot %s/%s path is incorrect: %s.",
+ snapshotRef.getNamespace(), snapshotRef.getName(), path));
+ }
+
+ return path;
+ }
+
+ protected static FlinkStateSnapshot createFlinkStateSnapshot(
+ KubernetesClient kubernetesClient,
+ String name,
+ FlinkStateSnapshotSpec spec,
+ SnapshotTriggerType triggerType) {
+ var metadata = new ObjectMeta();
+ metadata.setName(name);
+ metadata.getLabels().put(CrdConstants.LABEL_SNAPSHOT_TYPE, triggerType.name());
+
+ var snapshot = new FlinkStateSnapshot();
+ snapshot.setSpec(spec);
+ snapshot.setMetadata(metadata);
+
+ return kubernetesClient.resource(snapshot).create();
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param savepointPath savepoint path if any
+ * @param triggerType trigger type
+ * @param savepointFormatType format type
+ * @param disposeOnDelete should dispose of data on deletion
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createSavepointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ @Nullable String savepointPath,
+ SnapshotTriggerType triggerType,
+ SavepointFormatType savepointFormatType,
+ boolean disposeOnDelete) {
+ var savepointSpec =
+ SavepointSpec.builder()
+ .path(savepointPath)
+ .formatType(savepointFormatType)
+ .disposeOnDelete(disposeOnDelete)
+ .alreadyExists(triggerType == SnapshotTriggerType.UPGRADE)
+ .build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .savepoint(savepointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(SAVEPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Creates a checkpoint {@link FlinkStateSnapshot} resource on the Kubernetes cluster.
+ *
+ * @param kubernetesClient kubernetes client
+ * @param resource Flink resource associated
+ * @param checkpointType type of checkpoint
+ * @param triggerType trigger type
+ * @return created snapshot
+ */
+ public static FlinkStateSnapshot createCheckpointResource(
+ KubernetesClient kubernetesClient,
+ AbstractFlinkResource<?, ?> resource,
+ CheckpointType checkpointType,
+ SnapshotTriggerType triggerType) {
+ var checkpointSpec = CheckpointSpec.builder().checkpointType(checkpointType).build();
+
+ var snapshotSpec =
+ FlinkStateSnapshotSpec.builder()
+ .jobReference(JobReference.fromFlinkResource(resource))
+ .checkpoint(checkpointSpec)
+ .build();
+
+ var resourceName = getFlinkStateSnapshotName(CHECKPOINT, triggerType, resource);
+ return createFlinkStateSnapshot(kubernetesClient, resourceName, snapshotSpec, triggerType);
+ }
+
+ /**
+ * Based on job configuration and operator configuration, decide if {@link FlinkStateSnapshot}
+ * resources should be used or not. Operator configuration will disable the usage of the
+ * corresponding CRD was not installed on this Kubernetes cluster.
+ *
+ * @param operatorConfiguration operator config
+ * @param configuration job config
+ * @return true if snapshot resources should be created
+ */
+ public static boolean isSnapshotResourceEnabled(
+ FlinkOperatorConfiguration operatorConfiguration, Configuration configuration) {
+ return configuration.get(SNAPSHOT_RESOURCE_ENABLED)
+ && operatorConfiguration.isSnapshotResourcesEnabled();
+ }
+
+ /**
+ * Return a generated name for a {@link FlinkStateSnapshot} to be created.
+ *
+ * @param snapshotType type of snapshot
+ * @param triggerType trigger type of snapshot
+ * @param referencedResource referenced resource
+ * @return result name
+ */
+ public static String getFlinkStateSnapshotName(
+ SnapshotType snapshotType,
+ SnapshotTriggerType triggerType,
+ AbstractFlinkResource<?, ?> referencedResource) {
+ return String.format(
+ "%s-%s-%s-%s",
+ referencedResource.getMetadata().getName(),
+ snapshotType.name().toLowerCase(),
+ triggerType.name().toLowerCase(),
+ UUID.randomUUID());
Review Comment:
Hi @anupamaggarwal you are right, that makes more sense than a random UUID. I have pushed a commit for this.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1691974263
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -716,6 +695,48 @@ public CheckpointFetchResult fetchCheckpointInfo(
}
}
+ @Override
+ public CheckpointStatsResult fetchCheckpointStats(
+ String jobId, Long checkpointId, Configuration conf) {
+ try (RestClusterClient<String> clusterClient = getClusterClient(conf)) {
+ var checkpointStatusHeaders = CheckpointStatisticDetailsHeaders.getInstance();
+ var parameters = checkpointStatusHeaders.getUnresolvedMessageParameters();
+ parameters.jobPathParameter.resolve(JobID.fromHexString(jobId));
+
+ // This was needed because the parameter is protected
+ var checkpointIdPathParameter =
+ (CheckpointIdPathParameter) Iterables.getLast(parameters.getPathParameters());
+ checkpointIdPathParameter.resolve(checkpointId);
+
+ var response =
+ clusterClient.sendRequest(
+ checkpointStatusHeaders, parameters, EmptyRequestBody.getInstance());
+
+ var stats = response.get();
+ if (stats == null) {
+ throw new IllegalStateException("Checkpoint ID %d for job %s does not exist!");
+ } else if (stats instanceof CheckpointStatistics.CompletedCheckpointStatistics) {
+ return CheckpointStatsResult.completed(
+ ((CheckpointStatistics.CompletedCheckpointStatistics) stats)
+ .getExternalPath());
+ } else if (stats instanceof CheckpointStatistics.FailedCheckpointStatistics) {
+ return CheckpointStatsResult.error(
+ ((CheckpointStatistics.FailedCheckpointStatistics) stats)
+ .getFailureMessage());
+ } else if (stats instanceof CheckpointStatistics.PendingCheckpointStatistics) {
+ return CheckpointStatsResult.pending();
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown checkpoint statistics result class: %s",
+ stats.getClass().getSimpleName()));
+ }
+ } catch (Exception e) {
+ LOG.error("Exception while fetching checkpoint statistics", e);
Review Comment:
This commit made it so we ignore errors when fetching checkpoint stats, since at that point we have determined that the checkpoint was successful anyways. We just set an empty path: https://github.com/apache/flink-kubernetes-operator/pull/821/commits/637b3053b9644b3d6cadd1fb6a05e1f5aab75fa6
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#issuecomment-2271266449
Thank you @gyfora , I have done a lot of manual testing and I am quite happy with the current state of the PR. Once this is merged, I will start working on the other tickets in this FLIP, and create new PRs hopefully this week for the following:
- Docs
- E2E test case
- Snapshot history automatic cleanup
- FlinkStateSnapshot metrics integration
I will be unavailable for ~3 weeks starting from Aug 23, I hope to have the remaining PRs ready and in good state until then.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667256876
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkStateSnapshotStatus.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.api.status;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.diff.Diffable;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import io.fabric8.crd.generator.annotation.PrinterColumn;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+/** Last observed status of the Flink state snapshot. */
+@Experimental
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@EqualsAndHashCode
+@ToString
Review Comment:
see other comments
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java:
##########
@@ -410,6 +430,14 @@ public static String operatorConfigKey(String key) {
.withDescription(
"Max number of reconcile loops triggered within the rate limiter refresh period for each resource. Setting the limit <= 0 disables the limiter.");
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<Boolean> OPERATOR_JOB_UPGRADE_SAVEPOINT_DISPOSE_ON_DELETE =
+ operatorConfig("job.upgrade.savepoint.dispose-on-delete")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Savepoint data created by upgrade savepoints as FlinkStateSnapshot resources will be disposed of automatically when the Kubernetes resource gets deleted.");
+
Review Comment:
I think we should have a single dispose-on-delete config that generally applies to snapshot resources.
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/FlinkStateSnapshotSpec.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.api.spec;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.diff.Diffable;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+/** Spec that describes a FlinkStateSnapshot. */
+@Experimental
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode
+@ToString
Review Comment:
Data covers equals and toString
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/spec/CheckpointSpec.java:
##########
@@ -0,0 +1,41 @@
+/*
+ * 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.api.spec;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.status.CheckpointType;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+
+/** Spec for checkpoint state snapshots. */
+@Experimental
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+@EqualsAndHashCode
Review Comment:
`@Data` already adds EqualsAndHashCode, you probably copied this from other classes (so maybe I messed up first :D) but it should not be necessary
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/CommonStatus.java:
##########
@@ -27,11 +27,13 @@
import io.fabric8.crd.generator.annotation.PrinterColumn;
import lombok.AllArgsConstructor;
import lombok.Data;
+import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import org.apache.commons.lang3.StringUtils;
/** Last observed common status of the Flink deployment/Flink SessionJob. */
+@EqualsAndHashCode
Review Comment:
Data should cover this, did you hit some issue without it?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSavepointState(ctx);
+ }
+ }
+
+ private void observeSavepointState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ LOG.info("Observing savepoint state for resource {}...", resourceName);
+
+ if (StringUtils.isEmpty(triggerId)) {
+ LOG.debug("Trigger ID is not set for savepoint {} yet.", resourceName);
Review Comment:
we could remove this and move the Observing log after the if?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSavepointState(ctx);
+ }
+ }
+
+ private void observeSavepointState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ LOG.info("Observing savepoint state for resource {}...", resourceName);
Review Comment:
This should be debug
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSavepointState(ctx);
+ }
+ }
+
+ private void observeSavepointState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ LOG.info("Observing savepoint state for resource {}...", resourceName);
+
+ if (StringUtils.isEmpty(triggerId)) {
+ LOG.debug("Trigger ID is not set for savepoint {} yet.", resourceName);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s for savepoint %s was not found",
+ resource.getSpec().getJobReference(),
+ resourceName)));
+ var jobId = secondaryResource.getStatus().getJobStatus().getJobId();
+
+ var ctxFlinkDeployment =
+ ctxFactory.getResourceContext(
+ ctx.getReferencedJobFlinkDeployment(), ctx.getJosdkContext());
+ var observeConfig = ctx.getReferencedJobObserveConfig();
+
+ if (!ReconciliationUtils.isJobRunning(secondaryResource.getStatus())) {
+ snapshotAbandoned(resource);
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()) {
+ var savepointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchSavepointInfo(triggerId, jobId, observeConfig);
+ handleSavepoint(ctx, savepointInfo);
+ } else {
+ var checkpointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchCheckpointInfo(triggerId, jobId, observeConfig);
+ handleCheckpoint(ctx, checkpointInfo, ctxFlinkDeployment, jobId);
+ }
+ }
+
+ private void handleSavepoint(
+ FlinkStateSnapshotContext ctx, SavepointFetchResult savepointInfo) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+
+ if (savepointInfo.isPending()) {
+ LOG.debug(
+ "Savepoint '{}' with ID {} is pending",
+ resourceName,
+ resource.getStatus().getTriggerId());
+ } else if (savepointInfo.getError() != null) {
+ LOG.error("Savepoint {} failed: {}", resourceName, savepointInfo.getError());
Review Comment:
Unnecessary as there is an event for this already
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSavepointState(ctx);
+ }
+ }
+
+ private void observeSavepointState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ LOG.info("Observing savepoint state for resource {}...", resourceName);
+
+ if (StringUtils.isEmpty(triggerId)) {
+ LOG.debug("Trigger ID is not set for savepoint {} yet.", resourceName);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s for savepoint %s was not found",
+ resource.getSpec().getJobReference(),
+ resourceName)));
+ var jobId = secondaryResource.getStatus().getJobStatus().getJobId();
+
+ var ctxFlinkDeployment =
+ ctxFactory.getResourceContext(
+ ctx.getReferencedJobFlinkDeployment(), ctx.getJosdkContext());
+ var observeConfig = ctx.getReferencedJobObserveConfig();
+
+ if (!ReconciliationUtils.isJobRunning(secondaryResource.getStatus())) {
+ snapshotAbandoned(resource);
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()) {
+ var savepointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchSavepointInfo(triggerId, jobId, observeConfig);
+ handleSavepoint(ctx, savepointInfo);
+ } else {
+ var checkpointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchCheckpointInfo(triggerId, jobId, observeConfig);
+ handleCheckpoint(ctx, checkpointInfo, ctxFlinkDeployment, jobId);
+ }
+ }
+
+ private void handleSavepoint(
+ FlinkStateSnapshotContext ctx, SavepointFetchResult savepointInfo) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+
+ if (savepointInfo.isPending()) {
+ LOG.debug(
+ "Savepoint '{}' with ID {} is pending",
+ resourceName,
+ resource.getStatus().getTriggerId());
+ } else if (savepointInfo.getError() != null) {
+ LOG.error("Savepoint {} failed: {}", resourceName, savepointInfo.getError());
+ snapshotFailed(ctx.getKubernetesClient(), resource, savepointInfo.getError());
+ } else {
+ LOG.info("Savepoint {} successful: {}", resourceName, savepointInfo.getLocation());
+ snapshotSuccessful(resource, savepointInfo.getLocation());
+ }
+ }
+
+ private void handleCheckpoint(
+ FlinkStateSnapshotContext ctx,
+ CheckpointFetchResult checkpointInfo,
+ FlinkResourceContext<FlinkDeployment> ctxFlinkDeployment,
+ String jobId) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+
+ if (checkpointInfo.isPending()) {
+ LOG.debug(
+ "Checkpoint for {} with ID {} is pending",
+ resourceName,
+ resource.getStatus().getTriggerId());
+ return;
+ }
+
+ if (checkpointInfo.getError() != null) {
+ LOG.error("Checkpoint {} failed: {}", resourceName, checkpointInfo.getError());
+ snapshotFailed(ctx.getKubernetesClient(), resource, checkpointInfo.getError());
+ } else {
+ LOG.info(
+ "Checkpoint {} was successful, querying final checkpoint path...",
+ resourceName);
Review Comment:
I think this could be debug or removed
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,201 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSavepointState(ctx);
+ }
+ }
+
+ private void observeSavepointState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ LOG.info("Observing savepoint state for resource {}...", resourceName);
+
+ if (StringUtils.isEmpty(triggerId)) {
+ LOG.debug("Trigger ID is not set for savepoint {} yet.", resourceName);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s for savepoint %s was not found",
+ resource.getSpec().getJobReference(),
+ resourceName)));
+ var jobId = secondaryResource.getStatus().getJobStatus().getJobId();
+
+ var ctxFlinkDeployment =
+ ctxFactory.getResourceContext(
+ ctx.getReferencedJobFlinkDeployment(), ctx.getJosdkContext());
+ var observeConfig = ctx.getReferencedJobObserveConfig();
+
+ if (!ReconciliationUtils.isJobRunning(secondaryResource.getStatus())) {
+ snapshotAbandoned(resource);
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()) {
+ var savepointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchSavepointInfo(triggerId, jobId, observeConfig);
+ handleSavepoint(ctx, savepointInfo);
+ } else {
+ var checkpointInfo =
+ ctxFlinkDeployment
+ .getFlinkService()
+ .fetchCheckpointInfo(triggerId, jobId, observeConfig);
+ handleCheckpoint(ctx, checkpointInfo, ctxFlinkDeployment, jobId);
+ }
+ }
+
+ private void handleSavepoint(
+ FlinkStateSnapshotContext ctx, SavepointFetchResult savepointInfo) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+
+ if (savepointInfo.isPending()) {
+ LOG.debug(
+ "Savepoint '{}' with ID {} is pending",
+ resourceName,
+ resource.getStatus().getTriggerId());
+ } else if (savepointInfo.getError() != null) {
+ LOG.error("Savepoint {} failed: {}", resourceName, savepointInfo.getError());
+ snapshotFailed(ctx.getKubernetesClient(), resource, savepointInfo.getError());
+ } else {
+ LOG.info("Savepoint {} successful: {}", resourceName, savepointInfo.getLocation());
+ snapshotSuccessful(resource, savepointInfo.getLocation());
+ }
+ }
+
+ private void handleCheckpoint(
+ FlinkStateSnapshotContext ctx,
+ CheckpointFetchResult checkpointInfo,
+ FlinkResourceContext<FlinkDeployment> ctxFlinkDeployment,
+ String jobId) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+
+ if (checkpointInfo.isPending()) {
+ LOG.debug(
+ "Checkpoint for {} with ID {} is pending",
+ resourceName,
+ resource.getStatus().getTriggerId());
+ return;
+ }
+
+ if (checkpointInfo.getError() != null) {
+ LOG.error("Checkpoint {} failed: {}", resourceName, checkpointInfo.getError());
Review Comment:
Unnecessary as we trigger an event which logs this again
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667372364
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java:
##########
@@ -222,9 +235,12 @@ private void updateStatusBeforeFirstDeployment(
CR cr, SPEC spec, Configuration deployConfig, STATUS status, KubernetesClient client) {
if (spec.getJob() != null) {
var initialUpgradeMode = UpgradeMode.STATELESS;
+ var snapshotRef = spec.getJob().getFlinkStateSnapshotReference();
var initialSp = spec.getJob().getInitialSavepointPath();
- if (initialSp != null) {
+ if (snapshotRef != null) {
+ initialUpgradeMode = UpgradeMode.SAVEPOINT;
+ } else if (initialSp != null) {
status.getJobStatus()
.getSavepointInfo()
.setLastSavepoint(Savepoint.of(initialSp, SnapshotTriggerType.UNKNOWN));
Review Comment:
shouldn't the last savepoint info be updated in case the snapshot ref is provided? This logic is there to handle failures before the initial startup so should apply to both
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) throws Exception {
+ var source = ctx.getResource().getSpec().getJobReference();
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!FlinkStateSnapshotState.TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+ resource.getStatus().setState(FlinkStateSnapshotState.COMPLETED);
+ resource.getStatus().setPath(resource.getSpec().getSavepoint().getPath());
+ var time = DateTimeUtils.kubernetes(Instant.now());
+ resource.getStatus().setTriggerTimestamp(time);
+ resource.getStatus().setResultTimestamp(time);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s not found",
+ source)));
Review Comment:
Should this just abandon the snapshot?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -329,18 +459,26 @@ private void redeployWithSavepoint(
throws Exception {
LOG.info("Redeploying from savepoint");
cancelJob(ctx, UpgradeMode.STATELESS);
- var savepoint = currentDeploySpec.getJob().getInitialSavepointPath();
+ var snapshotRef = currentDeploySpec.getJob().getFlinkStateSnapshotReference();
currentDeploySpec.getJob().setUpgradeMode(UpgradeMode.SAVEPOINT);
+
+ var path = currentDeploySpec.getJob().getInitialSavepointPath();
+ if (snapshotRef != null && snapshotRef.getName() != null) {
+ path =
+ FlinkStateSnapshotUtils.getAndValidateFlinkStateSnapshotPath(
+ ctx.getKubernetesClient(), snapshotRef);
+ }
+
status.getJobStatus()
.getSavepointInfo()
- .setLastSavepoint(Savepoint.of(savepoint, SnapshotTriggerType.UNKNOWN));
+ .setLastSavepoint(Savepoint.of(path, SnapshotTriggerType.UNKNOWN));
Review Comment:
See my other long comment
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractJobReconciler.java:
##########
@@ -266,19 +304,31 @@ protected void restoreJob(
Optional<String> savepointOpt = Optional.empty();
if (spec.getJob().getUpgradeMode() != UpgradeMode.STATELESS) {
- savepointOpt =
- Optional.ofNullable(
- ctx.getResource()
- .getStatus()
- .getJobStatus()
- .getSavepointInfo()
- .getLastSavepoint())
- .flatMap(s -> Optional.ofNullable(s.getLocation()));
+ if (FlinkStateSnapshotUtils.shouldCreateSnapshotResource(
+ ctx.getOperatorConfig(), deployConfig)) {
+ savepointOpt = getLatestSavepointPathFromFlinkStateSnapshots(ctx);
+ } else {
+ savepointOpt =
+ Optional.ofNullable(
+ ctx.getResource()
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .getLastSavepoint())
+ .flatMap(s -> Optional.ofNullable(s.getLocation()));
Review Comment:
The problem with this logic is that the snapshotReference field allows us to specify the path directly. So if the user starts the job using that, in the code we don't update the lastSavepointInfo but we also won't find that path here. So we will simply restart from empty state in those cases which is not what should happen.
2 main ways to solve it:
1. We keep the current lastSavepoint logic in the status and update it whenever a snapshot is completed. This is a bit problematic because we would have to update the status of the FlinkDeployment/SessionJob from the snapshot controller, which is a bit weird.
2. Introduce a new field in the status to capture the savepoint reference used during a Savepoint deployment. I think this is the way to go because this is the functionality that is important to capture. When we submit with initialSavepointPath the first time / savepoint redeploy / savepoint upgrade, in the lastReconciledSpec we have upgradeMode: SAVEPOINT recorded and it is assumed that the lastSavepoint contains the savepoint used. Now we are getting rid of the lastSavepoint so we should put something more meaningful in place in the status to capture the actual upgrade related behaviour. One possibility here is simply to use the `job.snapshotReference` field in the lastReconciledSpec to capture the upgrade savepoint and clear it in other cases (same way of how we reuse the upgradeMode field to capture how we actually stopped the job in the lastReconciledSpec)
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/ApplicationReconciler.java:
##########
@@ -226,7 +230,36 @@ private void setJobIdIfNecessary(
@Override
protected void cancelJob(FlinkResourceContext<FlinkDeployment> ctx, UpgradeMode upgradeMode)
throws Exception {
- ctx.getFlinkService().cancelJob(ctx.getResource(), upgradeMode, ctx.getObserveConfig());
+ var conf = ctx.getObserveConfig() != null ? ctx.getObserveConfig() : new Configuration();
+ var savepointFormatType =
+ conf.get(KubernetesOperatorConfigOptions.OPERATOR_SAVEPOINT_FORMAT_TYPE);
+
+ var savepointOpt = ctx.getFlinkService().cancelJob(ctx.getResource(), upgradeMode, conf);
+ savepointOpt.ifPresent(
+ location -> {
+ if (FlinkStateSnapshotUtils.shouldCreateSnapshotResource(
+ ctx.getOperatorConfig(), conf)) {
+ FlinkStateSnapshotUtils.createUpgradeSavepointResource(
+ ctx.getKubernetesClient(),
+ ctx.getResource(),
+ location,
+ SavepointFormatType.valueOf(savepointFormatType.name()),
+ conf.get(
+ KubernetesOperatorConfigOptions
+ .OPERATOR_JOB_SAVEPOINT_DISPOSE_ON_DELETE));
+ } else {
+ Savepoint sp =
+ Savepoint.of(
+ location,
+ SnapshotTriggerType.UPGRADE,
+ SavepointFormatType.valueOf(savepointFormatType.name()));
+ ctx.getResource()
+ .getStatus()
+ .getJobStatus()
+ .getSavepointInfo()
+ .updateLastSavepoint(sp);
Review Comment:
see my other long comment
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/snapshot/StateSnapshotObserver.java:
##########
@@ -0,0 +1,197 @@
+/*
+ * 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.observer.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.observer.CheckpointFetchResult;
+import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+
+/** The observer of {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotObserver {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotObserver.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void observe(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var savepointState = resource.getStatus().getState();
+
+ if (FlinkStateSnapshotState.IN_PROGRESS.equals(savepointState)) {
+ observeSavepointState(ctx);
+ }
+ }
+
+ private void observeSavepointState(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ var resourceName = resource.getMetadata().getName();
+ var triggerId = resource.getStatus().getTriggerId();
+
+ if (StringUtils.isEmpty(triggerId)) {
+ return;
+ }
+
+ LOG.debug("Observing savepoint state for resource {}...", resourceName);
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s for savepoint %s was not found",
+ resource.getSpec().getJobReference(),
+ resourceName)));
Review Comment:
If the target is not found anymore to me that sounds similar to the not running case, and probably should be marked abandoned. What is your thinking here?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) throws Exception {
+ var source = ctx.getResource().getSpec().getJobReference();
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!FlinkStateSnapshotState.TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+ resource.getStatus().setState(FlinkStateSnapshotState.COMPLETED);
+ resource.getStatus().setPath(resource.getSpec().getSavepoint().getPath());
+ var time = DateTimeUtils.kubernetes(Instant.now());
+ resource.getStatus().setTriggerTimestamp(time);
+ resource.getStatus().setResultTimestamp(time);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s not found",
+ source)));
+ if (!ReconciliationUtils.isJobRunning(secondaryResource.getStatus())) {
+ LOG.warn(
+ "Target job {} for savepoint {} is not running, cannot trigger snapshot.",
+ secondaryResource.getMetadata().getName(),
+ resource.getMetadata().getName());
+ return;
+ }
Review Comment:
we have already observed and abandoned the snapshot in this case right?
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667680584
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) throws Exception {
+ var source = ctx.getResource().getSpec().getJobReference();
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!FlinkStateSnapshotState.TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+ resource.getStatus().setState(FlinkStateSnapshotState.COMPLETED);
+ resource.getStatus().setPath(resource.getSpec().getSavepoint().getPath());
+ var time = DateTimeUtils.kubernetes(Instant.now());
+ resource.getStatus().setTriggerTimestamp(time);
+ resource.getStatus().setResultTimestamp(time);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s not found",
+ source)));
+ if (!ReconciliationUtils.isJobRunning(secondaryResource.getStatus())) {
+ LOG.warn(
+ "Target job {} for savepoint {} is not running, cannot trigger snapshot.",
+ secondaryResource.getMetadata().getName(),
+ resource.getMetadata().getName());
+ return;
+ }
Review Comment:
I have added a commit which will make sure that we abandon snapshots if referenced Flink resource is not found, or the job is not running, in both observe and reconcile phase. An event will also be generated for the snapshot resource in these cases.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667390128
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) throws Exception {
+ var source = ctx.getResource().getSpec().getJobReference();
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!FlinkStateSnapshotState.TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+ resource.getStatus().setState(FlinkStateSnapshotState.COMPLETED);
+ resource.getStatus().setPath(resource.getSpec().getSavepoint().getPath());
+ var time = DateTimeUtils.kubernetes(Instant.now());
+ resource.getStatus().setTriggerTimestamp(time);
+ resource.getStatus().setResultTimestamp(time);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s not found",
+ source)));
+ if (!ReconciliationUtils.isJobRunning(secondaryResource.getStatus())) {
+ LOG.warn(
+ "Target job {} for savepoint {} is not running, cannot trigger snapshot.",
+ secondaryResource.getMetadata().getName(),
+ resource.getMetadata().getName());
+ return;
+ }
Review Comment:
The observer will only observe snapshots that are `IN_PROGRESS`, while the reconciler will only reconcile snapshots that have `TRIGGER_PENDING` state, so it's possible that the job is not running 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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667389539
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) throws Exception {
+ var source = ctx.getResource().getSpec().getJobReference();
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!FlinkStateSnapshotState.TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+ resource.getStatus().setState(FlinkStateSnapshotState.COMPLETED);
+ resource.getStatus().setPath(resource.getSpec().getSavepoint().getPath());
+ var time = DateTimeUtils.kubernetes(Instant.now());
+ resource.getStatus().setTriggerTimestamp(time);
+ resource.getStatus().setResultTimestamp(time);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s not found",
+ source)));
Review Comment:
This feels a bit weird, since after the observe phase, the code should not be able to run into this at all. I think it would help keep the code cleaner if we only abandon snapshots from the observer.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667389539
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) throws Exception {
+ var source = ctx.getResource().getSpec().getJobReference();
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!FlinkStateSnapshotState.TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+ resource.getStatus().setState(FlinkStateSnapshotState.COMPLETED);
+ resource.getStatus().setPath(resource.getSpec().getSavepoint().getPath());
+ var time = DateTimeUtils.kubernetes(Instant.now());
+ resource.getStatus().setTriggerTimestamp(time);
+ resource.getStatus().setResultTimestamp(time);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s not found",
+ source)));
Review Comment:
This feels a bit weird, since after the observe phase, the code should not be able to run into this at all. I think it would help keep the code cleaner if we only abandon snapshots from the observer.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "gyfora (via GitHub)" <gi...@apache.org>.
gyfora commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1666495408
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/deployment/AbstractFlinkResourceReconciler.java:
##########
@@ -122,12 +127,20 @@ public void reconcile(FlinkResourceContext<CR> ctx) throws Exception {
var deployConfig = ctx.getDeployConfig(spec);
updateStatusBeforeFirstDeployment(
cr, spec, deployConfig, status, ctx.getKubernetesClient());
- deploy(
- ctx,
- spec,
- deployConfig,
- Optional.ofNullable(spec.getJob()).map(JobSpec::getInitialSavepointPath),
- false);
+
+ Optional<String> savepointPathOpt =
+ Optional.ofNullable(spec.getJob()).map(JobSpec::getInitialSavepointPath);
+ if (spec.getJob() != null && spec.getJob().getFlinkStateSnapshotReference() != null) {
+ var snapshotRef = spec.getJob().getFlinkStateSnapshotReference();
+ if (snapshotRef.getName() != null) {
+ savepointPathOpt =
+ Optional.of(
+ FlinkStateSnapshotUtils.getAndValidateFlinkStateSnapshotPath(
+ ctx.getKubernetesClient(), snapshotRef));
+ }
+ }
Review Comment:
At this point we should capture this in a method
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/crd/CustomResourceDefinitionWatcherImpl.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.crd;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/** Default implementation for CustomResourceDefinitionWatcher. */
+@RequiredArgsConstructor
+public class CustomResourceDefinitionWatcherImpl implements CustomResourceDefinitionWatcher {
Review Comment:
The ` CustomResourceDefinitionWatcher` interface and the way we use it feels a bit overly complicated for the simple purpose we want. My recommendation would be to remove the generic interface and replace it with some specific utility that checks the Snapshot CRD and based on this we simply set a boolean base config in the config manager.
This way we can get if it's enabled from the operator config instead of having to pass and use the watcher from the controllers.
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSavepoint(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
Review Comment:
I think for FlinkStateSnapshot resources we should avoid using the custom status patching. We can just rely on the JOSDK built in Update control mechanism.
We are using a custom patching logic for the deployments because we are executing destructive actions before which we must record the status in some cases. Here I believe it's not the case in general.
##########
flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/status/FlinkStateSnapshotStatus.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.api.status;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.kubernetes.operator.api.diff.Diffable;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import io.fabric8.crd.generator.annotation.PrinterColumn;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+import lombok.experimental.SuperBuilder;
+
+/** Last observed status of the Flink state snapshot. */
+@Experimental
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+@EqualsAndHashCode
+@ToString
+@SuperBuilder
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class FlinkStateSnapshotStatus implements Diffable<FlinkStateSnapshotStatus> {
+
+ /** Current state of the snapshot. */
+ @PrinterColumn(name = "Snapshot Status")
Review Comment:
Can we simply call it `State` ?
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java:
##########
@@ -0,0 +1,207 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.observer.snapshot.StateSnapshotObserver;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.reconciler.snapshot.StateSnapshotReconciler;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.EventSourceUtils;
+import org.apache.flink.kubernetes.operator.utils.StatusRecorder;
+import org.apache.flink.kubernetes.operator.validation.FlinkResourceValidator;
+
+import io.javaoperatorsdk.operator.api.reconciler.Cleaner;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+/** Controller that runs the main reconcile loop for {@link FlinkStateSnapshot}. */
+@ControllerConfiguration
+public class FlinkStateSnapshotController
+ implements Reconciler<FlinkStateSnapshot>,
+ ErrorStatusHandler<FlinkStateSnapshot>,
+ EventSourceInitializer<FlinkStateSnapshot>,
+ Cleaner<FlinkStateSnapshot> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(FlinkStateSnapshotController.class);
+
+ private final Set<FlinkResourceValidator> validators;
+ private final FlinkResourceContextFactory ctxFactory;
+ private final StateSnapshotReconciler reconciler;
+ private final StateSnapshotObserver observer;
+ private final EventRecorder eventRecorder;
+ private final StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder;
+
+ public FlinkStateSnapshotController(
+ Set<FlinkResourceValidator> validators,
+ FlinkResourceContextFactory ctxFactory,
+ StateSnapshotReconciler reconciler,
+ StateSnapshotObserver observer,
+ EventRecorder eventRecorder,
+ StatusRecorder<FlinkStateSnapshot, FlinkStateSnapshotStatus> statusRecorder) {
+ this.validators = validators;
+ this.ctxFactory = ctxFactory;
+ this.reconciler = reconciler;
+ this.observer = observer;
+ this.eventRecorder = eventRecorder;
+ this.statusRecorder = statusRecorder;
+ }
+
+ @Override
+ public UpdateControl<FlinkStateSnapshot> reconcile(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ statusRecorder.updateStatusFromCache(flinkStateSnapshot);
+
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+
+ // observe
+ observer.observe(ctx);
+
+ // validate
+ if (!validateSavepoint(ctx)) {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ UpdateControl<FlinkStateSnapshot> updateControl = UpdateControl.noUpdate();
+ return updateControl.rescheduleAfter(
+ ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ // reconcile
+ try {
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ reconciler.reconcile(ctx);
+ } catch (Exception e) {
+ eventRecorder.triggerSnapshotEvent(
+ flinkStateSnapshot,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.SnapshotError,
+ EventRecorder.Component.Snapshot,
+ e.getMessage(),
+ josdkContext.getClient());
+ throw new ReconciliationException(e);
+ }
+
+ var updateControl = getUpdateControl(ctx);
+ statusRecorder.patchAndCacheStatus(flinkStateSnapshot, ctx.getKubernetesClient());
+ return updateControl;
+ }
+
+ @Override
+ public DeleteControl cleanup(
+ FlinkStateSnapshot flinkStateSnapshot, Context<FlinkStateSnapshot> josdkContext) {
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, josdkContext);
+ DeleteControl deleteControl;
+ try {
+ deleteControl = reconciler.cleanup(ctx);
+ } catch (Exception e) {
+ eventRecorder.triggerSnapshotEvent(
+ flinkStateSnapshot,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.CleanupFailed,
+ EventRecorder.Component.Snapshot,
+ e.getMessage(),
+ ctx.getKubernetesClient());
+ LOG.error(
+ "Error during cleanup of snapshot {}",
+ flinkStateSnapshot.getMetadata().getName(),
+ e);
+ return DeleteControl.noFinalizerRemoval()
+ .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ if (deleteControl.isRemoveFinalizer()) {
+ statusRecorder.removeCachedStatus(flinkStateSnapshot);
+ }
+ return deleteControl;
+ }
+
+ private UpdateControl<FlinkStateSnapshot> getUpdateControl(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+ if (FlinkStateSnapshotState.FAILED.equals(resource.getStatus().getState())) {
+ if (resource.getStatus().getFailures() > resource.getSpec().getBackoffLimit()) {
+ LOG.info(
+ "Snapshot {} failed and won't be retried as failure count exceeded the backoff limit",
+ resource.getMetadata().getName());
+ return UpdateControl.noUpdate();
+ } else {
+ long retrySeconds = 10L * (1L << resource.getStatus().getFailures() - 1);
+ LOG.info(
+ "Snapshot {} failed and will be retried in {} seconds...",
+ resource.getMetadata().getName(),
+ retrySeconds);
+ resource.getStatus().setState(FlinkStateSnapshotState.TRIGGER_PENDING);
+ return UpdateControl.<FlinkStateSnapshot>noUpdate()
+ .rescheduleAfter(Duration.ofSeconds(retrySeconds));
+ }
+ }
+ return UpdateControl.<FlinkStateSnapshot>noUpdate()
+ .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ }
+
+ @Override
+ public ErrorStatusUpdateControl<FlinkStateSnapshot> updateErrorStatus(
+ FlinkStateSnapshot flinkStateSnapshot,
+ Context<FlinkStateSnapshot> context,
+ Exception e) {
+ var ctx = ctxFactory.getFlinkStateSnapshotContext(flinkStateSnapshot, context);
+ return ReconciliationUtils.toErrorStatusUpdateControl(ctx, e, statusRecorder);
+ }
+
+ @Override
+ public Map<String, EventSource> prepareEventSources(
+ EventSourceContext<FlinkStateSnapshot> context) {
+ return EventSourceInitializer.nameEventSources(
+ EventSourceUtils.getFlinkStateSnapshotInformerEventSources(context));
+ }
+
+ private boolean validateSavepoint(FlinkStateSnapshotContext ctx) {
+ var savepoint = ctx.getResource();
+ for (var validator : validators) {
+ var validationError =
+ validator.validateStateSnapshot(savepoint, ctx.getSecondaryResource());
+ if (validationError.isPresent()) {
+ eventRecorder.triggerSnapshotEvent(
+ savepoint,
+ EventRecorder.Type.Warning,
+ EventRecorder.Reason.ValidationError,
+ EventRecorder.Component.Operator,
+ validationError.get(),
+ ctx.getKubernetesClient());
+ return true;
Review Comment:
we always seem to return "true" is this correct for the validation error case?
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1667680584
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.autoscaler.utils.DateTimeUtils;
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.CrdConstants;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotState;
+import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType;
+import org.apache.flink.kubernetes.operator.controller.FlinkResourceContext;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.util.Optional;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) throws Exception {
+ var source = ctx.getResource().getSpec().getJobReference();
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!FlinkStateSnapshotState.TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+ resource.getStatus().setState(FlinkStateSnapshotState.COMPLETED);
+ resource.getStatus().setPath(resource.getSpec().getSavepoint().getPath());
+ var time = DateTimeUtils.kubernetes(Instant.now());
+ resource.getStatus().setTriggerTimestamp(time);
+ resource.getStatus().setResultTimestamp(time);
+ return;
+ }
+
+ var secondaryResource =
+ ctx.getSecondaryResource()
+ .orElseThrow(
+ () ->
+ new RuntimeException(
+ String.format(
+ "Secondary resource %s not found",
+ source)));
+ if (!ReconciliationUtils.isJobRunning(secondaryResource.getStatus())) {
+ LOG.warn(
+ "Target job {} for savepoint {} is not running, cannot trigger snapshot.",
+ secondaryResource.getMetadata().getName(),
+ resource.getMetadata().getName());
+ return;
+ }
Review Comment:
I have added a commit which will make sure that we abandon snapshots if referenced Flink resource is not found, or the job is not running, in both observe and reconcile phase.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688100790
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/config/KubernetesOperatorConfigOptions.java:
##########
@@ -426,6 +438,14 @@ public static String operatorConfigKey(String key) {
.withDescription(
"Max allowed checkpoint age for initiating last-state upgrades on running jobs. If a checkpoint is not available within the desired age (and nothing in progress) a savepoint will be triggered.");
+ @Documentation.Section(SECTION_DYNAMIC)
+ public static final ConfigOption<Boolean> OPERATOR_JOB_SAVEPOINT_DISPOSE_ON_DELETE =
+ operatorConfig("savepoint.dispose-on-delete")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription(
+ "Savepoint data for FlinkStateSnapshot resources created by the operator during upgrades and periodic savepoints will be disposed of automatically when the generated Kubernetes resource is deleted.");
Review Comment:
I believe it's the correct way to say it (to dispose of 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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688262282
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotContext.java:
##########
@@ -0,0 +1,112 @@
+/*
+ * 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.controller;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.operator.api.AbstractFlinkResource;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkSessionJob;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.JobKind;
+import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus;
+import org.apache.flink.kubernetes.operator.config.FlinkConfigManager;
+import org.apache.flink.kubernetes.operator.config.FlinkOperatorConfiguration;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Optional;
+
+/** Context for reconciling a snapshot. */
+@Getter
+@RequiredArgsConstructor
+public class FlinkStateSnapshotContext {
+
+ private final FlinkStateSnapshot resource;
+ private final FlinkStateSnapshotStatus originalStatus;
+ private final Context<FlinkStateSnapshot> josdkContext;
+ private final FlinkConfigManager configManager;
+
+ private FlinkOperatorConfiguration operatorConfig;
+ private Configuration referencedJobObserveConfig;
+ private FlinkDeployment referencedJobFlinkDeployment;
+
+ /**
+ * @return Operator configuration for this resource.
+ */
+ public FlinkOperatorConfiguration getOperatorConfig() {
+ if (operatorConfig != null) {
+ return operatorConfig;
+ }
+ return operatorConfig =
+ configManager.getOperatorConfiguration(
+ getResource().getMetadata().getNamespace(), null);
Review Comment:
Added lazy getter as well, thank you!
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "anupamaggarwal (via GitHub)" <gi...@apache.org>.
anupamaggarwal commented on PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#issuecomment-2246776962
Thanks for contributing this super useful feature @mateczagany 👍 !.
I am still new to the operator codebase so pls feel free to ignore if something is not relevant, had some initial questions-
1) IIUC if a savepoint CR has a job reference (with an invalid job which does not exist/ is not running) we would set the status field to ABANDONED. Does it also make sense to forbid the snapshot CR creation altogether in these cases through the validation webhook ?
2) Would it be possible to add an example FlinkStateSnapshot CR to the examples folder ? (similar to [basic.yaml](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/try-flink-kubernetes-operator/quick-start/#submitting-a-flink-job) referenced in the quickstart guide). Pls ignore if you were planning on covering this as part of the documentation PR.
3) Does it also make sense to include some of the new configs (for example `snapshot.resource.enabled`) in `helm/*/conf/flink-conf.yaml` with the default values commented out? I was wondering if this would give more signal to first time users of this feature, but it seems we currently have a subset of operator properties exposed, so wasn't entirely sure.
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "mateczagany (via GitHub)" <gi...@apache.org>.
mateczagany commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1688472333
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java:
##########
@@ -716,6 +695,48 @@ public CheckpointFetchResult fetchCheckpointInfo(
}
}
+ @Override
+ public CheckpointStatsResult fetchCheckpointStats(
+ String jobId, Long checkpointId, Configuration conf) {
+ try (RestClusterClient<String> clusterClient = getClusterClient(conf)) {
+ var checkpointStatusHeaders = CheckpointStatisticDetailsHeaders.getInstance();
+ var parameters = checkpointStatusHeaders.getUnresolvedMessageParameters();
+ parameters.jobPathParameter.resolve(JobID.fromHexString(jobId));
+
+ // This was needed because the parameter is protected
+ var checkpointIdPathParameter =
+ (CheckpointIdPathParameter) Iterables.getLast(parameters.getPathParameters());
+ checkpointIdPathParameter.resolve(checkpointId);
+
+ var response =
+ clusterClient.sendRequest(
+ checkpointStatusHeaders, parameters, EmptyRequestBody.getInstance());
+
+ var stats = response.get();
+ if (stats == null) {
+ throw new IllegalStateException("Checkpoint ID %d for job %s does not exist!");
+ } else if (stats instanceof CheckpointStatistics.CompletedCheckpointStatistics) {
+ return CheckpointStatsResult.completed(
+ ((CheckpointStatistics.CompletedCheckpointStatistics) stats)
+ .getExternalPath());
+ } else if (stats instanceof CheckpointStatistics.FailedCheckpointStatistics) {
+ return CheckpointStatsResult.error(
+ ((CheckpointStatistics.FailedCheckpointStatistics) stats)
+ .getFailureMessage());
+ } else if (stats instanceof CheckpointStatistics.PendingCheckpointStatistics) {
+ return CheckpointStatsResult.pending();
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Unknown checkpoint statistics result class: %s",
+ stats.getClass().getSimpleName()));
+ }
+ } catch (Exception e) {
+ LOG.error("Exception while fetching checkpoint statistics", e);
Review Comment:
We should handle cases where the checkpoint statistics are no longer stored on the web server, and we get the following error:
```
Could not find checkpointing statistics for checkpoint 243.
```
--
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
Re: [PR] [FLINK-35265] Implement FlinkStateSnapshot custom resource [flink-kubernetes-operator]
Posted by "anupamaggarwal (via GitHub)" <gi...@apache.org>.
anupamaggarwal commented on code in PR #821:
URL: https://github.com/apache/flink-kubernetes-operator/pull/821#discussion_r1691872393
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/snapshot/StateSnapshotReconciler.java:
##########
@@ -0,0 +1,230 @@
+/*
+ * 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.reconciler.snapshot;
+
+import org.apache.flink.configuration.CheckpointingOptions;
+import org.apache.flink.kubernetes.operator.api.FlinkDeployment;
+import org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot;
+import org.apache.flink.kubernetes.operator.api.spec.FlinkStateSnapshotSpec;
+import org.apache.flink.kubernetes.operator.controller.FlinkStateSnapshotContext;
+import org.apache.flink.kubernetes.operator.exception.ReconciliationException;
+import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.kubernetes.operator.service.FlinkResourceContextFactory;
+import org.apache.flink.kubernetes.operator.utils.EventRecorder;
+import org.apache.flink.kubernetes.operator.utils.FlinkStateSnapshotUtils;
+import org.apache.flink.kubernetes.operator.utils.SnapshotUtils;
+import org.apache.flink.util.Preconditions;
+
+import io.javaoperatorsdk.operator.api.reconciler.DeleteControl;
+import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.TRIGGER_PENDING;
+
+/** The reconciler for the {@link org.apache.flink.kubernetes.operator.api.FlinkStateSnapshot}. */
+@RequiredArgsConstructor
+public class StateSnapshotReconciler {
+
+ private static final Logger LOG = LoggerFactory.getLogger(StateSnapshotReconciler.class);
+
+ private final FlinkResourceContextFactory ctxFactory;
+ private final EventRecorder eventRecorder;
+
+ public void reconcile(FlinkStateSnapshotContext ctx) {
+ var resource = ctx.getResource();
+
+ var savepointState = resource.getStatus().getState();
+ if (!TRIGGER_PENDING.equals(savepointState)) {
+ return;
+ }
+
+ if (resource.getSpec().isSavepoint()
+ && resource.getSpec().getSavepoint().getAlreadyExists()) {
+ LOG.info(
+ "Snapshot {} is marked as completed in spec, skipping triggering savepoint.",
+ resource.getMetadata().getName());
+
+ FlinkStateSnapshotUtils.snapshotSuccessful(
+ resource, resource.getSpec().getSavepoint().getPath(), true);
+ return;
+ }
+
+ if (FlinkStateSnapshotUtils.abandonSnapshotIfJobNotRunning(
+ ctx.getKubernetesClient(),
+ ctx.getResource(),
+ ctx.getSecondaryResource().orElse(null),
+ eventRecorder)) {
+ return;
+ }
+
+ var jobId = ctx.getSecondaryResource().orElseThrow().getStatus().getJobStatus().getJobId();
+
+ Optional<String> triggerIdOpt;
+ try {
+ triggerIdOpt = triggerCheckpointOrSavepoint(resource.getSpec(), ctx, jobId);
+ } catch (Exception e) {
+ LOG.error("Failed to trigger snapshot for resource {}", ctx.getResource(), e);
+ throw new ReconciliationException(e);
+ }
+
+ if (triggerIdOpt.isEmpty()) {
+ LOG.warn("Failed to trigger snapshot {}", resource.getMetadata().getName());
+ return;
+ }
+
+ FlinkStateSnapshotUtils.snapshotInProgress(resource, triggerIdOpt.get());
+ }
+
+ public DeleteControl cleanup(FlinkStateSnapshotContext ctx) throws Exception {
+ var resource = ctx.getResource();
+ var state = resource.getStatus().getState();
+ var resourceName = resource.getMetadata().getName();
+ LOG.info("Cleaning up resource {}...", resourceName);
+
+ if (resource.getSpec().isCheckpoint()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (!resource.getSpec().getSavepoint().getDisposeOnDelete()) {
+ return DeleteControl.defaultDelete();
+ }
+ if (resource.getSpec().getJobReference() == null
+ || resource.getSpec().getJobReference().getName() == null) {
+ return DeleteControl.defaultDelete();
+ }
+
+ switch (state) {
+ case IN_PROGRESS:
+ LOG.info(
+ "Cannot delete resource {} yet as savepoint is still in progress...",
+ resourceName);
+ return DeleteControl.noFinalizerRemoval()
+ .rescheduleAfter(ctx.getOperatorConfig().getReconcileInterval().toMillis());
+ case COMPLETED:
+ var flinkDeployment = getFlinkDeployment(ctx);
+ return handleSnapshotCleanup(resource, flinkDeployment, ctx);
Review Comment:
I was trying out the periodic snapshot feature today. Does it make sense to introduce a feature to auto-cleanup the `flinkstatesnapshots.flink.apache.org` CRs and only retain upto some max retention limit (we could still retain the savepoint dir?) IIUC `disposeOnDelete` for periodic savepoints is set to false by default (so it should be relatively safe to do so).
If the user sets the savepoint interval `kubernetes.operator.periodic.savepoint.interval` to a low value, and there are multiple jobs on the cluster it would lead to creation of many snapshot CRs. I am not familiar with K8s internals but could this increase the load on the API server?
--
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