gyfora commented on code in PR #860:
URL:
https://github.com/apache/flink-kubernetes-operator/pull/860#discussion_r1712179282
##########
flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/observer/SnapshotObserver.java:
##########
@@ -220,63 +235,179 @@ private void
observeTriggeredCheckpoint(FlinkResourceContext<CR> ctx, String job
/** Clean up and dispose savepoints according to the configured max
size/age. */
@VisibleForTesting
- void cleanupSavepointHistory(FlinkResourceContext<CR> ctx, SavepointInfo
currentSavepointInfo) {
+ void cleanupSavepointHistory(FlinkResourceContext<CR> ctx) {
+ if (!FlinkStateSnapshotUtils.isSnapshotResourceEnabled(
+ ctx.getOperatorConfig(), ctx.getObserveConfig())) {
+ cleanupSavepointHistoryLegacy(ctx);
+ return;
+ }
+
+ var snapshots =
ctx.getJosdkContext().getSecondaryResources(FlinkStateSnapshot.class);
+ if (CollectionUtil.isNullOrEmpty(snapshots)) {
+ return;
+ }
+ if (ctx.getObserveConfig().get(OPERATOR_SAVEPOINT_CLEANUP_ENABLED)) {
+ var savepointsToDelete =
+ getFlinkStateSnapshotsToCleanUp(
+ snapshots, ctx.getObserveConfig(),
ctx.getOperatorConfig(), SAVEPOINT);
+ var checkpointsToDelete =
+ getFlinkStateSnapshotsToCleanUp(
+ snapshots, ctx.getObserveConfig(),
ctx.getOperatorConfig(), CHECKPOINT);
+ Stream.concat(savepointsToDelete.stream(),
checkpointsToDelete.stream())
+ .forEach(
+ snapshot ->
+ ctx.getKubernetesClient()
+ .resource(snapshot)
+ .withTimeoutInMillis(0L)
+ .delete());
+ }
+ }
+
+ @VisibleForTesting
+ Set<FlinkStateSnapshot> getFlinkStateSnapshotsToCleanUp(
+ Collection<FlinkStateSnapshot> snapshots,
+ Configuration observeConfig,
+ FlinkOperatorConfiguration operatorConfig,
+ SnapshotType snapshotType) {
+ var snapshotList =
+ snapshots.stream()
+ .filter(s -> s.getStatus() != null)
+ .filter(s ->
COMPLETED.equals(s.getStatus().getState()))
+ .filter(
+ s ->
+ SnapshotTriggerType.PERIODIC.equals(
+
FlinkStateSnapshotUtils.getSnapshotTriggerType(s)))
Review Comment:
Feels like filtering everything out may leave a lot of lingering upgrade /
abandoned / unfinished savepoint CRs over time. How are we planning to address
that?
I think upgrade save points can be deleted as long as we have at least 1
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
