input {
  kafka {
    codec => "json"
    bootstrap_servers => "kafka1:9092,kafka2:9192,kafka3:9292"
    topics => ["db-slow-log"]
    id => "db-slow-log"
    consumer_threads => 3
    decorate_events => true
    auto_offset_reset => "latest"
  }
}

filter {
  grok {
    id => "db-slow-log-grok-sleep-drop"
    match => { "message" => "SELECT SLEEP" }
    add_tag => [ "sleep_drop" ]
    tag_on_failure => []
  }

  if "sleep_drop" in [tags] {
    drop {
      id => "sleep_drop"
    }
  }

  grok {
    id => "db-slow-log-grok-slowlog"
    match => [ "message", "# Time: \d+ %{TIME:time}\s+# User@Host: (?<user>[a-zA-Z0-9._-]*)\[(?:.*)\]\s+@\s+(?<client-domain>\S*)\s+\[%{IP:client-ip}*\]\s.*# Query_time: %{NUMBER:query_time}\s+Lock_time: %{NUMBER:lock_time}\s+Rows_sent: %{NUMBER:rows_sent}\s+Rows_examined: %{NUMBER:rows_examined}\s+SET\s+timestamp=%{NUMBER:timestamp_mysql};\s+(?<query>[\s\S]*)" ]
  }

  date {
    id => "db-slow-log-date"
    match => [ "timestamp", "UNIX" ]
    remove_field => [ "timestamp" ]
  }
}

output {
  elasticsearch {
    id => "db-slow-log-output-es"
    hosts => ["elasticsearch-master:9200","elasticsearch-data-1:9200","elasticsearch-data-2:9200","elasticsearch-data-3:9200"]
    user => "elastic"
    password => "changeme"
    index => "logstash-db-slow-%{+YYYY.MM.dd}"
    codec => "json"
  }
}