HorizonDB Durable Instance Health Gate with Kestra

id: horizondb-durable-instance-health-gate
namespace: company.team

description: |
  Audit an Azure HorizonDB durable function store for Failed and Running instance counts,
  classify the result HEALTHY, WARNING, CRITICAL or PROBE_FAILED, alert Slack with the real
  counts, optionally cancel one named stuck instance an operator has already identified, and
  log a final audit summary either way.

inputs:
  - id: database
    type: STRING
    displayName: HorizonDB database name
    description: Target database on the HorizonDB server. Maps to the ListInstances,
      Cancel, and GetStatus tasks' database field.
    defaults: "orchestrations"

  - id: primary_watch_status
    type: SELECT
    displayName: Status this gate watches
    description: Failed only classifies on the failed instance count, Running only
      classifies on the running backlog count, Both classifies on either.
    values:
      - "Failed"
      - "Running"
      - "Both"
    defaults: "Both"

  - id: failed_count_critical_threshold
    type: INT
    displayName: Failed instance count threshold
    description: Number of instances in Failed status at or above which the run is
      classified CRITICAL.
    defaults: 5

  - id: running_backlog_warning_threshold
    type: INT
    displayName: Running backlog threshold
    description: Number of instances still in Running status at or above which the
      run is classified WARNING, a sign of a stalled worker pool rather than
      normal in-flight work.
    defaults: 50

  - id: sample_limit
    type: INT
    displayName: Sample limit per probe
    description: Maximum rows pulled per status by each ListInstances call. Maps to
      the ListInstances task's limit field, which pg_durable itself caps at
      10000.
    defaults: 100

  - id: use_entra_id
    type: BOOL
    displayName: Authenticate with Azure Entra ID
    description: When true, connects with Entra ID instead of a static password.
      Maps to the HorizonDB tasks' useEntraId field.
    defaults: false

  - id: cancel_instance_id
    type: STRING
    displayName: Instance ID to cancel
    description: Leave empty to never cancel anything. When set, together with
      auto_remediate true and dry_run false, cancels exactly this one instance,
      the one an operator already identified from a prior alert on this gate,
      never a bulk cancel.
    defaults: ""

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: When true, together with dry_run false and cancel_instance_id set,
      cancel that one instance and verify the result.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate true, for the cancel to
      actually run.
    defaults: true

tasks:
  - id: probe_failed_instances
    type: io.kestra.plugin.azure.horizondb.durable.ListInstances
    description: Read-only. Counts instances currently in Failed status.
    allowFailure: true
    host: "{{ secret('HORIZONDB_HOST') }}"
    database: "{{ inputs.database }}"
    username: "{{ secret('HORIZONDB_USERNAME') }}"
    password: "{{ secret('HORIZONDB_PASSWORD') }}"
    useEntraId: "{{ inputs.use_entra_id }}"
    tenantId: "{{ secret('AZURE_TENANT_ID') }}"
    clientId: "{{ secret('AZURE_CLIENT_ID') }}"
    clientSecret: "{{ secret('AZURE_CLIENT_SECRET') }}"
    statusFilter: "Failed"
    limit: "{{ inputs.sample_limit }}"
    fetchType: FETCH

  - id: probe_running_instances
    type: io.kestra.plugin.azure.horizondb.durable.ListInstances
    description: Read-only. Counts instances currently in Running status.
    allowFailure: true
    host: "{{ secret('HORIZONDB_HOST') }}"
    database: "{{ inputs.database }}"
    username: "{{ secret('HORIZONDB_USERNAME') }}"
    password: "{{ secret('HORIZONDB_PASSWORD') }}"
    useEntraId: "{{ inputs.use_entra_id }}"
    tenantId: "{{ secret('AZURE_TENANT_ID') }}"
    clientId: "{{ secret('AZURE_CLIENT_ID') }}"
    clientSecret: "{{ secret('AZURE_CLIENT_SECRET') }}"
    statusFilter: "Running"
    limit: "{{ inputs.sample_limit }}"
    fetchType: FETCH

  - id: classify_orchestration_health
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when either probe produced no output at all, CRITICAL
      when the watched failed count meets its threshold, WARNING when the
      watched running backlog meets its threshold, HEALTHY otherwise.
    format: "{{ (outputs.probe_failed_instances is not defined or
      outputs.probe_running_instances is not defined) ? 'PROBE_FAILED' :
      ((((inputs.primary_watch_status == 'Failed' or inputs.primary_watch_status
      == 'Both') and (outputs.probe_failed_instances.size ?? 0) >=
      inputs.failed_count_critical_threshold)) ? 'CRITICAL' :
      ((((inputs.primary_watch_status == 'Running' or
      inputs.primary_watch_status == 'Both') and
      (outputs.probe_running_instances.size ?? 0) >=
      inputs.running_backlog_warning_threshold)) ? 'WARNING' : 'HEALTHY')) }}"

  - id: route_by_orchestration_health
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a healthy store. Branch B alerts Slack with the real
      failed and running counts when either threshold is met, or the probe
      itself failed.
    value: "{{ outputs.classify_orchestration_health.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "HorizonDB {{ inputs.database }} healthy. failed={{
            outputs.probe_failed_instances.size }} running={{
            outputs.probe_running_instances.size }}."
      "WARNING":
        - id: alert_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "HorizonDB " ~ inputs.database ~ " WARNING. Running backlog at " ~ outputs.probe_running_instances.size ~ " instances, at or above the " ~ inputs.running_backlog_warning_threshold ~ " threshold. Failed count is " ~ outputs.probe_failed_instances.size ~ ". This usually means a stalled worker pool, not a traffic spike."} | toJson }}
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "HorizonDB " ~ inputs.database ~ " CRITICAL. " ~ outputs.probe_failed_instances.size ~ " instance(s) in Failed status, at or above the " ~ inputs.failed_count_critical_threshold ~ " threshold. Running backlog is " ~ outputs.probe_running_instances.size ~ ". Query df.list_instances('Failed') directly to find the exact instance IDs."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "HorizonDB instance audit on database " ~ inputs.database ~ " did not return a result from one or both ListInstances calls. Check HORIZONDB_HOST, credentials, and use_entra_id."} | toJson }}
    defaults:
      - id: log_unhandled_health_status
        type: io.kestra.plugin.core.log.Log
        message: "Unhandled HorizonDB health status: {{
          outputs.classify_orchestration_health.value }}"

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    description: Runs only when an operator has set cancel_instance_id to a specific
      instance from a prior alert on this gate, and auto_remediate is true and
      dry_run is false. Never cancels in bulk.
    condition: "{{ inputs.cancel_instance_id != '' and inputs.auto_remediate and not
      inputs.dry_run }}"
    then:
      - id: cancel_target_instance
        type: io.kestra.plugin.azure.horizondb.durable.Cancel
        description: df.cancel() does not document a return value confirming the
          instance actually stopped, verify_cancel below checks the real status
          afterward.
        allowFailure: true
        host: "{{ secret('HORIZONDB_HOST') }}"
        database: "{{ inputs.database }}"
        username: "{{ secret('HORIZONDB_USERNAME') }}"
        password: "{{ secret('HORIZONDB_PASSWORD') }}"
        useEntraId: "{{ inputs.use_entra_id }}"
        tenantId: "{{ secret('AZURE_TENANT_ID') }}"
        clientId: "{{ secret('AZURE_CLIENT_ID') }}"
        clientSecret: "{{ secret('AZURE_CLIENT_SECRET') }}"
        instanceId: "{{ inputs.cancel_instance_id }}"
        reason: "Cancelled by horizondb-durable-instance-health-gate after operator
          review (execution {{ execution.id }})."

      - id: verify_cancel
        type: io.kestra.plugin.azure.horizondb.durable.GetStatus
        allowFailure: true
        host: "{{ secret('HORIZONDB_HOST') }}"
        database: "{{ inputs.database }}"
        username: "{{ secret('HORIZONDB_USERNAME') }}"
        password: "{{ secret('HORIZONDB_PASSWORD') }}"
        useEntraId: "{{ inputs.use_entra_id }}"
        tenantId: "{{ secret('AZURE_TENANT_ID') }}"
        clientId: "{{ secret('AZURE_CLIENT_ID') }}"
        clientSecret: "{{ secret('AZURE_CLIENT_SECRET') }}"
        instanceId: "{{ inputs.cancel_instance_id }}"

      - id: report_cancel_result
        type: io.kestra.plugin.core.flow.If
        description: Fails the flow when the cancel call or the verification probe did
          not come back, or the instance is still Running, so a cancel that did
          not take is never silently reported as done.
        condition: "{{ outputs.cancel_target_instance is not defined or
          outputs.verify_cancel is not defined or outputs.verify_cancel.status
          == 'Running' }}"
        then:
          - id: cancel_not_confirmed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "Cancel of instance {{ inputs.cancel_instance_id }} did not
              confirm as stopped (status {{ outputs.verify_cancel.status ??
              'unreachable' }})."
        else:
          - id: cancel_confirmed
            type: io.kestra.plugin.core.log.Log
            message: "Instance {{ inputs.cancel_instance_id }} cancelled and confirmed.
              status={{ outputs.verify_cancel.status }}."

  - 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: "HorizonDB durable instance health gate completed. database={{
      inputs.database }} status={{ outputs.classify_orchestration_health.value
      }} failed={{ outputs.probe_failed_instances.size }} running={{
      outputs.probe_running_instances.size }}"

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

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode by default.
    cron: "0 * * * *"
    disabled: true
    inputs:
      database: "orchestrations"
      primary_watch_status: "Both"
      failed_count_critical_threshold: 5
      running_backlog_warning_threshold: 50
      sample_limit: 100
      use_entra_id: false
      cancel_instance_id: ""
      auto_remediate: false
      dry_run: true

A durable orchestration that gets stuck in Running, or a batch that starts failing after a downstream dependency changes, looks identical to normal traffic from the outside, the HTTP trigger that started it already returned 200. HorizonDB's own df.list_instances() and df.status() know exactly what state every instance is actually in, but nothing reads that report on a schedule. This blueprint turns a HorizonDB durable store into a Day 2 health gate: it counts instances by status, alerts Slack with the real numbers, and lets an operator cancel one specific instance they have already identified, never a bulk cancel.

How it works

probe_failed_instances and probe_running_instances (io.kestra.plugin.azure.horizondb.durable.ListInstances) each query df.list_instances(status, limit) for one status, Failed and Running, with allowFailure: true so a connection problem does not stop the flow before it can be classified as PROBE_FAILED below.
classify_orchestration_health (io.kestra.plugin.core.debug.Return) compares probe_failed_instances.size and probe_running_instances.size against their own thresholds, gated by primary_watch_status, and returns PROBE_FAILED, CRITICAL, WARNING, or HEALTHY.
route_by_orchestration_health (io.kestra.plugin.core.flow.Switch) branches on that status: HEALTHY logs it, WARNING, CRITICAL, and PROBE_FAILED each alert Slack with the real failed and running counts, never an invented number.
remediate_gate (io.kestra.plugin.core.flow.If) runs only when cancel_instance_id is set, auto_remediate is true, and dry_run is false:
- cancel_target_instance calls df.cancel(id, reason) on exactly that one instance. df.cancel() does not document a return value confirming the instance actually stopped, so this step alone proves nothing.
- verify_cancel calls df.status() on the same instance, and report_cancel_result (io.kestra.plugin.core.flow.If) fails the flow with io.kestra.plugin.core.execution.Fail if it is still Running, so a cancel that did not take is never silently reported as done.
log_audit_summary always runs last and logs the final counts and classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real count of failed and in-flight durable instances, read straight from HorizonDB's own df.list_instances(), not an assumption that everything that started also finished.
A Slack alert carrying the exact counts that triggered it, never an invented number.
A deliberately narrow remediation path: this blueprint cancels exactly one named instance an operator already identified, it does not expose a way to cancel instances in bulk from a schedule.
A verified cancel: the flow checks the instance's real status after cancelling and fails loudly if it is still running, instead of trusting df.cancel()'s own silence.

Who it's for

Platform teams running durable orchestrations on Azure HorizonDB who need to know about a growing Failed count or a stalled Running backlog before it is a customer-facing incident.
Teams evaluating the HorizonDB plugin who want a working example of durable.ListInstances, durable.Cancel, and durable.GetStatus together, including the Entra ID and static-password authentication paths.

Why orchestrate this with Kestra

HorizonDB exposes df.list_instances(), df.status(), and df.cancel() as plain SQL functions, it has no concept of a schedule, a threshold, or an alert built from what they return. Kestra adds the recurring audit, the classification by status count, the Slack alert, the deliberately scoped single-instance cancel, and the verification that the cancel worked, all as one declarative YAML file you can version and review like code.

Prerequisites

An Azure HorizonDB server with a database containing durable function instances.
Either a HorizonDB username and password, or an Azure Entra ID identity (managed identity, service principal, or az login on the worker) with useEntraId: true.
A Slack incoming webhook for alerts.
HorizonDB speaks the standard PostgreSQL wire protocol and this plugin uses the standard org.postgresql JDBC driver, ssl: false is explicitly supported for local development against a non-TLS instance. The df schema functions this blueprint depends on (df.list_instances, df.status, df.cancel) are HorizonDB's own managed implementation, a vanilla local PostgreSQL container will accept the connection but does not have these functions, there is no open-source emulator for the df schema itself. Local testing needs a real, non-production HorizonDB instance.

Secrets

HORIZONDB_HOST: HorizonDB server hostname, without protocol or port.
HORIZONDB_USERNAME, HORIZONDB_PASSWORD: static credentials, unused when use_entra_id is true.
AZURE_TENANT_ID, AZURE_CLIENT_ID, AZURE_CLIENT_SECRET: service principal credentials for Entra ID authentication, unused when use_entra_id is false. Omit all three to fall back to the default Azure credential chain on the worker.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set all these as base64-encoded environment variables, for example SECRET_HORIZONDB_HOST, SECRET_HORIZONDB_USERNAME, SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`database`	STRING	`orchestrations`	any HorizonDB database name
`primary_watch_status`	SELECT	`Both`	`Failed`, `Running`, `Both`
`failed_count_critical_threshold`	INT	`5`	any non-negative integer
`running_backlog_warning_threshold`	INT	`50`	any non-negative integer
`sample_limit`	INT	`100`	`1` to `10000`
`use_entra_id`	BOOL	`false`	`true`, `false`
`cancel_instance_id`	STRING	`` (empty)	empty, or a real instance ID
`auto_remediate`	BOOL	`false`	`true`, `false`
`dry_run`	BOOL	`true`	`true`, `false`

Outputs

{{ outputs.probe_failed_instances.size }}: count of durable function instances currently in Failed status.
{{ outputs.probe_failed_instances.rows }}: list of raw instance records returned by the Failed probe (fetchType: FETCH).
{{ outputs.probe_running_instances.size }}: count of durable function instances currently in Running status.
{{ outputs.probe_running_instances.rows }}: list of raw instance records returned by the Running probe (fetchType: FETCH).
{{ outputs.classify_orchestration_health.value }}: health classification evaluated by the gate (HEALTHY, WARNING, CRITICAL, or PROBE_FAILED).
{{ outputs.cancel_target_instance.cancelled }}: boolean indicating whether the cancel statement executed cleanly.
{{ outputs.verify_cancel.status }}: verified runtime status of the target instance after remediation.

Quick start

Add the HORIZONDB_HOST, credentials, and SLACK_WEBHOOK secrets.
Run this flow manually with the defaults and read log_audit_summary for the counts.
If it reports CRITICAL or WARNING, query df.list_instances('Failed') or df.list_instances('Running') directly to find the specific instance worth investigating.
Once you have a specific instance ID to cancel, set cancel_instance_id, auto_remediate: true, and dry_run: false and run this flow again.
Enable the hourly trigger, which always ships in safe mode.

How to extend

If your df.list_instances() result columns include a start or last-updated timestamp in your HorizonDB version, read it from outputs.probe_running_instances.instances to classify by age instead of only count, that detail was not documented in the plugin source this blueprint was verified against, confirm your own schema before relying on it.
Loop over outputs.probe_failed_instances.instances to post each failed instance's own identifying fields to Slack individually, instead of only the count.
Chain a Signal task after a verified cancel to notify a compensating orchestration that the original instance was terminated.

Pitfalls

df.cancel() does not document a return value confirming the instance actually stopped. cancelled: true on cancel_target_instance's output means the call executed without the driver throwing, nothing more, trust verify_cancel's real status, not that flag.
limit on ListInstances is capped at 10000 by the task itself specifically so this gate can never accidentally load an unbounded number of instances into memory, raise sample_limit deliberately, do not assume it is unlimited.
useEntraId: true with no tenantId/clientId/clientSecret set falls back to whatever DefaultAzureCredential resolves on the worker (managed identity, environment variables, az login), which can silently differ between a developer's laptop and a production worker. Set the three explicitly for a service principal if that matters.
ssl defaults to true (sslmode=require), only set it to false against a non-production instance you control.

HorizonDB Durable Instance Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

HorizonDB Durable Instance Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

More Cloud Blueprints

New to Kestra?