managing-logstash

name	managing-logstash
description	Use when working with Logstash — logstash pipeline management with input/filter/output analysis, pipeline statistics, event processing metrics, and queue monitoring. Covers pipeline health, plugin performance, JVM stats, hot threads analysis, and configuration review. Use when managing Logstash pipelines, analyzing throughput, reviewing filter performance, or troubleshooting event processing.
connection_type	logstash
preload	false

Logstash Management Skill

Monitor and manage Logstash event processing pipelines via the monitoring API.

API Conventions

Authentication

Logstash monitoring API is typically unauthenticated (localhost access). Connection handles auth if configured.

Base URL

• Monitoring API: http://<host>:9600/
• Use connection-injected LOGSTASH_BASE_URL.

Output Rules

• TOKEN EFFICIENCY: Target <=50 lines per output
• Use jq to extract only relevant pipeline and plugin metrics
• NEVER dump full node stats — always filter to specific sections

Core Helper Function

#!/bin/bash

logstash_api() {
    local endpoint="$1"
    curl -s "${LOGSTASH_BASE_URL}${endpoint}"
}

logstash_stats() {
    local section="${1:-}"
    if [ -n "$section" ]; then
        logstash_api "/_node/stats/${section}"
    else
        logstash_api "/_node/stats"
    fi
}

logstash_pipelines() {
    logstash_api "/_node/stats/pipelines"
}

Parallel Execution

{
    logstash_api "/" &
    logstash_stats "jvm" &
    logstash_stats "process" &
    logstash_stats "pipelines" &
}
wait

Anti-Hallucination Rules

NEVER assume pipeline names, plugin IDs, or filter configurations. ALWAYS discover first.

Phase 1: Discovery

#!/bin/bash
echo "=== Logstash Info ==="
logstash_api "/" | jq '{version: .version, status: .status, pipeline: .pipeline}'

echo ""
echo "=== Pipeline Names ==="
logstash_stats "pipelines" | jq -r '.pipelines | keys[]'

echo ""
echo "=== Plugin Summary per Pipeline ==="
logstash_stats "pipelines" | jq -r '.pipelines | to_entries[] | "\(.key): inputs=\(.value.plugins.inputs | length) filters=\(.value.plugins.filters | length) outputs=\(.value.plugins.outputs | length)"'

Common Operations

Pipeline Health Overview

#!/bin/bash
echo "=== Pipeline Status ==="
logstash_stats "pipelines" | jq -r '.pipelines | to_entries[] | {
    pipeline: .key,
    events_in: .value.events.in,
    events_out: .value.events.out,
    events_filtered: .value.events.filtered,
    queue_events: .value.queue.events_count,
    queue_type: .value.queue.type
}'

echo ""
echo "=== Event Throughput ==="
logstash_stats "pipelines" | jq -r '.pipelines | to_entries[] | "\(.key)\tin:\(.value.events.in)\tout:\(.value.events.out)\tfiltered:\(.value.events.filtered)\tduration_ms:\(.value.events.duration_in_millis)"'

Input/Filter/Output Performance

#!/bin/bash
PIPELINE="${1:-main}"

echo "=== Input Plugins (${PIPELINE}) ==="
logstash_stats "pipelines" | jq -r ".pipelines.\"${PIPELINE}\".plugins.inputs[] | \"\(.name)\tid:\(.id)\tevents:\(.events.out)\"" | head -10

echo ""
echo "=== Filter Plugins (${PIPELINE}) ==="
logstash_stats "pipelines" | jq -r ".pipelines.\"${PIPELINE}\".plugins.filters[] | \"\(.name)\tid:\(.id)\tevents_in:\(.events.in)\tevents_out:\(.events.out)\tduration:\(.events.duration_in_millis)ms\"" | head -15

echo ""
echo "=== Output Plugins (${PIPELINE}) ==="
logstash_stats "pipelines" | jq -r ".pipelines.\"${PIPELINE}\".plugins.outputs[] | \"\(.name)\tid:\(.id)\tevents_in:\(.events.in)\tevents_out:\(.events.out)\tduration:\(.events.duration_in_millis)ms\"" | head -10

echo ""
echo "=== Slowest Filters ==="
logstash_stats "pipelines" | jq -r ".pipelines.\"${PIPELINE}\".plugins.filters | sort_by(-.events.duration_in_millis)[:5][] | \"\(.name)\t\(.id)\t\(.events.duration_in_millis)ms total\""

Queue Monitoring

#!/bin/bash
echo "=== Queue Status ==="
logstash_stats "pipelines" | jq -r '.pipelines | to_entries[] | {
    pipeline: .key,
    queue_type: .value.queue.type,
    events_count: .value.queue.events_count,
    queue_size_bytes: .value.queue.queue_size_in_bytes,
    max_queue_size: .value.queue.max_queue_size_in_bytes
}'

echo ""
echo "=== Queue Utilization ==="
logstash_stats "pipelines" | jq -r '.pipelines | to_entries[] | select(.value.queue.max_queue_size_in_bytes > 0) | "\(.key)\t\(.value.queue.queue_size_in_bytes / .value.queue.max_queue_size_in_bytes * 100 | round)% used"'

JVM & Process Health

#!/bin/bash
echo "=== JVM Memory ==="
{
    logstash_stats "jvm" | jq -r '.jvm.mem | {
        heap_used_pct: .heap_used_percent,
        heap_used_mb: (.heap_used_in_bytes / 1048576 | round),
        heap_max_mb: (.heap_max_in_bytes / 1048576 | round),
        non_heap_mb: (.non_heap_used_in_bytes / 1048576 | round)
    }' &

    echo "=== GC Stats ==="
    logstash_stats "jvm" | jq -r '.jvm.gc.collectors | to_entries[] | "\(.key)\tcollections:\(.value.collection_count)\ttime:\(.value.collection_time_in_millis)ms"' &

    echo "=== Process Stats ==="
    logstash_stats "process" | jq -r '.process | {
        cpu_percent: .cpu.percent,
        open_file_descriptors: .open_file_descriptors,
        max_file_descriptors: .max_file_descriptors,
        uptime_ms: .uptime_in_millis
    }' &
}
wait

Hot Threads Analysis

#!/bin/bash
echo "=== Hot Threads ==="
logstash_api "/_node/hot_threads?human=true" | head -40

echo ""
echo "=== Pipeline Workers ==="
logstash_stats "pipelines" | jq -r '.pipelines | to_entries[] | "\(.key)\tworkers:\(.value.workers)\tbatch_size:\(.value.batch_size)"'

Output Format

Present results as a structured report:

Managing Logstash Report
════════════════════════
Resources discovered: [count]

Resource       Status    Key Metric    Issues
──────────────────────────────────────────────
[name]         [ok/warn] [value]       [findings]

Summary: [total] resources | [ok] healthy | [warn] warnings | [crit] critical
Action Items: [list of prioritized findings]

Target ≤50 lines of output. Use tables for multi-resource comparisons.

Counter-Rationalizations

Shortcut	Counter	Why
"I'll skip discovery and check known resources"	Always run Phase 1 discovery first	Resource names change, new resources appear — assumed names cause errors
"The user only asked for a quick check"	Follow the full discovery → analysis flow	Quick checks miss critical issues; structured analysis catches silent failures
"Default configuration is probably fine"	Audit configuration explicitly	Defaults often leave logging, security, and optimization features disabled
"Metrics aren't needed for this"	Always check relevant metrics when available	API/CLI responses show current state; metrics reveal trends and intermittent issues
"I don't have access to that"	Try the command and report the actual error	Assumed permission failures prevent useful investigation; actual errors are informative

Common Pitfalls

• Pipeline names: Default pipeline is main — multi-pipeline setups use pipelines.yml for naming
• Plugin IDs: Auto-generated IDs are hashes — use custom id in config for readable names
• Queue types: memory (default, volatile), persisted (disk-backed, survives restarts)
• Duration metrics: duration_in_millis is cumulative total — divide by event count for per-event time
• Filter ordering: Filters execute in config order — a slow grok filter blocks downstream filters
• Dead letter queue: Failed events go to DLQ if enabled — check /data/dead_letter_queue/
• Port default: Monitoring API on port 9600 — must be enabled with api.enabled: true
• JVM pressure: If heap_used_percent consistently >75%, increase JVM heap or reduce batch size