ELK Stack — Input Plugins

5 minute read

Input Plugins

Input plugins are the starting point of every Logstash pipeline. They create events from external data sources — files, network streams, message queues, databases, and more.

stdin — Standard Input

The simplest input, useful for testing and debugging:

input {
  stdin {
    codec => plain
    type => "stdin"
  }
}

Type a line and press Enter — each line becomes one event.

file — Reading Log Files

The file input tails files like tail -F and tracks read positions using a sincedb file:

input {
  file {
    path => "/var/log/nginx/access.log"
    start_position => "beginning"      # "beginning" or "end" (default)
    sincedb_path => "/var/lib/logstash/sincedb_nginx"
    type => "nginx-access"

    # File discovery
    discover_interval => 15            # Seconds between file discovery scans
    stat_interval => 1                 # Seconds between file stat checks

    # Exclude patterns
    exclude => ["*.gz", "*.zip"]

    # File identity tracking
    file_completed_action => "log"     # What to do when file is done
    file_completed_log_path => "/var/log/logstash/completed.log"
  }
}

Key concepts:

Option	Description
`start_position`	`beginning` reads from start (first run only); `end` tails new lines
`sincedb_path`	Tracks file read positions so Logstash resumes where it left off after restart
`discover_interval`	How often to look for new files matching the path glob
`type`	Adds a `type` field to every event — useful for conditionals

Reading multiple files:

input {
  file {
    path => [
      "/var/log/app/*.log",
      "/var/log/syslog"
    ]
    start_position => "beginning"
    type => "application"
  }
}

beats — Receiving from Beats Agents

The beats input receives events from Filebeat, Metricbeat, and other Beat shippers:

input {
  beats {
    port => 5044
    host => "0.0.0.0"

    # SSL/TLS encryption
    ssl_enabled => true
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
    ssl_client_authentication => "required"
  }
}

Corresponding Filebeat configuration:

# filebeat.yml
output.logstash:
  hosts: ["logstash-server:5044"]
  ssl.certificate: "/etc/filebeat/certs/filebeat.crt"
  ssl.key: "/etc/filebeat/certs/filebeat.key"
  ssl.certificate_authorities: ["/etc/filebeat/certs/ca.crt"]

tcp — TCP Socket Input

Receive events over a TCP connection:

input {
  tcp {
    port => 5000
    host => "0.0.0.0"
    codec => json_lines            # Expect newline-delimited JSON

    # SSL settings
    ssl_enabled => false

    # Add metadata
    type => "tcp-json"
  }
}

Test with netcat:

echo '{"user": "alice", "action": "login"}' | nc localhost 5000

udp — UDP Socket Input

Similar to TCP but connectionless — useful for syslog and metrics:

input {
  udp {
    port => 5001
    codec => plain
    type => "udp-input"
  }
}

syslog — Syslog Messages

A specialized input that parses RFC3164 and RFC5424 syslog messages:

input {
  syslog {
    port => 514
    type => "syslog"
    timezone => "UTC"

    # Facility and severity labels are added automatically
    # Fields: facility, facility_label, severity, severity_label,
    #         priority, timestamp, logsource, program, pid, message
  }
}

Sample syslog input:

<34>Oct 11 22:14:15 mymachine su: 'su root' failed for user1 on /dev/pts/8

Parsed fields:

{
    "facility" => 4,
    "facility_label" => "security/authorization",
    "severity" => 2,
    "severity_label" => "Critical",
    "priority" => 34,
    "program" => "su",
    "message" => "'su root' failed for user1 on /dev/pts/8"
}

kafka — Apache Kafka Consumer

Consume messages from Kafka topics:

input {
  kafka {
    bootstrap_servers => "kafka-01:9092,kafka-02:9092,kafka-03:9092"
    topics => ["app-logs", "access-logs"]
    group_id => "logstash-consumer-group"
    codec => json

    # Consumer settings
    consumer_threads => 3
    auto_offset_reset => "latest"      # "latest" or "earliest"
    decorate_events => "extended"      # Adds Kafka metadata fields

    # 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/HTTPS Endpoint

Expose an HTTP endpoint to receive events via POST requests:

input {
  http {
    port => 8080
    host => "0.0.0.0"
    codec => json

    # Basic auth
    user => "logstash"
    password => "secret"

    # SSL
    ssl_enabled => true
    ssl_certificate => "/etc/logstash/certs/http.crt"
    ssl_key => "/etc/logstash/certs/http.key"

    # Response
    response_code => 200
    type => "webhook"
  }
}

Send data:

curl -X POST http://localhost:8080 \
  -u logstash:secret \
  -H "Content-Type: application/json" \
  -d '{"service": "payment", "status": "success", "amount": 99.99}'

jdbc — Database Input

Pull data from any JDBC-compatible database (MySQL, PostgreSQL, Oracle, etc.):

input {
  jdbc {
    jdbc_driver_library => "/opt/logstash/drivers/postgresql-42.2.18.jar"
    jdbc_driver_class => "org.postgresql.Driver"
    jdbc_connection_string => "jdbc:postgresql://db-host:5432/myapp"
    jdbc_user => "logstash_reader"
    jdbc_password => "secure_password"

    # SQL query — use :sql_last_value for incremental sync
    statement => "SELECT * FROM orders WHERE updated_at > :sql_last_value ORDER BY updated_at"

    # Scheduling (cron syntax)
    schedule => "*/5 * * * *"          # Every 5 minutes

    # Tracking column for incremental loads
    use_column_value => true
    tracking_column => "updated_at"
    tracking_column_type => "timestamp"
    last_run_metadata_path => "/var/lib/logstash/jdbc_last_run"

    type => "database"
  }
}

s3 — AWS S3 Bucket

Read objects from an S3 bucket:

input {
  s3 {
    bucket => "my-log-bucket"
    region => "us-east-1"
    prefix => "logs/production/"

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

    # Processing
    codec => plain
    type => "s3-logs"

    # Polling
    interval => 60                     # Seconds between polls

    # After processing
    delete => false                    # Don't delete from S3
    backup_to_bucket => "my-log-archive"
  }
}

elasticsearch — Elasticsearch Input

Read documents from an Elasticsearch index (useful for reindexing or cross-cluster sync):

input {
  elasticsearch {
    hosts => ["https://es-source:9200"]
    index => "old-logs-*"
    query => '{ "query": { "range": { "@timestamp": { "gte": "now-1d" } } } }'
    size => 500
    scroll => "5m"
    docinfo => true                    # Include _index, _type, _id metadata

    # Authentication
    user => "elastic"
    password => "changeme"
    ssl_enabled => true
    ssl_certificate_authorities => ["/etc/logstash/certs/ca.crt"]
  }
}

Comparison of Common Input Plugins

Plugin	Source	Use Case	Protocol
`stdin`	Terminal	Testing/debugging	—
`file`	Log files	Tailing local files	Filesystem
`beats`	Beat agents	Centralized collection from hosts	Lumberjack
`tcp`	Network	Custom TCP log shipping	TCP
`udp`	Network	Syslog, metrics, statsd	UDP
`syslog`	Syslog daemons	Centralized syslog server	TCP/UDP
`kafka`	Kafka topics	High-throughput streaming	Kafka
`http`	HTTP clients	Webhooks, REST APIs	HTTP(S)
`jdbc`	Databases	Periodic database polling	JDBC
`s3`	AWS S3	Cloud log storage	AWS SDK
`elasticsearch`	ES index	Reindexing, cross-cluster sync	HTTP(S)

Quick Reference

Concept	Description
`start_position`	`beginning` (read from start) or `end` (tail new lines only)
`sincedb`	Tracks file read positions for the `file` input
`codec`	Decodes incoming data format (plain, json, json_lines, multiline)
`type`	Adds a `type` field to events — useful for routing in filters/outputs
`beats` port	Default `5044` — receives data from Filebeat, Metricbeat, etc.
`jdbc` schedule	Cron expression for periodic database polling
`kafka` group_id	Consumer group for coordinated consumption across instances
`tags`	Adds tags to events for conditional processing

Tip: You can define multiple inputs in a single pipeline. Use the type or tags fields to distinguish events from different sources in your filters and outputs.

👉 Next: Filter Plugins

Share on

Twitter Facebook LinkedIn