Pomerium Identity Proxy Databroker & Session Sync Sentinel

id: pomerium-identity-proxy-databroker-sync-sentinel
namespace: company.security
description: |
  Audits Pomerium Zero Trust identity-aware proxy databroker storage errors and
  identity manager synchronization telemetry on a recurring schedule. Detects
  session directory desync and IdP communication failures, routes alerts to Slack,
  and gates automated incident response before user access outages occur.

triggers:
  - id: periodic_sync_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Periodically audits Pomerium databroker storage errors and
      identity synchronization telemetry. Shipped disabled by default; enable
      once pomerium_host and pomerium_metrics_port point to a running Pomerium
      instance."
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      pomerium_host: "localhost"
      pomerium_metrics_port: 9090
      max_storage_errors_threshold: 0
      max_idp_fetch_errors_threshold: 3
      min_synced_records_threshold: 1

inputs:
  - id: pomerium_host
    type: STRING
    displayName: "Pomerium Host"
    defaults: "localhost"
    description: "Hostname or IP address exposing Pomerium Prometheus operational metrics."

  - id: pomerium_metrics_port
    type: INT
    displayName: "Pomerium Metrics Port"
    defaults: 9090
    description: "TCP port configured for the Pomerium Prometheus metrics listener
      (default: 9090)."

  - id: max_storage_errors_threshold
    type: INT
    displayName: "Max Storage Errors Threshold"
    defaults: 0
    description: "Maximum allowable cumulative databroker storage errors before
      triggering a critical access proxy outage alert."

  - id: max_idp_fetch_errors_threshold
    type: INT
    displayName: "Max IdP Fetch Errors Threshold"
    defaults: 3
    description: "Maximum allowable identity manager user info fetch errors before
      flagging degraded directory synchronization."

  - id: min_synced_records_threshold
    type: INT
    displayName: "Minimum Synced Records Threshold"
    defaults: 1
    description: "Minimum expected count of synchronized identity and route records
      in the databroker state store."

tasks:
  - id: probe_pomerium_metrics
    type: io.kestra.plugin.core.http.Request
    description: "Queries Pomerium Prometheus metrics endpoint to inspect databroker
      state and identity manager health."
    uri: "http://{{ inputs.pomerium_host }}:{{ inputs.pomerium_metrics_port
      }}/metrics"
    method: GET
    headers:
      Accept: "text/plain"
    options:
      allowFailed: true

  - id: analyze_databroker_telemetry
    type: io.kestra.plugin.scripts.python.Script
    description: "Parses Prometheus metrics to extract databroker storage errors,
      IdP fetch failures, and synced record gauges."
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install --quiet kestra
    script: |
      import re
      import sys
      from kestra import Kestra

      metrics_raw = """{{ outputs.probe_pomerium_metrics.body }}"""
      max_storage_errors = {{ inputs.max_storage_errors_threshold }}
      max_idp_fetch_errors = {{ inputs.max_idp_fetch_errors_threshold }}
      min_synced_records = {{ inputs.min_synced_records_threshold }}

      databroker_storage_errors = 0.0
      idp_fetch_errors = 0.0
      synced_records = 0.0

      for line in metrics_raw.splitlines():
          line = line.strip()
          if not line or line.startswith('#'):
              continue
          m = re.match(r'^([a-zA-Z_:][a-zA-Z0-9_:]*)\s*(?:\{[^}]*\})?\s+([0-9eE\.\+\-]+)', line)
          if m:
              metric_name, val_str = m.group(1), m.group(2)
              try:
                  val = float(val_str)
              except ValueError:
                  continue
              if 'pomerium_databroker_storage_errors_total' in metric_name or 'databroker_storage_errors_total' in metric_name:
                  databroker_storage_errors += val
              elif 'pomerium_identity_manager_user_info_fetch_errors_total' in metric_name or 'user_info_fetch_errors_total' in metric_name:
                  idp_fetch_errors += val
              elif 'pomerium_databroker_synced_records' in metric_name or 'databroker_records' in metric_name:
                  synced_records = max(synced_records, val)

      status = "HEALTHY"
      summary = "Pomerium Databroker and Identity Provider synchronization operating nominally."

      if databroker_storage_errors > max_storage_errors:
          status = "CRITICAL_DATABROKER_ERROR"
          summary = f"CRITICAL: Pomerium Databroker storage errors ({int(databroker_storage_errors)}) exceeded threshold ({max_storage_errors}). Edge nodes failing to synchronize session state."
      elif idp_fetch_errors > max_idp_fetch_errors:
          status = "IDP_SYNC_WARNING"
          summary = f"WARNING: Identity Manager user info fetch errors ({int(idp_fetch_errors)}) exceeded threshold ({max_idp_fetch_errors}). Directory synchronization degraded."
      elif synced_records < min_synced_records:
          status = "IDP_SYNC_WARNING"
          summary = f"WARNING: Databroker synced record count ({int(synced_records)}) is below expected minimum ({min_synced_records}). Possible IdP desync or empty session directory."

      Kestra.outputs({
          "status": status,
          "storage_errors": databroker_storage_errors,
          "idp_fetch_errors": idp_fetch_errors,
          "synced_records": synced_records,
          "summary": summary
      })

  - id: log_telemetry_summary
    type: io.kestra.plugin.core.log.Log
    description: "Logs evaluated Pomerium telemetry and state classification."
    message: "Pomerium telemetry audit completed. Status: {{
      outputs.analyze_databroker_telemetry.vars.status }} | Storage Errors: {{
      outputs.analyze_databroker_telemetry.vars.storage_errors }} | IdP Fetch
      Errors: {{ outputs.analyze_databroker_telemetry.vars.idp_fetch_errors }} |
      Synced Records: {{
      outputs.analyze_databroker_telemetry.vars.synced_records }}"

  - id: classify_health_status
    type: io.kestra.plugin.core.flow.Switch
    description: "Evaluates databroker health status and dispatches targeted alerts
      or healthy confirmation."
    value: "{{ outputs.analyze_databroker_telemetry.vars.status }}"
    cases:
      CRITICAL_DATABROKER_ERROR:
        - id: alert_critical_databroker_outage
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Dispatches critical Slack alert when Pomerium databroker storage
            errors exceed threshold."
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "#security-ops",
              "text": ":rotating_light: *Pomerium Databroker Storage Failure Detected*",
              "blocks": [
                {
                  "type": "header",
                  "text": {
                    "type": "plain_text",
                    "text": ":rotating_light: Pomerium Databroker Critical Failure",
                    "emoji": true
                  }
                },
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": "*Cluster:* `{{ inputs.pomerium_host }}:{{ inputs.pomerium_metrics_port }}`\n*Status:* `CRITICAL_DATABROKER_ERROR`\n*Storage Errors:* `{{ outputs.analyze_databroker_telemetry.vars.storage_errors }}` (Threshold: `{{ inputs.max_storage_errors_threshold }}`)\n\n*Impact:* Edge proxies cannot persist or replicate session states. Users may experience sudden authentication drops or authorization denial."
                  }
                },
                {
                  "type": "context",
                  "elements": [
                    {
                      "type": "mrkdwn",
                      "text": "Kestra Execution: `{{ execution.id }}` | Namespace: `{{ flow.namespace }}`"
                    }
                  ]
                }
              ]
            }

        - id: fail_on_critical_databroker_outage
          type: io.kestra.plugin.core.execution.Fail
          description: "Fails execution to trigger downstream on-call incident workflows
            and escalation."
          errorMessage: "{{ outputs.analyze_databroker_telemetry.vars.summary }}"

      IDP_SYNC_WARNING:
        - id: alert_idp_sync_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Dispatches warning Slack alert when Pomerium identity manager
            encounters fetch errors."
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "#security-ops",
              "text": ":warning: *Pomerium Identity Provider Synchronization Warning*",
              "blocks": [
                {
                  "type": "header",
                  "text": {
                    "type": "plain_text",
                    "text": ":warning: Pomerium IdP Sync Degraded",
                    "emoji": true
                  }
                },
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": "*Cluster:* `{{ inputs.pomerium_host }}:{{ inputs.pomerium_metrics_port }}`\n*Status:* `IDP_SYNC_WARNING`\n*IdP Fetch Errors:* `{{ outputs.analyze_databroker_telemetry.vars.idp_fetch_errors }}` (Threshold: `{{ inputs.max_idp_fetch_errors_threshold }}`)\n*Synced Records:* `{{ outputs.analyze_databroker_telemetry.vars.synced_records }}`\n\n*Recommendation:* Verify upstream Identity Provider API rate limits, OAuth tokens, and network egress from the Pomerium control plane."
                  }
                },
                {
                  "type": "context",
                  "elements": [
                    {
                      "type": "mrkdwn",
                      "text": "Kestra Execution: `{{ execution.id }}` | Namespace: `{{ flow.namespace }}`"
                    }
                  ]
                }
              ]
            }

      HEALTHY:
        - id: log_healthy_state
          type: io.kestra.plugin.core.log.Log
          description: "Logs healthy status confirmation when all Pomerium components
            operate within nominal thresholds."
          message: "Pomerium databroker storage and identity manager synchronization are
            nominal (Storage Errors: {{
            outputs.analyze_databroker_telemetry.vars.storage_errors }}, IdP
            Fetch Errors: {{
            outputs.analyze_databroker_telemetry.vars.idp_fetch_errors }},
            Synced Records: {{
            outputs.analyze_databroker_telemetry.vars.synced_records }})."

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: "Finalizes execution and records audit log for security compliance
      verification."
    message: "Audit complete for Pomerium instance at {{ inputs.pomerium_host }}:{{
      inputs.pomerium_metrics_port }}. Evaluated state: {{
      outputs.analyze_databroker_telemetry.vars.status }}. Summary: {{
      outputs.analyze_databroker_telemetry.vars.summary }}"

errors:
  - id: alert_sentinel_probe_failed
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Dispatches urgent Slack notification if the Pomerium metrics
      endpoint is unreachable or encounters HTTP failure."
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "#security-ops",
        "text": ":rotating_light: *Pomerium Sentinel Probe Failure*",
        "blocks": [
          {
            "type": "header",
            "text": {
              "type": "plain_text",
              "text": ":rotating_light: Pomerium Unreachable",
              "emoji": true
            }
          },
          {
            "type": "section",
            "text": {
              "type": "mrkdwn",
              "text": "*Failed Target:* `http://{{ inputs.pomerium_host }}:{{ inputs.pomerium_metrics_port }}/metrics`\n*Error:* The Pomerium metrics endpoint failed to respond. The identity proxy or metrics listener may be down, misconfigured, or unreachable from the Kestra worker network."
            }
          },
          {
            "type": "context",
            "elements": [
              {
                "type": "mrkdwn",
                "text": "Kestra Execution: `{{ execution.id }}` | Flow: `{{ flow.id }}`"
              }
            ]
          }
        ]
      }

Continuously audits Pomerium Zero Trust identity-aware proxy deployments to detect databroker storage errors, monitor identity provider (IdP) directory synchronization health, verify synced record counts, and dispatch actionable Slack alerts before enterprise users encounter access outages.

How it works

periodic_sync_audit (io.kestra.plugin.core.trigger.Schedule) triggers every 15 minutes with configurable thresholds and connection parameters. It ships disabled by default so initial connectivity can be verified manually.
probe_pomerium_metrics (io.kestra.plugin.core.http.Request) queries the Pomerium Prometheus metrics endpoint (GET /metrics) to fetch real-time operational telemetry across all proxy subsystems.
analyze_databroker_telemetry (io.kestra.plugin.scripts.python.Script) parses the Prometheus text exposition stream to extract pomerium_databroker_storage_errors_total, pomerium_identity_manager_user_info_fetch_errors_total, and pomerium_databroker_synced_records. It classifies cluster health into CRITICAL_DATABROKER_ERROR, IDP_SYNC_WARNING, or HEALTHY.
log_telemetry_summary (io.kestra.plugin.core.log.Log) outputs a structured telemetry overview into the execution audit trail.
classify_health_status (io.kestra.plugin.core.flow.Switch) evaluates the calculated status:
- CRITICAL_DATABROKER_ERROR: Dispatches an urgent alert to #security-ops via io.kestra.plugin.slack.notifications.SlackIncomingWebhook highlighting storage persistence failures that risk session drops, then terminates via io.kestra.plugin.core.execution.Fail to trigger on-call escalation.
- IDP_SYNC_WARNING: Dispatches a Slack notification warning security engineers that directory synchronization or user info refresh is degraded while access proxy routing remains functional.
- HEALTHY: Confirms zero storage errors and positive synchronized record counts via io.kestra.plugin.core.log.Log.
log_audit_complete (io.kestra.plugin.core.log.Log) writes a final immutable compliance record documenting flow parameters and resulting cluster state.
The global errors handler catches connection timeouts or unreachable hosts to dispatch an urgent out-of-band Slack notification when Pomerium is down.

What you get

Early warning of distributed databroker storage failures before edge proxies drop active user sessions or deny policy authorization.
Continuous validation of Identity Provider (OIDC, Google, Okta, Azure AD, GitHub) directory refresh calls to catch expired service account tokens or rate limit exhaustion.
Automated severity-based triage routing critical databroker outages to incident response channels while logging routine identity warnings separately.
Fully automated immutable execution audit history in Kestra for Zero Trust security compliance and SOC 2 access reviews.

Who it's for

Security Engineers and DevSecOps teams responsible for Zero Trust network access (ZTNA) and identity-aware reverse proxies.
Platform and Site Reliability Engineers managing corporate single sign-on (SSO) gateways and internal developer portals.
Systems Administrators auditing directory synchronization stability across distributed identity infrastructure.

Why orchestrate this with Kestra

Pomerium does not include native threshold-based alerting or team collaboration webhooks out of the box. Relying on raw cron scripts or manual dashboard checks creates visibility gaps that surface only after employees are locked out of internal tools. Kestra delivers scheduled execution, declarative error handling, secure credential masking, and automated multi-branch alerting in a single version-controlled workflow.

Prerequisites

A running Pomerium instance (open-source or enterprise) exposing its Prometheus metrics listener. Pomerium provides an official open-source Docker image (pomerium/pomerium:latest) that runs locally without commercial licensing.
A Slack incoming webhook URL configured for security alerts.

Secrets

SLACK_WEBHOOK_URL: Incoming webhook URL used by alert_critical_databroker_outage, alert_idp_sync_warning, and alert_sentinel_probe_failed to send notifications to #security-ops.

Quick start

Start a local Pomerium container or mock metrics endpoint:

docker run -d --name pomerium-sentinel-test -p 9090:9090 pomerium/pomerium:latest

Add the SLACK_WEBHOOK_URL secret to your Kestra namespace.
Import this flow into Kestra and trigger an execution with default inputs.
Confirm that probe_pomerium_metrics succeeds and log_audit_complete confirms nominal status.
Test alerting by triggering an execution with max_storage_errors_threshold set to -1 to simulate a critical databroker incident.
Enable periodic_sync_audit to schedule continuous 15-minute background auditing.

How to extend

Add an automated remediation branch to restart the Pomerium databroker container via io.kestra.plugin.docker.Run or reload Kubernetes deployments via io.kestra.plugin.kubernetes.pods.Restart when critical storage errors persist.
Forward historical telemetry metrics to Prometheus Pushgateway or Datadog for long-term SSO trend analysis.
Integrate with PagerDuty via io.kestra.plugin.pagerduty.notifications.PagerDutyAlert to trigger high-priority paging during full proxy outages.

Pomerium Identity-Aware Proxy Databroker and Session Sync Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Pomerium Identity-Aware Proxy Databroker and Session Sync Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Fastly Month-to-Date Cost Guard

Faktory Job Enqueue Path Reachability Sentinel

Fastly Daily Traffic Digest to Slack

Feature-Flag Progressive Delivery with an Error-Rate Guard

Fastly Edge Error Spike Alert with the Stats Trigger

Falco Cloud Native Runtime Threat and Security Priority Event Sentinel

New to Kestra?