Cloud Tasks Queue Backlog and Failure Gate with Kestra

id: cloudtasks-queue-backlog-failure-gate
namespace: company.team

description: |
  Read a Cloud Tasks queue's real queue/depth and task_attempt_count Cloud Monitoring metrics,
  classify the queue HEALTHY, WARNING, CRITICAL or PROBE_FAILED from the actual values
  returned, alert Slack with those values, and log a final audit summary either way.

inputs:
  - id: queue_id
    type: STRING
    displayName: Cloud Tasks queue ID
    description: Name of the queue to audit. Used in the Cloud Monitoring filter's
      resource.labels.queue_id.
    defaults: "default-queue"

  - id: location
    type: STRING
    displayName: Queue location
    description: Region the queue runs in, for example us-central1. Used in the
      Cloud Monitoring filter's resource.labels.location.
    defaults: "us-central1"

  - id: window
    type: STRING
    displayName: Lookback window
    description: ISO-8601 duration looked back from now for both metrics. Maps to
      the Query task's window field.
    defaults: "PT10M"

  - id: queue_depth_critical_threshold
    type: INT
    displayName: Queue depth critical threshold
    description: queue/depth value at or above which the queue is classified
      CRITICAL for backlog.
    defaults: 1000

  - id: failed_attempt_critical_threshold
    type: INT
    displayName: Failed attempt count threshold
    description: Total non-ok task_attempt_count over the window at or above which
      the queue is classified CRITICAL for dispatch failures.
    defaults: 10

  - id: alert_on
    type: SELECT
    displayName: Signal to gate on
    description: Which metric drives the CRITICAL classification. depth only watches
      backlog size, failures only watches failed dispatch attempts, both watches
      either.
    values:
      - "depth"
      - "failures"
      - "both"
    defaults: "both"

tasks:
  - id: probe_queue_depth
    type: io.kestra.plugin.gcp.monitoring.Query
    description: Read-only. The queue/depth gauge metric, the number of tasks
      currently in the queue.
    allowFailure: true
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"
    projectId: "{{ secret('GCP_PROJECT_ID') }}"
    filter: "metric.type=\"cloudtasks.googleapis.com/queue/depth\" AND
      resource.type=\"cloud_tasks_queue\" AND resource.labels.queue_id=\"{{
      inputs.queue_id }}\" AND resource.labels.location=\"{{ inputs.location
      }}\""
    window: "{{ inputs.window }}"

  - id: probe_failed_attempts
    type: io.kestra.plugin.gcp.monitoring.Query
    description: Read-only. The task_attempt_count delta metric filtered to non-ok
      response codes, over the window.
    allowFailure: true
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT') }}"
    projectId: "{{ secret('GCP_PROJECT_ID') }}"
    filter: "metric.type=\"cloudtasks.googleapis.com/queue/task_attempt_count\" AND
      resource.type=\"cloud_tasks_queue\" AND resource.labels.queue_id=\"{{
      inputs.queue_id }}\" AND resource.labels.location=\"{{ inputs.location
      }}\" AND metric.labels.response_code!=\"ok\""
    window: "{{ inputs.window }}"

  - id: extract_metric_values
    type: io.kestra.plugin.core.output.OutputValues
    description: Pulls the latest queue/depth point and sums every non-ok
      task_attempt_count point, both as real numbers from the Cloud Monitoring
      response, not an estimate. Empty values when a probe did not return.
    values:
      queue_depth: >-
        {{ (outputs.probe_queue_depth is not defined or
        outputs.probe_queue_depth.series is not defined or
        (outputs.probe_queue_depth.series | length) == 0) ? '0' :
        (outputs.probe_queue_depth.series |
        jq('(.[0]?.points[0]?.value?.int64Value // 0) | tonumber') | first) }}
      failed_attempts: >-
        {{ (outputs.probe_failed_attempts is not defined or
        outputs.probe_failed_attempts.series is not defined or
        (outputs.probe_failed_attempts.series | length) == 0) ? '0' :
        (outputs.probe_failed_attempts.series |
        jq('([.[].points[]?.value.int64Value | tonumber] | add) // 0') | first)
        }}

  - id: classify_queue_health
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when either probe produced no output at all, CRITICAL
      when the signal selected by alert_on is at or above its threshold, HEALTHY
      otherwise.
    format: "{{ (outputs.probe_queue_depth is not defined or
      outputs.probe_queue_depth.series is not defined or
      outputs.probe_failed_attempts is not defined or
      outputs.probe_failed_attempts.series is not defined) ? 'PROBE_FAILED' :
      ((((inputs.alert_on == 'depth' or inputs.alert_on == 'both') and
      fromJson(outputs.extract_metric_values.values.queue_depth) >=
      inputs.queue_depth_critical_threshold) or ((inputs.alert_on == 'failures'
      or inputs.alert_on == 'both') and
      fromJson(outputs.extract_metric_values.values.failed_attempts) >=
      inputs.failed_attempt_critical_threshold)) ? 'CRITICAL' : 'HEALTHY') }}"

  - id: route_by_queue_health
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a healthy queue. Branch B alerts Slack when the queue
      is backlogged, dispatch attempts are failing, or the probe itself failed.
    value: "{{ outputs.classify_queue_health.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Cloud Tasks queue {{ inputs.queue_id }} healthy. depth={{
            outputs.extract_metric_values.values.queue_depth }}
            failed_attempts={{
            outputs.extract_metric_values.values.failed_attempts }} over {{
            inputs.window }}."
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Cloud Tasks queue " ~ inputs.queue_id ~ " (" ~ inputs.location ~ ") CRITICAL over the last " ~ inputs.window ~ ". depth=" ~ outputs.extract_metric_values.values.queue_depth ~ " (threshold " ~ inputs.queue_depth_critical_threshold ~ "). failed_attempts=" ~ outputs.extract_metric_values.values.failed_attempts ~ " (threshold " ~ inputs.failed_attempt_critical_threshold ~ ")."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Cloud Tasks queue probe for " ~ inputs.queue_id ~ " (" ~ inputs.location ~ ") did not return a result from one or both Cloud Monitoring queries. Check GCP_SERVICE_ACCOUNT, GCP_PROJECT_ID, and that the queue exists in that location."} | toJson }}
    defaults:
      - id: log_unknown_status
        type: io.kestra.plugin.core.log.Log
        message: "Cloud Tasks queue status unhandled: {{
          outputs.classify_queue_health.value ?? 'unknown' }}."

  - 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: "Cloud Tasks queue backlog and failure gate completed. queue={{
      inputs.queue_id }} status={{ outputs.classify_queue_health.value }}
      alert_on={{ inputs.alert_on }} window={{ inputs.window }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "Cloud Tasks queue backlog and failure 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. Match the cron interval to window for
      non-overlapping coverage.
    cron: "*/10 * * * *"
    disabled: true
    inputs:
      queue_id: "default-queue"
      location: "us-central1"
      window: "PT10M"
      queue_depth_critical_threshold: 1000
      failed_attempt_critical_threshold: 10
      alert_on: "both"

A Cloud Tasks queue backs up for one of two reasons: the handler it dispatches to is down or too slow, so task_attempt_count fills up with non-ok response codes, or the handler is fine and the producer side is simply creating tasks faster than they can be drained, so queue/depth climbs with a mostly-ok attempt history. The two causes need different responses and look identical from a glance at the queue. This blueprint reads both real Cloud Monitoring metrics, classifies what it actually finds, and alerts Slack with the exact depth and failure numbers instead of a single ambiguous "queue is big" signal.

How it works

probe_queue_depth and probe_failed_attempts (io.kestra.plugin.gcp.monitoring.Query) each query one cloudtasks.googleapis.com metric scoped to queue_id and location over window, with allowFailure: true so a credentials or project problem does not stop the flow before it can be classified as PROBE_FAILED below.
- probe_queue_depth reads the queue/depth gauge metric, the number of tasks currently in the queue.
- probe_failed_attempts reads the queue/task_attempt_count delta metric, filtered to response_code!="ok", so only failed dispatch attempts count.
extract_metric_values (io.kestra.plugin.core.output.OutputValues) pulls the real number out of each query's raw Cloud Monitoring JSON: the most recent point's value.int64Value for queue_depth, and the sum of every matching point's value.int64Value for failed_attempts (a delta metric needs summing across the window, a gauge only needs its latest point). Both default to 0 if the matching probe produced output but found no time series at all, which Cloud Monitoring returns for a genuinely empty or healthy queue, not an error.
classify_queue_health (io.kestra.plugin.core.debug.Return) returns PROBE_FAILED when either probe produced no output at all (guarded with is not defined), CRITICAL when the signal selected by alert_on meets its own threshold, HEALTHY otherwise.
route_by_queue_health (io.kestra.plugin.core.flow.Switch) branches on that status: HEALTHY logs the real depth and failure numbers, CRITICAL and PROBE_FAILED each alert Slack with those same real numbers, never an invented count.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real distinction between a queue that is backlogged and one whose dispatches are failing, from the same two Cloud Monitoring metrics Google itself documents for this purpose.
A Slack alert carrying the exact queue/depth and task_attempt_count values that triggered it, never an invented number.
A choice of which signal matters for a given queue (alert_on), a queue with a bursty producer may only care about depth, one calling a flaky downstream handler may only care about failed attempts.

Who it's for

Backend and platform teams running Cloud Tasks queues who need to know the difference between "the producer is outrunning the consumer" and "the consumer is down", not just that the queue looks unhealthy.
Teams evaluating the GCP Monitoring plugin who want a working example of gcp.monitoring.Query against a real Google Cloud metric namespace with resource and metric label filters, including how to extract a value out of the raw time series JSON.

Why orchestrate this with Kestra

Cloud Monitoring can fire a native alerting policy, but turning that into a message that says "which queue, how deep, how many failures, over what window" without also wiring an additional notification channel is its own project. Kestra adds the schedule, the single declarative query against both metrics, the classification that extracts the real numbers out of the raw time series response, and a Slack alert built from them, all as one YAML file you can version and review like code next to the rest of your data pipelines.

Prerequisites

A Cloud Tasks queue and a GCP service account with monitoring.timeSeries.list permission (roles/monitoring.viewer is enough, this blueprint only reads).
A Slack incoming webhook for alerts.
This plugin calls the Cloud Monitoring API directly from the Kestra worker process, there is no sub-container taskRunner and no Docker network to join, any machine Kestra runs on that can reach the Google Cloud Monitoring API works. Cloud Tasks and Cloud Monitoring are both managed Google Cloud services with no self-hostable open-source emulator for either.

Secrets

GCP_SERVICE_ACCOUNT, GCP_PROJECT_ID: used by both io.kestra.plugin.gcp.monitoring.Query tasks.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set all three as base64-encoded environment variables: SECRET_GCP_SERVICE_ACCOUNT, SECRET_GCP_PROJECT_ID, SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`queue_id`	STRING	`default-queue`	any Cloud Tasks queue ID
`location`	STRING	`us-central1`	any GCP region running Cloud Tasks
`window`	STRING	`PT10M`	any ISO-8601 duration
`queue_depth_critical_threshold`	INT	`1000`	any non-negative integer
`failed_attempt_critical_threshold`	INT	`10`	any non-negative integer
`alert_on`	SELECT	`both`	`depth`, `failures`, `both`

Outputs

{{ outputs.extract_metric_values.values.queue_depth }}: most recent queue depth count (0 if no time series returned).
{{ outputs.extract_metric_values.values.failed_attempts }}: sum of non-ok task attempt failures across window (0 if no time series returned).
{{ outputs.classify_queue_health.value }}: health classification (HEALTHY, CRITICAL, or PROBE_FAILED).

Quick start

Add the GCP and Slack secrets.
Set queue_id and location to a real queue and run this flow manually.
Read log_audit_summary for the real depth and failed-attempt numbers.
Enable the every_ten_minutes trigger, whose cron matches the default window.

How to extend

Split probe_failed_attempts into one query per specific non-ok response_code (for example UNAVAILABLE versus DEADLINE_EXCEEDED) to tell a slow handler apart from a down one in the alert text.
Read cloudtasks.googleapis.com/queue/task_attempt_delays, a distribution metric, to alert on growing dispatch latency even before attempts start failing outright.
Add a second pair of probes for a second queue and reuse the same classification logic to monitor several queues from one flow.

Pitfalls

queue/depth is a gauge, read its latest point. task_attempt_count is a delta, sum every point across the window, reading only its latest point badly undercounts failures that happened earlier in the window.
Cloud Monitoring can return zero time series for a metric when there has been no activity at all in the window, this blueprint treats that as 0, a healthy empty result, not a reason to alert, confirm that matches your expectations for a queue that is sometimes genuinely idle.
resource.labels.queue_id and resource.labels.location must match exactly, a typo in either silently returns zero time series rather than an error, which this gate reports as HEALTHY with depth=0, not PROBE_FAILED.
This blueprint never pauses, purges, or redirects the queue, there is no remediation task in this flow, it is alert-only by design, Cloud Tasks queue management belongs in the service that owns the queue's configuration.

Cloud Tasks Queue Backlog and Failure 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?

Cloud Tasks Queue Backlog and Failure 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

API Contract Smoke Test with Slack Alerts

Argo CD Application Drift Sentinel

Single-tenant Argo CD rollout with waves, guarded sync strategy, and Status-based readiness gates

Monitor API Uptime in Parallel and Alert Slack with the Exact Status of Each Endpoint

API Rate Limit Observer

Apify Actor Run Failure Sentinel

More Cloud Blueprints

Ansible Scalability, Deploy Configuration Files to Multiple Servers at Scale

AI Prompt Drift and Quality Sentinel

Evaluate RAG Pipelines for Hallucinations and Faithfulness

AWS CloudTrail Unauthorized API Call Sentinel

Detect CloudTrail Management-Event Logging Gaps

High-Performance Computing, Run Parallel Python Workloads on AWS Batch with ECS Fargate

New to Kestra?