filter {
  if "mysql" in [tags] {
    grok {
      match => { "message" => "# User@Host: %{WORD:user}\[%{WORD}\] @ (%{HOSTNAME:client_hostname}|) \[(%{IP:client_ip}|)\]" }
    }

    grok {
      match => { "message" => "# Thread_id: %{NUMBER:thread_id:int} \s*Schema: (%{WORD:schema}| ) \s*QC_hit: %{WORD:qc_hit}" }
    }

    grok {
      match => { "message" => "# Query_time: %{NUMBER:query_time:float} \s*Lock_time: %{NUMBER:lock_time:float} \s*Rows_sent: %{NUMBER:rows_sent:int} \s*Rows_examined: %{NUMBER:rows_examined:int}" }
    }

    grok {
      match => { "message" => "(?m)SET timestamp=%{NUMBER:timestamp};%{GREEDYDATA:logmessage}" }
    }

    geoip {
      source => "clientip"
    }

    date {
      match => [ "timestamp", "UNIX" ]
    }

    mutate {
      remove_field => "timestamp"
    }

    mutate {
      gsub => [ "logmessage", "^\n", "" ]
      add_field => { "module" => "mysql" }
      add_field => { "loglevel" => "WARNING" }
    }
  }
}