How to Log and Analyze MQTT Messages
with the ELK Stack?

Introduction

The ELK Stack (Elasticsearch, Logstash, and Kibana) is a powerful open-source log management and analytics platform. Elasticsearch, its core component, supports replication, allowing data to be copied across multiple nodes. This ensures high availability by preventing data loss in case of node failures.

Replication in Elasticsearch improves query performance by distributing search requests across replicas. It also enhances scalability, allowing systems to handle more queries efficiently. By default, Elasticsearch creates one replica per shard, but this can be adjusted based on requirements. Proper replication strategies help in load balancing, fault tolerance, and overall system reliability.

Elasticsearch

Elasticsearch is a distributed search and analytics engine designed for handling large volumes of data in real time. It stores data in a NoSQL format using JSON documents and provides fast full-text search capabilities. Data is indexed using an inverted index, allowing quick retrieval and analysis. Elasticsearch is scalable, supporting horizontal scaling by adding more nodes to a cluster. It offers replication and sharding, ensuring data redundancy and high availability. Common use cases include log analysis, security monitoring, and business intelligence.

Logstash

Logstash is a data processing pipeline that collects, transforms, and sends data to various destinations, including Elasticsearch. It can ingest data from multiple sources such as logs, metrics, databases, and message queues. Logstash processes data using a powerful filtering system, allowing transformation, enrichment, and parsing. It supports various plugins to handle different data formats, including JSON, CSV, and syslog. Logstash is often used for log collection, data transformation, and real-time data processing before storage.

Kibana

Kibana is a visualization and analytics tool designed to interact with data stored in Elasticsearch. It provides an intuitive dashboard interface for creating visualizations like charts, graphs, and maps. Kibana enables real-time data exploration, making it easy to analyze logs and monitor system performance. It includes features like machine learning, security monitoring, and alerting. Users can build interactive dashboards and set up alerts for anomalies in data. Kibana is widely used for log analysis, security event monitoring, and business intelligence reporting.

Before we begin, ensure you have:

• A Linux server (Ubuntu 20.04 recommended)
• MQTTRoute installed and running
• Java 8 or later (required for ELK and Kafka)
• Elasticsearch, Logstash, and Kibana (ELK Stack) installed

Step 1: Install Java 8 for ELK

• Update the package list : sudo apt update
• Install OpenJDK 8 : sudo apt install -y openjdk-8-jdk
• Verify the instalation : java -version
• Ensure the output shows java8 installed.

If Java is installed on the machine, the following output will be displayed:

Step 2: Install and Configure the ELK Stack

• Install Elasticsearch :

Check the status to see if Elasticsearch is running properly; the result should appear as shown below:

• elasticsearch.service - Elasticsearch

Generate Password for elasticsearch. By Default the username and password is necessary to connect the elasticsearch version 8.17.3

Use the following command to generate the Elasticsearch password:

Verify Installation:

After the password is generated successfully, run the following command to verify it:

If the elasticsearch has properly configured the below results exists :

TLS/SSL configuration in elasticsearch 8.17.3 :

Elasticsearch 8.17.3 includes built-in security and requires TLS/SSL for communication. Follow these steps to configure TLS/SSL for both HTTP and transport layers.

• Generate SSL certificate :

Elasticsearch includes a built-in tool to generate certificates. Use the following command to generate a certificate:

Extract the certificates :

2. Configure the elasticsearch for TLS/SSL:

Edit the elasticsearch configuration file using the below command:

Add or update the settings:

If the features below are not present in elasticsearch.yml, add them. If they already exist, set or enable them to true.

# Enable security features
xpack.security.enabled: true
xpack.security.enrollment.enabled: true

# Enable encryption and mutual authentication between cluster nodes
xpack.security.transport.ssl:
enabled: true
verification_mode: certificate
keystore.path: certs/transport.p12
truststore.path: certs/transport.p12

# Create a new cluster with the current node only
# Additional nodes can still join the cluster later
cluster.initial_master_nodes: ["hostname"]

# Allow HTTP API connections from anywhere
# Connections are encrypted and require user authentication
http.host: 0.0.0.0

# Allow other nodes to join the cluster from anywhere
# Connections are encrypted and mutually authenticated
#transport.host: 0.0.0.0

Restart Elasticsearch :

Restart the elasticsearch to apply the changes :

Afterward, check whether Elasticsearch is running using the command below: :

Verify Elasticsearch TLS/SSL Configuration :

By default our certificate exists in the below directory

Use the following command to verify the TLS/SSL installation:

When Elasticsearch is configured correctly for TLS/SSL, the result appears as shown below:

Install Logstash :

Download Logstash 8.17.3

Install Logstash 8.17.3

Verify Logstash Installation :

Check whether Logstash version 8.17.3 is installed using the command below. If the version is displayed, it means Logstash is installed correctly on our machine.

The result looks like :

Start and Enable logstash :

Check the status for Logstash installation :

The output is as follows:

Check Logstash Logs in Log File

1. Logstash writes logs to /var/log/logstash/logstash-plain.log. View logs using:

2. To check Logstash logs is using journalctl

Download Kibana 8.17.3

To download Kibana 8.17.3, use the below command:

Install Kibana 8.17.3 :

Configure kibana 8.17.3 :

Follow the below steps to configure Kibana
Step-1:

Step-2:
Uncomment and modify the below setting :

After modifying the settings, save and exit the configuration.

Start and Enable Kibana:

To check whether Kibana is configured properly, the results appear as shown below:

• kibana.service - Kibana

Open kibana in browser :

Once Kibana is configured properly, open it using the following URL:

http://<server_ip>:5601

For example :http://localhost:5601

Configure kibana 8.17.3 with TLS/SSL:

If Elasticsearch is already configured with TLS/SSL (HTTPS), you need to configure Kibana to communicate securely with it.

Now, let's see how to set up Kibana with TLS/SSL:

Step-1 :

Step-2 :
Copy the certficate from the elasticsearch certs directory to the kibana certs directory :

Set the corrector permissions for certificates :

chown kibana:kibana /etc/kibana/elastic-stack-ca.crt

chmod 600 /etc/kibana/elastic-stack-ca.crt

Verify the file ownership and permissions :

ls -l /etc/kibana/elastic-stack-ca.crt

The output appears as follows:
-rwxr-x--- 1 kibana kibana 1915 Mar 25 13:56 /etc/kibana/elastic-stack-ca.crt

Step-3 :
Uncomment and modify the below setting :

server.port: 5601
server.host : "localhost"
elasticsearch.hosts: ["https://localhost:9200"]
elasticsearch.username: "kibana_system"
elasticsearch.password: "password"
elasticsearch.maxSockets: 1024
elasticsearch.ssl.certificateAuthorities: [ "/etc/kibana/http_ca.crt" ]

Finally save and exit the kibana configurations.

If you want to access Kibana over HTTPS instead of http://localhost:5601, configure TLS in kibana.yml:

Add these settings:

server.ssl.enabled: true
server.ssl.certificate: "/etc/kibana/kibana.crt"
server.ssl.key: "/etc/kibana/kibana.key"

Step-4 :
Restart the Kibana to get the changes applied:

If you do not have the Elasticsearch user, use the command below to manually reset the password.

Manually Set Password for User

Use the following command in your terminal:

step-4 : Try Accessing Kibana

If you're using Kibana HTTPS, use:

https://localhost:5601

If Kibana SSL is disabled, use:

http://localhost:5601

If Kibana is running on a remote server, replace localhost with the server's IP:

http://your-server-ip:5601

Download MQTT Broker and Running in our machine and it looks like the

Elasticsearch Connector Configuration

The Elasticsearch Connector in CrystalMQ allows seamless data retrieval from the broker and efficient indexing into Elasticsearch. It ensures secure authentication, encrypted communication, and reliable data storage, making it an essential component for managing and analyzing IoT data.

• Host: Enter the hostname or IP address of the Elasticsearch server.
• Port: Specify the port on which the Elasticsearch server is listening (default: 9200).
• Index: Enter the name of the Elasticsearch index for storing or retrieving data.
• Authentication: Select the authentication method:
  Basic Auth:Authenticate using a username and password.
  PKI Auth:Authenticate using Public Key Infrastructure (PKI) with client certificates.
  JWT Auth:Authenticate using a JSON Web Token (JWT).
• Access Token: If using JWT authentication, enter a valid JWT access token.
• SSL: Enable SSL/TLS for a secure connection. If enabled, provide the required certificate files.
• CA Certificate: Upload the Certificate Authority (CA) certificate if using a self-signed certificate.
• Client Certificate & Key: If using PKI authentication, upload the required client certificate and client key

1. Authentication in Elasticsearch Connector

Elasticsearch supports multiple authentication methods to ensure secure access.

Basic Authentication

By default, Elasticsearch requires a username and password for authentication. Without valid credentials, the connection will fail, leading to authentication errors. This method is simple and widely used for securing access.

PKI (Public Key Infrastructure) Authentication

PKI authentication uses digital certificates instead of passwords for secure access. In this approach:

• A Client Certificate and Client Key are required to establish a trusted connection with Elasticsearch.
• The server verifies the certificate against a trusted Certificate Authority (CA) to ensure authenticity.
• This method enhances security by eliminating the need for passwords while ensuring encrypted and authenticated communication.

JWT (JSON Web Token) Authentication

JWT authentication provides token-based access control, where an Access Token is used instead of traditional credentials.

• Authentication providers like Google or Keycloak generate an access token for users.
• Elasticsearch validates the token using a secret key stored in its configuration.
• This method enables single sign-on (SSO) and secure, scalable authentication.

2. Encryption: SSL/TLS in Elasticsearch

To protect data in transit, Elasticsearch supports SSL/TLS encryption, ensuring that communication between clients and the server remains private and secure.

Setting Up SSL/TLS Encryption

• OpenSSL is used to generate security certificates required for encrypted communication.
• Elasticsearch requires specific certificates:

  1. CA Certificate - A trusted certificate authority that signs other certificates.

  2. Client Certificate - Identifies the client to Elasticsearch.

  3. Client Key - A private key used to authenticate the client certificate.

By configuring authentication and encryption properly, the Elasticsearch Connector in CrystalMQ provides a secure, scalable, and efficient solution for managing IoT data.

In the Elasticsearch connector, basic authentication (username and password) is required, so it can be configured as the default.

Enter the configurations in the Elasticsearch connector, including the Elasticsearch index, to establish the connection.

Once all configuration fields are completed, select 'Save' to save the configurations to the database.

If the Elasticsearch connector is connected successfully, the Enable/Disable toggle button will show green. If there is an issue with the connector, the button will display red.

In the above configuration, only Basic Authentication is used, without enabling TLS/SSL encryption.

Use Logstash to view data from Elasticsearch. Let’s see how we can use the Logstash pipeline to capture this data.

Logstash offers various methods to output data, depending on how you want to store, visualize, or forward it. Some common output options include:

• Elasticsearch
• File
• Standard Output (stdout)
• Kafka
• Redis
• AWS S3 / Google Cloud Storage
• InfluxDB
• Email
• TCP / UDP
• Http
• Graphite
• WebSocket
• Google BigQuery

Step -1 :

Configure Logstash to retrieve data from Elasticsearch:

• Create a Logstash configuration file
  Create a file named elasticsearch_logstash.conf in the Logstash
  configuration directory

• Edit the Configuration File
  Open elasticsearch_logstash.conf and add the following configuration:
  

Run Logstash

Start Logstash using the following command:

In case Logstash has been installed through a package manager:

For example : sudo /usr/share/logstash/bin/logstash -f

~/Documents/elasticsearch_logstash.conf

After configuring Logstash, use the command below to verify that the configuration has been applied correctly.

If the configuration is correct, the output is as follows:

Check the Logstash logs using the command below to verify that data is being retrieved from Elasticsearch and displayed in Logstash.

The Logstash log appears as shown below:

Using bundled JDK: /usr/share/logstash/jdk

Use Kibana to visualize the data once Elasticsearch has received it from the CrystalMQ Elasticsearch Connector.

Step-1:

Open the browser url like http://<host_ip>:5601

Step-2: Create an Index Pattern

Before visualizing data, you need to define an index pattern to access your MQTT logs.

• Go to "Stack Management" → "Index Patterns".
• Click "Create Index Pattern".
• Enter your index name (e.g., offset_mqtt-*).
• Select the timestamp field (@timestamp if available).
• Click "Create index pattern".

Step-3: Chart for MQTT Messages Over Time

Before visualizing data, you need to define an index pattern to access your MQTT logs.

• Go to "Visualize Library" → Click "Create visualization".
• Select "Line Chart".
• Choose your index pattern (offset_mqtt-*).
• Configure:
  X-axis: Select @timestamp and set interval (e.g., 1m, 1h).
  Y-axis: Select Count to count MQTT messages.
• Click "Save" and name your visualization.

Step-4: The chart looks as follows:

Bar chart :

Pie-chart :

Table View :

Now, we will use Elasticsearch with TLS/SSL encryption to establish a connection and send messages through Data Connectors in CrystalMQ:

Step-1 : Verify that all certificate paths in the Elasticsearch configuration are provided correctly. If any changes are made, restart Elasticsearch to apply them.

Step-2 : Open the Kibana configuration file and verify that all settings are properly configured for TLS/SSL.

Step-3 : Run CrystalMQ and use the Elasticsearch connector in Data Connectors to connect to Elasticsearch with SSL/TLS encryption.

Step-4 : For SSL/TLS encryption, the CA root certificate is required to connect to Elasticsearch. The client certificate and client key are optional—uploading just the CA root certificate is sufficient to establish the connection.

Step-5 : After uploading all the certificates, save the configuration and enable the connector using the Enable/Disable toggle button.

Step-6 : The Elasticsearch connector with TLS/SSL encryption has now been successfully connected in CrystalMQ.

Use Logstash to view the logs from the Elasticsearch connector:

Configure Logstash to Retrieve Data from Elasticsearch:

• Create a Logstash configuration file
  Create a file named elasticsearch_logstash.conf in the Logstash
  configuration directory
• Edit the Configuration File
  Open elasticsearch_logstash.conf and add the following configuration:

Save the configuration and use the command below to verify whether it has been established correctly:

Use Logstash to capture the data from elasticsearch with TLS/SSL :

If you've installed Logstash via a package manager:

For example : sudo /usr/share/logstash/bin/logstash -f
~/Documents/elasticsearch_logstash.conf

Now Configure the Kibana with elasticsearch TLS/SSL:

Refer to the steps above to configure kibana.yml for Elasticsearch with TLS/SSL encryption.

Open the browser at https://<host_ip>:5601 and visualize the data in Kibana using the instructions above, with options such as bar charts, pie charts, and data views.

In Elasticsearch version 8.17.3, we are using the basic license, so PKI and JWT cannot be used to establish the connection.

To use JWT and PKI authentication, an upgrade to the Platinum license is required.

Using PKI authentication in CrystalMQ:

In PKI authentication, the client certificate and client key are required, while the CA root certificate is optional.

Save the configuration and enable the connector toggle button to connect to Elasticsearch using PKI authentication.

Elasticsearch is now configured with PKI authentication.

If we want to encrypt the data, we can use SSL/TLS with PKI authentication.

To encrypt the data, SSL/TLS can be used along with PKI authentication.

In PKI authentication with SSL/TLS, the client certificate and client key are required, while for SSL-only connections, only the CA root certificate is needed to connect to Elasticsearch.

Using JWT authentication in CrystalMQ:

In JWT (JSON Web Tokens) authentication, the access token is used to connect to Elasticsearch.

Use an authentication provider, such as Keycloak or Google Console, to obtain the access key.

Now, use the access key with the command below to obtain the access token:

In this command, the Keycloak authentication provider is used to obtain the authentication token.

After obtaining the access token, use it in the Elasticsearch data connector to connect to Elasticsearch using JWT authentication.

If we want to encrypt the data, we can use SSL/TLS with PKI authentication.

When using SSL/TLS with JWT, simply upload the CA root certificate to connect to Elasticsearch.