# Sending only Json format of log to Kafka server from logs containing both json and non json format in logstash config
**URL:** https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168
**Category:** Logstash
**Created:** [August 6, 2018, 12:26pm UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168 "2018-08-06T12:26:26Z")
**Posts on this page:** 7
**Page:** 1
### Author:  [@Anshika](https://discuss.elastic.co/u/Anshika)
#### Post date: [August 6, 2018, 12:26pm UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168/1 "2018-08-06T12:26:27Z")
I am trying to send the Json format of the log from the logs containing both json and non json format to Kafka server.
example of logs: 880 \<14\>1 2018-08-06T10:49:05.89677+00:00 dev.hello-world 24f9ade2-1bdb-46c2-bc5c-4b25a277832e [APP/PROC/WEB/0] - - 2018-08-06 10:49:05.894 INFO 24 --- [nio-8080-exec-2] classnmae : {"Status":"from employee first page method","TransactionAfter":{"empId":"1","name":"emp1","designation":"manager","salary":3000.0},"Category":null,"Messages":{"Value":"EIPCLELOGS","Name":"Identifier"},"Header":{"TransactionType":"INFO","ServiceName":"class controllers.TestController","BusinessID2":"1","Hostname":"0e2edd3e-c649-472c-539d-6dcb/10.255.223.92","ComponentName":"firstPage","ApplicationID":"abc","Timestamp":"2018-08-06T10:49:05.883+0000","TransactionDomain":"Employee","BusinessID":"1","TransactionID":"1","ApplicationDomain":"Employee"},"TimeDuration":null,"TransactionBefore":"emp1","DataEncoding":null,"LogLevel":"INFO"}
tried solution:
filter {
if "EIPCLELOGS" in [message] {
grok {
match =\> {
"message" =\> [
"(?[0-9-]+) \<(?[0-9]+)\>(?[0-9]+) %{TIMESTAMP\_ISO8601:UTCtimestamp} %{JAVACLASS:class}-(?[a-z]+) (?[a-z0-9-]+) \*[%{DATA:thread}] - - %{DATA:timestamp1} \*%{LOGLEVEL:level} %{DATA:pid} --- \*[%{DATA:thread2}] %{JAVACLASS:class2} \*: %{GREEDYDATA:log}"
]
}
}
```
json {
source => "log"
target => "parsedJson"
remove_field=>[log"]
}
mutate {
add_field => {
"message" => ["%{[parsedJson][message]}"]
}
}
```
}
}
expected OutPut in elastic search :
{"Status":"from employee first page method","TransactionAfter":{"empId":"1","name":"emp1","designation":"manager","salary":3000.0},"Category":null,"Messages":{"Value":"EIPCLELOGS","Name":"Identifier"},"Header":{"TransactionType":"INFO","ServiceName":"class .controllers.TestController","BusinessID2":"1","Hostname":"1554f7af-5d9c-4f19-4c48-0ca3/10.255.223.51","ComponentName":"firstPage","ApplicationID":"eip","Timestamp":"2018-08-06T10:15:58.483+0000","TransactionDomain":"Employee","BusinessID":"1","TransactionID":"1","ApplicationDomain":"Employee"},"TimeDuration":null,"TransactionBefore":"emp1","DataEncoding":null,"LogLevel":"INFO"}
also, tried prune
prune {
whitelist\_names =\> ["using the wanted json fromat"]
}
The If with grok works fine in the filter.I get grok filters correctly,But the later doesn't work. I have tried removing the fields in mutate directly as well which still does not work.
All the above solution does not seem to be working.Any help would be appreciated.Thanks
---
### Author:  [@spinscale](https://discuss.elastic.co/u/spinscale)
#### Post date: [August 6, 2018, 3:54pm UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168/2 "2018-08-06T15:54:09Z")
I moved this over to the logstash forum. You might want to take some more time and properly format this post, it is pretty hard to read.
---
### Author:  [@Badger](https://discuss.elastic.co/u/Badger)
#### Post date: [August 6, 2018, 5:16pm UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168/3 "2018-08-06T17:16:39Z")
> [@Anshika](#):
>
> ```
> mutate { add_field => { "message" => ["%{[parsedJson][message]}"] } }
>
> ```
There is no field nested in parsedJson called message, so this will not work.
---
### Author:  [@Anshika](https://discuss.elastic.co/u/Anshika)
#### Post date: [August 7, 2018, 6:03am UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168/4 "2018-08-07T06:03:49Z")
@spinscale Sorry about the format.I will make changes.
---
### Author:  [@Anshika](https://discuss.elastic.co/u/Anshika)
#### Post date: [August 7, 2018, 6:05am UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168/5 "2018-08-07T06:05:42Z")
@Badger i am not looking for nested field. I want the whole JSON as a response.Are these steps correct for getting only the JSOn from the log input??
---
### Author:  [@Anshika](https://discuss.elastic.co/u/Anshika)
#### Post date: [August 7, 2018, 1:17pm UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168/6 "2018-08-07T13:17:31Z")
Hi,
I have tried more on this and I am almost near to what I am looking for.
**My logstash config looks like** :
input {
tcp {
port =\> 5000
type =\> syslog
}
udp {
port =\> 5000
type =\> syslog
}
}
filter {
if "EIPCLELOGS" in [message] {
grok {
match =\> {
"message" =\> [
"(?[0-9-]+) \<(?[0-9]+)\>(?[0-9]+) %{TIMESTAMP\_ISO8601:UTCtimestamp} %{JAVACLASS:class}-(?[a-z]+) (?[a-z0-9-]+) \*[%{DATA:thread}] - - %{DATA:timestamp1} \*%{LOGLEVEL:level} %{DATA:pid} --- \*[%{DATA:thread2}] %{JAVACLASS:class2} \*: %{GREEDYDATA:cleLog}"
]
}
}
```
mutate {
remove_field => ["timestamp1","pid","port","thread","thread2","level","class2","class","UTCtimestamp","Fields1","Fields2","Fields3","Fields5","Fields6","host","type","message","@version","@timestamp"]
remove_tag => ["timestamp1","pid","port","thread","thread2","level","class2","class","UTCtimestamp","Fields1","Fields2","Fields3","Fields5","Fields6","host","type","message","@version","@timestamp"]
```
}
}
}
output {
if "eip" in [Header][ApplicationID]{
kafka {
codec =\> json{}
bootstrap\_servers =\> "kafka servers"
topic\_id =\> "cle-logs-eip"
}
}
}
**My Output in kafka is** :
{"cleLog":"{"Status":"from employee first page method","TransactionAfter":{"empId":"1","name":"emp1","designation":"manager","salary":3000.0},"Category":null,"Messages":{"Value":"EIPCLELOGS","Name":"Identifier"},"Header":{"TransactionType":"INFO","ServiceName":"class com.pepsico.eip.controllers.TestController","BusinessID2":"1","Hostname":"b484b154-2d07-473e-4cd0-f641/10.255.223.4","ComponentName":"firstPage","ApplicationID":"eip","Timestamp":"2018-08-07T12:27:01.730+0000","TransactionDomain":"Employee","BusinessID":"1","TransactionID":"1","ApplicationDomain":"Employee"},"TimeDuration":null,"TransactionBefore":"emp1","DataEncoding":null,"LogLevel":"INFO"}"}
**expected output** :
{"Status":"from employee first page method","TransactionAfter":{"empId":"1","name":"emp1","designation":"manager","salary":3000.0},"Category":null,"Messages":{"Value":"EIPCLELOGS","Name":"Identifier"},"Header":{"TransactionType":"INFO","ServiceName":"class .eip.controllers.TestController","BusinessID2":"1","Hostname":"b484b154-2d07-473e-4cd0-f641/10.255.223.4","ComponentName":"firstPage","ApplicationID":"eip","Timestamp":"2018-08-07T12:27:01.730+0000","TransactionDomain":"Employee","BusinessID":"1","TransactionID":"1","ApplicationDomain":"Employee"},"TimeDuration":null,"TransactionBefore":"emp1","DataEncoding":null,"LogLevel":"INFO"}
Basically, I need 2 things here,
1. need to remove the cleLog wrapper from the output and
2. need to send the logs to kafka only when the Header.ApplicationID is "eip"
It would be great help if someone can help me on this.
---
### Author:  [@system](https://discuss.elastic.co/u/system)
#### Post date: [September 4, 2018, 1:28pm UTC](https://discuss.elastic.co/t/sending-only-json-format-of-log-to-kafka-server-from-logs-containing-both-json-and-non-json-format-in-logstash-config/143168/7 "2018-09-04T13:28:51Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.