Report the four DORA metrics per service every week and alert when delivery degrades

id: dora-metrics-weekly-report
namespace: company.team

description: |
  Weekly DORA metrics per service: deployment frequency, lead time for changes, change failure
  rate and time to restore, computed from deployments, the commits they shipped and the
  incidents they caused. Each service is placed in a DORA performance band, compared with its
  previous report over the same window, kept in KV, and flagged when a metric degrades past a threshold.

inputs:
  - id: week_ending
    type: STRING
    displayName: Week ending (Sunday)
    defaults: "2026-10-04"
    validator: ^\d{4}-\d{2}-\d{2}$

  - id: window_days
    type: INT
    displayName: Window (days)
    description: Metrics are computed over this rolling window, ending on
      week_ending. 28 days smooths a quiet week.
    defaults: 28

  - id: degrade_pct
    type: FLOAT
    displayName: Alert on degradation (%)
    description: A metric worse than the previous report by more than this
      percentage is flagged.
    defaults: 25.0

  - id: scenario
    type: SELECT
    displayName: Demo scenario
    description: ISSUES adds deployments with no commit linked and an incident with
      a resolution before its start, which must stop the report.
    values:
      - CLEAN
      - ISSUES
    defaults: CLEAN

variables:
  state_key: dora_history
  pack_dir: "dora/{{ inputs.window_days }}d/{{ inputs.week_ending }}"

concurrency:
  limit: 1

tasks:
  - id: state
    type: io.kestra.plugin.core.output.OutputValues
    values:
      previous: >-
        {{ (kv(vars.state_key, errorOnMissing=false) ?? {'weeks': {}}) |
        jq('.weeks | to_entries | map(select((.key | startswith("' ~
        inputs.window_days ~ 'd/")) and .key < "' ~ inputs.window_days ~ 'd/' ~
        inputs.week_ending ~ '")) | sort_by(.key) | last | .value // {}') |
        first | toJson }}

  - id: metrics
    type: io.kestra.plugin.jdbc.duckdb.Queries
    description: >
      One DuckDB session: deployments, the commits each one shipped and
      incidents linked to the deployment that caused them (demo generator or
      your CI/CD, Git and incident tool), then the four metrics per service, the
      bands and the comparison with the previous report.
    outputFiles:
      - services
      - failed_changes
      - blockers
    fetchType: FETCH_ONE
    sql: |
      SET VARIABLE wk_end = DATE '{{ inputs.week_ending }}';
      SET VARIABLE w_start = getvariable('wk_end') - {{ inputs.window_days }} + 1;
      -- The demo history is fixed: six weeks ending 2026-10-04, whatever week you report on.
      SET VARIABLE anchor = DATE '2026-10-04';

      -- =======================================================================
      -- Demo data, deterministic. Replace with your deployment log, Git and
      -- incident tool.
      -- =======================================================================
      CREATE TABLE services (service VARCHAR, team VARCHAR);
      INSERT INTO services VALUES ('checkout', 'Payments'), ('search', 'Discovery'), ('ledger', 'Finance Eng'), ('mobile-api', 'Mobile');

      -- Deployments to production. Checkout ships several times a day, ledger every two weeks.
      CREATE TABLE deployments AS
      SELECT s.service || '-' || lpad(i::VARCHAR, 4, '0') AS deploy_id, s.service,
             (getvariable('anchor') - 41)::TIMESTAMP + to_hours(i * s.every_h) AS deployed_at, 'SUCCESS' AS status
      FROM (VALUES ('checkout', 7), ('search', 26), ('ledger', 330), ('mobile-api', 80)) s(service, every_h),
           range(0, 1000) t(i)
      WHERE i * s.every_h < 42 * 24;
      -- A deploy that failed in the pipeline never reached users: it is not a production change.
      UPDATE deployments SET status = 'FAILED' WHERE service = 'search' AND hash(deploy_id) % 9 = 0;

      -- Commits shipped by each deployment: the commits since the previous successful deploy.
      -- Search started batching its changes in the last two weeks, so its lead time grew.
      CREATE TABLE commits AS
      SELECT d.deploy_id, d.service, 'c' || hash(d.deploy_id || k) % 1000000 AS sha,
             d.deployed_at - to_minutes((CASE d.service WHEN 'checkout' THEN 90 WHEN 'search' THEN (CASE WHEN d.deployed_at >= getvariable('anchor') - 13 THEN 4200 ELSE 1300 END)
               WHEN 'ledger' THEN 9000 ELSE 2600 END * (0.4 + (hash(d.deploy_id || k) % 120) / 100.0))::BIGINT) AS committed_at
      FROM deployments d, range(0, 3) r(k) WHERE d.status = 'SUCCESS';
      {% if inputs.scenario == 'ISSUES' %}
      INSERT INTO deployments VALUES ('mobile-api-hotfix-1', 'mobile-api', getvariable('anchor') - 2 + INTERVAL 3 HOUR, 'SUCCESS'),
                                     ('mobile-api-hotfix-2', 'mobile-api', getvariable('anchor') - 1 + INTERVAL 5 HOUR, 'SUCCESS');
      {% endif %}

      -- Incidents caused by a deployment (a rollback, a hotfix or an incident linked to the change).
      -- Mobile had a bad fortnight: three failed changes, one of them a long outage.
      CREATE TABLE incidents (incident_id VARCHAR, service VARCHAR, deploy_id VARCHAR, started_at TIMESTAMP, resolved_at TIMESTAMP, severity VARCHAR);
      INSERT INTO incidents
      SELECT 'INC-' || row_number() OVER (ORDER BY deployed_at), service, deploy_id, deployed_at + INTERVAL 20 MINUTE,
             deployed_at + INTERVAL 20 MINUTE + to_minutes(CASE service WHEN 'checkout' THEN 35 WHEN 'ledger' THEN 300 ELSE 140 END), 'SEV2'
      FROM deployments
      WHERE status = 'SUCCESS' AND ((service = 'checkout' AND hash(deploy_id) % 23 = 0) OR (service = 'search' AND hash(deploy_id) % 11 = 0)
         OR (service = 'ledger' AND deployed_at::DATE = getvariable('anchor') - 41 + 13));
      INSERT INTO incidents
      SELECT 'INC-M' || row_number() OVER (ORDER BY deployed_at), service, deploy_id, deployed_at + INTERVAL 15 MINUTE,
             deployed_at + INTERVAL 15 MINUTE + to_minutes(CASE WHEN row_number() OVER (ORDER BY deployed_at) = 1 THEN 1260 ELSE 95 END), 'SEV1'
      FROM (SELECT * FROM deployments WHERE service = 'mobile-api' AND deployed_at >= getvariable('anchor') - 13 ORDER BY deployed_at LIMIT 3);
      {% if inputs.scenario == 'ISSUES' %}
      INSERT INTO incidents VALUES ('INC-BAD', 'search', NULL, getvariable('anchor') - 3 + INTERVAL 10 HOUR, getvariable('anchor') - 3 + INTERVAL 9 HOUR, 'SEV3');
      {% endif %}

      -- =======================================================================
      -- Checks. A deploy with no commit has no lead time and hides failures from
      -- the change failure rate; a negative restore time breaks the median.
      -- =======================================================================
      CREATE TABLE blockers AS
      SELECT 'DEPLOY_WITHOUT_COMMITS' AS code, deploy_id AS ref, 'Successful production deploy with no commit linked' AS detail
        FROM deployments d WHERE status = 'SUCCESS' AND deployed_at::DATE BETWEEN getvariable('w_start') AND getvariable('wk_end')
        AND NOT EXISTS (SELECT 1 FROM commits c WHERE c.deploy_id = d.deploy_id)
      UNION ALL SELECT 'RESOLVED_BEFORE_START', incident_id, 'Resolved at ' || resolved_at || ', before it started at ' || started_at FROM incidents WHERE resolved_at < started_at
      UNION ALL SELECT 'UNKNOWN_SERVICE', deploy_id, 'Service ' || service || ' is not in the catalog' FROM deployments d WHERE NOT EXISTS (SELECT 1 FROM services s WHERE s.service = d.service);

      -- =======================================================================
      -- The four metrics over the window, per service.
      -- =======================================================================
      CREATE TABLE prod AS SELECT * FROM deployments WHERE status = 'SUCCESS' AND deployed_at::DATE BETWEEN getvariable('w_start') AND getvariable('wk_end');
      CREATE TABLE m AS
      SELECT s.service, s.team,
             (SELECT count(*) FROM prod p WHERE p.service = s.service) AS deploys,
             round((SELECT count(*) FROM prod p WHERE p.service = s.service) * 7.0 / {{ inputs.window_days }}, 2) AS deploys_per_week,
             -- Lead time: median of commit-to-production over every commit shipped in the window.
             round((SELECT median(date_diff('minute', c.committed_at, p.deployed_at)) FROM prod p JOIN commits c USING (deploy_id) WHERE p.service = s.service) / 60.0, 1) AS lead_time_h,
             (SELECT count(DISTINCT i.deploy_id) FROM incidents i JOIN prod p USING (deploy_id) WHERE p.service = s.service) AS failed_changes,
             round((SELECT median(date_diff('minute', i.started_at, i.resolved_at)) FROM incidents i
                    WHERE i.service = s.service AND i.started_at::DATE BETWEEN getvariable('w_start') AND getvariable('wk_end') AND i.resolved_at >= i.started_at) / 60.0, 2) AS restore_h
      FROM services s;
      ALTER TABLE m ADD COLUMN cfr_pct DOUBLE; UPDATE m SET cfr_pct = round(100.0 * failed_changes / nullif(deploys, 0), 1);

      -- DORA bands (Accelerate State of DevOps). Overall band = the weakest of the four.
      CREATE TABLE banded AS
      SELECT *,
             CASE WHEN deploys_per_week >= 7 THEN 4 WHEN deploys_per_week >= 1 THEN 3 WHEN deploys_per_week >= 0.25 THEN 2 ELSE 1 END AS b_df,
             CASE WHEN lead_time_h IS NULL THEN NULL WHEN lead_time_h < 24 THEN 4 WHEN lead_time_h < 168 THEN 3 WHEN lead_time_h < 720 THEN 2 ELSE 1 END AS b_lt,
             CASE WHEN cfr_pct IS NULL THEN NULL WHEN cfr_pct <= 5 THEN 4 WHEN cfr_pct <= 10 THEN 3 WHEN cfr_pct <= 15 THEN 2 ELSE 1 END AS b_cfr,
             CASE WHEN restore_h IS NULL THEN 4 WHEN restore_h < 1 THEN 4 WHEN restore_h < 24 THEN 3 WHEN restore_h < 168 THEN 2 ELSE 1 END AS b_mttr
      FROM m;
      CREATE TABLE prev AS SELECT key AS service, value->>'lead_time_h' AS lt, value->>'cfr_pct' AS cfr, value->>'restore_h' AS mttr, value->>'deploys_per_week' AS df
        FROM json_each('{{ outputs.state.values.previous }}');
      CREATE TABLE report AS
      SELECT b.service, b.team, b.deploys, b.deploys_per_week, b.lead_time_h, b.failed_changes, b.cfr_pct, b.restore_h,
             ['', 'LOW', 'MEDIUM', 'HIGH', 'ELITE'][least(b.b_df, coalesce(b.b_lt, 4), coalesce(b.b_cfr, 4), b.b_mttr) + 1] AS band,
             p.df::DOUBLE AS prev_deploys_per_week, p.lt::DOUBLE AS prev_lead_time_h, p.cfr::DOUBLE AS prev_cfr_pct, p.mttr::DOUBLE AS prev_restore_h,
             concat_ws('; ',
               CASE WHEN p.df::DOUBLE > 0 AND b.deploys_per_week < p.df::DOUBLE * (1 - {{ inputs.degrade_pct }} / 100) THEN 'deploy frequency ' || p.df || ' -> ' || b.deploys_per_week || '/week' END,
               CASE WHEN p.lt::DOUBLE > 0 AND b.lead_time_h > p.lt::DOUBLE * (1 + {{ inputs.degrade_pct }} / 100) THEN 'lead time ' || p.lt || 'h -> ' || b.lead_time_h || 'h' END,
               CASE WHEN p.cfr IS NOT NULL AND b.cfr_pct > greatest(p.cfr::DOUBLE * (1 + {{ inputs.degrade_pct }} / 100), p.cfr::DOUBLE + 2) THEN 'change failure rate ' || p.cfr || '% -> ' || b.cfr_pct || '%' END,
               CASE WHEN p.mttr::DOUBLE > 0 AND b.restore_h > p.mttr::DOUBLE * (1 + {{ inputs.degrade_pct }} / 100) THEN 'time to restore ' || p.mttr || 'h -> ' || b.restore_h || 'h' END) AS degraded
      FROM banded b LEFT JOIN prev p USING (service) ORDER BY b.service;

      COPY report TO '{{ outputFiles.services }}' (HEADER, DELIMITER ',');
      COPY (SELECT i.service, i.deploy_id, i.incident_id, i.severity, i.started_at, i.resolved_at, round(date_diff('minute', i.started_at, i.resolved_at) / 60.0, 2) AS restore_h
            FROM incidents i JOIN prod USING (deploy_id) ORDER BY i.started_at) TO '{{ outputFiles.failed_changes }}' (HEADER, DELIMITER ',');
      COPY blockers TO '{{ outputFiles.blockers }}' (HEADER, DELIMITER ',');

      SELECT
        (SELECT string_agg(service || ' ' || band || ' (' || deploys_per_week || '/wk, lead ' || coalesce(lead_time_h::VARCHAR || 'h', '-') || ', CFR ' || coalesce(cfr_pct::VARCHAR || '%', '-') || ', restore ' || coalesce(restore_h::VARCHAR || 'h', 'no incident') || ')', '; ' ORDER BY service) FROM report) AS summary,
        (SELECT coalesce(string_agg(service || ': ' || degraded, ' | ' ORDER BY service), '') FROM report WHERE degraded <> '') AS degraded,
        (SELECT count(*) FROM report WHERE degraded <> '')::INT AS degraded_count,
        (SELECT sum(deploys) FROM report)::INT AS deploys,
        (SELECT json_group_object(service, json_object('deploys_per_week', deploys_per_week, 'lead_time_h', lead_time_h, 'cfr_pct', cfr_pct, 'restore_h', restore_h, 'band', band)) FROM report)::VARCHAR AS snapshot,
        (SELECT count(*) FROM blockers)::INT AS blockers,
        (SELECT coalesce(string_agg(code || ' ' || ref, ', ' ORDER BY code, ref), '') FROM blockers) AS blocker_detail;

  - id: result
    type: io.kestra.plugin.core.output.OutputValues
    values:
      r: "{{ (outputs.metrics.outputs | last).row | toJson }}"

  - id: log_report
    type: io.kestra.plugin.core.log.Log
    message: |
      DORA, {{ inputs.window_days }} days to {{ inputs.week_ending }} ({{ inputs.scenario }}), {{ outputs.result.values.r | jq('.deploys') | first }} production deploys.
      {{ outputs.result.values.r | jq('.summary') | first }}
      Degraded vs previous report: {{ outputs.result.values.r | jq('if .degraded == "" then "none" else .degraded end') | first }}
      Blockers: {{ outputs.result.values.r | jq('.blockers') | first }} {{ outputs.result.values.r | jq('.blocker_detail') | first }}

  - id: evidence_blockers
    type: io.kestra.plugin.core.namespace.UploadFiles
    namespace: "{{ flow.namespace }}"
    filesMap:
      "{{ render(vars.pack_dir) }}/blockers.csv": "{{ outputs.metrics.outputFiles.blockers }}"

  - id: gate
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.result.values.r | jq('.blockers > 0') | first }}"
    then:
      - id: blocked
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: "DORA report not published, the data is incomplete: {{
          outputs.result.values.r | jq('.blocker_detail') | first }}"

  - id: evidence
    type: io.kestra.plugin.core.namespace.UploadFiles
    namespace: "{{ flow.namespace }}"
    filesMap:
      "{{ render(vars.pack_dir) }}/services.csv": "{{ outputs.metrics.outputFiles.services }}"
      "{{ render(vars.pack_dir) }}/failed-changes.csv": "{{ outputs.metrics.outputFiles.failed_changes }}"

  - id: save
    type: io.kestra.plugin.core.kv.Set
    key: "{{ vars.state_key }}"
    kvType: JSON
    value: >-
      {{ {'old': (kv(vars.state_key, errorOnMissing=false) ?? {'weeks': {}}),
      'wk': inputs.window_days ~ 'd/' ~ inputs.week_ending, 'snap':
      fromJson(fromJson(outputs.result.values.r).snapshot)}
         | jq('{weeks: (.old.weeks + {(.wk): .snap})}') | first | toJson }}

  - id: degraded_gate
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.result.values.r | jq('.degraded_count > 0') | first }}"
    then:
      - id: alert
        type: io.kestra.plugin.core.log.Log
        level: WARN
        message: "DORA metrics degraded: {{ outputs.result.values.r | jq('.degraded') |
          first }}"

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 7 * * 1"
    disabled: true
    inputs:
      week_ending: "{{ trigger.date | dateAdd(-1, 'DAYS') | date('yyyy-MM-dd') }}"