PostHog Error Burst and Feature Flag Blast Radius Incident Gate

id: posthog-error-burst-and-feature-flag-blast-radius-gate
namespace: company.analytics

inputs:
  - id: posthog_host
    type: STRING
    displayName: PostHog Ingestion Host
    description: PostHog ingestion host URL used by project-API-key tasks
      (EvaluateFeatureFlag, Capture), e.g. https://us.i.posthog.com or a
      self-hosted URL.
    defaults: "https://us.i.posthog.com"

  - id: posthog_app_host
    type: STRING
    displayName: PostHog App Host
    description: PostHog app host used by the personal-API-key NewEvent trigger,
      e.g. https://us.posthog.com, https://eu.posthog.com, or a self-hosted URL.
      This is a different host than posthog_host.
    defaults: "https://us.posthog.com"

  - id: flag_key
    type: STRING
    displayName: Target Feature Flag Key
    description: Feature flag key currently in rollout that you want to monitor for
      regression correlation (e.g. checkout-redesign-v2).
    defaults: "checkout-redesign-v2"

  - id: correlation_threshold_pct
    type: INT
    displayName: Flag Correlation Threshold (%)
    description: Percentage of error-experiencing users who have the feature flag
      enabled required to classify the flag as the culprit.
    defaults: 70

  - id: min_error_count
    type: INT
    displayName: Minimum Error Count
    description: Minimum number of error events required to trigger blast radius
      correlation evaluation.
    defaults: 3

  - id: gate_on_regression
    type: BOOL
    displayName: Enforce Release Gate on Regression
    description: When true, automatically fails the execution if the error burst
      strongly correlates with the feature flag.
    defaults: true

  - id: notify_slack
    type: BOOL
    displayName: Send Slack Alert
    description: Dispatches high-severity incident notifications to Slack. Requires
      the SLACK_WEBHOOK_URL secret.
    defaults: false

  - id: sample_error_events
    type: JSON
    displayName: Sample Error Events (Fallback / Manual Run)
    description: Seed error events used when manually testing without waiting for
      the NewEvent polling trigger.
    defaults: |
      [
        {
          "id": "evt-101",
          "distinctId": "usr-prod-alpha-1",
          "event": "$exception",
          "properties": {"$exception_message": "TypeError: Cannot read properties of undefined (reading 'billing_address')", "browser": "Chrome", "path": "/checkout"}
        },
        {
          "id": "evt-102",
          "distinctId": "usr-prod-alpha-2",
          "event": "$exception",
          "properties": {"$exception_message": "TypeError: Cannot read properties of undefined (reading 'billing_address')", "browser": "Firefox", "path": "/checkout"}
        },
        {
          "id": "evt-103",
          "distinctId": "usr-prod-alpha-3",
          "event": "$exception",
          "properties": {"$exception_message": "TypeError: Cannot read properties of undefined (reading 'billing_address')", "browser": "Safari", "path": "/checkout"}
        },
        {
          "id": "evt-104",
          "distinctId": "usr-prod-control-9",
          "event": "$exception",
          "properties": {"$exception_message": "NetworkError: Failed to fetch pricing tier", "browser": "Chrome", "path": "/pricing"}
        }
      ]

triggers:
  - id: on_new_posthog_errors
    type: io.kestra.plugin.posthog.NewEvent
    description: Continuously polls PostHog for newly ingested client exception
      events and initiates incident triage.
    personalApiKey: "{{ secret('POSTHOG_PERSONAL_API_KEY') }}"
    projectId: "{{ secret('POSTHOG_PROJECT_ID') }}"
    host: "{{ inputs.posthog_app_host }}"
    eventName: "$exception"
    maxEvents: 50
    interval: PT5M

tasks:
  - id: extract_error_cluster
    type: io.kestra.plugin.scripts.python.Script
    description: Ingests error events from trigger or manual input, deduplicates
      affected distinct IDs, and extracts error signatures.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install kestra
    env:
      TRIGGER_EVENTS: "{{ trigger.events ? (trigger.events | toJson) : '' }}"
      FALLBACK_EVENTS: "{{ inputs.sample_error_events | toJson }}"
      MIN_ERROR_THRESHOLD: "{{ inputs.min_error_count }}"
    script: |
      import os
      import json
      from kestra import Kestra

      trigger_events_raw = os.environ.get("TRIGGER_EVENTS", "").strip()
      fallback_events_raw = os.environ.get("FALLBACK_EVENTS", "").strip()

      events = []
      if trigger_events_raw:
          try:
              events = json.loads(trigger_events_raw)
          except Exception:
              events = []

      if not events and fallback_events_raw:
          try:
              events = json.loads(fallback_events_raw)
          except Exception:
              events = []

      total_errors = len(events)
      distinct_users = list({e.get("distinctId") for e in events if e.get("distinctId")})

      signatures = {}
      for e in events:
          props = e.get("properties", {})
          msg = props.get("$exception_message", props.get("message", "UnknownException"))
          signatures[msg] = signatures.get(msg, 0) + 1

      min_threshold = int(os.environ.get("MIN_ERROR_THRESHOLD", "3"))
      has_incident_volume = total_errors >= min_threshold

      print(f"Processed {total_errors} errors across {len(distinct_users)} unique users.")
      print(f"Top signatures: {signatures}")

      Kestra.outputs({
          "total_errors": total_errors,
          "unique_user_count": len(distinct_users),
          "distinct_ids": distinct_users,
          "has_incident_volume": has_incident_volume,
          "top_signatures": signatures
      })

  - id: audit_user_feature_flags
    type: io.kestra.plugin.core.flow.Loop
    description: Evaluates the target feature flag state for each unique user
      experiencing client exceptions.
    values: "{{ outputs.extract_error_cluster.vars.distinct_ids }}"
    outputs:
      - id: distinct_id
        type: STRING
        value: "{{ item.value }}"
      - id: enabled
        type: BOOL
        value: "{{ outputs.evaluate_flag_for_user.enabled }}"
    tasks:
      - id: evaluate_flag_for_user
        type: io.kestra.plugin.posthog.EvaluateFeatureFlag
        description: Evaluates the feature flag for the specific user via PostHog API.
        apiKey: "{{ secret('POSTHOG_API_KEY') }}"
        host: "{{ inputs.posthog_host }}"
        distinctId: "{{ item.value }}"
        flagKey: "{{ inputs.flag_key }}"

  - id: correlate_blast_radius
    type: io.kestra.plugin.scripts.python.Script
    description: Computes statistical correlation between feature flag state and
      error occurrences to determine causality.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install kestra
    env:
      TOTAL_ERRORS: "{{ outputs.extract_error_cluster.vars.total_errors }}"
      UNIQUE_USERS: "{{ outputs.extract_error_cluster.vars.unique_user_count }}"
      EVAL_RESULTS: "{{ outputs.audit_user_feature_flags.outputs | toJson }}"
      FLAG_KEY: "{{ inputs.flag_key }}"
      THRESHOLD_PCT: "{{ inputs.correlation_threshold_pct }}"
      HAS_VOLUME: "{{ outputs.extract_error_cluster.vars.has_incident_volume }}"
    script: |
      import os
      import json
      from kestra import Kestra

      total_errors = int(os.environ.get("TOTAL_ERRORS", "0"))
      unique_users = int(os.environ.get("UNIQUE_USERS", "0"))
      threshold_pct = float(os.environ.get("THRESHOLD_PCT", "70.0"))
      has_volume = os.environ.get("HAS_VOLUME", "false").lower() == "true"
      flag_key = os.environ.get("FLAG_KEY", "unknown-flag")

      eval_results_raw = os.environ.get("EVAL_RESULTS", "[]")
      try:
          eval_results = json.loads(eval_results_raw)
      except Exception:
          eval_results = []

      flag_enabled_count = 0
      flag_disabled_count = 0

      for entry in eval_results:
          iteration_outputs = entry.get("outputs", {}) if isinstance(entry, dict) else {}
          if iteration_outputs.get("enabled") is True:
              flag_enabled_count += 1
          else:
              flag_disabled_count += 1

      total_evaluated = flag_enabled_count + flag_disabled_count
      correlation_pct = (flag_enabled_count / total_evaluated * 100.0) if total_evaluated > 0 else 0.0

      verdict = "BASELINE_LOW_SEVERITY"
      summary = ""

      if not has_volume:
          verdict = "BASELINE_LOW_SEVERITY"
          summary = f"Error count ({total_errors}) is below minimum threshold. Normal operational baseline."
      elif correlation_pct >= threshold_pct:
          verdict = "FLAG_CORRELATED_REGRESSION"
          summary = f"CRITICAL: {correlation_pct:.1f}% of failing users ({flag_enabled_count}/{total_evaluated}) have flag '{flag_key}' ENABLED. Release regression detected."
      else:
          verdict = "ORGANIC_INFRASTRUCTURE_SURGE"
          summary = f"SURGE: {total_errors} errors observed, but only {correlation_pct:.1f}% have flag '{flag_key}' enabled. Indicates broader infrastructure outage."

      print(f"Blast Radius Result: {verdict} ({correlation_pct:.1f}% flag correlation)")

      Kestra.outputs({
          "verdict": verdict,
          "correlation_pct": round(correlation_pct, 2),
          "flag_enabled_users": flag_enabled_count,
          "flag_disabled_users": flag_disabled_count,
          "total_evaluated": total_evaluated,
          "summary": summary
      })

  - id: route_incident_action
    type: io.kestra.plugin.core.flow.Switch
    description: Routes mitigation, telemetry markers, and incident reporting based
      on the classified regression verdict.
    value: "{{ outputs.correlate_blast_radius.vars.verdict }}"
    cases:
      FLAG_CORRELATED_REGRESSION:
        - id: log_regression_warning
          type: io.kestra.plugin.core.log.Log
          description: Logs critical alert identifying the feature flag as the root cause
            of the error spike.
          message: "🚨 FLAG REGRESSION IDENTIFIED: {{
            outputs.correlate_blast_radius.vars.summary }}"

        - id: capture_incident_marker
          type: io.kestra.plugin.posthog.Capture
          description: Emits an incident telemetry marker event back to PostHog annotating
            the regression.
          apiKey: "{{ secret('POSTHOG_API_KEY') }}"
          host: "{{ inputs.posthog_host }}"
          distinctId: "kestra-release-gate"
          event: "feature_flag_regression_detected"
          properties:
            flag_key: "{{ inputs.flag_key }}"
            correlation_pct: "{{ outputs.correlate_blast_radius.vars.correlation_pct }}"
            affected_users: "{{ outputs.correlate_blast_radius.vars.total_evaluated }}"
            kestra_execution_id: "{{ execution.id }}"
            status: "ROLLBACK_RECOMMENDED"

        - id: persist_incident_manifest
          type: io.kestra.plugin.core.storage.Write
          description: Persists full JSON incident triage dossier to internal storage for
            post-mortem analysis.
          extension: .json
          content: |
            {
              "verdict": "FLAG_CORRELATED_REGRESSION",
              "flagKey": "{{ inputs.flag_key }}",
              "correlationPercentage": {{ outputs.correlate_blast_radius.vars.correlation_pct }},
              "flagEnabledFailingUsers": {{ outputs.correlate_blast_radius.vars.flag_enabled_users }},
              "flagDisabledFailingUsers": {{ outputs.correlate_blast_radius.vars.flag_disabled_users }},
              "totalErrors": {{ outputs.extract_error_cluster.vars.total_errors }},
              "executionId": "{{ execution.id }}",
              "actionRequired": "KILL_SWITCH_ENGAGE_OR_ROLLBACK"
            }

        - id: check_slack_alert
          type: io.kestra.plugin.core.flow.If
          description: Dispatches high-priority Slack notification to on-call engineers if
            enabled.
          condition: "{{ inputs.notify_slack }}"
          then:
            - id: notify_slack_channel
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              description: Posts incident triage summary to Slack release channel.
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "text": "🚨 *PostHog Feature Flag Regression Alert* | Execution `{{ execution.id }}`\n• *Flag:* `{{ inputs.flag_key }}`\n• *Correlation:* `{{ outputs.correlate_blast_radius.vars.correlation_pct }}%` of failing users\n• *Summary:* {{ outputs.correlate_blast_radius.vars.summary }}\n• *Action:* Recommended emergency kill switch"
                }

      ORGANIC_INFRASTRUCTURE_SURGE:
        - id: log_infrastructure_notice
          type: io.kestra.plugin.core.log.Log
          description: Exonerates the feature flag and notes general infrastructure anomaly.
          message: "INFRASTRUCTURE ANOMALY: {{ outputs.correlate_blast_radius.vars.summary
            }}"

        - id: persist_surge_manifest
          type: io.kestra.plugin.core.storage.Write
          description: Saves anomaly telemetry manifest to internal storage.
          extension: .json
          content: |
            {
              "verdict": "ORGANIC_INFRASTRUCTURE_SURGE",
              "flagKey": "{{ inputs.flag_key }}",
              "correlationPercentage": {{ outputs.correlate_blast_radius.vars.correlation_pct }},
              "totalErrors": {{ outputs.extract_error_cluster.vars.total_errors }},
              "executionId": "{{ execution.id }}",
              "note": "Flag exonerated; investigate backend services"
            }

      BASELINE_LOW_SEVERITY:
        - id: log_baseline_healthy
          type: io.kestra.plugin.core.log.Log
          description: Confirms error volume is within acceptable baseline bounds.
          message: "Baseline check passed: {{ outputs.correlate_blast_radius.vars.summary
            }}"

  - id: enforce_release_gate
    type: io.kestra.plugin.core.flow.If
    description: Fails the execution if gate_on_regression is enabled and a flag
      regression was confirmed.
    condition: "{{ inputs.gate_on_regression and
      outputs.correlate_blast_radius.vars.verdict ==
      'FLAG_CORRELATED_REGRESSION' }}"
    then:
      - id: fail_deployment_gate
        type: io.kestra.plugin.core.execution.Fail
        description: Fails the pipeline execution to stop continuous delivery rollouts
          or trip release alarms.
        errorMessage: "Release gate tripped: Feature flag '{{ inputs.flag_key }}'
          strongly correlates with active error burst ({{
          outputs.correlate_blast_radius.vars.correlation_pct }}% correlation)."