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 containers (including openshift containers)
  [sources.input_myinfra_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_myinfra_container_meta]
  type = "remap"
  inputs = ["input_myinfra_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
  }
  
      '''
  
  # 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_mypipeline_my_labels_0]
  type = "remap"
  inputs = ["input_myinfra_container_meta","input_mytestapp_container_meta"]
  source = '''
    ._internal.openshift.labels = .openshift.labels = {"key1":"value1","key2":"value2"}
  '''
  
  [transforms.pipeline_mypipeline_viaq_1]
  type = "remap"
  inputs = ["pipeline_mypipeline_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 .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_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_mypipeline_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"