Pipelines or "if else" conditions for output

[phil.zac (Phil)]

Hi,

I am using

elasticsearch-5.6.1
kibana-5.6.1
logstash-5.6.1
metricbeat-5.6.1
packetbeat-5.6.1
x-pack-5.6.1

for Windows Server 2012 R2

---

My issue - I cant figure out how to make multiple output for logstash.conf

output {
          #if "%beat%" in [type]   # not working currently  need to check...
	if [type] =~ /^.*beat.*$/   # not working currently  need to check...
    {
		  elasticsearch {
		  hosts => "http://${HOSTNAME}:${PORT}"
		  user => "******"
		  password => "******"
		  manage_template => false
		  index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
		  document_type => "%{[@metadata][type]}"
       }
	} else {
		elasticsearch { 
		hosts => "http://${HOSTNAME}:${PORT}"
		user => "******"
		password => "*******"
		index => "%{type}-%{+YYYY.MM.dd}"
		document_id => "%{sessionnumber}-%{seqnum}"
		doc_as_upsert => "true"
    	}
    }
	
# display to the console any of the output stream, this can be commented for performance reason
	
  stdout { codec => rubydebug }
}

if I put only the beat part of the output, it works fine.

Any idea on how to make it work?

Thanks.

Phil

[pjanzen (Paul Janzen)]

I think you need to write it as

if [type] =~ /beat/ {

Then it will work. (I hope )

[magnusbaeck (Magnus Bäck)]

Please show an example event that you want to process. Use the output from stdout { codec => rubydebug }.

[phil.zac (Phil)]

I have tried to do that, but same issue. I am thinking the only options would be pipelines.yml. There is not much documents regarding pipelines.yml for 5.6.1 version. for other latest versions, there are documents related to that but might not be relevant for 5.6.1 version.

[pjanzen (Paul Janzen)]

And you are sure the type is beat? you can try if [tags] =~ /beats/ see if that will work?

[phil.zac (Phil)]

When you say event, do you mean the input type section.

Input part consists 1) using jdbc getting details from Database (can't put here for security reason)
and 2) for beat part.

The input beat part is followed in the tutorials

beats{
    	port => 5044
    }

The jdbc part works fine as long as the output part below as

output {
   		elasticsearch { 
		hosts => "http://${HOSTNAME}:${PORT}"
		user => "******"
		password => "********"
		index => "%{type}-%{+YYYY.MM.dd}"
		document_id => "%{sessionnumber}-%{seqnum}"
		doc_as_upsert => "true"
    }
	
# display to the console any of the output stream, this can be commented for performance reason
	
  stdout { codec => rubydebug }
}

Similarly Beats works fine when the output is (with no changes to the input section which includes jdbc input part)

output {
       		  elasticsearch {
		  hosts => "http://${HOSTNAME}:${PORT}"
		  user => "******"
		  password => "******"
		  manage_template => false
		  index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
		  document_type => "%{[@metadata][type]}"
       }
# display to the console any of the output stream, this can be commented for performance reason
	
  stdout { codec => rubydebug }
}

Hope this clears things.

[phil.zac (Phil)]

yes, I have tried to use tags. below is my code

input {
   jdbc {
      // hidden and correct
   }
  beats{
    	port => 5044
        tags => "beats"       # tried with and without this line.
    }
}

output {
   	if [tags] =~ /beat/
    {
      # similar to above code for output beats
   } else {
       # similar to above code for output for jdbc
  }

# display to the console any of the output stream, this can be commented for performance reason
	
  stdout { codec => rubydebug }
}

Not getting correctly for beats....

[magnusbaeck (Magnus Bäck)]

When you say event, do you mean the input type section.

No, I meant that I want to see the output from stdout { codec => rubydebug }.

[phil.zac (Phil)]

@phil.zac

stdout { codec => rubydebug }

ok When I run

.\logstash.bat -e 'output stdout with codec output {=> rubydebug'

Log files

[2018-03-18T21:27:01,147][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"***************/fb_apache/configuration"}
[2018-03-18T21:27:01,151][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>*******************/netflow/configuration"}
[2018-03-18T21:27:01,181][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"arcsight", :directory=>"****************jruby/1.9/gems/x-pack-5.6.1-java/modules/arcsight/configuration"}
[2018-03-18T21:27:01,787][ERROR][logstash.agent           ] Cannot create pipeline {:reason=>"Expected one of #, { at line 1, column 8 (byte 8) after output "}
[2018-03-18T21:27:02,134][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://elastic:xxxxxx@localhost:9200/]}}
[2018-03-18T21:27:02,138][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://elastic:xxxxxx@localhost:9200/, :path=>"/"}
[2018-03-18T21:27:02,359][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://elastic:xxxxxx@localhost:9200/"}
[2018-03-18T21:27:02,361][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["http://localhost:9200"]}
[2018-03-18T21:27:02,363][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>".monitoring-logstash", "pipeline.workers"=>1, "pipeline.batch.size"=>2, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>2}
[2018-03-18T21:27:02,367][INFO ][logstash.pipeline        ] Pipeline .monitoring-logstash started
[2018-03-18T21:27:02,497][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2018-03-18T21:27:05,382][WARN ][logstash.agent           ] stopping pipeline {:id=>".monitoring-logstash"}

On the otherhand.... I tried to comment out the if and else part and leave the stdout { codec => rubydebug } uncommented, I get a long list of input/output coming in and not fit to put in here......

output {
    #if "%beat%" in [type]
	#if [type] =~ /^.*beat.*$/   # not working currently  need to check...
	#if [tags] =~ /beat/
        # commenting......remaining...
  stdout { codec => rubydebug }
}

Let me know if I need to send some other short information?

Thanks.

Phil

[magnusbaeck (Magnus Bäck)]

ok When I run

.\logstash.bat -e 'output stdout with codec output {=> rubydebug'

???

I just want you to run Logstash as usual but show us the output from the stdout output you already have in your configuration.

[phil.zac (Phil)]

@magnusbaeck

stdout output

Sorry for the late reply.

When I run it, here is the log. (hidden the unnecessary ones)

[2018-03-26T16:38:27,125][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"*********/fb_apache/configuration"}
[2018-03-26T16:38:27,130][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"*********/netflow/configuration"}
[2018-03-26T16:38:27,144][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"arcsight", :directory=>"*********/bundle/jruby/1.9/gems/x-pack-5.6.1-java/modules/arcsight/configuration"}
[2018-03-26T16:38:28,067][INFO ][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=>{:removed=>[], :added=>[http://elastic:xxxxxx@localhost:9200/]}}
[2018-03-26T16:38:28,071][INFO ][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck_url=>http://elastic:xxxxxx@localhost:9200/, :path=>"/"}
[2018-03-26T16:38:28,339][WARN ][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=>"http://elastic:xxxxxx@localhost:9200/"}
[2018-03-26T16:38:28,341][INFO ][logstash.outputs.elasticsearch] New Elasticsearch output {:class=>"LogStash::Outputs::ElasticSearch", :hosts=>["http://localhost:9200"]}
[2018-03-26T16:38:28,342][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>".monitoring-logstash", "pipeline.workers"=>1, "pipeline.batch.size"=>2, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>2}
[2018-03-26T16:38:28,345][INFO ][logstash.pipeline        ] Pipeline .monitoring-logstash started
[2018-03-26T16:38:28,356][INFO ][logstash.pipeline        ] Starting pipeline {"id"=>"main", "pipeline.workers"=>8, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>1000}
[2018-03-26T16:38:28,933][INFO ][logstash.inputs.beats    ] Beats inputs: Starting input listener {:address=>"0.0.0.0:5044"}
[2018-03-26T16:38:29,002][INFO ][logstash.pipeline        ] Pipeline main started
[2018-03-26T16:38:29,019][INFO ][org.logstash.beats.Server] Starting server on port: 5044

truncated due to word limit

On the prompt, here is some of the prompt captured.....as it is rolling continuously.

},
 "version" => 4,
"response" => {
       "code" => 0,
    "message" => "EchoReply(0)",
       "type" => 0
}
},
    "tags" => [
[0] "beats",
[1] "beats_input_raw_event"
],
"path" => "***********"
  "@timestamp" => 2018-03-26T21:46:20.005Z,
   "bytes_out" => 8,
    "beat" => {
"name" => "************"
"hostname" => "************"
 "version" => "5.6.1"
},
"@version" => "1",
"host" => "**********"
"responsetime" => 0,
"path" => "***********"
  "status" => "OK"
}
{
"bytes_in" => 8,
"path" => "***********"
    "type" => "icmp",
    "icmp" => {
 "request" => {
       "code" => 0,
    "message" => "EchoRequest(0)",
       "type" => 8
},
 "version" => 4,
"response" => {
       "code" => 0,
    "message" => "EchoReply(0)",
       "type" => 0
}
},
    "tags" => [
[0] "beats",
[1] "beats_input_raw_event"
],
"path" => "***********"
  "@timestamp" => 2018-03-26T21:46:40.002Z,
   "bytes_out" => 8,
    "beat" => {
"name" => "************"
"hostname" => "************"
 "version" => "5.6.1"
},
"@version" => "1",
"host" => "**********"
"responsetime" => 0,
"path" => "***********"
  "status" => "OK"
}
{
   "duration" => {
"us" => 2998
},
 "@timestamp" => 2018-03-26T21:47:50.001Z,
"resolve_rtt" => {
"us" => 998
},
"path" => "***********"
"host" => "**********"
   "beat" => {
"name" => "************"
"hostname" => "************"
 "version" => "5.6.1"
},
   "@version" => "1",
"monitor" => "icmp-host-ip@*******",
     "up" => false,
  "error" => {
   "type" => "io",
"message" => "ping timeout"
},
   "type" => "icmp",
   "tags" => [
[0] "beats",
[1] "beats_input_raw_event"
]
}
{
"bytes_in" => 8,
"path" => "***********"
    "icmp" => {
 "request" => {
       "code" => 0,
    "message" => "EchoRequest(0)",
       "type" => 8
},
 "version" => 4,
"response" => {
       "code" => 0,
    "message" => "EchoReply(0)",
       "type" => 0
}
},
    "type" => "icmp",
    "tags" => [
[0] "beats",
[1] "beats_input_raw_event"
],
"path" => "***********"
  "@timestamp" => 2018-03-26T21:46:20.006Z,
   "bytes_out" => 8,
    "beat" => {
"name" => "************"
"hostname" => "************"
 "version" => "5.6.1"
},
"@version" => "1",
"host" => "**********"
"responsetime" => 0,
"path" => "***********"
  "status" => "OK"
}
{
"bytes_in" => 8,
"path" => "***********"
    "type" => "icmp",
    "icmp" => {
 "request" => {
       "code" => 0,
    "message" => "EchoRequest(0)",
       "type" => 8
},
 "version" => 4,
"response" => {
       "code" => 0,
    "message" => "EchoReply(0)",
       "type" => 0
}
},
    "tags" => [
[0] "beats",
[1] "beats_input_raw_event"
],
"path" => "***********"
  "@timestamp" => 2018-03-26T21:46:40.002Z,
   "bytes_out" => 8,
    "beat" => {
"name" => "************"
"hostname" => "************"
 "version" => "5.6.1"
},
"@version" => "1",
"host" => "**********"
"responsetime" => 0,
"path" => "***********"
  "status" => "OK"
}
{
"bytes_in" => 8,
"path" => "***********"
    "icmp" => {
 "request" => {
       "code" => 0,
    "message" => "EchoRequest(0)",
       "type" => 8
},
 "version" => 4,
"response" => {
       "code" => 0,
    "message" => "EchoReply(0)",
       "type" => 0
}
},
    "type" => "icmp",
    "tags" => [
[0] "beats",
[1] "beats_input_raw_event"
],
"path" => "***********"
  "@timestamp" => 2018-03-26T21:46:40.003Z,
   "bytes_out" => 8,
    "beat" => {
"name" => "************"
"hostname" => "************"
 "version" => "5.6.1"
},

truncated due to word limit

Let me know.

Once again thanks.

Phil

[magnusbaeck (Magnus Bäck)]

Okay, so the type field contains e.g. "icmp", not anything related to beats. If you want to route all events from Packetbeat (or whatever is producing the data) you can use this:

if "beats" in [tags] {
  ...
}

[phil.zac (Phil)]

Thanks alot.