# How to send a json object to elasticsearch throw logstash
**URL:** https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516
**Category:** Logstash
**Created:** [March 21, 2017, 9:59pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516 "2017-03-21T21:59:52Z")
**Posts on this page:** 16
**Page:** 1
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 21, 2017, 9:59pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/1 "2017-03-21T21:59:52Z")
I am creating "bill" feature in my nodejs application that basically will save in Elasticsearch the username every time any user access any rest service.
I have struggling for the last two days with no success. It is my first time working with ELK.
Here is my last tentative.
step 1)
I start the logstash with ./logstash-5.2.2/bin/logstash -f "logstash.conf"
logstash.conf contains:
input {
tcp {
port =\> 5000
type =\> document\_type
}
}
filter {
grok {
match =\> { "message" =\> "data=%{GREEDYDATA:request}"}
}
json{
source =\> "request"
target =\> "parsedJson"
remove\_field=\>["message"]
}
mutate {
add\_field =\> {
"firstname" =\> "%{[parsedJson][firstname]}}"
"surname" =\> "%{[parsedJson][surname]}}"
}
}
kv {
source =\> "message"
remove\_field =\> ["message"]
}
}
output {
elasticsearch {
codec =\> "json"
hosts =\> ["127.0.0.1:9200"]
index =\> "my\_index\_previously\_mapped"
}
}
step 2)
nodejs sends the json object to logstash
var Logstash = require('logstash-client');
var logstash = new Logstash({
type: tcp,
host: localhost,
port: 5000
});
var user = {
firstname: req.body.username,
surname: req.body.surname
};
logstash.send(user);
step 3)
I was expecting to get "two columns", firstname and surname, the datetime from server when it was saved and not get message "column" which seems to me to be redudant (pointless in my scenario). The next step would be to agregate and filter something like how many access for certain user from this datetime to that datetime. But I am facing several issues naturally because I am dummy on ELK.
PS. Obviously, I am going to add few more info when finished but in order to keep things simple, lets say just log firstname and surname are enough.
First issue: why message "column" is still there?
Second issue: why firstname "column" just contain a static string "%{[parsedJson][surname]}}" (surname as well)?
Third issue: why @timestamp is always 3 hours ahead of my system datetime?
Fourth issue: I understand that I have to create a mapping if I want to aggregate. Why the bellow command is trying to re-create the index?
curl -XPUT '[http://localhost:9200/my\_index\_previously\_mapped/](http://localhost:9200/my_index_previously_mapped/)' -d '
> {
> "mappings" : {
> "my\_document\_type" : {
> "properties" : {
> "firstname" : { "type" : "text" } }
> }
> }
> }'
> {"error":{"root\_cause":[{"type":"index\_already\_exists\_exception","reason":"index [my\_index\_previously\_mapped/\_GiiT8JGSruBt9ytm8L6zQ] already exists","index\_uuid":"\_GiiT8JGSruBt9ytm8L6zQ","index":"my\_index\_previously\_mapped"}],"type":"index\_already\_exists\_exception","reason":"index [my\_index\_previously\_mapped/\_GiiT8JGSruBt9ytm8L6zQ] already exists","index\_uuid":"\_GiiT8JGSruBt9ytm8L6zQ","index":"my\_index\_previously\_mapped"},"status":400}
\*\*\* Added in March 22 2017 at 11am UTC -3
I started the logstash with
input {
tcp {
port =\> 5000
type =\> document\_type
}
}
output {
stdout { codec =\> rubydebug }
}
then I got:
{
"@timestamp" =\> 2017-03-22T13:43:01.443Z,
"port" =\> 58794,
"@version" =\> "1",
"host" =\> "127.0.0.1",
"message" =\> "{"firstname":"a","surname":"a"}",
"type" =\> "document\_type"
}
As a previous SQL and NoSql (mongodb) user, my intention is to get three "columns" in ElastiSearch. One for timestamp, another for firstname and other one for surname. Then I can search aggregating how many access were done by certain user in certain period or, just as didactic example, let's say query how many users with firstname john are in Elasticsearch ignoring the surname. If I can learrn how to achieve these two tasks it will probably be a large step forward.
\*\*\* Added March 22 at 12am UTC -3
{
"@timestamp" =\> 2017-03-22T14:56:53.064Z,
"port" =\> 33666,
"@version" =\> "1",
"host" =\> "127.0.0.1",
"message" =\> "{"firstname":"a","surname":"a"}",
"type" =\> "document\_type",
"tags" =\> [
[0] "\_grokparsefailure"
]
}
is the result of
filter {
grok {
match =\> { "message" =\> "%{TIMESTAMP\_ISO8601:time} \ %{GREEDYDATA:msg}" }
}
}
\*\*\* Added in Mar 22 at 12:10 PM UTC -3
my filter is:
filter {
json {
source =\> "message"
}
}
and the output is:
{
"firstname" =\> "a",
"@timestamp" =\> 2017-03-22T15:04:29.108Z,
"port" =\> 34102,
"surname" =\> "a",
"@version" =\> "1",
"host" =\> "127.0.0.1",
"message" =\> "{"firstname":"a","surname":"a"}",
"type" =\> "document\_type"
}
I successfully search using:
{"query":{"bool":{"must":[{"range":{"@timestamp":{"gte":"2017-03-22","lte":"2017-03-22"}}}],"must\_not":,"should":}},"from":0,"size":10,"sort":,"aggs":{}}
Nevertheless, I get such error while trying to aggregate:
... Fielddata is disabled on text fields by default. Set fielddata=true on [firstname] in order to load fielddata in memory by uninverting the inverted index...
Then, I tried:
curl -XPUT '[http://localhost:9200/greencard\_indice/\_mapping/cpfTipo](http://localhost:9200/greencard_indice/_mapping/cpfTipo)
> {
> "properties": {
> "firstname": {
> "type": "text",
> "fielddata": true
> }
> }
> }'
> curl: (3) [globbing] nested brace in column 75
and I am still getting the same error while aggregating.
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [March 22, 2017, 6:18am UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/2 "2017-03-22T06:18:03Z")
Remove `codec => "json"` from your elasticsearch output.
Please show exactly what you're posting to Logstash. Suggestion: Comment out all your filters and outputs and add a `stdout { codec => rubydebug }` output that'll dump the events to one of the log files. Then we can start adding back filters.
> Second issue: why firstname "column" just contain a static string "%{[parsedJson][surname]}}" (surname as well)?
That indicates that there was no `[parsedJson][surname]` field so the string was taken literally. This indicates that the json filter failed, perhaps because the grok filter failed. This is exactly why I advocate a gradual build-up of filters.
> Third issue: why @timestamp is always 3 hours ahead of my system datetime?
Perhaps because your local timezone is UTC-3? `@timestamp` is always UTC.
> Fourth issue: I understand that I have to create a mapping if I want to aggregate. Why the bellow command is trying to re-create the index?
A PUT operation always tries to create an index. To update the mappings of an existing index, see [Update mapping API | Elasticsearch Guide [8.11] | Elastic](https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-put-mapping.html).
---
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 22, 2017, 2:13pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/3 "2017-03-22T14:13:01Z")
> [@DemeCarvO](#):
>
> { "@timestamp" =\> 2017-03-22T13:43:01.443Z, "port" =\> 58794, "@version" =\> "1", "host" =\> "127.0.0.1", "message" =\> "{"firstname":"a","surname":"a"}", "type" =\> "document\_type"}
I edited my question adding the console output. Basically , it is:
{
"@timestamp" =\> 2017-03-22T13:43:01.443Z,
"port" =\> 58794,
"@version" =\> "1",
"host" =\> "127.0.0.1",
"message" =\> "{"firstname":"a","surname":"a"}",
"type" =\> "document\_type"
}
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [March 22, 2017, 2:38pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/4 "2017-03-22T14:38:44Z")
`message` only contains a JSON object so your grok filter won't work. Because of that no `request` field is extracted so your json filter fails, and because of that your mutate filter fails.
---
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 22, 2017, 2:51pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/5 "2017-03-22T14:51:24Z")
Magnusbaeck, thank you, But, how to achieve what I want? I mean, I want three columns: timestamp, firstname and surname. I guess that the most relevant part from my question could be translate as "how to split the json message in order to persist each field". Is my mistake either on nodejs side or elasticsearch side? I guess not. I guess it is some error in logstash.conf related to filter somehow.
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [March 22, 2017, 2:52pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/6 "2017-03-22T14:52:52Z")
Just drop the grok filter and parse the `message` field. You used the grok filter to strip a "data=" prefix but it didn't actually exist.
---
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 22, 2017, 2:59pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/7 "2017-03-22T14:59:41Z")
please, can you provide any example? I just added what I tried on my original question.
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [March 22, 2017, 3:03pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/8 "2017-03-22T15:03:37Z")
```nohighlight
json {
source => "message"
}
```
---
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 22, 2017, 3:17pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/9 "2017-03-22T15:17:47Z")
Thanks. What is wrong with:
curl -XPUT '[http://localhost:9200/greencard\_indice/\_mapping/cpfTipo](http://localhost:9200/greencard_indice/_mapping/cpfTipo)
> {
> "properties": {
> "firstname": {
> "type": "text",
> "fielddata": true
> }
> }
> }'
> curl: (3) [globbing] nested brace in column 75
Basically, I want to aggregate/sum/group by (I want to know how many users with same firstname)
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [March 22, 2017, 3:19pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/10 "2017-03-22T15:19:38Z")
The JSON stuff is supposed to be posted in the body of the request, i.e. as an argument to a `-d` option to curl. See examples in the docs.
---
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 22, 2017, 3:39pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/11 "2017-03-22T15:39:39Z")
Sorry, I didn't see such silly error with curl. Now it is working. You gave wonderful help. Please, a last and the most important part from my question, can you comment on such design: it is my first time using ELK. And after I read a lot some user cases, I want to use ELK for "bill" feature. I mean, every time an user reach any rest service, I will log from NodeJs to ElasticSearch throw LogStash basically his identification, time and the service involved.
Are these assumptions good "rule of thumb"?
1 - I decided not to log straight from NodeJs to ElasticSearch mainly beacuse, if I understood correctly, LogStash is not "thread-block". I mean, LogStash provides similar advantage as we get when we use asynchronous message queue to log. Additionally, Logstash will make my life easier to manage in/output, filter, re-process and so on the logs.
2 - I decided not to rely on Kibana for create the visualizations per user for create the invoice because it seems to me that Kibana will fit better analytic studies for business decision since it shows graphs, dashboard and tables easily.
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [March 22, 2017, 3:48pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/12 "2017-03-22T15:48:11Z")
> I decided not to log straight from NodeJs to Elasticsearch mainly beacuse, if I understood correctly, LogStash is not "thread-block". I mean, LogStash provides similar advantage as we get when we use asynchronous message queue to log.
Logstash has a very limited internal queue so you shouldn't rely on it to not block. I generally recommend having applications log to disk that Filebeat can tail and ship to Logstash. Then the logfile itself because the buffer.
> I decided not to rely on Kibana for create the visualizations per user for create the invoice because it seems to me that Kibana will fit better analytic studies for business decision since it shows graphs, dashboard and tables easily.
I don't know what kind of visualization you're after but yes, showing invoices isn't something Kibana excels at.
---
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 22, 2017, 4:02pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/13 "2017-03-22T16:02:51Z")
Does it make sense for invoice control based on service access (my focus is only related to performance, not business at all):
1 - create an index with three fields: user\_identification (text), access\_time (timestamp), accessed\_service (text)
2 - log in files via logstash all three fields accordingly to the access
3 - extract periodically from such files to ElasticSearch (using Filebeat for instance)
4 - (the main part of my question) create a mapping to the user\_identification so I can aggregate it and generate the user invoice
PS. I read:
"Fielddata can consume a lot of heap space, especially when loading high cardinality text fields. Once fielddata has been loaded into the heap, it remains there for the lifetime of the segment. Also, loading fielddata is an expensive process which can cause users to experience latency hits. This is why fielddata is disabled by default"
in [https://www.elastic.co/guide/en/elasticsearch/reference/current/fielddata.html](https://www.elastic.co/guide/en/elasticsearch/reference/current/fielddata.html)
which drive me to the idea that I am doing something wrong in my step 4.
---
### Author:  [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)
#### Post date: [March 22, 2017, 8:10pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/14 "2017-03-22T20:10:27Z")
Your approach is reasonable.
> which drive me to the idea that I am doing something wrong in my step 4.
No, you're not doing anything wrong.
---
### Author:  [@DemeCarvO](https://discuss.elastic.co/u/DemeCarvO)
#### Post date: [March 22, 2017, 8:25pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/15 "2017-03-22T20:25:30Z")
Thanks. Wonderful help! I still have a lot of new doubts but I need to study, try bymyself before start a new question on this forum. Now, at least, I am able to do the basic operations with Logstah+ElasticSearch and I am filling I have a good north.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [April 19, 2017, 8:25pm UTC](https://discuss.elastic.co/t/how-to-send-a-json-object-to-elasticsearch-throw-logstash/79516/16 "2017-04-19T20:25:51Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.