Clean orphaned Spark output staging safely

id: spark-orphaned-staging-output-cleanup
namespace: company.team

description: Remove abandoned Spark output staging trees only after their age
  and cluster liveness checks pass.

concurrency:
  limit: 1
  behavior: QUEUE

inputs:
  - id: output_path
    type: STRING
    defaults: "s3a://sstage/output"
    description: One Spark output root. The flow never deletes outside this URI.
  - id: s3_endpoint
    type: STRING
    defaults: "{{ secret('SPARK_S3_ENDPOINT') }}"
    description: S3A endpoint reachable from the SparkCLI task runner.
  - id: s3_access_key
    type: STRING
    defaults: "{{ secret('SPARK_S3_ACCESS_KEY') }}"
    description: S3 access key with list and delete permission on output_path.
  - id: s3_secret_key
    type: STRING
    defaults: "{{ secret('SPARK_S3_SECRET_KEY') }}"
    description: S3 secret key with list and delete permission on output_path.
  - id: spark_master_web_url
    type: STRING
    defaults: "{{ secret('SPARK_MASTER_WEB_URL') }}"
    description: Spark standalone master JSON endpoint used to spare paths
      containing a live application id.
  - id: task_network
    type: STRING
    defaults: "bridge"
    description: Docker network for SparkCLI task runners.
  - id: max_job_runtime_seconds
    type: INT
    defaults: 3600
    min: 0
    max: 604800
    description: Longest permitted Spark job runtime. A staging tree is eligible
      only when older than this value.
  - id: auto_remediate
    type: BOOL
    defaults: false
    description: Permit deletion after approval. Keep false for an audit-only run.
  - id: dry_run
    type: BOOL
    defaults: true
    description: Keep true to prevent all deletes.

variables:
  audit_script: |
    import datetime
    import hashlib
    import json
    import os
    import re
    import subprocess
    import sys

    tree = open(sys.argv[1], encoding="utf-8").read().splitlines()
    output = os.environ["OUTPUT_PATH"].rstrip("/")
    cutoff = int(os.environ["MAX_JOB_RUNTIME_SECONDS"])
    live = set()
    try:
        raw = subprocess.check_output(["curl", "-fsS", os.environ["SPARK_MASTER_WEB_URL"] + "/json"], text=True)
        payload = json.loads(raw)
        for key in ("activeapps", "activeApps"):
            for app in payload.get(key, []) or []:
                if app.get("id"):
                    live.add(app["id"])
    except Exception as exc:
        raise SystemExit("cannot read the Spark master, so live applications are unknown: %s" % exc)

    staging = {}
    committed = []
    line_re = re.compile(r"(\d{4}-\d{2}-\d{2})\s+(\d{2}:\d{2})")
    for line in tree:
        if not line or line.startswith("Found "):
            continue
        fields = line.split()
        if len(fields) < 8 or not (fields[0].startswith("d") or fields[0].startswith("-") or fields[0].startswith("l")):
            continue
        path = fields[-1]
        if not path.startswith(output + "/") and path != output:
            continue
        match = line_re.search(line)
        if not match:
            continue
        try:
            mtime = datetime.datetime.strptime(match.group(1) + " " + match.group(2), "%Y-%m-%d %H:%M").replace(tzinfo=datetime.timezone.utc).timestamp()
        except ValueError:
            continue
        try:
            size = int(fields[4])
        except (ValueError, IndexError):
            size = 0
        relative = path[len(output) + 1:] if path != output else ""
        parts = relative.split("/") if relative else []
        marker = next((i for i, part in enumerate(parts) if part in ("_temporary", "_SPARK_STAGING")), None)
        if marker is None:
            committed.append("%s|%s" % (path, size))
            continue
        end = min(len(parts), marker + 2)
        root = output + "/" + "/".join(parts[:end])
        entry = staging.setdefault(root, {"mtime": 0, "bytes": 0})
        entry["mtime"] = max(entry["mtime"], mtime)
        if not (fields[0].startswith("d") or fields[0].startswith("l")):
            entry["bytes"] += size

    now = datetime.datetime.now(datetime.timezone.utc).timestamp()
    candidates = []
    for path, item in sorted(staging.items()):
        age = int(max(0, now - item["mtime"]))
        live_hit = any(app_id in path for app_id in live)
        if age >= cutoff and not live_hit:
            candidates.append({"path": path, "age_seconds": age, "bytes": item["bytes"]})
    committed.sort()
    digest = hashlib.sha256("\\n".join(committed).encode()).hexdigest()
    print(json.dumps({
        "candidate_count": len(candidates),
        "candidates": candidates,
        "candidate_paths": [item["path"] for item in candidates],
        "candidates_json": json.dumps(candidates, separators=(",", ":")),
        "staging_count": len(staging),
        "staging_bytes": sum(item["bytes"] for item in staging.values()),
        "candidate_bytes": sum(item["bytes"] for item in candidates),
        "committed_digest": digest,
        "live_application_count": len(live)
    }), flush=True)

  verify_script: |
    import json
    import os
    import subprocess
    import sys
    expected_digest = os.environ["EXPECTED_COMMITTED_DIGEST"]
    expected_candidates = int(os.environ["EXPECTED_CANDIDATE_COUNT"])
    require_clean = os.environ.get("EXPECT_CLEAN") == "1"
    tree = open(sys.argv[1], encoding="utf-8").read().splitlines()
    output = os.environ["OUTPUT_PATH"].rstrip("/")
    staging = set()
    committed = []
    for line in tree:
        fields = line.split()
        if len(fields) < 8 or not (fields[0].startswith("d") or fields[0].startswith("-") or fields[0].startswith("l")):
            continue
        path = fields[-1]
        if not path.startswith(output + "/"):
            continue
        relative = path[len(output) + 1:]
        parts = relative.split("/")
        marker = next((i for i, part in enumerate(parts) if part in ("_temporary", "_SPARK_STAGING")), None)
        if marker is None:
            committed.append("%s|%s" % (path, fields[4]))
        else:
            staging.add(output + "/" + "/".join(parts[:min(len(parts), marker + 2)]))
    import hashlib
    committed.sort()
    digest = hashlib.sha256("\\n".join(committed).encode()).hexdigest()
    planned = set(json.loads(os.environ.get("PLANNED_PATHS", "[]")))
    leftover = [p for p in staging if p in planned and not any(s != p and s.startswith(p + "/") for s in staging)]
    survivors = len(leftover) if require_clean else len(staging)
    committed_unchanged = digest == expected_digest
    if require_clean and (survivors != 0 or not committed_unchanged):
        raise SystemExit("verification failed: survivors=%d committed_unchanged=%s" % (survivors, committed_unchanged))
    if not require_clean and (survivors != expected_candidates or not committed_unchanged):
        raise SystemExit("no-change verification failed: survivors=%d expected=%d committed_unchanged=%s" % (survivors, expected_candidates, committed_unchanged))
    print("::" + json.dumps({"outputs": {"surviving_staging_count": survivors, "committed_unchanged": committed_unchanged, "expected_candidate_count": expected_candidates}}) + "::", flush=True)

triggers:
  - id: daily_cleanup
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 2 * * *"
    disabled: true
    description: Daily review, disabled until an operator enables it after a dry run.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: inspect_before
    type: io.kestra.plugin.spark.SparkCLI
    description: List the complete output tree with hadoop fs and select only old
      staging roots with no live Spark application id.
    containerImage: apache/spark:3.5.1
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: "{{ inputs.task_network }}"
      user: root
      memory:
        memory: 1500MB
    env:
      AWS_ACCESS_KEY_ID: "{{ inputs.s3_access_key }}"
      AWS_SECRET_ACCESS_KEY: "{{ inputs.s3_secret_key }}"
      S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
      OUTPUT_PATH: "{{ inputs.output_path }}"
      MAX_JOB_RUNTIME_SECONDS: "{{ inputs.max_job_runtime_seconds }}"
      SPARK_MASTER_WEB_URL: "{{ inputs.spark_master_web_url }}"
    beforeCommands:
      - "mkdir -p /tmp/hconf && printf '<configuration/>' >
        /tmp/hconf/core-site.xml && export HADOOP_CONF_DIR=/tmp/hconf"
      - "curl -fsSL -o /opt/spark/jars/hadoop-aws-3.3.4.jar
        https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoo\
        p-aws-3.3.4.jar && curl -fsSL -o
        /opt/spark/jars/aws-java-sdk-bundle-1.12.262.jar
        https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.2\
        62/aws-java-sdk-bundle-1.12.262.jar"
      - "printf '%s\\n' '#!/bin/sh' 'if [ \"$1\" = fs ]; then shift; fi' 'exec
        /opt/spark/bin/spark-class org.apache.hadoop.fs.FsShell
        -Dfs.s3a.endpoint=\"$S3_ENDPOINT\" -Dfs.s3a.path.style.access=true
        -Dfs.s3a.connection.ssl.enabled=false \"$@\"' > /tmp/hadoop && chmod +x
        /tmp/hadoop && export PATH=/tmp:$PATH"
    inputFiles:
      audit.py: "{{ vars.audit_script }}"
    outputFiles:
      - audit.json
    commands:
      - "hadoop fs -ls -R \"$OUTPUT_PATH\" > /tmp/tree-before.txt"
      - "python3 audit.py /tmp/tree-before.txt > \"{{ outputFiles['audit.json']
        }}\""

  - id: remediation_gate
    type: io.kestra.plugin.core.flow.If
    description: Require explicit non-dry approval only when the detector found a
      real orphan.
    condition: "{{ inputs.auto_remediate and not inputs.dry_run and
      fromJson(read(outputs.inspect_before.outputFiles['audit.json'])).candidat\
      e_count > 0 }}"
    then:
      - id: hold
        type: io.kestra.plugin.core.flow.Pause
        description: Review the exact staging paths, ages and byte total before deleting
          them.
        onResume:
          - id: approved
            type: BOOL
            defaults: false
            description: Approve removal of the listed staging roots.
      - id: approval_branch
        type: io.kestra.plugin.core.flow.If
        description: Continue to the recheck and delete only when the operator approved
          the pause.
        condition: "{{ outputs.hold.onResume.approved ?? false }}"
        then:
          - id: prewrite_recheck
            type: io.kestra.plugin.spark.SparkCLI
            description: Re-list the tree and Spark master immediately after approval so a
              changed target becomes a failed run.
            containerImage: apache/spark:3.5.1
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              networkMode: "{{ inputs.task_network }}"
              user: root
              memory:
                memory: 1500MB
            env:
              AWS_ACCESS_KEY_ID: "{{ inputs.s3_access_key }}"
              AWS_SECRET_ACCESS_KEY: "{{ inputs.s3_secret_key }}"
              S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
              OUTPUT_PATH: "{{ inputs.output_path }}"
              MAX_JOB_RUNTIME_SECONDS: "{{ inputs.max_job_runtime_seconds }}"
              SPARK_MASTER_WEB_URL: "{{ inputs.spark_master_web_url }}"
            beforeCommands:
              - "mkdir -p /tmp/hconf && printf '<configuration/>' >
                /tmp/hconf/core-site.xml && export HADOOP_CONF_DIR=/tmp/hconf"
              - "curl -fsSL -o /opt/spark/jars/hadoop-aws-3.3.4.jar
                https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3\
                .4/hadoop-aws-3.3.4.jar && curl -fsSL -o
                /opt/spark/jars/aws-java-sdk-bundle-1.12.262.jar
                https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundl\
                e/1.12.262/aws-java-sdk-bundle-1.12.262.jar"
              - "printf '%s\\n' '#!/bin/sh' 'if [ \"$1\" = fs ]; then shift; fi'
                'exec /opt/spark/bin/spark-class org.apache.hadoop.fs.FsShell
                -Dfs.s3a.endpoint=\"$S3_ENDPOINT\"
                -Dfs.s3a.path.style.access=true
                -Dfs.s3a.connection.ssl.enabled=false \"$@\"' > /tmp/hadoop &&
                chmod +x /tmp/hadoop && export PATH=/tmp:$PATH"
            inputFiles:
              audit.py: "{{ vars.audit_script }}"
            outputFiles:
              - audit.json
            commands:
              - "hadoop fs -ls -R \"$OUTPUT_PATH\" > /tmp/tree-recheck.txt"
              - "python3 audit.py /tmp/tree-recheck.txt > \"{{
                outputFiles['audit.json'] }}\""
          - id: stale_plan_guard
            type: io.kestra.plugin.core.flow.If
            description: Refuse the delete when either the candidate set or committed tree
              changed during the approval pause.
            condition: "{{
              fromJson(read(outputs.prewrite_recheck.outputFiles['audit.json'])\
              ).committed_digest !=
              fromJson(read(outputs.inspect_before.outputFiles['audit.json'])).\
              committed_digest or
              (fromJson(read(outputs.prewrite_recheck.outputFiles['audit.json']\
              )).candidate_paths | toJson) !=
              (fromJson(read(outputs.inspect_before.outputFiles['audit.json']))\
              .candidate_paths | toJson) }}"
            then:
              - id: fail_stale_plan
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "The staging plan changed during approval; no delete was
                  attempted."
            else:
              - id: remediate
                type: io.kestra.plugin.spark.SparkCLI
                description: Remove only the rechecked staging roots with SparkCLI hadoop fs -rm
                  -r.
                containerImage: apache/spark:3.5.1
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                  networkMode: "{{ inputs.task_network }}"
                  user: root
                  memory:
                    memory: 1500MB
                env:
                  AWS_ACCESS_KEY_ID: "{{ inputs.s3_access_key }}"
                  AWS_SECRET_ACCESS_KEY: "{{ inputs.s3_secret_key }}"
                  S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
                  OUTPUT_PATH: "{{ inputs.output_path }}"
                  PLANNED_CANDIDATES: "{{
                    fromJson(read(outputs.prewrite_recheck.outputFiles['audit.j\
                    son'])).candidates | toJson }}"
                beforeCommands:
                  - "mkdir -p /tmp/hconf && printf '<configuration/>' >
                    /tmp/hconf/core-site.xml && export
                    HADOOP_CONF_DIR=/tmp/hconf"
                  - "curl -fsSL -o /opt/spark/jars/hadoop-aws-3.3.4.jar
                    https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws\
                    /3.3.4/hadoop-aws-3.3.4.jar && curl -fsSL -o
                    /opt/spark/jars/aws-java-sdk-bundle-1.12.262.jar
                    https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-b\
                    undle/1.12.262/aws-java-sdk-bundle-1.12.262.jar"
                  - "printf '%s\\n' '#!/bin/sh' 'if [ \"$1\" = fs ]; then shift;
                    fi' 'exec /opt/spark/bin/spark-class
                    org.apache.hadoop.fs.FsShell
                    -Dfs.s3a.endpoint=\"$S3_ENDPOINT\"
                    -Dfs.s3a.path.style.access=true
                    -Dfs.s3a.connection.ssl.enabled=false \"$@\"' > /tmp/hadoop
                    && chmod +x /tmp/hadoop && export PATH=/tmp:$PATH"
                commands:
                  - "python3 -c 'import json,os,subprocess;
                    out=os.environ[\"OUTPUT_PATH\"];
                    planned=json.loads(os.environ[\"PLANNED_CANDIDATES\"]);
                    [subprocess.run([\"hadoop\",\"fs\",\"-rm\",\"-r\",\"-f\",x[\
                    \"path\"]],check=True) for x in planned if
                    x[\"path\"].startswith(out+\"/\") and (\"/_temporary/\" in
                    x[\"path\"] or \"/_SPARK_STAGING/\" in x[\"path\"])] ;
                    [subprocess.run([\"hadoop\",\"fs\",\"-rm\",\"-r\",\"-f\",ou\
                    t+\"/\"+m],check=True) for m in
                    (\"_temporary\",\"_SPARK_STAGING\") if not
                    any((out+\"/\"+m+\"/\") in line for line in
                    subprocess.run([\"hadoop\",\"fs\",\"-ls\",\"-R\",out+\"/\"+\
                    m],capture_output=True,text=True).stdout.splitlines())]'"

        else:
          - id: log_declined
            type: io.kestra.plugin.core.log.Log
            message: "Approval was declined; staging trees remain and the verification below
              must show no change."

  - id: verify
    type: io.kestra.plugin.spark.SparkCLI
    description: Re-list the tree and fail if a planned staging path survived or any
      committed path changed.
    containerImage: apache/spark:3.5.1
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: "{{ inputs.task_network }}"
      user: root
      memory:
        memory: 1500MB
    env:
      AWS_ACCESS_KEY_ID: "{{ inputs.s3_access_key }}"
      AWS_SECRET_ACCESS_KEY: "{{ inputs.s3_secret_key }}"
      S3_ENDPOINT: "{{ inputs.s3_endpoint }}"
      OUTPUT_PATH: "{{ inputs.output_path }}"
      EXPECTED_COMMITTED_DIGEST: "{{
        fromJson(read(outputs.inspect_before.outputFiles['audit.json'])).commit\
        ted_digest }}"
      EXPECTED_CANDIDATE_COUNT: "{{
        fromJson(read(outputs.inspect_before.outputFiles['audit.json'])).stagin\
        g_count }}"
      PLANNED_PATHS: "{{
        fromJson(read(outputs.inspect_before.outputFiles['audit.json'])).candid\
        ate_paths | toJson }}"
      EXPECT_CLEAN: "{{ (inputs.auto_remediate and not inputs.dry_run and
        (outputs.hold.onResume.approved ?? false)) ? '1' : '0' }}"
    beforeCommands:
      - "mkdir -p /tmp/hconf && printf '<configuration/>' >
        /tmp/hconf/core-site.xml && export HADOOP_CONF_DIR=/tmp/hconf"
      - "curl -fsSL -o /opt/spark/jars/hadoop-aws-3.3.4.jar
        https://repo1.maven.org/maven2/org/apache/hadoop/hadoop-aws/3.3.4/hadoo\
        p-aws-3.3.4.jar && curl -fsSL -o
        /opt/spark/jars/aws-java-sdk-bundle-1.12.262.jar
        https://repo1.maven.org/maven2/com/amazonaws/aws-java-sdk-bundle/1.12.2\
        62/aws-java-sdk-bundle-1.12.262.jar"
      - "printf '%s\\n' '#!/bin/sh' 'if [ \"$1\" = fs ]; then shift; fi' 'exec
        /opt/spark/bin/spark-class org.apache.hadoop.fs.FsShell
        -Dfs.s3a.endpoint=\"$S3_ENDPOINT\" -Dfs.s3a.path.style.access=true
        -Dfs.s3a.connection.ssl.enabled=false \"$@\"' > /tmp/hadoop && chmod +x
        /tmp/hadoop && export PATH=/tmp:$PATH"
    inputFiles:
      verify.py: "{{ vars.verify_script }}"
    commands:
      - "hadoop fs -ls -R \"$OUTPUT_PATH\" > /tmp/tree-after.txt"
      - "python3 verify.py /tmp/tree-after.txt"

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert the data platform channel when the audit, stale-plan guard or
      verification fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "Spark staging cleanup failed for {{ inputs.output_path }} in execution {{ execution.id }}. No committed path is allowed to change."
      }