Kestra Execution Log Analytics and Incident Error Triage

id: kestra-execution-log-analytics-and-error-triage
namespace: company.team

description: |
  Audit recent execution logs across namespaces, summarize error and warning counts,
  and output an operational triage summary.

triggers:
  - id: every_hour
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scan execution logs hourly for error spikes
    cron: "0 * * * *"

inputs:
  - id: kestra_url
    type: STRING
    displayName: Kestra API URL
    description: Base URL of the Kestra instance
    defaults: "http://localhost:8080"

  - id: namespace
    type: STRING
    displayName: Target Namespace
    description: Namespace prefix to filter logs
    defaults: "company.team"

  - id: min_level
    type: STRING
    displayName: Minimum Log Level
    description: Minimum log severity level to fetch
    defaults: "WARN"

tasks:
  - id: fetch_logs
    type: io.kestra.plugin.core.http.Request
    description: Query recent execution logs from the Kestra REST API
    uri: "{{ inputs.kestra_url }}/api/v1/logs/search"
    method: GET
    params:
      namespace: "{{ inputs.namespace }}"
      minLevel: "{{ inputs.min_level }}"
      size: 50

  - id: triage_summary
    type: io.kestra.plugin.scripts.python.Script
    description: Analyze log severity breakdown and print an operational triage summary
    containerImage: python:3.11-slim
    inputFiles:
      logs.json: "{{ outputs.fetch_logs.body }}"
    script: |
      import json
      from collections import Counter

      with open("logs.json", "r") as f:
          content = f.read().strip()
          data = json.loads(content) if content else {}

      logs = data if isinstance(data, list) else data.get("results", data.get("content", []))
      errors = [l for l in logs if str(l.get("level", "")).upper() == "ERROR"]
      warnings = [l for l in logs if str(l.get("level", "")).upper() == "WARN"]

      print("=== Execution Log Triage Summary ===")
      print(f"Total inspected entries: {len(logs)}")
      print(f"Error count: {len(errors)}")
      print(f"Warning count: {len(warnings)}")

      if errors:
          top_error_flows = Counter(l.get("flowId", "unknown") for l in errors).most_common(3)
          print(f"Top error flows: {top_error_flows}")
          print("Latest error message:", errors[0].get("message", "N/A"))
      else:
          print("No ERROR level logs found in the sampled window.")

Audit execution log telemetry, calculate error ratios, and generate actionable triage summaries for platform operations.

This blueprint queries Kestra's execution log API for recent WARN and ERROR events, computes error counts, and logs an operational triage summary directly in Kestra.

How it works

The every_hour schedule triggers regular log health scans.
fetch_logs queries /api/v1/logs/search for recent warnings and errors in the target namespace.
triage_summary inspects the response, counts error and warning entries, identifies offending flows, and outputs the operational summary.

What you get

Automated audit of recent error and warning logs across executions.
Identification of top error-producing workflows.
Fast triage context without manual log scanning.

Who it's for

Platform and DevOps engineers monitoring Kestra operational health.
Workflow authors needing quick visibility into recent execution failures.

Prerequisites

A running Kestra instance with API access (no external secrets required).

Quick start

Import the blueprint into your namespace.
Set inputs.kestra_url if your Kestra API is not at http://localhost:8080.
Execute manually or let the hourly schedule run.

How to extend

To visualize these metrics in real time with interactive charts, you can paste the following dashboard YAML directly into Kestra's Dashboards UI. It covers the Logs and LogsKPI KPIs (error and warning density), TimeSeries (incidents over time by flow), Bar charts (ranked error counts and noisy flows), and the recent incident triage Table:

id: execution_log_analytics_and_error_triage
title: Execution Log Analytics & Error Triage
description: Real-time monitoring of execution log severity distributions, error density ratios, top noisy flows, and live incident triage streams
timeWindow:
  defaults: P7D
  max: P365D

charts:
  - id: kpi_error_log_density
    type: io.kestra.plugin.core.dashboard.chart.KPI
    chartOptions:
      displayName: Error Log Density
      description: Percentage of all emitted log events that are at ERROR severity. Compare against your own baseline to detect anomalous error spikes.
      numberType: PERCENTAGE
      width: 6
    data:
      type: io.kestra.plugin.core.dashboard.data.LogsKPI
      columns:
        field: LEVEL
        agg: COUNT
      numerator:
        - field: LEVEL
          type: EQUAL_TO
          value: ERROR

  - id: kpi_warn_log_density
    type: io.kestra.plugin.core.dashboard.chart.KPI
    chartOptions:
      displayName: Warning Log Density
      description: Percentage of log events at WARN severity. Compare against your own baseline to monitor potential degradation before hard pipeline failures occur.
      numberType: PERCENTAGE
      width: 6
    data:
      type: io.kestra.plugin.core.dashboard.data.LogsKPI
      columns:
        field: LEVEL
        agg: COUNT
      numerator:
        - field: LEVEL
          type: EQUAL_TO
          value: WARN

  - id: errors_over_time_by_flow
    type: io.kestra.plugin.core.dashboard.chart.TimeSeries
    chartOptions:
      displayName: Error and Warning Incidents Over Time
      description: Chronological volume of ERROR and WARN logs broken down by flow. Compare against your own baseline to identify when incidents started and which flows were impacted.
      legend:
        enabled: true
      column: date
      colorByColumn: flowId
    data:
      type: io.kestra.plugin.core.dashboard.data.Logs
      columns:
        date:
          field: DATE
          displayName: Timestamp
        flowId:
          field: FLOW_ID
          displayName: Flow
        total:
          displayName: Incident Logs
          agg: COUNT
          graphStyle: BARS
      where:
        - field: LEVEL
          type: IN
          values:
            - ERROR
            - WARN

  - id: errors_by_flow_bar
    type: io.kestra.plugin.core.dashboard.chart.Bar
    chartOptions:
      displayName: Error Log Count by Flow
      description: Ranking of flows generating the highest count of ERROR log lines. Compare against your own baseline to isolate flows driving platform instability.
      legend:
        enabled: false
      column: flowId
    data:
      type: io.kestra.plugin.core.dashboard.data.Logs
      columns:
        flowId:
          field: FLOW_ID
          displayName: Flow ID
        total:
          field: LEVEL
          displayName: Error Count
          agg: COUNT
      where:
        - field: LEVEL
          type: EQUAL_TO
          value: ERROR
      orderBy:
        - column: total
          order: DESC

  - id: top_noisy_flows_bar
    type: io.kestra.plugin.core.dashboard.chart.Bar
    chartOptions:
      displayName: Top Noisy Flows (Total Log Volume)
      description: Flows generating the highest aggregate volume of log lines across all severities. Compare against your own baseline to detect runaway verbosity or tight polling loops.
      legend:
        enabled: true
      column: flowId
      colorByColumn: level
    data:
      type: io.kestra.plugin.core.dashboard.data.Logs
      columns:
        flowId:
          field: FLOW_ID
          displayName: Flow ID
        level:
          field: LEVEL
        total:
          displayName: Log Entries
          agg: COUNT
      orderBy:
        - column: total
          order: DESC

  - id: recent_error_warn_logs_table
    type: io.kestra.plugin.core.dashboard.chart.Table
    chartOptions:
      displayName: Recent Incident Triage Log Stream
      description: Real-time stream of latest ERROR and WARN log entries across all namespaces. Inspect message, task, and flow details for active incident triage.
    data:
      type: io.kestra.plugin.core.dashboard.data.Logs
      columns:
        date:
          field: DATE
          displayName: Timestamp
        level:
          field: LEVEL
          displayName: Severity
        namespace:
          field: NAMESPACE
          displayName: Namespace
        flowId:
          field: FLOW_ID
          displayName: Flow
        taskId:
          field: TASK_ID
          displayName: Task
        message:
          field: MESSAGE
          displayName: Message
      where:
        - field: LEVEL
          type: IN
          values:
            - ERROR
            - WARN
      orderBy:
        - column: date
          order: DESC

You can also extend the flow itself:

Add notifications via Slack or email when error counts exceed a critical threshold.
Write error metrics to external monitoring databases like DuckDB or Postgres.