refactor(inkless:delete): improve retention-enforcement observability - #725
Merged
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
Improves observability around diskless retention enforcement scheduling/execution and file-cleanup runs by adding a new scheduler lag gauge, redefining retention query timing to be per-partition, and adjusting logs to emit intent/result pairs around potentially blocking work.
Changes:
- Added
RetentionEnforcementScheduleLagMsgauge fed by the scheduler head entry, with locking to avoidPriorityQueueraces. - Changed retention-enforcement query timing so the duration callback fires once per partition (one transaction per partition).
- Updated retention/file-cleaner logs to emit clearer intent/result pairs; fixed partition attribution in per-partition retention logs.
Reviewed changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| storage/inkless/src/test/java/io/aiven/inkless/delete/RetentionEnforcementSchedulerTest.java | Adds coverage for the new schedule-lag gauge behavior. |
| storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/EnforceRetentionJobTest.java | Adds coverage ensuring the duration callback fires once per partition. |
| storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcerMetrics.java | Introduces the new schedule-lag metric and wires it into the metrics group. |
| storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcer.java | Wires the schedule-lag supplier into metrics; improves intent/result logging and per-partition attribution. |
| storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcementScheduler.java | Adds queue locking and exposes schedule lag calculation for the new gauge. |
| storage/inkless/src/main/java/io/aiven/inkless/delete/FileCleaner.java | Updates file-cleaner logs to be work-gated and to include runtime duration. |
| storage/inkless/src/main/java/io/aiven/inkless/control_plane/postgres/EnforceRetentionJob.java | Moves duration measurement to per-partition calls (one callback per request). |
| docs/inkless/metrics.rst | Documents the new RetentionEnforcer schedule-lag gauge. |
jeqo
force-pushed
the
jeqo/enforce-per-partition-metrics
branch
from
July 28, 2026 15:01
7a4c29d to
9e27f3c
Compare
Contributor
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (2)
storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcerMetrics.java:77
scheduleLagMsSupplieris used directly when registering the gauge. If this is ever passed as null (e.g., in tests or future call sites), this will fail at runtime with an NPE during metric registration or polling. Consider enforcing non-null at construction and using the validated reference for the gauge.
public RetentionEnforcerMetrics(final Supplier<Long> scheduleLagMsSupplier) {
retentionEnforcementTotalTime = metricsGroup.newHistogram(RETENTION_ENFORCEMENT_TOTAL_TIME, true, Map.of());
metricsGroup.newGauge(RETENTION_ENFORCEMENT_RATE, retentionEnforcementRate::intValue);
metricsGroup.newGauge(RETENTION_ENFORCEMENT_TOTAL_BATCHES_DELETED, retentionEnforcementTotalBatchesDeleted::intValue);
metricsGroup.newGauge(RETENTION_ENFORCEMENT_TOTAL_BYTES_DELETED, retentionEnforcementTotalBytesDeleted::intValue);
metricsGroup.newGauge(RETENTION_ENFORCEMENT_ERROR_RATE, retentionEnforcementErrorRate::intValue);
metricsGroup.newGauge(RETENTION_ENFORCEMENT_SCHEDULE_LAG_MS, scheduleLagMsSupplier);
}
storage/inkless/src/main/java/io/aiven/inkless/delete/RetentionEnforcementScheduler.java:189
dumpQueue()sorts only bynextEnforcementTime. When multiple partitions share the same scheduled time, the resulting order can still vary becausePriorityQueueiteration order is unspecified and the comparator treats ties as equal. Adding deterministic tie-breakers keeps this method stable and better matches the intent of making assertions independent of heap layout.
List<TopicIdPartitionWithNextEnforcementTime> dumpQueue() {
synchronized (queueLock) {
return partitionsByNextEnforcementTime.stream()
.sorted(TopicIdPartitionWithNextEnforcementTime.timeComparator())
.toList();
Add a RetentionEnforcementScheduleLagMs gauge (0 when on schedule; a sustained positive value means enforcement is falling behind its per-partition cadence), fed by the scheduler's oldest past-due queue entry read under a lock so the metrics/JMX thread and the enforcement thread do not race the PriorityQueue.
EnforceRetentionJob runs one transaction per partition (post-#705), yet EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side RetentionEnforcementTotalTime, which already measures the whole enforce wave. Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per partition) instead, matching the other *QueryTime metrics (one DB operation each) and yielding per-partition enforce latency without a new metric. The whole-wave total stays RetentionEnforcementTotalTime on the broker. Note: this redefines two existing metrics -- EnforceRetentionQueryTime from per-wave to per-partition, and EnforceRetentionQueryRate from wave count to partitions-enforced count (wave count remains RetentionEnforcementRate).
…t/result pairs
Retention enforcer: promote the per-cycle summary from DEBUG to INFO and add a
matching intent line before the (potentially blocking) enforce call, so a stuck
enforce shows as intent-without-completion:
"Enforcing retention for {n} partitions"
"Enforced retention for {n} partitions in {ms} ms: {b} batches, {y} bytes deleted"
One pair per non-empty cycle; empty cycles stay silent (no per-tick heartbeat).
Also fix a latent partition misattribution: the per-partition log lines indexed
readyPartitions while responses align with the delete-policy-filtered requests, so a
filtered-out partition could be mislabeled -- keep an enforcedPartitions list aligned
1:1 with requests/responses instead.
File cleaner: replace the per-run "Running file cleaner at {now}" heartbeat
(redundant with the framework timestamp and the outcome line) with a work-gated
intent/result pair:
"Running file cleaner: deleting {n} of {m} marked files" (m-n still within grace)
"File cleaner deleted {n} files in {ms} ms"
Partition/size counts are not available at this layer (FileToDelete carries only the
object key; batches are already deleted by cleanup time).
jeqo
force-pushed
the
jeqo/enforce-per-partition-metrics
branch
from
July 28, 2026 15:51
9e27f3c to
1bc3e35
Compare
jeqo
marked this pull request as ready for review
July 28, 2026 16:20
giuseppelillo
approved these changes
Jul 29, 2026
giuseppelillo
pushed a commit
that referenced
this pull request
Jul 29, 2026
…#725) * feat(inkless:retention): surface enforcement schedule lag Add a RetentionEnforcementScheduleLagMs gauge (0 when on schedule; a sustained positive value means enforcement is falling behind its per-partition cadence), fed by the scheduler's oldest past-due queue entry read under a lock so the metrics/JMX thread and the enforcement thread do not race the PriorityQueue. * feat(inkless:retention): meter enforce latency per partition EnforceRetentionJob runs one transaction per partition (post-#705), yet EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side RetentionEnforcementTotalTime, which already measures the whole enforce wave. Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per partition) instead, matching the other *QueryTime metrics (one DB operation each) and yielding per-partition enforce latency without a new metric. The whole-wave total stays RetentionEnforcementTotalTime on the broker. Note: this redefines two existing metrics -- EnforceRetentionQueryTime from per-wave to per-partition, and EnforceRetentionQueryRate from wave count to partitions-enforced count (wave count remains RetentionEnforcementRate). * feat(inkless:retention): make enforcement and file-cleaner logs intent/result pairs Retention enforcer: promote the per-cycle summary from DEBUG to INFO and add a matching intent line before the (potentially blocking) enforce call, so a stuck enforce shows as intent-without-completion: "Enforcing retention for {n} partitions" "Enforced retention for {n} partitions in {ms} ms: {b} batches, {y} bytes deleted" One pair per non-empty cycle; empty cycles stay silent (no per-tick heartbeat). Also fix a latent partition misattribution: the per-partition log lines indexed readyPartitions while responses align with the delete-policy-filtered requests, so a filtered-out partition could be mislabeled -- keep an enforcedPartitions list aligned 1:1 with requests/responses instead. File cleaner: replace the per-run "Running file cleaner at {now}" heartbeat (redundant with the framework timestamp and the outcome line) with a work-gated intent/result pair: "Running file cleaner: deleting {n} of {m} marked files" (m-n still within grace) "File cleaner deleted {n} files in {ms} ms" Partition/size counts are not available at this layer (FileToDelete carries only the object key; batches are already deleted by cleanup time).
giuseppelillo
pushed a commit
that referenced
this pull request
Jul 30, 2026
…#725) * feat(inkless:retention): surface enforcement schedule lag Add a RetentionEnforcementScheduleLagMs gauge (0 when on schedule; a sustained positive value means enforcement is falling behind its per-partition cadence), fed by the scheduler's oldest past-due queue entry read under a lock so the metrics/JMX thread and the enforcement thread do not race the PriorityQueue. * feat(inkless:retention): meter enforce latency per partition EnforceRetentionJob runs one transaction per partition (post-#705), yet EnforceRetentionQueryTime wrapped the whole batch -- duplicating the broker-side RetentionEnforcementTotalTime, which already measures the whole enforce wave. Record EnforceRetentionQueryTime/QueryRate per enforce_retention_v2 call (one per partition) instead, matching the other *QueryTime metrics (one DB operation each) and yielding per-partition enforce latency without a new metric. The whole-wave total stays RetentionEnforcementTotalTime on the broker. Note: this redefines two existing metrics -- EnforceRetentionQueryTime from per-wave to per-partition, and EnforceRetentionQueryRate from wave count to partitions-enforced count (wave count remains RetentionEnforcementRate). * feat(inkless:retention): make enforcement and file-cleaner logs intent/result pairs Retention enforcer: promote the per-cycle summary from DEBUG to INFO and add a matching intent line before the (potentially blocking) enforce call, so a stuck enforce shows as intent-without-completion: "Enforcing retention for {n} partitions" "Enforced retention for {n} partitions in {ms} ms: {b} batches, {y} bytes deleted" One pair per non-empty cycle; empty cycles stay silent (no per-tick heartbeat). Also fix a latent partition misattribution: the per-partition log lines indexed readyPartitions while responses align with the delete-policy-filtered requests, so a filtered-out partition could be mislabeled -- keep an enforcedPartitions list aligned 1:1 with requests/responses instead. File cleaner: replace the per-run "Running file cleaner at {now}" heartbeat (redundant with the framework timestamp and the outcome line) with a work-gated intent/result pair: "Running file cleaner: deleting {n} of {m} marked files" (m-n still within grace) "File cleaner deleted {n} files in {ms} ms" Partition/size counts are not available at this layer (FileToDelete carries only the object key; batches are already deleted by cleanup time).
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Improves observability of diskless retention enforcement and file cleanup. No change to enforcement or cleanup behavior. Three focused commits, all in the
io.aiven.inkless.delete/ control-plane area.What changed
RetentionEnforcementScheduleLagMsgauge: milliseconds the most overdue diskless partition is past its scheduled enforcement time (0 when on schedule; a sustained positive value means enforcement is falling behind its per-partition cadence). Fed by the scheduler's oldest past-due queue entry, read under a lock so the metrics/JMX thread and the enforcement thread don't race thePriorityQueue.EnforceRetentionJobruns one transaction per partition (post-fix(inkless:retention): decouple enforce_retention_v2 from the log lock #705), soEnforceRetentionQueryTime/QueryRateare now recorded per partition instead of once per wave. No new metric.Metrics
New:
RetentionEnforcer.RetentionEnforcementScheduleLagMs— gauge (ms).Redefined (operator impact — re-baseline any dashboards/alerts on these):
PostgresControlPlane.EnforceRetentionQueryTimeandEnforceRetentionQueryRatenow measure a singleenforce_retention_v2call (one partition) rather than the whole wave.QueryRaterises by roughly partitions-per-wave with no config change;QueryTimepercentiles become per-partition latencies. Nothing is lost — the whole-wave duration remainsRetentionEnforcer.RetentionEnforcementTotalTimeand the wave count remainsRetentionEnforcementRate. Per-transaction latency is the more accurate thing to measure since each partition is its own transaction.Logs
Enforcing retention for {n} partitions/Enforced retention for {n} partitions in {ms} ms: {b} batches, {y} bytes deleted. One pair per non-empty cycle; empty cycles stay silent (no per-tick heartbeat, since the enforcer ticks every 500ms).Running file cleaner at {now}heartbeat (redundant with the log framework timestamp and the outcome line) with a work-gated pair —Running file cleaner: deleting {n} of {m} marked files/File cleaner deleted {n} files in {ms} ms(them - nfiles are still within the retention grace period).readyPartitionswhile responses align with the delete-policy-filteredrequests); now uses a list kept aligned 1:1 withrequests/responses. Latent today (diskless doesn't support compaction, so the lists don't diverge) but made correct.Operator notes
EnforceRetentionQueryTime/EnforceRetentionQueryRate(semantics changed as above).RetentionEnforcementScheduleLagMsreturns 0 for both "on schedule" and "empty queue" (no diskless delete-policy partitions yet, or during startup); pair it with queue activity if that distinction matters.FileToDeletecarries only the object key, and batches are already deleted by cleanup time.Testing
scheduleLagTracksOverduePartition(empty -> 0, future head -> 0, advance past due -> exact lag).durationCallbackFiresOncePerPartition(3 callbacks for 3 requests, including a non-existent partition).metrics.rstregenerated; checkstyle (main + test) clean.