Installing and Configuring Logstash for Kafka-to-Elasticsearch Pipelines

Package Installation

Add the Elastic PGP key and set up the repository:

sudo rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

sudo tee /etc/yum.repos.d/logstash.repo <<EOF
[logstash-8.x]
name=Elastic repository for 8.x packages
baseurl=https://artifacts.elastic.co/packages/8.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF

sudo yum install logstash -y

Key directories after a package-based install:

  • Service unit: /etc/systemd/system/logstash.service
  • Installation root: /usr/share/logstash
  • Binaries: /usr/share/logstash/bin
  • Configuration: /etc/logstash
  • Logs: /var/log/logstash/
  • Plugins: /usr/share/logstash/plugins
  • Perssitent data: /var/lib/logstash

Configuration Layout

Pipeline definition files belong under /etc/logstash/conf.d. Runtime and startup options are managed by:

  • /etc/logstash/logstash.yml – Logstash runtime settings
  • /etc/logstash/pipelines.yml – multi-pipeline definitions
  • /etc/logstash/jvm.options – JVM heap and tuning flags
  • /etc/logstash/startup.options – service initialization parameters

Verifying the Installation

Run a minimal stdin-to-stdout pipeline:

cd /usr/share/logstash/
bin/logstash -e 'input { stdin {} } output { stdout {} }'

Type hello world into the interactive shell. A successful installation produces a structured event similar to:

{
  "@version" : "1",
  "message" : "hello world",
  "@timestamp" : "2022-04-26T09:18:26.485741Z",
  "event" : { "original" : "hello world" },
  "host" : { "hostname" : "10-52-6-111" }
}

Kafka to Elasticsearch Pipeline

Stack versions used in thiss setup:

  • Logstash 8.1.3 (bundled JDK)
  • Elasticsearch 7.16.3
  • Kafka 2.7.1

Create /etc/logstash/conf.d/kafka_to_es.conf:

input {
  kafka {
    bootstrap_servers => "106:9093,10.1:9093,10:9093"
    topics => "vslogs"
    group_id => "vslogs_group_id_1"
    client_id => "vslogs_client_id_1"
    auto_offset_reset => "latest"
    consumer_threads => 3
    decorate_events => true
    type => "vslogs"
    codec => "json"
    sasl_mechanism => "SCRAM-SHA-256"
    security_protocol => "SASL_PLAINTEXT"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='' password='';"
  }

  kafka {
    bootstrap_servers => "6:9093,2.31:9093,1.4.0.112:9093"
    topics => "vsulblog"
    group_id => "vsulblog_group_id_1"
    client_id => "vsulblog_client_id_1"
    auto_offset_reset => "latest"
    consumer_threads => 3
    decorate_events => true
    type => "vsulblog"
    codec => "json"
    sasl_mechanism => "SCRAM-SHA-256"
    security_protocol => "SASL_PLAINTEXT"
    sasl_jaas_config => "org.apache.kafka.common.security.scram.ScramLoginModule required username='' password='';"
  }
}

filter {}

output {
  if [type] == "vslogs" {
    elasticsearch {
      hosts => [ ":9200", ":9200", ":9200" ]
      index => "vs-vslogs"
      user => ""
      password => ""
    }
  }
  if [type] == "vsulblog" {
    elasticsearch {
      hosts => [ ":9200", ":9200", "1.4.1.9:9200" ]
      index => "vs-vsulblog"
      user => ""
      password => ""
    }
  }
}

When a single pipeline consumes multiple Kafka topics, each input must define a distinct group_id and a unique client_id. Legacy Kafka inputs require topic_id instead of topics, and use zk_connect instead of bootstrap_servers.

File to Kafka Example

Read a local file and forward lines to Kafka:

input {
  file {
    codec => plain { charset => "UTF-8" }
    path => "/root/logserver/gamelog.txt"
    discover_interval => 5
    start_position => "beginning"
  }
}

output {
  kafka {
    topic_id => "gamelogs"
    codec => plain { format => "%{message}" charset => "UTF-8" }
    bootstrap_servers => "node01:9092,node02:9092,node03:9092"
  }
}

Utility Commands

List installed plugins:

/usr/share/logstash/bin/logstash-plugin list

Validate a pipeline configuration:

/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka_to_es.conf --config.test_and_exit

Running as a Systemd Service

Create the service override:

# /etc/systemd/system/logstash.service
[Unit]
Description=logstash

[Service]
Type=simple
User=logstash
Group=logstash
EnvironmentFile=-/etc/default/logstash
EnvironmentFile=-/etc/sysconfig/logstash
ExecStart=/usr/share/logstash/bin/logstash "--path.settings" "/etc/logstash"
Restart=always
WorkingDirectory=/
Nice=19
LimitNOFILE=16384
TimeoutStopSec=infinity

[Install]
WantedBy=multi-user.target

Containerised Deployment

Pull the official image:

docker pull docker.elastic.co/logstash/logstash:8.1.3

Run a transient container with a custom pipeline directory:

docker run --rm -it \
  -v ~/pipeline/:/usr/share/logstash/pipeline/ \
  docker.elastic.co/logstash/logstash:8.1.3

Override logstash.yml:

docker run --rm -it \
  -v ~/settings/logstash.yml:/usr/share/logstash/config/logstash.yml \
  docker.elastic.co/logstash/logstash:8.1.3

In a Docker environment, all Logstash artefacts live under /usr/share/logstash. Confgiuration files map as follows:

  • /usr/share/logstash/config – replaces /etc/logstash
  • /usr/share/logstash/pipeline – replaces /etc/logstash/conf.d

By default the image enables X-Pack management, which causes license errors if Elasticsearch is unreachable. Disable it in logstash.yml:

xpack.management.enabled: false

Foreground debugging:

docker run --rm -it \
  -v /root/docker-logstash-config/logstash.yml:/usr/share/logstash/config/logstash.yml \
  docker.elastic.co/logstash/logstash:8.1.3

Daemonised container:

docker run -itd --name logstash \
  -v /root/docker-logstash-config/logstash.yml:/usr/share/logstash/config/logstash.yml \
  docker.elastic.co/logstash/logstash:8.1.3

docker logs -f logstash

Custom Image Build

A Dockerfile bundles configuration and removes the sample pipeline:

FROM docker.elastic.co/logstash/logstash:8.1.3
RUN rm -f /usr/share/logstash/pipeline/logstash.conf
ADD pipeline/ /usr/share/logstash/pipeline/
ADD config/ /usr/share/logstash/config/

Directory structure before building:

├── config
│   └── logstash.yml
├── Dockerfile
└── pipeline
    └── kafka_to_es.conf

Build the image and tag it:

docker build -t logstash-8.1.3-kafka-to-es:v1 .

Keep the build context minimal to avoid sending unnecessary files to the Docker daemon.

Tags: Logstash Kafka elasticsearch data-pipeline configuration

Posted on Mon, 21 Sep 2026 16:55:23 +0000 by hyperpcs

Hot Tags

©2026 Freaks City