Configure Logstash to consume logs from Kafka, transform the data, and securely output to Elasticsearch with SSL/TLS.
The pipeline config has three sections: input, filter, and output.
cat /etc/logstash/conf.d/no-filter.confConfigures Logstash to act as a consumer for the Kafka topic.
input {
kafka {
bootstrap_servers => "192.168.20.208:9092"
topics => ["application_logs"]
codec => json { ecs_compatibility => disabled }
group_id => "logstash-consumer-group"
auto_offset_reset => "earliest"
}
}| Setting | Value | Explanation |
|---|---|---|
bootstrap_servers | 192.168.20.208:9092 | Address and port of the Kafka broker |
topics | application_logs | The Kafka topic Logstash will read from |
codec | json { ... } | Decodes incoming Kafka messages as JSON, creating structured fields |
group_id | logstash-consumer-group | Consumer group ID — allows multiple Logstash instances to scale across partitions |
auto_offset_reset | earliest | Starts reading from oldest available message on first startup |
Standardizes data before it reaches Elasticsearch.
filter {
date {
match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
timezone => "Asia/Riyadh"
target => "@timestamp"
}
uuid {
target => "doc_id"
}
mutate {
remove_field => ["timestamp", "@version", "event"]
}
}| Plugin | Config | Explanation |
|---|---|---|
| date | match, timezone, target | Parses the app timestamp, adjusts to Asia/Riyadh timezone, sets it as @timestamp |
| uuid | target => "doc_id" | Generates a unique ID for every log document — ensures document uniqueness in Elasticsearch |
| mutate | remove_field | Removes temporary fields: original timestamp, Logstash's @version, and event |
Sends cleaned, structured logs to your secure Elasticsearch cluster.
output {
elasticsearch {
hosts => ["https://localhost:9200"]
user => "elastic"
password => "7=04elu9ZfBiltoN=KFS"
ssl_enabled => true
ssl_certificate_authorities => ["/etc/logstash/elasticsearch-ca.crt"]
index => "applogs-%{+YYYY.MM.dd}"
document_id => "%{doc_id}"
}
stdout { codec => rubydebug }
}| Setting | Value | Explanation |
|---|---|---|
hosts | https://localhost:9200 | Elasticsearch endpoint — https indicates secure connection |
user / password | elastic / password | Credentials for authentication with the Elasticsearch cluster |
ssl_enabled | true | Required when using https hosts |
ssl_certificate_authorities | cert path | CA certificate location to trust the Elasticsearch SSL certificate |
index | applogs-%{+YYYY.MM.dd} | Daily index naming — standard practice for time-series log data |
document_id | %{doc_id} | UUID from filter stage used as unique primary key in Elasticsearch |
stdout | rubydebug | Debug output — prints every processed log to the Logstash console |
input {
kafka {
bootstrap_servers => "192.168.20.208:9092"
topics => ["application_logs"]
codec => json { ecs_compatibility => disabled }
group_id => "logstash-consumer-group"
auto_offset_reset => "earliest"
}
}
filter {
date {
match => ["timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
timezone => "Asia/Riyadh"
target => "@timestamp"
}
uuid {
target => "doc_id"
}
mutate {
remove_field => ["timestamp", "@version", "event"]
}
}
output {
elasticsearch {
hosts => ["https://localhost:9200"]
user => "elastic"
password => "7=04elu9ZfBiltoN=KFS"
ssl_enabled => true
ssl_certificate_authorities => ["/etc/logstash/elasticsearch-ca.crt"]
index => "applogs-%{+YYYY.MM.dd}"
document_id => "%{doc_id}"
}
stdout { codec => rubydebug }
}Insufficient memory is the most common reason for Logstash instability. Set heap size in /etc/logstash/jvm.options.
sudo sed -i 's/^-Xms.*/-Xms4g/' /etc/logstash/jvm.options
sudo sed -i 's/^-Xmx.*/-Xmx4g/' /etc/logstash/jvm.optionscd /etc/elasticsearch/certs/
openssl s_client -showcerts -connect ELK-SERVER:9200 </dev/null 2>/dev/null \
| openssl x509 > /etc/logstash/elasticsearch-ca.crtopenssl s_client -connect ELK-SERVER:9200 — connects to Elasticsearch over TLS and prints the certificate chainopenssl x509 — extracts the X.509 certificate onlycd /etc/elasticsearch/certs/
openssl x509 -in http_ca.crt -noout -fingerprint -sha256 \
| cut -d '=' -f2 | tr -d ':' | tr 'A-F' 'a-f' >> /etc/filebeat/fingerprint.txt
cat /etc/filebeat/fingerprint.txt7a212d974cd16b78bdd5e3e3541d5579558914ae6ef52bc6c0a9d3811a46dd6boutput.elasticsearch:
hosts: ["ELK-SERVER:9200"]
preset: balanced
protocol: "https"
ssl:
enabled: true
ca_trusted_fingerprint: "7a212d974cd16b78bdd5e3e3541d5579558914ae6ef52bc6c0a9d3811a46dd6b"
username: "elastic"
password: "abcd@1234"sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash -tchmod 644 /etc/logstash/conf.d/filebeat-client-ubuntu.conf
chown -R logstash:logstash /usr/share/logstash/datachown prevents permission errors when pipelines write to disk.systemctl daemon-reload
systemctl enable --now logstash
systemctl start logstash
systemctl restart logstash
systemctl status logstash
# Verify Logstash is listening on port 5044
ss -tulnp | grep 5044cd /etc/elasticsearch/certs/
openssl s_client -showcerts -connect ELK-SERVER:9200 </dev/null 2>/dev/null \
| openssl x509 > /etc/logstash/elasticsearch-ca.crt
sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash -t
chmod 644 /etc/logstash/conf.d/filebeat-client-ubuntu.conf
chown -R logstash:logstash /usr/share/logstash/data
sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash -t
sudo -u logstash /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/filebeat-client-ubuntu.conf
systemctl daemon-reload
systemctl enable --now logstash
systemctl start logstash
systemctl restart logstash
systemctl status logstash
ss -tulnp | grep 5044