Fivetran Connector Freshness Sentinel for Kestra

id: fivetran-connector-freshness-sentinel
namespace: company.team
description: |
  On a schedule, read-only check a Fivetran connector's sync state, setup
  state, pause state, and freshness (nothing is synced, paused, or
  resumed) and classify it as HEALTHY, WARNING, CRITICAL, or UNREACHABLE.
  Log clean runs and page Slack the moment a connector fails, falls
  behind schedule, or sits paused without anyone noticing.

inputs:
  - id: connector_id
    type: STRING
    defaults: "<fivetran_connector_id>"
    displayName: "Fivetran Connector ID"
    description: "The Fivetran connector identifier to check, as shown in the
      Fivetran dashboard or API."
  - id: freshness_buffer
    type: STRING
    defaults: "PT30M"
    displayName: "Freshness Buffer"
    description: "ISO-8601 duration added on top of the connector's own sync
      frequency when Fivetran decides whether its last sync is still fresh."

triggers:
  - id: connector_freshness_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the connector every 6 hours. Shipped disabled; enable it once
      connector_id points at a real Fivetran connector and a manual run has been
      verified.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      freshness_buffer: "PT30M"

tasks:
  - id: check_connector_status
    type: io.kestra.plugin.fivetran.connectors.Status
    description: Read the connector's current sync state, setup state, pause state,
      and freshness in a single non-blocking call. Does not trigger a sync,
      pause, or resume the connector.
    apiKey: "{{ secret('FIVETRAN_API_KEY') }}"
    apiSecret: "{{ secret('FIVETRAN_API_SECRET') }}"
    connectorIds:
      - "{{ inputs.connector_id }}"
    freshnessBuffer: "{{ inputs.freshness_buffer }}"
    wait: false
    allowFailure: true

  - id: classify_connector_health
    type: io.kestra.plugin.core.debug.Return
    description: Classify the connector's reported state in one jq query, treating a
      missing output as an API or connector lookup failure, an active failure or
      broken setup as critical regardless of pause state, and a paused or stale
      connector as a drift warning.
    format: "{{ outputs.check_connector_status.connectors is not defined ?
      'UNREACHABLE' : (outputs.check_connector_status.connectors |
      jq('(to_entries[0].value) as $c | if $c == null then \"UNREACHABLE\" elif
      ($c.hasFailed == true) or ($c.setupState != \"connected\") then
      \"CRITICAL\" elif ($c.paused == true) or ($c.fresh == false) then
      \"WARNING\" else \"HEALTHY\" end') | first) }}"

  - id: route_by_connector_health
    type: io.kestra.plugin.core.flow.Switch
    description: Route to a quiet log on a fresh, synced connector, an informational
      Slack message when it is paused or stale, or a Slack page for an active
      failure, broken setup, or an unreachable API.
    value: "{{ outputs.classify_connector_health.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Fivetran connector {{ inputs.connector_id }} is healthy: synced and
            fresh, setupState connected."
      "WARNING":
        - id: alert_warning_connector
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Post an informational Slack message when the connector is paused or
            has fallen behind its expected sync schedule, without an active sync
            failure.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Fivetran freshness sentinel WARNING: connector {{ inputs.connector_id }} is either paused or stale beyond its freshness buffer of {{ inputs.freshness_buffer }}. No active sync failure was reported, but downstream data may be going stale. Execution {{ execution.id }}."
            }
      "CRITICAL":
        - id: alert_critical_connector
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the connector is actively failing to sync or its
            setup is no longer connected, regardless of pause state.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Fivetran freshness sentinel CRITICAL: connector {{ inputs.connector_id }} is reporting an active sync failure or a setupState other than connected. Downstream tables fed by this connector are very likely going stale right now. Execution {{ execution.id }}."
            }
      "UNREACHABLE":
        - id: alert_unreachable_connector
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Page Slack when the status check could not return connector state
            at all, indicating the Fivetran API is unreachable, credentials are
            invalid, or the connector ID does not exist.
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {
              "text": "Fivetran freshness sentinel UNREACHABLE: checking connector {{ inputs.connector_id }} did not return connector state (API unreachable, invalid API key/secret, or connector not found). Execution {{ execution.id }}."
            }
    defaults:
      - id: log_unclassified
        type: io.kestra.plugin.core.log.Log
        message: "Fivetran freshness sentinel completed with an unexpected
          classification: {{ outputs.classify_connector_health.value }}."

  - id: log_audit_result
    type: io.kestra.plugin.core.log.Log
    description: Record a single audit line for every run regardless of branch
      taken, including the raw sync/setup state and pause/freshness flags, so
      the check's own history stays queryable independent of Slack.
    message: "Fivetran freshness sentinel audit: connector={{ inputs.connector_id }}
      status={{ outputs.classify_connector_health.value }} details={{
      outputs.check_connector_status.connectors is defined ?
      (outputs.check_connector_status.connectors | jq('(to_entries[0].value) |
      {syncState, setupState, paused, fresh, hasFailed, succeededAt, failedAt}')
      | first) : 'n/a' }} execution={{ execution.id }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the sentinel flow itself fails outright, so a
      broken check is never mistaken for a healthy connector.
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "text": "Fivetran freshness sentinel flow {{ flow.id }} FAILED (execution {{ execution.id }}). Health of connector {{ inputs.connector_id }} is now unknown; check the task logs and the Fivetran dashboard manually."
      }

outputs:
  - id: status
    type: STRING
    value: "{{ outputs.classify_connector_health.value }}"

A paused or quietly failing Fivetran connector looks identical to a healthy one from the outside: the dashboards built on top of it just keep showing yesterday's numbers until someone notices. This blueprint turns a scheduled, non-blocking status read into a Day 2 data-pipeline health gate: io.kestra.plugin.fivetran.connectors.Status reports the connector's sync state, setup state, pause state, and Fivetran's own freshness calculation without triggering, pausing, or resuming anything, a single jq query classifies the combination, and the result routes to a quiet log, an informational Slack note, or a Slack page depending on severity.

How it works

check_connector_status (io.kestra.plugin.fivetran.connectors.Status) reads connector_id's state from the Fivetran API: syncState, setupState, paused, hasFailed, succeededAt, failedAt, and fresh (Fivetran's own freshness verdict, computed from the connector's syncFrequency plus freshness_buffer). wait: false makes this a single non-blocking read instead of letting the task poll internally until fresh, since this flow's own Schedule trigger provides the recurring cadence instead. allowFailure: true keeps an API or connector lookup failure from aborting the flow outright.
classify_connector_health (io.kestra.plugin.core.debug.Return) first guards for a missing connectors output (UNREACHABLE), then pulls the single requested connector's state with a jq query and classifies it: an active sync failure (hasFailed) or a setupState other than connected becomes CRITICAL regardless of pause state, since a broken connector matters more than whether someone paused it on purpose; a paused connector or one Fivetran itself reports as not fresh becomes WARNING; otherwise HEALTHY.
route_by_connector_health (io.kestra.plugin.core.flow.Switch) branches on that classification: HEALTHY logs a quiet confirmation; WARNING posts an informational Slack note; CRITICAL and UNREACHABLE each page Slack; a defaults fallback logs the raw value for any classification outside the four known cases.
log_audit_result (io.kestra.plugin.core.log.Log) writes one audit line per run, independent of which Switch branch fired, carrying the full raw state (sync state, setup state, paused, fresh, hasFailed, succeededAt, failedAt) so the check's own history stays queryable even if Slack delivery fails.
The flow-level errors block pages Slack separately if the flow fails outright, so a broken sentinel is never mistaken for a healthy connector.
The connector_freshness_schedule trigger checks every 6 hours and ships disabled: true, with an explicit inputs: override for freshness_buffer so the scheduled run's tolerance is pinned regardless of this flow's own defaults.

What you get

A scheduled, non-blocking health check for any Fivetran connector with zero custom Fivetran API scripting.
A single status output (HEALTHY / WARNING / CRITICAL / UNREACHABLE) other flows can gate on.
Three distinct Slack messages, one per severity, each explaining what was detected.
An audit log line on every run, separate from Slack, carrying the full raw connector state, so the check's own history survives a webhook outage.

Who it's for

Data engineering teams running Fivetran who have no alert for a connector that silently paused or started failing between dashboard glances.
Analytics engineers who want to know a source stopped syncing before a stakeholder notices the numbers look wrong.
Teams standardizing Day 2 data-pipeline governance across multiple ingestion tools inside Kestra instead of relying on Fivetran's own email digest alone.

Why orchestrate this with Kestra

Fivetran can email on sync failures, but that email lives outside your orchestration layer and says nothing about a connector someone paused and forgot about. Kestra's scheduled, non-blocking Status check folds sync failure, setup health, pause state, and freshness into one typed classification, with the schedule, the check, and the Slack call all captured as one auditable execution history instead of an inbox rule.

Prerequisites

A Fivetran account with an existing connector identified by connector_id.
A Fivetran API key and secret with read access to that connector.
A Slack incoming webhook for the WARNING, CRITICAL, UNREACHABLE, and flow-failure alerts.

Secrets

FIVETRAN_API_KEY, FIVETRAN_API_SECRET: Fivetran API credentials, paired for HTTP Basic authentication.
SLACK_WEBHOOK: Slack incoming webhook used for every alert branch and the flow-failure alert.

In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }}, for example SECRET_SLACK_WEBHOOK=$(echo -n 'https://hooks.slack.com/services/<placeholder>' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Quick start

Fivetran is a managed SaaS with no official local Docker image; use its free Starter tier and connect a small demo source such as its own sample data connector to get a real connector_id without a paid plan.
If Kestra runs in Docker and you proxy Fivetran API calls through a local gateway for testing, check docker network ls for Kestra's network and attach that gateway container to it: docker network connect YOUR_KESTRA_NETWORK <container_name> Avoid host port 8080 for any such local proxy since Kestra's own UI already uses it.
Generate a Fivetran API key and secret from the Fivetran dashboard under Account Settings > API Config.
Add FIVETRAN_API_KEY, FIVETRAN_API_SECRET, and SLACK_WEBHOOK as secrets, set connector_id to a real connector, and run the flow once manually to confirm status reads HEALTHY.
Pause the connector in the Fivetran dashboard and run again to confirm status reads WARNING.
Enable connector_freshness_schedule once pointed at a real connector.

How to extend

Pass multiple IDs in connectorIds and change the jq query from to_entries[0].value to iterate every entry, reporting the worst classification across a whole connector fleet in one run.
Chain a CRITICAL branch into a ticketing system with the connector's failedAt timestamp attached for faster triage.
Push status into a KV store and alert only on state transitions instead of every run.
Set wait: true with a short maxDuration instead of false if a specific execution should actively wait for freshness rather than just sampling it.

Inputs

connector_id: the Fivetran connector to check (type STRING).
freshness_buffer: ISO-8601 duration added on top of the connector's own sync frequency before it is considered stale (type STRING, default PT30M).

Outputs

status: one of HEALTHY, WARNING, CRITICAL, UNREACHABLE.
Example path: {{ outputs.classify_connector_health.value }}, also exposed as the flow output {{ outputs.status }}.

Pitfalls

fresh reflects Fivetran's own calculation from syncFrequency plus freshness_buffer; a connector with a long sync interval (e.g. a 24-hour schedule) can look WARNING for most of a normal day if the buffer is set too tight. Tune freshness_buffer against the connector's real schedule before enabling the schedule trigger.
This flow only ever calls the connector status read; it never calls Fivetran's sync-trigger, pause, or resume endpoints, so there is nothing destructive to default to dry-run for.
wait: false is deliberate: setting it to true turns this single scheduled probe into a task that can block for up to maxDuration (default one hour) waiting for freshness, which defeats the point of a lightweight recurring check.
Fivetran API keys are account-wide by default; scope a dedicated key to this check's read-only needs where your Fivetran plan allows it, rather than reusing an administrative key.