One Input Event - 2 Output Events

[cawoodm (Marc)]

We have tomcat access logs which we read from a FileBeat input and pass to an ElasticSearch output IndexA (fields a, b, c).

We would however like to have some of these log entries (the 400 and 500 errors) additionally in a different IndexB on the same ElasticSearch server. The documents should have a different format (fields d, e, f).

What are the options for doing this?

Some ideas:

1. Write the selected error docs to a file and feed this file to filebeat->logstash?
2. Write the selected error docs to logstash directly (via TCP/syslog)?
3. Somehow split the event in the logstash pipeline?
4. Use an ElasticSearch query to get the error docs from IndexA? How to load only new docs?

[magnusbaeck (Magnus Bäck)]

Somehow split the event in the logstash pipeline?

Yes, this. Use a clone filter.

[cawoodm (Marc)]

Can't get clone to do anything. I've added the following inside my filter {} block but I never get a 2nd event:

clone {
    add_field => {"foo2" => "bar2"}
}

Nor does this accomplish the feat:

clone {}

[magnusbaeck (Magnus Bäck)]

Its clones option must contain at least one type name.

$ cat test.config 
input { stdin { } }
output { stdout { codec => rubydebug } }
filter { clone { clones => ["sometype"] } }
$ echo hello | /opt/logstash/bin/logstash -f test.config
Settings: Default pipeline workers: 8
Pipeline main started
{
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-04-13T09:49:19.735Z",
          "host" => "lnxolofon"
}
{
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-04-13T09:49:19.735Z",
          "host" => "lnxolofon",
          "type" => "sometype"
}
Pipeline main has been shutdown
stopping pipeline {:id=>"main"}

[cawoodm (Marc)]

Thanks Magnus!

Unfortunately for me I was using the field "type" to actually determine if I need to clone or not so it gets pretty ugly here:

	if [type] == "ERROR" {
		mutate { add_field => {"type2" => "%{type}" } }
		clone {
			clones => ["cloned"]
			add_field => {
				"int1" => "%{status}"
			}
			remove_field => [ "ip", "method", "status", "url", "path", "userAgent" ]
		}
		if [type] == "cloned" {
			mutate {
				update => {
					"type" => "%{type2}"
					"[@metadata][index]" => "logs"
				} 
				remove_field => ["type2"]
			}
		}
	}