id: k8s-terminated-pod-garbage-collector
namespace: company.team
description: Archive ownerless failed Pod status diagnostics before
retention-aged, identity-pinned removal.
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: cluster_id
type: STRING
defaults: primary
- id: namespace
type: STRING
defaults: default
- id: api_url
type: STRING
defaults: "{{ secret('K8S_MASTER_URL') }}"
- id: token
type: SECRET
defaults: "{{ secret('K8S_TOKEN') }}"
- id: ca_cert_data
type: STRING
defaults: "{{ secret('K8S_CA_CERT_DATA') }}"
- id: trust_certs
type: BOOL
defaults: false
- id: retention_seconds
type: INT
defaults: 86400
description: Minimum hold after complete container termination or first observed
Failed status.
- id: max_pods
type: INT
defaults: 5
- id: max_inventory
type: INT
defaults: 1000
- id: max_archive_bytes
type: INT
defaults: 262144
- id: dry_run
type: BOOL
defaults: true
- id: auto_remediate
type: BOOL
defaults: false
variables:
state_key: "pod-diagnostics-{{ inputs.cluster_id }}-{{ inputs.namespace }}"
helper: |
import hashlib, json, re
from datetime import datetime, timezone
PIN = 'diagnostics.example.com/collector'
LABEL = 'diagnostics.example.com/collect'
def stamp(value):
if not isinstance(value, str):
raise ValueError('Missing timestamp')
d = datetime.fromisoformat(value.replace('Z', '+00:00'))
if d.tzinfo is None:
raise ValueError('Timestamp has no timezone')
return d.timestamp()
def eligible(meta, status, namespace):
if meta.get('namespace') != namespace or status.get('phase') != 'Failed':
return False
if not meta.get('uid') or not meta.get('resourceVersion'):
return False
if (meta.get('labels') or {}).get(LABEL) != 'true' or meta.get('ownerReferences'):
return False
if 'kubernetes.io/config.mirror' in (meta.get('annotations') or {}):
return False
finals = meta.get('finalizers') or []
if finals not in ([], [PIN]):
return False
if meta.get('deletionTimestamp') and finals != [PIN]:
return False
for field in ('containerStatuses', 'initContainerStatuses', 'ephemeralContainerStatuses'):
for c in status.get(field) or []:
if not isinstance(c, dict):
raise ValueError('Malformed container status')
state = c.get('state') or {}
if 'running' in state or 'waiting' in state:
return False
return True
def anchor(status, first):
containers = []
for field in ('containerStatuses', 'initContainerStatuses', 'ephemeralContainerStatuses'):
containers.extend(status.get(field) or [])
times = [stamp(c['state']['terminated']['finishedAt']) for c in containers
if isinstance(c.get('state', {}).get('terminated'), dict)
and c['state']['terminated'].get('finishedAt')]
if containers and len(times) == len(containers):
return max(times), 'finishedAt'
return first, 'observedFailed'
def pairs(payload):
if payload.get('size') == 0:
return []
metas, statuses = payload.get('metadataItems'), payload.get('statusItems')
if not isinstance(metas, list) or not isinstance(statuses, list) or len(metas) != len(statuses):
raise ValueError('Incomplete native Pod inventory')
return [(m, s.get('status') or {}) for m, s in zip(metas, statuses)]
def emit(**values):
print('::' + json.dumps({'outputs': values}, separators=(',', ':')) + '::')
tasks:
- id: validate_inputs
type: io.kestra.plugin.scripts.python.Script
containerImage: python:3.13-slim@sha256:70729b46c69b4f1e97c4822c1af3df53a1476cf5ddc6c087c0c10bc3a5678c2f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
memory:
memory: 128MB
timeout: PT30S
env:
CONFIG: "{{ {'cluster': inputs.cluster_id, 'namespace': inputs.namespace,
'retention': inputs.retention_seconds, 'count': inputs.max_pods,
'inventory': inputs.max_inventory, 'bytes': inputs.max_archive_bytes} |
toJson }}"
script: |
import json, os, re
c = json.loads(os.environ['CONFIG'])
assert re.fullmatch(r'[a-z0-9][a-z0-9-]{0,39}', c['cluster']), 'Invalid cluster identity'
assert re.fullmatch(r'[a-z0-9]([-a-z0-9]{0,61}[a-z0-9])?', c['namespace']), 'Invalid namespace'
assert c['namespace'] not in ('kube-system', 'kube-public', 'kube-node-lease'), 'System namespace refused'
assert 1 <= c['retention'] <= 31536000, 'Invalid retention'
assert 1 <= c['count'] <= 50 and c['count'] <= c['inventory'] <= 10000, 'Invalid count ceiling'
assert 1024 <= c['bytes'] <= 1048576, 'Invalid archive ceiling'
- id: list_pods
type: io.kestra.plugin.kubernetes.kubectl.Get
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
fetchType: FETCH
- id: plan
type: io.kestra.plugin.scripts.python.Script
description: Evaluate complete termination or observed-Failed age; preserve
successful and protected Pods.
containerImage: python:3.13-slim@sha256:70729b46c69b4f1e97c4822c1af3df53a1476cf5ddc6c087c0c10bc3a5678c2f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
memory:
memory: 128MB
timeout: PT30S
inputFiles:
helper.py: "{{ vars.helper }}"
inventory.json: "{{ outputs.list_pods | toJson }}"
state.json: "{{ kv(render(vars.state_key), errorOnMissing=false) ?? '{}' }}"
env:
NAMESPACE: "{{ inputs.namespace }}"
RETENTION: "{{ inputs.retention_seconds }}"
MAX_PODS: "{{ inputs.max_pods }}"
MAX_INVENTORY: "{{ inputs.max_inventory }}"
MAX_BYTES: "{{ inputs.max_archive_bytes }}"
script: |
import json, os, time
from helper import *
payload = json.load(open('inventory.json'))
previous = json.load(open('state.json'))
if not isinstance(previous, dict):
raise ValueError('Malformed observation ledger')
rows = pairs(payload)
if len(rows) > int(os.environ['MAX_INVENTORY']):
raise ValueError('Inventory ceiling exceeded')
now = time.time()
state, candidates, held = {}, [], 0
for meta, status in rows:
if not eligible(meta, status, os.environ['NAMESPACE']):
continue
uid = meta['uid']
first = previous.get(uid, {}).get('first', now)
if not isinstance(first, (int, float)) or not 0 < first <= now:
raise ValueError('Invalid observation timestamp')
state[uid] = {'first': first, 'name': meta['name']}
end, basis = anchor(status, first)
if end <= 0 or end > now:
raise ValueError('Invalid terminal timestamp')
if now - end < int(os.environ['RETENTION']):
held += 1
continue
if len(json.dumps({'metadata': meta, 'status': status}).encode()) > int(os.environ['MAX_BYTES']):
raise ValueError('Diagnostic archive ceiling exceeded')
candidates.append({'name': meta['name'], 'uid': uid, 'first': first, 'basis': basis, 'end': end})
candidates.sort(key=lambda x: (x['end'], x['uid']))
targets = candidates[:int(os.environ['MAX_PODS'])]
emit(targets=targets, target_count=len(targets), held=held, deferred=len(candidates)-len(targets), state=json.dumps(state))
- id: save_observation
type: io.kestra.plugin.core.kv.Set
key: "{{ render(vars.state_key) }}"
value: "{{ outputs.plan.vars.state }}"
overwrite: true
- id: summary
type: io.kestra.plugin.core.log.Log
message: "Failed-Pod retention: eligible={{ outputs.plan.vars.target_count }},
held={{ outputs.plan.vars.held }}, deferred={{ outputs.plan.vars.deferred
}}."
- id: live_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ inputs.auto_remediate and not inputs.dry_run and
outputs.plan.vars.target_count > 0 }}"
then:
- id: approve
type: io.kestra.plugin.core.flow.Pause
pauseDuration: PT1H
onResume:
- id: approved
type: BOOL
defaults: false
description: Authorize archiving status diagnostics and removing the listed
failed Pods.
- id: approved_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.approve.onResume.approved ?? false }}"
then:
- id: collect
type: io.kestra.plugin.core.flow.Loop
values: "{{ outputs.plan.vars.targets }}"
concurrencyLimit: 1
tasks:
- id: reread
type: io.kestra.plugin.kubernetes.kubectl.Get
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourcesNames: [ "{{ fromJson(item.value).name }}" ]
fetchType: FETCH_ONE
- id: check_current
type: io.kestra.plugin.scripts.python.Script
containerImage: python:3.13-slim@sha256:70729b46c69b4f1e97c4822c1af3df53a1476cf5ddc6c087c0c10bc3a5678c2f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
memory:
memory: 128MB
timeout: PT30S
inputFiles:
helper.py: "{{ vars.helper }}"
current.json: "{{ outputs.reread | toJson }}"
target.json: "{{ item.value }}"
env:
NAMESPACE: "{{ inputs.namespace }}"
RETENTION: "{{ inputs.retention_seconds }}"
script: |
import json, os, time
from helper import *
p, t = json.load(open('current.json')), json.load(open('target.json'))
m, s = p.get('metadataItem') or {}, (p.get('statusItem') or {}).get('status') or {}
absent = p.get('size') == 0 or m.get('uid') != t['uid']
if absent:
emit(absent=True, needs_pin=False, patch='[]')
else:
assert eligible(m, s, os.environ['NAMESPACE']), 'Target no longer eligible'
end, basis = anchor(s, t['first'])
assert end <= time.time() - int(os.environ['RETENTION']), 'Retention changed'
patch = [{'op':'test','path':'/metadata/uid','value':t['uid']},
{'op':'test','path':'/metadata/resourceVersion','value':m['resourceVersion']},
{'op':'test','path':'/status/phase','value':'Failed'},
{'op':'add','path':'/metadata/finalizers','value':[PIN]}]
emit(absent=False, needs_pin=(m.get('finalizers') or []) == [], patch=json.dumps(patch))
- id: present_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ not outputs.check_current.vars.absent }}"
then:
- id: pin_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.check_current.vars.needs_pin }}"
then:
- id: pin_identity
type: io.kestra.plugin.kubernetes.kubectl.Patch
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourceName: "{{ fromJson(item.value).name }}"
patchStrategy: JSON_PATCH
patch: "{{ outputs.check_current.vars.patch }}"
- id: read_pinned
type: io.kestra.plugin.kubernetes.kubectl.Get
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourcesNames: [ "{{ fromJson(item.value).name }}" ]
fetchType: FETCH_ONE
- id: prepare_archive
type: io.kestra.plugin.scripts.python.Script
containerImage: python:3.13-slim@sha256:70729b46c69b4f1e97c4822c1af3df53a1476cf5ddc6c087c0c10bc3a5678c2f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
memory:
memory: 128MB
timeout: PT30S
inputFiles:
helper.py: "{{ vars.helper }}"
pinned.json: "{{ outputs.read_pinned | toJson }}"
target.json: "{{ item.value }}"
env:
NAMESPACE: "{{ inputs.namespace }}"
MAX_BYTES: "{{ inputs.max_archive_bytes }}"
script: |
import json, os
from helper import *
p, t = json.load(open('pinned.json')), json.load(open('target.json'))
m, s = p['metadataItem'], p['statusItem']['status']
assert m['uid'] == t['uid'] and eligible(m,s,os.environ['NAMESPACE']) and m['finalizers'] == [PIN], 'Identity pin invalid'
archive = json.dumps({'metadata':m,'status':s,'retentionBasis':t['basis'],'anchor':t['end']}, sort_keys=True, separators=(',', ':'))
assert len(archive.encode()) <= int(os.environ['MAX_BYTES']), 'Archive too large'
emit(archive=archive, sha256=hashlib.sha256(archive.encode()).hexdigest())
- id: archive
type: io.kestra.plugin.core.storage.Write
extension: .json
content: "{{ outputs.prepare_archive.vars.archive }}"
- id: verify_archive
type: io.kestra.plugin.scripts.python.Script
containerImage: python:3.13-slim@sha256:70729b46c69b4f1e97c4822c1af3df53a1476cf5ddc6c087c0c10bc3a5678c2f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
memory:
memory: 128MB
timeout: PT30S
inputFiles:
archive.json: "{{ read(outputs.archive.uri) }}"
env:
HASH: "{{ outputs.prepare_archive.vars.sha256 }}"
UID: "{{ fromJson(item.value).uid }}"
script: |
import hashlib, json, os
data = open('archive.json','rb').read()
assert hashlib.sha256(data).hexdigest() == os.environ['HASH'], 'Archive readback mismatch'
assert json.loads(data)['metadata']['uid'] == os.environ['UID'], 'Archive identity mismatch'
- id: archive_receipt
type: io.kestra.plugin.core.kv.Set
key: "pod-diagnostic-{{ inputs.cluster_id }}-{{ fromJson(item.value).uid }}"
value: "{{ {'uid': fromJson(item.value).uid, 'uri': outputs.archive.uri,
'sha256': outputs.prepare_archive.vars.sha256} | toJson
}}"
overwrite: true
- id: before_delete
type: io.kestra.plugin.kubernetes.kubectl.Get
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourcesNames: [ "{{ fromJson(item.value).name }}" ]
fetchType: FETCH_ONE
- id: fence_delete
type: io.kestra.plugin.scripts.python.Script
containerImage: python:3.13-slim@sha256:70729b46c69b4f1e97c4822c1af3df53a1476cf5ddc6c087c0c10bc3a5678c2f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
memory:
memory: 128MB
timeout: PT30S
inputFiles:
helper.py: "{{ vars.helper }}"
current.json: "{{ outputs.before_delete | toJson }}"
archived.json: "{{ read(outputs.archive.uri) }}"
receipt.json: "{{ kv('pod-diagnostic-' ~ inputs.cluster_id ~ '-' ~
fromJson(item.value).uid) }}"
env:
UID: "{{ fromJson(item.value).uid }}"
NAMESPACE: "{{ inputs.namespace }}"
script: |
import json, os
from helper import *
p = json.load(open('current.json'))
m, s = p['metadataItem'], p['statusItem']['status']
a, r = json.load(open('archived.json')), json.load(open('receipt.json'))
assert m['uid'] == os.environ['UID'] and eligible(m,s,os.environ['NAMESPACE']) and m['finalizers'] == [PIN], 'Pre-delete fence invalid'
assert s == a['status'], 'Diagnostic status changed after archive'
assert r['uid'] == m['uid'] and r['sha256'] == hashlib.sha256(open('archived.json','rb').read()).hexdigest(), 'Archive receipt readback invalid'
- id: delete_pod
type: io.kestra.plugin.kubernetes.kubectl.Delete
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourcesNames: [ "{{ fromJson(item.value).name }}" ]
- id: read_deleting
type: io.kestra.plugin.kubernetes.kubectl.Get
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourcesNames: [ "{{ fromJson(item.value).name }}" ]
fetchType: FETCH_ONE
- id: prepare_release
type: io.kestra.plugin.scripts.python.Script
containerImage: python:3.13-slim@sha256:70729b46c69b4f1e97c4822c1af3df53a1476cf5ddc6c087c0c10bc3a5678c2f
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
memory:
memory: 128MB
timeout: PT30S
inputFiles:
helper.py: "{{ vars.helper }}"
deleting.json: "{{ outputs.read_deleting | toJson }}"
env:
UID: "{{ fromJson(item.value).uid }}"
NAMESPACE: "{{ inputs.namespace }}"
script: |
import json, os
from helper import *
p = json.load(open('deleting.json'))
m, s = p['metadataItem'], p['statusItem']['status']
assert m['uid'] == os.environ['UID'] and eligible(m,s,os.environ['NAMESPACE']), 'Deleting target changed'
assert m.get('deletionTimestamp') and m['finalizers'] == [PIN], 'Delete did not mark pinned Pod'
emit(patch=json.dumps([{'op':'test','path':'/metadata/uid','value':m['uid']},
{'op':'test','path':'/metadata/resourceVersion','value':m['resourceVersion']},
{'op':'test','path':'/status/phase','value':'Failed'},
{'op':'test','path':'/metadata/finalizers','value':[PIN]},
{'op':'remove','path':'/metadata/finalizers/0'}]))
- id: release_pin
type: io.kestra.plugin.kubernetes.kubectl.Patch
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourceName: "{{ fromJson(item.value).name }}"
patchStrategy: JSON_PATCH
patch: "{{ outputs.prepare_release.vars.patch }}"
- id: settle
type: io.kestra.plugin.core.flow.Sleep
duration: PT2S
- id: read_after
type: io.kestra.plugin.kubernetes.kubectl.Get
timeout: PT30S
connection:
masterUrl: "{{ inputs.api_url }}"
oauthToken: "{{ inputs.token }}"
caCertData: "{{ inputs.ca_cert_data }}"
trustCerts: "{{ inputs.trust_certs }}"
namespace: "{{ inputs.namespace }}"
resourceType: pods
resourcesNames: [ "{{ fromJson(item.value).name }}" ]
fetchType: FETCH_ONE
- id: verify_absent
type: io.kestra.plugin.core.flow.If
condition: "{{ (outputs.read_after.metadataItem.uid ?? '') ==
fromJson(item.value).uid }}"
then:
- id: recovery_failed
type: io.kestra.plugin.core.execution.Fail
errorMessage: Archived failed Pod UID remains in the Kubernetes API.
triggers:
- id: daily
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 8 * * *"
disabled: true
inputs:
dry_run: true
auto_remediate: false
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Failed-Pod diagnostic retention failed. Execution {{ execution.id
}}. Inspect collector-pinned Pods before retrying."
