Self-Service Kubernetes Namespace Provisioning with Policy Tiers, Approval, Verification and TTL Lifecycle

id: kubernetes-namespace-self-service-provisioning
namespace: company.platform
description: |
  Golden-path namespaces on demand. A request (portal webhook or the Execute form)
  is checked against the platform team's tiers, approved automatically or by a
  person, applied as a namespace with quota, limits, RBAC, network policy and Pod
  Security labels, read back from the cluster to verify it, registered with an
  owner and expiry, and torn down after approval when it expires.

labels:
  team: platform
  use-case: namespace-self-service

inputs:
  - id: mode
    type: SELECT
    displayName: Action
    description: "`request` provisions a namespace. `sweep` handles expiring
      namespaces (the daily schedule does this)."
    values:
      - request
      - sweep
    defaults: request
  - id: team
    type: STRING
    displayName: Team
    required: false
  - id: app
    type: STRING
    displayName: Application
    required: false
  - id: environment
    type: SELECT
    displayName: Environment
    values:
      - dev
      - staging
      - prod
    defaults: dev
  - id: tier
    type: SELECT
    displayName: Size tier
    values:
      - small
      - medium
      - large
    defaults: small
  - id: owner_group
    type: STRING
    displayName: Owner group
    description: Identity provider group that gets the edit role in the namespace.
    required: false
  - id: ttl_days
    type: INT
    displayName: Days until expiry
    description: Ignored for prod, where namespaces don't expire.
    defaults: 30
  - id: reason
    type: STRING
    displayName: Reason
    required: false
  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Validate and render the manifests without applying anything.
    defaults: false

variables:
  # Size tiers: requests and limits for the whole namespace, plus per-container defaults.
  tiers:
    small:
      {
        cpu: "2",
        memory: 4Gi,
        limits_cpu: "4",
        limits_memory: 8Gi,
        pods: 20,
        pvcs: 5,
        default_cpu: 250m,
        default_memory: 256Mi
      }
    medium:
      {
        cpu: "8",
        memory: 16Gi,
        limits_cpu: "16",
        limits_memory: 32Gi,
        pods: 60,
        pvcs: 15,
        default_cpu: 500m,
        default_memory: 512Mi
      }
    large:
      {
        cpu: "32",
        memory: 64Gi,
        limits_cpu: "64",
        limits_memory: 128Gi,
        pods: 200,
        pvcs: 40,
        default_cpu: "1",
        default_memory: 1Gi
      }
  # Tiers approved without a person, per environment. Everything else waits for approval.
  auto_approve:
    dev: [ small, medium ]
    staging: [ small ]
    prod: []
  # Longest lifetime per environment; 0 means no expiry.
  max_ttl_days: { dev: 30, staging: 90, prod: 0 }
  pod_security: { dev: baseline, staging: baseline, prod: restricted }
  owner_role: edit
  reminder_days: 3

triggers:
  - id: portal_request
    type: io.kestra.plugin.core.trigger.Webhook
    description: Your developer portal, Backstage or Jira automation posts a JSON
      request with team, app, environment, tier, owner_group, ttl_days, reason
      and requested_by.
    key: "{{ secret('NAMESPACE_REQUEST_WEBHOOK_KEY') }}"
    inputs:
      mode: request

  - id: daily_lifecycle
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 8 * * *"
    inputs:
      mode: sweep

tasks:
  - id: by_mode
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ inputs.mode ?? 'request' }}"
    cases:
      request:
        - id: existing
          type: io.kestra.plugin.scripts.shell.Commands
          description: List namespaces with kubectl, to refuse a name that already exists
            and to make a repeated request a no-op. Namespaces are
            cluster-scoped, which the Kubernetes plugin's kubectl tasks don't
            read.
          taskRunner:
            type: io.kestra.plugin.scripts.runner.docker.Docker
            user: "0"
          containerImage: bitnami/kubectl
          env:
            K8S_SERVER: "{{ secret('K8S_MASTER_URL') }}"
            K8S_TOKEN: "{{ secret('K8S_TOKEN') }}"
            K8S_CA_CERT: "{{ secret('K8S_CA_CERT') }}"
          commands:
            - echo "$K8S_CA_CERT" | base64 -d > ca.crt
            - kubectl --server="$K8S_SERVER" --token="$K8S_TOKEN"
              --certificate-authority=ca.crt get namespaces -o json >
              namespaces.json
          outputFiles:
            - namespaces.json
          retry:
            type: exponential
            interval: PT10S
            maxInterval: PT1M
            maxAttempts: 3

        - id: validate
          type: io.kestra.plugin.scripts.python.Script
          description: Check the request against the naming, tier and lifetime policy, and
            decide whether a person must approve it.
          dependencies:
            - kestra
          inputFiles:
            request.json: |
              {"team": {{ (trigger.body.team ?? inputs.team ?? '') | toJson }},
               "app": {{ (trigger.body.app ?? inputs.app ?? '') | toJson }},
               "environment": {{ (trigger.body.environment ?? inputs.environment ?? 'dev') | toJson }},
               "tier": {{ (trigger.body.tier ?? inputs.tier ?? 'small') | toJson }},
               "owner_group": {{ (trigger.body.owner_group ?? inputs.owner_group ?? '') | toJson }},
               "ttl_days": {{ (trigger.body.ttl_days ?? inputs.ttl_days ?? 30) | toJson }},
               "reason": {{ (trigger.body.reason ?? inputs.reason ?? '') | toJson }},
               "requested_by": {{ (trigger.body.requested_by ?? 'Kestra UI') | toJson }}}
            policy.json: |
              {"tiers": {{ vars.tiers | toJson }}, "auto_approve": {{ vars.auto_approve | toJson }},
               "max_ttl_days": {{ vars.max_ttl_days | toJson }}}
            existing.json: "{{ outputs.existing.outputFiles['namespaces.json'] }}"
          script: |
            import json, re
            from kestra import Kestra

            r = json.load(open("request.json"))
            policy = json.load(open("policy.json"))
            problems = []
            label = re.compile(r"^[a-z0-9]([a-z0-9-]*[a-z0-9])?$")
            for field in ("team", "app", "owner_group"):
                r[field] = str(r[field]).strip().lower()
                if not r[field]:
                    problems.append(f"{field} is required")
                elif not label.fullmatch(r[field]) or len(r[field]) > 40:
                    problems.append(f"{field} '{r[field]}' must be lowercase letters, digits and dashes, at most 40 characters")
            if r["environment"] not in policy["max_ttl_days"]:
                problems.append(f"environment must be one of {', '.join(policy['max_ttl_days'])}")
            if r["tier"] not in policy["tiers"]:
                problems.append(f"tier must be one of {', '.join(policy['tiers'])}")
            name = f"{r['team']}-{r['app']}-{r['environment']}"
            if len(name) > 63:
                problems.append(f"namespace name {name} is longer than 63 characters")
            if r["team"].startswith("kube") or name in ("default", "kube-system", "kube-public", "kube-node-lease"):
                problems.append(f"{name} is a reserved name")
            if problems:
                raise SystemExit("Request refused: " + "; ".join(problems) + ".")

            max_ttl = int(policy["max_ttl_days"][r["environment"]])
            ttl = int(r["ttl_days"] or 0)
            reasons = []
            if r["tier"] not in policy["auto_approve"].get(r["environment"], []):
                reasons.append(f"tier {r['tier']} in {r['environment']} needs approval")
            if max_ttl and (ttl <= 0 or ttl > max_ttl):
                reasons.append(f"{ttl} days is outside the {max_ttl}-day limit for {r['environment']}")
            ttl = 0 if not max_ttl else ttl

            current = next((i["metadata"] for i in json.load(open("existing.json"))["items"] if i["metadata"]["name"] == name), None)
            state = "new"
            if current:
                labels = current.get("labels") or {}
                if labels.get("app.kubernetes.io/managed-by") != "kestra":
                    raise SystemExit(f"Namespace {name} already exists and was not created by this flow. Pick another app name or ask the platform team.")
                same = labels.get("platform.kestra.io/tier") == r["tier"] and labels.get("platform.kestra.io/owner-group") == r["owner_group"]
                state = "exists" if same else "conflict"
                if state == "conflict":
                    raise SystemExit(f"Namespace {name} already exists with a different tier or owner. Change it with the platform team instead of requesting it again.")
            print(f"{name}: {state}; " + ("approval needed: " + "; ".join(reasons) if reasons else "auto-approved by policy"))
            Kestra.outputs({"name": name, "request": r, "ttl_days": ttl, "state": state,
                            "needs_approval": bool(reasons), "approval_reasons": reasons})

        - id: already_there
          type: io.kestra.plugin.core.flow.If
          description: A repeated request for a namespace this flow already created ends
            here without changes.
          condition: "{{ outputs.validate.vars.state == 'exists' }}"
          then:
            - id: no_change
              type: io.kestra.plugin.core.execution.Exit
              state: SUCCESS

        - id: approval
          type: io.kestra.plugin.core.flow.If
          condition: "{{ outputs.validate.vars.needs_approval and not (inputs.dry_run ??
            false) }}"
          then:
            - id: ask_platform_team
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              retry:
                type: constant
                interval: PT10S
                maxAttempts: 3
              messageText: |
                🙋 *Namespace request:* `{{ outputs.validate.vars.name }}` ({{ outputs.validate.vars.request.tier }}) for `{{ outputs.validate.vars.request.owner_group }}`, requested by {{ outputs.validate.vars.request.requested_by }}.
                Needs approval: {{ outputs.validate.vars.approval_reasons | join('; ') }}.{% if outputs.validate.vars.request.reason != '' %} Reason: {{ outputs.validate.vars.request.reason }}{% endif %}
                {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}

            - id: approve
              type: io.kestra.plugin.core.flow.Pause
              description: The platform team approves, rejects, or adjusts the tier and
                lifetime. Without an answer in 3 days nothing is created.
              pauseDuration: P3D
              behavior: CANCEL
              onResume:
                - id: decision
                  type: SELECT
                  displayName: Decision
                  values:
                    - Approve
                    - Reject
                - id: tier
                  type: SELECT
                  displayName: Tier to grant (leave empty to keep the requested tier)
                  values:
                    - small
                    - medium
                    - large
                  required: false
                - id: ttl_days
                  type: INT
                  displayName: Days until expiry (leave empty to keep the requested value)
                  required: false
                - id: note
                  type: STRING
                  displayName: Note for the requester
                  required: false

            - id: rejected
              type: io.kestra.plugin.core.flow.If
              condition: "{{ outputs.approve.onResume.decision == 'Reject' }}"
              then:
                - id: tell_rejected
                  type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                  url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                  retry:
                    type: constant
                    interval: PT10S
                    maxAttempts: 3
                  messageText: "❌ Namespace request `{{ outputs.validate.vars.name }}` was
                    rejected. {{ outputs.approve.onResume.note ?? '' }}"
                - id: stop_rejected
                  type: io.kestra.plugin.core.execution.Exit
                  state: SUCCESS

        - id: render
          type: io.kestra.plugin.scripts.python.Script
          description: Render the golden-path manifests. Namespace-scoped objects are
            rendered separately, since they are applied inside the new
            namespace.
          dependencies:
            - kestra
          inputFiles:
            plan.json: |
              {"name": {{ outputs.validate.vars.name | toJson }}, "request": {{ outputs.validate.vars.request | toJson }},
               "ttl_days": {{ outputs.validate.vars.ttl_days }}, "tiers": {{ vars.tiers | toJson }},
               "pod_security": {{ vars.pod_security | toJson }}, "owner_role": {{ vars.owner_role | toJson }},
               "override": {{ (outputs.approve.onResume ?? {}) | toJson }}, "execution": {{ execution.id | toJson }}}
          outputFiles:
            - namespace.yaml
            - policies.yaml
          script: |
            import json
            from datetime import date, timedelta
            from kestra import Kestra

            p = json.load(open("plan.json"))
            r, name, o = p["request"], p["name"], p["override"]
            tier = o.get("tier") or r["tier"]
            ttl = int(o["ttl_days"]) if o.get("ttl_days") not in (None, "") else int(p["ttl_days"])
            if r["environment"] == "prod":
                ttl = 0
            t = p["tiers"][tier]
            expires = (date.today() + timedelta(days=ttl)).isoformat() if ttl else ""
            approver = "platform team" if o else "policy (auto-approved)"
            env = r["environment"]

            namespace = {"apiVersion": "v1", "kind": "Namespace", "metadata": {
                "name": name,
                "labels": {"app.kubernetes.io/managed-by": "kestra", "platform.kestra.io/team": r["team"],
                           "platform.kestra.io/environment": env, "platform.kestra.io/tier": tier,
                           "platform.kestra.io/owner-group": r["owner_group"],
                           "pod-security.kubernetes.io/enforce": p["pod_security"][env],
                           "pod-security.kubernetes.io/enforce-version": "latest"},
                "annotations": {"platform.kestra.io/expires-at": expires or "never",
                                "platform.kestra.io/requested-by": r["requested_by"],
                                "platform.kestra.io/approved-by": approver, "platform.kestra.io/reason": r["reason"],
                                "platform.kestra.io/request-execution": p["execution"]}}}
            limits = dict(cpu=str(t["default_cpu"]), memory=str(t["default_memory"]))
            policies = [
                {"apiVersion": "v1", "kind": "ResourceQuota", "metadata": dict(name=f"tier-{tier}"),
                 "spec": {"hard": {"requests.cpu": str(t["cpu"]), "requests.memory": str(t["memory"]),
                                   "limits.cpu": str(t["limits_cpu"]), "limits.memory": str(t["limits_memory"]),
                                   "pods": str(t["pods"]), "persistentvolumeclaims": str(t["pvcs"])}}},
                {"apiVersion": "v1", "kind": "LimitRange", "metadata": dict(name="container-defaults"),
                 "spec": {"limits": [dict(type="Container", default=limits, defaultRequest=limits)]}},
                {"apiVersion": "rbac.authorization.k8s.io/v1", "kind": "RoleBinding", "metadata": dict(name=f"owners-{p['owner_role']}"),
                 "roleRef": dict(apiGroup="rbac.authorization.k8s.io", kind="ClusterRole", name=p["owner_role"]),
                 "subjects": [dict(apiGroup="rbac.authorization.k8s.io", kind="Group", name=r["owner_group"])]},
                {"apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicy", "metadata": dict(name="default-deny-ingress"),
                 "spec": dict(podSelector=dict(), policyTypes=["Ingress"], ingress=[dict(**{"from": [dict(podSelector=dict())]})])},
            ]
            # JSON is valid YAML, so the manifests need no YAML library.
            open("namespace.yaml", "w").write(json.dumps(namespace, indent=2) + "\n")
            open("policies.yaml", "w").write("\n---\n".join(json.dumps(d, indent=2) for d in policies) + "\n")
            Kestra.outputs({"tier": tier, "ttl_days": ttl, "expires": expires, "approver": approver,
                            "expected": {"quota": f"tier-{tier}", "pods": str(t["pods"]), "cpu": str(t["cpu"]),
                                         "rolebinding": f"owners-{p['owner_role']}", "group": r["owner_group"]}})

        - id: dry_run_stop
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.dry_run ?? false }}"
          then:
            - id: show_plan
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              retry:
                type: constant
                interval: PT10S
                maxAttempts: 3
              messageText: "🧪 Dry run: would create `{{ outputs.validate.vars.name }}` ({{
                outputs.render.vars.tier }}, expires {{
                outputs.render.vars.expires != '' ? outputs.render.vars.expires
                : 'never' }}, approval {{ outputs.validate.vars.needs_approval ?
                'required' : 'not required' }}) with a ResourceQuota,
                LimitRange, RoleBinding for `{{
                outputs.validate.vars.request.owner_group }}` and a default-deny
                NetworkPolicy. Manifests are in the execution outputs."
            - id: stop_dry_run
              type: io.kestra.plugin.core.execution.Exit
              state: SUCCESS

        - id: apply_namespace
          type: io.kestra.plugin.scripts.shell.Commands
          description: Create the namespace with its labels and annotations (server-side
            apply, so a re-run doesn't fail).
          taskRunner:
            type: io.kestra.plugin.scripts.runner.docker.Docker
            user: "0"
          containerImage: bitnami/kubectl
          env:
            K8S_SERVER: "{{ secret('K8S_MASTER_URL') }}"
            K8S_TOKEN: "{{ secret('K8S_TOKEN') }}"
            K8S_CA_CERT: "{{ secret('K8S_CA_CERT') }}"
          inputFiles:
            namespace.yaml: "{{ outputs.render.outputFiles['namespace.yaml'] }}"
          commands:
            - echo "$K8S_CA_CERT" | base64 -d > ca.crt
            - kubectl --server="$K8S_SERVER" --token="$K8S_TOKEN"
              --certificate-authority=ca.crt apply --server-side
              --field-manager=kestra -f namespace.yaml
          retry:
            type: exponential
            interval: PT10S
            maxInterval: PT1M
            maxAttempts: 3

        - id: apply_policies
          type: io.kestra.plugin.scripts.shell.Commands
          description: Quota, container defaults, the owner group's RoleBinding and the
            default-deny NetworkPolicy, applied inside the new namespace in one
            kubectl apply.
          taskRunner:
            type: io.kestra.plugin.scripts.runner.docker.Docker
            user: "0"
          containerImage: bitnami/kubectl
          env:
            K8S_SERVER: "{{ secret('K8S_MASTER_URL') }}"
            K8S_TOKEN: "{{ secret('K8S_TOKEN') }}"
            K8S_CA_CERT: "{{ secret('K8S_CA_CERT') }}"
          inputFiles:
            policies.yaml: "{{ outputs.render.outputFiles['policies.yaml'] }}"
          commands:
            - echo "$K8S_CA_CERT" | base64 -d > ca.crt
            - kubectl --server="$K8S_SERVER" --token="$K8S_TOKEN"
              --certificate-authority=ca.crt apply --server-side
              --field-manager=kestra -n "{{ outputs.validate.vars.name }}" -f
              policies.yaml
          retry:
            type: exponential
            interval: PT10S
            maxInterval: PT1M
            maxAttempts: 3

        - id: read_back
          type: io.kestra.plugin.core.flow.Parallel
          description: Read the objects back from the cluster, so the namespace is only
            reported ready if it really has its guardrails.
          tasks:
            - id: get_quota
              type: io.kestra.plugin.kubernetes.kubectl.Get
              connection:
                masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                oauthToken: "{{ secret('K8S_TOKEN') }}"
                caCertData: "{{ secret('K8S_CA_CERT') }}"
              namespace: "{{ outputs.validate.vars.name }}"
              resourceType: resourcequotas
              fetchType: FETCH
              retry:
                type: exponential
                interval: PT10S
                maxInterval: PT1M
                maxAttempts: 3
            - id: get_netpol
              type: io.kestra.plugin.kubernetes.kubectl.Get
              connection:
                masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                oauthToken: "{{ secret('K8S_TOKEN') }}"
                caCertData: "{{ secret('K8S_CA_CERT') }}"
              namespace: "{{ outputs.validate.vars.name }}"
              resourceType: networkpolicies
              apiGroup: networking.k8s.io
              apiVersion: v1
              fetchType: FETCH
              retry:
                type: exponential
                interval: PT10S
                maxInterval: PT1M
                maxAttempts: 3
            - id: get_rolebinding
              type: io.kestra.plugin.kubernetes.kubectl.Get
              connection:
                masterUrl: "{{ secret('K8S_MASTER_URL') }}"
                oauthToken: "{{ secret('K8S_TOKEN') }}"
                caCertData: "{{ secret('K8S_CA_CERT') }}"
              namespace: "{{ outputs.validate.vars.name }}"
              resourceType: rolebindings
              apiGroup: rbac.authorization.k8s.io
              apiVersion: v1
              fetchType: FETCH
              retry:
                type: exponential
                interval: PT10S
                maxInterval: PT1M
                maxAttempts: 3

        - id: verify
          type: io.kestra.plugin.scripts.python.Script
          description: Compare what the cluster holds with what was requested.
          dependencies:
            - kestra
          inputFiles:
            found.json: |
              {"quotas": {{ outputs.get_quota.metadataItems | toJson }}, "quota_status": {{ (outputs.get_quota.statusItems ?? []) | toJson }},
               "netpols": {{ outputs.get_netpol.metadataItems | toJson }}, "rolebindings": {{ outputs.get_rolebinding.metadataItems | toJson }},
               "expected": {{ outputs.render.vars.expected | toJson }}}
          script: |
            import json
            from kestra import Kestra

            f = json.load(open("found.json"))
            e = f["expected"]
            names = lambda items: {i.get("name") for i in items}
            problems = []
            if e["quota"] not in names(f["quotas"]):
                problems.append(f"ResourceQuota {e['quota']} is missing")
            else:
                hard = next((s.get("hard") or {} for s in f["quota_status"] if s), {})
                if hard and (str(hard.get("pods")) != e["pods"] or str(hard.get("requests.cpu")) != e["cpu"]):
                    problems.append(f"ResourceQuota limits differ from the tier: pods={hard.get('pods')}, requests.cpu={hard.get('requests.cpu')}")
            if "default-deny-ingress" not in names(f["netpols"]):
                problems.append("NetworkPolicy default-deny-ingress is missing")
            if e["rolebinding"] not in names(f["rolebindings"]):
                problems.append(f"RoleBinding {e['rolebinding']} is missing")
            print("Verified" if not problems else "Verification failed: " + "; ".join(problems))
            Kestra.outputs({"ok": not problems, "problems": problems})

        - id: rollback_if_unverified
          type: io.kestra.plugin.core.flow.If
          description: Never leave a namespace without its guardrails. Delete the
            namespace this run created and fail loudly.
          condition: "{{ not outputs.verify.vars.ok }}"
          then:
            - id: delete_unverified
              type: io.kestra.plugin.scripts.shell.Commands
              description: Delete the namespace this run created.
              taskRunner:
                type: io.kestra.plugin.scripts.runner.docker.Docker
                user: "0"
              containerImage: bitnami/kubectl
              env:
                K8S_SERVER: "{{ secret('K8S_MASTER_URL') }}"
                K8S_TOKEN: "{{ secret('K8S_TOKEN') }}"
                K8S_CA_CERT: "{{ secret('K8S_CA_CERT') }}"
              commands:
                - echo "$K8S_CA_CERT" | base64 -d > ca.crt
                - kubectl --server="$K8S_SERVER" --token="$K8S_TOKEN"
                  --certificate-authority=ca.crt delete namespace "{{
                  outputs.validate.vars.name }}" --ignore-not-found --wait=false
              retry:
                type: exponential
                interval: PT10S
                maxInterval: PT1M
                maxAttempts: 3

            - id: fail_unverified
              type: io.kestra.plugin.core.execution.Fail
              errorMessage: "Namespace {{ outputs.validate.vars.name }} did not verify and was
                deleted: {{ outputs.verify.vars.problems | join('; ') }}"

        - id: register
          type: io.kestra.plugin.core.kv.Set
          description: Owner, tier and expiry, used by the daily lifecycle sweep.
          key: "{{ 'namespace-' ~ outputs.validate.vars.name }}"
          kvType: JSON
          value: |
            {"name": {{ outputs.validate.vars.name | toJson }}, "team": {{ outputs.validate.vars.request.team | toJson }},
             "environment": {{ outputs.validate.vars.request.environment | toJson }}, "tier": {{ outputs.render.vars.tier | toJson }},
             "owner_group": {{ outputs.validate.vars.request.owner_group | toJson }}, "requested_by": {{ outputs.validate.vars.request.requested_by | toJson }},
             "approved_by": {{ outputs.render.vars.approver | toJson }}, "expires": {{ outputs.render.vars.expires | toJson }},
             "created": "{{ now() | date('yyyy-MM-dd') }}", "execution": {{ execution.id | toJson }}}

        - id: tell_ready
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          retry:
            type: constant
            interval: PT10S
            maxAttempts: 3
          messageText: |
            ✅ Namespace `{{ outputs.validate.vars.name }}` is ready. Group `{{ outputs.validate.vars.request.owner_group }}` has the {{ vars.owner_role }} role in it.
            Tier {{ outputs.render.vars.tier }}: {{ vars.tiers[outputs.render.vars.tier].cpu }} CPU and {{ vars.tiers[outputs.render.vars.tier].memory }} requested, {{ vars.tiers[outputs.render.vars.tier].pods }} pods. Ingress from other namespaces is denied by default. Expires: {{ outputs.render.vars.expires != '' ? outputs.render.vars.expires : 'never' }}. Approved by {{ outputs.render.vars.approver }}.{% if (outputs.approve.onResume.note ?? '') != '' %} Note: {{ outputs.approve.onResume.note }}{% endif %}

      sweep:
        - id: registry_keys
          type: io.kestra.plugin.core.kv.GetKeys
          prefix: namespace-

        - id: cluster_namespaces
          type: io.kestra.plugin.scripts.shell.Commands
          description: List namespaces with kubectl to compare the cluster with the
            registry.
          taskRunner:
            type: io.kestra.plugin.scripts.runner.docker.Docker
            user: "0"
          containerImage: bitnami/kubectl
          env:
            K8S_SERVER: "{{ secret('K8S_MASTER_URL') }}"
            K8S_TOKEN: "{{ secret('K8S_TOKEN') }}"
            K8S_CA_CERT: "{{ secret('K8S_CA_CERT') }}"
          commands:
            - echo "$K8S_CA_CERT" | base64 -d > ca.crt
            - kubectl --server="$K8S_SERVER" --token="$K8S_TOKEN"
              --certificate-authority=ca.crt get namespaces -o json >
              namespaces.json
          outputFiles:
            - namespaces.json
          retry:
            type: exponential
            interval: PT10S
            maxInterval: PT1M
            maxAttempts: 3

        - id: lifecycle
          type: io.kestra.plugin.scripts.python.Script
          description: Find namespaces that expire soon or have expired, and registry
            entries or managed namespaces that no longer match each other.
          dependencies:
            - kestra
          inputFiles:
            registry.json: "[{% for k in outputs.registry_keys.keys %}{{ kv(k) | toJson }},
              {% endfor %}null]"
            cluster.json: "{{ outputs.cluster_namespaces.outputFiles['namespaces.json'] }}"
            settings.json: '{"reminder_days": {{ vars.reminder_days }}}'
          script: |
            import json
            from datetime import date
            from kestra import Kestra

            registry = {r["name"]: r for r in json.load(open("registry.json")) if r}
            cluster = {i["metadata"]["name"]: i["metadata"] for i in json.load(open("cluster.json"))["items"]}
            managed = {n for n, m in cluster.items() if (m.get("labels") or {}).get("app.kubernetes.io/managed-by") == "kestra"}
            today, soon = date.today(), json.load(open("settings.json"))["reminder_days"]
            expired, expiring, lines = [], [], []
            for name, r in sorted(registry.items()):
                if name not in cluster:
                    lines.append(f"• `{name}` is registered but no longer in the cluster; its record will be removed.")
                    expired.append({"name": name, "gone": True})
                    continue
                if not r.get("expires"):
                    continue
                left = (date.fromisoformat(r["expires"]) - today).days
                if left < 0:
                    expired.append({"name": name, "gone": False})
                    lines.append(f"• `{name}` ({r['owner_group']}) expired on {r['expires']}.")
                elif left <= soon:
                    expiring.append(name)
                    lines.append(f"• `{name}` ({r['owner_group']}) expires in {left} day(s), on {r['expires']}.")
            for name in sorted(managed - set(registry)):
                lines.append(f"• `{name}` is labelled as managed by Kestra but has no registry record; check it by hand.")
            Kestra.outputs({"digest": "\n".join(lines), "expired": [e["name"] for e in expired if not e["gone"]],
                            "gone": [e["name"] for e in expired if e["gone"]], "expiring": expiring})

        - id: post_lifecycle
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          runIf: "{{ outputs.lifecycle.vars.digest != '' }}"
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          retry:
            type: constant
            interval: PT10S
            maxAttempts: 3
          messageText: |
            🗓️ *Namespace lifecycle*
            {{ outputs.lifecycle.vars.digest }}
            {% if (outputs.lifecycle.vars.expired | length) > 0 %}Decide on the expired ones: {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}{% endif %}

        - id: forget_gone
          type: io.kestra.plugin.core.flow.Loop
          description: Remove registry records of namespaces that were deleted outside
            this flow.
          values: "{{ outputs.lifecycle.vars.gone }}"
          tasks:
            - id: delete_record
              type: io.kestra.plugin.core.kv.Delete
              key: "{{ 'namespace-' ~ item.value }}"
              errorOnMissing: false

        - id: teardown
          type: io.kestra.plugin.core.flow.If
          condition: "{{ (outputs.lifecycle.vars.expired | length) > 0 }}"
          then:
            - id: decide_expired
              type: io.kestra.plugin.core.flow.Pause
              description: Pick expired namespaces to delete and ones to extend. Without an
                answer in 2 days nothing is deleted and they are listed again
                tomorrow.
              pauseDuration: P2D
              behavior: CANCEL
              onResume:
                - id: delete
                  type: MULTISELECT
                  displayName: Delete these namespaces (everything inside them is removed)
                  expression: "{{ outputs.lifecycle.vars.expired }}"
                  required: false
                - id: extend
                  type: MULTISELECT
                  displayName: Extend these namespaces
                  expression: "{{ outputs.lifecycle.vars.expired }}"
                  required: false
                - id: extend_days
                  type: INT
                  displayName: Extend by this many days
                  defaults: 14

            - id: plan_teardown
              type: io.kestra.plugin.scripts.python.Script
              dependencies:
                - kestra
              inputFiles:
                answer.json: "{{ outputs.decide_expired.onResume | toJson }}"
                registry.json: "[{% for k in outputs.registry_keys.keys %}{{ kv(k) | toJson }},
                  {% endfor %}null]"
              script: |
                import json
                from datetime import date, timedelta
                from kestra import Kestra

                a = json.load(open("answer.json"))
                registry = {r["name"]: r for r in json.load(open("registry.json")) if r}
                delete = list(a.get("delete") or [])
                extend = [n for n in a.get("extend") or [] if n not in delete]
                until = (date.today() + timedelta(days=int(a.get("extend_days") or 14))).isoformat()
                updates = {n: dict(registry[n], expires=until) for n in extend}
                Kestra.outputs({"delete": delete, "extend": extend, "updates": updates, "until": until})

            - id: delete_expired
              type: io.kestra.plugin.core.flow.Loop
              values: "{{ outputs.plan_teardown.vars.delete }}"
              concurrencyLimit: 1
              tasks:
                - id: delete_namespace
                  type: io.kestra.plugin.scripts.shell.Commands
                  description: Delete the expired namespace and everything in it.
                  taskRunner:
                    type: io.kestra.plugin.scripts.runner.docker.Docker
                    user: "0"
                  containerImage: bitnami/kubectl
                  env:
                    K8S_SERVER: "{{ secret('K8S_MASTER_URL') }}"
                    K8S_TOKEN: "{{ secret('K8S_TOKEN') }}"
                    K8S_CA_CERT: "{{ secret('K8S_CA_CERT') }}"
                  commands:
                    - echo "$K8S_CA_CERT" | base64 -d > ca.crt
                    - kubectl --server="$K8S_SERVER" --token="$K8S_TOKEN"
                      --certificate-authority=ca.crt delete namespace "{{
                      item.value }}" --ignore-not-found --wait=false
                  retry:
                    type: exponential
                    interval: PT10S
                    maxInterval: PT1M
                    maxAttempts: 3

                - id: forget_namespace
                  type: io.kestra.plugin.core.kv.Delete
                  key: "{{ 'namespace-' ~ item.value }}"
                  errorOnMissing: false

            - id: extend_expired
              type: io.kestra.plugin.core.flow.Loop
              values: "{{ outputs.plan_teardown.vars.extend }}"
              tasks:
                - id: extend_record
                  type: io.kestra.plugin.core.kv.Set
                  key: "{{ 'namespace-' ~ item.value }}"
                  kvType: JSON
                  value: "{{ outputs.plan_teardown.vars.updates[item.value] | toJson }}"

            - id: report_teardown
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              retry:
                type: constant
                interval: PT10S
                maxAttempts: 3
              messageText: "🧹 Deleted: {{ (outputs.plan_teardown.vars.delete | length) > 0 ?
                outputs.plan_teardown.vars.delete | join(', ') : 'none' }}.
                Extended until {{ outputs.plan_teardown.vars.until }}: {{
                (outputs.plan_teardown.vars.extend | length) > 0 ?
                outputs.plan_teardown.vars.extend | join(', ') : 'none' }}."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert with the failed task, the error and a link.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3
    messageText: |
      ❌ *Namespace self-service FAILED* ({{ inputs.mode ?? 'request' }}; flow {{ flow.namespace }}.{{ flow.id }}, execution {{ execution.id }}).
      Failed task: {{ (errorLogs() | length) > 0 ? errorLogs()[0].taskId : 'unknown' }}. Error: {{ ((errorLogs() | length) > 0 and (errorLogs()[0].message ?? 'null') != 'null') ? errorLogs()[0].message : 'see the execution logs' }}
      {{ kestra.url ?? '' }}/ui/main/executions/{{ flow.namespace }}/{{ flow.id }}/{{ execution.id }}

outputs:
  - id: namespace
    type: STRING
    displayName: Namespace
    value: "{{ outputs.validate.vars.name ?? '' }}"
  - id: expires
    type: STRING
    displayName: Expiry date (empty for no expiry)
    value: "{{ outputs.render.vars.expires ?? '' }}"