vector.toml

data_dir = "/var/lib/vector"

# ------------------ SYSLOG message pipe ------------------ #

[sources.syslog_pipe_in]
    type = "socket"
    mode = "udp"
    address = "127.0.0.1:3333"

[transforms.syslog_pipe_parsing]
    type = "grok_parser"
    inputs = ["syslog_pipe_in"]
    pattern = "%{SYSLOG5424LINE:syslog_message}"
    drop_field = true

[transforms.syslog_pipe_cleanup]
    type = "remove_fields"
    inputs = ["syslog_pipe_parsing"]
    fields = [
        "syslog_message", "host", "syslog5424_sd",
        "timestamp", "syslog5424_pri",
        "syslog5424_msgid", "syslog5424_ts", "syslog5424_ver"
    ]

[transforms.syslog_pipe_remap]
    type = "lua"
    inputs = ["syslog_pipe_cleanup"]

    source = """
event["message"] = event["syslog5424_msg"]
event["host"] = event["syslog5424_host"]
event["procid"] = event["syslog5424_proc"]
event["syslog5424_host"] = nil
event["syslog5424_msg"] = nil
event["syslog5424_proc"] = nil
"""

[transforms.syslog_pipe_type_correction]
    type = "coercer"
    inputs = ["syslog_pipe_remap"]

    [transforms.syslog_pipe_type_correction.types]
        "procid" = "int"

[transforms.syslog_pipe_out]
    type = "regex_parser"
    inputs = ["syslog_pipe_type_correction"]
    regex = "^(?P<app_name>[\\w\\.\\d-]+)\\|(?P<app_version>[\\w\\.\\d-]+)\\|(?P<target_format>[\\w\\.\\d-]+)\\|(?P<target_parser>[\\w\\.\\d-]+)$"
    field = "syslog5424_app"
    drop_field = true

# ------------------ SYSLOG message pipe ------------------ #

# ------------------ JSON Pipe ---------------------------- #

[transforms.json_pipe_in]
    type = 'field_filter'
    inputs = ["syslog_pipe_out"]
    field = "target_format"
    value = "json"

[transforms.json_parser]
    type = "json_parser"
    inputs = ["json_pipe_in"]
    drop_invalid = false

# ------------------ JSON NGINX Pipe ---------------------- #

[transforms.json_nginx_pipe_in]
    type = 'field_filter'
    inputs = ["json_parser"]
    field = "target_parser"
    value = "nginx"

[transforms.json_nginx_parser]
    type = "regex_parser"
    inputs = ["json_nginx_pipe_in"]
    field = "http.nginx_request"
    regex = "(?P<__http_request_method>[a-zA-Z]+) (?P<__http_path>[^\\?]+)(\\?(?P<__http_query>.+))? HTTP/1.1"
    drop_failed = false
    drop_field = false

[transforms.json_nginx_pipe_out]
    type = "lua"
    inputs = ["json_nginx_parser"]

    source = """
if event["http.nginx_request"] ~= nil then
    if event["http.remote_address"] == "-" then
        event["http.remote_address"] = nil
    end
    if event["__http_request_method"] ~= nil then
        event["http.request_method"] = event["__http_request_method"]
        event["__http_request_method"] = nil
    else
        event["http.request_method"] = "WHAT"
    end
    if event["__http_path"] ~= nil then
        event["http.path"] = event["__http_path"]
        event["__http_path"] = nil
    else
        event["http.path"] = "/"
    end
    if event["__http_query"] ~= nil then
        event["http.query"] = event["__http_query"]
        event["__http_query"] = nil
    else
        event["http.query"] = ""
    end
    if event["http.timestamp"] ~= nil then
        event["timestamp"] = event["http.timestamp"]
        event["http.timestamp"] = nil 
    end
    event["http.nginx_request"] = nil
end
"""
# ------------------ JSON NGINX Pipe ---------------------- #

# ------------------ JSON General Pipe -------------------- #

[transforms.json_general_pipe_out]
    type = 'field_filter'
    inputs = ["json_parser"]
    field = "target_parser"
    value = "general"

# ------------------ JSON General Pipe -------------------- #

# ------------------ JSON Traefik Pipe -------------------- #

[transforms.json_traefik_pipe_in]
    type = 'field_filter'
    inputs = ["json_parser"]
    field = "target_parser"
    value = "traefik"

[transforms.json_traefik_field_remap]
    type = "lua"
    inputs = ["json_traefik_pipe_in"]

    source = """
event["RequestProtocol"] = nil
event["RequestLine"] = nil

event["timestamp"] = event["time"]
event["time"] = nil

if event["msg"] ~= nil then
    event["event"] = event["msg"]
    event["msg"] = nil
end


event["http.response_length"] = event["traefik.OriginContentSize"]
event["http.request_method"] = event["traefik.RequestMethod"]
event["http.referrer"] = event["traefik.request_Referer"]
event["http.user_agent"] = event["traefik.request_User-Agent"]
event["http.status"] = event["traefik.OriginStatus"]
event["http.remote_address"] = event["traefik.ClientHost"]
if event["traefik.Duration"] ~= nil then
    event["http.request_time"] = tonumber(event["traefik.Duration"]) / 1000000000
else
    event["traefik.Duration"] = nil
end

for key, _ in pairs(event) do
    if key:lower() ~= key then
        event["traefik." .. key] = event[key]
        event[key] = nil
    end
end
"""

[transforms.json_traefik_path_and_query_split]
    type = "regex_parser"
    inputs = ["json_traefik_field_remap"]
    regex = "(?P<__http_path>[^\\?]+)(\\?(?P<__http_query>[^\\?]+))?"
    field = "traefik.RequestPath"

[transforms.json_traefik_path_and_query_remap]
    type = "lua"
    inputs = ["json_traefik_path_and_query_split"]
    source = """
event["http.path"] = event["__http_path"]
event["http.query"] = event["__http_query"]
event["__http_query"] = nil
event["__http_path"] = nil
"""

[transforms.json_traefik_pipe_out]
  type = "coercer"
  inputs = ["json_traefik_path_and_query_remap"]
  

  [transforms.json_traefik_pipe_out.types]

    "http.response_length" = "int"
    "http.status" = "int"
    "http.request_time" = "float"


# ------------------ JSON Traefik Pipe -------------------- #

[transforms.json_pipe_out]
    type = "lua"
    inputs = ["json_nginx_pipe_out", 'json_general_pipe_out', 'json_traefik_pipe_out']

    source = """
if event["http.remote_address"] == "-" then
    event["http.remote_address"] = nil
end
"""

# ------------------ JSON Pipe ---------------------------- #

# ------------------ Text Pipe ---------------------------- #

[transforms.text_pipe_in]
    type = 'field_filter'
    inputs = ["syslog_pipe_out"]
    field = "target_format"
    value = "text"

# ------------------ Text PostgreSQL Pipe ----------------- #

[transforms.text_postgresql_pipe]
    type = 'field_filter'
    inputs = ["text_pipe_in"]
    field = "target_parser"
    value = "postgresql"

[transforms.text_postgresql_pipe_parse]
    type = "grok_parser"
    inputs = ["text_postgresql_pipe"]
    pattern = "%{DATESTAMP:timestamp} %{TZ} \\[%{NUMBER:postgresql.process.pid}(-%{BASE16FLOAT:postgresql.log.core_id})?\\] ((\\[%{USERNAME:user.name}\\]@\\[%{[a-zA-Z0-9_]+[a-zA-Z0-9_\\$]*:postgresql.log.database}\\]|%{USERNAME:user.name}@%{[a-zA-Z0-9_]+[a-zA-Z0-9_\\\\$]*:postgresql.log.database}) )?%{WORD:log.level}:  (?:%{NUMBER:postgresql.log.error.code}|%{SPACE})(duration: %{NUMBER:postgresql.duration} ms  %{WORD:postgresql.log.query_step}(?: <unnamed>| %{WORD:postgresql.log.query_name})?: %{GREEDYDATA:postgresql.log.query}|: %{GREEDYDATA:message}|%{GREEDYDATA:message})"
    drop_field = true

    [transforms.text_postgresql_pipe_parse.types]
        "postgresql.log.core_id" = "int"
        "postgresql.log.error.code" = "int"
        "postgresql.process.pid" = "int"
        "postgresql.duration" = "float"
        "timestamp" = "timestamp|%y-%m-%d %H:%M:%S%.3f"


[transforms.text_postgresql_pipe_processing]
    type = "lua"
    inputs = ["text_postgresql_pipe_parse"]
    source = """
if event["postgresql.duration"] ~= nil then
    event["postgresql.duration"] = event["postgresql.duration"] / 1000
end
event["name18"] = nil
event["name29"] = nil
event["level"] = string.lower(event["log.level"])
event["log.level"] = nil
"""

[transforms.text_postgresql_pipe_out]
  type = "coercer"
  inputs = ["text_postgresql_pipe_processing"]
  

  [transforms.text_postgresql_pipe_out.types]
    "postgresql.duration" = "float"


# ------------------ Text PostgreSQL Pipe ----------------- #

# ------------------ Text Pipe ---------------------------- #

# ------------------ Extra Pipe --------------------------- #

# ------------------ Extra Pipe --------------------------- #