input {
  file {
    path => "/mnt/wasabi-logs/*"  # Use absolute paths; wildcards are supported
    mode => "read"
    start_position => "beginning"
    sincedb_path => "/var/lib/logstash/sincedb-wasabi"
    discover_interval => 5
    stat_interval => 1
  }
}

filter {
  # Drop non-record lines
  if [message] =~ /^Record format:/ { drop {} }
  if [message] =~ /^\s*=+\s*$/ { drop {} }

  grok {
    match => {
      "message" => [
        '^%{NOTSPACE:bucket_owner} %{NOTSPACE:bucket} \[%{HTTPDATE:timestamp}\] %{IPORHOST:client_ip} %{NOTSPACE:requester} %{NOTSPACE:request_id} %{NOTSPACE:operation} %{NOTSPACE:key} %{QS:request_uri} %{NUMBER:http_status:int} %{NOTSPACE:error_code} %{NOTSPACE:bytes_sent} %{NOTSPACE:object_size} %{NOTSPACE:total_time} %{NOTSPACE:turn_around_time} %{QS:referrer} %{QS:user_agent}(?: (?:(?:%{QS:tls_version})|(?:%{NOTSPACE:tls_version})))? %{NOTSPACE:version_id}\s*$'
      ]
    }
    tag_on_failure => ["_grok_accesslog_failed"]
  }

  if "_grok_accesslog_failed" not in [tags] {
    
    urldecode { field => "key" } 
    urldecode { field => "request_uri" }

    # Normalize "-" to empty, then convert where possible
    mutate {
      gsub => [
        "bytes_sent", "^-?$", "",
        "object_size", "^-?$", "",
        "total_time", "^-?$", "",
        "turn_around_time", "^-?$", ""
      ]
      convert => {
        "bytes_sent" => "integer"
        "object_size" => "integer"
        "total_time" => "integer"
        "turn_around_time" => "integer"
      }
    }

    # If TLS wasn't present in the record, set a stable placeholder
    if ![tls_version] or [tls_version] == "" {
      mutate { add_field => { "tls_version" => "-" } }
    }

    date {
      match  => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
      target => "@timestamp"
    }

    mutate {
      replace => {
      "syslog_msg" => "bucket_owner=%{bucket_owner} bucket=%{bucket} ts=%{timestamp} ip=%{client_ip} requester=%{requester} request_id=%{request_id} op=%{operation} key=%{key} uri=%{request_uri} status=%{http_status} error=%{error_code} bytes_sent=%{bytes_sent} object_size=%{object_size} total_time=%{total_time} turn_around_time=%{turn_around_time} referrer=%{referrer} ua=%{user_agent} tls=%{tls_version} version_id=%{version_id}"
      }
    }

  } else {
    mutate { replace => { "syslog_msg" => "unparsed=%{message}" } }
  }
}

output {
  syslog {
    host => "10.1.2.3" # Replace with your destination SEM server's IP or hostname
    port => 514 # The standard syslog port
    facility => "local3" # This may not match the actual facility number sent to SEM
    protocol => "udp"
    message => "%{syslog_msg}"
  }
}