In a previous post, we went through a few input plugins like the file input plugin, the TCP/UDP input plugins, etc for collecting data using Logstash. In this post, we will see a few more useful input plugins like the HTTP, HTTP poller, dead letter queue, twitter input plugins, and see how these input plugins work.
This will clone the repository to the folder /etc/logstash/conf.d/logstash-input-plugins folder.
Heartbeat Input plugin
Heartbeat input plugin is one of the simple but most useful input plugins available for Logstash. This is one simple way by which we can check whether the Logstash is up and running without any issues. Put simply, we can check Logstash’s pulse using this plugin.
This will send periodic messages to the target Elasticsearch or any other destination. The intervals at which these messages are sent can also be defined by us.
Let’s look at the configuration for sending a status message every 5 seconds, by going to the following link:
You can see the field named “message” with the specified string “ok” in the document.
Now let’s explore other two different options that can be given to the “message” setting.
If we give the value “epoch” in the “message” setting, it will emit the time of the event as an epoch value under a field called “clock”. Let’s replace the “ok” with the “epoch” for the “message” setting in our configuration.
Give some time for Logstash to run. After it has started running, wait for some time and then hit CTRL+C to quit the Logstash screen and to stop it. Let’s inspect a single document from the index which we created now by typing in the request:
In this document, you can see that there is a field called “clock” which has the time in “epoch” representation. This is really helpful in scenarios where we need to calculate the delays in the ingestion pipelines using Logstash. The actual time of generation of an event and the time of ingestion can be subtracted to find the time delay in ingestion.
Also, the “message” setting can be set to have the value “sequence”, which basically will generate incrementing numbers in sequence under the field “clock”. The configuration of the logstash for this can be found in this link:
Here the field “clock” can be seen with incrementing numbers and has been sent in every 5 seconds. This is also helpful in determining missed events.
Generate Input plugin
Sometimes for testing or other purposes it would be great to have some specified data to be inserted. The conditions might vary like you might want to have some specific data generated for longer times, or some data generated for some iterations etc. Of course we can write custom scripts or programs to generate such data and send to a file or even Logstash directly. But there is a much more simple way to do such things within Logstash itself. We can use the generator plugin for logstash to generate random/custom log events.
Let’s dive into an example first.
The configuration for the example can be seen in the link
In the configuration, under the “lines” section, two JSON documents were given and also for the Logstash to understand it is JSON, we have specified the “codec” value as JSON. Now, “count” parameter is set to 0, which basically tells the Logstash to generate an infinite number of events with the values in the “lines” array. If we set any other number for the “count” parameter, Logstash will generate each line that many times.
Let’s run Logstash with this configuration file using the command:
This plugin is very useful when we need to have documents generated for testing. We can give multiple lines and test grok patterns or filters and see how they are getting indexed using this plugin.
Dead letter queue
This is one important input plugin and is an immensely helpful one too. So far we have seen the cases of successful processing of events by Logstash. Now what happens when Logstash fails to process an event? Normally the event is dropped and hence we will lose it.
In many cases it would be helpful to collect those documents for further inspection and do the necessary actions so that in future the event dropping won’t happen.
To make this possible Logstash provides us with a queue called dead-letter-queue. The events which are dropped would get collected here.
Now, once in the dead-letter-queue, with the help of this dead-letter-queue input plugin , we can process these documents using another Logstash configuration and make the necessary changes and then index them back to Elasticsearch.
You can see the flow of unprocessed events to dead-letter-queue and then to Elasticsearch in this diagram.
So that we have seen what is “dead-letter-queue” in Logstash, let’s move on to a simple use case involving the dead letter queue input plugin.
Since we are using the dead-letter-queue input plugin, we need to do two things prior. First, we need to enable the dead-letter-queue.
Second, we need to set the path to the dead-letter-queue.
Let’s create a folder named dlq to store the dead-letter-queue data by typing in the following command
mkdir /home/student/dlq
Now, Logstash creates a user named “logstash” during the installation and performs the actions using this user. So let’s change the ownership of the above folder to the user named “logstash” by typing in the following command
sudo chown -R logstash:logstash /home/student/dlq
Now let’s enable the dead_letter_queue and also specify its path in the logstash’s setting by editing the settings file by typing in
sudo nano /etc/logstash/logstash.yml
Now uncomment the setting dead_letter_queue and give the value as true
Also give set the path to store the queue data, by uncommenting the path.dead_letter_queue as /home/student/dlq
The sections of the logstash.yml file will look like this after we have made the necessary edits
You can see that, in the first 10 documents, the “age” field has integer values. But from the 11th document, it is mistakenly given as boolean. So what happens is that Elasticsearch will assign a type “long” data type used for the integers for the “age” field. This is because, when we are indexing the documents the first document which has the “age” as integers are indexed first, so Elasticsearch will assign them the “age” field. Now, when the 11th document comes with the boolean value for “age” , Elasticsearch will reject the document since a field cannot contain two data types. So it does for all the documents with “age” as a boolean value.
So let’s index the sample data and see what happens.
For that let’s run the configuration file for logstash. Let’s have a look at the configuration file by going to the following link:
In the command we typed, there is an extra parameter called “path.settings” and its value “/etc/logstash”. This is given so that Logstash will understand where the logstash.yml file is located and will take that settings. Give some time for Logstash to run. After it has started running, wait for some time and then hit CTRL+C to quit the Logstash screen and to stop it.
Let’s inspect how many documents were inserted to the index. There was a total of 20 documents and out of it only 10 should have been indexed as the other 10 would have the mapping issue.
In the response you can see that the hits.total.value is equal to 10 (marked in bold in the above response).
Now you can see that the remaining 10 documents are not indexed.So these documents are dropped and since we have enabled the dead_letter_queue, these documents should be there. Let’s use the input plugin for reading from the dead_letter_queue.
The configuration for reading from the dead_letter_queue can be seen in this link:
In the configuration, we have pointed to the path where we had set the dead_letter_queue. Now if we run the configuration Logstash will read the contents in the dead_letter_queue and process it and then index to Elasticsearch.
For the sake of simplicity , in the configuration, I have not added any filters. We definitely can add filters depending up on the use case and then run the configuration if we want.
Let’s run Logstash with this configuration file using the command:
You can see that there are 10 documents that were indexed. And upon inspection on the sample document, it has returned the document which had the boolean value for the field “age”.
Note that with this configuration, after running the Logstash, the queue contents would not be cleared. If you want the documents which were read not to be indexed again, the flag “commit_offsets” to true, just under the “path” setting.
Http_poller plugin
The http_poller plugin is a very useful input plugin that comes with Logstash. As the name indicates, this plugin can be used to poll http endpoints periodically and let us store the response in Elasticsearch.
This can be helpful if we have health APIs exposed for our applications and need periodic status monitoring of them. Also we can use it to collect the status like weather, match details etc periodically.
In our case let’s look at polling two http APIs for the purpose of understanding the plugin. We will first call a POST method to an external API at an interval of 5 seconds, and will store it to an index named “http-poller-api”. Then we will also call the Elasticsearch’s cluster health API, which is a GET request , in every second and store it in an index named “http-poller-es-health”.
Let’s first familiarize with both of the APIs first.
The first API is a simple online free API, which gives some response when we give a POST request.
Open the website apitester.com and a web page will open like this. We open this website simply to test one of the APIs we are trying to call.
Now, fill the box marked as URL with this address https://jsonplaceholder.typicode.com/posts In the section post data, add the following JSON lines
{ "title": "foo", "body": "bar", "userId": "1"}
And add “content-type” and “application/json” as the key value pairs for the boxes named “name” and “value” respectively. The filled form details would look like this:
What this API does is that it will create a record in the remote database and send the details with the record id as the response body. You can also see a pretty long response header also along with the response.
Our second API is a GET request supplied by our own Elasticsearch itself, which gives us the cluster status. You can test the API by typing in the following request in the terminal
In the configuration you can see two “http_poller” sections. One is named as “external_api” and the other is named as “es_health_status”. In the “external_api” section , you can see the URL we have called in the first example, with the method, and the content-type mentioned in the “headers” section. Apart from that, I have given an identifier named “external-api” in the “tags” to identify the response from this section.
Also the periodic scheduling is done every 5 seconds. This is specified under the “schedule” section. Another important setting is the “metadata_target”. I have specified that to be “http_poller_metadata”. This will ensure that the response headers will be stored in the field named “http_poller_metadata” when storing in ES.
Similar settings are applied to the “es_health_status” section too. Here the difference is that the “method” is a GET method and also the “Schedule” section is given a cron expression as the value. Also the value of “tags” is “es_health”.
Now coming to the output section, there is a check on the “tags” field. If the data is coming from the “external-api” tag, the index where the data is stored is “http-poller-api”. But if the data contains “tags” value as “es_health”, it will be stored in the index named “http-poller-es-health”.
Since we have a good understanding of the Logstash configuration, let’s run this configuration by typing in this command
Wait for some time for the logstash to run. Since it is a polling operation, Logstash won’t finish running,but will keep running. So after about 5 mins, you can stop the Logstash by hitting CTRL+C
Now let’s check whether we have data in both of the indices. Lets first query the index “http-poller-api” with this query:
In the response, you can see the “http_poller_metadata” field with the necessary details of both the request and response headers. Also there is the field named “id” returned by the server.
Now coming to the index “http-poller-es-health”, lets query the same query like this:
Here also you can see the same. So by using this input plugin we can call HTTP calls periodically index the response with necessary data to a single index or multiple indices.
Twitter plugin
Another interesting input plugin which is provided by Logstash is the Twitter plugin. Twitter input plugin allows us to stream Twitter events directly to Elasticsearch or any output that Logstash support.
Let’s explore the Twitter input plugin and see it in action.
For this to work, you need to have a Twitter account. If you don’t have one, don’t worry, just go to Twitter and create one using the signup option.
Click on the “create an app” button. If you are creating an app for the first time, it would ask to “apply” and it might take a day or two to get the approval. So once you have the approval, the “create an app” clicking would redirect to a screen like this:
“api key” which we copied to the notepad to that of the “consumer_key”
“api key secret” to that of the “consumer_secret”
“Access token” to that of the “oauth_token”
“Access token secret” to that of the “oauth_token_secret”
You can enter any string values in the “keywords” section which will get you the tweets which contains those strings. Here I have given “covid”, “corona” as the strings. This will give tweets which contain either “covid” OR “corona” in their body.
The “full_tweet” value is set to true to get the full information about the tweet as given by the Twitter
Let’s run Logstash with this configuration file using the command:
As you can see there is a lot of information per tweet. It would be interesting to collect tweets on the topic you are interested and do some analytics on them.
Cleaning up
Now let’s clean up the indices we have created in this data. You can keep them if you want.
curl -XDELETE heartbeat
curl -XDELETE generator
curl -XDELETE dlq-sample-data
curl -XDELETE dlq-01
curl -XDELETE http-poller-api
curl -XDELETE http-poller-es-health
curl -XDELETE twitter
Conclusion
In this post, we have seen some of the most common and helpful input plugins out there. There are many more input plugins in Logstash’s arsenal, but most of them follow similar patterns of the ones we have seen.
It is commonplace for organizations to restrict their IT systems from having direct or unsolicited access to external networks or the Internet, with network proxies serving…
AWS Systems Manager and CloudWatch Agent provide an integrated approach to observability and managing your AWS infrastructure efficiently. In this tutorial, I will show you how…
Infrastructure as Code is an increasingly popular DevOps paradigm. IaC has the ability to abstract away the details of server provisioning. This tutorial will look at…
Oh no!We didn’t find anything for “”,
