Send fluentd logs through otel (#109)

This sends logs collected by fluentd to otelcol via fluentforward. There are some limitations noted by TODOs that I will file issues to track but they should not affect the common cases. It's mostly around configuring of hec_exporter TLS settings. It still attaches k8s metadata on the fluentd side as it uses various annotations to construct source/sourcetype in some cases. May not be worth trying to fix with move to filelog receiver.
signalfx · May 4, 2021 · 9b37f46 · 9b37f46
1 parent 79d6701
commit 9b37f46
Show file tree

Hide file tree

Showing 7 changed files with 111 additions and 140 deletions.
diff --git a/helm-charts/splunk-otel-collector/templates/_helpers.tpl b/helm-charts/splunk-otel-collector/templates/_helpers.tpl
@@ -58,6 +58,19 @@ Get Splunk ingest host
 {{- .Values.ingestHost | default (printf "ingest.%s.signalfx.com" .Values.splunkRealm) }}
 {{- end -}}
 
+{{/*
+Get Splunk log URL
+*/}}
+{{- define "splunk-otel-collector.logUrl" -}}
+{{- $host := include "splunk-otel-collector.ingestHost" . }}
+{{- $endpoint := printf "%s://%s" .Values.ingestProtocol $host }}
+{{- if or (and (eq .Values.ingestProtocol "http") (ne (toString .Values.ingestPort) "80")) (and (eq .Values.ingestProtocol "https") (ne (toString .Values.ingestPort) "443")) }}
+{{- printf "%s:%s/v1/log" $endpoint (toString .Values.ingestPort) }}
+{{- else }}
+{{- $endpoint }}
+{{- end }}
+{{- end -}}
+
 {{/*
 Get Splunk ingest URL
 */}}

diff --git a/helm-charts/splunk-otel-collector/templates/config/_otel-agent.tpl b/helm-charts/splunk-otel-collector/templates/config/_otel-agent.tpl
@@ -14,6 +14,9 @@ extensions:
 
 receivers:
   {{- include "splunk-otel-collector.otelTraceReceivers" . | nindent 2 }}
+  fluentforward:
+    endpoint: 0.0.0.0:8006
+
   # Prometheus receiver scraping metrics from the pod itself
   prometheus/agent:
     config:
@@ -175,14 +178,20 @@ processors:
 exporters:
 
   {{- if .Values.otelCollector.enabled }}
-  # If collector is enabled, metrics and traces will be sent to collector
+  # If collector is enabled, metrics, logs and traces will be sent to collector
   otlp:
     endpoint: {{ include "splunk-otel-collector.fullname" . }}:4317
     insecure: true
   {{- else }}
-  # If collector is disabled, metrics and traces will be set to to SignalFx backend
+  # If collector is disabled, metrics, logs and traces will be sent to to SignalFx backend
   {{- include "splunk-otel-collector.otelSapmExporter" . | nindent 2 }}
+  splunk_hec:
+    endpoint: {{ include "splunk-otel-collector.logUrl" . }}
+    token: "${SPLUNK_HEC_TOKEN}"
+    index: "{{ .Values.logsBackend.hec.indexName }}"
+    insecure_skip_verify: {{ .Values.logsBackend.hec.insecureSSL | default false }}
   {{- end }}
+
   signalfx:
     correlation:
     {{- if .Values.otelCollector.enabled }}
@@ -203,8 +212,24 @@ service:
   # The default pipelines should to be changed. You can add any custom pipeline instead.
   # In order to disable a default pipeline just set it to `null` in otelAgent.config overrides.
   pipelines:
+    logs:
+      receivers: [fluentforward]
+      processors:
+        - memory_limiter
+        - batch
+        - resource
+        - resourcedetection
+        {{- if .Values.environment }}
+        - resource/add_environment
+        {{- end }}
+      exporters:
+        {{- if .Values.otelCollector.enabled }}
+        - otlp
+        {{- else }}
+        - splunk_hec
+        {{- end }}
 
-    # default traces pipeline
+    # Default traces pipeline.
     traces:
       receivers: [otlp, jaeger, smartagent/signalfx-forwarder, zipkin]
       processors:
@@ -224,7 +249,7 @@ service:
         {{- end }}
         - signalfx
 
-    # default metrics pipeline
+    # Default metrics pipeline.
     metrics:
       receivers: [hostmetrics, kubeletstats, receiver_creator]
       processors:

diff --git a/helm-charts/splunk-otel-collector/templates/config/_otel-collector.tpl b/helm-charts/splunk-otel-collector/templates/config/_otel-collector.tpl
@@ -71,6 +71,11 @@ processors:
 
 exporters:
   {{- include "splunk-otel-collector.otelSapmExporter" . | nindent 2 }}
+  splunk_hec:
+    endpoint: {{ include "splunk-otel-collector.logUrl" . }}
+    token: "${SPLUNK_HEC_TOKEN}"
+    index: "{{ .Values.logsBackend.hec.indexName }}"
+    insecure_skip_verify: {{ .Values.logsBackend.hec.insecureSSL | default false }}
   signalfx:
     ingest_url: {{ include "splunk-otel-collector.ingestUrl" . }}
     api_url: {{ include "splunk-otel-collector.apiUrl" . }}
@@ -104,6 +109,12 @@ service:
 
     # default logs pipeline
     logs:
+      receivers: [otlp, fluentforward]
+      processors: [memory_limiter, batch]
+      exporters: [splunk_hec]
+
+    # logs pipeline for receiving and exporting SignalFx events
+    logs/signalfx-events:
       receivers: [signalfx]
       processors: [memory_limiter, batch]
       exporters: [signalfx]

diff --git a/helm-charts/splunk-otel-collector/templates/configmap-fluentd.yaml b/helm-charts/splunk-otel-collector/templates/configmap-fluentd.yaml
@@ -203,12 +203,6 @@ data:
       </match>
     </label>
     <label @SPLUNK>
-      <filter **>
-        @type record_transformer
-        <record>
-          host.name "#{ENV['K8S_NODE_NAME']}"
-        </record>
-      </filter>
       # Enrich log with k8s metadata
       <filter tail.containers.**>
         @type kubernetes_metadata
@@ -221,49 +215,55 @@ data:
         <record>
           # set the sourcetype from splunk.com/sourcetype pod annotation or set it to kube:container:CONTAINER_NAME
           sourcetype ${record.dig("kubernetes", "annotations", "splunk.com/sourcetype") ? "kube:"+record.dig("kubernetes", "annotations", "splunk.com/sourcetype") : "kube:container:"+record.dig("kubernetes","container_name")}
-          container_name ${record.dig("kubernetes","container_name")}
-          namespace ${record.dig("kubernetes","namespace_name")}
-          pod ${record.dig("kubernetes","pod_name")}
-          container_id ${record.dig("docker","container_id")}
-          pod_uid ${record.dig("kubernetes","pod_id")}
-          node_name "#{ENV['K8S_NODE_NAME']}"
-          cluster_name {{ .Values.clusterName }}
-          container_image ${record.dig("kubernetes","container_image")}
           # set the index field to the value found in the pod splunk.com/index annotations. if not set, use namespace annotation, or default to the default_index
-          index ${record.dig("kubernetes", "annotations", "splunk.com/index") ? record.dig("kubernetes", "annotations", "splunk.com/index") : record.dig("kubernetes", "namespace_annotations", "splunk.com/index") ? (record["kubernetes"]["namespace_annotations"]["splunk.com/index"]) : ("{{ .Values.logsBackend.hec.indexName | default "main"}}")}
+          index ${record.dig("kubernetes", "annotations", "splunk.com/index") ? record.dig("kubernetes", "annotations", "splunk.com/index") : record.dig("kubernetes", "namespace_annotations", "splunk.com/index")}
+
+          k8s.container.name ${record.dig("kubernetes","container_name")}
+          k8s.namespace.name ${record.dig("kubernetes","namespace_name")}
+          k8s.pod.name ${record.dig("kubernetes","pod_name")}
+          container.id ${record.dig("docker","container_id")}
+          k8s.pod.uid ${record.dig("kubernetes","pod_id")}
+          container.image.name ${record.dig("kubernetes","container_image")}
+
           {{- range .Values.extraAttributes.podLabels }}
           k8s.pod.labels.{{ . }} ${record.dig("kubernetes","labels","{{ . }}")}
           {{- end }}
-          blacklist ${record.dig("kubernetes", "annotations", "splunk.com/exclude") ? record.dig("kubernetes", "annotations", "splunk.com/exclude") : record.dig("kubernetes", "namespace_annotations", "splunk.com/exclude") ? (record["kubernetes"]["namespace_annotations"]["splunk.com/exclude"]) : ("false")}
-          {{- range .Values.extraAttributes.custom }}
-          {{ .name }} "{{ .value }}"
-          {{- end }}
-          {{- if .Values.environment }}
-          deployment.environment "{{ .Values.environment }}"
-          {{- end }}
+
+          denylist ${record.dig("kubernetes", "annotations", "splunk.com/exclude") ? record.dig("kubernetes", "annotations", "splunk.com/exclude") : record.dig("kubernetes", "namespace_annotations", "splunk.com/exclude") ? (record["kubernetes"]["namespace_annotations"]["splunk.com/exclude"]) : ("false")}
         </record>
       </filter>
       <filter tail.containers.**>
-        # Exclude all logs that are blacklisted
+        # Exclude all logs that are denylisted
         @type grep
         <exclude>
-          key blacklist
+          key denylist
           pattern /^true$/
         </exclude>
       </filter>
       # 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
       {{- if eq .Values.fluentd.config.containers.logFormatType "cri" }}
       <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) | .index = {{ .Values.logsBackend.hec.indexName | default "main" | quote }}'
+        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 }}
 
       # 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.")) | .record.cluster_name = "{{ .Values.clusterName }}" | .record.index = {{ .Values.logsBackend.hec.indexName | default "main" | quote }} {{- range .Values.extraAttributes.custom }}| .record.{{ .name }} = "{{ .value }}" {{- end }} |.record'
+        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 }}
 
@@ -272,7 +272,7 @@ data:
       # extract sourcetype
       <filter tail.file.**>
         @type jq_transformer
-        jq '.record.sourcetype = (.tag | ltrimstr("tail.file.")) | .record.cluster_name = "{{ .Values.clusterName }}" | .record.index = {{ .Values.logsBackend.hec.indexName | default "main" | quote }} {{- range .Values.extraAttributes.custom }}| .record.{{ .name }} = "{{ .value }}" {{- end }} | .record'
+        jq '.record.sourcetype = (.tag | ltrimstr("tail.file.")) {{- range .Values.extraAttributes.custom }}| .record.{{ .name }} = "{{ .value }}" {{- end }} | .record'
       </filter>
       {{- end }}
 
@@ -288,91 +288,25 @@ data:
       {{- 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") : ""}
+          com.splunk.index ${record.dig("index") ? record.dig("index") : "{{ .Values.logsBackend.hec.indexName | default "main"}}"}
+        </record>
+        remove_keys denylist,docker,kubernetes,source,sourcetype,index
+      </filter>
+
       # = output =
       <match **>
-        {{- if .Values.logsBackend.splunkIngestAPI.ingestAPIHost }}
-        @type splunk_ingest_api
-        {{- with .Values.logsBackend.splunkIngestAPI.serviceClientIdentifier }}
-        service_client_identifier {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.splunkIngestAPI.serviceClientSecretKey }}
-        service_client_secret_key {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.splunkIngestAPI.tokenEndpoint }}
-        token_endpoint {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.splunkIngestAPI.ingestAuthHost }}
-        ingest_auth_host {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.splunkIngestAPI.ingestAPIHost }}
-        ingest_api_host {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.splunkIngestAPI.tenant }}
-        ingest_api_tenant {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.splunkIngestAPI.eventsEndpoint }}
-        ingest_api_events_endpoint {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.splunkIngestAPI.debugIngestAPI }}
-        debug_http {{ . }}
-        {{- end }}
-        {{- else }}
-        @type splunk_hec
-        {{- with .Values.logsBackend.hec.protocol | default .Values.ingestProtocol }}
-        protocol {{ . }}
-        {{- end }}
-        {{- with .Values.logsBackend.hec.host | default (include "splunk-otel-collector.ingestHost" .) }}
-        hec_host {{ . | quote }}
-        {{- end }}
-        {{- with .Values.logsBackend.hec.port | default .Values.ingestPort }}
-        hec_port {{ . }}
-        {{- end }}
-        hec_token "#{ENV['SPLUNK_HEC_TOKEN']}"
-        index_key index
-        insecure_ssl {{ .Values.logsBackend.hec.insecureSSL | default false }}
-        {{- if .Values.logsBackend.hec.clientCert }}
-        client_cert /fluentd/etc/splunk/hec_client_cert
-        {{- end }}
-        {{- if .Values.logsBackend.hec.clientKey }}
-        client_key /fluentd/etc/splunk/hec_client_key
-        {{- end }}
-        {{- if .Values.logsBackend.hec.caFile }}
-        ca_file /fluentd/etc/splunk/hec_ca_file
-        {{- end }}
-        {{- end }}
-        host "#{ENV['K8S_NODE_NAME']}"
-        source_key source
-        sourcetype_key sourcetype
-        <fields>
-          # 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
-          {{- if eq .Values.fluentd.config.containers.logFormatType "cri"}}
-          container_retry
-          {{- else }}
-          container.image.name container_image
-          {{- end }}
-          k8s.pod.uid pod_uid
-          k8s.pod.name pod
-          k8s.container.name container_name
-          k8s.namespace.name namespace
-          k8s.node.name node_name
-          k8s.cluster.name cluster_name
-          container.id container_id
-          host.name
-          {{- range .Values.extraAttributes.custom }}
-          {{ .name }}
-          {{- end }}
-          {{- range .Values.indexFields }}
-          {{ . }}
-          {{- end }}
-          {{- range .Values.extraAttributes.podLabels }}
-          k8s.pod.labels.{{ . }}
-          {{- end }}
-          {{- if .Values.environment }}
-          deployment.environment
-          {{- end }}
-        </fields>
+        @type forward
+        heartbeat_type udp
+        <server>
+          host 127.0.0.1
+          port 8006
+        </server>
         {{- with .Values.fluentd.config.buffer }}
         <buffer>
         {{- range $parameter, $value := . }}

diff --git a/helm-charts/splunk-otel-collector/templates/daemonset.yaml b/helm-charts/splunk-otel-collector/templates/daemonset.yaml
@@ -63,14 +63,6 @@ spec:
             valueFrom:
               fieldRef:
                 fieldPath: metadata.name
-          - name: SPLUNK_HEC_TOKEN
-            valueFrom:
-              secretKeyRef:
-                name: {{ template "splunk-otel-collector.secret" . }}
-                key: splunk_hec_token
-          {{- with .Values.fluentd.extraEnvs }}
-          {{- . | toYaml | nindent 10 }}
-          {{- end }}
         resources:
           {{- toYaml .Values.fluentd.resources | nindent 10 }}
         volumeMounts:
@@ -155,9 +147,15 @@ spec:
           - name: HOST_DEV
             value: /hostfs/dev
           {{- end }}
+          - name: SPLUNK_HEC_TOKEN
+            valueFrom:
+              secretKeyRef:
+                name: {{ template "splunk-otel-collector.secret" . }}
+                key: splunk_hec_token
           {{- with .Values.otelAgent.extraEnvs }}
           {{- . | toYaml | nindent 10 }}
           {{- end }}
+
         readinessProbe:
           httpGet:
             path: /

diff --git a/helm-charts/splunk-otel-collector/templates/deployment-collector.yaml b/helm-charts/splunk-otel-collector/templates/deployment-collector.yaml
@@ -70,6 +70,11 @@ spec:
               secretKeyRef:
                 name: {{ include "splunk-otel-collector.secret" . }}
                 key: splunk_access_token
+          - name: SPLUNK_HEC_TOKEN
+            valueFrom:
+              secretKeyRef:
+                name: {{ template "splunk-otel-collector.secret" . }}
+                key: splunk_hec_token
           {{- with .Values.otelCollector.extraEnvs }}
           {{- . | toYaml | nindent 10 }}
           {{- end }}