logstash.conf

# Author: Luca Memini
# Email: luca@memini.it
# Last Update:
#
# This conf file is based on accepting logs for dnstap json events

input {
  udp {
    port => 6052
    codec => json
    type => "dnstap"
  }
}

filter {
  if [type] == "dnstap" {
    date {
      match => [ "timestamp", "UNIX" ]
    }

    ##### Section "answars"
    grok {
        match => [ "answers", '%{DATA:fqdn}.%{SPACE}%{NUMBER:ttl}%{SPACE}(?<rrclass>.{2})\s%{DATA:querytype}\s(%{IP:answer_ip}|"%{GREEDYDATA:answer}"|%{GREEDYDATA:answer}\.)' ]
        tag_on_failure => [ "_no_answers" ]
    }

    ##### Section "question"
    grok {
        match => [ "question", '"%{DATA:query}\.%{SPACE}(?<question_rrclass>.{2})\s%{GREEDYDATA:question_querytype}"' ]
        tag_on_failure => [ "_no_question" ]
    }

    # I recommend renaming the fields below to be consistent with other log sources.  This makes it easy to "pivot" between logs
    mutate {
      rename => {
        "query_address" => "source_ip"
        "response_address" => "destination_ip"
        "query_port" => "source_port"
        "response_port" => "destination_port"
        "identity" => "syslog_hostname"
        "version" => "syslog_program"
      }
    }
  }

 if [source_ip] and [question] and [answers] and [rcode_string] {
   mutate {
     replace => { "message" => "%{rcode_string}: %{source_ip} (%{question} -> %{answers})" }
   }
 } else {
   mutate {
     replace => { "message" => "%{rcode_string}: %{source_ip} (%{question})" }
   }
 }

  if [type] == "dnstap" {
    if "_grokparsefailure" not in [tags] and "_jsonparsefailure" not in [tags] {
      mutate {
        remove_field => [ "authorities" ]
      }
    }
  }
}

output {
  if [type] == "dnstap" {

    elasticsearch {
      ilm_enabled => false
      index => "logstash-%{+YYYY.MM.dd}"
    }

# debug only ;-)
#   file {
#        path => "/var/log/logstash/dnstap.log"
#        codec => "json"
#   }
  }
}