Send all data to one index and specific fields to another at same time

Elastic Stack Logstash
1

Hello,

I've been reading for a long time in different sources if it's possible to send all data processed by logstash to one index and at the same time send just some specific fields of this data to another different index.

I tried to use the clone filter plugin to achieve it but it doesn't work for me.

e.g, I have this logstash configuration:

input { stdin { } }

filter {
  clone {
    clones => ["cloned"]
  }
  grok {
    match => { "message" => "%{NUMBER:timestamp}\s+%{NUMBER:duration}\s%{IP:client_address}\s%{WORD:cache_result}/%{POSINT:status_code}\s%{NUMBER:bytes}\s%{WORD:request_method}\s%{NOTSPACE:url}\s(%{NOTSPACE:user}|-)\s%{WORD:hierarchy_code}/%{IPORHOST:server}\s%{NOTSPACE:content_type}"}
  }

  if [type] == "cloned" {
    prune {
      whitelist_names => ["url"]
    }
  }
}

output {
  if [type] == "cloned" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "cloned_index"
    }
  }
  elasticsearch {  
    hosts => ["localhost:9200"]
    index => "http_examples"     
  }
  stdout { codec => rubydebug }
}

Sending the following data:

1598405553.742    289 127.0.0.1 TCP_MISS/301 440 GET http://this-is-a-test/ - HIER_DIRECT/13.107.42.14 -

I have the following output:

{
    "url" => "http://this-is-a-test/"
}
{
      "content_type" => "-",
       "status_code" => "301",
             "bytes" => "440",
           "message" => "1598405553.742    289 127.0.0.1 TCP_MISS/301 440 GET http://this-is-a-test/ - HIER_DIRECT/13.107.42.12 -",
      "cache_result" => "TCP_MISS",
    "client_address" => "127.0.0.1",
          "duration" => "289",
        "@timestamp" => 2022-04-14T01:16:18.624Z,
    "hierarchy_code" => "HIER_DIRECT",
         "timestamp" => "1598405553.742",
            "server" => "13.107.42.12",
    "request_method" => "GET",
               "url" => "http://this-is-a-test/",
          "@version" => "1",
              "host" => "elastic",
              "user" => "-"
}

I see it prints the two different outputs but I don't find the way to create the index in the index pattern area of the Stack management in Kibana.

Any idea of what can be happening?

Thanks a lot for your help.

2

This condition will never be true:

  if [type] == "cloned" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "cloned_index"
    }
  }

In the prune filter that applies to events that have type = cloned, the whitelist_names only includes the url filter, so you won't have any event with type = cloned after this prune filter.

Try to change your whitelist_names to ["url", "type"] and see if this do what you want.

You would also need to filter out the events with type = cloned so those events are not indexed in the http_examples, your output block should look something like this:

output {
    if [type] == "cloned" {
        elasticsearch {
            hosts => ["localhost:9200"]
            index => "cloned_index"
        }
    } else {
        elasticsearch {
            hosts => ["localhost:9200"]
            index => "http_examples"
        }
    }
}
1 Like
3

Your explanation makes totally sense.

I've changed it and everything works perfectly.

Thank you very much @leandrojmp

4

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