Splunk-connect-for-kubernetes: How can my application send logs to different index's for compliance purposes

[pc-rshetty]

What happened:
Currently i can send all the logs from a pod to a single index.
What you expected to happen:
We are migrating a lot of EC2 based application to EKS. Some security related logs on ec2 were written on file system and then sent to different index due to retention policy being higher compared to regular log file. In containers all logs go to a single index. Is there a way my application running on a k8s pod can send this to a different index ?
How to reproduce it (as minimally and precisely as possible):

Anything else we need to know?:

Environment:
EKS

• Kubernetes version (use kubectl version):
  1.14
• Ruby version (use ruby --version):
• OS (e.g: cat /etc/os-release):
  Amaxon Linux 2
• Splunk version:
• Others:

[njferrant]

From the SCK 1.4.0 docs, you can do it with Annotations. Either at the pod or the namespace level (or both):
https://github.com/splunk/splunk-connect-for-kubernetes/tree/release/1.4.0#managing-sck-log-ingestion-by-using-annotations

[pc-rshetty]

thanks @njferrant i had read this but i though this is at a pod/namespace level. What i need is on the same pod some logs go to index1 and some go to index2.

[matthewmodestino]

Hi @pc-rshetty!

What ultimately works for you depends on your particular log stream...

Can you confirm that the events you want to route share the same source fields in Splunk?

In other words, do the logs get emitted by the same container in the pod?

[pc-rshetty]

yes @matthewmodestino the same container will emit the log just that some logs which need security teams attention and need longer retention period need to go to different index. Rest will go to a common index.

[matthewmodestino]

ok, we can do this in fluentd filters in your configmap or on the splunk destination side with props/transforms.

For example, lets say the dynamically generated sourcetype is kube:container:foo

you can either, create a fluentd record transformer filter for this container that sets the index when it matches your security event, or you can create a sourcetype config on the Splunk side to override the index on events that match your condition.

Got an example i can see?

[pc-rshetty]

ok, we can do this in fluentd filters in your configmap or on the splunk destination side with props/transforms.

For example, lets say the dynamically generated sourcetype is kube:container:foo

you can either, create a fluentd record transformer filter for this container that sets the index when it matches your security event, or you can create a sourcetype config on the Splunk side to override the index on events that match your condition.

Got an example i can see?

thanks @matthewmodestino The logs that should go to different index is some thing like this, it is prefixed with
[ABC_SECURITY] - INFO foobar event type

Can you please give me an example of how we can do this "_it matches your security event, or you can create a sourcetype config on the Splunk side to override the index on events that match your condition._"

[pc-rshetty]

@matthewmodestino just a gentle reminder

[limlengchye]

We have similar need in our organisation too. In our case, the output sent to more than one hec tokens. I made a PR to allow BYO match config. https://github.com/splunk/splunk-connect-for-kubernetes/pull/399

[matthewmodestino]

@limlengchye why do you need multiple tokens?? Are the tokens all from the same Splunk enviro??

This sounds like micro optimization... you already have the option to add custom filters..and adding more rules and outputs and sending to multiple backends will likely degrade fluentd perf. Have you done any perf testing on your PR?

That seems like an overhead nightmare and a recipe for issues and errors and likely sounds like you need a solution like Data Stream Processor....

In general, the HEC token here is more of a transport layer item and shoyldn’t be used for identity/rbac imo. Beacuse SCK ensures the payload is ready to be indexed, you get automagic index routing.

Can you please explain your requirements?

I usually tell customers NOT to create a multiple HEC token nightmare and use one per cluster...

What benefit are you under the impression you are getting, or what “requirements” are you alluding to?

[limlengchye]

@matthewmodestino
Multiple HEC tokens allow different type payloads of events to be indexed based on the configuration of the token ( as long as the token is valid ). This is from security point of view. We have certain indexes are restricted, and it can only be received by HEC ( and indexed ) if a token is valid. We are currently using multiple HEC tokens in production, and has been running fine so far for more than a year.

[matthewmodestino]

Understood, but this integration sends "cooked" payloads, so we suggest one token per cluster to allow for the index routing logic we have built to take over. You can still protect your tokens and indexes but you will need to balance security with usability as always.

[rockb1017]

@limlengchye could you share the custom conf you are providing to send use multiple HEC token? i think it could be helpful if we can make it more native as well.

[pc-mreeves]

@matthewmodestino - I'm @pc-rshetty 's colleague...

ok, we can do this in fluentd filters in your configmap or on the splunk destination side with props/transforms.

For example, lets say the dynamically generated sourcetype is kube:container:foo

you can either, create a fluentd record transformer filter for this container that sets the index when it matches your security event, or you can create a sourcetype config on the Splunk side to override the index on events that match your condition.

Got an example i can see?

You have already been a huge help WRT the multiline solution.

So if I take your meaning I cannot directly modify the index the same way I did multiline?

E.g. (how I wish it worked shown below where I could specify alternate indices for a given container/pod)

logs:
  foobar-app:
      from:
          pod: "foobar-service-*"
          container: "foobar-service"
      multiline:
          firstline: /^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}/
      // my imaginary wishlist
      index_route:
          index: cool_log_index
          pattern: /^\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2} \[COOL_LOG_TYPE\]/

Can I somehow annotate the log via a regex here and then elsewhere use customFilters to select a different index to send them to?

https://github.com/splunk/splunk-connect-for-kubernetes/blob/develop/helm-chart/splunk-connect-for-kubernetes/values.yaml#L388

Thanks!

[ljuaneda]

Hello,

We have the exact same need and we've figured how to do it, but it based on parsing the log for a specific pattern :

      <filter tail.containers.**>
        @type record_transformer
        enable_ruby
        <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")}
          container_image ${record.dig("kubernetes","container_image")}
          # set the cluster_name field to the configured value, or default to "cluster_name"
          cluster_name cluster_name
          # 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"]) : ("main")}
          index ${record.dig("kubernetes", "annotations", "splunk.com/indexbt") ? record.dig("kubernetes", "annotations", "splunk.com/indexbt") : record.dig("kubernetes", "namespace_annotations", "splunk.com/indexbt") ? (record["kubernetes"]["namespace_annotations"]["splunk.com/indexbt"]) : ("business-trace")}
          label_app ${record.dig("kubernetes","labels","app")}
          label_k8s-app ${record.dig("kubernetes","labels","k8s-app")}
          label_release ${record.dig("kubernetes","labels","release")}
          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")}
        </record>
      </filter>

Then you'll have to parse the records for a defined pattern to match your need, for example "business.trace" :

      <filter tail.containers.**>
        @type jq_transformer
        jq '.record.index = (.record.index as $index | .record.indexbt as $indexbt | .record.log | if contains("business.trace") then $indexbt else $index end) | .record'
      </filter>