Build a Secure, Scalable ELK + Kafka Logging Architecture on CentOS

Architecture Overview

The solution uses Filebeat on each log‑generating server to ship logs to a Kafka cluster; Logstash reads from Kafka, optionally filters the data, and writes to an Elasticsearch cluster; Kibana provides a web UI for searching, aggregating and visualizing the indexed data.

Preparation

Hostname: es83 – devices: 192.168.100.83 – roles: filebeat, es, logstash, kafka, kibana – OS: CentOS 7.6

Hostname: es86 – devices: 192.168.100.86 – roles: es, logstash, kafka – OS: CentOS 7.6

Hostname: es87 – devices: 192.168.100.87 – roles: es, logstash, kafka – OS: CentOS 7.6

The ELK stack version is 7.2.0 and Kafka version is 2.12‑2.3.0.

Environment Configuration

Disable SELinux, stop firewalld, turn off swap, increase file‑descriptor limits, set JVM options, create a non‑privileged user, and configure password‑less SSH between the nodes.

#!/bin/bash
# Update /etc/hosts
cat >> /etc/hosts <<EOF
192.168.100.83 es83
192.168.100.86 es86
192.168.100.87 es87
EOF

# Stop firewalld
systemctl stop firewalld
systemctl disable firewalld

# Disable SELinux
setenforce 0
sed -i "s/SELINUX=enforcing/SELINUX=disabled/g" /etc/selinux/config

# Disable swap
swapoff -a

# Increase limits
cat > /etc/security/limits.conf <<EOF
* soft nofile 65536
* hard nofile 131072
* soft nproc 65536
* hard nproc 65536
* soft memlock unlimited
* hard memlock unlimited
EOF

# Adjust sysctl
cat >> /etc/sysctl.conf <<EOF
vm.max_map_count=562144
EOF
sysctl -p

# Create elkuser
useradd elkuser
echo 123456 | passwd --stdin elkuser

# Setup SSH keys
ssh-keygen -q -N "" -f ~/.ssh/id_rsa
ssh-copy-id 192.168.100.83
ssh-copy-id 192.168.100.86
ssh-copy-id 192.168.100.87

Elasticsearch Cluster Deployment

Download, extract, and configure Elasticsearch with 30 GB heap, G1 GC, and X‑Pack security enabled. Generate a CA, node certificates, and configure elasticsearch.yml for a three‑node master‑data cluster.

# Download
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-7.2.0-linux-x86_64.tar.gz
# Extract
tar -xvf elasticsearch-7.2.0-linux-x86_64.tar.gz
# Adjust JVM heap
sed -i -e 's/1g/30g/g' -e '36,38s/^-/#&/' ./config/jvm.options
# Create certificates
./bin/elasticsearch-certutil ca
./bin/elasticsearch-certutil cert --ca elastic-stack-ca.p12 --ip 192.168.100.83
# (repeat for 192.168.100.86 and .87)
# Copy certificates
cp *.p12 ./config/
# elasticsearch.yml
cat > ./config/elasticsearch.yml <<EOF
cluster.name: chilu_elk
node.name: es83
node.master: true
node.data: true
path.data: /logdata/data1,/logdata/data2,/logdata/data3,/logdata/data4,/logdata/data5,/logdata/data6
bootstrap.memory_lock: true
network.host: 192.168.100.83
http.port: 9200
transport.tcp.port: 9300
discovery.zen.ping.unicast.hosts: ["192.168.100.83:9300","192.168.100.86:9300","192.168.100.87:9300"]
cluster.initial_master_nodes: ["192.168.100.83:9300","192.168.100.86:9300","192.168.100.87:9300"]
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: es83.p12
xpack.security.transport.ssl.truststore.path: elastic-stack-ca.p12
xpack.security.http.ssl.enabled: true
xpack.security.http.ssl.keystore.path: es83.p12
xpack.security.http.ssl.truststore.path: elastic-stack-ca.p12
xpack.security.http.ssl.client_authentication: optional
EOF
# Distribute and start
scp -r ./elasticsearch-7.2.0 192.168.100.86:/home/elkuser/
scp -r ./elasticsearch-7.2.0 192.168.100.87:/home/elkuser/
ssh 192.168.100.86 "sed -i 's/es83/es86/g' /home/elkuser/elasticsearch-7.2.0/config/elasticsearch.yml"
ssh 192.168.100.87 "sed -i 's/es83/es87/g' /home/elkuser/elasticsearch-7.2.0/config/elasticsearch.yml"
# Adjust ownership
chown -R elkuser:elkuser /logdata ./elasticsearch-7.2.0
# Start as elkuser
su elkuser -c "./elasticsearch-7.2.0/bin/elasticsearch -d"
# Generate passwords
./elasticsearch-7.2.0/bin/elasticsearch-setup-passwords auto | tee elk_pwd.log

Kafka Cluster Deployment

Deploy a three‑node Kafka cluster with an embedded Zookeeper, enable SASL/PLAIN authentication, and configure ACLs for producer and consumer.

# Download and extract
wget https://archive.apache.org/dist/kafka/2.3.0/kafka_2.12-2.3.0.tgz
tar -xvf kafka_2.12-2.3.0.tgz
cd kafka_2.12-2.3.0
# Zookeeper config
cat > ./config/zookeeper.properties <<EOF
dataDir=/opt/zookeeper
clientPort=2181
maxClientCnxns=0
tickTime=2000
initLimit=10
syncLimit=5
server.1=192.168.100.83:2888:3888
server.2=192.168.100.86:2888:3888
server.3=192.168.100.87:2888:3888
EOF
mkdir /opt/zookeeper && echo 1 > /opt/zookeeper/myid
# Kafka config
cat > ./config/server.properties <<EOF
broker.id=83
listeners=SASL_PLAINTEXT://192.168.100.83:9092
advertised.listeners=SASL_PLAINTEXT://192.168.100.83:9092
num.network.threads=5
num.io.threads=8
socket.send.buffer.bytes=1024000
socket.receive.buffer.bytes=1024000
socket.request.max.bytes=1048576000
log.dirs=/logdata/kfkdata1,/logdata/kfkdata2,/logdata/kfkdata3,/logdata/kfkdata4,/logdata/kfkdata5
num.partitions=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
log.retention.hours=72
log.cleaner.enable=true
zookeeper.connect=192.168.100.83:2181,192.168.100.86:2181,192.168.100.87:2181
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
super.users=User:admin;User:kafka
EOF
# JAAS files
cat > ./config/zk_server_jaas.conf <<EOF
Server {
  org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="chilu@rljie" user_kafka="chilu@rljie" user_producer="chilu@rljie";
};
EOF
cat > ./config/kafka_server_jaas.conf <<EOF
KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="chilu@rljie" user_admin="chilu@rljie" user_producer="chilu@rljie" user_consumer="chilu@rljie";
};
EOF
# Start services
./bin/zookeeper-server-start.sh -daemon ./config/zookeeper.properties
./bin/kafka-server-start.sh -daemon ./config/server.properties
# Distribute to other nodes and adjust IDs
scp -r ./kafka_2.12-2.3.0 192.168.100.86:/opt/
scp -r ./kafka_2.12-2.3.0 192.168.100.87:/opt/
ssh 192.168.100.86 "echo 2 > /opt/zookeeper/myid && sed -i 's/83/86/' /opt/kafka_2.12-2.3.0/config/server.properties"
ssh 192.168.100.87 "echo 3 > /opt/zookeeper/myid && sed -i 's/83/87/' /opt/kafka_2.12-2.3.0/config/server.properties"
# Create ACLs
cat > kfkacls.sh <<'EOF'
#!/bin/bash
/opt/kafka_2.12-2.3.0/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=mykafka:2181 \
  --add --allow-principal User:producer --allow-host 0.0.0.0 --operation Read --operation Write --topic elk
/opt/kafka_2.12-2.3.0/bin/kafka-acls.sh --authorizer-properties zookeeper.connect=mykafka:2181 \
  --add --allow-principal User:consumer --allow-host 0.0.0.0 --operation Read --operation Write --topic elk
EOF
bash kfkacls.sh

Logstash Service Deployment

Install Logstash 7.2.0, allocate 30 GB heap, copy the CA certificate, and configure input from Kafka and output to Elasticsearch with TLS.

# Download and extract
wget https://artifacts.elastic.co/downloads/logstash/logstash-7.2.0.tar.gz
tar -xvf logstash-7.2.0.tar.gz
# Adjust JVM
sed -i 's/1g/30g/g' ./config/jvm.options
# Copy CA
cp /home/elkuser/elasticsearch-7.2.0/root.pem ./config/
# logstash.yml
cat > ./config/logstash.yml <<EOF
http.host: "192.168.100.83"
node.name: "logstash83"
xpack.monitoring.elasticsearch.hosts: ["https://192.168.100.83:9200"]
xpack.monitoring.enabled: true
xpack.monitoring.elasticsearch.username: "elastic"
xpack.monitoring.elasticsearch.password: "NvOBRGpUE3DoaSbYaUP3"
xpack.monitoring.elasticsearch.ssl.certificate_authority: config/root.pem
xpack.monitoring.elasticsearch.ssl.verification_mode: certificate
xpack.monitoring.collection.interval: 30s
EOF
# Kafka client JAAS
cat > ./config/kafka-client-jaas.conf <<EOF
KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required username="consumer" password="chilu@rljie";
};
EOF
# Example pipeline
cat > ./config/pipeline.conf <<EOF
input {
  kafka {
    bootstrap_servers => "192.168.100.83:9092,192.168.100.86:9092,192.168.100.87:9092"
    client_id => "chilu83"
    topics => ["elk"]
    group_id => "chilu"
    security_protocol => "SASL_PLAINTEXT"
    sasl_mechanism => "PLAIN"
    jaas_path => "/home/elkuser/logstash-7.2.0/config/kafka-client-jaas.conf"
  }
}
output {
  elasticsearch {
    hosts => ["https://192.168.100.83:9200"]
    user => "elastic"
    password => "NvOBRGpUE3DoaSbYaUP3"
    ssl => true
    cacert => "/home/elkuser/logstash-7.2.0/config/root.pem"
    index => "chilu_elk-%{+YYYY.MM.dd}"
  }
}
EOF
# Start Logstash
../bin/logstash -r -f ./pipeline.conf

Kibana Service Deployment

Install Kibana 7.2.0, allocate 8 GB heap, generate self‑signed certificates, and configure it to connect securely to the Elasticsearch cluster.

# Download and extract
wget https://artifacts.elastic.co/downloads/kibana/kibana-7.2.0-linux-x86_64.tar.gz
tar -xvf kibana-7.2.0-linux-x86_64.tar.gz
# Increase Node memory
sed -i 's/warnings/warnings --max_old_space_size=8096/' ./bin/kibana
# Copy client certificate
cp /home/elkuser/elasticsearch-7.2.0/client.p12 ./config/
# Extract PEM files
openssl pkcs12 -in client.p12 -nocerts -nodes -out client.key
openssl pkcs12 -in client.p12 -clcerts -nokeys -out client.cer
openssl pkcs12 -in client.p12 -cacerts -nokeys -chain -out client-ca.cer
# Generate self‑signed server cert
openssl req -newkey rsa:2048 -nodes -keyout server.key -x509 -days 3650 -out server.crt -subj "/C=CN/ST=guangzhou/L=rljie/O=chilu/OU=linux/"
# kibana.yml
cat > ./config/kibana.yml <<EOF
server.name: kibana
server.host: "192.168.100.83"
elasticsearch.hosts: ["https://192.168.100.83:9200"]
elasticsearch.username: "elastic"
elasticsearch.password: "NvOBRGpUE3DoaSbYaUP3"
xpack.security.enabled: true
elasticsearch.ssl.certificateAuthorities: config/client-ca.cer
elasticsearch.ssl.verificationMode: certificate
xpack.security.encryptionKey: "4297f44b13955235245b2497399d7a93"
xpack.reporting.encryptionKey: "4297f44b13955235245b2497399d7a93"
server.ssl.enabled: true
server.ssl.certificate: server.crt
server.ssl.key: server.key
EOF
# Start Kibana (background)
nohup ./bin/kibana --allow-root &

Filebeat Service Deployment

Install Filebeat 7.2.0, configure it to read log files and forward them to the Kafka cluster using SASL/PLAIN authentication.

# Download and extract
wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.2.0-linux-x86_64.tar.gz
tar -xvf filebeat-7.2.0-linux-x86_64.tar.gz
cd filebeat-7.2.0-linux-x86_64
# filebeat.yml
cat > filebeat.yml <<EOF
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /var/log/access.log
  close_timeout: 1h
  clean_inactive: 3h
  ignore_older: 2h

filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: true

setup.template.settings:
  index.number_of_shards: 3

output.kafka:
  hosts: ["192.168.100.83:9092","192.168.100.86:9092","192.168.100.87:9092"]
  topic: elk
  required_acks: 1
  username: "producer"
  password: "chilu@rljie"
EOF
# Start Filebeat
nohup ./filebeat -e -c filebeat.yml &

After completing all steps, access Kibana at https://192.168.100.83 and log in with the elastic user and its generated password to explore the collected logs.