Stream Analytics Job Health Gate

id: stream-analytics-job-health-gate
namespace: company.team

description: |
  Read an Azure Stream Analytics job's real jobState and provisioningState, classify the job
  HEALTHY, WARNING, CRITICAL or PROBE_FAILED against the state it is expected to be in, alert
  Slack with the exact states returned, and log a final audit summary either way.

inputs:
  - id: resource_group
    type: STRING
    displayName: Resource group name
    description: Resource group containing the Stream Analytics job. Maps to the
      GetJob task's resourceGroup field.
    defaults: "stream-analytics-rg"

  - id: job_name
    type: STRING
    displayName: Job name
    description: Name of the Stream Analytics job to audit. Maps to the GetJob
      task's jobName field.
    defaults: "clickstream-processor"

  - id: expected_job_state
    type: SELECT
    displayName: Expected job state
    description: The jobState this job should be in during normal operation. Most
      jobs should be Running continuously, some are intentionally Stopped
      between scheduled batch windows.
    values:
      - "Running"
      - "Stopped"
    defaults: "Running"

  - id: degraded_is_critical
    type: BOOL
    displayName: Treat degraded as critical
    description: When true, a Degraded jobState classifies CRITICAL, the same as
      Failed. When false, classifies WARNING instead, useful for a job with a
      known tolerance for brief degradation.
    defaults: true

tasks:
  - id: probe_job_status
    type: io.kestra.plugin.azure.streamanalytics.GetJob
    description: Read-only. Reads the job's real jobState and provisioningState.
    allowFailure: true
    tenantId: "{{ secret('AZURE_TENANT_ID') }}"
    clientId: "{{ secret('AZURE_CLIENT_ID') }}"
    clientSecret: "{{ secret('AZURE_CLIENT_SECRET') }}"
    subscriptionId: "{{ secret('AZURE_SUBSCRIPTION_ID') }}"
    resourceGroup: "{{ inputs.resource_group }}"
    jobName: "{{ inputs.job_name }}"

  - id: classify_job_health
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when the probe produced no output at all, CRITICAL on
      Failed or a failed/canceled provisioning state, CRITICAL or WARNING on
      Degraded depending on degraded_is_critical, HEALTHY when jobState matches
      expected_job_state and provisioning succeeded, WARNING for every other
      state such as a job still Starting or Scaling.
    format: "{{ outputs.probe_job_status is not defined ? 'PROBE_FAILED' :
      ((outputs.probe_job_status.jobState == 'Failed' or
      outputs.probe_job_status.provisioningState == 'Failed' or
      outputs.probe_job_status.provisioningState == 'Canceled') ? 'CRITICAL' :
      (outputs.probe_job_status.jobState == 'Degraded' ?
      (inputs.degraded_is_critical ? 'CRITICAL' : 'WARNING') :
      ((outputs.probe_job_status.jobState == inputs.expected_job_state and
      outputs.probe_job_status.provisioningState == 'Succeeded') ? 'HEALTHY' :
      'WARNING'))) }}"

  - id: route_by_job_health
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a healthy job. Branch B alerts Slack when the job
      failed, is degraded, or is in an unexpected state, or the probe itself
      failed.
    value: "{{ outputs.classify_job_health.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Stream Analytics job {{ inputs.job_name }} healthy. jobState={{
            outputs.probe_job_status.jobState }} provisioningState={{
            outputs.probe_job_status.provisioningState }}."
      "WARNING":
        - id: alert_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Stream Analytics job " ~ inputs.job_name ~ " WARNING. jobState=" ~ outputs.probe_job_status.jobState ~ " (expected " ~ inputs.expected_job_state ~ "), provisioningState=" ~ outputs.probe_job_status.provisioningState ~ "."} | toJson }}
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Stream Analytics job " ~ inputs.job_name ~ " CRITICAL. jobState=" ~ outputs.probe_job_status.jobState ~ ", provisioningState=" ~ outputs.probe_job_status.provisioningState ~ ". Streaming output has likely stopped or fallen behind."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Stream Analytics job probe for " ~ inputs.job_name ~ " in " ~ inputs.resource_group ~ " did not return a result. Check resource_group, job_name and Azure credentials."} | toJson }}

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "Stream Analytics job health gate completed. job={{ inputs.job_name }}
      status={{ outputs.classify_job_health.value }} expected_job_state={{
      inputs.expected_job_state }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "Stream Analytics job health gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: every_ten_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled.
    cron: "*/10 * * * *"
    disabled: true
    inputs:
      resource_group: "stream-analytics-rg"
      job_name: "clickstream-processor"
      expected_job_state: "Running"
      degraded_is_critical: true