ELK Stack — Output Plugins

Output Plugins

Output plugins are the final stage of the Logstash pipeline. They send processed events to one or more destinations — Elasticsearch, files, message queues, HTTP endpoints, and more.

stdout — Console Output

The stdout output prints events to the terminal. Essential for testing and debugging:

output {
  stdout {
    codec => rubydebug
  }
}

Rubydebug output:

{
    "@timestamp" => 2026-04-01T09:06:00.000Z,
       "message" => "User login successful",
          "user" => "alice",
        "action" => "login",
      "@version" => "1"
}

Other codec options:

# JSON output (one event per line)
output {
  stdout {
    codec => json_lines
  }
}

# Simple line output
output {
  stdout {
    codec => line {
      format => "%{@timestamp} [%{level}] %{message}"
    }
  }
}

# Dots — print a dot per event (throughput testing)
output {
  stdout {
    codec => dots
  }
}

elasticsearch — The Primary Output

The most common output sends events to Elasticsearch:

output {
  elasticsearch {
    hosts => ["https://es-node1:9200", "https://es-node2:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"

    # Authentication — Basic
    user => "elastic"
    password => "changeme"

    # SSL/TLS
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]

    # Document settings
    document_id => "%{[@metadata][_id]}"    # Use for upserts / deduplication
    action => "index"                        # index, create, update, delete

    # Index template
    manage_template => true
    template_name => "app-logs"
    template => "/etc/logstash/templates/app-logs.json"
    template_overwrite => true

    # Performance
    bulk_path => "/_bulk"
  }
}

Authentication with API key:

output {
  elasticsearch {
    hosts => ["https://es-node1:9200"]
    index => "secure-logs-%{+YYYY.MM.dd}"

    # API Key authentication
    api_key => "id:api_key_value"

    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
  }
}

Authentication with cloud_id (Elastic Cloud):

output {
  elasticsearch {
    cloud_id => "my-deployment:dXMtY2VudHJhbC0xLmdjcC5jbG91ZC5lcy5..."
    cloud_auth => "elastic:password"
    index => "cloud-logs-%{+YYYY.MM.dd}"
  }
}

Dynamic index naming:

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    user => "elastic"
    password => "changeme"

    # Route events to different indices based on fields
    index => "%{[type]}-%{+YYYY.MM.dd}"
    # Produces: apache-access-2026.04.01, syslog-2026.04.01, etc.
  }
}

Using data streams (recommended for time-series data):

output {
  elasticsearch {
    hosts => ["https://localhost:9200"]
    user => "elastic"
    password => "changeme"

    # Data stream mode
    data_stream => true
    data_stream_type => "logs"
    data_stream_dataset => "apache"
    data_stream_namespace => "production"
    # Produces data stream: logs-apache-production
  }
}

file — Write to Files

Write events to files on disk:

output {
  file {
    path => "/var/log/logstash/output/%{type}-%{+YYYY-MM-dd}.log"
    codec => json_lines

    # Rotation
    filename_failure => "_filepath_failures"

    # Write mode
    write_behavior => "append"

    # Flush
    flush_interval => 2             # Seconds between flushes
  }
}

CSV output:

output {
  file {
    path => "/var/log/logstash/output/events-%{+YYYY-MM-dd}.csv"
    codec => line {
      format => "%{@timestamp},%{level},%{service},%{message}"
    }
  }
}

kafka — Produce to Kafka

Send events to Kafka topics:

output {
  kafka {
    bootstrap_servers => "kafka-01:9092,kafka-02:9092"
    topic_id => "processed-logs"
    codec => json

    # Producer settings
    acks => "all"                    # "0", "1", or "all"
    retries => 3
    batch_size => 16384
    linger_ms => 5

    # Partitioning
    message_key => "%{service}"      # Route same service to same partition

    # Compression
    compression_type => "snappy"     # none, gzip, snappy, lz4

    # Security
    security_protocol => "SASL_SSL"
    sasl_mechanism => "PLAIN"
    sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='user' password='pass';"
  }
}

http — HTTP POST Output

Send events to an HTTP endpoint (webhooks, APIs, custom services):

output {
  http {
    url => "https://api.example.com/ingest"
    http_method => "post"
    format => "json"

    # Headers
    headers => {
      "Content-Type" => "application/json"
      "Authorization" => "Bearer eyJhbGciOiJIUzI1NiIs..."
    }

    # Retry settings
    retry_failed => true
    retryable_codes => [429, 500, 502, 503]
    automatic_retries => 3

    # Connection
    connect_timeout => 10
    request_timeout => 60

    # Pool
    pool_max => 50
    pool_max_per_route => 25
  }
}

email — Email Notifications

Send email alerts for specific events:

output {
  if "server_error" in [tags] {
    email {
      to => "ops-team@example.com"
      from => "logstash@example.com"
      subject => "Server Error Alert — %{service}"
      body => "Error detected at %{@timestamp}\n\nService: %{service}\nMessage: %{message}\nHost: %{host}"

      # SMTP settings
      address => "smtp.example.com"
      port => 587
      username => "logstash@example.com"
      password => "smtp_password"
      use_tls => true
      authentication => "plain"
    }
  }
}

s3 — AWS S3 Output

Archive events to S3 buckets:

output {
  s3 {
    region => "us-east-1"
    bucket => "log-archive-bucket"
    prefix => "logstash/logs/%{+YYYY/MM/dd}"

    # Credentials (or use IAM role)
    access_key_id => "AKIAIOSFODNN7EXAMPLE"
    secret_access_key => "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"

    # File settings
    codec => json_lines
    encoding => "gzip"               # Compress files

    # Rotation
    size_file => 5242880             # 5 MB per file
    time_file => 15                  # Minutes per file

    # Storage class
    storage_class => "STANDARD_IA"   # STANDARD, STANDARD_IA, GLACIER, etc.
  }
}

Multiple Outputs

A single pipeline can send events to multiple destinations:

output {
  # Always send to Elasticsearch
  elasticsearch {
    hosts => ["https://localhost:9200"]
    index => "app-logs-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "changeme"
  }

  # Debug to console in dev
  if [@metadata][env] == "development" {
    stdout {
      codec => rubydebug
    }
  }

  # Archive to S3
  s3 {
    bucket => "log-archive"
    prefix => "raw-logs/%{+YYYY/MM/dd}"
    codec => json_lines
    encoding => "gzip"
    region => "us-east-1"
  }

  # Send errors to a separate index
  if [level] == "ERROR" or [level] == "FATAL" {
    elasticsearch {
      hosts => ["https://localhost:9200"]
      index => "error-logs-%{+YYYY.MM.dd}"
      user => "elastic"
      password => "changeme"
    }
  }
}

Quick Reference

Output	Destination	Key Options
stdout	Console	codec => rubydebug for debugging
elasticsearch	Elasticsearch	hosts, index, user, password
file	Local files	path (supports field references), codec
kafka	Kafka topics	bootstrap_servers, topic_id, codec
http	HTTP endpoints	url, http_method, format, headers
email	Email (SMTP)	to, subject, body, SMTP settings
s3	AWS S3	bucket, prefix, region, encoding

Concept	Description
index	Elasticsearch index name — supports date math %{+YYYY.MM.dd}
document_id	Set explicit document ID for upserts/dedup
data_stream	Use Elasticsearch data streams for time-series data
codec	Encode output format (json_lines, rubydebug, line)
api_key	Elasticsearch API key auth format: "id:key"
Multiple outputs	Events are sent to ALL matching outputs in the block

Tip: Use [@metadata] fields to store values that you need for routing logic but don’t want indexed in Elasticsearch. Metadata fields are automatically dropped from the output event.

👉 Next: Multiple Pipelines