id: minio-noncurrent-version-bloat-prune-guard
namespace: company.team
description: |
Turning on bucket versioning is one click and it is the right call. Without a lifecycle expiry rule
it is also a storage bill that only grows, because every overwrite and every delete keeps its
predecessor forever. A key written daily for a year holds 365 noncurrent versions plus a delete
marker, all billed, none visible in a normal object listing. This flow measures that hidden mass
with the MinIO plugin, selects the noncurrent versions that are past an age floor and outside a
keep-last-N rollback window, deletes them by version id behind an approval, then re-reads the
version listing and fails the run if any targeted version survived. The current version of every
key is never a candidate.
inputs:
- id: dry_run
type: BOOL
defaults: true
description: When true the flow only reports the prunable versions and never
deletes one.
- id: auto_remediate
type: BOOL
defaults: false
description: When true (and dry_run is false) the flow deletes the selected
versions after approval.
- id: bucket
type: STRING
description: The versioned bucket to audit.
- id: prefix
type: STRING
defaults: ""
description: Restrict the audit to one key prefix. Empty audits the whole bucket.
- id: keep_noncurrent_versions
type: INT
defaults: 3
description: >-
How many noncurrent versions to keep per key, newest first, whatever their
age. This is the rollback window. Set it to 0 only when you are certain no
key needs a predecessor.
- id: min_noncurrent_age_minutes
type: INT
defaults: 43200
description: >-
A noncurrent version younger than this is never pruned. The default 43200
minutes is 30 days. The floor exists so a version written minutes ago,
which is very likely the thing somebody is about to roll back to, is out
of reach of this flow.
- id: delete_marker_policy
type: SELECT
values:
- keep
- prune_noncurrent
defaults: keep
description: >-
A delete marker is itself a version. keep (the default) never removes one.
prune_noncurrent removes only delete markers that are NOT the latest
version of their key, which means the key was written again after the
delete so the marker is historical. A latest delete marker is never
removed under any setting, because removing it resurrects an object
somebody deleted.
- id: max_versions_to_prune
type: INT
defaults: 200
description: >-
Ceiling on how many versions one run may delete. More selected than this
usually means a wrong bucket, a wrong prefix or a keep window set to zero
by accident, so the flow refuses to write and alerts instead.
- id: minio_endpoint
type: STRING
defaults: http://minio.internal:9000
description: S3 API endpoint URL. The native MinIO plugin and the AWS CLI both
connect to it.
- id: minio_access_key
type: STRING
defaults: "{{ secret('MINIO_ACCESS_KEY_ID') }}"
description: Access key. Defaults to the MINIO_ACCESS_KEY_ID secret and needs
s3:ListBucketVersions plus s3:DeleteObjectVersion.
- id: minio_secret_key
type: STRING
defaults: "{{ secret('MINIO_SECRET_KEY_ID') }}"
description: Secret key. Defaults to the MINIO_SECRET_KEY_ID secret.
- id: awscli_image
type: STRING
defaults: public.ecr.aws/aws-cli/aws-cli:latest
description: >-
Container image carrying the AWS CLI. Neither the MinIO plugin nor the S3
plugin ships a task that deletes one specific object version, so the
version inventory and the per-version delete run through this CLI.
- id: docker_network
type: STRING
defaults: bridge
description: >-
Docker network the CLI container joins so it can reach the endpoint. The
default bridge reaches any routable address. Set it to your own network
when the endpoint is a container name.
triggers:
- id: daily
type: io.kestra.plugin.core.trigger.Schedule
description: Measure version bloat every morning. Safe mode is forced here, so a
scheduled run only reports.
cron: "0 5 * * *"
disabled: true
inputs:
bucket: my-versioned-bucket
dry_run: true
auto_remediate: false
tasks:
- id: list_current_objects
type: io.kestra.plugin.minio.List
description: >-
List the bucket the way a normal client sees it, current versions only.
This is the count the console shows and the number everybody reasons
about.
endpoint: "{{ inputs.minio_endpoint }}"
accessKeyId: "{{ inputs.minio_access_key }}"
secretKeyId: "{{ inputs.minio_secret_key }}"
bucket: "{{ inputs.bucket }}"
prefix: "{{ inputs.prefix }}"
recursive: true
includeVersions: false
filter: FILES
- id: list_version_entries
type: io.kestra.plugin.minio.List
description: >-
List the same bucket with version listing on. Every stored version and
every delete marker comes back as its own entry. The gap against the
previous task is the mass you are paying for and cannot see.
endpoint: "{{ inputs.minio_endpoint }}"
accessKeyId: "{{ inputs.minio_access_key }}"
secretKeyId: "{{ inputs.minio_secret_key }}"
bucket: "{{ inputs.bucket }}"
prefix: "{{ inputs.prefix }}"
recursive: true
includeVersions: true
filter: FILES
- id: audit_versions
type: io.kestra.plugin.scripts.shell.Commands
description: >-
Read the full version inventory, which carries the version ids the plugin
listing does not expose, then select what may be pruned. A version is a
candidate only when it is not the latest version of its key, it falls
outside the keep-last-N window for that key, it is older than the age
floor and it has a real version id. A connection or credential failure
fails the task, so an unreachable endpoint can never read as a clean
bucket.
containerImage: "{{ inputs.awscli_image }}"
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: "{{ inputs.docker_network }}"
entryPoint: []
outputFiles:
- plan.json
env:
HOME: /tmp
AWS_ACCESS_KEY_ID: "{{ inputs.minio_access_key }}"
AWS_SECRET_ACCESS_KEY: "{{ inputs.minio_secret_key }}"
AWS_DEFAULT_REGION: us-east-1
AWS_EC2_METADATA_DISABLED: "true"
ENDPOINT: "{{ inputs.minio_endpoint }}"
BUCKET: "{{ inputs.bucket }}"
PREFIX: "{{ inputs.prefix }}"
KEEP_N: "{{ inputs.keep_noncurrent_versions }}"
AGE_MINUTES: "{{ inputs.min_noncurrent_age_minutes }}"
DM_POLICY: "{{ inputs.delete_marker_policy }}"
commands:
- |
set -e
if ! aws --endpoint-url "$ENDPOINT" s3api get-bucket-versioning --bucket "$BUCKET" > /tmp/ver.json; then
echo "cannot reach the endpoint $ENDPOINT or the credentials were rejected" >&2
exit 1
fi
aws --endpoint-url "$ENDPOINT" s3api list-object-versions \
--bucket "$BUCKET" --prefix "$PREFIX" --output json > /tmp/versions.json
python3 - <<'PY'
import datetime
import json
import os
keep_n = int(os.environ['KEEP_N'])
age_minutes = int(os.environ['AGE_MINUTES'])
policy = os.environ['DM_POLICY']
now = datetime.datetime.now(datetime.timezone.utc)
cutoff = now - datetime.timedelta(minutes=age_minutes)
def when(row):
raw = row.get('LastModified')
if not raw:
return None
return datetime.datetime.fromisoformat(raw.replace('Z', '+00:00'))
def real_id(row):
vid = row.get('VersionId')
return vid not in (None, '', 'null')
inventory = json.load(open('/tmp/versions.json'))
versioning = json.load(open('/tmp/ver.json')).get('Status', 'Disabled')
versions = inventory.get('Versions') or []
markers = inventory.get('DeleteMarkers') or []
current = [v for v in versions if v.get('IsLatest')]
noncurrent = [v for v in versions if not v.get('IsLatest')]
latest_markers = [m for m in markers if m.get('IsLatest')]
old_markers = [m for m in markers if not m.get('IsLatest')]
grouped = {}
for row in noncurrent:
grouped.setdefault(row['Key'], []).append(row)
targets = []
held_window = 0
held_age = 0
held_unversioned = 0
for key in sorted(grouped):
rows = grouped[key]
rows.sort(key=lambda r: (when(r) or cutoff), reverse=True)
held_window += len(rows[:keep_n])
for row in rows[keep_n:]:
if not real_id(row):
held_unversioned += 1
continue
stamp = when(row)
if stamp is None or stamp > cutoff:
held_age += 1
continue
targets.append({
'key': key,
'version_id': row['VersionId'],
'kind': 'version',
'size': row.get('Size') or 0,
'last_modified': row.get('LastModified'),
})
marker_pruned = 0
for row in old_markers:
if policy != 'prune_noncurrent':
continue
if not real_id(row):
held_unversioned += 1
continue
stamp = when(row)
if stamp is None or stamp > cutoff:
held_age += 1
continue
marker_pruned += 1
targets.append({
'key': row['Key'],
'version_id': row['VersionId'],
'kind': 'delete_marker',
'size': 0,
'last_modified': row.get('LastModified'),
})
fingerprint = sorted(
'%s|%s|%s' % (v['Key'], v.get('VersionId'), (v.get('ETag') or '').strip('"'))
for v in current
)
marker_tips = sorted('%s|%s' % (m['Key'], m.get('VersionId')) for m in latest_markers)
plan = {
'targets': targets,
'current_fingerprint': fingerprint,
'latest_delete_markers': marker_tips,
}
with open('plan.json', 'w') as handle:
json.dump(plan, handle)
keys = sorted(set(row['key'] for row in targets))
report = {
'versioning_status': versioning,
'current_object_count': len(current),
'total_version_entries': len(versions) + len(markers),
'noncurrent_version_count': len(noncurrent),
'noncurrent_delete_marker_count': len(old_markers),
'latest_delete_marker_count': len(latest_markers),
'hidden_entry_count': len(noncurrent) + len(markers),
'prunable_count': len(targets),
'prunable_bytes': sum(row['size'] for row in targets),
'prunable_delete_markers': marker_pruned,
'held_by_keep_window': held_window,
'held_by_age_floor': held_age,
'held_unversioned': held_unversioned,
'prune_rows': [json.dumps(row, separators=(',', ':')) for row in targets],
'keys_affected': keys,
'keys_affected_csv': ', '.join(keys) if keys else 'none',
}
print('audit bucket=%s versioning=%s current=%d entries=%d noncurrent=%d markers=%d prunable=%d bytes=%d held_window=%d held_age=%d' % (
os.environ['BUCKET'], versioning, len(current), len(versions) + len(markers),
len(noncurrent), len(markers), len(targets), report['prunable_bytes'], held_window, held_age))
print('::' + json.dumps({'outputs': report}) + '::')
PY
- id: log_bloat
type: io.kestra.plugin.core.log.Log
description: Put the two listings side by side, so the invisible mass is on the
record whether or not anything is pruned.
message: >-
Bucket {{ inputs.bucket }} versioning={{
outputs.audit_versions.vars.versioning_status }}. A normal listing shows
{{ (outputs.list_current_objects.objects ?? []) | length }} object(s).
Version listing returns {{ (outputs.list_version_entries.objects ?? []) |
length }} entries, of which {{
outputs.audit_versions.vars.noncurrent_version_count }} are noncurrent
versions, {{ outputs.audit_versions.vars.noncurrent_delete_marker_count }}
are historical delete markers plus {{
outputs.audit_versions.vars.latest_delete_marker_count }} latest delete
markers. {{ outputs.audit_versions.vars.prunable_count }} version(s) are
prunable ({{ outputs.audit_versions.vars.prunable_bytes }} bytes) across
keys [{{ outputs.audit_versions.vars.keys_affected_csv }}]. Held back by
the keep window: {{ outputs.audit_versions.vars.held_by_keep_window }}.
Held back by the age floor: {{
outputs.audit_versions.vars.held_by_age_floor }}.
- id: gate
type: io.kestra.plugin.core.flow.If
description: Delete a version only when auto_remediate is true, dry_run is false
and something is actually prunable. Any other combination reports and
stops.
condition: "{{ inputs.auto_remediate and (inputs.dry_run == false) and
((outputs.audit_versions.vars.prune_rows | length) > 0) }}"
then:
- id: ceiling_check
type: io.kestra.plugin.core.flow.If
description: Refuse the run when the selection is larger than the ceiling, which
usually means a wrong bucket, a wrong prefix or a zero keep window.
condition: "{{ (outputs.audit_versions.vars.prune_rows | length) <=
inputs.max_versions_to_prune }}"
then:
- id: confirm
type: io.kestra.plugin.core.flow.Pause
description: >-
Hold for a human before any version is deleted. A version delete
is not recoverable, there is no undo and no recycle bin, so the
exact selection printed above is a person's call. The current
version of every key is not in that selection.
onResume:
- id: approved
type: BOOL
description: Set true to delete the selected noncurrent versions. Leave false or
unset to make no change.
- id: apply
type: io.kestra.plugin.core.flow.If
description: Only proceed past the pause when it was explicitly approved. An
unanswered pause never reads as approval.
condition: "{{ outputs.confirm.onResume.approved ?? false }}"
then:
- id: remediate_versions
type: io.kestra.plugin.core.flow.Loop
description: Delete each selected version by its own version id. Serial, because
concurrent version deletes on one bucket race on the same key.
values: "{{ outputs.audit_versions.vars.prune_rows }}"
concurrencyLimit: 1
tasks:
- id: delete_version
type: io.kestra.plugin.scripts.shell.Commands
description: >-
Delete one specific object version. The version id is
mandatory on this call, so it can only ever remove the
version that was selected. Omitting it would create a
delete marker instead, which is the opposite of what this
flow is for.
containerImage: "{{ inputs.awscli_image }}"
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: "{{ inputs.docker_network }}"
entryPoint: []
env:
HOME: /tmp
AWS_ACCESS_KEY_ID: "{{ inputs.minio_access_key }}"
AWS_SECRET_ACCESS_KEY: "{{ inputs.minio_secret_key }}"
AWS_DEFAULT_REGION: us-east-1
AWS_EC2_METADATA_DISABLED: "true"
ENDPOINT: "{{ inputs.minio_endpoint }}"
BUCKET: "{{ inputs.bucket }}"
ROW: "{{ item.value }}"
commands:
- |
set -e
KEY=$(printf '%s' "$ROW" | jq -r '.key')
VID=$(printf '%s' "$ROW" | jq -r '.version_id')
KIND=$(printf '%s' "$ROW" | jq -r '.kind')
if [ -z "$VID" ] || [ "$VID" = "null" ]; then
echo "refusing to delete $KEY with no version id" >&2
exit 1
fi
echo "deleting $KIND $KEY version $VID"
aws --endpoint-url "$ENDPOINT" s3api delete-object \
--bucket "$BUCKET" --key "$KEY" --version-id "$VID" > /dev/null
- id: verify_pruned
type: io.kestra.plugin.scripts.shell.Commands
description: >-
Read the version listing again and compare it with the plan.
Two things have to hold. No targeted version id may still be
present. Every current version recorded before the prune must
still be there with the same version id and the same ETag.
This is the authoritative check, not the delete task's exit
code.
containerImage: "{{ inputs.awscli_image }}"
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: "{{ inputs.docker_network }}"
entryPoint: []
inputFiles:
plan.json: "{{ outputs.audit_versions.outputFiles['plan.json'] }}"
env:
HOME: /tmp
AWS_ACCESS_KEY_ID: "{{ inputs.minio_access_key }}"
AWS_SECRET_ACCESS_KEY: "{{ inputs.minio_secret_key }}"
AWS_DEFAULT_REGION: us-east-1
AWS_EC2_METADATA_DISABLED: "true"
ENDPOINT: "{{ inputs.minio_endpoint }}"
BUCKET: "{{ inputs.bucket }}"
PREFIX: "{{ inputs.prefix }}"
commands:
- |
set -e
aws --endpoint-url "$ENDPOINT" s3api list-object-versions \
--bucket "$BUCKET" --prefix "$PREFIX" --output json > /tmp/after.json
python3 - <<'PY'
import json
plan = json.load(open('plan.json'))
after = json.load(open('/tmp/after.json'))
versions = after.get('Versions') or []
markers = after.get('DeleteMarkers') or []
present = set()
for row in versions + markers:
present.add('%s|%s' % (row['Key'], row.get('VersionId')))
survivors = [
'%s|%s' % (row['key'], row['version_id'])
for row in plan['targets']
if '%s|%s' % (row['key'], row['version_id']) in present
]
current_now = sorted(
'%s|%s|%s' % (v['Key'], v.get('VersionId'), (v.get('ETag') or '').strip('"'))
for v in versions if v.get('IsLatest')
)
lost = [row for row in plan['current_fingerprint'] if row not in current_now]
marker_tips_now = sorted(
'%s|%s' % (m['Key'], m.get('VersionId'))
for m in markers if m.get('IsLatest')
)
lost_tips = [row for row in plan['latest_delete_markers'] if row not in marker_tips_now]
verdict = 'clean' if not survivors and not lost and not lost_tips else 'dirty'
result = {
'verdict': verdict,
'targeted': len(plan['targets']),
'survivor_count': len(survivors),
'survivor_csv': ', '.join(survivors) if survivors else 'none',
'current_lost_count': len(lost) + len(lost_tips),
'current_lost_csv': ', '.join(lost + lost_tips) if (lost or lost_tips) else 'none',
'entries_after': len(versions) + len(markers),
'current_after': len(current_now),
}
print('reverify verdict=%s targeted=%d survivors=%d current_lost=%d entries_after=%d' % (
verdict, len(plan['targets']), len(survivors), len(lost) + len(lost_tips),
len(versions) + len(markers)))
print('::' + json.dumps({'outputs': result}) + '::')
PY
- id: assert_clean
type: io.kestra.plugin.core.flow.If
description: Fail the run when a targeted version survived or a current version
went missing, so a prune that did not land is never reported
as success.
condition: "{{ outputs.verify_pruned.vars.verdict == 'clean' }}"
then:
- id: pruned_ok
type: io.kestra.plugin.core.log.Log
message: >-
Pruned {{ outputs.verify_pruned.vars.targeted }}
version(s). Version entries are now {{
outputs.verify_pruned.vars.entries_after }} and all {{
outputs.verify_pruned.vars.current_after }} current
version(s) are intact.
else:
- id: prune_failed
type: io.kestra.plugin.core.execution.Fail
errorMessage: >-
Prune did not land cleanly. Surviving targeted versions:
{{ outputs.verify_pruned.vars.survivor_csv }}. Missing
current versions: {{
outputs.verify_pruned.vars.current_lost_csv }}.
else:
- id: declined
type: io.kestra.plugin.core.log.Log
message: >-
Approval was declined, so {{
outputs.audit_versions.vars.prunable_count }} prunable
version(s) across [{{
outputs.audit_versions.vars.keys_affected_csv }}] were left in
place.
else:
- id: too_many
type: io.kestra.plugin.core.execution.Fail
errorMessage: >-
{{ outputs.audit_versions.vars.prunable_count }} selected
version(s) exceeds max_versions_to_prune ({{
inputs.max_versions_to_prune }}). Refusing to delete. Check the
bucket, the prefix and keep_noncurrent_versions.
else:
- id: no_write
type: io.kestra.plugin.core.log.Log
message: >-
No version was deleted. dry_run={{ inputs.dry_run }},
auto_remediate={{ inputs.auto_remediate }}, prunable={{
outputs.audit_versions.vars.prunable_count }} ({{
outputs.audit_versions.vars.prunable_bytes }} bytes across [{{
outputs.audit_versions.vars.keys_affected_csv }}]). Set dry_run=false
and auto_remediate=true to prune them after approval.
- id: report
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Post a run summary to Slack.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"text": "MinIO noncurrent version bloat guard on {{ inputs.bucket }}: {{ outputs.audit_versions.vars.current_object_count }} visible object(s) but {{ outputs.audit_versions.vars.total_version_entries }} version entries. {{ outputs.audit_versions.vars.prunable_count }} prunable ({{ outputs.audit_versions.vars.prunable_bytes }} bytes), {{ outputs.audit_versions.vars.held_by_keep_window }} held by the keep window, {{ outputs.audit_versions.vars.held_by_age_floor }} held by the age floor. dry_run={{ inputs.dry_run }}, auto_remediate={{ inputs.auto_remediate }} (execution {{ execution.id }})."
}
errors:
- id: alert_on_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: Alert Slack when the audit fails, a version delete errors or the
re-verify finds the prune did not land.
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{
"text": "MinIO noncurrent version bloat guard FAILED on {{ inputs.bucket }} (execution {{ execution.id }}). Either the endpoint was unreachable, a version delete was rejected or a targeted version survived the prune. Check the logs."
}
