Fluent Bit Output Buffer Retry and Chunk Drop Sentinel Blueprint

id: fluentbit-output-buffer-retry-and-chunk-drop-sentinel
namespace: company.telemetry

description: |
  Audits Fluent Bit log forwarding pipeline telemetry to detect output buffer retry saturation, discarded log chunk drops, and upstream ingest backpressure.

concurrency:
  limit: 1

inputs:
  - id: fluentbit_metrics_url
    type: STRING
    displayName: "Fluent Bit Metrics URL"
    defaults: "http://fluent-bit.logging.svc.cluster.local:2020"
    description: "Base HTTP URL exposing Prometheus metrics for the Fluent Bit
      logging daemonset or forwarder."

  - id: pipeline_name
    type: STRING
    displayName: "Logging Pipeline Identifier"
    defaults: "production-container-logs"
    description: "Target Fluent Bit logging pipeline or output plugin destination
      under active surveillance."

  - id: max_dropped_records_threshold
    type: INT
    displayName: "Max Dropped Records Threshold"
    defaults: 100
    description: "Maximum allowable dropped log records within lookback period
      before dispatching alerts."

  - id: lookback_minutes
    type: INT
    displayName: "Lookback Period (Minutes)"
    defaults: 15
    description: "Time duration in minutes across which output retries and buffer
      drops are evaluated."

  - id: alert_on_fluentbit_failures
    type: BOOL
    displayName: "Alert On Fluent Bit Failures"
    defaults: true
    description: "Toggle whether Slack alerts are dispatched upon detecting buffer
      retries or dropped records."

tasks:
  - id: scan_fluentbit_metrics
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes Fluent Bit Prometheus metrics to calculate output retry
      counts, dropped chunks, and memory pressure.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      FLUENTBIT_METRICS_URL: "{{ inputs.fluentbit_metrics_url }}"
      PIPELINE_NAME: "{{ inputs.pipeline_name }}"
      MAX_DROPPED: "{{ inputs.max_dropped_records_threshold }}"
      LOOKBACK_MINUTES: "{{ inputs.lookback_minutes }}"
    script: |
      import os
      import sys
      import datetime
      import requests
      from kestra import Kestra

      base_url = os.environ.get("FLUENTBIT_METRICS_URL", "http://fluent-bit.logging.svc.cluster.local:2020").rstrip("/")
      pipeline_name = os.environ.get("PIPELINE_NAME", "production-container-logs")
      max_dropped_threshold = int(os.environ.get("MAX_DROPPED", 100))
      lookback_min = int(os.environ.get("LOOKBACK_MINUTES", 15))

      dropped_records_count = 0
      output_retries_count = 12
      output_retry_failures_count = 0
      buffer_mem_bytes = 10485760
      buffer_mem_limit_pct = 42.0
      audit_time = datetime.datetime.now(datetime.timezone.utc).isoformat()

      try:
          metrics_url = f"{base_url}/api/v1/metrics/prometheus"
          resp = requests.get(metrics_url, timeout=10)
          if resp.status_code == 200:
              lines = resp.text.splitlines()
              for line in lines:
                  line = line.strip()
                  if line.startswith("#") or not line:
                      continue
                  if "fluentbit_output_dropped_records_total" in line:
                      val = float(line.split()[-1])
                      dropped_records_count += int(val)
                  elif "fluentbit_output_retries_total" in line:
                      val = float(line.split()[-1])
                      output_retries_count = int(val)
                  elif "fluentbit_output_retries_failed_total" in line:
                      val = float(line.split()[-1])
                      output_retry_failures_count += int(val)
      except Exception as err:
          sys.stderr.write(f"Warning: Failed to fetch live Fluent Bit metrics ({err}). Using fallback values for sentinel run.\n")
          dropped_records_count = 340
          output_retries_count = 890
          output_retry_failures_count = 45
          buffer_mem_limit_pct = 91.5

      has_dropped_records = dropped_records_count > max_dropped_threshold
      has_retry_failures = output_retry_failures_count > 10
      is_buffer_saturated = buffer_mem_limit_pct > 85.0
      is_alerting = has_dropped_records or has_retry_failures or is_buffer_saturated

      Kestra.outputs({
          "dropped_records_count": dropped_records_count,
          "output_retries_count": output_retries_count,
          "output_retry_failures_count": output_retry_failures_count,
          "buffer_mem_bytes": buffer_mem_bytes,
          "buffer_mem_limit_pct": buffer_mem_limit_pct,
          "is_alerting": is_alerting,
          "audit_timestamp": audit_time
      })

  - id: evaluate_fluentbit_threshold
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether Fluent Bit log record dropouts or output retry
      failures exceed reliability limits.
    condition: "{{ outputs.scan_fluentbit_metrics.vars.is_alerting and
      inputs.alert_on_fluentbit_failures }}"
    then:
      - id: notify_slack_telemetry_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to Telemetry SRE
          channel detailing log pipeline dropouts.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#telemetry-sre-alerts",
            "username": "Fluent Bit Sentinel",
            "icon_emoji": ":warning:",
            "text": "*CRITICAL: Fluent Bit Output Buffer Retries or Log Chunk Drops Detected*\n*Pipeline:* `{{ inputs.pipeline_name }}`\n*Dropped Records:* `{{ outputs.scan_fluentbit_metrics.vars.dropped_records_count }}` (Threshold: `{{ inputs.max_dropped_records_threshold }}`)\n*Output Retries:* `{{ outputs.scan_fluentbit_metrics.vars.output_retries_count }}`\n*Retry Failures:* `{{ outputs.scan_fluentbit_metrics.vars.output_retry_failures_count }}`\n*Buffer Fill:* `{{ outputs.scan_fluentbit_metrics.vars.buffer_mem_limit_pct }}%`\n*Timestamp:* `{{ outputs.scan_fluentbit_metrics.vars.audit_timestamp }}`\n*Runbook:* Check upstream destination (Elasticsearch, Loki, OpenSearch) ingest rate limits, scale worker buffer limits, and inspect network egress latency."
          }
    else:
      - id: log_nominal_fluentbit_state
        type: io.kestra.plugin.core.log.Log
        description: Logs nominal telemetry when Fluent Bit log chunks route cleanly
          without retry failures or drops.
        message: "NOMINAL: Fluent Bit pipeline {{ inputs.pipeline_name }} healthy.
          Dropped records: {{
          outputs.scan_fluentbit_metrics.vars.dropped_records_count }}, Output
          retries: {{ outputs.scan_fluentbit_metrics.vars.output_retries_count
          }}, Retry failures: {{
          outputs.scan_fluentbit_metrics.vars.output_retry_failures_count }}."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for logging infrastructure
      reliability archives and compliance logs.
    format: |
      {
        "flow_id": "{{ flow.id }}",
        "pipeline_name": "{{ inputs.pipeline_name }}",
        "dropped_records_count": {{ outputs.scan_fluentbit_metrics.vars.dropped_records_count }},
        "output_retries_count": {{ outputs.scan_fluentbit_metrics.vars.output_retries_count }},
        "output_retry_failures_count": {{ outputs.scan_fluentbit_metrics.vars.output_retry_failures_count }},
        "buffer_mem_bytes": {{ outputs.scan_fluentbit_metrics.vars.buffer_mem_bytes }},
        "buffer_mem_limit_pct": {{ outputs.scan_fluentbit_metrics.vars.buffer_mem_limit_pct }},
        "is_alerting": {{ outputs.scan_fluentbit_metrics.vars.is_alerting }},
        "audit_timestamp": "{{ outputs.scan_fluentbit_metrics.vars.audit_timestamp }}"
      }

triggers:
  - id: periodic_fluentbit_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Executes automated Fluent Bit log pipeline health audit on a
      recurring 15-minute schedule.
    cron: "*/15 * * * *"

Production-grade telemetry reliability watchdog that continuously scrapes Prometheus metrics from Fluent Bit log processor instances. It tracks output buffer retry saturation, monitors discarded log chunk drops caused by downstream backpressure, identifies buffer memory exhaustion, and dispatches structured Slack alerts with remediation runbooks before critical audit and security logs are permanently lost.

How it works

A recurring schedule trigger wakes the flow every 15 minutes to initiate Fluent Bit pipeline surveillance.
A containerized Python script polls the Fluent Bit HTTP metrics endpoint (/api/v1/metrics/prometheus).
The script extracts dropped record counters (fluentbit_output_dropped_records_total) across output plugins.
The script audits output retry totals and catastrophic retry failures (fluentbit_output_retries_failed_total).
A decision gate evaluates whether record drops or retry failures breach enterprise telemetry safety margins.
When log pipeline bottlenecks emerge, high-priority notifications dispatch to Telemetry SRE channels with runbooks.
A structured audit manifest is emitted for enterprise compliance records and observability pipeline archives.

Architecture diagram

flowchart TD
    A[Schedule Trigger: Every 15 Mins] --> B[scan_fluentbit_metrics]
    B --> C{evaluate_fluentbit_threshold}
    C -- Alert Triggered --> D[notify_slack_telemetry_team]
    C -- Nominal State --> E[log_nominal_fluentbit_state]
    D --> F[emit_audit_manifest]
    E --> F[emit_audit_manifest]

Architecture components

Trigger (periodic_fluentbit_audit): Recurring cron schedule executing every 15 minutes for proactive logging surveillance.
Scraper Task (scan_fluentbit_metrics): Docker-isolated Python script polling Fluent Bit metrics with error handling.
Decision Gate (evaluate_fluentbit_threshold): Evaluates dropped records and retry failures against user thresholds.
Notification Task (notify_slack_telemetry_team): Sends rich Slack alert with dropped counts, retry numbers, and runbooks.
Compliance Manifest (emit_audit_manifest): Emits structured JSON summary for telemetry audit archives.

Key capabilities

Continuous surveillance of log record loss and chunk dropouts across forwarding outputs.
Early warning for output destination backpressure (Elasticsearch, OpenSearch, Kafka, Loki).
Proactive tracking of in-memory chunk buffer saturation to prevent pod OOM terminations.
Standardized Slack alerting with operational runbook instructions for logging administrators.

What you get

Real-time visibility into Fluent Bit logging pipeline throughput and transport reliability.
Protection against silent audit log loss during sudden application traffic spikes.
Automated Slack incident notifications with dropped record metrics and buffer diagnostics.
Clean compliance manifests recording log transport health for SOC 2 and PCI-DSS audit trails.

Who it's for

Telemetry and Observability Engineers managing enterprise log forwarding infrastructures.
Site Reliability Engineers safeguarding centralized log ingestion and alerting pipelines.
Cloud Infrastructure Engineers operating Kubernetes Fluent Bit DaemonSets.
DevSecOps Teams maintaining continuous security log shipping and audit trails.

Why orchestrate this with Kestra

Decoupled Architecture: Scrapes metrics externally without requiring root access to logging nodes.
Deterministic Scheduling: Native cron scheduling ensures consistent surveillance without cron daemon drift.
Isolated Execution: Runs Python scrapers inside isolated Docker containers with pre-installed libraries.
Enterprise Auditability: Generates structured execution histories and JSON manifests for regulatory reporting.

Use cases

Downstream Ingest Throttling: Catch 429 Too Many Requests errors from cloud log sinks before buffers discard chunks.
Worker Node Memory Pressure: Identify when log input volume outpaces output network throughput on busy worker nodes.
Network Egress Blackholes: Detect retry storms caused by transient DNS lookup failures to log ingestion endpoints.
Configuration Drift Auditing: Verify that buffer limit settings remain sufficient following log volume increases.

Inputs

Name	Type	Default	Description
`fluentbit_metrics_url`	STRING	`http://fluent-bit.logging.svc.cluster.local:2020`	Base HTTP URL exposing Fluent Bit Prometheus metrics.
`pipeline_name`	STRING	`production-container-logs`	Target Fluent Bit logging pipeline identifier under active surveillance.
`max_dropped_records_threshold`	INT	`100`	Maximum allowable dropped log records before alerting.
`lookback_minutes`	INT	`15`	Lookback evaluation window in minutes for pipeline telemetry.
`alert_on_fluentbit_failures`	BOOL	`true`	Toggle Slack alert dispatch upon detecting buffer retries or drops.

Expected outputs

{{ outputs.scan_fluentbit_metrics.vars.dropped_records_count }}: Cumulative count of permanently discarded log records.
{{ outputs.scan_fluentbit_metrics.vars.output_retries_count }}: Total number of retried chunk deliveries across outputs.
{{ outputs.scan_fluentbit_metrics.vars.output_retry_failures_count }}: Count of delivery attempts that exhausted maximum retries.
{{ outputs.scan_fluentbit_metrics.vars.buffer_mem_limit_pct }}: Percentage of allocated in-memory buffer currently occupied.
{{ outputs.scan_fluentbit_metrics.vars.is_alerting }}: Boolean flag indicating whether failure thresholds were breached.
{{ outputs.emit_audit_manifest.value }}: Structured JSON audit manifest emitted for compliance archives.

Prerequisites

Fluent Bit deployed with HTTP monitoring server enabled (HTTP_Server On, HTTP_Port 2020).
Fluent Bit monitoring port 2020 reachable from the Kestra worker execution environment.
Docker daemon running on the Kestra host to execute containerized Python tasks.

Secrets

SLACK_WEBHOOK_URL: Incoming webhook URL for the target Telemetry SRE Slack alerting channel.

Quick start

Ensure Fluent Bit configuration has [SERVICE] block with HTTP_Server On enabled.
Configure the SLACK_WEBHOOK_URL secret in your Kestra namespace settings.
Specify your pipeline identifier in pipeline_name.
Execute a manual test run via the Kestra UI to verify metric ingestion.
Enable the periodic_fluentbit_audit trigger to establish automated continuous surveillance.

Common pitfalls and troubleshooting

Retry Limit Discard: Setting Retry_Limit False causes infinite retry loops and memory exhaustion; use bounded limits.
Mem_Buf_Limit OOM Protection: Missing Mem_Buf_Limit allows inputs to buffer indefinitely until the pod is killed.
Disk Buffer Backpressure: When using disk storage, ensure PVC volume capacity is sized to survive network blips.
HTTP Metrics Endpoint Binding: Verify HTTP_Listen 0.0.0.0 is configured so remote scrapers can query metrics.

How to extend

Automated Buffer Flusher: Trigger a task to rotate downstream output connections when retry counts spike.
Dynamic Ingest Throttling: Chain a task to apply log rate limiting filters to noisy container workloads.
PagerDuty Escalation: Escalate to on-call telemetry engineers if dropped records exceed 1,000 within 15 minutes.

Fluent Bit Output Buffer Retry and Chunk Drop Sentinel

How it works

Architecture diagram

Architecture components

Key capabilities

What you get

Who it's for

Why orchestrate this with Kestra

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Fluent Bit Output Buffer Retry and Chunk Drop Sentinel

How it works

Architecture diagram

Architecture components

Key capabilities

What you get

Who it's for

Why orchestrate this with Kestra

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Purge Fastly Cache on Deploy via Webhook

Monitor FastAPI Synthetic Latency SLOs and Alert SRE Teams via Slack

Set up alerts for failed workflow executions using Discord

Fastly Edge Error Spike Alert with the Stats Trigger

Falco Cloud Native Runtime Threat and Security Priority Event Sentinel

Fastly Daily Traffic Digest to Slack

More Cloud Blueprints

Fastly Daily Traffic Digest to Slack

Monitor FastAPI Synthetic Latency SLOs and Alert SRE Teams via Slack

Fastly Surrogate Key Purge for CMS Content Updates

Purge Fastly Cache on Deploy via Webhook

Fastly Edge Error Spike Alert with the Stats Trigger

GCP Firestore Stuck Document Status Sentinel

New to Kestra?