input {
  file {
    id => "bytedesk-app-log"
    path => ["/var/log/bytedesk/bytedeskim.log*", "/var/log/bytedesk-source/bytedeskim.log*"]
    start_position => "beginning"
    mode => "tail"
    sincedb_path => "/usr/share/logstash/data/bytedeskim.sincedb"
    codec => multiline {
      pattern => "^-*\[RID:"
      negate => true
      what => "previous"
      auto_flush_interval => 2
    }
  }
}

filter {
  mutate {
    gsub => [
      "message", "\u001B\[[0-9;]*[A-Za-z]", ""
    ]
  }

  grok {
    match => {
      "message" => [
        "^-\\[RID:%{DATA:requestId} TRACEID:%{DATA:traceId}\\]-%{TIMESTAMP_ISO8601:log_timestamp}-%{SPACE}%{LOGLEVEL:level} %{INT:pid} --- \\[%{DATA:thread}\\] %{DATA:logger} : %{GREEDYDATA:parsed_message}$",
        "^\\[RID:%{DATA:requestId} TRACEID:%{DATA:traceId}\\] %{TIMESTAMP_ISO8601:log_timestamp} %{SPACE}%{LOGLEVEL:level}(?: %{INT:pid})? --- \\[%{DATA:thread}\\] %{DATA:logger} : %{GREEDYDATA:parsed_message}$"
      ]
    }
    tag_on_failure => ["_bytedesk_parse_failure"]
  }

  if [log_timestamp] {
    date {
      match => ["log_timestamp", "yyyy-MM-dd HH:mm:ss.SSS"]
      target => "@timestamp"
    }
  }

  mutate {
    add_field => {
      "service.name" => "bytedesk"
      "event.dataset" => "bytedesk.application"
    }
  }

  if [parsed_message] {
    mutate {
      replace => {
        "message" => "%{parsed_message}"
      }
    }
  }

  if [level] {
    mutate {
      add_field => {
        "[log][level]" => "%{level}"
      }
    }
  }

  if [logger] {
    mutate {
      add_field => {
        "[log][logger]" => "%{logger}"
      }
    }
  }

  if [thread] {
    mutate {
      add_field => {
        "[process][thread][name]" => "%{thread}"
      }
    }
  }

  if [pid] {
    mutate {
      convert => {
        "pid" => "integer"
      }
      add_field => {
        "[process][pid]" => "%{pid}"
      }
    }
  }

  mutate {
    strip => ["level", "logger", "thread", "message", "requestId", "traceId"]
    remove_field => ["parsed_message", "log_timestamp", "host", "path", "level", "logger", "thread", "pid"]
  }
}

output {
  elasticsearch {
    hosts => ["${ELASTICSEARCH_HOSTS}"]
    user => "${ELASTICSEARCH_USERNAME}"
    password => "${ELASTICSEARCH_PASSWORD}"
    index => "bytedesk-logs-%{+YYYY.MM.dd}"
    ilm_enabled => false
  }
}