Pipelines or "if else" conditions for output

phil.zac · March 16, 2018, 10:01pm

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 · March 16, 2018, 10:26pm

I think you need to write it as

if [type] =~ /beat/ {

Then it will work. (I hope )

magnusbaeck · March 17, 2018, 1:45pm

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

phil.zac · March 18, 2018, 6:35pm

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 · March 18, 2018, 6:58pm

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

phil.zac · March 18, 2018, 7:04pm

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 · March 18, 2018, 7:29pm

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 · March 18, 2018, 8:48pm

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 · March 19, 2018, 1:42am

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 · March 19, 2018, 7:13am

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 · March 26, 2018, 10:00pm

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 · March 27, 2018, 6:44am

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 · March 29, 2018, 2:41pm

Thanks alot.

system · April 26, 2018, 2:42pm

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.

Topic		Replies	Views
If-else at Multiple Pipeline Logstash	4	2664	April 12, 2019
If/Else condition for Output to Elasticsearch Beats docker , filebeat	4	2918	July 12, 2021
Clarification on if/else behavior in output Logstash	3	379	October 21, 2019
If conditional with multiple outputs Logstash	7	13150	May 22, 2019
Multiple outputs in the filebeats.yml to the logstash Beats filebeat	3	7691	December 18, 2019

Pipelines or "if else" conditions for output

Related topics