Halt a Failing OTA Firmware Rollout at the Canary Cohort

id: firmware-ota-canary
namespace: company.team
description: |
  OTA firmware rollouts that push a bad build everywhere at once can brick a
  fleet before anyone can react. A deploy pipeline POSTs canary cohort
  telemetry to a keyed webhook, Python computes the canary error and reboot
  failure rates against a configurable threshold, and an If decides whether
  to halt. A failing canary alerts Slack and, when auto halt is on, turns the
  execution red before the rollout widens; a healthy canary is logged. The
  rollout decision, the error rates and the cohort counters are published as
  outputs.

triggers:
  - id: on_canary_telemetry
    type: io.kestra.plugin.core.trigger.Webhook
    description: The OTA deploy pipeline POSTs canary cohort telemetry here as soon
      as the canary wave finishes, so the halt decision lands before the rollout
      widens.
    key: "{{ secret('OTA_CANARY_WEBHOOK_KEY') }}"
  - id: daily_rollout_recheck
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily 04:00 UTC fallback net that rescores the staged canary
      telemetry, shipped disabled until an operator switches it on.
    cron: "0 4 * * *"
    disabled: true

inputs:
  - id: rollout_name
    type: STRING
    defaults: fw-2.14.0
    description: Firmware version the canary wave belongs to, used when the posted
      telemetry omits its own rollout identifier.
  - id: error_threshold_pct
    type: FLOAT
    defaults: 2.0
    description: Canary error rate percentage at or above which the rollout decision
      flips to HALT.
  - id: auto_halt
    type: BOOL
    defaults: true
    description: Whether a failing canary turns the execution red after the Slack
      alert, flip to false for observe only runs.

tasks:
  # --- 1. canary workspace ---
  - id: canary_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the staged canary telemetry and the
      evaluator read the same file for the whole run.
    tasks:
      # --- 2. canary fixture ---
      - id: stage_canary_metrics
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write canary cohort telemetry with a heredoc, a 250 device canary
          wave whose reboot failures and error events breach the default
          threshold beside a healthy control cohort, so trial runs always have
          telemetry to evaluate.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: alpine:3.20
        commands:
          - |
            set -u
            cat > canary_metrics.json <<'JSON'
            {
              "rollout": "fw-2.14.0",
              "device_model": "sensor-hub-v3",
              "cohorts": {
                "canary": {"devices": 250, "updated": 246, "reboot_failures": 9, "error_events": 17, "unresponsive": 3},
                "control": {"devices": 4750, "updated": 4701, "reboot_failures": 4, "error_events": 21}
              }
            }
            JSON
            echo "staged $(wc -c < canary_metrics.json) bytes of canary telemetry"

      # --- 3. canary evaluation ---
      - id: evaluate_canary
        type: io.kestra.plugin.scripts.python.Script
        description: Turn the canary and control cohorts into error and reboot failure
          rates, compare the canary rate against the threshold, and emit the
          halt decision and the cohort counters through the outputs protocol.
        containerImage: python:3.12-slim
        env:
          ROLLOUT_JSON: "{{ trigger.body ?? '' }}"
          ERROR_THRESHOLD_PCT: "{{ inputs.error_threshold_pct }}"
          ROLLOUT_NAME: "{{ inputs.rollout_name }}"
        script: |
          import json
          import os

          DEFAULT_TELEMETRY = """
          {
            "rollout": "fw-2.14.0",
            "device_model": "sensor-hub-v3",
            "cohorts": {
              "canary": {"devices": 250, "updated": 246, "reboot_failures": 9, "error_events": 17, "unresponsive": 3},
              "control": {"devices": 4750, "updated": 4701, "reboot_failures": 4, "error_events": 21}
            }
          }
          """


          def as_int(value, fallback=0):
              try:
                  return int(value)
              except (TypeError, ValueError):
                  try:
                      return int(float(value))
                  except (TypeError, ValueError):
                      return fallback


          def as_float(value, fallback=0.0):
              try:
                  return float(value)
              except (TypeError, ValueError):
                  return fallback


          raw = (os.environ.get("ROLLOUT_JSON") or "").strip() or DEFAULT_TELEMETRY.strip()
          threshold = max(0.0, as_float(os.environ.get("ERROR_THRESHOLD_PCT"), 2.0))
          fallback_rollout = (os.environ.get("ROLLOUT_NAME") or "unknown").strip() or "unknown"

          outputs = {
              "data_ok": False,
              "rollout": fallback_rollout,
              "device_model": "unknown",
              "decision": "HALT",
              "halted": True,
              "canary_devices": 0,
              "updated_devices": 0,
              "reboot_failures": 0,
              "canary_errors": 0,
              "canary_error_rate": 0.0,
              "reboot_failure_rate": 0.0,
              "control_error_rate": 0.0,
          }

          try:
              document = json.loads(raw)
          except ValueError:
              document = None

          cohorts = document.get("cohorts") if isinstance(document, dict) else None
          canary = cohorts.get("canary") if isinstance(cohorts, dict) else None
          control = cohorts.get("control") if isinstance(cohorts, dict) else None

          if isinstance(canary, dict):
              devices = max(0, as_int(canary.get("devices")))
              updated = max(0, as_int(canary.get("updated")))
              outputs["canary_devices"] = devices
              if updated > 0:
                  reboot_failures = max(0, as_int(canary.get("reboot_failures")))
                  error_events = max(0, as_int(canary.get("error_events")))
                  unresponsive = max(0, as_int(canary.get("unresponsive")))
                  errors = reboot_failures + error_events + unresponsive
                  canary_rate = round(errors * 100.0 / updated, 2)
                  reboot_rate = round(reboot_failures * 100.0 / updated, 2)
                  control_rate = 0.0
                  if isinstance(control, dict):
                      control_updated = max(0, as_int(control.get("updated")))
                      if control_updated > 0:
                          control_errors = (max(0, as_int(control.get("reboot_failures")))
                                            + max(0, as_int(control.get("error_events"))))
                          control_rate = round(control_errors * 100.0 / control_updated, 2)
                  halted = canary_rate >= threshold
                  outputs.update({
                      "data_ok": True,
                      "rollout": str(document.get("rollout") or fallback_rollout),
                      "device_model": str(document.get("device_model") or "unknown"),
                      "decision": "HALT" if halted else "GO",
                      "halted": halted,
                      "canary_devices": devices,
                      "updated_devices": updated,
                      "reboot_failures": reboot_failures,
                      "canary_errors": errors,
                      "canary_error_rate": canary_rate,
                      "reboot_failure_rate": reboot_rate,
                      "control_error_rate": control_rate,
                  })

          print("data_ok=%s rollout=%s decision=%s canary_rate=%.2f%% reboot_rate=%.2f%% threshold=%.2f%%" % (
              outputs["data_ok"], outputs["rollout"], outputs["decision"],
              outputs["canary_error_rate"], outputs["reboot_failure_rate"], threshold))
          print("::" + json.dumps({"outputs": outputs}) + "::")

  # --- 4. halt gate ---
  - id: halt_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch on the rollout decision, so a failing or unevaluable canary
      halts before the rollout widens while a healthy canary is logged as
      cleared to proceed.
    condition: "{{ outputs.evaluate_canary.vars.halted }}"
    then:
      # --- 5. halt escalation ---
      - id: alert_halt
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the rollout, the canary error rate against the threshold, the
          reboot failures and the control rate so the release owner stops the
          rollout immediately.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "OTA firmware rollout HALT for {{ outputs.evaluate_canary.vars.rollout }} ({{ outputs.evaluate_canary.vars.device_model }}) - canary error rate {{ outputs.evaluate_canary.vars.canary_error_rate }} percent against the {{ inputs.error_threshold_pct }} percent threshold, {{ outputs.evaluate_canary.vars.reboot_failures }} reboot failures and {{ outputs.evaluate_canary.vars.canary_errors }} total error events across {{ outputs.evaluate_canary.vars.updated_devices }} updated devices, control rate {{ outputs.evaluate_canary.vars.control_error_rate }} percent. data_ok {{ outputs.evaluate_canary.vars.data_ok }}. Do not widen the rollout. Execution {{ execution.id }}."
          }
      - id: enforce_halt
        type: io.kestra.plugin.core.flow.If
        description: Apply the auto_halt policy, turning the run red for release
          pipelines and staying observe only when the input is flipped off.
        condition: "{{ inputs.auto_halt }}"
        then:
          - id: fail_rollout
            type: io.kestra.plugin.core.execution.Fail
            description: Fail the execution so the release pipeline stops instead of pushing
              the firmware to the remaining fleet.
            errorMessage: "OTA rollout {{ outputs.evaluate_canary.vars.rollout }} halted at
              the canary cohort with a {{
              outputs.evaluate_canary.vars.canary_error_rate }} percent error
              rate against the {{ inputs.error_threshold_pct }} percent
              threshold."
        else:
          - id: log_observe_only
            type: io.kestra.plugin.core.log.Log
            description: Record the halt verdict while keeping the run green, for estates
              that want the decision reported rather than enforced.
            message: "OTA rollout {{ outputs.evaluate_canary.vars.rollout }} canary failure
              not enforced - auto_halt is false, {{
              outputs.evaluate_canary.vars.canary_error_rate }} percent error
              rate alerted only."
    else:
      - id: log_canary_healthy
        type: io.kestra.plugin.core.log.Log
        description: Record the passing canary with its rates so a cleared rollout still
          leaves evidence the gate ran.
        message: "OTA rollout {{ outputs.evaluate_canary.vars.rollout }} canary healthy
          - {{ outputs.evaluate_canary.vars.canary_error_rate }} percent error
          rate under the {{ inputs.error_threshold_pct }} percent threshold
          across {{ outputs.evaluate_canary.vars.updated_devices }} updated
          devices, cleared to widen."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: rollout_decision
    type: STRING
    description: Gate verdict for the canary wave, either HALT when the error rate
      met the threshold or when telemetry could not be evaluated, or GO when it
      passed.
    value: "{{ outputs.evaluate_canary.vars.decision }}"
  - id: halted
    type: BOOL
    description: True whenever the rollout decision is HALT, the single condition
      the halt gate branches on.
    value: "{{ outputs.evaluate_canary.vars.halted }}"
  - id: canary_error_rate
    type: FLOAT
    description: Percentage of updated canary devices that reported reboot failures,
      error events or went unresponsive, for example 11.79 on the bundled
      telemetry.
    value: "{{ outputs.evaluate_canary.vars.canary_error_rate }}"
  - id: reboot_failure_rate
    type: FLOAT
    description: Percentage of updated canary devices that failed to reboot after
      taking the firmware, tracked separately because bricked devices never
      report errors.
    value: "{{ outputs.evaluate_canary.vars.reboot_failure_rate }}"
  - id: control_error_rate
    type: FLOAT
    description: Error rate percentage of the untouched control cohort, the baseline
      that shows whether errors come from the firmware or the environment.
    value: "{{ outputs.evaluate_canary.vars.control_error_rate }}"