id: k8s-paused-deployment-rollout-resume-guard
namespace: company.team
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: k8s_master_url
type: STRING
defaults: "{{ secret('K8S_MASTER_URL') }}"
description: Kubernetes API URL reachable by the Kestra worker.
- id: k8s_token
type: STRING
defaults: "{{ secret('K8S_TOKEN') }}"
description: Token allowed to run read pods and patch the selected Deployment.
- id: target_namespace
type: STRING
defaults: production
description: Namespace containing the target and the read-only ServiceAccount.
- id: deployment_name
type: STRING
description: Exactly one Deployment to inspect and possibly resume.
- id: reader_service_account
type: STRING
defaults: kestra-rollout-reader
description: Namespace ServiceAccount with get/list on Deployments, ReplicaSets
and Pods only.
- id: min_paused_age_minutes
type: INT
defaults: 30
min: 1
max: 10080
- id: max_replicas
type: INT
defaults: 10
min: 1
max: 100
- id: max_objects
type: INT
defaults: 100
min: 10
max: 500
- id: verify_seconds
type: INT
defaults: 120
min: 10
max: 600
- id: dry_run
type: BOOL
defaults: true
- id: auto_remediate
type: BOOL
defaults: false
- id: trust_certs
type: BOOL
defaults: false
- id: reader_image
type: STRING
defaults: python:3.12-slim@sha256:dddfd7e07f9d15aeeca61529320492139d21cac7f0070c00609243e51e4e0016
- id: file_sidecar_image
type: STRING
defaults: busybox:1.36@sha256:73aaf090f3d85aa34ee199857f03fa3a95c8ede2ffd4cc2cdb5b94e566b11662
- id: slack_webhook_url
type: STRING
defaults: "{{ secret('SLACK_WEBHOOK_URL') }}"
tasks:
- id: inventory
type: io.kestra.plugin.kubernetes.core.PodCreate
connection:
masterUrl: '{{ inputs.k8s_master_url }}'
oauthToken: '{{ inputs.k8s_token }}'
trustCerts: '{{ inputs.trust_certs }}'
namespace: '{{ inputs.target_namespace }}'
waitUntilRunning: PT60S
waitRunning: PT12M
waitForLogInterval: PT1S
timeout: PT14M
fileSidecar:
image: '{{ inputs.file_sidecar_image }}'
outputFiles:
- plan.json
spec:
serviceAccountName: '{{ inputs.reader_service_account }}'
restartPolicy: Never
containers:
- name: reader
image: '{{ inputs.reader_image }}'
imagePullPolicy: IfNotPresent
resources:
requests:
memory: 32Mi
cpu: 50m
limits:
memory: 128Mi
cpu: 300m
env:
- name: MODE
value: inventory
- name: TARGET_NAMESPACE
value: '{{ inputs.target_namespace }}'
- name: DEPLOYMENT_NAME
value: '{{ inputs.deployment_name }}'
- name: GRACE_MINUTES
value: '{{ inputs.min_paused_age_minutes }}'
- name: MAX_REPLICAS
value: '{{ inputs.max_replicas }}'
- name: MAX_OBJECTS
value: '{{ inputs.max_objects }}'
- name: VERIFY_SECONDS
value: '{{ inputs.verify_seconds }}'
command:
- python
- -c
- |
import datetime
import json
import os
import pathlib
import ssl
import time
import urllib.error
import urllib.parse
import urllib.request
ROOT = pathlib.Path('/kestra/working-dir')
NS = os.environ['TARGET_NAMESPACE']
NAME = os.environ['DEPLOYMENT_NAME']
BASE = 'https://' + os.environ['KUBERNETES_SERVICE_HOST'] + ':' + os.environ['KUBERNETES_SERVICE_PORT_HTTPS']
SA = pathlib.Path('/var/run/secrets/kubernetes.io/serviceaccount')
CTX = ssl.create_default_context(cafile=str(SA / 'ca.crt'))
TOKEN = (SA / 'token').read_text().strip()
GRACE = int(os.environ['GRACE_MINUTES'])
CEILING = int(os.environ['MAX_REPLICAS'])
MAX_OBJECTS = int(os.environ['MAX_OBJECTS'])
MODE = os.environ['MODE']
APPROVE = 'kestra.io/resume-paused-rollout'
HOLD = 'kestra.io/hold-paused-rollout'
def get(path):
req = urllib.request.Request(BASE + path, headers={'Authorization': 'Bearer ' + TOKEN})
with urllib.request.urlopen(req, context=CTX, timeout=10) as r:
return json.load(r)
def deployment():
return get('/apis/apps/v1/namespaces/' + NS + '/deployments/' + NAME)
def template(spec):
result = json.loads(json.dumps(spec['template']))
result.get('metadata', {}).get('labels', {}).pop('pod-template-hash', None)
return result
def resources(dep):
selector = dep['spec']['selector']
if selector.get('matchExpressions') or not selector.get('matchLabels'):
raise RuntimeError('Only a nonempty matchLabels selector is supported')
labels = ','.join(k + '=' + v for k, v in sorted(selector['matchLabels'].items()))
query = '?' + urllib.parse.urlencode({'labelSelector': labels, 'limit': MAX_OBJECTS + 1})
sets = get('/apis/apps/v1/namespaces/' + NS + '/replicasets' + query)
pods = get('/api/v1/namespaces/' + NS + '/pods' + query)
for data in [sets, pods]:
if data.get('metadata', {}).get('continue') or len(data['items']) > MAX_OBJECTS:
raise RuntimeError('Resource read exceeds max_objects')
uid = dep['metadata']['uid']
sets = [s for s in sets['items'] if any(o.get('uid') == uid and o.get('controller') for o in s['metadata'].get('ownerReferences', []))]
return sets, pods['items']
def classify(dep, sets):
meta, spec, status = dep['metadata'], dep['spec'], dep.get('status', {})
annotations = meta.get('annotations', {})
paused = next((c for c in status.get('conditions', []) if c['type'] == 'Progressing' and c.get('reason') == 'DeploymentPaused'), None)
matching = [s for s in sets if template(s['spec']) == template(spec)]
age = None
if paused and paused.get('lastUpdateTime'):
since = datetime.datetime.fromisoformat(paused['lastUpdateTime'].replace('Z', '+00:00'))
age = (datetime.datetime.now(datetime.timezone.utc) - since).total_seconds() / 60
desired = spec.get('replicas', 1)
if spec.get('paused') is not True:
verdict = 'not_paused'
elif annotations.get(APPROVE) != 'true':
verdict = 'not_opted_in'
elif HOLD in annotations:
verdict = 'intentional_pause_protected'
elif meta.get('deletionTimestamp'):
verdict = 'deleting'
elif not paused or age is None:
verdict = 'no_pause_timestamp'
elif age < GRACE:
verdict = 'within_grace'
elif desired < 1 or desired > CEILING:
verdict = 'replica_bound'
elif not sets:
verdict = 'no_existing_replicaset'
elif matching:
verdict = 'no_pending_template'
else:
verdict = 'eligible'
return {'name': NAME, 'namespace': NS, 'uid': meta['uid'], 'generation': meta['generation'], 'annotations': annotations, 'spec': spec, 'desired': desired, 'verdict': verdict, 'age_minutes': age, 'replicasets': [{'name': s['metadata']['name'], 'uid': s['metadata']['uid'], 'replicas': s['spec'].get('replicas', 0)} for s in sets]}
def save(name, obj):
(ROOT / name).write_text(json.dumps(obj, sort_keys=True))
dep = deployment()
sets, pods = resources(dep)
plan = classify(dep, sets)
if MODE == 'inventory':
save('plan.json', plan)
print(json.dumps({k: plan[k] for k in ['name', 'verdict', 'age_minutes', 'desired']}), flush=True)
elif MODE == 'recheck':
original = json.loads((ROOT / 'approved-plan.json').read_text())
for key in ['uid', 'generation', 'spec', 'annotations']:
if plan[key] != original[key]:
raise RuntimeError('Approved snapshot changed: ' + key)
if plan['verdict'] != 'eligible':
raise RuntimeError('Target is no longer eligible: ' + plan['verdict'])
patch = [{'op': 'test', 'path': '/metadata/uid', 'value': plan['uid']}, {'op': 'test', 'path': '/metadata/generation', 'value': plan['generation']}, {'op': 'test', 'path': '/metadata/annotations', 'value': plan['annotations']}, {'op': 'test', 'path': '/spec', 'value': plan['spec']}, {'op': 'replace', 'path': '/spec/paused', 'value': False}]
save('patch.json', patch)
print('Approved snapshot still eligible. Conditional patch prepared.', flush=True)
elif MODE == 'verify':
original = json.loads((ROOT / 'approved-plan.json').read_text())
deadline = time.monotonic() + int(os.environ['VERIFY_SECONDS'])
result = {}
while True:
dep = deployment()
sets, pods = resources(dep)
meta, spec, status = dep['metadata'], dep['spec'], dep.get('status', {})
intended = template(original['spec'])
if meta['uid'] != original['uid'] or template(spec) != intended or spec.get('replicas', 1) != original['desired']:
raise RuntimeError('Target identity, template or replica count changed after approval')
matching = [s for s in sets if template(s['spec']) == intended]
set_uids = {s['metadata']['uid'] for s in sets}
intended_uids = {s['metadata']['uid'] for s in matching}
owned = [p for p in pods if not p['metadata'].get('deletionTimestamp') and any(o.get('uid') in set_uids and o.get('controller') for o in p['metadata'].get('ownerReferences', []))]
ready = [p for p in owned if any(o.get('uid') in intended_uids and o.get('controller') for o in p['metadata'].get('ownerReferences', [])) and any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in p.get('status', {}).get('conditions', [])) and [(c['name'], c['image']) for c in p['spec']['containers']] == [(c['name'], c['image']) for c in intended['spec']['containers']] and all(c.get('ready') and c.get('imageID') for c in p.get('status', {}).get('containerStatuses', []))]
desired = original['desired']
result = {'paused': spec.get('paused', False), 'generation': meta['generation'], 'observedGeneration': status.get('observedGeneration', 0), 'updatedReplicas': status.get('updatedReplicas', 0), 'replicas': status.get('replicas', 0), 'availableReplicas': status.get('availableReplicas', 0), 'ready_intended_pods': len(ready), 'owned_live_pods': len(owned), 'desired': desired, 'images': [c['image'] for c in intended['spec']['containers']], 'pod_uids': [p['metadata']['uid'] for p in ready]}
ok = not result['paused'] and result['generation'] == original['generation'] + 1 and result['observedGeneration'] == result['generation'] and all(result[k] == desired for k in ['updatedReplicas', 'replicas', 'availableReplicas', 'ready_intended_pods', 'owned_live_pods']) and len(matching) == 1
if ok:
save('verification.json', result)
print('Intended rollout completed: ' + json.dumps(result), flush=True)
break
if time.monotonic() >= deadline:
save('verification.json', result)
raise RuntimeError('Intended rollout did not complete within bound: ' + json.dumps(result))
time.sleep(2)
else:
raise RuntimeError('Unknown mode')
- id: report
type: io.kestra.plugin.core.log.Log
message: '{{ read(outputs.inventory.outputFiles["plan.json"]) }}'
- id: eligible
type: io.kestra.plugin.core.flow.If
condition: '{{ (read(outputs.inventory.outputFiles["plan.json"]) |
jq(".verdict") | first) == "eligible" }}'
then:
- id: notify_plan
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: '{{ inputs.slack_webhook_url }}'
messageText: Paused rollout found for {{ inputs.target_namespace }}/{{
inputs.deployment_name }}. dry_run={{ inputs.dry_run }}. Review the
inventory before approving.
- id: may_change
type: io.kestra.plugin.core.flow.If
condition: '{{ inputs.auto_remediate and not inputs.dry_run }}'
then:
- id: approve_resume
type: io.kestra.plugin.core.flow.Pause
timeout: PT1H
onResume:
- id: approved
type: BOOL
defaults: false
description: Resume the exact template and replica count recorded in inventory?
- id: approved
type: io.kestra.plugin.core.flow.If
condition: '{{ outputs.approve_resume.onResume.approved ?? false }}'
then:
- id: recheck
type: io.kestra.plugin.kubernetes.core.PodCreate
connection:
masterUrl: '{{ inputs.k8s_master_url }}'
oauthToken: '{{ inputs.k8s_token }}'
trustCerts: '{{ inputs.trust_certs }}'
namespace: '{{ inputs.target_namespace }}'
waitUntilRunning: PT60S
waitRunning: PT12M
waitForLogInterval: PT1S
timeout: PT14M
fileSidecar:
image: '{{ inputs.file_sidecar_image }}'
outputFiles:
- patch.json
spec:
serviceAccountName: '{{ inputs.reader_service_account }}'
restartPolicy: Never
containers:
- name: reader
image: '{{ inputs.reader_image }}'
imagePullPolicy: IfNotPresent
resources:
requests:
memory: 32Mi
cpu: 50m
limits:
memory: 128Mi
cpu: 300m
env:
- name: MODE
value: recheck
- name: TARGET_NAMESPACE
value: '{{ inputs.target_namespace }}'
- name: DEPLOYMENT_NAME
value: '{{ inputs.deployment_name }}'
- name: GRACE_MINUTES
value: '{{ inputs.min_paused_age_minutes }}'
- name: MAX_REPLICAS
value: '{{ inputs.max_replicas }}'
- name: MAX_OBJECTS
value: '{{ inputs.max_objects }}'
- name: VERIFY_SECONDS
value: '{{ inputs.verify_seconds }}'
command:
- python
- -c
- |
import datetime
import json
import os
import pathlib
import ssl
import time
import urllib.error
import urllib.parse
import urllib.request
ROOT = pathlib.Path('/kestra/working-dir')
NS = os.environ['TARGET_NAMESPACE']
NAME = os.environ['DEPLOYMENT_NAME']
BASE = 'https://' + os.environ['KUBERNETES_SERVICE_HOST'] + ':' + os.environ['KUBERNETES_SERVICE_PORT_HTTPS']
SA = pathlib.Path('/var/run/secrets/kubernetes.io/serviceaccount')
CTX = ssl.create_default_context(cafile=str(SA / 'ca.crt'))
TOKEN = (SA / 'token').read_text().strip()
GRACE = int(os.environ['GRACE_MINUTES'])
CEILING = int(os.environ['MAX_REPLICAS'])
MAX_OBJECTS = int(os.environ['MAX_OBJECTS'])
MODE = os.environ['MODE']
APPROVE = 'kestra.io/resume-paused-rollout'
HOLD = 'kestra.io/hold-paused-rollout'
def get(path):
req = urllib.request.Request(BASE + path, headers={'Authorization': 'Bearer ' + TOKEN})
with urllib.request.urlopen(req, context=CTX, timeout=10) as r:
return json.load(r)
def deployment():
return get('/apis/apps/v1/namespaces/' + NS + '/deployments/' + NAME)
def template(spec):
result = json.loads(json.dumps(spec['template']))
result.get('metadata', {}).get('labels', {}).pop('pod-template-hash', None)
return result
def resources(dep):
selector = dep['spec']['selector']
if selector.get('matchExpressions') or not selector.get('matchLabels'):
raise RuntimeError('Only a nonempty matchLabels selector is supported')
labels = ','.join(k + '=' + v for k, v in sorted(selector['matchLabels'].items()))
query = '?' + urllib.parse.urlencode({'labelSelector': labels, 'limit': MAX_OBJECTS + 1})
sets = get('/apis/apps/v1/namespaces/' + NS + '/replicasets' + query)
pods = get('/api/v1/namespaces/' + NS + '/pods' + query)
for data in [sets, pods]:
if data.get('metadata', {}).get('continue') or len(data['items']) > MAX_OBJECTS:
raise RuntimeError('Resource read exceeds max_objects')
uid = dep['metadata']['uid']
sets = [s for s in sets['items'] if any(o.get('uid') == uid and o.get('controller') for o in s['metadata'].get('ownerReferences', []))]
return sets, pods['items']
def classify(dep, sets):
meta, spec, status = dep['metadata'], dep['spec'], dep.get('status', {})
annotations = meta.get('annotations', {})
paused = next((c for c in status.get('conditions', []) if c['type'] == 'Progressing' and c.get('reason') == 'DeploymentPaused'), None)
matching = [s for s in sets if template(s['spec']) == template(spec)]
age = None
if paused and paused.get('lastUpdateTime'):
since = datetime.datetime.fromisoformat(paused['lastUpdateTime'].replace('Z', '+00:00'))
age = (datetime.datetime.now(datetime.timezone.utc) - since).total_seconds() / 60
desired = spec.get('replicas', 1)
if spec.get('paused') is not True:
verdict = 'not_paused'
elif annotations.get(APPROVE) != 'true':
verdict = 'not_opted_in'
elif HOLD in annotations:
verdict = 'intentional_pause_protected'
elif meta.get('deletionTimestamp'):
verdict = 'deleting'
elif not paused or age is None:
verdict = 'no_pause_timestamp'
elif age < GRACE:
verdict = 'within_grace'
elif desired < 1 or desired > CEILING:
verdict = 'replica_bound'
elif not sets:
verdict = 'no_existing_replicaset'
elif matching:
verdict = 'no_pending_template'
else:
verdict = 'eligible'
return {'name': NAME, 'namespace': NS, 'uid': meta['uid'], 'generation': meta['generation'], 'annotations': annotations, 'spec': spec, 'desired': desired, 'verdict': verdict, 'age_minutes': age, 'replicasets': [{'name': s['metadata']['name'], 'uid': s['metadata']['uid'], 'replicas': s['spec'].get('replicas', 0)} for s in sets]}
def save(name, obj):
(ROOT / name).write_text(json.dumps(obj, sort_keys=True))
dep = deployment()
sets, pods = resources(dep)
plan = classify(dep, sets)
if MODE == 'inventory':
save('plan.json', plan)
print(json.dumps({k: plan[k] for k in ['name', 'verdict', 'age_minutes', 'desired']}), flush=True)
elif MODE == 'recheck':
original = json.loads((ROOT / 'approved-plan.json').read_text())
for key in ['uid', 'generation', 'spec', 'annotations']:
if plan[key] != original[key]:
raise RuntimeError('Approved snapshot changed: ' + key)
if plan['verdict'] != 'eligible':
raise RuntimeError('Target is no longer eligible: ' + plan['verdict'])
patch = [{'op': 'test', 'path': '/metadata/uid', 'value': plan['uid']}, {'op': 'test', 'path': '/metadata/generation', 'value': plan['generation']}, {'op': 'test', 'path': '/metadata/annotations', 'value': plan['annotations']}, {'op': 'test', 'path': '/spec', 'value': plan['spec']}, {'op': 'replace', 'path': '/spec/paused', 'value': False}]
save('patch.json', patch)
print('Approved snapshot still eligible. Conditional patch prepared.', flush=True)
elif MODE == 'verify':
original = json.loads((ROOT / 'approved-plan.json').read_text())
deadline = time.monotonic() + int(os.environ['VERIFY_SECONDS'])
result = {}
while True:
dep = deployment()
sets, pods = resources(dep)
meta, spec, status = dep['metadata'], dep['spec'], dep.get('status', {})
intended = template(original['spec'])
if meta['uid'] != original['uid'] or template(spec) != intended or spec.get('replicas', 1) != original['desired']:
raise RuntimeError('Target identity, template or replica count changed after approval')
matching = [s for s in sets if template(s['spec']) == intended]
set_uids = {s['metadata']['uid'] for s in sets}
intended_uids = {s['metadata']['uid'] for s in matching}
owned = [p for p in pods if not p['metadata'].get('deletionTimestamp') and any(o.get('uid') in set_uids and o.get('controller') for o in p['metadata'].get('ownerReferences', []))]
ready = [p for p in owned if any(o.get('uid') in intended_uids and o.get('controller') for o in p['metadata'].get('ownerReferences', [])) and any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in p.get('status', {}).get('conditions', [])) and [(c['name'], c['image']) for c in p['spec']['containers']] == [(c['name'], c['image']) for c in intended['spec']['containers']] and all(c.get('ready') and c.get('imageID') for c in p.get('status', {}).get('containerStatuses', []))]
desired = original['desired']
result = {'paused': spec.get('paused', False), 'generation': meta['generation'], 'observedGeneration': status.get('observedGeneration', 0), 'updatedReplicas': status.get('updatedReplicas', 0), 'replicas': status.get('replicas', 0), 'availableReplicas': status.get('availableReplicas', 0), 'ready_intended_pods': len(ready), 'owned_live_pods': len(owned), 'desired': desired, 'images': [c['image'] for c in intended['spec']['containers']], 'pod_uids': [p['metadata']['uid'] for p in ready]}
ok = not result['paused'] and result['generation'] == original['generation'] + 1 and result['observedGeneration'] == result['generation'] and all(result[k] == desired for k in ['updatedReplicas', 'replicas', 'availableReplicas', 'ready_intended_pods', 'owned_live_pods']) and len(matching) == 1
if ok:
save('verification.json', result)
print('Intended rollout completed: ' + json.dumps(result), flush=True)
break
if time.monotonic() >= deadline:
save('verification.json', result)
raise RuntimeError('Intended rollout did not complete within bound: ' + json.dumps(result))
time.sleep(2)
else:
raise RuntimeError('Unknown mode')
inputFiles:
approved-plan.json: '{{ read(outputs.inventory.outputFiles["plan.json"]) }}'
- id: resume
type: io.kestra.plugin.kubernetes.kubectl.Patch
connection:
masterUrl: '{{ inputs.k8s_master_url }}'
oauthToken: '{{ inputs.k8s_token }}'
trustCerts: '{{ inputs.trust_certs }}'
namespace: '{{ inputs.target_namespace }}'
resourceType: Deployment
apiGroup: apps
apiVersion: v1
resourceName: '{{ inputs.deployment_name }}'
patchStrategy: JSON_PATCH
patch: '{{ read(outputs.recheck.outputFiles["patch.json"]) }}'
timeout: PT30S
- id: verify
type: io.kestra.plugin.kubernetes.core.PodCreate
connection:
masterUrl: '{{ inputs.k8s_master_url }}'
oauthToken: '{{ inputs.k8s_token }}'
trustCerts: '{{ inputs.trust_certs }}'
namespace: '{{ inputs.target_namespace }}'
waitUntilRunning: PT60S
waitRunning: PT12M
waitForLogInterval: PT1S
timeout: PT14M
fileSidecar:
image: '{{ inputs.file_sidecar_image }}'
outputFiles:
- verification.json
spec:
serviceAccountName: '{{ inputs.reader_service_account }}'
restartPolicy: Never
containers:
- name: reader
image: '{{ inputs.reader_image }}'
imagePullPolicy: IfNotPresent
resources:
requests:
memory: 32Mi
cpu: 50m
limits:
memory: 128Mi
cpu: 300m
env:
- name: MODE
value: verify
- name: TARGET_NAMESPACE
value: '{{ inputs.target_namespace }}'
- name: DEPLOYMENT_NAME
value: '{{ inputs.deployment_name }}'
- name: GRACE_MINUTES
value: '{{ inputs.min_paused_age_minutes }}'
- name: MAX_REPLICAS
value: '{{ inputs.max_replicas }}'
- name: MAX_OBJECTS
value: '{{ inputs.max_objects }}'
- name: VERIFY_SECONDS
value: '{{ inputs.verify_seconds }}'
command:
- python
- -c
- |
import datetime
import json
import os
import pathlib
import ssl
import time
import urllib.error
import urllib.parse
import urllib.request
ROOT = pathlib.Path('/kestra/working-dir')
NS = os.environ['TARGET_NAMESPACE']
NAME = os.environ['DEPLOYMENT_NAME']
BASE = 'https://' + os.environ['KUBERNETES_SERVICE_HOST'] + ':' + os.environ['KUBERNETES_SERVICE_PORT_HTTPS']
SA = pathlib.Path('/var/run/secrets/kubernetes.io/serviceaccount')
CTX = ssl.create_default_context(cafile=str(SA / 'ca.crt'))
TOKEN = (SA / 'token').read_text().strip()
GRACE = int(os.environ['GRACE_MINUTES'])
CEILING = int(os.environ['MAX_REPLICAS'])
MAX_OBJECTS = int(os.environ['MAX_OBJECTS'])
MODE = os.environ['MODE']
APPROVE = 'kestra.io/resume-paused-rollout'
HOLD = 'kestra.io/hold-paused-rollout'
def get(path):
req = urllib.request.Request(BASE + path, headers={'Authorization': 'Bearer ' + TOKEN})
with urllib.request.urlopen(req, context=CTX, timeout=10) as r:
return json.load(r)
def deployment():
return get('/apis/apps/v1/namespaces/' + NS + '/deployments/' + NAME)
def template(spec):
result = json.loads(json.dumps(spec['template']))
result.get('metadata', {}).get('labels', {}).pop('pod-template-hash', None)
return result
def resources(dep):
selector = dep['spec']['selector']
if selector.get('matchExpressions') or not selector.get('matchLabels'):
raise RuntimeError('Only a nonempty matchLabels selector is supported')
labels = ','.join(k + '=' + v for k, v in sorted(selector['matchLabels'].items()))
query = '?' + urllib.parse.urlencode({'labelSelector': labels, 'limit': MAX_OBJECTS + 1})
sets = get('/apis/apps/v1/namespaces/' + NS + '/replicasets' + query)
pods = get('/api/v1/namespaces/' + NS + '/pods' + query)
for data in [sets, pods]:
if data.get('metadata', {}).get('continue') or len(data['items']) > MAX_OBJECTS:
raise RuntimeError('Resource read exceeds max_objects')
uid = dep['metadata']['uid']
sets = [s for s in sets['items'] if any(o.get('uid') == uid and o.get('controller') for o in s['metadata'].get('ownerReferences', []))]
return sets, pods['items']
def classify(dep, sets):
meta, spec, status = dep['metadata'], dep['spec'], dep.get('status', {})
annotations = meta.get('annotations', {})
paused = next((c for c in status.get('conditions', []) if c['type'] == 'Progressing' and c.get('reason') == 'DeploymentPaused'), None)
matching = [s for s in sets if template(s['spec']) == template(spec)]
age = None
if paused and paused.get('lastUpdateTime'):
since = datetime.datetime.fromisoformat(paused['lastUpdateTime'].replace('Z', '+00:00'))
age = (datetime.datetime.now(datetime.timezone.utc) - since).total_seconds() / 60
desired = spec.get('replicas', 1)
if spec.get('paused') is not True:
verdict = 'not_paused'
elif annotations.get(APPROVE) != 'true':
verdict = 'not_opted_in'
elif HOLD in annotations:
verdict = 'intentional_pause_protected'
elif meta.get('deletionTimestamp'):
verdict = 'deleting'
elif not paused or age is None:
verdict = 'no_pause_timestamp'
elif age < GRACE:
verdict = 'within_grace'
elif desired < 1 or desired > CEILING:
verdict = 'replica_bound'
elif not sets:
verdict = 'no_existing_replicaset'
elif matching:
verdict = 'no_pending_template'
else:
verdict = 'eligible'
return {'name': NAME, 'namespace': NS, 'uid': meta['uid'], 'generation': meta['generation'], 'annotations': annotations, 'spec': spec, 'desired': desired, 'verdict': verdict, 'age_minutes': age, 'replicasets': [{'name': s['metadata']['name'], 'uid': s['metadata']['uid'], 'replicas': s['spec'].get('replicas', 0)} for s in sets]}
def save(name, obj):
(ROOT / name).write_text(json.dumps(obj, sort_keys=True))
dep = deployment()
sets, pods = resources(dep)
plan = classify(dep, sets)
if MODE == 'inventory':
save('plan.json', plan)
print(json.dumps({k: plan[k] for k in ['name', 'verdict', 'age_minutes', 'desired']}), flush=True)
elif MODE == 'recheck':
original = json.loads((ROOT / 'approved-plan.json').read_text())
for key in ['uid', 'generation', 'spec', 'annotations']:
if plan[key] != original[key]:
raise RuntimeError('Approved snapshot changed: ' + key)
if plan['verdict'] != 'eligible':
raise RuntimeError('Target is no longer eligible: ' + plan['verdict'])
patch = [{'op': 'test', 'path': '/metadata/uid', 'value': plan['uid']}, {'op': 'test', 'path': '/metadata/generation', 'value': plan['generation']}, {'op': 'test', 'path': '/metadata/annotations', 'value': plan['annotations']}, {'op': 'test', 'path': '/spec', 'value': plan['spec']}, {'op': 'replace', 'path': '/spec/paused', 'value': False}]
save('patch.json', patch)
print('Approved snapshot still eligible. Conditional patch prepared.', flush=True)
elif MODE == 'verify':
original = json.loads((ROOT / 'approved-plan.json').read_text())
deadline = time.monotonic() + int(os.environ['VERIFY_SECONDS'])
result = {}
while True:
dep = deployment()
sets, pods = resources(dep)
meta, spec, status = dep['metadata'], dep['spec'], dep.get('status', {})
intended = template(original['spec'])
if meta['uid'] != original['uid'] or template(spec) != intended or spec.get('replicas', 1) != original['desired']:
raise RuntimeError('Target identity, template or replica count changed after approval')
matching = [s for s in sets if template(s['spec']) == intended]
set_uids = {s['metadata']['uid'] for s in sets}
intended_uids = {s['metadata']['uid'] for s in matching}
owned = [p for p in pods if not p['metadata'].get('deletionTimestamp') and any(o.get('uid') in set_uids and o.get('controller') for o in p['metadata'].get('ownerReferences', []))]
ready = [p for p in owned if any(o.get('uid') in intended_uids and o.get('controller') for o in p['metadata'].get('ownerReferences', [])) and any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in p.get('status', {}).get('conditions', [])) and [(c['name'], c['image']) for c in p['spec']['containers']] == [(c['name'], c['image']) for c in intended['spec']['containers']] and all(c.get('ready') and c.get('imageID') for c in p.get('status', {}).get('containerStatuses', []))]
desired = original['desired']
result = {'paused': spec.get('paused', False), 'generation': meta['generation'], 'observedGeneration': status.get('observedGeneration', 0), 'updatedReplicas': status.get('updatedReplicas', 0), 'replicas': status.get('replicas', 0), 'availableReplicas': status.get('availableReplicas', 0), 'ready_intended_pods': len(ready), 'owned_live_pods': len(owned), 'desired': desired, 'images': [c['image'] for c in intended['spec']['containers']], 'pod_uids': [p['metadata']['uid'] for p in ready]}
ok = not result['paused'] and result['generation'] == original['generation'] + 1 and result['observedGeneration'] == result['generation'] and all(result[k] == desired for k in ['updatedReplicas', 'replicas', 'availableReplicas', 'ready_intended_pods', 'owned_live_pods']) and len(matching) == 1
if ok:
save('verification.json', result)
print('Intended rollout completed: ' + json.dumps(result), flush=True)
break
if time.monotonic() >= deadline:
save('verification.json', result)
raise RuntimeError('Intended rollout did not complete within bound: ' + json.dumps(result))
time.sleep(2)
else:
raise RuntimeError('Unknown mode')
inputFiles:
approved-plan.json: '{{ read(outputs.inventory.outputFiles["plan.json"]) }}'
else:
- id: declined
type: io.kestra.plugin.core.log.Log
message: Approval declined. Deployment unchanged.
else:
- id: safe_mode
type: io.kestra.plugin.core.log.Log
message: Plan only. Set dry_run=false and auto_remediate=true to request
approval.
else:
- id: spared
type: io.kestra.plugin.core.log.Log
message: No eligible paused rollout. Read inventory for the precise exclusion.
errors:
- id: notify_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: '{{ inputs.slack_webhook_url }}'
messageText: Paused rollout guard failed for {{ inputs.target_namespace }}/{{
inputs.deployment_name }}. Execution {{ execution.id }}. Check the current
Deployment and inventory before retrying.
triggers:
- id: daily_check
type: io.kestra.plugin.core.trigger.Schedule
cron: 0 8 * * *
disabled: true
inputs:
dry_run: true
auto_remediate: false
deployment_name: replace-me
