FluentD to Splunk

[venksel]

Hi There,
I'm trying to get the logs forwarded from containers in Kubernetes over to Splunk using HEC. Fluentd has been deployed and fluent.conf is updated with the below in the Config Map. FluentD deployed has the HEC plugin installed that I got from here https://github.com/splunk/fluent-plugin-splunk-hec

kind: ConfigMap
apiVersion: v1
metadata:
  name: fluentd-config
  namespace: fluentd-logging
  selfLink: /api/v1/namespaces/fluentd-logging/configmaps/fluentd-config
  uid: 8769fb6d-6c7a-4ba4-bce9-597e373fafe3
  resourceVersion: '85955'
  creationTimestamp: '2020-07-29T10:26:44Z'
  labels:
    k8s-app: fluentd-splunk
  annotations:
    kubectl.kubernetes.io/last-applied-configuration: >
      {"apiVersion":"v1","data":{"fluent.conf":"\u003cmatch fluent.**\u003e\n 
      @type null\n\u003c/match\u003e\n\u003csource\u003e\n  @type tail\n 
      @log_level fatal\n  @id in_tail_container_logs\n  path
      /var/log/containers/*.log\n  pos_file
      /var/log/fluentd-containers.log.pos\n  tag
      \"#{ENV['FLUENT_CONTAINER_TAIL_TAG'] || 'kubernetes.*'}\"\n 
      read_from_head true\n  \u003cparse\u003e\n    @type multi_format\n   
      \u003cpattern\u003e\n      format json\n      time_key time\n     
      time_format %Y-%m-%dT%H:%M:%S.%NZ\n    \u003c/pattern\u003e\n   
      \u003cpattern\u003e\n      format /^(?\u003ctime\u003e.+)
      (?\u003cstream\u003estdout|stderr) [^ ]* (?\u003clog\u003e.*)$/\n     
      time_format %Y-%m-%dT%H:%M:%S.%N%:z\n    \u003c/pattern\u003e\n 
      \u003c/parse\u003e\n\u003c/source\u003e\n\u003cfilter **\u003e\n  @type
      concat\n  key log\n  separator \"\"\n  multiline_start_regexp /^\\d{2}/\n 
      flush_interval 1\n  timeout_label
      @NORMAL\n\u003c/filter\u003e\n\u003cmatch **\u003e\n  @type relabel\n 
      @label @NORMAL\n\u003c/match\u003e\n\u003clabel @NORMAL\u003e\n 
      \u003cfilter **\u003e\n    @type record_transformer\n    @id
      filter_containers_stream_transformer\n    \u003crecord\u003e\n     
      stream_name ${tag_parts[3]}\n    \u003c/record\u003e\n    remove_keys
      $.kubernetes.pod_id, $.kubernetes.master_url,
      $.kubernetes.container_image_id, $.kubernetes.namespace_id,
      $.kubernetes.labels, $.docker, $.stream, $.stream_name,
      $.kubernetes.container_name, $.kubernetes.container_image,
      $.kubernetes.host, $.kubernetes.namespace_name\n  \u003c/filter\u003e\n 
      \u003cfilter kubernetes.**\u003e\n    @type kubernetes_metadata\n    @id
      filter_kube_metadata\n    kubernetes_url
      \"#{ENV['FLUENT_FILTER_KUBERNETES_URL'] || 'https://' +
      ENV.fetch('KUBERNETES_SERVICE_HOST') + ':' +
      ENV.fetch('KUBERNETES_SERVICE_PORT') + '/api'}\"\n    verify_ssl
      \"#{ENV['KUBERNETES_VERIFY_SSL'] || true}\"\n    ca_file
      \"#{ENV['KUBERNETES_CA_FILE']}\"\n  \u003c/filter\u003e\n  \u003cfilter
      **\u003e\n    @type record_transformer\n    enable_ruby true\n   
      \u003crecord\u003e\n      kubernetes
      ${record[\"kubernetes\"].merge({\"cluster_name\":\"test-fluentd-cnative1\"})}\n   
      \u003c/record\u003e\n  \u003c/filter\u003e\n  \u003cmatch
      kubernetes.**\u003e\n    @type rewrite_tag_filter\n   
      \u003crule\u003e\n      key $['kubernetes']['namespace_name']\n     
      pattern ^(.+)$\n      tag $1.${tag}\n    \u003c/rule\u003e\n 
      \u003c/match\u003e\n  \u003cmatch default.kubernetes.**\u003e\n    @type
      file\n    path /var/log/fluent/upgrade_default*.log\n   
      \u003cformat\u003e\n      @type json\n    \u003c/format\u003e\n   
      buffer_type file\n    buffer_path /var/log/upgrade_default*.log\n   
      time_slice_format %Y-%m-%d.%H%M\n    append true\n    flush_interval 
      1s\n 
      \u003c/match\u003e\n\u003c/label\u003e\n"},"kind":"ConfigMap","metadata":{"annotations":{},"labels":{"k8s-app":"fluentd-splunk"},"name":"fluentd-config","namespace":"fluentd-logging"}}
data:
  fluent.conf: |
    <match fluent.**>
      @type null
    </match>
    <source>
      @type tail
      @log_level fatal
      @id in_tail_container_logs
      path /var/log/containers/*.log
      pos_file /var/log/fluentd-containers.log.pos
      tag "#{ENV['FLUENT_CONTAINER_TAIL_TAG'] || 'kubernetes.*'}"
      read_from_head true
      <parse>
        @type multi_format
        <pattern>
          format json
          time_key time
          time_format %Y-%m-%dT%H:%M:%S.%NZ
        </pattern>
        <pattern>
          format /^(?<time>.+) (?<stream>stdout|stderr) [^ ]* (?<log>.*)$/
          time_format %Y-%m-%dT%H:%M:%S.%N%:z
        </pattern>
      </parse>
    </source>
    <filter **>
      @type concat
      key log
      separator ""
      multiline_start_regexp /^\d{2}/
      flush_interval 1
      timeout_label @NORMAL
    </filter>
    <match **>
      @type relabel
      @label @NORMAL
    </match>
    <label @NORMAL>
      <filter **>
        @type record_transformer
        @id filter_containers_stream_transformer
        <record>
          stream_name ${tag_parts[3]}
        </record>
        remove_keys $.kubernetes.pod_id, $.kubernetes.master_url, $.kubernetes.container_image_id, $.kubernetes.namespace_id, $.kubernetes.labels, $.docker, $.stream, $.stream_name, $.kubernetes.container_name, $.kubernetes.container_image, $.kubernetes.host, $.kubernetes.namespace_name
      </filter>
      <filter kubernetes.**>
        @type kubernetes_metadata
        @id filter_kube_metadata
        kubernetes_url "#{ENV['FLUENT_FILTER_KUBERNETES_URL'] || 'https://' + ENV.fetch('KUBERNETES_SERVICE_HOST') + ':' + ENV.fetch('KUBERNETES_SERVICE_PORT') + '/api'}"
        verify_ssl "#{ENV['KUBERNETES_VERIFY_SSL'] || true}"
        ca_file "#{ENV['KUBERNETES_CA_FILE']}"
      </filter>
      <filter **>
        @type record_transformer
        enable_ruby true
        <record>
          kubernetes ${record["kubernetes"].merge({"cluster_name":"test-fluentd-cnative1"})}
        </record>
      </filter>
      <match kubernetes.**>
        @type rewrite_tag_filter
        <rule>
          key $['kubernetes']['namespace_name']
          pattern ^(.+)$
          tag $1.${tag}
        </rule>
      </match>
      <match default.kubernetes.**>
        @type splunk_hec
        hec_host splunk-host.com
        hec_port 8088
        hec_token xxxxxxxxxx
        index test
        source ${tag}
        sourcetype _json
        <format>
          @type json
        </format>
      </match>
    </label>

 Currently, however,  it's not working for some reason. Any insights on what config change is required to the above will be greatly appreciated.

Cheers,

Rachael