ITD - ELK Multi Server Deployment

Procedure :

UAT Server Details

ELK_Kafka_Cluster1_UAT  dcnluelkkafcl01 10.176.16.121
ELK_Elastic_Search1_UAT dcnluelkelse01 10.176.16.122
ELK_Logtash dcnlelklog 10.176.16.123
ELK_Kibana dcnlelkkib 10.176.16.124

PDC Server Details

ELK_Kafka_Node1_PDC	10.176.25.233	dcplekafka01
ELK_Kafka_Node2_PDC	10.176.25.234	dcplekafka02
ELK_Elastic_Search_Master_Node_1	10.176.25.235		dcplelkesmas01
ELK_Elastic_Search_Master_Node_2	10.176.25.236		dcplelkesmas02
ELK_Elastic_Search_Master_Node_3	10.176.25.237		dcplelkesmas03
ELK_Elastic_Search_Slave_Node_1	10.176.25.238		dcplelkesdat01
ELK_Elastic_Search_Slave_Node_2	10.176.25.239		dcplelkesdat02
ELK_Logstash Node 1	10.176.25.240	dcplelklogst01 
ELK_Logstash Node 2	10.176.25.241	dcplelklogst02
ELK_Kibana Node	10.176.25.242	dcplelkkibana

BCP Server Details

ELK_Kafka_Node1_BCP	10.144.24.233	drplekafka01
ELK_Kafka_Node2_BCP	10.144.24.234	drplekafka02
ELK_Elastic_Search_Master_Node_1	10.144.24.235		drplelkesmas01
ELK_Elastic_Search_Master_Node_2	10.144.24.236		drplelkesmas02
ELK_Elastic_Search_Master_Node_3	10.144.24.237		drplelkesmas03
ELK_Elastic_Search_Slave_Node_1	10.144.24.238		drplelkesdat01
ELK_Elastic_Search_Slave_Node_2	10.144.24.239		drplelkesdat02
ELK_Logstash Node 1	10.144.24.240	drplelklogst01 
ELK_Logstash Node 2	10.144.24.241	drplelklogst02 
ELK_Kibana Node	10.144.25.242	drplelkkibana

1. Install/Configure ELK Multi Components Architecture - Elasticsearch, Logstash, Kibana and Kafka all on different nodes

NOTE: Installations/Configurations are same for both DC and DR.

INSTALL KAFKA

Installing pre-requisities,

yum install java-1.8.0-openjdk
yum install java-1.8.0-openjdk-devel

wget https://mirrors.estointernet.in/apache/kafka/2.4.1/kafka_2.12-2.4.1.tgz

tar -xzf kafka_2.12-2.4.1.tgz
cp -r kafka_2.12-2.4.1 /opt/kafka

cd /opt/kafka/
bin/zookeeper-server-start.sh config/zookeeper.properties &
/opt/kafka/bin/kafka-server-start.sh  /opt/kafka/config/server.properties &

yum install net-tools
netstat -ntplu 			// check ports and services.

Enable Kafka at boot time

sudo vi /etc/systemd/system/zookeeper.service

[Unit]
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
ExecStart=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties
ExecStop=/opt/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

Next, create the systemd service file for kafka:

sudo vi /etc/systemd/system/kafka.service

[Unit]
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple

ExecStart=/bin/sh -c '/opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/server.properties > /opt/kafka/kafka.log 2>&1'
ExecStop=/opt/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

sudo systemctl start kafka

To ensure that the server has started successfully, check the journal logs for the kafka unit:
 journalctl -u kafka
 sudo systemctl enable kafka

firewall-cmd --permanent --add-port=9092/tcp
firewall-cmd --reload

Configure Filebeat in Kafka machine.

Configuration of Yum for electicsearch

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

vi /etc/yum.repos.d/elasticsearch.repo
[elasticsearch]
name=Elasticsearch repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

yum install filebeat
systemctl enable filebeat
systemctl start filebeat

vi /etc/filebeat/filebeat.yml

change 
#output.elasticsearch:
 #hosts: ["localhost:9200"]
output.kafka:
 hosts: ["10.176.25.135:9092"]
 topic: "filebeat"
 codec.json:
   pretty: false

filebeat modules enable system
filebeat setup --pipelines --modules system
systemctl restart filebeat

INSTALL ELASTICSEARCH MASTER NODE

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

vi /etc/yum.repos.d/elasticsearch.repo
 [elasticsearch]
 name=Elasticsearch repository for 7.x packages
 baseurl=https://artifacts.elastic.co/packages/7.x/yum
 gpgcheck=1
 gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
 enabled=1
 autorefresh=1
 type=rpm-md

yum install java-1.8.0-openjdk
yum install java-1.8.0-openjdk-devel

yum install elasticsearch
vi /etc/elasticsearch/elasticsearch.yml
 network.host: 0.0.0.0
 discovery.seed_hosts: ["10.176.25.135", "127.0.0.1"]                ## Just one Node in UAT ###
 . . .
 Save and close elasticsearch.yml. 
 Change Value of java Heap Size in /etc/elasticsearch/jvm.options
 -Xms1g
 -Xmx1g
to 
 #-Xms1g
 #-Xmx1g
 -Xms4g
 -Xmx4g

 Then, start the Elasticsearch service with systemctl:
 systemctl start elasticsearch
 Next, run the following command to enable Elasticsearch to start up every time your server boots:
 systemctl enable elasticsearch
 You can test whether your Elasticsearch service is running by sending an HTTP request:

 curl -X GET "localhost:9200"

INSTALL KIBANA

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

vi /etc/yum.repos.d/elasticsearch.repo
[elasticsearch]
name=Elasticsearch repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

yum install java-1.8.0-openjdk
yum install java-1.8.0-openjdk-devel

yum install kibana

vi /etc/kibana/kibana.yml and change 
elasticsearch.hosts: ["http://dcplelkesmas01:9200"]

sudo systemctl enable kibana
sudo systemctl start kibana

Nginx installation on Kibana

wget https://dl.fedoraproject.org/pub/epel/epel-release-latest-7.noarch.rpm
yum install epel-release-latest-7.noarch.rpm
yum install nginx
systemctl status nginx
systemctl enable  nginx
systemctl start  nginx
Now nginx as reverse proxy should be running on the Server . 

Set the Selinux and Firewall setting 
setsebool httpd_can_network_connect 1 -P
firewall-cmd --zone=public --permanent --add-service=http
firewall-cmd --reload 

Then check the configuration for syntax errors:

nginx -t
systemctl restart nginx //to restart

INSTALL LOGSTASH

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

vi /etc/yum.repos.d/elasticsearch.repo 
[elasticsearch]
name=Elasticsearch repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

yum install java-1.8.0-openjdk
yum install java-1.8.0-openjdk-devel 
yum install logstash

sudo vi /etc/logstash/conf.d/02-beats-input.conf
Insert the following input configuration. This specifies a beats input that will listen on TCP port 5044.
/etc/logstash/conf.d/02-beats-input.conf
input {
 beats {
   port => 5044
  }
}

/etc/logstash/conf.d/10-syslog-filter.conf
filter {
 if [fileset][module] == "system" {
   if [fileset][name] == "auth" {
     grok {
       match => { "message" => ["%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth] [ssh][event]} %{DATA:[system][auth][ssh][method]} for (invalid user )?%{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]} port %{NUMBER:[system][auth][ssh][port]} ssh2(: %{GREEDYDATA:[system][auth][ssh][signature]})?",
                 "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: %{DATA:[system][auth][ssh][event]} user %{DATA:[system][auth][user]} from %{IPORHOST:[system][auth][ssh][ip]}",
                 "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sshd(?:\[%{POSINT:[system][auth][pid]}\])?: Did not receive identification string  from %{IPORHOST:[system][auth][ssh][dropped_ip]}",
                 "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} sudo(?:\[%{POSINT:[system][auth][pid]}\])?: \s*%{DATA:[system][auth][user]} :( %{DATA:[system][auth][sudo][error]} ;)? TTY=%{DATA:[system][auth][sudo][tty]} ; PWD=%{DATA:[system][auth][sudo][pwd]} ; USER=%{DATA:[system][auth][sudo][user]} ; COMMAND=%{GREEDYDATA:[system][auth][sudo][command]}",
                 "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} groupadd(?:\[%{POSINT:[system][auth][pid]}\])?: new group: name=%{DATA:system.auth.groupadd.name}, GID=%{NUMBER:system.auth.groupadd.gid}",
                 "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} useradd(?:\[%{POSINT:[system][auth][pid]}\])?: new user: name=%{DATA:[system][auth][user][add][name]}, UID=%{NUMBER:[system][auth][user][add][uid]}, GID=%{NUMBER:[system][auth][user][add][gid]}, home=%{DATA:[system][auth][user][add][home]}, shell=%{DATA:[system][auth][user][add][shell]}$",
                 "%{SYSLOGTIMESTAMP:[system][auth][timestamp]} %{SYSLOGHOST:[system][auth][hostname]} %{DATA:[system][auth][program]}(?:\[%{POSINT:[system][auth][pid]}\])?: %{GREEDYMULTILINE:[system][auth][message]}"] }
       pattern_definitions => {
         "GREEDYMULTILINE"=> "(.|\n)*"
       }
       remove_field => "message"
     }
     date {
       match => [ "[system][auth][timestamp]", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
     }
     geoip {
       source => "[system][auth][ssh][ip]"
       target => "[system][auth][ssh][geoip]"
     }
   }
   else if [fileset][name] == "syslog" {
     grok {
       match => { "message" => ["%{SYSLOGTIMESTAMP:[system][syslog][timestamp]} %{SYSLOGHOST:[system][syslog][hostname]} %{DATA:[system][syslog][program]}(?:\[%{POSINT:[system][syslog] [pid]}\])?: %{GREEDYMULTILINE:[system][syslog][message]}"] }
       pattern_definitions => { "GREEDYMULTILINE" => "(.|\n)*" }
       remove_field => "message"
     }
     date {
       match => [ "[system][syslog][timestamp]", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss" ]
     }
   }
  }
}

vi /etc/logstash/conf.d/30-elasticsearch-output.conf
input {
  kafka {
   bootstrap_servers => "dcplekafka01:9092, dcplekafka02:9092"
   topics => ["filebeat", "oracle"]
   codec => json
  }
}

filter {

 mutate {
   add_field => {"[@metadata][index]" => "%{[kafka][topic]}"}
  }
}

output {
 elasticsearch {
   hosts => ["dcplelkesmas01:9200", "dcplelkesmas02:9200", "dcplelkesmas03:9200"]
   index => "logstash-%{[@metadata][index]}-%{+YYYY.MM.dd}"
  }
}
Save and close the file.
If you want to add filters for other applications that use the Filebeat input, be sure to name the files so they’re sorted between the input and the output configuration, meaning that   the file names should begin with a two-digit number between 02 and 30 .
Test your Logstash configuration with this command:
sudo -u logstash /usr/share/logstash/bin/logstash --path.settings /etc/logstash -t

ES master and Data node setup

On all master nodes, Update elasticsearch.yml

cluster.name: es-cluster
node.name: ${HOSTNAME}
node.master: true
node.data: false 
##bootstrap.memory_lock: true
network.host: 0.0.0.0
cluster.initial_master_nodes: ["dcplelkesmas01", "dcplelkesmas02", "dcplelkesmas03"]
discovery.zen.minimum_master_nodes: 2
discovery.zen.ping.unicast.hosts: ["dcplelkesmas01", "dcplelkesmas02", "dcplelkesmas03"]

Next,

cat > /etc/default/elasticsearch << EOF
ES_STARTUP_SLEEP_TIME=5

MAX_OPEN_FILES=131070 MAX_LOCKED_MEMORY=unlimited EOF

cat > /etc/security/limits.conf << EOF
elasticsearch soft memlock unlimited
elasticsearch hard memlock unlimited
EOF

sysctl -w vm.max_map_count=262144
sysctl -p /etc/sysctl.conf 
mkdir /usr/share/elasticsearch/data
chown -R elasticsearch:elasticsearch /usr/share/elasticsearch/data 

systemctl daemon-reload
systemctl enable elasticsearch
systemctl restart elasticsearch 

On Data Nodes

chown -R elasticsearch:elasticsearch /data
mkdir /data/nodes
chown -R elasticsearch:elasticsearch /data/nodes

Add following /etc/elasticsearch/elasticsearch.yml

cluster.name: es-cluster
node.name: ${HOSTNAME}
node.master: false
node.data: true
path.data: /data
network.host: 0.0.0.0
cluster.initial_master_nodes: ["dcplelkesmas01", "dcplelkesmas02", "dcplelkesmas03"]
discovery.zen.minimum_master_nodes: 2
discovery.zen.ping.unicast.hosts: ["dcplelkesmas01", "dcplelkesmas02", "dcplelkesmas03"]

Remove node.path /var/lib/elasticsearch

On Logstash Nodes

vi /etc/logstash/conf.d/30-elasticsearch-output.conf

input {
 kafka {
   bootstrap_servers => "dcplekafka01:9092, dcplekafka02:9092"
   topics => ["filebeat", "oracle"]
   codec => json
 }
}

filter {    
 mutate {
   add_field => {"[@metadata][index]" => "%{[kafka][topic]}"}
 }
}

output {
 elasticsearch {
   hosts => ["dcplelkesmas01:9200", "dcplelkesmas02:9200", "dcplelkesmas03:9200"]
   index => "logstash-%{[@metadata][index]}-%{+YYYY.MM.dd}"
 }
}

ELK nodes integration with Cluster Setup Zookeeper Cluster for Kafka On Both Kafka

mkdir /home/kafka
mkdir /home/zookeeper
firewall-cmd --zone=trusted --add-source=10.176.25.233 --permanent
firewall-cmd --zone=trusted --add-source=10.176.25.234 --permanent

On Kafka Node 1

cd /home/zookeeper
touch myid
echo "1" >> myid
For /opt/kafka/config/zookeeper.properties on Kafka01
dataDir=/home/zookeeper
clientPort=2181
maxClientCnxns=0
initLimit=5
syncLimit=2
tickTime=2000
# list of servers
server.1=0.0.0.0:2888:3888
server.2=10.176.25.234:2888:3888
Restart zookeeper
systemctl restart zookeeper

On Kafka Node 2

cd /home/zookeeper
touch myid
echo "2" >> myid
For /opt/kafka/config/zookeeper.properties on kafka02
dataDir=/home/zookeeper
clientPort=2181
maxClientCnxns=0
initLimit=5
syncLimit=2
tickTime=2000
# list of servers
server.1=10.176.25.233:3888
server.2=0.0.0.0:2888:3888

Restart zookeeper

systemctl restart zookeeper

Now, Set Kafka to connect to multi node zookeeper cluster

For /opt/kafka/config/server.properties for kafka01

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=10

1. A comma separated list of directories under which to store log files

log.dirs=/home/kafka
#advertised.host.name=10.176.25.233

1. Add all 2 zookeeper instances IP’s here

zookeeper.connect=localhost:2181,10.176.25.234:2181
zookeeper.connection.timeout.ms=6000

1. Restart Kafka - systemctl restart kafka.service

For /opt/kafka/config/server.properties for kafka02

1. The id of the broker. This must be set to a unique integer for each broker.

broker.id=20

1. A comma seperated list of directories under which to store log files

log.dirs=/home/kafka
#advertised.host.name=10.176.25.234

# Add all 2 zookeeper instances IP’s here
zookeeper.connect=localhost:2181,10.176.25.233:2181
zookeeper.connection.timeout.ms=6000

1. Restart Kafka - systemctl restart kafka.service

Testing

On the first machine, produce some messages on the test topic. You can use this command.

$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
Now, Type some messages

On the second machine, consume those messages.

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

You should be receiving the messages you have typed in the producer command.