Managing Logstash

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

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