logstash.conf

input {
    udp {
        port => 1984
    }
}

filter {
    json {
        source => "message"
    }

    if "_jsonparsefailure" in [tags] {
        mutate {
            add_field => {
                client => "error"
                metric => "metric_error"
            }
        }
    }
}

output {
    elasticsearch {
        hosts => ["http://10.240.0.2:9200"]
        index => ".measure-%{client}-%{+YYYY.MM.dd}"
        document_type => "%{metric}"
    }
}