Check that a GTFS-Realtime vehicle feed is fresh and alert when it goes stale

id: gtfs-realtime-feed-health-check
namespace: company.team

inputs:
  - id: feed_url
    type: STRING
    displayName: GTFS-Realtime vehicle positions URL
    description: Public URL of a GTFS-Realtime VehiclePositions feed. The default is
      the MBTA feed, which needs no API key.
    defaults: https://cdn.mbta.com/realtime/VehiclePositions.pb

  - id: max_age_seconds
    type: INT
    displayName: Maximum feed age in seconds
    description: The feed counts as stale when its header timestamp is older than this.
    defaults: 120

  - id: min_vehicles
    type: INT
    displayName: Minimum vehicles with a position
    description: The feed counts as unhealthy when fewer vehicles than this report a
      position.
    defaults: 1

  - id: alert_after
    type: INT
    displayName: Unhealthy checks before alerting
    description: Alert only after this many unhealthy checks in a row, so one slow
      fetch does not page anyone.
    defaults: 2

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack
    description: Post to Slack when the alert threshold is reached. Needs the
      SLACK_WEBHOOK_URL secret.
    defaults: false

  - id: fail_on_alert
    type: BOOL
    displayName: Fail on alert
    description: End the run as FAILED when the alert threshold is reached.
    defaults: false

triggers:
  - id: every_five_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check the feed every five minutes. Riders notice a frozen feed
      within minutes, so check often.
    cron: "*/5 * * * *"

tasks:
  - id: download
    type: io.kestra.plugin.core.http.Download
    description: Fetch the current feed snapshot.
    uri: "{{ inputs.feed_url }}"

  - id: inspect
    type: io.kestra.plugin.scripts.python.Script
    description: Decode the protobuf and check how old the feed is and how many
      vehicles report a position.
    dependencies:
      - kestra
      - gtfs-realtime-bindings
    inputFiles:
      feed.pb: "{{ outputs.download.uri }}"
    script: |
      import time
      from google.transit import gtfs_realtime_pb2
      from kestra import Kestra

      max_age = {{ inputs.max_age_seconds }}
      min_vehicles = {{ inputs.min_vehicles }}

      feed = gtfs_realtime_pb2.FeedMessage()
      with open("feed.pb", "rb") as f:
          feed.ParseFromString(f.read())

      now = int(time.time())
      age = now - feed.header.timestamp if feed.header.timestamp else -1
      vehicles = [e.vehicle for e in feed.entity if e.HasField("vehicle")]
      positioned = sum(1 for v in vehicles if v.HasField("position"))
      stale_vehicles = sum(1 for v in vehicles if v.timestamp and now - v.timestamp > max_age)

      problems = []
      if age < 0:
          problems.append("the feed header has no timestamp")
      elif age > max_age:
          problems.append(f"the feed is {age}s old")
      if positioned < min_vehicles:
          problems.append(f"only {positioned} vehicles report a position")

      Kestra.outputs({
          "healthy": not problems,
          "problems": "; ".join(problems) or "none",
          "feed_age_seconds": age,
          "vehicles": len(vehicles),
          "vehicles_with_position": positioned,
          "stale_vehicles": stale_vehicles,
      })

  - id: previous_streak
    type: io.kestra.plugin.core.kv.Get
    description: Read how many unhealthy checks happened in a row before this one.
    key: "{{ flow.id }}_unhealthy_streak"
    errorOnMissing: false

  - id: health_gate
    type: io.kestra.plugin.core.flow.If
    description: Count unhealthy checks in a row and alert once the streak reaches
      alert_after.
    condition: "{{ outputs.inspect.vars.healthy }}"
    then:
      - id: reset_streak
        type: io.kestra.plugin.core.kv.Set
        description: A healthy check resets the streak.
        key: "{{ flow.id }}_unhealthy_streak"
        kvType: NUMBER
        value: 0

      - id: log_healthy
        type: io.kestra.plugin.core.log.Log
        message: "Feed healthy: {{ outputs.inspect.vars.feed_age_seconds }}s old, {{
          outputs.inspect.vars.vehicles_with_position }} vehicles with a
          position, {{ outputs.inspect.vars.stale_vehicles }} stale vehicle
          reports."
    else:
      - id: bump_streak
        type: io.kestra.plugin.core.kv.Set
        description: Add this check to the unhealthy streak.
        key: "{{ flow.id }}_unhealthy_streak"
        kvType: NUMBER
        value: "{{ (outputs.previous_streak.value ?? 0) + 1 }}"

      - id: alert_gate
        type: io.kestra.plugin.core.flow.If
        description: Alert only when the streak reaches alert_after.
        condition: "{{ (outputs.previous_streak.value ?? 0) + 1 >= inputs.alert_after }}"
        then:
          - id: notify_alert
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.notify_slack }}"
            then:
              - id: slack_alert
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "text": "GTFS-Realtime feed {{ inputs.feed_url }} has been unhealthy for {{ (outputs.previous_streak.value ?? 0) + 1 }} checks in a row: {{ outputs.inspect.vars.problems }}. Execution {{ execution.id }}."
                  }

          - id: fail_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.fail_on_alert }}"
            then:
              - id: fail_run
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "GTFS-Realtime feed unhealthy for {{
                  (outputs.previous_streak.value ?? 0) + 1 }} checks in a row:
                  {{ outputs.inspect.vars.problems }}"
        else:
          - id: log_unhealthy
            type: io.kestra.plugin.core.log.Log
            level: WARN
            message: "Feed unhealthy ({{ outputs.inspect.vars.problems }}), check {{
              (outputs.previous_streak.value ?? 0) + 1 }} of {{
              inputs.alert_after }} before alerting."