kind: ConfigMap
apiVersion: v1
metadata:
  name: fluentd
  namespace: openshift-logging
data:
  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>


    # In each section below, pre- and post- includes don't include anything
    initially;

    # they exist to enable future additions to openshift conf as needed.


    ## sources

    ## ordered so that syslog always runs last...

    <source>
      @type prometheus
      bind "#{ENV['POD_IP']}"
      <ssl>
        enable true
        certificate_path "#{ENV['METRICS_CERT'] || '/etc/fluent/metrics/tls.crt'}"
        private_key_path "#{ENV['METRICS_KEY'] || '/etc/fluent/metrics/tls.key'}"
      </ssl>
    </source>


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


    # excluding prometheus_tail_monitor

    # since it leaks namespace/pod info

    # via file paths

    # including new 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>

    #journal logs to gather node

    <source>
      @type systemd
      @id systemd-input
      @label @MEASURE
      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>

    # container logs

    <source>
      @type tail
      @id container-input
      path "/var/log/containers/*robot-shop*.log"
      exclude_path ["/var/log/containers/fluentd-*_openshift-logging_*.log", "/var/log/containers/elasticsearch-*_openshift-logging_*.log", "/var/log/containers/kibana-*_openshift-logging_*.log"]
      pos_file "/var/lib/fluentd/pos/es-containers.log.pos"
      refresh_interval 5
      rotate_wait 5
      tag kubernetes.*
      read_from_head "true"
      skip_refresh_on_startup true
      @label @MEASURE
      <parse>
        @type multi_format
        <pattern>
          format json
          time_format '%Y-%m-%dT%H:%M:%S.%N%Z'
          keep_time_key true
        </pattern>
        <pattern>
          format regexp
          expression /^(?<time>[^\s]+) (?<stream>stdout|stderr)( (?<logtag>.))? (?<log>.*)$/
          time_format '%Y-%m-%dT%H:%M:%S.%N%:z'
          keep_time_key true
        </pattern>
      </parse>
    </source>


    <label @MEASURE>
      <filter **>
        @type record_transformer
        enable_ruby
        <record>
          msg_size ${record.to_s.length}
        </record>
      </filter>
      <filter **>
        @type prometheus
        <metric>
          name cluster_logging_collector_input_record_total
          type counter
          desc The total number of incoming records
          <labels>
            tag ${tag}
            hostname ${hostname}
          </labels>
        </metric>
      </filter>
      <filter **>
        @type prometheus
        <metric>
          name cluster_logging_collector_input_record_bytes
          type counter
          desc The total bytes of incoming records
          key msg_size
          <labels>
            tag ${tag}
            hostname ${hostname}
          </labels>
        </metric>
      </filter>
      <filter **>
        @type record_transformer
        remove_keys msg_size
      </filter>
      <match journal>
        @type relabel
        @label @INGRESS
      </match>
      <match *audit.log>
        @type relabel
        @label @INGRESS
       </match>
      <match kubernetes.**>
        @type relabel
        @label @CONCAT
      </match>
    </label>


    <label @CONCAT>
      <filter kubernetes.**>
        @type concat
        key log
        partial_key logtag
        partial_value P
        separator ''
      </filter>
      <match kubernetes.**>
        @type relabel
        @label @INGRESS
      </match>
    </label>


    #syslog input config here


    <label @INGRESS>

      ## filters
      <filter journal>
       @type grep
       <exclude>
          key PRIORITY
          pattern ^7$
       </exclude>
      </filter>

      <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>

      <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>

      <filter kubernetes.**>
        @type kubernetes_metadata
        kubernetes_url 'https://kubernetes.default.svc'
        cache_size '1000'
        watch 'false'
        use_journal 'nil'
        ssl_partial_chain 'true'
      </filter>

      <filter kubernetes.journal.**>
        @type parse_json_field
        merge_json_log 'false'
        preserve_json_log 'true'
        json_fields 'log,MESSAGE'
      </filter>

      <filter kubernetes.var.log.containers.**>
        @type parse_json_field
        merge_json_log 'false'
        preserve_json_log 'true'
        json_fields 'log,MESSAGE'
      </filter>

      <filter kubernetes.var.log.containers.eventrouter-** kubernetes.var.log.containers.cluster-logging-eventrouter-**>
        @type parse_json_field
        merge_json_log true
        preserve_json_log true
        json_fields 'log,MESSAGE'
      </filter>

      <filter **kibana**>
        @type record_transformer
        enable_ruby
        <record>
          log ${record['err'] || record['msg'] || record['MESSAGE'] || record['log']}
        </record>
        remove_keys req,res,msg,name,level,v,pid,err
      </filter>

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

      <filter **>
        @type viaq_data_model
        elasticsearch_index_prefix_field 'viaq_index_name'
        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
        extra_keep_fields ''
        keep_empty_fields 'message'
        use_undefined false
        undefined_name 'undefined'
        rename_time true
        rename_time_if_missing false
        src_time_name 'time'
        dest_time_name '@timestamp'
        pipeline_type 'collector'
        undefined_to_string 'false'
        undefined_dot_replace_char 'UNUSED'
        undefined_max_num_fields '-1'
        process_kubernetes_events 'false'
        <formatter>
          tag "system.var.log**"
          type sys_var_log
          remove_keys host,pid,ident
        </formatter>
        <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.journal.container**"
          type k8s_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.containers.eventrouter-** kubernetes.var.log.containers.cluster-logging-eventrouter-** k8s-audit.log** openshift-audit.log** ovn-audit.log**"
          type k8s_json_file
          remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
          process_kubernetes_events 'true'
        </formatter>
        <formatter>
          tag "kubernetes.var.log.containers**"
          type k8s_json_file
          remove_keys log,stream,CONTAINER_ID_FULL,CONTAINER_NAME
        </formatter>
        <elasticsearch_index_name>
          enabled 'true'
          tag "journal.system** system.var.log** **_default_** **_kube-*_** **_openshift-*_** **_openshift_**"
          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 static
          static_index_name app-write
        </elasticsearch_index_name>
      </filter>

      <filter **>
        @type elasticsearch_genid_ext
        hash_id_key viaq_msg_id
        alt_key kubernetes.event.metadata.uid
        alt_tags 'kubernetes.var.log.containers.logging-eventrouter-*.** kubernetes.var.log.containers.eventrouter-*.** kubernetes.var.log.containers.cluster-logging-eventrouter-*.** kubernetes.journal.container._default_.kubernetes-event'
      </filter>

      # Relabel specific source tags to specific intermediary labels for copy processing
      # Earlier matchers remove logs so they don't fall through to later ones.
      # A log source matcher may be null if no pipeline wants that type of log.
      <match **_default_** **_kube-*_** **_openshift-*_** **_openshift_** journal.** system.var.log**>
        @type relabel
        @label @_INFRASTRUCTURE
      </match>
      <match kubernetes.**>
        @type relabel
        @label @_APPLICATION
      </match>
      <match linux-audit.log** k8s-audit.log** openshift-audit.log** ovn-audit.log**>
        @type null
      </match>

      <match **>
        @type stdout
      </match>

    </label>


    # Relabel specific sources (e.g. logs.apps) to multiple pipelines

    <label @_APPLICATION>
      <filter **>
        @type record_modifier
        <record>
          log_type application
        </record>
      </filter>
      <match **>
        @type copy
        <store>
          @type relabel
          @label @PIPELINE_0_
        </store>
      </match>
    </label>

    <label @_INFRASTRUCTURE>
      <filter **>
        @type record_modifier
        <record>
          log_type infrastructure
        </record>
      </filter>
      <match **>
        @type copy
        <store>
          @type relabel
          @label @PIPELINE_0_
        </store>
      </match>
    </label>



    # Relabel specific pipelines to multiple, outputs (e.g. ES, kafka stores)

    <label @PIPELINE_0_>
      <match **>
        @type copy
        <store>
          @type relabel
          @label @DEFAULT
        </store>
      </match>
    </label>


    # Ship logs to specific outputs

    <label @DEFAULT>
      <filter **>
        @type record_modifier
      remove_keys structured
      </filter>
      #flatten labels to prevent field explosion in ES
      <filter ** >
        @type record_transformer
        enable_ruby true
        <record>
          kubernetes ${!record['kubernetes'].nil? ? record['kubernetes'].merge({"flat_labels": (record['kubernetes']['labels']||{}).map{|k,v| "#{k}=#{v}"}}) : {} }
        </record>
        remove_keys $.kubernetes.labels
      </filter>
      <match retry_default>
        @type copy
        <store>
          @type elasticsearch
          @id retry_default
          host elasticsearch.openshift-logging.svc
          port 9200
          verify_es_version_at_startup false
          scheme https
          ssl_version TLSv1_2
          client_key '/var/run/ocp-collector/secrets/fluentd/tls.key'
          client_cert '/var/run/ocp-collector/secrets/fluentd/tls.crt'
          ca_file '/var/run/ocp-collector/secrets/fluentd/ca-bundle.crt'
          target_index_key viaq_index_name
          id_key viaq_msg_id
          remove_keys viaq_index_name
          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
            flush_at_shutdown true
            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'] ||  8589934592 }" #8G
            chunk_limit_size "#{ENV['BUFFER_SIZE_LIMIT'] || '8m'}"
            overflow_action block
          </buffer>
        </store>
      </match>
      <match **>
        @type copy
        <store>
          @type elasticsearch
          @id default
          host elasticsearch.openshift-logging.svc
          port 9200
          verify_es_version_at_startup false
          scheme https
          ssl_version TLSv1_2
          client_key '/var/run/ocp-collector/secrets/fluentd/tls.key'
          client_cert '/var/run/ocp-collector/secrets/fluentd/tls.crt'
          ca_file '/var/run/ocp-collector/secrets/fluentd/ca-bundle.crt'
          target_index_key viaq_index_name
          id_key viaq_msg_id
          remove_keys viaq_index_name
          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
            flush_at_shutdown true
            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'] ||  8589934592 }" #8G
            chunk_limit_size "#{ENV['BUFFER_SIZE_LIMIT'] || '8m'}"
            overflow_action block
          </buffer>
        </store>
      </match>
    </label>

  run.sh: >+

    #!/bin/bash


    CFG_DIR=/etc/fluent/configs.d


    fluentdargs="--no-supervisor"

    # find the sniffer class file

    sniffer=$( gem contents fluent-plugin-elasticsearch|grep
    elasticsearch_simple_sniffer.rb )

    if [ -z "$sniffer" ] ; then
        sniffer=$( rpm -ql rubygem-fluent-plugin-elasticsearch|grep elasticsearch_simple_sniffer.rb )
    fi

    if [ -n "$sniffer" -a -f "$sniffer" ] ; then
        fluentdargs="$fluentdargs -r $sniffer"
    fi


    set -e

    fluentdargs="--suppress-config-dump $fluentdargs"



    issue_deprecation_warnings() {
        : # none at the moment
    }


    IPADDR4=${NODE_IPV4:-$( /usr/sbin/ip -4 addr show dev eth0 | grep inet | sed
    -e "s/[ \t]*inet \([0-9.]*\).*/\1/" )}

    IPADDR6=${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 "")}


    export IPADDR4 IPADDR6


    # Check bearer_token_file for fluent-plugin-kubernetes_metadata_filter.

    if [ ! -s /var/run/secrets/kubernetes.io/serviceaccount/token ] ; then
        echo "ERROR: Bearer_token_file (/var/run/secrets/kubernetes.io/serviceaccount/token) to access the Kubernetes API server is missing or empty."
        exit 1
    fi


    # If FILE_BUFFER_PATH exists and it is not a directory, mkdir fails with the
    error.

    FILE_BUFFER_PATH=/var/lib/fluentd

    mkdir -p $FILE_BUFFER_PATH

    FLUENT_CONF=$CFG_DIR/user/fluent.conf

    if [ ! -f "$FLUENT_CONF" ] ; then
        echo "ERROR: The configuration $FLUENT_CONF does not exist"
        exit 1
    fi


    ###

    # Calculate the max allowed for each output buffer given the number of

    # buffer file paths

    ###


    NUM_OUTPUTS=$(grep "path.*'$FILE_BUFFER_PATH" $FLUENT_CONF | wc -l)

    if [ $NUM_OUTPUTS -eq 0 ]; then
        # Reset to default single output if log forwarding outputs all invalid
        NUM_OUTPUTS=1
    fi


    # Get the available disk size.

    DF_LIMIT=$(df -B1 $FILE_BUFFER_PATH | grep -v Filesystem | awk '{print $2}')

    DF_LIMIT=${DF_LIMIT:-0}

    if [ $DF_LIMIT -eq 0 ]; then
        echo "ERROR: No disk space is available for file buffer in $FILE_BUFFER_PATH."
        exit 1
    fi


    # Default to 15% of disk which is approximately 18G

    ALLOWED_PERCENT_OF_DISK=${ALLOWED_PERCENT_OF_DISK:-15}

    if [ $ALLOWED_PERCENT_OF_DISK -gt 100 ] || [ $ALLOWED_PERCENT_OF_DISK -le 0
    ] ; then
      ALLOWED_PERCENT_OF_DISK=15
      echo ALLOWED_PERCENT_OF_DISK is out of the allowed range. Setting to ${ALLOWED_PERCENT_OF_DISK}%
    fi

    # Determine allowed total given the number of outputs we have.

    ALLOWED_DF_LIMIT=$(expr $DF_LIMIT \* $ALLOWED_PERCENT_OF_DISK / 100) || :


    # TOTAL_LIMIT_SIZE per buffer

    TOTAL_LIMIT_SIZE=$(echo ${TOTAL_LIMIT_SIZE:-0})

    if [ $TOTAL_LIMIT_SIZE -eq 0 ]; then
        TOTAL_LIMIT_SIZE=$(expr $ALLOWED_DF_LIMIT / $NUM_OUTPUTS) || :
    fi

    echo "Setting each total_size_limit for $NUM_OUTPUTS buffers to
    $TOTAL_LIMIT_SIZE bytes"

    export TOTAL_LIMIT_SIZE


    ##

    # Calculate the max number of queued chunks given the size of each chunk

    # and the max allowed space per buffer

    ##

    BUFFER_SIZE_LIMIT=$(echo ${BUFFER_SIZE_LIMIT:-8388608})

    BUFFER_QUEUE_LIMIT=$(expr $TOTAL_LIMIT_SIZE / $BUFFER_SIZE_LIMIT)

    echo "Setting queued_chunks_limit_size for each buffer to
    $BUFFER_QUEUE_LIMIT"

    export BUFFER_QUEUE_LIMIT

    echo "Setting chunk_limit_size for each buffer to $BUFFER_SIZE_LIMIT"

    export BUFFER_SIZE_LIMIT


    issue_deprecation_warnings


    # this should be the last thing before launching fluentd so as not to use

    # jemalloc with any other processes

    if type -p jemalloc-config > /dev/null 2>&1 ; then
        export LD_PRELOAD=$( jemalloc-config --libdir )/libjemalloc.so.$( jemalloc-config --revision )
        export LD_BIND_NOW=1 # workaround for https://bugzilla.redhat.com/show_bug.cgi?id=1544815
    fi


    # In case of an update to secure fluentd container, copy the fluentd pos
    files to their new

    # locations under /var/lib/fluentd/pos. Moving old pos files is not possible
    since /var/log

    # is mounted read-only in the secure fluentd container.

    #

    POS_FILES_DIR=${FILE_BUFFER_PATH}/pos

    mkdir -p $POS_FILES_DIR

    if [ -f /var/log/openshift-apiserver/audit.log.pos -a ! -f
    ${POS_FILES_DIR}/oauth-apiserver.audit.log ] ; then
        cp /var/log/openshift-apiserver/audit.log.pos ${POS_FILES_DIR}/oauth-apiserver.audit.log
    fi

    declare -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
    )

    for POS_FILE in es-containers.log.pos journal_pos.json
    oauth-apiserver.audit.log

    do
      POS_FILES_FROM_TO["/var/log/$pos_file"]="${POS_FILES_DIR}/$pos_file"
    done

    for FROM in "${!POS_FILES_FROM_TO[@]}"

    do
        TO=${POS_FILES_FROM_TO[$FROM]}
        if [ -f "$FROM" -a ! -f "$TO" ] ; then
          cp "$FROM" "$TO"
        fi
    done


    exec fluentd $fluentdargs

