sources:
  kamailio_file:
    type: file
    include:
      - /var/log/kamailio.log
    read_from: end

transforms:
  kamailio_only:
    type: filter
    inputs:
      - kamailio_file
    condition: |
      contains(string!(.message), "/usr/sbin/kamailio") &&
      contains(string!(.message), "\"module\": \"xlog\"") &&
      contains(string!(.message), "\"level\": \"INFO\"") &&
      !contains(string!(.message), "[RTPENGINE]")

  parse_kamailio:
    type: remap
    inputs:
      - kamailio_only
    source: |
      raw = string!(.message)

      parsed_syslog = parse_regex!(
        raw,
        r'^(?P<syslog_ts>\S+)\s+(?P<syslog_host>\S+)\s+(?P<process>[^\[]+)\[(?P<pid>\d+)\]:\s+(?P<kam_json>\{.*\})$'
      )

      event_ts = parse_timestamp!(
        parsed_syslog.syslog_ts,
        format: "%+"
      )

      kam = parse_json!(parsed_syslog.kam_json)
      kam_msg = string!(kam.message)
      original_message = .message

      . = {}

      .timestamp = event_ts
      .log_timestamp = parsed_syslog.syslog_ts
      .syslog_host = parsed_syslog.syslog_host
      .process = parsed_syslog.process
      .pid = to_int!(parsed_syslog.pid)

      .event = ""
      .calldirection = ""
      .srcip = ""
      .srcport = ""
      .callid = ""
      .ruri = ""
      .method = ""
      .respcode = ""
      .respclass = ""
      .respmess = ""
      .cseq = ""
      .fromuser = ""
      .fromtag = ""
      .touser = ""
      .totag = ""
      .dsturi = ""
      .dstname = ""
      .callend = ""
      .label = ""

      .kam_level = kam.level
      .kam_module = kam.module
      .kam_file = kam.file
      .kam_line = kam.line
      .kam_function = kam.function
      .logprefix = kam.logprefix
      .raw = original_message

      if contains(kam_msg, "sl:filtered-ack ACK to local reply absorbed") {
        pfx = parse_regex!(
          string!(kam.logprefix),
          r'^\{(?P<route_type>\S+)\s+(?P<cseq>\S+)\s+(?P<method>\S+)\s+(?P<callid>[^\}]+)\}'
        )

        .event = "local_ack_absorbed"
        .callid = pfx.callid
        .method = pfx.method
        .cseq = pfx.cseq
        .label = "ACK to local reply absorbed"
      } else {
        msg = replace(kam_msg, r'^[A-Z]+:\s+', "")

        kv, err = parse_regex(
          msg,
          r'^event=(?P<event>\S+) calldirection=(?P<calldirection>\S+) srcip=(?P<srcip>\S+) srcport=(?P<srcport>\S+) callid=(?P<callid>\S+) ruri=(?P<ruri>\S+) method=(?P<method>\S+) respcode=(?P<respcode>\S+) respmess=(?P<respmess>.*?) cseq=(?P<cseq>\S+) fromuser=(?P<fromuser>\S+) fromtag=(?P<fromtag>\S+) touser=(?P<touser>\S+) totag=(?P<totag>\S+) dsturi=(?P<dsturi>\S+) callend=(?P<callend>\S*) label=(?P<label>.*)$'
        )
        if err != null {
          abort
        }
        . = merge(., kv)

        .label = strip_whitespace(to_string(.label))
        .label = replace(.label, "'", "")

      }

      for_each(object(.)) -> |key, value| {
        if value == "<null>" {
          . = set!(., [key], "")
        }
      }

      if !exists(.callend) || .callend == null {
        .callend = ""
      }

      if exists(.srcport) && .srcport != "" {
        .srcport = to_int!(.srcport)
      }

      if exists(.respcode) && .respcode != "" {
        .respcode = to_int!(.respcode)
      }

      if .respcode == "" {
        .respclass = "none"
      } else if starts_with(to_string(.respcode), "1") {
        .respclass = "1xx"
      } else if starts_with(to_string(.respcode), "2") {
        .respclass = "2xx"
      } else if starts_with(to_string(.respcode), "3") {
        .respclass = "3xx"
      } else if starts_with(to_string(.respcode), "4") {
        .respclass = "4xx"
      } else if starts_with(to_string(.respcode), "5") {
        .respclass = "5xx"
      } else if starts_with(to_string(.respcode), "6") {
        .respclass = "6xx"
      } else {
        .respclass = "other"
      }

      if exists(.cseq) && .cseq != "" {
        .cseq = to_int!(.cseq)
      }

      if .dsturi == "sip:192.168.10.24:5060" {
        .dstname = "MS1"
      } else if .dsturi == "sip:192.168.10.25:5060" {
        .dstname = "MS2"
      } else if .dsturi == "sip:192.168.10.26:5060" {
        .dstname = "MS3"
      } else {
        .dstname = ""
      }


  skip_reinvite_messages:
    type: filter
    inputs:
      - parse_kamailio
    condition: |
      cseq = to_int(.cseq) ?? 0
      totag = to_string(.totag) ?? ""
      method = to_string(.method) ?? ""
      event = to_string(.event) ?? ""
      respcode = to_int(.respcode) ?? 0

      is_reinvite_invite = method == "INVITE" && cseq > 1 && totag != "" && (event == "rx_request" || event == "tx_request")
      is_reinvite_reply = method == "INVITE" && cseq > 1 && totag != "" && (event == "rx_reply" || event == "tx_reply") && (respcode == 100 || respcode == 200)
      is_reinvite_ack = method == "ACK" && cseq > 1 && totag != "" && (event == "rx_request" || event == "tx_request")

      if is_reinvite_invite || is_reinvite_reply || is_reinvite_ack {
        false
      } else {
        true
      }

sinks:
  loki:
    type: loki
    inputs:
      - skip_reinvite_messages
    endpoint: "http://192.168.0.230:3100"
    healthcheck:
        enabled: false
    buffer:
        type: "disk"
        max_size: 268435488
        when_full: "block"

    encoding:
      codec: json

    labels:
      app: kamailio
      host: "{{ syslog_host }}"
      event: "{{ event }}"
      method: "{{ method }}"
      respcode: "{{ respcode }}"
      respclass: "{{ respclass }}"
      dstname: "{{ dstname }}"
      calldirection: "{{ calldirection }}"
      callend: "{{ callend }}"
      cseq: "{{ cseq }}"
