content: expire_metrics_secs = 60
  data_dir = "/var/lib/vector/openshift-logging/my-forwarder"
  
  [api]
  enabled = true
  address = "0.0.0.0:24686"
  
  # Load sensitive data from files
  [secret.kubernetes_secret]
  type = "directory"
  path = "/var/run/ocp-collector/secrets"
  
  [sources.internal_metrics]
  type = "internal_metrics"
  
  # Logs from host audit
  [sources.input_audit_host]
  type = "file"
  include = ["/var/log/audit/audit.log"]
  host_key = "hostname"
  glob_minimum_cooldown_ms = 15000
  ignore_older_secs = 3600
  max_line_bytes = 3145728
  max_read_bytes =  262144
  rotate_wait_secs = 5
  
  [transforms.input_audit_host_meta]
  type = "remap"
  inputs = ["input_audit_host"]
  source = '''
    . = {"_internal": .}
    ._internal.log_source = "auditd"
    ._internal.log_type = "audit"
    ._internal.hostname = get_env_var("VECTOR_SELF_NODE_NAME") ?? ""
    ._internal.openshift = { "cluster_id": "${OPENSHIFT_CLUSTER_ID:-}"}
    ._internal.openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  
    if !exists(._internal.level) {
    level = null
    message = ._internal.message
  
    # attempt 1: parse as logfmt (e.g. level=error msg="Failed to connect")
  
    parsed_logfmt, err = parse_logfmt(message)
    if err == null && is_string(parsed_logfmt.level) {
      level = downcase!(parsed_logfmt.level)
    }
  
    # attempt 2: parse as klog (e.g. I0920 14:22:00.089385 1 scheduler.go:592] "Successfully bound pod to node")
    if level == null {
      parsed_klog, err = parse_klog(message)
      if err == null && is_string(parsed_klog.level) {
        level = parsed_klog.level
      }
    }
  
    # attempt 3: parse with groks template (if previous attempts failed) for classic text logs like Logback, Log4j etc.
  
    if level == null {
      parsed_grok, err = parse_groks(message,
        patterns: [
          "%{common_prefix} %{_message}"
        ],
        aliases: {
          "common_prefix": "%{_timestamp} %{_loglevel}",
          "_timestamp": "%{TIMESTAMP_ISO8601:timestamp}",
          "_loglevel": "%{LOGLEVEL:level}",
          "_message": "%{GREEDYDATA:message}"
        }
      )
  
      if err == null && is_string(parsed_grok.level) {
        level = downcase!(parsed_grok.level)
      }
    }
  
    if level == null {
      level = "default"
  
      # attempt 4: Match on well known structured patterns
      # Order: emergency, alert, critical, error, warn, notice, info, debug, trace
  
      if match!(message, r'^EM[0-9]+|level=emergency|Value:emergency|"level":"emergency"') {
        level = "emergency"
      } else if match!(message, r'^A[0-9]+|level=alert|Value:alert|"level":"alert"') {
        level = "alert"
      } else if match!(message, r'^C[0-9]+|level=critical|Value:critical|"level":"critical"') {
        level = "critical"
      } else if match!(message, r'^E[0-9]+|level=error|Value:error|"level":"error"') {
        level = "error"
      } else if match!(message, r'^W[0-9]+|level=warn|Value:warn|"level":"warn"') {
        level = "warn"
      } else if match!(message, r'^N[0-9]+|level=notice|Value:notice|"level":"notice"') {
        level = "notice"
      } else if match!(message, r'^I[0-9]+|level=info|Value:info|"level":"info"') {
        level = "info"
      } else if match!(message, r'^D[0-9]+|level=debug|Value:debug|"level":"debug"') {
        level = "debug"
      } else if match!(message, r'^T[0-9]+|level=trace|Value:trace|"level":"trace"') {
        level = "trace"
      }
  
      # attempt 5: Match on the keyword that appears earliest in the message
      if level == "default" {
  	  level_patterns = r'(?i)(?<emergency>emergency|<emergency>)|(?<alert>alert|<alert>)|(?<critical>critical|<critical>)|(?<error>error|<error>)|(?<warn>warn(?:ing)?|<warn>)|(?<notice>notice|<notice>)|(?:\b(?<info>info)\b|<info>)|(?<debug>debug|<debug>)|(?<trace>trace|<trace>)'
  	  parsed, err = parse_regex(message, level_patterns)
  	  if err == null {
  		if is_string(parsed.emergency) {
  		  level = "emergency"
  		} else if is_string(parsed.alert) {
  		  level = "alert"
  		} else if is_string(parsed.critical) {
  		  level = "critical"
  		} else if is_string(parsed.error) {
  		  level = "error"
  		} else if is_string(parsed.warn) {
  		  level = "warn"
  		} else if is_string(parsed.notice) {
  		  level = "notice"
  		} else if is_string(parsed.info) {
  		  level = "info"
  		} else if is_string(parsed.debug) {
  		  level = "debug"
  		} else if is_string(parsed.trace) {
  		  level = "trace"
  		}
  	  }
  	}
    }
    ._internal.level = level
  }
  
  
    match1 = parse_regex(._internal.message, r'type=(?P<type>[^ ]+)') ?? {}
    envelop = {}
    envelop |= {"type": match1.type}
  
    match2, err = parse_regex(._internal.message, r'msg=audit\((?P<ts_record>[^ ]+)\):')
    if err == null {
      sp, err = split(match2.ts_record,":")
      if err == null && length(sp) == 2 {
          ts = parse_timestamp(sp[0],"%s.%3f") ?? ""
          if ts != "" { ._internal.timestamp = ts }
          envelop |= {"record_id": sp[1]}
          ._internal |= {"audit.linux" : envelop}
          ._internal.timestamp =  format_timestamp(ts,"%+") ?? ""
      }
    } else {
      log("could not parse host audit msg. err=" + err, rate_limit_secs: 0)
    }
  
  '''
  
  # Logs from kubernetes audit
  [sources.input_audit_kube]
  type = "file"
  include = ["/var/log/kube-apiserver/audit.log"]
  host_key = "hostname"
  glob_minimum_cooldown_ms = 15000
  ignore_older_secs = 3600
  max_line_bytes = 3145728
  max_read_bytes =  262144
  rotate_wait_secs = 5
  
  [transforms.input_audit_kube_meta]
  type = "remap"
  inputs = ["input_audit_kube"]
  source = '''
    . = {"_internal": .}
  
    ._internal.structured = parse_json!(string!(._internal.message))
    ._internal = merge!(._internal,._internal.structured)
  
    ._internal.log_source = "kubeAPI"
    ._internal.log_type = "audit"
    ._internal.hostname = get_env_var("VECTOR_SELF_NODE_NAME") ?? ""
    ._internal.openshift = { "cluster_id": "${OPENSHIFT_CLUSTER_ID:-}"}
    ._internal.openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  '''
  
  # Logs from openshift audit
  [sources.input_audit_openshift]
  type = "file"
  include = ["/var/log/oauth-apiserver/audit.log","/var/log/openshift-apiserver/audit.log","/var/log/oauth-server/audit.log"]
  host_key = "hostname"
  glob_minimum_cooldown_ms = 15000
  ignore_older_secs = 3600
  max_line_bytes = 3145728
  max_read_bytes =  262144
  rotate_wait_secs = 5
  
  [transforms.input_audit_openshift_meta]
  type = "remap"
  inputs = ["input_audit_openshift"]
  source = '''
    . = {"_internal": .}
  
    ._internal.structured = parse_json!(string!(._internal.message))
    ._internal = merge!(._internal,._internal.structured)
  
    ._internal.log_source = "openshiftAPI"
    ._internal.log_type = "audit"
    ._internal.hostname = get_env_var("VECTOR_SELF_NODE_NAME") ?? ""
    ._internal.openshift = { "cluster_id": "${OPENSHIFT_CLUSTER_ID:-}"}
    ._internal.openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  '''
  
  # Logs from ovn audit
  [sources.input_audit_ovn]
  type = "file"
  include = ["/var/log/ovn/acl-audit-log.log"]
  host_key = "hostname"
  glob_minimum_cooldown_ms = 15000
  ignore_older_secs = 3600
  max_line_bytes = 3145728
  max_read_bytes =  262144
  rotate_wait_secs = 5
  
  [transforms.input_audit_ovn_meta]
  type = "remap"
  inputs = ["input_audit_ovn"]
  source = '''
    . = {"_internal": .}
    ._internal.log_source = "ovn"
    ._internal.log_type = "audit"
    ._internal.hostname = get_env_var("VECTOR_SELF_NODE_NAME") ?? ""
    ._internal.openshift = { "cluster_id": "${OPENSHIFT_CLUSTER_ID:-}"}
    ._internal.openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  
    if !exists(._internal.level) {
    level = null
    message = ._internal.message
  
    # attempt 1: parse as logfmt (e.g. level=error msg="Failed to connect")
  
    parsed_logfmt, err = parse_logfmt(message)
    if err == null && is_string(parsed_logfmt.level) {
      level = downcase!(parsed_logfmt.level)
    }
  
    # attempt 2: parse as klog (e.g. I0920 14:22:00.089385 1 scheduler.go:592] "Successfully bound pod to node")
    if level == null {
      parsed_klog, err = parse_klog(message)
      if err == null && is_string(parsed_klog.level) {
        level = parsed_klog.level
      }
    }
  
    # attempt 3: parse with groks template (if previous attempts failed) for classic text logs like Logback, Log4j etc.
  
    if level == null {
      parsed_grok, err = parse_groks(message,
        patterns: [
          "%{common_prefix} %{_message}"
        ],
        aliases: {
          "common_prefix": "%{_timestamp} %{_loglevel}",
          "_timestamp": "%{TIMESTAMP_ISO8601:timestamp}",
          "_loglevel": "%{LOGLEVEL:level}",
          "_message": "%{GREEDYDATA:message}"
        }
      )
  
      if err == null && is_string(parsed_grok.level) {
        level = downcase!(parsed_grok.level)
      }
    }
  
    if level == null {
      level = "default"
  
      # attempt 4: Match on well known structured patterns
      # Order: emergency, alert, critical, error, warn, notice, info, debug, trace
  
      if match!(message, r'^EM[0-9]+|level=emergency|Value:emergency|"level":"emergency"') {
        level = "emergency"
      } else if match!(message, r'^A[0-9]+|level=alert|Value:alert|"level":"alert"') {
        level = "alert"
      } else if match!(message, r'^C[0-9]+|level=critical|Value:critical|"level":"critical"') {
        level = "critical"
      } else if match!(message, r'^E[0-9]+|level=error|Value:error|"level":"error"') {
        level = "error"
      } else if match!(message, r'^W[0-9]+|level=warn|Value:warn|"level":"warn"') {
        level = "warn"
      } else if match!(message, r'^N[0-9]+|level=notice|Value:notice|"level":"notice"') {
        level = "notice"
      } else if match!(message, r'^I[0-9]+|level=info|Value:info|"level":"info"') {
        level = "info"
      } else if match!(message, r'^D[0-9]+|level=debug|Value:debug|"level":"debug"') {
        level = "debug"
      } else if match!(message, r'^T[0-9]+|level=trace|Value:trace|"level":"trace"') {
        level = "trace"
      }
  
      # attempt 5: Match on the keyword that appears earliest in the message
      if level == "default" {
  	  level_patterns = r'(?i)(?<emergency>emergency|<emergency>)|(?<alert>alert|<alert>)|(?<critical>critical|<critical>)|(?<error>error|<error>)|(?<warn>warn(?:ing)?|<warn>)|(?<notice>notice|<notice>)|(?:\b(?<info>info)\b|<info>)|(?<debug>debug|<debug>)|(?<trace>trace|<trace>)'
  	  parsed, err = parse_regex(message, level_patterns)
  	  if err == null {
  		if is_string(parsed.emergency) {
  		  level = "emergency"
  		} else if is_string(parsed.alert) {
  		  level = "alert"
  		} else if is_string(parsed.critical) {
  		  level = "critical"
  		} else if is_string(parsed.error) {
  		  level = "error"
  		} else if is_string(parsed.warn) {
  		  level = "warn"
  		} else if is_string(parsed.notice) {
  		  level = "notice"
  		} else if is_string(parsed.info) {
  		  level = "info"
  		} else if is_string(parsed.debug) {
  		  level = "debug"
  		} else if is_string(parsed.trace) {
  		  level = "trace"
  		}
  	  }
  	}
    }
    ._internal.level = level
  }
  
  '''
  
  # Logs from containers (including openshift containers)
  [sources.input_infrastructure_container]
  type = "kubernetes_logs"
  max_read_bytes = 3145728
  glob_minimum_cooldown_ms = 15000
  auto_partial_merge = true
  include_paths_glob_patterns = ["/var/log/pods/default_*/*/*.log", "/var/log/pods/kube*_*/*/*.log", "/var/log/pods/openshift*_*/*/*.log"]
  exclude_paths_glob_patterns = ["/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.log.*", "/var/log/pods/*/*/*.tmp", "/var/log/pods/openshift-logging_*/gateway/*.log", "/var/log/pods/openshift-logging_*/loki*/*.log", "/var/log/pods/openshift-logging_*/opa/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/openshift-logging_logfilesmetricexporter-*/*/*.log"]
  pod_annotation_fields.pod_labels = "kubernetes.labels"
  pod_annotation_fields.pod_namespace = "kubernetes.namespace_name"
  pod_annotation_fields.pod_annotations = "kubernetes.annotations"
  pod_annotation_fields.pod_uid = "kubernetes.pod_id"
  pod_annotation_fields.pod_node_name = "hostname"
  namespace_annotation_fields.namespace_uid = "kubernetes.namespace_id"
  rotate_wait_secs = 5
  use_apiserver_cache = true
  
  [transforms.input_infrastructure_container_meta]
  type = "remap"
  inputs = ["input_infrastructure_container"]
  source = '''
    . = {"_internal": .}
    if exists(._internal.stream) {._internal.kubernetes.container_iostream = ._internal.stream}
    ._internal.log_source = "container"
  
      # If namespace is infra, label log_type as infra
      if match_any(string!(._internal.kubernetes.namespace_name), [r'^default$', r'^openshift(-.+)?$', r'^kube(-.+)?$']) {
          ._internal.log_type = "infrastructure"
      } else {
          ._internal.log_type = "application"
      }
  
    ._internal.hostname = get_env_var("VECTOR_SELF_NODE_NAME") ?? ""
    ._internal.openshift = { "cluster_id": "${OPENSHIFT_CLUSTER_ID:-}"}
    ._internal.openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  
    if !exists(._internal.level) {
    level = null
    message = ._internal.message
  
    # attempt 1: parse as logfmt (e.g. level=error msg="Failed to connect")
  
    parsed_logfmt, err = parse_logfmt(message)
    if err == null && is_string(parsed_logfmt.level) {
      level = downcase!(parsed_logfmt.level)
    }
  
    # attempt 2: parse as klog (e.g. I0920 14:22:00.089385 1 scheduler.go:592] "Successfully bound pod to node")
    if level == null {
      parsed_klog, err = parse_klog(message)
      if err == null && is_string(parsed_klog.level) {
        level = parsed_klog.level
      }
    }
  
    # attempt 3: parse with groks template (if previous attempts failed) for classic text logs like Logback, Log4j etc.
  
    if level == null {
      parsed_grok, err = parse_groks(message,
        patterns: [
          "%{common_prefix} %{_message}"
        ],
        aliases: {
          "common_prefix": "%{_timestamp} %{_loglevel}",
          "_timestamp": "%{TIMESTAMP_ISO8601:timestamp}",
          "_loglevel": "%{LOGLEVEL:level}",
          "_message": "%{GREEDYDATA:message}"
        }
      )
  
      if err == null && is_string(parsed_grok.level) {
        level = downcase!(parsed_grok.level)
      }
    }
  
    if level == null {
      level = "default"
  
      # attempt 4: Match on well known structured patterns
      # Order: emergency, alert, critical, error, warn, notice, info, debug, trace
  
      if match!(message, r'^EM[0-9]+|level=emergency|Value:emergency|"level":"emergency"') {
        level = "emergency"
      } else if match!(message, r'^A[0-9]+|level=alert|Value:alert|"level":"alert"') {
        level = "alert"
      } else if match!(message, r'^C[0-9]+|level=critical|Value:critical|"level":"critical"') {
        level = "critical"
      } else if match!(message, r'^E[0-9]+|level=error|Value:error|"level":"error"') {
        level = "error"
      } else if match!(message, r'^W[0-9]+|level=warn|Value:warn|"level":"warn"') {
        level = "warn"
      } else if match!(message, r'^N[0-9]+|level=notice|Value:notice|"level":"notice"') {
        level = "notice"
      } else if match!(message, r'^I[0-9]+|level=info|Value:info|"level":"info"') {
        level = "info"
      } else if match!(message, r'^D[0-9]+|level=debug|Value:debug|"level":"debug"') {
        level = "debug"
      } else if match!(message, r'^T[0-9]+|level=trace|Value:trace|"level":"trace"') {
        level = "trace"
      }
  
      # attempt 5: Match on the keyword that appears earliest in the message
      if level == "default" {
  	  level_patterns = r'(?i)(?<emergency>emergency|<emergency>)|(?<alert>alert|<alert>)|(?<critical>critical|<critical>)|(?<error>error|<error>)|(?<warn>warn(?:ing)?|<warn>)|(?<notice>notice|<notice>)|(?:\b(?<info>info)\b|<info>)|(?<debug>debug|<debug>)|(?<trace>trace|<trace>)'
  	  parsed, err = parse_regex(message, level_patterns)
  	  if err == null {
  		if is_string(parsed.emergency) {
  		  level = "emergency"
  		} else if is_string(parsed.alert) {
  		  level = "alert"
  		} else if is_string(parsed.critical) {
  		  level = "critical"
  		} else if is_string(parsed.error) {
  		  level = "error"
  		} else if is_string(parsed.warn) {
  		  level = "warn"
  		} else if is_string(parsed.notice) {
  		  level = "notice"
  		} else if is_string(parsed.info) {
  		  level = "info"
  		} else if is_string(parsed.debug) {
  		  level = "debug"
  		} else if is_string(parsed.trace) {
  		  level = "trace"
  		}
  	  }
  	}
    }
    ._internal.level = level
  }
  
  '''
  
  [sources.input_infrastructure_journal]
  type = "journald"
  journal_directory = "/var/log/journal"
  
  [transforms.input_infrastructure_journal_meta]
  type = "remap"
  inputs = ["input_infrastructure_journal"]
  source = '''
    . = {"_internal": .}
    ._internal.log_source = "node"
    ._internal.log_type = "infrastructure"
    ._internal.openshift = { "cluster_id": "${OPENSHIFT_CLUSTER_ID:-}"}
    ._internal.openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  
    if ._internal.PRIORITY == "8" || ._internal.PRIORITY == 8 {
    	._internal.level = "trace"
    } else {
    	priority = to_int!(._internal.PRIORITY)
    	._internal.level, err = to_syslog_level(priority)
    	if err != null {
    		log("Unable to determine level from PRIORITY: " + err, level: "error")
    		log(., level: "error")
    		._internal.level = "unknown"
    	} else {
    		del(._internal.PRIORITY)
    	}
    }
  
    if exists(._internal.MESSAGE) {._internal.message = del(._internal.MESSAGE)}
  
    # systemd’s kernel-specific metadata.
    # .systemd.k = {}
    if exists(._internal.KERNEL_DEVICE) { ._internal.systemd.k.KERNEL_DEVICE = del(._internal.KERNEL_DEVICE) }
    if exists(._internal.KERNEL_SUBSYSTEM) { ._internal.systemd.k.KERNEL_SUBSYSTEM = del(._internal.KERNEL_SUBSYSTEM) }
    if exists(._internal.UDEV_DEVLINK) { ._internal.systemd.k.UDEV_DEVLINK = del(._internal.UDEV_DEVLINK) }
    if exists(._internal.UDEV_DEVNODE) { ._internal.systemd.k.UDEV_DEVNODE = del(._internal.UDEV_DEVNODE) }
    if exists(._internal.UDEV_SYSNAME) { ._internal.systemd.k.UDEV_SYSNAME = del(._internal.UDEV_SYSNAME) }
  
  
    # trusted journal fields, fields that are implicitly added by the journal and cannot be altered by client code.
    ._internal.systemd.t = {}
    if exists(._internal._AUDIT_LOGINUID) { ._internal.systemd.t.AUDIT_LOGINUID = del(._internal._AUDIT_LOGINUID) }
    if exists(._internal._BOOT_ID) { ._internal.systemd.t.BOOT_ID = del(._internal._BOOT_ID) }
    if exists(._internal._AUDIT_SESSION) { ._internal.systemd.t.AUDIT_SESSION = del(._internal._AUDIT_SESSION) }
    if exists(._internal._CAP_EFFECTIVE) { ._internal.systemd.t.CAP_EFFECTIVE = del(._internal._CAP_EFFECTIVE) }
    if exists(._internal._CMDLINE) { ._internal.systemd.t.CMDLINE = del(._internal._CMDLINE) }
    if exists(._internal._COMM) { ._internal.systemd.t.COMM = del(._internal._COMM) }
    if exists(._internal._EXE) { ._internal.systemd.t.EXE = del(._internal._EXE) }
    if exists(._internal._GID) { ._internal.systemd.t.GID = del(._internal._GID) }
    if exists(._internal._HOSTNAME) { ._internal.systemd.t.HOSTNAME = ._internal.hostname }
    if exists(._internal._LINE_BREAK) { ._internal.systemd.t.LINE_BREAK = del(._internal._LINE_BREAK) }
    if exists(._internal._MACHINE_ID) { ._internal.systemd.t.MACHINE_ID = del(._internal._MACHINE_ID) }
    if exists(._internal._PID) { ._internal.systemd.t.PID = del(._internal._PID) }
    if exists(._internal._SELINUX_CONTEXT) { ._internal.systemd.t.SELINUX_CONTEXT = del(._internal._SELINUX_CONTEXT) }
    if exists(._internal._SOURCE_REALTIME_TIMESTAMP) { ._internal.systemd.t.SOURCE_REALTIME_TIMESTAMP = del(._internal._SOURCE_REALTIME_TIMESTAMP) }
    if exists(._internal._STREAM_ID) { ._internal.systemd.t.STREAM_ID = ._internal._STREAM_ID }
    if exists(._internal._SYSTEMD_CGROUP) { ._internal.systemd.t.SYSTEMD_CGROUP = del(._internal._SYSTEMD_CGROUP) }
    if exists(._internal._SYSTEMD_INVOCATION_ID) {._internal.systemd.t.SYSTEMD_INVOCATION_ID = ._internal._SYSTEMD_INVOCATION_ID}
    if exists(._internal._SYSTEMD_OWNER_UID) { ._internal.systemd.t.SYSTEMD_OWNER_UID = del(._internal._SYSTEMD_OWNER_UID) }
    if exists(._internal._SYSTEMD_SESSION) { ._internal.systemd.t.SYSTEMD_SESSION = del(._internal._SYSTEMD_SESSION) }
    if exists(._internal._SYSTEMD_SLICE) { ._internal.systemd.t.SYSTEMD_SLICE = del(._internal._SYSTEMD_SLICE) }
    if exists(._internal._SYSTEMD_UNIT) { ._internal.systemd.t.SYSTEMD_UNIT = del(._internal._SYSTEMD_UNIT) }
    if exists(._internal._SYSTEMD_USER_UNIT) { ._internal.systemd.t.SYSTEMD_USER_UNIT = del(._internal._SYSTEMD_USER_UNIT) }
    if exists(._internal._TRANSPORT) { ._internal.systemd.t.TRANSPORT = del(._internal._TRANSPORT) }
    if exists(._internal._UID) { ._internal.systemd.t.UID = del(._internal._UID) }
  
  
    # fields that are directly passed from clients and stored in the journal.
    ._internal.systemd.u = {}
    if exists(._internal.CODE_FILE) { ._internal.systemd.u.CODE_FILE = del(._internal.CODE_FILE) }
    if exists(._internal.CODE_FUNC) { ._internal.systemd.u.CODE_FUNCTION = del(._internal.CODE_FUNC) }
    if exists(._internal.CODE_LINE) { ._internal.systemd.u.CODE_LINE = del(._internal.CODE_LINE) }
    if exists(._internal.ERRNO) { ._internal.systemd.u.ERRNO = del(._internal.ERRNO) }
    if exists(._internal.MESSAGE_ID) { ._internal.systemd.u.MESSAGE_ID = del(._internal.MESSAGE_ID) }
    if exists(._internal.SYSLOG_FACILITY) { ._internal.systemd.u.SYSLOG_FACILITY = del(._internal.SYSLOG_FACILITY) }
    if exists(._internal.SYSLOG_IDENTIFIER) { ._internal.systemd.u.SYSLOG_IDENTIFIER = del(._internal.SYSLOG_IDENTIFIER) }
    if exists(._internal.SYSLOG_PID) { ._internal.systemd.u.SYSLOG_PID = del(._internal.SYSLOG_PID) }
    if exists(._internal.RESULT) { ._internal.systemd.u.RESULT = del(._internal.RESULT) }
    if exists(._internal.UNIT) { ._internal.systemd.u.UNIT = del(._internal.UNIT) }
  
  '''
  
  # Logs from containers (including openshift containers)
  [sources.input_mytestapp_container]
  type = "kubernetes_logs"
  max_read_bytes = 3145728
  glob_minimum_cooldown_ms = 15000
  auto_partial_merge = true
  include_paths_glob_patterns = ["/var/log/pods/test-ns_*/*/*.log"]
  exclude_paths_glob_patterns = ["/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.log.*", "/var/log/pods/*/*/*.tmp", "/var/log/pods/default_*/*/*.log", "/var/log/pods/kube*_*/*/*.log", "/var/log/pods/openshift*_*/*/*.log"]
  pod_annotation_fields.pod_labels = "kubernetes.labels"
  pod_annotation_fields.pod_namespace = "kubernetes.namespace_name"
  pod_annotation_fields.pod_annotations = "kubernetes.annotations"
  pod_annotation_fields.pod_uid = "kubernetes.pod_id"
  pod_annotation_fields.pod_node_name = "hostname"
  namespace_annotation_fields.namespace_uid = "kubernetes.namespace_id"
  rotate_wait_secs = 5
  use_apiserver_cache = true
  
  [transforms.input_mytestapp_container_meta]
  type = "remap"
  inputs = ["input_mytestapp_container"]
  source = '''
    . = {"_internal": .}
    if exists(._internal.stream) {._internal.kubernetes.container_iostream = ._internal.stream}
    ._internal.log_source = "container"
  
      # If namespace is infra, label log_type as infra
      if match_any(string!(._internal.kubernetes.namespace_name), [r'^default$', r'^openshift(-.+)?$', r'^kube(-.+)?$']) {
          ._internal.log_type = "infrastructure"
      } else {
          ._internal.log_type = "application"
      }
  
    ._internal.hostname = get_env_var("VECTOR_SELF_NODE_NAME") ?? ""
    ._internal.openshift = { "cluster_id": "${OPENSHIFT_CLUSTER_ID:-}"}
    ._internal.openshift.sequence = to_unix_timestamp(now(), unit: "nanoseconds")
  
    if !exists(._internal.level) {
    level = null
    message = ._internal.message
  
    # attempt 1: parse as logfmt (e.g. level=error msg="Failed to connect")
  
    parsed_logfmt, err = parse_logfmt(message)
    if err == null && is_string(parsed_logfmt.level) {
      level = downcase!(parsed_logfmt.level)
    }
  
    # attempt 2: parse as klog (e.g. I0920 14:22:00.089385 1 scheduler.go:592] "Successfully bound pod to node")
    if level == null {
      parsed_klog, err = parse_klog(message)
      if err == null && is_string(parsed_klog.level) {
        level = parsed_klog.level
      }
    }
  
    # attempt 3: parse with groks template (if previous attempts failed) for classic text logs like Logback, Log4j etc.
  
    if level == null {
      parsed_grok, err = parse_groks(message,
        patterns: [
          "%{common_prefix} %{_message}"
        ],
        aliases: {
          "common_prefix": "%{_timestamp} %{_loglevel}",
          "_timestamp": "%{TIMESTAMP_ISO8601:timestamp}",
          "_loglevel": "%{LOGLEVEL:level}",
          "_message": "%{GREEDYDATA:message}"
        }
      )
  
      if err == null && is_string(parsed_grok.level) {
        level = downcase!(parsed_grok.level)
      }
    }
  
    if level == null {
      level = "default"
  
      # attempt 4: Match on well known structured patterns
      # Order: emergency, alert, critical, error, warn, notice, info, debug, trace
  
      if match!(message, r'^EM[0-9]+|level=emergency|Value:emergency|"level":"emergency"') {
        level = "emergency"
      } else if match!(message, r'^A[0-9]+|level=alert|Value:alert|"level":"alert"') {
        level = "alert"
      } else if match!(message, r'^C[0-9]+|level=critical|Value:critical|"level":"critical"') {
        level = "critical"
      } else if match!(message, r'^E[0-9]+|level=error|Value:error|"level":"error"') {
        level = "error"
      } else if match!(message, r'^W[0-9]+|level=warn|Value:warn|"level":"warn"') {
        level = "warn"
      } else if match!(message, r'^N[0-9]+|level=notice|Value:notice|"level":"notice"') {
        level = "notice"
      } else if match!(message, r'^I[0-9]+|level=info|Value:info|"level":"info"') {
        level = "info"
      } else if match!(message, r'^D[0-9]+|level=debug|Value:debug|"level":"debug"') {
        level = "debug"
      } else if match!(message, r'^T[0-9]+|level=trace|Value:trace|"level":"trace"') {
        level = "trace"
      }
  
      # attempt 5: Match on the keyword that appears earliest in the message
      if level == "default" {
  	  level_patterns = r'(?i)(?<emergency>emergency|<emergency>)|(?<alert>alert|<alert>)|(?<critical>critical|<critical>)|(?<error>error|<error>)|(?<warn>warn(?:ing)?|<warn>)|(?<notice>notice|<notice>)|(?:\b(?<info>info)\b|<info>)|(?<debug>debug|<debug>)|(?<trace>trace|<trace>)'
  	  parsed, err = parse_regex(message, level_patterns)
  	  if err == null {
  		if is_string(parsed.emergency) {
  		  level = "emergency"
  		} else if is_string(parsed.alert) {
  		  level = "alert"
  		} else if is_string(parsed.critical) {
  		  level = "critical"
  		} else if is_string(parsed.error) {
  		  level = "error"
  		} else if is_string(parsed.warn) {
  		  level = "warn"
  		} else if is_string(parsed.notice) {
  		  level = "notice"
  		} else if is_string(parsed.info) {
  		  level = "info"
  		} else if is_string(parsed.debug) {
  		  level = "debug"
  		} else if is_string(parsed.trace) {
  		  level = "trace"
  		}
  	  }
  	}
    }
    ._internal.level = level
  }
  
  '''
  
  [transforms.pipeline_pipeline_my_labels_0]
  type = "remap"
  inputs = ["input_audit_host_meta","input_audit_kube_meta","input_audit_openshift_meta","input_audit_ovn_meta","input_infrastructure_container_meta","input_infrastructure_journal_meta","input_mytestapp_container_meta"]
  source = '''
    ._internal.openshift.labels = .openshift.labels = {"key1":"value1","key2":"value2"}
  '''
  
  [transforms.pipeline_pipeline_viaq_1]
  type = "remap"
  inputs = ["pipeline_pipeline_my_labels_0"]
  source = '''
    if exists(._internal.hostname) { .hostname = ._internal.hostname }
    .log_type = ._internal.log_type
    .log_source = ._internal.log_source
  
    if exists(._internal.openshift) {.openshift = ._internal.openshift}
    if exists(._internal.dedot_openshift_labels) {.openshift.labels = del(._internal.dedot_openshift_labels) }
  
  
    if ._internal.log_type == "audit" && ._internal.log_source == "auditd" {
    if !exists(._internal.structured) {
      .message = ._internal.message
    }
    ."audit.linux" = ._internal."audit.linux"
    .level = "default"
    }
  
  
    if ._internal.log_type == "audit" && ._internal.log_source == "kubeAPI" {
    .k8s_audit_level = ._internal.structured.level
    }
  
  
    if ._internal.log_type == "audit" && ._internal.log_source == "openshiftAPI" {
    .openshift_audit_level = ._internal.structured.level
    }
  
  
    if ._internal.log_type == "audit" && ._internal.log_source == "ovn" {
    if !exists(._internal.structured) {
      .message = ._internal.message
    }
    .level = ._internal.level
    }
  
  
    if .log_source == "container" {
      if exists(._internal.kubernetes.pod_name) && starts_with(string!(._internal.kubernetes.pod_name), "eventrouter-") {
      parsed, err = parse_json(._internal.message)
      if err != null {
        log("Unable to process EventRouter log: " + err, level: "info")
      } else {
        ._internal.event = parsed
        if exists(._internal.event.event) && is_object(._internal.event.event) {
            ._internal.kubernetes.event = del(._internal.event.event)
    		._internal.kubernetes.event.verb = ._internal.event.verb
            ._internal.message = del(._internal.kubernetes.event.message)
            ._internal."@timestamp" = .kubernetes.event.metadata.creationTimestamp
        } else {
          log("Unable to merge EventRouter log message into record: " + err, level: "info")
        }
      }
    }
    if ._internal.log_source == "container" {
      if exists(._internal.kubernetes.namespace_labels) {
        ._internal.dedot_namespace_labels = {}
        for_each(object!(._internal.kubernetes.namespace_labels)) -> |key,value| {
          newkey = replace(key, r'[\./]', "_")
          ._internal.dedot_namespace_labels = set!(._internal.dedot_namespace_labels,[newkey],value)
        }
      }
      if exists(._internal.kubernetes.labels) {
        ._internal.dedot_labels = {}
        for_each(object!(._internal.kubernetes.labels)) -> |key,value| {
          newkey = replace(key, r'[\./]', "_")
          ._internal.dedot_labels = set!(._internal.dedot_labels,[newkey],value)
        }
      }
    }
    if exists(._internal.openshift.labels) {for_each(object!(._internal.openshift.labels)) -> |key,value| {
      ._internal.dedot_openshift_labels = {}
      newkey = replace(key, r'[\./]', "_")
      ._internal.dedot_openshift_labels = set!(._internal.dedot_openshift_labels,[newkey],value)
    }}
    .kubernetes = ._internal.kubernetes
    if exists(._internal.dedot_labels) {.kubernetes.labels = del(._internal.dedot_labels) }
    if exists(._internal.dedot_namespace_labels) {.kubernetes.namespace_labels = del(._internal.dedot_namespace_labels) }
    del(.kubernetes.node_labels)
    del(.kubernetes.container_image_id)
    del(.kubernetes.pod_ips)
    if !exists(._internal.structured) {
      .message = ._internal.message
    }
    }
  
  
    if ._internal.log_source == "node" {
      .tag = ".journal.system"
  
    .hostname = del(._internal.host)
  
    .systemd = ._internal.systemd
  
    if !exists(._internal.structured) {
      .message = ._internal.message
    }
    }
  
  
    if ._internal.log_type == "audit" && exists(._internal.structured) {. = merge!(.,._internal.structured) }
    if ._internal.log_source == "syslog" && exists(._internal.structured) {. = merge!(.,._internal.structured) }
    if ._internal.log_source == "container" && exists(._internal.structured) {.structured = ._internal.structured }
  
    .timestamp = ._internal.timestamp
    ."@timestamp" = ._internal.timestamp
  
    if ._internal.log_type != "audit" && exists(._internal.level) {
      .level = ._internal.level
    }
  
  '''
  
  # Kafka Topic
  [transforms.output_kafka_receiver_topic]
  type = "remap"
  inputs = ["pipeline_pipeline_viaq_1"]
  source = '''
    ._internal.output_kafka_receiver_topic = "topic"
  
  '''
  
  [sinks.output_kafka_receiver]
  type = "kafka"
  inputs = ["output_kafka_receiver_topic"]
  bootstrap_servers = "broker1-kafka.svc.messaging.cluster.local:9092"
  topic = "{{ _internal.output_kafka_receiver_topic }}"
  healthcheck.enabled = false
  
  
  [sinks.output_kafka_receiver.encoding]
  codec = "json"
  timestamp_format = "rfc3339"
  except_fields = ["_internal"]
  
  
  
  
  [sinks.output_kafka_receiver.tls]
  enabled = true
  min_tls_version = "VersionTLS12"
  ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"
  key_file = "/var/run/ocp-collector/secrets/kafka-receiver-1/tls.key"
  crt_file = "/var/run/ocp-collector/secrets/kafka-receiver-1/tls.crt"
  ca_file = "/var/run/ocp-collector/secrets/kafka-receiver-1/ca-bundle.crt"
  
  [transforms.add_nodename_to_metric]
  type = "remap"
  inputs = ["internal_metrics"]
  source = '''
  .tags.hostname = get_env_var!("VECTOR_SELF_NODE_NAME")
  '''
  
  [sinks.prometheus_output]
  type = "prometheus_exporter"
  inputs = ["add_nodename_to_metric"]
  address = "[::]:24231"
  default_namespace = "collector"
  
  [sinks.prometheus_output.tls]
  enabled = true
  key_file = "/etc/collector/metrics/tls.key"
  crt_file = "/etc/collector/metrics/tls.crt"
  min_tls_version = "VersionTLS12"
  ciphersuites = "TLS_AES_128_GCM_SHA256,TLS_AES_256_GCM_SHA384,TLS_CHACHA20_POLY1305_SHA256,ECDHE-ECDSA-AES128-GCM-SHA256,ECDHE-RSA-AES128-GCM-SHA256,ECDHE-ECDSA-AES256-GCM-SHA384,ECDHE-RSA-AES256-GCM-SHA384,ECDHE-ECDSA-CHACHA20-POLY1305,ECDHE-RSA-CHACHA20-POLY1305,DHE-RSA-AES128-GCM-SHA256,DHE-RSA-AES256-GCM-SHA384"
  