I have implemented the kafka with logstash input and elasticsearch output. its working fine in kibana.. I want to filter the data based on statuscode

Question

This is kibana dashboard json Data.. Here i have to filter the based on response statuscode with in the message json data field..

{
  "_index": "rand-topic",
  "_type": "_doc",
  "_id": "ulF8uH0BK9MbBSR7DPEw",
  "_version": 1,
  "_score": null,
  "fields": {
    "@timestamp": [
    "2021-12-14T10:27:56.956Z"
],
"@version": [
  "1"
],
"@version.keyword": [
  "1"
],
"message": [
  "{\"requestMethod\":\"GET\",\"headers\":{\"content-type\":\"application/json\",\"user-agent\":\"PostmanRuntime/7.28.4\",\"accept\":\"*/*\",\"postman-token\":\"977fc94b-38c8-4df4-ad73-814871a32eca\",\"host\":\"localhost:5600\",\"accept-encoding\":\"gzip, deflate, br\",\"connection\":\"keep-alive\",\"content-length\":\"44\"},\"body\":{\"category\":\"CAT\",\"noise\":\"purr\"},\"query\":{},\"requestUrl\":\"http://localhost:5600/kafka\",\"protocol\":\"HTTP/1.1\",\"remoteIp\":\"1\",\"requestSize\":302,\"userAgent\":\"PostmanRuntime/7.28.4\",\"statusCode\":200,\"response\":{\"success\":true,\"message\":\"Kafka Details are added\",\"data\":{\"kafkaData\":{\"_id\":\"61b871ac69be37078a9c1a79\",\"category\":\"DOG\",\"noise\":\"bark\",\"__v\":0},\"postData\":{\"category\":\"DOG\",\"noise\":\"bark\"}}},\"latency\":{\"seconds\":0,\"nanos\":61000000},\"responseSize\":193}"]},"sort[1639477676956]}

Expected output like this Here added the statuscode field from message field

{
  "_index": "rand-topic",
  "_type": "_doc",
  "_id": "ulF8uH0BK9MbBSR7DPEw",
  "_version": 1,
  "_score": null,
 "fields": {
   "@timestamp": [
    "2021-12-14T10:27:56.956Z"
   ],
 "@version": [
    "1"
 ],
 "@version.keyword": [
    "1"
 ],
"statusCode": [
  200
],
"message": [
   "{\"requestMethod\":\"GET\",\"headers\":{\"content- 
 type\":\"application/json\",\"user- 
 agent\":\"PostmanRuntime/7.28.4\",\"accept\":\"*/*\",\"postman- 
 token\":\"977fc94b-38c8-4df4-ad73- 
 814871a32eca\",\"host\":\"localhost:5600\",\"accept- 
 encoding\":\"gzip, deflate, br\",\"connection\":\"keep- 
 alive\",\"content-length\":\"44\"},\"body\": 
 {\"category\":\"CAT\",\"noise\":\"purr\"},\"query\": {}, \"requestUrl\":\"http://localhost:5600/kafka\",\"protocol\":\"HTTP/1.1\",\"remoteIp\":\"1\",\"requestSize\":302,\"userAgent\":\"PostmanRuntime/7.28.4\",\"statusCode\":200,\"response\":{\"success\":true,\"message\":\"Kafka Details are added\",\"data\":{\"kafkaData\":{\"_id\":\"61b871ac69be37078a9c1a79\",\"category\":\"DOG\",\"noise\":\"bark\",\"__v\":0},\"postData\":{\"category\":\"DOG\",\"noise\":\"bark\"}}},\"latency\":{\"seconds\":0,\"nanos\":61000000},\"responseSize\":193}"

]},"sort": [1639477676956]}

Please help me how to configure logstash filter for statusCode

input {
  kafka {
    topics => ["randtopic"]
    bootstrap_servers => "192.168.29.138:9092"
  }
}

filter{
  mutate {
    add_field => { 
        "statusCode" => "%{[status]}" 
    }
  }
}

output {
 elasticsearch {
    hosts => ["192.168.29.138:9200"]
    index => "rand-topic"
    workers => 1
 }
}

masoud rafiee · Accepted Answer · 2021-12-28 06:20:15Z

1

 output {
  if [message][0][statusCode] == "200" {
     Do Somethings ....
    stdout { codec => ""}
  }
}

edited Dec 28, 2021 at 6:20

answered Dec 15, 2021 at 11:37

masoud rafiee

4471 gold badge3 silver badges9 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

Emmanuel R Over a year ago

I need to filter based on statuscode from this response with in message field json (\"statusCode\":200,)and set it in filter configuration in log stash to display it as a separate field and let the message field as it is like a string not as a json in kibana Now i have used it. but wont work mutate { add_field => { "statusCode" => "%{[status]}" } }

masoud rafiee Over a year ago

try this :mutate { add_field => { "StatusCode" => "%[message][statusCode]" } }

Emmanuel R Over a year ago

i tried it like 2 way.. but its not worked. mutate { add_field => { "statusCode" => "%{[message][0][statusCode]}" } }. and mutate { add_field => { "statusCode" => "%{[message][statusCode]}" } } Output statusCode : %{[message][0][status]} Here status Code value is not set that variable. Its like returned given string

Collectives™ on Stack Overflow

I have implemented the kafka with logstash input and elasticsearch output. its working fine in kibana.. I want to filter the data based on statuscode

1 Answer 1

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related