[FLINK-36109][snapshot] Add labels to snapshot resources

Author: mateczagany

docs/content/docs/custom-resource/snapshots.md

@@ -127,7 +127,6 @@ This however requires the referenced Flink resource to be alive, as this operati
 
 This feature is not available for checkpoints.
 
-
 ## Triggering snapshots
 
 Upgrade savepoints are triggered automatically by the system during the upgrade process as we have seen in the previous sections.
@@ -208,3 +207,19 @@ Legacy savepoints found in FlinkDeployment/FlinkSessionJob CRs under the depreca
 - For max count and FlinkStateSnapshot resources **disabled**, it will be cleaned up when `savepointHistory` exceeds max count
 - For max count and FlinkStateSnapshot resources **enabled**, it will be cleaned up when `savepointHistory` + number of FlinkStateSnapshot CRs related to the job exceed max count
 
+
+## Advanced Snapshot Filtering
+
+At the end of each snapshot reconciliation phase, the operator will update its labels to reflect the latest status and spec of the resources.
+This will allow the Kubernetes API server to filter snapshots without having to query all resources, since filtering by status or spec fields of custom resources is not supported in Kubernetes by default.
+Example queries with label selectors using `kubectl`:
+```shell
+# Query all checkpoints
+kubectl -n flink get flinksnp -l 'snapshot.type=CHECKPOINT'
+
+# Query all savepoints with states
+kubectl -n flink get flinksnp -l 'snapshot.state in (COMPLETED,ABANDONED),snapshot.type=SAVEPOINT'
+
+# Query all savepoints/checkpoints with job reference
+kubectl -n flink get flinksnp -l 'job-reference.kind=FlinkDeployment,job-reference.name=test-job'
+```

flink-kubernetes-operator-api/src/main/java/org/apache/flink/kubernetes/operator/api/CrdConstants.java

@@ -30,4 +30,8 @@ public class CrdConstants {
     public static final String EPHEMERAL_STORAGE = "ephemeral-storage";
 
     public static final String LABEL_SNAPSHOT_TYPE = "snapshot.type";
+    public static final String LABEL_SNAPSHOT_TRIGGER_TYPE = "snapshot.trigger-type";
+    public static final String LABEL_SNAPSHOT_STATE = "snapshot.state";
+    public static final String LABEL_SNAPSHOT_JOB_REFERENCE_KIND = "job-reference.kind";
+    public static final String LABEL_SNAPSHOT_JOB_REFERENCE_NAME = "job-reference.name";
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotContext.java

@@ -33,6 +33,7 @@
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 
+import java.util.Map;
 import java.util.Optional;
 
 /** Context for reconciling a snapshot. */
@@ -42,6 +43,7 @@ public class FlinkStateSnapshotContext {
 
     private final FlinkStateSnapshot resource;
     private final FlinkStateSnapshotStatus originalStatus;
+    private final Map<String, String> originalLabels;
     private final Context<FlinkStateSnapshot> josdkContext;
     private final FlinkConfigManager configManager;
 

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/controller/FlinkStateSnapshotController.java

@@ -45,6 +45,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
@@ -82,6 +83,8 @@ public UpdateControl<FlinkStateSnapshot> reconcile(
             reconciler.reconcile(ctx);
         }
 
+        updateLabels(ctx);
+
         notifyListeners(ctx);
         return getUpdateControl(ctx);
     }
@@ -154,13 +157,27 @@ public Map<String, EventSource> prepareEventSources(
                 EventSourceUtils.getFlinkStateSnapshotInformerEventSources(context));
     }
 
+    /**
+     * Checks whether status and/or labels were changed on this resource, and returns an
+     * UpdateControl instance accordingly. Unless the snapshot state is terminal, the update control
+     * will be configured to reschedule the reconciliation.
+     *
+     * @param ctx snapshot context
+     * @return update control
+     */
     private UpdateControl<FlinkStateSnapshot> getUpdateControl(FlinkStateSnapshotContext ctx) {
         var resource = ctx.getResource();
-        UpdateControl<FlinkStateSnapshot> updateControl;
-        if (!ctx.getOriginalStatus().equals(resource.getStatus())) {
+        var updateControl = UpdateControl.<FlinkStateSnapshot>noUpdate();
+
+        var labelsChanged = resourceLabelsChanged(ctx);
+        var statusChanged = resourceStatusChanged(ctx);
+
+        if (labelsChanged && statusChanged) {
+            updateControl = UpdateControl.updateResourceAndPatchStatus(resource);
+        } else if (labelsChanged) {
+            updateControl = UpdateControl.updateResource(resource);
+        } else if (statusChanged) {
             updateControl = UpdateControl.patchStatus(resource);
-        } else {
-            updateControl = UpdateControl.noUpdate();
         }
 
         switch (resource.getStatus().getState()) {
@@ -174,7 +191,7 @@ private UpdateControl<FlinkStateSnapshot> getUpdateControl(FlinkStateSnapshotCon
     }
 
     private void notifyListeners(FlinkStateSnapshotContext ctx) {
-        if (!ctx.getOriginalStatus().equals(ctx.getResource().getStatus())) {
+        if (resourceStatusChanged(ctx)) {
             statusRecorder.notifyListeners(ctx.getResource(), ctx.getOriginalStatus());
         }
     }
@@ -197,4 +214,38 @@ private boolean validateSnapshot(FlinkStateSnapshotContext ctx) {
         }
         return true;
     }
+
+    /**
+     * Updates FlinkStateSnapshot resource labels with labels that represent its current state and
+     * spec.
+     *
+     * @param ctx snapshot context
+     */
+    private void updateLabels(FlinkStateSnapshotContext ctx) {
+        var labels = new HashMap<>(ctx.getResource().getMetadata().getLabels());
+        labels.putAll(
+                FlinkStateSnapshotUtils.getSnapshotLabels(
+                        ctx.getResource(), ctx.getSecondaryResource()));
+        ctx.getResource().getMetadata().setLabels(labels);
+    }
+
+    /**
+     * Checks if the resource status has changed since the start of reconciliation.
+     *
+     * @param ctx snapshot context
+     * @return true if resource status changed
+     */
+    private boolean resourceStatusChanged(FlinkStateSnapshotContext ctx) {
+        return !ctx.getOriginalStatus().equals(ctx.getResource().getStatus());
+    }
+
+    /**
+     * Checks if the resource labels have changed since the start of reconciliation.
+     *
+     * @param ctx snapshot context
+     * @return true if resource labels changed
+     */
+    private boolean resourceLabelsChanged(FlinkStateSnapshotContext ctx) {
+        return !ctx.getOriginalLabels().equals(ctx.getResource().getMetadata().getLabels());
+    }
 }

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/reconciler/SnapshotTriggerTimestampStore.java

@@ -80,7 +80,7 @@ public Instant getLastPeriodicTriggerInstant(
                                                                 .getLabels()
                                                                 .get(
                                                                         CrdConstants
-                                                                                .LABEL_SNAPSHOT_TYPE)))
+                                                                                .LABEL_SNAPSHOT_TRIGGER_TYPE)))
                         .map(
                                 s ->
                                         DateTimeUtils.parseKubernetes(

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkResourceContextFactory.java

@@ -35,6 +35,8 @@
 import org.apache.flink.kubernetes.operator.utils.EventRecorder;
 import org.apache.flink.util.concurrent.ExecutorThreadFactory;
 
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableMap;
+
 import io.javaoperatorsdk.operator.api.reconciler.Context;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import org.slf4j.Logger;
@@ -76,7 +78,11 @@ public FlinkResourceContextFactory(
     public FlinkStateSnapshotContext getFlinkStateSnapshotContext(
             FlinkStateSnapshot savepoint, Context<FlinkStateSnapshot> josdkContext) {
         return new FlinkStateSnapshotContext(
-                savepoint, savepoint.getStatus().toBuilder().build(), josdkContext, configManager);
+                savepoint,
+                savepoint.getStatus().toBuilder().build(),
+                ImmutableMap.copyOf(savepoint.getMetadata().getLabels()),
+                josdkContext,
+                configManager);
     }
 
     public <CR extends AbstractFlinkResource<?, ?>> FlinkResourceContext<CR> getResourceContext(

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/EventSourceUtils.java

@@ -61,7 +61,7 @@ public class EventSourceUtils {
                         .map(Enum::name)
                         .collect(Collectors.joining(","));
         var labelSelector =
-                String.format("%s in (%s)", CrdConstants.LABEL_SNAPSHOT_TYPE, labelFilters);
+                String.format("%s in (%s)", CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE, labelFilters);
         var configuration =
                 InformerConfiguration.from(FlinkStateSnapshot.class, context)
                         .withLabelSelector(labelSelector)

flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/utils/FlinkStateSnapshotUtils.java

@@ -42,6 +42,8 @@
 import javax.annotation.Nullable;
 
 import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 import java.util.function.Supplier;
@@ -66,7 +68,7 @@ protected static FlinkStateSnapshot createFlinkStateSnapshot(
         var metadata = new ObjectMeta();
         metadata.setNamespace(namespace);
         metadata.setName(name);
-        metadata.getLabels().put(CrdConstants.LABEL_SNAPSHOT_TYPE, triggerType.name());
+        metadata.getLabels().put(CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE, triggerType.name());
 
         var snapshot = new FlinkStateSnapshot();
         snapshot.setSpec(spec);
@@ -84,7 +86,7 @@ protected static FlinkStateSnapshot createFlinkStateSnapshot(
      */
     public static SnapshotTriggerType getSnapshotTriggerType(FlinkStateSnapshot snapshot) {
         var triggerTypeStr =
-                snapshot.getMetadata().getLabels().get(CrdConstants.LABEL_SNAPSHOT_TYPE);
+                snapshot.getMetadata().getLabels().get(CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE);
         try {
             return SnapshotTriggerType.valueOf(triggerTypeStr);
         } catch (NullPointerException | IllegalArgumentException e) {
@@ -345,7 +347,9 @@ public static void snapshotSuccessful(
     public static void snapshotInProgress(FlinkStateSnapshot snapshot, String triggerId) {
         snapshot.getMetadata()
                 .getLabels()
-                .putIfAbsent(CrdConstants.LABEL_SNAPSHOT_TYPE, SnapshotTriggerType.MANUAL.name());
+                .putIfAbsent(
+                        CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE,
+                        SnapshotTriggerType.MANUAL.name());
         snapshot.getStatus().setState(IN_PROGRESS);
         snapshot.getStatus().setTriggerId(triggerId);
         snapshot.getStatus().setTriggerTimestamp(DateTimeUtils.kubernetes(Instant.now()));
@@ -361,6 +365,52 @@ public static void snapshotTriggerPending(FlinkStateSnapshot snapshot) {
         snapshot.getStatus().setState(TRIGGER_PENDING);
     }
 
+    /**
+     * Creates a map of labels that can be applied to a snapshot resource based on its current spec
+     * and status. As described in FLINK-36109, we should set up selectable spec fields instead of
+     * labels once the Kubernetes feature is GA and widely supported.
+     *
+     * @param snapshot snapshot instance
+     * @param secondaryResourceOpt optional referenced Flink resource
+     * @return map of auto-generated labels
+     */
+    public static Map<String, String> getSnapshotLabels(
+            FlinkStateSnapshot snapshot,
+            Optional<AbstractFlinkResource<?, ?>> secondaryResourceOpt) {
+        var labels = new HashMap<String, String>();
+        labels.put(
+                CrdConstants.LABEL_SNAPSHOT_TYPE,
+                snapshot.getSpec().isSavepoint()
+                        ? SnapshotType.SAVEPOINT.name()
+                        : SnapshotType.CHECKPOINT.name());
+        labels.put(
+                CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE,
+                snapshot.getMetadata()
+                        .getLabels()
+                        .getOrDefault(
+                                CrdConstants.LABEL_SNAPSHOT_TRIGGER_TYPE,
+                                SnapshotTriggerType.MANUAL.name()));
+
+        Optional.ofNullable(snapshot.getStatus())
+                .ifPresent(
+                        status ->
+                                labels.put(
+                                        CrdConstants.LABEL_SNAPSHOT_STATE,
+                                        status.getState().name()));
+
+        secondaryResourceOpt.ifPresent(
+                secondaryResource -> {
+                    labels.put(
+                            CrdConstants.LABEL_SNAPSHOT_JOB_REFERENCE_KIND,
+                            secondaryResource.getKind());
+                    labels.put(
+                            CrdConstants.LABEL_SNAPSHOT_JOB_REFERENCE_NAME,
+                            secondaryResource.getMetadata().getName());
+                });
+
+        return labels;
+    }
+
     /**
      * Extracts the namespace of the job reference from a snapshot resource. This is either
      * explicitly specified in the job reference, or it will fallback to the namespace of the