OpenMetadata Catalog Health and Change-Event Sentinel

id: openmetadata-catalog-health-and-change-event-sentinel
namespace: company.team

description: |
  Continuously monitors enterprise OpenMetadata catalog platforms by probing system health endpoints, auditing backend storage and search indexer cluster readiness (Elasticsearch/OpenSearch and PostgreSQL/MySQL), validating change-event subscription delivery, classifying operational posture, routing real-time Slack incident alerts, and logging compliance audit records.

inputs:
  - id: openmetadata_host
    type: STRING
    displayName: OpenMetadata Host
    description: Hostname or IP address of the OpenMetadata server instance
      reachable by the Kestra worker.
    defaults: "localhost"

  - id: openmetadata_port
    type: INT
    displayName: OpenMetadata Port
    description: HTTP port of the OpenMetadata REST API service (defaults to 8585).
    defaults: 8585

  - id: openmetadata_scheme
    type: STRING
    displayName: Protocol Scheme
    description: URI protocol scheme (http or https).
    defaults: "http"

  - id: expected_status
    type: STRING
    displayName: Expected System Status
    description: Expected string value of the system operational health attribute
      (defaults to healthy).
    defaults: "healthy"

tasks:
  - id: probe_system_status
    type: io.kestra.plugin.core.http.Request
    description: Queries OpenMetadata /api/v1/system/status to audit core server
      health, metadata database connectivity, and search indexer cluster
      operational state.
    uri: "{{ inputs.openmetadata_scheme }}://{{ inputs.openmetadata_host }}:{{
      inputs.openmetadata_port }}/api/v1/system/status"
    method: GET
    headers:
      Accept: "application/json"
      Authorization: "Bearer {{ secret('OPENMETADATA_JWT_TOKEN') }}"
    options:
      allowFailed: true
    allowFailure: true

  - id: probe_system_version
    type: io.kestra.plugin.core.http.Request
    description: Queries OpenMetadata /api/v1/system/version to verify API gateway
      responsiveness and deployment version consistency.
    uri: "{{ inputs.openmetadata_scheme }}://{{ inputs.openmetadata_host }}:{{
      inputs.openmetadata_port }}/api/v1/system/version"
    method: GET
    headers:
      Accept: "application/json"
      Authorization: "Bearer {{ secret('OPENMETADATA_JWT_TOKEN') }}"
    options:
      allowFailed: true
    allowFailure: true

  - id: probe_event_subscriptions
    type: io.kestra.plugin.core.http.Request
    description: Queries OpenMetadata /api/v1/events/subscriptions to verify that
      change-event notification sinks and webhook subscriptions are active and
      healthy.
    uri: "{{ inputs.openmetadata_scheme }}://{{ inputs.openmetadata_host }}:{{
      inputs.openmetadata_port }}/api/v1/events/subscriptions?limit=10"
    method: GET
    headers:
      Accept: "application/json"
      Authorization: "Bearer {{ secret('OPENMETADATA_JWT_TOKEN') }}"
    options:
      allowFailed: true
    allowFailure: true

  - id: classify_catalog_posture
    type: io.kestra.plugin.core.debug.Return
    description: Evaluates HTTP status codes across system status, version, and
      event subscription probes to classify catalog operational posture.
    format: |
      {{ outputs.probe_system_status.code is not defined or outputs.probe_system_version.code is not defined ? 'INSTANCE_UNREACHABLE' : (outputs.probe_system_status.code != 200 ? 'SYSTEM_UNHEALTHY' : ((outputs.probe_system_status.body | string) contains 'degraded' or (outputs.probe_system_status.body | string) contains 'unhealthy' ? 'SEARCH_OR_DATABASE_DEGRADED' : (outputs.probe_event_subscriptions.code != 200 ? 'EVENT_SUBSCRIPTION_DEGRADED' : 'HEALTHY_COMPLIANT'))) }}

  - id: route_by_posture
    type: io.kestra.plugin.core.flow.Switch
    description: Routes alerts and logs based on the classified OpenMetadata catalog
      operational posture.
    value: "{{ outputs.classify_catalog_posture.value }}"
    cases:
      "HEALTHY_COMPLIANT":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "OpenMetadata catalog at {{ inputs.openmetadata_scheme }}://{{
            inputs.openmetadata_host }}:{{ inputs.openmetadata_port }} is
            HEALTHY_COMPLIANT. System status: {{
            outputs.probe_system_status.code }}. Version: {{
            outputs.probe_system_version.body | json | jq('.version') | first ??
            'N/A' }}. Subscriptions code: {{
            outputs.probe_event_subscriptions.code }}."

      "SEARCH_OR_DATABASE_DEGRADED":
        - id: alert_indexer_degraded_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "OpenMetadata Sentinel CRITICAL ALERT: Search indexer or metadata database degraded on " ~ inputs.openmetadata_host ~ ":" ~ inputs.openmetadata_port ~ "! /api/v1/system/status reported subsystem degradation. OpenSearch/Elasticsearch cluster unready or SQL connection pool exhausted. Data lineage and catalog queries impaired! Execution: " ~ execution.id ~ "."} | toJson }}

        - id: log_indexer_degraded
          type: io.kestra.plugin.core.log.Log
          message: "CRITICAL: OpenMetadata backend storage or search indexer cluster is
            degraded."

      "EVENT_SUBSCRIPTION_DEGRADED":
        - id: alert_event_subscription_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "OpenMetadata Sentinel WARNING: Change-event subscription pipeline degraded on " ~ inputs.openmetadata_host ~ ":" ~ inputs.openmetadata_port ~ "! /api/v1/events/subscriptions returned HTTP " ~ (outputs.probe_event_subscriptions.code ?? 'NONE') ~ ". Webhook dispatchers or event streaming stalled. Downstream metadata sync delayed! Execution: " ~ execution.id ~ "."} | toJson }}

        - id: log_event_subscription
          type: io.kestra.plugin.core.log.Log
          message: "WARNING: OpenMetadata /api/v1/events/subscriptions returned non-200
            code: {{ outputs.probe_event_subscriptions.code }}."

      "SYSTEM_UNHEALTHY":
        - id: alert_system_unhealthy_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "OpenMetadata Sentinel CRITICAL: Core server unhealthy on " ~ inputs.openmetadata_host ~ ":" ~ inputs.openmetadata_port ~ "! /api/v1/system/status returned HTTP " ~ (outputs.probe_system_status.code ?? 'NONE') ~ ". Core API server process failing health checks. Execution: " ~ execution.id ~ "."} | toJson }}

        - id: log_system_unhealthy
          type: io.kestra.plugin.core.log.Log
          message: "CRITICAL: OpenMetadata /api/v1/system/status returned non-200 code: {{
            outputs.probe_system_status.code }}."

      "INSTANCE_UNREACHABLE":
        - id: alert_unreachable_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {{ {"text": "OpenMetadata Sentinel OUTAGE: Cannot reach OpenMetadata at " ~ inputs.openmetadata_scheme ~ "://" ~ inputs.openmetadata_host ~ ":" ~ inputs.openmetadata_port ~ ". Container offline, network partitioned, or reverse-proxy connection refused. Execution: " ~ execution.id ~ "."} | toJson }}

        - id: log_unreachable
          type: io.kestra.plugin.core.log.Log
          message: "OUTAGE: OpenMetadata instance unreachable at {{
            inputs.openmetadata_scheme }}://{{ inputs.openmetadata_host }}:{{
            inputs.openmetadata_port }}."

  - id: audit_summary
    type: io.kestra.plugin.core.output.OutputValues
    description: Exports structured telemetry attributes and endpoint verification
      metrics for compliance audit reporting.
    values:
      target_instance: "{{ inputs.openmetadata_scheme }}://{{ inputs.openmetadata_host
        }}:{{ inputs.openmetadata_port }}"
      status_http_code: "{{ outputs.probe_system_status.code ?? 0 }}"
      system_version: "{{ outputs.probe_system_version.body | json | jq('.version') |
        first ?? 'N/A' }}"
      subscriptions_http_code: "{{ outputs.probe_event_subscriptions.code ?? 0 }}"
      catalog_posture: "{{ outputs.classify_catalog_posture.value }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts on unexpected workflow runtime execution failure.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {"text": "OpenMetadata Sentinel workflow failed unexpectedly during execution " ~ execution.id ~ " in namespace " ~ flow.namespace ~ "."} | toJson }}

triggers:
  - id: hourly_canary
    type: io.kestra.plugin.core.trigger.Schedule
    description: Executes automated OpenMetadata health and change-event pipeline
      audit at the top of every hour. Shipped disabled.
    cron: "0 * * * *"
    disabled: true
    inputs:
      openmetadata_host: "localhost"
      openmetadata_port: 8585
      openmetadata_scheme: "http"
      expected_status: "healthy"

Continuously audits enterprise OpenMetadata data catalog platforms by probing system health endpoints, auditing underlying metadata database and search indexer cluster readiness (Elasticsearch/OpenSearch and PostgreSQL/MySQL), validating change-event subscription delivery, classifying operational posture, and dispatching real-time incident escalation alerts to Slack.

This blueprint was created by Mohit Jeswani.

How it works

probe_system_status (io.kestra.plugin.core.http.Request) queries /api/v1/system/status to audit core application server readiness, metadata database connectivity (PostgreSQL/MySQL), and search indexing cluster health (OpenSearch/Elasticsearch).
probe_system_version (io.kestra.plugin.core.http.Request) queries /api/v1/system/version to verify API gateway routing and deployment release consistency.
probe_event_subscriptions (io.kestra.plugin.core.http.Request) queries /api/v1/events/subscriptions to verify that change-event notification dispatchers, audit webhooks, and real-time metadata streaming pipelines are operational.
classify_catalog_posture (io.kestra.plugin.core.debug.Return) evaluates endpoint response codes and payload attributes using a declarative Pebble conditional expression:
- Classifies as INSTANCE_UNREACHABLE if endpoints cannot be reached over the network.
- Classifies as SYSTEM_UNHEALTHY if /api/v1/system/status returns a non-200 code.
- Classifies as SEARCH_OR_DATABASE_DEGRADED if the response payload reports internal subsystem degradation (such as search index cluster yellow/red state or connection pool exhaustion).
- Classifies as EVENT_SUBSCRIPTION_DEGRADED if change-event subscription endpoints return non-200.
- Classifies as HEALTHY_COMPLIANT when all subsystem health and event pipeline probes succeed.
route_by_posture (io.kestra.plugin.core.flow.Switch) routes execution based on the classification:
- HEALTHY_COMPLIANT: Logs successful audit confirmation with system version and status metrics.
- SEARCH_OR_DATABASE_DEGRADED: Posts an urgent Slack alert warning that the search indexer or metadata database is degraded and catalog queries are impacted.
- EVENT_SUBSCRIPTION_DEGRADED: Alerts Slack that change-event webhooks and downstream metadata sync are stalled.
- SYSTEM_UNHEALTHY / INSTANCE_UNREACHABLE: Dispatches high-priority incident notifications for server or network failure.
audit_summary (io.kestra.plugin.core.output.OutputValues) exports structured metrics capturing endpoint HTTP status codes and catalog operational posture for compliance logging.
errors sends a priority failure notification to Slack if unhandled workflow execution errors occur.

What you get

Automated, continuous validation of critical data governance and metadata catalog infrastructure.
Early warning of OpenSearch/Elasticsearch indexing cluster degradation and database connection pool exhaustion before data analysts experience query stalls.
Active verification of change-event subscriptions so downstream lineage pipelines and data governance webhooks never drop schema events silently.
Contextual Slack alerts detailing exact endpoint failure codes, target hostnames, and affected subsystems.
Auditable execution history satisfying SOC2, ISO27001, and BCBS 239 data management operational controls.

Who it's for

Data Platform Engineers and Analytics Platform Leads managing centralized enterprise data catalogs.
Data Governance Teams and Data Stewards responsible for metadata lineage, glossary enforcement, and PII tagging.
Site Reliability Engineers (SREs) monitoring high-availability data platform control planes.

Why orchestrate this with Kestra

OpenMetadata serves as the single source of truth for an organization's data assets, schema changes, and lineage graphs. When OpenMetadata's search indexer or event dispatchers stall, schema migrations go undetected and data quality alerts fail to propagate. Simple port monitors cannot inspect whether OpenSearch is responsive or whether event subscriptions are actively processing events. Kestra orchestrates multi-stage HTTP inspection, declarative Pebble classification, failure escalation to Slack, and centralized compliance audit trails in a single workflow.

Prerequisites

A running OpenMetadata instance (Docker container or Kubernetes cluster).
An OpenMetadata JWT or Service Bot token for authenticated API inspection.
An incoming Slack webhook URL for incident notifications.

Secrets

OPENMETADATA_JWT_TOKEN: OpenMetadata JWT token or Service Account bot token used for authenticated API inspection.
SLACK_WEBHOOK_URL: Incoming Slack webhook URL used by alert_indexer_degraded_slack, alert_event_subscription_slack, alert_system_unhealthy_slack, alert_unreachable_slack, and alert_flow_failure.

Quick start

Configure OPENMETADATA_JWT_TOKEN and SLACK_WEBHOOK_URL in your Kestra namespace secrets.

Start a local OpenMetadata instance using Docker Compose:

git clone https://github.com/open-metadata/OpenMetadata.git && cd OpenMetadata/docker/development
docker compose up -d

Import this blueprint into Kestra and click Execute with default inputs (openmetadata_host: "localhost").
Confirm that classify_catalog_posture classifies HEALTHY_COMPLIANT and inspect the structured telemetry in audit_summary.
Enable the hourly_canary schedule to automate continuous metadata catalog health monitoring.

How to extend

Add automated data quality test suite auditing by querying /api/v1/dataQuality/testSuites to alert on sudden spikes in failed dataset assertions.
Integrate PagerDuty or Opsgenie alerts for out-of-hours on-call paging during SEARCH_OR_DATABASE_DEGRADED or INSTANCE_UNREACHABLE incidents.
Expand multi-environment monitoring by looping over an array of development, staging, and production OpenMetadata cluster hostnames using io.kestra.plugin.core.flow.ForEach.

OpenMetadata Catalog Health and Change-Event 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?

OpenMetadata Catalog Health and Change-Event 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 Data Blueprints

Dlt to DuckDB Analytics Chain: Ingest, Aggregate, Publish

Dlt Dataset Freshness Monitor

Ingest a REST API into DuckDB with dlt

Build a Python ELT pipeline to extract Google Analytics data and load it into DuckDB with dlt

Load Pipedrive CRM data into BigQuery with a scheduled dlt pipeline

Run a scheduled ELT pipeline in Python: extract Chess.com API data and load it into DuckDB with dltHub

More Infrastructure Blueprints

Dex LDAP User Bind Timeout Sentinel

Blue-Green DNS Cutover on DigitalOcean Domains

Dex IdP SAML Connector Authentication Error Sentinel

Nightly DigitalOcean Droplet Snapshot Backups by Tag

DigitalOcean Account Balance Depletion Sentinel

DigitalOcean Alert Policy Compliance Gate

New to Kestra?