-
Notifications
You must be signed in to change notification settings - Fork 416
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[FLINK-36110][snapshot] Store periodic snapshot trigger timestamps in…
… memory
- Loading branch information
1 parent
c8a6ca8
commit 4811b19
Showing
8 changed files
with
265 additions
and
68 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
97 changes: 97 additions & 0 deletions
97
...n/java/org/apache/flink/kubernetes/operator/reconciler/SnapshotTriggerTimestampStore.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,97 @@ | ||
| package org.apache.flink.kubernetes.operator.reconciler; | ||
|
|
||
| import org.apache.flink.autoscaler.utils.DateTimeUtils; | ||
| 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.status.SnapshotInfo; | ||
|
|
||
| import io.fabric8.kubernetes.api.model.HasMetadata; | ||
| import lombok.RequiredArgsConstructor; | ||
|
|
||
| import javax.annotation.Nullable; | ||
|
|
||
| import java.time.Instant; | ||
| import java.util.Comparator; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.concurrent.ConcurrentHashMap; | ||
|
|
||
| import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED; | ||
| import static org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType.PERIODIC; | ||
| import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT; | ||
|
|
||
| /** Class used to store latest timestamps of periodic checkpoint/savepoint. */ | ||
| @RequiredArgsConstructor | ||
| public class SnapshotTriggerTimestampStore { | ||
| private final SnapshotType snapshotType; | ||
| private final ConcurrentHashMap<String, Instant> lastTriggered = new ConcurrentHashMap<>(); | ||
|
|
||
| /** | ||
| * Returns the time a periodic snapshot was last triggered for this resource. This is stored in | ||
| * memory, on operator start it will use the latest completed FlinkStateSnapshot CR creation | ||
| * timestamp. If none found, the return value will be the max of the resource creation timestamp | ||
| * and the latest triggered legacy snapshot. | ||
| * | ||
| * @param resource Flink resource | ||
| * @param snapshots related snapshot resources | ||
| * @return instant of last trigger | ||
| */ | ||
| public Instant getLastPeriodicTriggerInstant( | ||
| AbstractFlinkResource<?, ?> resource, @Nullable Set<FlinkStateSnapshot> snapshots) { | ||
| if (lastTriggered.containsKey(resource.getMetadata().getUid())) { | ||
| return lastTriggered.get(resource.getMetadata().getUid()); | ||
| } | ||
|
|
||
| return Optional.ofNullable(snapshots).orElse(Set.of()).stream() | ||
| .filter(s -> s.getStatus() != null && COMPLETED.equals(s.getStatus().getState())) | ||
| .filter(s -> (snapshotType == SAVEPOINT) == s.getSpec().isSavepoint()) | ||
| .filter( | ||
| s -> | ||
| PERIODIC.name() | ||
| .equals( | ||
| s.getMetadata() | ||
| .getLabels() | ||
| .get(CrdConstants.LABEL_SNAPSHOT_TYPE))) | ||
| .map(s -> DateTimeUtils.parseKubernetes(s.getMetadata().getCreationTimestamp())) | ||
| .max(Comparator.naturalOrder()) | ||
| .orElseGet( | ||
| () -> { | ||
| var legacyTs = getLegacyTimestamp(resource); | ||
| var creationTs = | ||
| Instant.parse(resource.getMetadata().getCreationTimestamp()); | ||
|
|
||
| if (legacyTs.compareTo(creationTs) > 0) { | ||
| return legacyTs; | ||
| } else { | ||
| return creationTs; | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Updates the time a periodic snapshot was last triggered for this resource. | ||
| * | ||
| * @param resource Kubernetes resource | ||
| * @param instant new timestamp | ||
| */ | ||
| public void updateLastPeriodicTriggerTimestamp(HasMetadata resource, Instant instant) { | ||
| lastTriggered.put(resource.getMetadata().getUid(), instant); | ||
| } | ||
|
|
||
| private Instant getLegacyTimestamp(AbstractFlinkResource<?, ?> resource) { | ||
| SnapshotInfo snapshotInfo; | ||
| switch (snapshotType) { | ||
| case SAVEPOINT: | ||
| snapshotInfo = resource.getStatus().getJobStatus().getSavepointInfo(); | ||
| break; | ||
| case CHECKPOINT: | ||
| snapshotInfo = resource.getStatus().getJobStatus().getCheckpointInfo(); | ||
| break; | ||
| default: | ||
| throw new IllegalArgumentException("Unsupported snapshot type: " + snapshotType); | ||
| } | ||
|
|
||
| return Instant.ofEpochMilli(snapshotInfo.getLastPeriodicTriggerTimestamp()); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
100 changes: 100 additions & 0 deletions
100
...va/org/apache/flink/kubernetes/operator/reconciler/SnapshotTriggerTimestampStoreTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| package org.apache.flink.kubernetes.operator.reconciler; | ||
|
|
||
| import org.apache.flink.autoscaler.utils.DateTimeUtils; | ||
| import org.apache.flink.kubernetes.operator.TestUtils; | ||
| 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.SavepointSpec; | ||
| import org.apache.flink.kubernetes.operator.api.status.Checkpoint; | ||
| import org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus; | ||
| import org.apache.flink.kubernetes.operator.api.status.Savepoint; | ||
| import org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType; | ||
|
|
||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import java.time.Instant; | ||
| import java.util.Set; | ||
|
|
||
| import static org.apache.flink.kubernetes.operator.api.status.FlinkStateSnapshotStatus.State.COMPLETED; | ||
| import static org.apache.flink.kubernetes.operator.api.status.SnapshotTriggerType.PERIODIC; | ||
| import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.CHECKPOINT; | ||
| import static org.apache.flink.kubernetes.operator.reconciler.SnapshotType.SAVEPOINT; | ||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
|
|
||
| class SnapshotTriggerTimestampStoreTest { | ||
|
|
||
| @Test | ||
| public void testCheckpointTimestampStore() { | ||
| testTimestampStore(CHECKPOINT); | ||
| } | ||
|
|
||
| @Test | ||
| public void testSavepointTimestampStore() { | ||
| testTimestampStore(SAVEPOINT); | ||
| } | ||
|
|
||
| private void testTimestampStore(SnapshotType snapshotType) { | ||
| var resource = TestUtils.buildApplicationCluster(); | ||
| var store = new SnapshotTriggerTimestampStore(snapshotType); | ||
|
|
||
| var instantCreation = Instant.ofEpochMilli(1); | ||
| resource.getMetadata().setCreationTimestamp(DateTimeUtils.kubernetes(instantCreation)); | ||
|
|
||
| assertEquals(instantCreation, store.getLastPeriodicTriggerInstant(resource, Set.of())); | ||
|
|
||
| var instantLegacy = Instant.ofEpochMilli(2); | ||
| if (snapshotType == SAVEPOINT) { | ||
| resource.getStatus() | ||
| .getJobStatus() | ||
| .getSavepointInfo() | ||
| .updateLastSavepoint(new Savepoint(2L, "", PERIODIC, null, null)); | ||
| } else { | ||
| resource.getStatus() | ||
| .getJobStatus() | ||
| .getCheckpointInfo() | ||
| .updateLastCheckpoint(new Checkpoint(2L, PERIODIC, null, null)); | ||
| } | ||
| assertEquals(instantLegacy, store.getLastPeriodicTriggerInstant(resource, Set.of())); | ||
|
|
||
| var snapshots = Set.of(createSnapshot(snapshotType, SnapshotTriggerType.PERIODIC, 3L)); | ||
| assertEquals( | ||
| Instant.ofEpochMilli(3), store.getLastPeriodicTriggerInstant(resource, snapshots)); | ||
|
|
||
| snapshots = | ||
| Set.of( | ||
| createSnapshot(snapshotType, SnapshotTriggerType.PERIODIC, 200L), | ||
| createSnapshot(snapshotType, SnapshotTriggerType.PERIODIC, 300L), | ||
| createSnapshot(snapshotType, SnapshotTriggerType.MANUAL, 10000L), | ||
| createSnapshot(snapshotType, SnapshotTriggerType.PERIODIC, 0L)); | ||
| assertEquals( | ||
| Instant.ofEpochMilli(300), | ||
| store.getLastPeriodicTriggerInstant(resource, snapshots)); | ||
|
|
||
| var instantInMemory = Instant.ofEpochMilli(111L); | ||
| store.updateLastPeriodicTriggerTimestamp(resource, instantInMemory); | ||
| assertEquals(instantInMemory, store.getLastPeriodicTriggerInstant(resource, snapshots)); | ||
|
|
||
| instantInMemory = Instant.ofEpochMilli(11L); | ||
| store.updateLastPeriodicTriggerTimestamp(resource, instantInMemory); | ||
| assertEquals(instantInMemory, store.getLastPeriodicTriggerInstant(resource, snapshots)); | ||
| } | ||
|
|
||
| private FlinkStateSnapshot createSnapshot( | ||
| SnapshotType snapshotType, SnapshotTriggerType triggerType, Long timestamp) { | ||
| var snapshot = new FlinkStateSnapshot(); | ||
| snapshot.getMetadata() | ||
| .setCreationTimestamp(DateTimeUtils.kubernetes(Instant.ofEpochMilli(timestamp))); | ||
| if (snapshotType == SAVEPOINT) { | ||
| snapshot.getSpec().setSavepoint(new SavepointSpec()); | ||
| } else { | ||
| snapshot.getSpec().setCheckpoint(new CheckpointSpec()); | ||
| } | ||
| snapshot.getMetadata() | ||
| .getLabels() | ||
| .put(CrdConstants.LABEL_SNAPSHOT_TYPE, triggerType.name()); | ||
| snapshot.setStatus(new FlinkStateSnapshotStatus()); | ||
| snapshot.getStatus().setState(COMPLETED); | ||
| return snapshot; | ||
| } | ||
| } |
Oops, something went wrong.