json.river

/*
Module: log-format-json
Description: Log Processing for Generic JSON
*/
argument "forward_to" {
  // comment = "Must be a list(LogsReceiver) where collected logs should be forwarded to"
  optional = false
}

export "process" {
  value = loki.process.log_format_json
}

loki.process "log_format_json" {
  forward_to = argument.forward_to.value

  // check logs.agent.grafana.com/log-format annotation, if the log_type is empty the line hasn't been processed, if it contains json and the line matches the format, then process the line as json
  stage.match {
    pipeline_name = "pipeline for annotation || logs.agent.grafana.com/log-format: json"
    selector = "{log_type=\"\", logs_agent_grafana_com_log_format=~\"(?i).*((generic-?)?json).*\"} |~ \"^\\\\s*\\\\{.+\\\\}\\\\s*$\""

    // set the log_type
    stage.static_labels{
      values = {
        log_type = "json",
      }
    }

    // extract the level
    stage.json {
      expressions = {
        level = "level || lvl || loglevel || LogLevel || log_level || logLevel || log_lvl || logLvl || levelname || levelName || LevelName",
      }
    }

    // set the extracted level as a label
    stage.labels {
      values = {
        level = "",
      }
    }

    // check logs.agent.grafana.com/scrub-timestamp annotation, if true remove the timestamp from the log line
    // this can reduce the overall # of bytes sent and stored in Loki
    // remove timestamp from the log line, depending on the entry it can be "start_time" or "time"
    stage.match {
      selector = "{logs_agent_grafana_com_scrub_timestamp=\"true\"}"
      pipeline_name = "pipeline for annotation || logs.agent.grafana.com/scrub-timestamp: true"

      // remove timestamp from the log line
      // unescaped regex: (?i)("(timestamp|ts|logdate|time)"\s*:\s*"[^"]+",?)
      stage.replace {
        expression = "(?i)(\"(timestamp|ts|logdate|time)\"\\s*:\\s*\"[^\"]+\",?)"
        replace = ""
      }
    }

    // check logs.agent.grafana.com/scrub-level annotation, if true remove the level from the log line (it is still a label)
    // this can reduce the overall # of bytes sent and stored in Loki
    stage.match {
      selector = "{logs_agent_grafana_com_scrub_level=~\"(?i)true\"}"
      pipeline_name = "pipeline for annotation || logs.agent.grafana.com/scrub-level: true"

      // remove level from the log line
      stage.replace {
        // unescaped regex: (?i)"(log)?(level|lvl)"\s*:\s*"[^"]+",?
        expression = "(?i)(\"(log)?(level|lvl)\"\\s*:\\s*\"[^\"]+\",?)"
        replace = ""
      }
    }

  }

}