configmap-fluentd.yaml

{{ $agent := fromYaml (include "splunk-otel-collector.agent" .) }}
{{- if and (eq (include "splunk-otel-collector.logsEnabled" .) "true") $agent.enabled (eq .Values.logsEngine "fluentd") }}
apiVersion: v1
kind: ConfigMap
metadata:
  name: {{ template "splunk-otel-collector.fullname" . }}-fluentd
  namespace: {{ template "splunk-otel-collector.namespace" . }}
  labels:
    {{- include "splunk-otel-collector.commonLabels" . | nindent 4 }}
    app: {{ template "splunk-otel-collector.name" . }}
    chart: {{ template "splunk-otel-collector.chart" . }}
    release: {{ .Release.Name }}
    heritage: {{ .Release.Service }}
data:
  fluent.conf: |-
    @include system.conf
    @include source.containers.conf
    @include source.files.conf
    @include source.journald.conf
    @include output.conf
    @include prometheus.conf

  system.conf: |-
    # system wide configurations
    <system>
      log_level {{ .Values.fluentd.config.logLevel }}
      root_dir /tmp/fluentd
    </system>

  prometheus.conf: |-
    # input plugin that exports metrics
    <source>
      @type prometheus
    </source>

    # input plugin that collects metrics from MonitorAgent
    <source>
      @type prometheus_monitor
    </source>

    # input plugin that collects metrics for output plugin
    <source>
      @type prometheus_output_monitor
    </source>

  source.containers.conf: |-
    # This configuration file for Fluentd / td-agent is used
    # to watch changes to Docker log files. The kubelet creates symlinks that
    # capture the pod name, namespace, container name & Docker container ID
    # to the docker logs for pods in the /var/log/containers directory on the host.
    # If running this fluentd configuration in a Docker container, the /var/log
    # directory should be mounted in the container.
    # reading kubelet logs from journal
    #
    # Reference:
    # https://github.com/kubernetes/community/blob/20d2f6f5498a5668bae2aea9dcaf4875b9c06ccb/contributors/design-proposals/node/kubelet-cri-logging.md
    #
    # Json Log Example:
    # {"log":"[info:2016-02-16T16:04:05.930-08:00] Some log text here\n","stream":"stdout","time":"2016-02-17T00:04:05.931087621Z"}
    # CRI Log Example (not supported):
    # 2016-02-17T00:04:05.931087621Z stdout P { 'long': { 'json', 'object output' },
    # 2016-02-17T00:04:05.931087621Z stdout F 'splitted': 'partial-lines' }
    # 2016-02-17T00:04:05.931087621Z stdout F [info:2016-02-16T16:04:05.930-08:00] Some log text here
    <source>
      @id containers.log
      @type tail
      @label @CONCAT
      tag tail.containers.*
      path {{ .Values.fluentd.config.path | default "/var/log/containers/*.log" }}
      {{- if .Values.fluentd.config.excludePath }}
      exclude_path {{ .Values.fluentd.config.excludePath | toJson }}
      {{- end }}
      pos_file {{ .Values.fluentd.config.posFilePrefix }}-containers.log.pos
      path_key source
      read_from_head true
      enable_stat_watcher {{ .Values.fluentd.config.enableStatWatcher | default true }}
      refresh_interval {{ .Values.fluentd.config.refreshInterval | default 60 }}
      <parse>
        @include source.containers.parse.conf
        time_key time
        time_type string
        localtime false
      </parse>
    </source>

  source.files.conf: |-
    # This fluentd conf file contains sources for log files other than container logs.
    {{- $checks := dict "hasFileLog" false }}
    {{- range $name, $logDef := .Values.fluentd.config.logs }}
    {{- if $logDef.from.file }}
    {{- set $checks "hasFileLog" true | and nil }}
    <source>
      @id tail.file.{{ $name }}
      @type tail
      @label @CONCAT
      tag tail.file.{{ or $logDef.sourcetype $name }}
      path {{ $logDef.from.file.path }}
      pos_file {{ $.Values.fluentd.config.posFilePrefix }}-{{ $name }}.pos
      read_from_head true
      path_key source
      {{- if $logDef.multiline }}
      multiline_flush_interval {{ $logDef.multiline.flushInterval | default "5s" }}
      {{- end }}
      <parse>
        {{- if $logDef.multiline }}
        @type multiline
        format_firstline {{ $logDef.multiline.firstline }}
        {{- if $logDef.timestampExtraction }}
        format1 /^(?<log>{{ $logDef.timestampExtraction.regexp }}.*)$/
        time_key time
        time_type string
        time_format {{ $logDef.timestampExtraction.format }}
        {{- end }}
        {{- else if $logDef.timestampExtraction }}
        @type regexp
        expression /^(?<log>{{ $logDef.timestampExtraction.regexp }}.*)$/
        time_key time
        time_type string
        time_format {{ $logDef.timestampExtraction.format }}
        {{- else }}
        @type none
        message_key log
        {{- end }}
      </parse>
    </source>
    {{- end }}
    {{- end }}

  source.journald.conf: |-
    # This fluentd conf file contains configurations for reading logs from systemd journal.
    {{- range $name, $logDef := .Values.fluentd.config.logs }}
    {{- if $logDef.from.journald }}
    {{- set $checks "hasJournald" true | and nil }}
    <source>
      @id journald-{{ $name }}
      @type systemd
      @label @CONCAT
      tag journald.{{ or $logDef.sourcetype $name }}
      path {{ $.Values.fluentd.config.journalLogPath | quote }}
      matches [{ "_SYSTEMD_UNIT": {{ $logDef.from.journald.unit | quote }} }]
      read_from_head true
      <storage>
        @type local
        persistent true
        path /var/log/splunkd-fluentd-journald-{{ $name }}.pos.json
      </storage>
      <entry>
        field_map {"MESSAGE": "log", "_SYSTEMD_UNIT": "source"}
        field_map_strict true
      </entry>
    </source>
    {{- end }}
    {{- end }}

  output.conf: |-
    #Events are emitted to the CONCAT label from the container, file and journald sources for multiline processing.
    <label @CONCAT>
      @include output.filter.conf
      # = handle custom multiline logs =
      {{- range $name, $logDef := .Values.fluentd.config.logs }}
      {{- if and $logDef.from.pod $logDef.multiline }}
      {{- $filenameGlob := regexReplaceAll "\\*+" (printf "%s*%s*" $logDef.from.pod ($logDef.from.container | default "")) "*" }}
      <filter tail.containers.var.log.containers.{{ $filenameGlob }}.log>
        @type concat
        key log
        timeout_label @SPLUNK
        stream_identity_key stream
        multiline_start_regexp {{ $logDef.multiline.firstline }}
        flush_interval {{ $logDef.multiline.flushInterval | default "5s" }}
        separator {{ $logDef.multiline.separator | default "" | quote }}
        use_first_timestamp true
      </filter>
      {{- end }}
      {{- end }}
      # = filters for journald logs =
      {{- range $name, $logDef := .Values.fluentd.config.logs }}
      {{- if and $logDef.from.journald $logDef.multiline }}
      <filter journald.{{ or $logDef.sourcetype $name }}>
        @type concat
        key log
        timeout_label @SPLUNK
        multiline_start_regexp {{ $logDef.multiline.firstline }}
        flush_interval {{ $logDef.multiline.flushInterval | default "5s" }}
      </filter>
      {{- end }}
      {{- end }}
      # Events are relabeled then emitted to the SPLUNK label
      <match **>
        @type relabel
        @label @SPLUNK
      </match>
    </label>
    <label @SPLUNK>
      # Extract k8s metadata from container logs source paths. Use original logs source
      # "/var/log/containers/<k8s.pod.k8s>_<k8s.namespace.name>_<k8s.container.name>-<container.id>.log"
      # first then check symlinks to the new k8s logs format
      # "/var/log/pods/<k8s.namespace.name>_<k8s.pod.name>_<k8s.pod.uid>/<k8s.container.name>/<k8s.container.restart_count>.log"
      # to fetch "k8s.pod.uid" that will be used to get other k8s metadata by otel-collector from k8s API.
      <filter tail.containers.**>
        @type record_modifier
        <record>
          pods_source ${File.readlink(record['source'])}
        </record>
      </filter>
      <filter tail.containers.**>
        @type jq_transformer
        jq '.record | . + (.source | capture("^/var/log/containers/(?<k8s.pod.name>[^_]+)_(?<k8s.namespace.name>[^_]+)_(?<k8s.container.name>[-0-9a-z]+)-(?<container.id>[^.]+).log$")) | . + (.pods_source | capture("^/var/log/pods/[^_]+_[^_]+_(?<k8s.pod.uid>[^/]+)/[^._]+/[0-9]+.log$") // {}) | .sourcetype = ("kube:container:" + .["k8s.container.name"])'
      </filter>

      @include output.transform.conf

      # create source and sourcetype
      {{- if $checks.hasJournald }}
      <filter journald.**>
        @type jq_transformer
        jq '.record.source = "{{ .Values.fluentd.config.journalLogPath }}/" + .record.source | .record.sourcetype = (.tag | ltrimstr("journald.")) {{- range .Values.extraAttributes.custom }}| .record.{{ .name }} = "{{ .value }}" {{- end }} |.record'
      </filter>
      {{- end }}

      # = filters for non-container log files =
      {{- if $checks.hasFileLog }}
      # extract sourcetype
      <filter tail.file.**>
        @type jq_transformer
        jq '.record.sourcetype = (.tag | ltrimstr("tail.file.")) {{- range .Values.extraAttributes.custom }}| .record.{{ .name }} = "{{ .value }}" {{- end }} | .record'
      </filter>
      {{- end }}

      # = custom filters specified by users =
      {{- range $name, $filterDef := .Values.fluentd.config.customFilters }}
      {{- if and $filterDef.tag $filterDef.type }}
      <filter {{ $filterDef.tag }}>
        @type {{ $filterDef.type }}
        {{- if $filterDef.body }}
        {{- $filterDef.body | nindent 8 }}
        {{- end }}
      </filter>
      {{- end }}
      {{- end }}

      <filter **>
        @type record_transformer
        enable_ruby
        <record>
          com.splunk.sourcetype ${record.dig("sourcetype") ? record.dig("sourcetype") : ""}
          com.splunk.source ${record.dig("source") ? record.dig("source") : ""}
        </record>
        remove_keys pods_source,source,sourcetype
      </filter>

      # = output =
      <match **>
        @type forward
        heartbeat_type udp
        <server>
          host 127.0.0.1
          port 8006
        </server>
        {{- with .Values.fluentd.config.buffer }}
        <buffer>
        {{- range $parameter, $value := . }}
          {{ $parameter }} {{ $value }}
        {{- end }}
        </buffer>
        {{- end }}
        <format>
          # we just want to keep the raw logs, not the structure created by docker or journald
          @type single_value
          message_key log
          add_newline false
        </format>
      </match>
    </label>

  {{- /*
    Default configuration files for CRI container engine.
    Can be overridden for docker engine by prepare-fluentd-config container.
  */}}
  source.containers.parse.conf: |-
    @type regexp
    expression /^(?<time>.+) (?<stream>stdout|stderr)( (?<partial_flag>[FP]))? (?<log>.*)$/
    time_format {{ .Values.fluentd.config.containers.criTimeFormat }}
  output.filter.conf: |-
    # = handle cri/containerd multiline format =
    <filter tail.containers.var.log.containers.**>
      @type concat
      key log
      partial_key partial_flag
      partial_value P
      separator ''
      timeout_label @SPLUNK
    </filter>
  output.transform.conf: |-
    # extract pod_uid and container_name for CRIO runtime
    # currently CRI does not produce log paths with all the necessary
    # metadata to parse out pod, namespace, container_name, container_id.
    # this may be resolved in the future by this issue: https://github.com/kubernetes/kubernetes/issues/58638#issuecomment-385126031
    <filter tail.containers.var.log.pods.**>
      @type jq_transformer
      jq '.record | . + (.source | capture("/var/log/pods/(?<pod_uid>[^/]+)/(?<container_name>[^/]+)/(?<container_retry>[0-9]+).log")) | .sourcetype = ("kube:container:" + .container_name)'
    </filter>
    # rename pod_uid and container_name to otel semantics.
    <filter tail.containers.var.log.pods.**>
      @type record_transformer
      <record>
        k8s.pod.uid ${record["pod_uid"]}
        k8s.container.name ${record["container_name"]}
      </record>
    </filter>
{{- end }}