Logstash field split

[wolfghost (Abhishek)]

How to split

"content" => "command hitesh.restaurants command: drop { drop: "restaurants" } keyUpdates:0 writeConflicts:0 numYields:0 reslen:81 locks:{ Global: { acquireCount: { r: 1, w: 1 } }, Database: { acquireCount: { W: 1 } } } protocol:op_query 39ms\r"

into different fields????

[magnusbaeck (Magnus Bäck)]

You can definitely do it with a grok filter but a dissect filter might be faster or easier to maintain.

[wolfghost (Abhishek)]

filter {
grok {

      match => { message => 

"%{TIMESTAMP_ISO8601:@timestamp} %{MONGO3_SEVERITY:severity} %{MONGO3_COMPONENT:component}%{SPACE}(?:[%{DATA:context}])? %{GREEDYDATA:content}" }

        }   

if [component] != "COMMAND" {
drop { }
}
mutate{
remove_field => "message"
}
}

This is my logstash conf file

and its output is

{
"severity" => "I",
"path" => "C:/data/log/mongo.log",
"component" => "COMMAND",
"@timestamp" => 2017-03-15T11:42:35.230Z,
"@version" => "1",
"host" => "DESKTOP-PKMSR1Q",
"context" => "conn3",
"content" => "command hitesh.restaurants command: insert { insert: "restaurants", ordered: false, documents: 1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 numYields:0 reslen:40 locks:{ Global: { acquireCount: { r: 17, w: 17 } }, Database: { acquireCount: { w: 16, W: 1 } }, Collection: { acquireCount: { w: 16, W: 1 } } } protocol:op_query 194ms\r"
}

I wants to split content field

[magnusbaeck (Magnus Bäck)]

Multiple problems:

• The field you want to parse is content but you've configured the grok filter to parse the message field.
• Your grok filter doesn't even resemble what you want to parse. For starters, your messages doesn't begin with a timestamp but you've configured grok to begin by looking for an ISO8601 timestamp.

[wolfghost (Abhishek)]

my log ooks like

2017-03-14T17:08:12.615+0530 I COMMAND [conn2] command hitesh.zips command: insert { insert: "zips", ordered: false, documents: 1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 numYields:0 reslen:40 locks:{ Global: { acquireCount: { r: 17, w: 17 } }, Database: { acquireCount: { w: 16, W: 1 } }, Collection: { acquireCount: { w: 16, W: 1 } } } protocol:op_query 541ms

This is message field for logstash

And inside message field content field so how to split only content field???

[wolfghost (Abhishek)]

As you told me i tried dissect filter

dissect {
mapping => {
"content" => "%{query} %{+command} %{+ninserted} %{+keyUpdates} %{+writeConflicts} %{+numYields} %{+reslen} %{+locks} %{+database} %{+collection} %{+protocol} [%{pid}]: %{content}"
}
}

and it is giving output as

       "context" => "conn6",
        "reslen" => "\"zips\","

}
{
"severity" => "I",
"numYields" => "insert:",
"query" => "command",
"pid" => "",
"collection" => "documents:",
"locks" => "ordered:",
"content" => "1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 num
Yields:0 reslen:40 locks:{ Global: { acquireCount: { r: 16, w: 16 } }, Database:
{ acquireCount: { W: 16 }, acquireWaitCount: { W: 14 }, timeAcquiringMicros: {
W: 9529 } }, Collection: { acquireCount: { w: 16 } } } protocol:op_query 81ms\r"
,
"command" => "admin.zips",
"tags" => [],
"path" => "C:/Data/log/mongodb2.log",
"writeConflicts" => "{",
"component" => "COMMAND",
"database" => "false,",
"protocol" => "",
"@timestamp" => 2017-03-16T06:41:29.146Z,
"ninserted" => "command:",
"keyUpdates" => "insert",
"@version" => "1",
"host" => "Admin-PC",
"context" => "conn2",
"reslen" => ""zips","
}
{
"severity" => "I",
"numYields" => "insert:",
"query" => "command",
"pid" => "",
"collection" => "documents:",
"locks" => "ordered:",
"content" => "1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 num
Yields:0 reslen:40 locks:{ Global: { acquireCount: { r: 16, w: 16 } }, Database:
{ acquireCount: { W: 16 }, acquireWaitCount: { W: 2 }, timeAcquiringMicros: { W
: 1723 } }, Collection: { acquireCount: { w: 16 } } } protocol:op_query 12ms\r",

       "command" => "admin.zips",
          "tags" => [],
          "path" => "C:/Data/log/mongodb2.log",
"writeConflicts" => "{",
     "component" => "COMMAND",
      "database" => "false,",
      "protocol" => "",
    "@timestamp" => 2017-03-16T06:41:29.147Z,
     "ninserted" => "command:",
    "keyUpdates" => "insert",
      "@version" => "1",
          "host" => "Admin-PC",
       "context" => "conn2",
        "reslen" => "\"zips\","

}
{
"severity" => "I",
"numYields" => "insert:",
"query" => "command",
"pid" => "",
"collection" => "documents:",
"locks" => "ordered:",
"content" => "1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 num
Yields:0 reslen:40 locks:{ Global: { acquireCount: { r: 16, w: 16 } }, Database:
{ acquireCount: { W: 16 }, acquireWaitCount: { W: 8 }, timeAcquiringMicros: { W
: 4354 } }, Collection: { acquireCount: { w: 16 } } } protocol:op_query 17ms\r",

       "command" => "admin.zips",
          "tags" => [],
          "path" => "C:/Data/log/mongodb2.log",
"writeConflicts" => "{",
     "component" => "COMMAND",
      "database" => "false,",
      "protocol" => "",
    "@timestamp" => 2017-03-16T06:41:29.147Z,
     "ninserted" => "command:",
    "keyUpdates" => "insert",
      "@version" => "1",
          "host" => "Admin-PC",
       "context" => "conn6",
        "reslen" => "\"zips\","

}
{
"severity" => "I",
"numYields" => "insert:",
"query" => "command",
"pid" => "",
"collection" => "documents:",
"locks" => "ordered:",
"content" => "1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 num
Yields:0 reslen:40 locks:{ Global: { acquireCount: { r: 16, w: 16 } }, Database:
{ acquireCount: { W: 16 }, acquireWaitCount: { W: 11 }, timeAcquiringMicros: {
W: 6108 } }, Collection: { acquireCount: { w: 16 } } } protocol:op_query 18ms\r"
,
"command" => "admin.zips",
"tags" => [],
"path" => "C:/Data/log/mongodb2.log",
"writeConflicts" => "{",
"component" => "COMMAND",
"database" => "false,",
"protocol" => "",
"@timestamp" => 2017-03-16T06:41:29.147Z,
"ninserted" => "command:",
"keyUpdates" => "insert",
"@version" => "1",
"host" => "Admin-PC",
"context" => "conn3",
"reslen" => ""zips","
}
{
"severity" => "I",
"numYields" => "insert:",
"query" => "command",
"pid" => "",
"collection" => "documents:",
"locks" => "ordered:",
"content" => "1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 num
Yields:0 reslen:40 locks:{ Global: { acquireCount: { r: 16, w: 16 } }, Database:
{ acquireCount: { W: 16 }, acquireWaitCount: { W: 16 }, timeAcquiringMicros: {
W: 9847 } }, Collection: { acquireCount: { w: 16 } } } protocol:op_query 19ms\r"
,
"command" => "admin.zips",
"tags" => [],
"path" => "C:/Data/log/mongodb2.log",
"writeConflicts" => "{",
"component" => "COMMAND",
"database" => "false,",
"protocol" => "",
"@timestamp" => 2017-03-16T06:41:29.147Z,
"ninserted" => "command:",
"keyUpdates" => "insert",
"@version" => "1",
"host" => "Admin-PC",
"context" => "conn5",
"reslen" => ""zips","
}
{
"severity" => "I",
"numYields" => "insert:",
"query" => "command",
"pid" => "",
"collection" => "documents:",
"locks" => "ordered:",
"content" => "1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 num
Yields:0 reslen:40 locks:{ Global: { acquireCount: { r: 16, w: 16 } }, Database:
{ acquireCount: { W: 16 }, acquireWaitCount: { W: 12 }, timeAcquiringMicros: {
W: 8421 } }, Collection: { acquireCount: { w: 16 } } } protocol:op_query 17ms\r"
,
"command" => "admin.zips",
"tags" => [],
"path" => "C:/Data/log/mongodb2.log",
"writeConflicts" => "{",
"component" => "COMMAND",
"database" => "false,",
"protocol" => "",
"@timestamp" => 2017-03-16T06:41:29.148Z,
"ninserted" => "command:",
"keyUpdates" => "insert",
"@version" => "1",
"host" => "Admin-PC",
"context" => "conn2",
"reslen" => ""zips","
}

but the values are not exact

[magnusbaeck (Magnus Bäck)]

It seems you and @Nikparab are asking the exact same question. Let's deal with the problem in one thread, please. See Logstash configuration.

[wolfghost (Abhishek)]

Sure, no problem on that.

But, can i only split the last field from,

2017-03-14T17:08:12.615+0530 I COMMAND [conn2] command hitesh.zips command: insert { insert: "zips", ordered: false, documents: 1000 } ninserted:1000 keyUpdates:0 writeConflicts:0 numYields:0 reslen:40 locks:{ Global: { acquireCount: { r: 17, w: 17 } }, Database: { acquireCount: { w: 16, W: 1 } }, Collection: { acquireCount: { w: 16, W: 1 } } } protocol:op_query 541ms

I just want that ms and it's value as '541' and i am not having consistent logs.

[magnusbaeck (Magnus Bäck)]

I'm focusing on the other thread, but if you only care about the final millisecond value that's very easy to do with a grok filter.

[wolfghost (Abhishek)]

I am having one grok for splitting that one line of log.

How to split content with grok again ,i am confused here.

can you guide me,please?

[magnusbaeck (Magnus Bäck)]

You can have multiple grok fields in your configuration. Each filter can parse different fields, and the fields produced by one filter can be parsed by a second filter. If you don't care about the stuff you've extracted to the content field (except the millisecond duration at the end) you can just add .*%{INT:duration:int}$ to the end of your first grok filter.

[wolfghost (Abhishek)]

grok {

      match => { message => "%{TIMESTAMP_ISO8601:@timestamp} %{MONGO3_SEVERITY:severity} %{MONGO3_COMPONENT:component}%{SPACE}(?:\[%{DATA:context}\])? %{GREEDYDATA:content}.*%{INT:duration:int}ms" }
        }            

is this the correct way?

[magnusbaeck (Magnus Bäck)]

You forgot the final dollar sign but otherwise yes. If you don't care about the other stuff you're capturing with %{GREEDYDATA:content} you can just delete it.

[wolfghost (Abhishek)]

sorry that i forgot the $ sign.

and now i tried to remove %{GREEDYDATA:content} field but it was not giving output:-

C:\ELK\logstash-5.2.2>bin\logstash -f bin\logstash.conf
Could not find log4j2 configuration at path /ELK/logstash-5.2.2/config/log4j2.properties. Using default config which logs to console
13:15:30.548 [[main]-pipeline-manager] INFO logstash.outputs.elasticsearch - Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://localhost:9200/]}}
13:15:30.548 [[main]-pipeline-manager] INFO logstash.outputs.elasticsearch - Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://localhost:9200/, :path=>"/"}
13:15:30.688 [[main]-pipeline-manager] WARN logstash.outputs.elasticsearch - Restored connection to ES instance {:url=>#<URI::HTTP:0x355c3142 URL:http://localhost:9200/>}
13:15:30.688 [[main]-pipeline-manager] INFO logstash.outputs.elasticsearch - Using mapping template from {:path=>nil}
13:15:30.970 [[main]-pipeline-manager] INFO logstash.outputs.elasticsearch - Attempting to install template {:manage_template=>{"template"=>"logstash-", "version"=>50001, "settings"=>{"index.refresh_interval"=>"5s"}, "mappings"=>{"default"=>{"_all"=>{"enabled"=>true, "norms"=>false}, "dynamic_templates"=>[{"message_field"=>{"path_match"=>"message", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false}}}, {"string_fields"=>{"match"=>"", "match_mapping_type"=>"string", "mapping"=>{"type"=>"text", "norms"=>false, "fields"=>{"keyword"=>{"type"=>"keyword"}}}}}], "properties"=>{"@timestamp"=>{"type"=>"date", "include_in_all"=>false}, "@version"=>{"type"=>"keyword", "include_in_all"=>false}, "geoip"=>{"dynamic"=>true, "properties"=>{"ip"=>{"type"=>"ip"}, "location"=>{"type"=>"geo_point"}, "latitude"=>{"type"=>"half_float"}, "longitude"=>{"type"=>"half_float"}}}}}}}}
13:15:30.970 [[main]-pipeline-manager] INFO logstash.outputs.elasticsearch - New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>[#<URI::Generic:0x2063c524 URL://localhost:9200>]}
13:15:31.079 [[main]-pipeline-manager] INFO logstash.pipeline - Starting pipeline {"id"=>"main", "pipeline.workers"=>4, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>500}
13:15:31.360 [[main]-pipeline-manager] INFO logstash.pipeline - Pipeline main started
13:15:31.454 [Api Webserver] INFO logstash.agent - Successfully started Logstash API endpoint {:port=>9600}
after this it is not showing anything

[magnusbaeck (Magnus Bäck)]

That's unrelated to the GREEDYDATA removal. If you show your input configuration we can probably help.

[wolfghost (Abhishek)]

input {
file {
path => "C:/Data/mongobdg1.log"
start_position => "beginning" }
}
filter {
grok {

      match => { message => "%{TIMESTAMP_ISO8601:@timestamp} %{MONGO3_SEVERITY:severity} %{MONGO3_COMPONENT:component}%{SPACE}(?:\[%{DATA:context}\])? .*%{INT:duration:int}$" }
        }            

mutate{
remove_field => "message"
}
if [component] != "COMMAND" {
drop { }
}
}
output {
elasticsearch{ hosts => ["localhost:9200"] index => "deepak2" }
stdout {codec => "rubydebug" }
}

This is my logstash configuration file.

[magnusbaeck (Magnus Bäck)]

Logstash is tailing the input file. Set sincedb_path => "nul" in your file input to disable the sincedb functionality.

[wolfghost (Abhishek)]

Still it is not giving any output.

input {
file {
path => "C:/Data/mongobdg14.log"
start_position => "beginning"
sincedb_path => "nul"
}
}

[magnusbaeck (Magnus Bäck)]

Depending on what Logstash you're running you may need to adjust the ignore_older option (see the documentation). If that doesn't help bump up the log level to get more clues.

[wolfghost (Abhishek)]

This is my grok filter
grok {

      match => { message => "%{TIMESTAMP_ISO8601:timestamp} %{MONGO3_SEVERITY:severity} %{MONGO3_COMPONENT:component}%{SPACE}(?:\[%{DATA:context}\])? %{GREEDYDATA:content}.*%{INT:duration:int}ms"}
     }

for log

2017-03-07T00:56:57.473-0800 I COMMAND [conn44599] command hgthanka.Recognition command: aggregate { aggregate: "Recognition", pipeline: [ { $match: { Status: "Active", SuppressInFeed: false, Template.Type: { $in: [ "Recognition", "KeyResult", "Objective", "Milestone", "Award", "Context", "Thanks", "Quick", "Congrats", "Newsfeed", "ProductItem", "PollResult", "GoalKeyResultUpdate" ] }, Template.GroupId: "c6ce5dc0-1041-11e5-b713-cf230176158d", $and: [ { $or: [ { VisibilityMemberIds: null }, { VisibilityMemberIds: { $in: [ "5b389ed2-1055-11e5-b2e5-75f447e35752" ] } } ] }, { $or: [ { VisibilityLocations.hgId: null }, { VisibilityLocations.hgId: { $in: [ "5b09c660-1055-11e5-b2e5-75f447e35752" ] } } ] } ] } }, { $group: { _id: "$BatchId", max: { $max: "$ModifiedDate" } } }, { $sort: { max: -1 } }, { $skip: 40 }, { $limit: 10 }, { $project: { _id: 0, BatchId: "$_id" } } ] } keyUpdates:0 writeConflicts:0 numYields:198 reslen:648 locks:{ Global: { acquireCount: { r: 404 } }, Database: { acquireCount: { r: 202 } }, Collection: { acquireCount: { r: 202 } } } protocol:op_query 157ms

and its output is

{
"severity" => "I",
"duration" => 7,
"path" => "C:/data/log/mongo225.log",
"component" => "COMMAND",
"host" => "DESKTOP-PKMSR1Q",
"context" => "conn44599",
"content" => "command hgthanka.Recognition command: aggregate { aggregate: "Recognition", pipeline: [ { $match: { Status: "Active", SuppressInFeed: false, Template.Type: { $in: [ "Recognition", "KeyResult", "Objective", "Milestone", "Award", "Context", "Thanks", "Quick", "Congrats", "Newsfeed", "ProductItem", "PollResult", "GoalKeyResultUpdate" ] }, Template.GroupId: "c6ce5dc0-1041-11e5-b713-cf230176158d", $and: [ { $or: [ { VisibilityMemberIds: null }, { VisibilityMemberIds: { $in: [ "5b389ed2-1055-11e5-b2e5-75f447e35752" ] } } ] }, { $or: [ { VisibilityLocations.hgId: null }, { VisibilityLocations.hgId: { $in: [ "5b09c660-1055-11e5-b2e5-75f447e35752" ] } } ] } ] } }, { $group: { _id: "$BatchId", max: { $max: "$ModifiedDate" } } }, { $sort: { max: -1 } }, { $skip: 40 }, { $limit: 10 }, { $project: { _id: 0, BatchId: "$_id" } } ] } keyUpdates:0 writeConflicts:0 numYields:198 reslen:648 locks:{ Global: { acquireCount: { r: 404 } }, Database: { acquireCount: { r: 202 } }, Collection: { acquireCount: { r: 202 } } } protocol:op_query 15",
"timestamp" => "2017-03-07T00:56:57.473-0800"
}

in duration it is showing as 7
but i want 157ms?

I think it is only showing last integer number instead of showing whole duration field.