logs-collector/vector.yaml

# Set global options
data_dir: "/var/lib/vector"

# Vector's API (disabled by default)
# Enable and try it out with the `vector top` command
# NOTE: this is _enabled_ for helm chart deployments, see:.apps://github.com/vectordotdev/helm-charts/blob/develop/charts/vector/examples/datadog-values.yaml#L78-L81
api:
  enabled: false
# address = "127.0.0.1:8686"

sources:
  from_stdin:
    type: stdin

transforms:
  pack:
    type: remap
    inputs:
      - from_stdin
    source: >-
      l = .

      #log("!!! 1", level:"info", rate_limit_secs:0)

      #log(l, level:"info", rate_limit_secs:0)

      .app = {}

      .app.timestamp = to_unix_timestamp(now())

      .app.trace_id = ""

      .app.span_id = ""
      
      .app.trace_flags = 0

      .app.severity_text = "INFO"
      
      .app.severity_number = 0

      .app.attributes = {}

      .app.resources = {}
      
      .app.body = ""

      #log("!!! 2", level:"info")

      m = if (exists(.message)) {
        to_string!(.message)
      } else {
        "NO MESSAGE"
      }

      msg, err = parse_json(m)

      if (err == null) {
        #log("!!! 2.1: {{m}}", level:"info", rate_limit_secs:0)

        .app.body = if (exists(msg.message)) {
          msg.message
        } else {
          m
        }

        #log("!!! 2.1.1: {{m}}", level:"info", rate_limit_secs:0)

        trace = if (exists(msg.message)) {
          msgjson, e = parse_regex(msg.message, r'request_id=(?P<trace_id>\S+)')

          if (e == null) {
            msgjson
          } else {
            {}
          }
        } else {
          {}
        }

        #log("!!! 2.1.2: {{m}}", level:"info", rate_limit_secs:0)

        .app.trace_id = if (exists(trace.trace_id) && trace.trace_id != "") {
          trace.trace_id
        } else if (msg.trace_id != "") {
          msg.trace_id
        } else {
          ""
        }

        #log("!!! 2.1.3: {{m}}", level:"info", rate_limit_secs:0)

        .app.trace_id = if (exists(.app.trace_id) && .app.trace_id != null && .app.trace_id != "") {
          span = parse_regex!(.app.trace_id, r'(?<a1>\w+)-(?<a2>\w+)-(?<a3>\w+)-(?<a4>\w+)-(?<a5>\w+)')

          a1 = span.a1

          a2 = span.a2

          a3 = span.a3

          a4 = span.a4

          a5 = span.a5

          "{{a1}}{{a2}}{{a3}}{{a4}}{{a5}}"
        } else {
          ""
        }

        #log("!!! 2.1.4: {{m}}", level:"info", rate_limit_secs:0)

        .app.severity_text = if (exists(msg.level)) {
          s, e = to_string(msg.level)

          if (e == null) {
            s
          } else {
            "BAD"
          }
        } else {
          "INFO"
        }

        #log("!!! 2.1.5: {{m}}", level:"info", rate_limit_secs:0)

        .app.severity_number = if (exists(msg.level_value)) {
          i, e = to_int(msg.level_value)

          if (e == null) {
            i
          } else {
            0
          }
        }
        
        #log("!!! 2.1.6: {{m}}", level:"info", rate_limit_secs:0)

        .app.attributes.message = msg
        
        #log("!!! 2.1.7: {{m}}", level:"info", rate_limit_secs:0)

        .app.attributes.logger_name = if (exists(msg.logger_name)) {
          msg.logger_name
        } else {
          "none"
        }

        #log("!!! 2.1.8: {{m}}", level:"info", rate_limit_secs:0)
        
        .app.attributes._timestamp = if (exists(msg.@timestamp)) {
          msg.@timestamp
        } else {
          now()
        }
        
        #log("!!! 2.1.9: {{m}}", level:"info", rate_limit_secs:0)

        .app.attributes.thread_name = if (exists(msg.thread_name)) {
          msg.thread_name
        } else {
          "none"
        }
        
        #log("!!! 2.1.10: {{m}}", level:"info", rate_limit_secs:0)

        .app.attributes.hostname = "$HOSTNAME"

        #log("!!! 2.1.11: {{m}}", level:"info", rate_limit_secs:0)
        
        .app.attributes.app_name = "${APP_NAME?Specify env.APP_NAME}"
      } else {
        #log("!!! 2.2: {{m}}", level:"warn", rate_limit_secs:0)

        .app.body = m

        #log(err, level:"error", rate_limit_secs:0)
      }

      #log("!!! 3", level:"warn", rate_limit_secs:0)

      del(.message)

sinks:
  to_stdout:
    type: console
    inputs:
      - pack
    encoding: 
      codec: json
      json:
        pretty: true
  to_rabbitmq:
    type: amqp
    inputs:
      - pack
    connection_string: amqp://rmuser:${RABBITMQ_PASS}@${LOGS_CONTAINER_HOST}:5672/%2f?timeout=10
    exchange: logs.add
    encoding:
      codec: json
      json:
        pretty: true