apiVersion: v1
data:
  cleanInValidJson.rb: |2+

    #!/usr/bin/env ruby

    require 'yajl'
    require 'json'


    #example pos file where issue was reported - FILE = "/var/lib/fluentd/pos/journal_pos.json"

    ARGV.each do |filename|

     input = File.read(filename)

     puts "checking if #{filename} a valid json by calling yajl parser"

     @default_options ||= {:symbolize_keys => false}

     begin
       Yajl::Parser.parse(input, @default_options )
     rescue Yajl::ParseError => e
       raise e.message
     end

    end

  fluent.conf: |-
    ## CLO GENERATED CONFIGURATION ###
    # This file is a copy of the fluentd configuration entrypoint
    # which should normally be supplied in a configmap.

    <system>
      log_level "#{ENV['LOG_LEVEL'] || 'warn'}"
    </system>

    # Prometheus Monitoring
    <source>
      @type prometheus
      bind "#{ENV['PROM_BIND_IP']}"
      <transport tls>
        cert_path /etc/collector/metrics/tls.crt
        private_key_path /etc/collector/metrics/tls.key
      </transport>
    </source>

    <source>
      @type prometheus_monitor
      <labels>
        hostname ${hostname}
      </labels>
    </source>

    # excluding prometheus_tail_monitor
    # since it leaks namespace/pod info
    # via file paths

    # tail_monitor plugin which publishes log_collected_bytes_total
    <source>
      @type collected_tail_monitor
      <labels>
        hostname ${hostname}
      </labels>
    </source>

    # This is considered experimental by the repo
    <source>
      @type prometheus_output_monitor
      <labels>
        hostname ${hostname}
      </labels>
    </source>

    # Logs from linux journal
    <source>
      @type systemd
      @id systemd-input
      @label @INGRESS
      path '/var/log/journal'
      <storage>
        @type local
        persistent true
        # NOTE: if this does not end in .json, fluentd will think it
        # is the name of a directory - see fluentd storage_local.rb
        path '/var/lib/fluentd/pos/journal_pos.json'
      </storage>
      matches "#{ENV['JOURNAL_FILTERS_JSON'] || '[]'}"
      tag journal
      read_from_head "#{if (val = ENV.fetch('JOURNAL_READ_FROM_HEAD','')) && (val.length > 0); val; else 'false'; end}"
    </source>

    # Logs from containers (including openshift containers)
    <source>
      @type tail
      @id container-input
      path "/var/log/pods/*/*/*.log"
      exclude_path ["/var/log/pods/openshift-logging_collector-*/*/*.log", "/var/log/pods/openshift-logging_elasticsearch-*/*/*.log", "/var/log/pods/openshift-logging_kibana-*/*/*.log", "/var/log/pods/openshift-logging_*/loki*/*.log", "/var/log/pods/openshift-logging_*/gateway/*.log", "/var/log/pods/openshift-logging_*/opa/*.log", "/var/log/pods/*/*/*.gz", "/var/log/pods/*/*/*.tmp"]
      pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
      follow_inodes true
      refresh_interval 5
      rotate_wait 5
      tag kubernetes.*
      read_from_head "true"
      skip_refresh_on_startup true
      @label @CONCAT
      <parse>
        @type regexp
        expression /^(?<@timestamp>[^\s]+) (?<stream>stdout|stderr) (?<logtag>[F|P]) (?<message>.*)$/
        time_key '@timestamp'
        keep_time_key true
      </parse>
    </source>

    # linux audit logs
    <source>
      @type tail
      @id audit-input
      @label @INGRESS
      path "/var/log/audit/audit.log"
      pos_file "/var/lib/fluentd/pos/audit.log.pos"
      follow_inodes true
      tag linux-audit.log
      <parse>
        @type viaq_host_audit
      </parse>
    </source>

    # k8s audit logs
    <source>
      @type tail
      @id k8s-audit-input
      @label @INGRESS
      path "/var/log/kube-apiserver/audit.log"
      pos_file "/var/lib/fluentd/pos/kube-apiserver.audit.log.pos"
      follow_inodes true
      tag k8s-audit.log
      <parse>
        @type json
        time_key requestReceivedTimestamp
        # In case folks want to parse based on the requestReceivedTimestamp key
        keep_time_key true
        time_format %Y-%m-%dT%H:%M:%S.%N%z
      </parse>
    </source>

    # Openshift audit logs
    <source>
      @type tail
      @id openshift-audit-input
      @label @INGRESS
      path /var/log/oauth-apiserver/audit.log,/var/log/openshift-apiserver/audit.log,/var/log/oauth-server/audit.log
      pos_file /var/lib/fluentd/pos/oauth-apiserver.audit.log
      follow_inodes true
      tag openshift-audit.log
      <parse>
        @type json
        time_key requestReceivedTimestamp
        # In case folks want to parse based on the requestReceivedTimestamp key
        keep_time_key true
        time_format %Y-%m-%dT%H:%M:%S.%N%z
      </parse>
    </source>

    # Openshift Virtual Network (OVN) audit logs
    <source>
      @type tail
      @id ovn-audit-input
      @label @INGRESS
      path "/var/log/ovn/acl-audit-log.log"
      pos_file "/var/lib/fluentd/pos/acl-audit-log.pos"
      follow_inodes true
      tag ovn-audit.log
      refresh_interval 5
      rotate_wait 5
      read_from_head true
      <parse>
        @type none
      </parse>
    </source>

    # Concat log lines of container logs, and send to INGRESS pipeline
    <label @CONCAT>
      <filter kubernetes.**>
        @type concat
        key message
        partial_key logtag
        partial_value P
        separator ''
      </filter>

      <match kubernetes.**>
        @type relabel
        @label @INGRESS
      </match>
    </label>

    # Ingress pipeline
    <label @INGRESS>
      # Filter out PRIORITY from journal logs
      <filter journal>
        @type grep
        <exclude>
          key PRIORITY
          pattern ^7$
        </exclude>
      </filter>

      # Process OVN logs
      <filter ovn-audit.log**>
        @type record_modifier
        <record>
          @timestamp ${DateTime.parse(record['message'].split('|')[0]).rfc3339(6)}
          level ${record['message'].split('|')[3].downcase}
        </record>
      </filter>

      # Process Kube and OpenShift Audit logs
      <filter k8s-audit.log openshift-audit.log>
        @type record_modifier
        <record>
          @timestamp ${record['requestReceivedTimestamp']}
        </record>
      </filter>

      # Retag Journal logs to specific tags
      <match journal>
        @type rewrite_tag_filter
        # skip to @INGRESS label section
        @label @INGRESS

        # see if this is a kibana container for special log handling
        # looks like this:
        # k8s_kibana.a67f366_logging-kibana-1-d90e3_logging_26c51a61-2835-11e6-ad29-fa163e4944d5_f0db49a2
        # we filter these logs through the kibana_transform.conf filter
        <rule>
          key CONTAINER_NAME
          pattern ^k8s_kibana\.
          tag kubernetes.journal.container.kibana
        </rule>

        <rule>
          key CONTAINER_NAME
          pattern ^k8s_[^_]+_logging-eventrouter-[^_]+_
          tag kubernetes.journal.container._default_.kubernetes-event
        </rule>

        # mark logs from default namespace for processing as k8s logs but stored as system logs
        <rule>
          key CONTAINER_NAME
          pattern ^k8s_[^_]+_[^_]+_default_
          tag kubernetes.journal.container._default_
        </rule>

        # mark logs from kube-* namespaces for processing as k8s logs but stored as system logs
        <rule>
          key CONTAINER_NAME
          pattern ^k8s_[^_]+_[^_]+_kube-(.+)_
          tag kubernetes.journal.container._kube-$1_
        </rule>

        # mark logs from openshift-* namespaces for processing as k8s logs but stored as system logs
        <rule>
          key CONTAINER_NAME
          pattern ^k8s_[^_]+_[^_]+_openshift-(.+)_
          tag kubernetes.journal.container._openshift-$1_
        </rule>

        # mark logs from openshift namespace for processing as k8s logs but stored as system logs
        <rule>
          key CONTAINER_NAME
          pattern ^k8s_[^_]+_[^_]+_openshift_
          tag kubernetes.journal.container._openshift_
        </rule>

        # mark fluentd container logs
        <rule>
          key CONTAINER_NAME
          pattern ^k8s_.*fluentd
          tag kubernetes.journal.container.fluentd
        </rule>

        # this is a kubernetes container
        <rule>
          key CONTAINER_NAME
          pattern ^k8s_
          tag kubernetes.journal.container
        </rule>

        # not kubernetes - assume a system log or system container log
        <rule>
          key _TRANSPORT
          pattern .+
          tag journal.system
        </rule>
      </match>

      # Invoke kubernetes apiserver to get kubernetes metadata
      <filter kubernetes.**>
        @id kubernetes-metadata
        @type kubernetes_metadata
        kubernetes_url 'https://kubernetes.default.svc'
        annotation_match ["^containerType\.logging\.openshift\.io\/.*$"]
        allow_orphans false
        cache_size '1000'
        ssl_partial_chain 'true'
      </filter>

      # Parse Json fields for container, journal and eventrouter logs
      <filter kubernetes.var.log.pods.**_eventrouter-**>
        @type parse_json_field
        merge_json_log true
        preserve_json_log true
        json_fields 'message'
      </filter>

      # Fix level field in audit logs
      <filter k8s-audit.log**>
        @type record_modifier
        <record>
          k8s_audit_level ${record['level']}
        </record>
      </filter>

      <filter openshift-audit.log**>
        @type record_modifier
        <record>
          openshift_audit_level ${record['level']}
        </record>
      </filter>

      # Viaq Data Model
      <filter **>
        @type viaq_data_model
        enable_flatten_labels true
        enable_prune_empty_fields false
        default_keep_fields CEE,time,@timestamp,aushape,ci_job,collectd,docker,fedora-ci,file,foreman,geoip,hostname,ipaddr4,ipaddr6,kubernetes,level,message,namespace_name,namespace_uuid,offset,openstack,ovirt,pid,pipeline_metadata,rsyslog,service,systemd,tags,testcase,tlog,viaq_msg_id
        keep_empty_fields 'message'
        rename_time true
        pipeline_type 'collector'
        process_kubernetes_events false
        <level>
          name warn
          match 'Warning|WARN|^W[0-9]+|level=warn|Value:warn|"level":"warn"'
        </level>
        <level>
          name info
          match 'Info|INFO|^I[0-9]+|level=info|Value:info|"level":"info"'
        </level>
        <level>
          name error
          match 'Error|ERROR|^E[0-9]+|level=error|Value:error|"level":"error"'
        </level>
        <level>
          name critical
          match 'Critical|CRITICAL|^C[0-9]+|level=critical|Value:critical|"level":"critical"'
        </level>
        <level>
          name debug
          match 'Debug|DEBUG|^D[0-9]+|level=debug|Value:debug|"level":"debug"'
        </level>
        <formatter>
          tag "journal.system**"
          type sys_journal
          remove_keys log,stream,MESSAGE,_SOURCE_REALTIME_TIMESTAMP,__REALTIME_TIMESTAMP,CONTAINER_ID,CONTAINER_ID_FULL,CONTAINER_NAME,PRIORITY,_BOOT_ID,_CAP_EFFECTIVE,_CMDLINE,_COMM,_EXE,_GID,_HOSTNAME,_MACHINE_ID,_PID,_SELINUX_CONTEXT,_SYSTEMD_CGROUP,_SYSTEMD_SLICE,_SYSTEMD_UNIT,_TRANSPORT,_UID,_AUDIT_LOGINUID,_AUDIT_SESSION,_SYSTEMD_OWNER_UID,_SYSTEMD_SESSION,_SYSTEMD_USER_UNIT,CODE_FILE,CODE_FUNCTION,CODE_LINE,ERRNO,MESSAGE_ID,RESULT,UNIT,_KERNEL_DEVICE,_KERNEL_SUBSYSTEM,_UDEV_SYSNAME,_UDEV_DEVNODE,_UDEV_DEVLINK,SYSLOG_FACILITY,SYSLOG_IDENTIFIER,SYSLOG_PID
        </formatter>
        <formatter>
          tag "kubernetes.var.log.pods.**_eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
          type k8s_json_file
          remove_keys stream
          process_kubernetes_events 'true'
        </formatter>
        <formatter>
          tag "kubernetes.var.log.pods**"
          type k8s_json_file
          remove_keys stream
        </formatter>
      </filter>

      # Generate elasticsearch id
      <filter **>
        @type elasticsearch_genid_ext
        hash_id_key viaq_msg_id
        alt_key kubernetes.event.metadata.uid
        alt_tags 'kubernetes.var.log.pods.**_eventrouter-*.** kubernetes.journal.container._default_.kubernetes-event'
      </filter>

      # Include Infrastructure logs
      <match kubernetes.var.log.pods.openshift_** kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.** system.var.log**>
        @type relabel
        @label @_INFRASTRUCTURE
      </match>

      # Include Application logs
      <match kubernetes.**>
        @type relabel
        @label @_APPLICATION
      </match>

      # Include Audit logs
      <match linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**>
        @type relabel
        @label @_AUDIT
      </match>

      # Send any remaining unmatched tags to stdout
      <match **>
        @type stdout
      </match>
    </label>

    # Routing Application to pipelines
    <label @_APPLICATION>
      <filter **>
        @type record_modifier
        <record>
          log_type application
        </record>
      </filter>

      <match **>
        @type label_router
        <route>
          @label @OL_LOGS
          <match>
            namespaces jbsb-pt, jbsb-sit, jbsb-nft, di-jbsb-sandbox, di-jbsb-sit, di-jbsb-nft, di-sl-dev, sl-pt, sl-sit, sl-nft
          </match>
        </route>

        <route>
          @label @_APPLICATION_ALL
          <match>

          </match>
        </route>
      </match>
    </label>

    # Sending unrouted application to pipelines
    <label @_APPLICATION_ALL>
      <match **>
        @type relabel
        @label @APP_LOGS
      </match>
    </label>

    # Sending infrastructure source type to pipeline
    <label @_INFRASTRUCTURE>
      <filter **>
        @type record_modifier
        <record>
          log_type infrastructure
        </record>
      </filter>

      <match **>
        @type relabel
        @label @INFRA_LOGS
      </match>
    </label>

    # Sending audit source type to pipeline
    <label @_AUDIT>
      <filter **>
        @type record_modifier
        <record>
          log_type audit
        </record>
      </filter>

      <match **>
        @type relabel
        @label @AUDIT_LOGS
      </match>
    </label>

    # Copying pipeline app-logs to outputs
    <label @APP_LOGS>
      <match **>
        @type relabel
        @label @DEFAULT
      </match>
    </label>

    # Copying pipeline audit-logs to outputs
    <label @AUDIT_LOGS>
      <match **>
        @type copy
        copy_mode deep
        <store>
          @type relabel
          @label @CW
        </store>

        <store>
          @type relabel
          @label @DEFAULT
        </store>
      </match>
    </label>

    # Copying pipeline infra-logs to outputs
    <label @INFRA_LOGS>
      <match **>
        @type copy
        copy_mode deep
        <store>
          @type relabel
          @label @CW
        </store>

        <store>
          @type relabel
          @label @DEFAULT
        </store>
      </match>
    </label>

    # Copying pipeline ol-logs to outputs
    <label @OL_LOGS>
      <match **>
        @type relabel
        @label @OL_ELASTICSEARCH
      </match>
    </label>

    # Ship logs to specific outputs
    <label @CW>
      <filter kubernetes.var.log.pods.openshift_** kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** journal.** system.var.log** var.log.pods.openshift-*_** var.log.pods.default_** var.log.pods.kube-*_**>
        @type record_modifier
        char_encoding utf-8:utf-8
        <record>
          cw_group_name managed.infrastructure
          cw_stream_name ${record['hostname']}.${tag}
        </record>
      </filter>

      <filter /^(?!(kubernetes\.|)var\.log\.pods\.openshift-.+_|(kubernetes\.|)var\.log\.pods\.default_|(kubernetes\.|)var\.log\.pods\.kube-.+_|journal\.|system\.var\.log|linux-audit\.log|k8s-audit\.log|openshift-audit\.log|ovn-audit\.log).+/>
        @type record_modifier
        char_encoding utf-8:utf-8
        <record>
          cw_group_name managed.application
          cw_stream_name ${tag}
        </record>
      </filter>

      <filter linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**>
        @type record_modifier
        char_encoding utf-8:utf-8
        <record>
          cw_group_name managed.audit
          cw_stream_name ${record['hostname']}.${tag}
        </record>
      </filter>

      #dedot namespace_labels and rebuild message field if present
      <filter **>
        @type record_modifier
        <record>
          _dummy_ ${if m=record.dig("kubernetes","namespace_labels");record["kubernetes"]["namespace_labels"]={}.tap{|n|m.each{|k,v|n[k.gsub(/[.\/]/,'_')]=v}};end}
          _dummy2_ ${if m=record.dig("kubernetes","labels");record["kubernetes"]["labels"]={}.tap{|n|m.each{|k,v|n[k.gsub(/[.\/]/,'_')]=v}};end}
          _dummy3_ ${if m=record.dig("kubernetes","flat_labels");record["kubernetes"]["flat_labels"]=[].tap{|n|m.each_with_index{|s, i|n[i] = s.gsub(/[.\/]/,'_')}};end}
        </record>
        remove_keys _dummy_, _dummy2_, _dummy3_
      </filter>

      <match **>
        @type cloudwatch_logs
        auto_create_stream true
        region eu-west-2
        log_group_name_key cw_group_name
        log_stream_name_key cw_stream_name
        remove_log_stream_name_key true
        remove_log_group_name_key true
        concurrency 2
        aws_key_id "#{open('/var/run/ocp-collector/secrets/cw-secret/aws_access_key_id','r') do |f|f.read.strip end}"
        aws_sec_key "#{open('/var/run/ocp-collector/secrets/cw-secret/aws_secret_access_key','r') do |f|f.read.strip end}"
        include_time_key true
        log_rejected_request true

        <buffer>
          disable_chunk_backup true
        </buffer>
      </match>
    </label>

    <label @OL_ELASTICSEARCH>
      #dedot namespace_labels
      <filter **>
        @type record_modifier
        <record>
          _dummy_ ${if m=record.dig("kubernetes","namespace_labels");record["kubernetes"]["namespace_labels"]={}.tap{|n|m.each{|k,v|n[k.gsub(/[.\/]/,'_')]=v}};end}
          _dummy2_ ${if m=record.dig("kubernetes","labels");record["kubernetes"]["labels"]={}.tap{|n|m.each{|k,v|n[k.gsub(/[.\/]/,'_')]=v}};end}
          _dummy3_ ${if m=record.dig("kubernetes","flat_labels");record["kubernetes"]["flat_labels"]=[].tap{|n|m.each_with_index{|s, i|n[i] = s.gsub(/[.\/]/,'_')}};end}
        </record>
        remove_keys _dummy_, _dummy2_, _dummy3_
      </filter>

      # Viaq Data Model
      <filter **>
        @type viaq_data_model
        enable_openshift_model false
        enable_prune_empty_fields false
        rename_time false
        undefined_dot_replace_char UNUSED
        elasticsearch_index_prefix_field 'viaq_index_name'
        <elasticsearch_index_name>
          enabled 'true'
          tag "kubernetes.var.log.pods.openshift_** kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** var.log.pods.openshift_** var.log.pods.openshift-*_** var.log.pods.default_** var.log.pods.kube-*_** journal.system** system.var.log**"
          name_type static
          static_index_name infra-write
        </elasticsearch_index_name>
        <elasticsearch_index_name>
          enabled 'true'
          tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
          name_type static
          static_index_name audit-write
        </elasticsearch_index_name>
        <elasticsearch_index_name>
          enabled 'true'
          tag "**"
          name_type structured
          static_index_name app-write
        </elasticsearch_index_name>
      </filter>
      <filter **>
        @type viaq_data_model
        enable_prune_labels true
        enable_openshift_model false
        rename_time false
        undefined_dot_replace_char UNUSED
        prune_labels_exclusions app_kubernetes_io_name,app_kubernetes_io_instance,app_kubernetes_io_version,app_kubernetes_io_component,app_kubernetes_io_part-of,app_kubernetes_io_managed-by,app_kubernetes_io_created-by
      </filter>

      #rebuild message field if present
      <filter **>
        @type record_modifier
        <record>
          _dummy_ ${(require 'json';record['message']=JSON.dump(record['structured'])) if record['structured'] and record['viaq_index_name'] == 'app-write'}
        </record>
        remove_keys _dummy_
      </filter>

      #remove structured field if present
      <filter **>
        @type record_modifier
        char_encoding utf-8:utf-8
        remove_keys structured
      </filter>

      <match retry_ol_elasticsearch>
        @type elasticsearch
        @id retry_ol_elasticsearch
        host jbsb-es-es-http.jbsb-monitoring.svc.cluster.local
        port 9200
        scheme http
        user "#{File.exists?('/var/run/ocp-collector/secrets/ol-clf-user/username') ? open('/var/run/ocp-collector/secrets/ol-clf-user/username','r') do |f|f.read end : ''}"
        password "#{File.exists?('/var/run/ocp-collector/secrets/ol-clf-user/password') ? open('/var/run/ocp-collector/secrets/ol-clf-user/password','r') do |f|f.read end : ''}"
        target_index_key viaq_index_name
        id_key viaq_msg_id
        remove_keys viaq_index_name
        verify_es_version_at_startup false

        type_name _doc
        http_backend typhoeus
        write_operation create
        reload_connections 'true'
        # https://github.com/uken/fluent-plugin-elasticsearch#reload-after
        reload_after '200'
        # https://github.com/uken/fluent-plugin-elasticsearch#sniffer-class-name
        sniffer_class_name 'Fluent::Plugin::ElasticsearchSimpleSniffer'
        reload_on_failure false
        # 2 ^ 31
        request_timeout 2147483648
        <buffer>
          @type file
          path '/var/lib/fluentd/retry_ol_elasticsearch'
          flush_mode interval
          flush_interval 1s
          flush_thread_count 2
          retry_type exponential_backoff
          retry_wait 1s
          retry_max_interval 60s
          retry_timeout 60m
          queued_chunks_limit_size "#{ENV['BUFFER_QUEUE_LIMIT'] || '32'}"
          total_limit_size "#{ENV['TOTAL_LIMIT_SIZE_PER_BUFFER'] || '8589934592'}"
          chunk_limit_size "#{ENV['BUFFER_SIZE_LIMIT'] || '8m'}"
          overflow_action block
          disable_chunk_backup true
        </buffer>
      </match>

      <match **>
        @type elasticsearch
        @id ol_elasticsearch
        host jbsb-es-es-http.jbsb-monitoring.svc.cluster.local
        port 9200
        scheme http
        user "#{File.exists?('/var/run/ocp-collector/secrets/ol-clf-user/username') ? open('/var/run/ocp-collector/secrets/ol-clf-user/username','r') do |f|f.read end : ''}"
        password "#{File.exists?('/var/run/ocp-collector/secrets/ol-clf-user/password') ? open('/var/run/ocp-collector/secrets/ol-clf-user/password','r') do |f|f.read end : ''}"
        target_index_key viaq_index_name
        id_key viaq_msg_id
        remove_keys viaq_index_name
        verify_es_version_at_startup false

        type_name _doc
        retry_tag retry_ol_elasticsearch
        http_backend typhoeus
        write_operation create
        reload_connections 'true'
        # https://github.com/uken/fluent-plugin-elasticsearch#reload-after
        reload_after '200'
        # https://github.com/uken/fluent-plugin-elasticsearch#sniffer-class-name
        sniffer_class_name 'Fluent::Plugin::ElasticsearchSimpleSniffer'
        reload_on_failure false
        # 2 ^ 31
        request_timeout 2147483648
        <buffer>
          @type file
          path '/var/lib/fluentd/ol_elasticsearch'
          flush_mode interval
          flush_interval 1s
          flush_thread_count 2
          retry_type exponential_backoff
          retry_wait 1s
          retry_max_interval 60s
          retry_timeout 60m
          queued_chunks_limit_size "#{ENV['BUFFER_QUEUE_LIMIT'] || '32'}"
          total_limit_size "#{ENV['TOTAL_LIMIT_SIZE_PER_BUFFER'] || '8589934592'}"
          chunk_limit_size "#{ENV['BUFFER_SIZE_LIMIT'] || '8m'}"
          overflow_action block
          disable_chunk_backup true
        </buffer>
      </match>
    </label>

    <label @DEFAULT>
      #dedot namespace_labels
      <filter **>
        @type record_modifier
        <record>
          _dummy_ ${if m=record.dig("kubernetes","namespace_labels");record["kubernetes"]["namespace_labels"]={}.tap{|n|m.each{|k,v|n[k.gsub(/[.\/]/,'_')]=v}};end}
          _dummy2_ ${if m=record.dig("kubernetes","labels");record["kubernetes"]["labels"]={}.tap{|n|m.each{|k,v|n[k.gsub(/[.\/]/,'_')]=v}};end}
          _dummy3_ ${if m=record.dig("kubernetes","flat_labels");record["kubernetes"]["flat_labels"]=[].tap{|n|m.each_with_index{|s, i|n[i] = s.gsub(/[.\/]/,'_')}};end}
        </record>
        remove_keys _dummy_, _dummy2_, _dummy3_
      </filter>

      # Viaq Data Model
      <filter **>
        @type viaq_data_model
        enable_openshift_model false
        enable_prune_empty_fields false
        rename_time false
        undefined_dot_replace_char UNUSED
        elasticsearch_index_prefix_field 'viaq_index_name'
        <elasticsearch_index_name>
          enabled 'true'
          tag "kubernetes.var.log.pods.openshift_** kubernetes.var.log.pods.openshift-*_** kubernetes.var.log.pods.default_** kubernetes.var.log.pods.kube-*_** var.log.pods.openshift_** var.log.pods.openshift-*_** var.log.pods.default_** var.log.pods.kube-*_** journal.system** system.var.log**"
          name_type static
          static_index_name infra-write
        </elasticsearch_index_name>
        <elasticsearch_index_name>
          enabled 'true'
          tag "linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
          name_type static
          static_index_name audit-write
        </elasticsearch_index_name>
        <elasticsearch_index_name>
          enabled 'true'
          tag "**"
          name_type structured
          static_index_name app-write
        </elasticsearch_index_name>
      </filter>
      <filter **>
        @type viaq_data_model
        enable_prune_labels true
        enable_openshift_model false
        rename_time false
        undefined_dot_replace_char UNUSED
        prune_labels_exclusions app_kubernetes_io_name,app_kubernetes_io_instance,app_kubernetes_io_version,app_kubernetes_io_component,app_kubernetes_io_part-of,app_kubernetes_io_managed-by,app_kubernetes_io_created-by
      </filter>

      #rebuild message field if present
      <filter **>
        @type record_modifier
        <record>
          _dummy_ ${(require 'json';record['message']=JSON.dump(record['structured'])) if record['structured'] and record['viaq_index_name'] == 'app-write'}
        </record>
        remove_keys _dummy_
      </filter>

      #remove structured field if present
      <filter **>
        @type record_modifier
        char_encoding utf-8:utf-8
        remove_keys structured
      </filter>

      <match retry_default>
        @type elasticsearch
        @id retry_default
        host elasticsearch
        port 9200
        scheme https
        ssl_version TLSv1_2
        client_key '/var/run/ocp-collector/secrets/collector/tls.key'
        client_cert '/var/run/ocp-collector/secrets/collector/tls.crt'
        ca_file '/var/run/ocp-collector/secrets/collector/ca-bundle.crt'
        target_index_key viaq_index_name
        id_key viaq_msg_id
        remove_keys viaq_index_name
        verify_es_version_at_startup false

        type_name _doc
        http_backend typhoeus
        write_operation create
        reload_connections 'true'
        # https://github.com/uken/fluent-plugin-elasticsearch#reload-after
        reload_after '200'
        # https://github.com/uken/fluent-plugin-elasticsearch#sniffer-class-name
        sniffer_class_name 'Fluent::Plugin::ElasticsearchSimpleSniffer'
        reload_on_failure false
        # 2 ^ 31
        request_timeout 2147483648
        <buffer>
          @type file
          path '/var/lib/fluentd/retry_default'
          flush_mode interval
          flush_interval 1s
          flush_thread_count 2
          retry_type exponential_backoff
          retry_wait 1s
          retry_max_interval 60s
          retry_timeout 60m
          queued_chunks_limit_size "#{ENV['BUFFER_QUEUE_LIMIT'] || '32'}"
          total_limit_size "#{ENV['TOTAL_LIMIT_SIZE_PER_BUFFER'] || '8589934592'}"
          chunk_limit_size "#{ENV['BUFFER_SIZE_LIMIT'] || '8m'}"
          overflow_action block
          disable_chunk_backup true
        </buffer>
      </match>

      <match **>
        @type elasticsearch
        @id default
        host elasticsearch
        port 9200
        scheme https
        ssl_version TLSv1_2
        client_key '/var/run/ocp-collector/secrets/collector/tls.key'
        client_cert '/var/run/ocp-collector/secrets/collector/tls.crt'
        ca_file '/var/run/ocp-collector/secrets/collector/ca-bundle.crt'
        target_index_key viaq_index_name
        id_key viaq_msg_id
        remove_keys viaq_index_name
        verify_es_version_at_startup false

        type_name _doc
        retry_tag retry_default
        http_backend typhoeus
        write_operation create
        reload_connections 'true'
        # https://github.com/uken/fluent-plugin-elasticsearch#reload-after
        reload_after '200'
        # https://github.com/uken/fluent-plugin-elasticsearch#sniffer-class-name
        sniffer_class_name 'Fluent::Plugin::ElasticsearchSimpleSniffer'
        reload_on_failure false
        # 2 ^ 31
        request_timeout 2147483648
        <buffer>
          @type file
          path '/var/lib/fluentd/default'
          flush_mode interval
          flush_interval 1s
          flush_thread_count 2
          retry_type exponential_backoff
          retry_wait 1s
          retry_max_interval 60s
          retry_timeout 60m
          queued_chunks_limit_size "#{ENV['BUFFER_QUEUE_LIMIT'] || '32'}"
          total_limit_size "#{ENV['TOTAL_LIMIT_SIZE_PER_BUFFER'] || '8589934592'}"
          chunk_limit_size "#{ENV['BUFFER_SIZE_LIMIT'] || '8m'}"
          overflow_action block
          disable_chunk_backup true
        </buffer>
      </match>
    </label>
  run.sh: "\n#!/bin/bash\n\nCFG_DIR=/etc/fluent/configs.d\n\nfluentdargs=\"--umask
    0077 --no-supervisor\"\n# find the sniffer class file\nsniffer=$( gem contents
    fluent-plugin-elasticsearch|grep elasticsearch_simple_sniffer.rb )\nif [ -z \"$sniffer\"
    ] ; then\n    sniffer=$( rpm -ql rubygem-fluent-plugin-elasticsearch|grep elasticsearch_simple_sniffer.rb
    )\nfi\nif [ -n \"$sniffer\" -a -f \"$sniffer\" ] ; then\n    fluentdargs=\"$fluentdargs
    -r $sniffer\"\nfi\n\nset -e\nfluentdargs=\"--suppress-config-dump $fluentdargs\"\n\n\nissue_deprecation_warnings()
    {\n    : # none at the moment\n}\n\nIPADDR4=${NODE_IPV4:-$( /usr/sbin/ip -4 addr
    show dev eth0 | grep inet | sed -e \"s/[ \\t]*inet \\([0-9.]*\\).*/\\1/\" )}\nIPADDR6=${NODE_IPV6:-$(
    /usr/sbin/ip -6 addr show dev eth0 | grep inet | sed -e \"s/[ \\t]*inet6 \\([a-z0-9::]*\\).*/\\1/\"
    | grep -v ^fe80 | grep -v ^::1 || echo \"\")}\n\nexport IPADDR4 IPADDR6\n\nexport
    PROM_BIND_IP=$(\nruby <<EOF\n  require 'ipaddr'\n  def get_ip (ipstrs)\n\tisIPV4
    = false\n\tisIPV6 = false\n\tfor ipstr in ipstrs do\n\t  begin\n\t\tipaddr = IPAddr.new
    ipstr.strip\n\t  rescue => e\n\t\treturn \"#{ipstr} is invalid ip. exception #{e.class}
    occurred with message #{e.message}\"\n\t  else\n\t\tisIPV4 |= ipaddr.ipv4?\n\t\tisIPV6
    |= ipaddr.ipv6?\n\t  end\n\tend\n\treturn \"[::]\" if isIPV6\n\treturn \"0.0.0.0\"
    if isIPV4\n\treturn \"invalid-ip\"\n  end\n  puts get_ip(\"${POD_IPS}\".split(\",\"))\nEOF\n)\n\necho
    \"POD_IPS: ${POD_IPS}, PROM_BIND_IP: ${PROM_BIND_IP}\"\n\n\n\n# Check bearer_token_file
    for fluent-plugin-kubernetes_metadata_filter.\nif [ ! -s /var/run/secrets/kubernetes.io/serviceaccount/token
    ] ; then\n    echo \"ERROR: Bearer_token_file (/var/run/secrets/kubernetes.io/serviceaccount/token)
    to access the Kubernetes API server is missing or empty.\"\n    exit 1\nfi\n\n#
    If FILE_BUFFER_PATH exists and it is not a directory, mkdir fails with the error.\nFILE_BUFFER_PATH=/var/lib/fluentd\nmkdir
    -p $FILE_BUFFER_PATH\nFLUENT_CONF=$CFG_DIR/user/fluent.conf\nif [ ! -f \"$FLUENT_CONF\"
    ] ; then\n    echo \"ERROR: The configuration $FLUENT_CONF does not exist\"\n
    \   exit 1\nfi\n\n###\n# Calculate the max allowed for each output buffer given
    the number of\n# buffer file paths\n###\n\nNUM_OUTPUTS=$(grep \"path.*'$FILE_BUFFER_PATH\"
    $FLUENT_CONF | wc -l)\nif [ $NUM_OUTPUTS -eq 0 ]; then\n    # Reset to default
    single output if log forwarding outputs all invalid\n    NUM_OUTPUTS=1\nfi\n\n#
    Get the available disk size.\nDF_LIMIT=$(df -B1 $FILE_BUFFER_PATH | grep -v Filesystem
    | awk '{print $2}')\nDF_LIMIT=${DF_LIMIT:-0}\nif [ $DF_LIMIT -eq 0 ]; then\n    echo
    \"ERROR: No disk space is available for file buffer in $FILE_BUFFER_PATH.\"\n
    \   exit 1\nfi\n\n# Default to 15% of disk which is approximately 18G\nALLOWED_PERCENT_OF_DISK=${ALLOWED_PERCENT_OF_DISK:-15}\nif
    [ $ALLOWED_PERCENT_OF_DISK -gt 100 ] || [ $ALLOWED_PERCENT_OF_DISK -le 0 ] ; then\n
    \ ALLOWED_PERCENT_OF_DISK=15\n  echo ALLOWED_PERCENT_OF_DISK is out of the allowed
    range. Setting to ${ALLOWED_PERCENT_OF_DISK}%\nfi\n# Determine allowed total given
    the number of outputs we have.\nALLOWED_DF_LIMIT=$(expr $DF_LIMIT \\* $ALLOWED_PERCENT_OF_DISK
    / 100) || :\n\n# total limit size allowed  per buffer\nTOTAL_LIMIT_SIZE_ALLOWED_PER_BUFFER=$(expr
    $ALLOWED_DF_LIMIT / $NUM_OUTPUTS) || :\n\nTOTAL_LIMIT_SIZE_ALLOWED_PER_BUFFER=${TOTAL_LIMIT_SIZE_ALLOWED_PER_BUFFER:-0}\nTOTAL_LIMIT_SIZE_PER_BUFFER=${TOTAL_LIMIT_SIZE_PER_BUFFER:-0}\nTOTAL_LIMIT_SIZE_PER_BUFFER=$(echo
    $TOTAL_LIMIT_SIZE_PER_BUFFER |  sed -e \"s/[Kk]/*1024/g;s/[Mm]/*1024*1024/g;s/[Gg]/*1024*1024*1024/g;s/i//g\"
    | bc) || :\nif [[ $TOTAL_LIMIT_SIZE_PER_BUFFER -lt $TOTAL_LIMIT_SIZE_ALLOWED_PER_BUFFER
    ]];\nthen\n   if [[ $TOTAL_LIMIT_SIZE_PER_BUFFER -eq 0 ]]; then\n       TOTAL_LIMIT_SIZE_PER_BUFFER=$TOTAL_LIMIT_SIZE_ALLOWED_PER_BUFFER\n
    \  fi\nelse\n    echo \"Requested buffer size per output $TOTAL_LIMIT_SIZE_PER_BUFFER
    for $NUM_OUTPUTS buffers exceeds maximum available size  $TOTAL_LIMIT_SIZE_ALLOWED_PER_BUFFER
    bytes per output\"\n    TOTAL_LIMIT_SIZE_PER_BUFFER=$TOTAL_LIMIT_SIZE_ALLOWED_PER_BUFFER\nfi\necho
    \"Setting each total_size_limit for $NUM_OUTPUTS buffers to $TOTAL_LIMIT_SIZE_PER_BUFFER
    bytes\"\nexport TOTAL_LIMIT_SIZE_PER_BUFFER\n\n##\n# Calculate the max number
    of queued chunks given the size of each chunk\n# and the max allowed space per
    buffer\n##\nBUFFER_SIZE_LIMIT=$(echo ${BUFFER_SIZE_LIMIT:-8388608})\nBUFFER_QUEUE_LIMIT=$(expr
    $TOTAL_LIMIT_SIZE_PER_BUFFER / $BUFFER_SIZE_LIMIT)\necho \"Setting queued_chunks_limit_size
    for each buffer to $BUFFER_QUEUE_LIMIT\"\nexport BUFFER_QUEUE_LIMIT\necho \"Setting
    chunk_limit_size for each buffer to $BUFFER_SIZE_LIMIT\"\nexport BUFFER_SIZE_LIMIT\n\nissue_deprecation_warnings\n\n#
    this should be the last thing before launching fluentd so as not to use\n# jemalloc
    with any other processes\nif type -p jemalloc-config > /dev/null 2>&1 ; then\n
    \   export LD_PRELOAD=$( jemalloc-config --libdir )/libjemalloc.so.$( jemalloc-config
    --revision )\n    export LD_BIND_NOW=1 # workaround for https://bugzilla.redhat.com/show_bug.cgi?id=1544815\nfi\n\n#
    In case of an update to secure fluentd container, copy the fluentd pos files to
    their new\n# locations under /var/lib/fluentd/pos. Moving old pos files is not
    possible since /var/log\n# is mounted read-only in the secure fluentd container.\n#\nPOS_FILES_DIR=${FILE_BUFFER_PATH}/pos\nmkdir
    -p $POS_FILES_DIR\nif [ -f /var/log/openshift-apiserver/audit.log.pos -a ! -f
    ${POS_FILES_DIR}/oauth-apiserver.audit.log ] ; then\n    cp /var/log/openshift-apiserver/audit.log.pos
    ${POS_FILES_DIR}/oauth-apiserver.audit.log\nfi\ndeclare -A POS_FILES_FROM_TO=(
    [/var/log/audit/audit.log.pos]=${POS_FILES_DIR}/audit.log.pos [/var/log/kube-apiserver/audit.log.pos]=${POS_FILES_DIR}/kube-apiserver.audit.log.pos
    )\nfor POS_FILE in es-containers.log.pos journal_pos.json oauth-apiserver.audit.log\ndo\n
    \ POS_FILES_FROM_TO[\"/var/log/$pos_file\"]=\"${POS_FILES_DIR}/$pos_file\"\ndone\nfor
    FROM in \"${!POS_FILES_FROM_TO[@]}\"\ndo\n    TO=${POS_FILES_FROM_TO[$FROM]}\n
    \   if [ -f \"$FROM\" -a ! -f \"$TO\" ] ; then\n      cp \"$FROM\" \"$TO\"\n    fi\ndone\n\nFILE=\"/var/lib/fluentd/pos/journal_pos.json\"\n\nif
    test -f \"$FILE\"; then\n    echo \"$FILE exists, checking if yajl parser able
    to parse this json file without any error.\"\n\n    ruby -v /etc/fluent/configs.d/user/cleanInValidJson.rb
    \ $FILE || \n    if [ $? -ne 0 ]; then\n      echo \"$FILE contains invalid json
    content so removing it as leads to crashloopbackoff error in fluentd pod\"\n      rm
    $FILE\n    fi\nfi\n\nexec fluentd $fluentdargs\n\n"
kind: ConfigMap
metadata:
  creationTimestamp: "2022-10-11T10:29:31Z"
  name: collector
  namespace: openshift-logging
  ownerReferences:
  - apiVersion: logging.openshift.io/v1
    controller: true
    kind: ClusterLogging
    name: instance
    uid: 33d9098a-1d5a-45fc-8410-90ee8e362aee
  resourceVersion: "589319617"
  uid: 525c6182-6abe-4e03-b345-ab4bc5839881
