How to understand how the reduce transform works with kubernetes_logs?

[bmday]

Hi!

We have a Kubernetes cluster v1.29.5, Vector 0.40.0 (helm chart), OpenSearch 2.15 and we want to get pod logs, where the .message field can contain data in JSON format, as well as multi-line data. Many services, many logs in different formats.

The idea according to which we configure Vector:

1. Get data from pods using the kubernetes_logs source.
2. Pass data from kubernetes_logs to the remap transform.
   Using the remap transform, we determine the type of data in the .message field (JSON or not) in order to parse the JSON, remove unnecessary fields, etc.
3. Pass data from remap to the reduce transform to process the .message field in order to merge multi-line logs with a regular expression pattern.
4. Pass data to the Elasticsearch (Opensearch) sink.

Simply put: K8s - remap (parse .message as JSON if .message is_json = true, otherwise do nothing and pass the event on) - reduce (if .message matches the regexp pattern, otherwise do nothing and pass the event on) - Opensearch.

The main problem is that after reduce we lose some fields.

We performed the following experiment. In parallel, we took data from each transform in Opensearch and saw duplicate events in it, which differ in the set of fields (see to screens).

screen_1

screen_2

screen_3

The screenshots show that the events are different, processed by different transforms, and their set of fields is different (.kubernetes.node_labels, .kubernetes.pod_labels, etc.)

We want to understand, is this normal behavior, or a bug? How can we achieve behavior when after the reduce transform we will save the fields that get into the reduce input?

Here is the Vector configuration, from which we wanted to get the expected behavior.
values.yaml:

role: "Agent"

service:
  enabled: false

customConfig:
    data_dir: /vector-data-dir
    api:
      enabled: false
      address: 127.0.0.1:8686
      playground: false
    log_schema:
      timestamp_key: "@timestamp"
    sources:
      kubernetes_logs:
        type: kubernetes_logs
        extra_namespace_label_selector: "kubernetes.io/metadata.name notin (kube-system)"
#        node_annotation_fields:
#          node_labels: ""
#        namespace_annotation_fields:
#          namespace_labels: ""
#        pod_annotation_fields:
#          pod_annotations: ""
#          pod_owner: ""
#          pod_ips: ""
#          pod_uid: ""


    transforms:
      kubernetes_logs_transformer:
        type: remap
        inputs:
          - kubernetes_logs
#          - reduce_multiline
        source: |
            .kubernetes.pod_labels = map_keys(object!(.kubernetes.pod_labels), recursive: true) -> |key| { replace(key, ".", "_") }
            if exists(.message) {
               #.message = string!(.message)
               if is_json(string!(.message)) {
                    parsed, err = parse_json(string!(.message))
                        if err != null {
                             log(err, level: "error")
                        }
                          del(.message)
                          .json = parsed
               } # else {
                #    .wrong_json = .message
               #}
            }
            .transformer = "kubernetes_logs_transformer (remap)" # for test dropped labels via reduce transform
            if exists(.kubernetes.pod_labels.creationTimestamp) {
                 .kubernetes.pod_labels.creationTimestamp = from_unix_timestamp!(to_int!(.kubernetes.pod_labels.creationTimestamp))
            }
            if exists(.json.ts) {
                 del(.json.ts)
            }
            if exists(.json.level) && !is_string(.json.level) {
                 .json.level = to_string!(.json.level)
            }
            if exists(.kubernetes.pod_labels."pod-template-hash") {
                 del(.kubernetes.pod_labels."pod-template-hash")
            }


      reduce_multiline:
        type: reduce
        inputs:
          - kubernetes_logs
#          - kubernetes_logs_transformer
        merge_strategies:
          message: concat_newline
        starts_when: |
            #.message = string!(.message)
            if exists(.message) && match(string!(.message), r'^\{|^time=|^\[33m|^\[[0-9]{4}-[0-9]{2}-[0-9]{2}|^[0-9]{4}-[0-9]{2}-[0-9]{2}|^[0-9]{2}/[[:alpha:]]+/[0-9]{4}|^\[[0-9]{2}-[[:alpha:]]+-[0-9]{4}|^\d+\.\d+\.\d+\.\d+|^ts=|^\[GIN\]|^\d+/\d+/\d+[[:space:]]\d+:\d+:\d+|^[[:alpha:]]\d+[[:space:]]\d+:\d+:\d+\.\d+') {
                #if err != null {
                #  false;
                #} else {
                #  matched;
                #}
                 true;
            } else {
                 false;
              }
        max_events: 100
        group_by:
          - file

    sinks:
      stdout:
        type: console
        inputs: [kubernetes_logs]
        encoding:
          codec: json
      es_cluster:
        inputs:
#          - "kubernetes_logs_transformer"
#          - "kubernetes_logs"
          - "reduce_multiline"
#           - "*" # for test
        type: "elasticsearch"
        api_version: v8
        endpoints:
          - "https://test-opensearch.example.com:9200"
        bulk:
          index: "vector-test-%Y-%m-%d"
        tls:
          ca_file: "/etc/ssl/ca.crt"
          verify_hostname: false
        auth:
          strategy: "basic"
          user: "vector_user"
          password: "vector_password"
        encoding:
          timestamp_format: "rfc3339"

[jszwedko]

Could you try v0.40.1 of Vector? There was a bug fix to reduce that might be impacting you. I'm also curious if you see different behavior with v0.39.0.

[bmday]

Hi, @jszwedko!
Thanks for your help! We deployed version 0.40.1 and the reduce transform behavior has indeed changed and is similar to what we expected.

However, in the Vector console we started seeing errors that confused us, and we don't quite understand what caused them. In the remap transform we remove the dots and substitute underscores instead to avoid field conflicts between Vector and Opensearch. This works as expected if we don't use the reduce transform. If we include the reduce transform after our remap, we encounter the errors that we have listed below.

2024-08-28T09:26:51.037323Z ERROR transform{component_kind="transform" component_id=reduce_multiline component_type=reduce}: vector::internal_events::reduce: Internal log [Event field could not be reduced.] has been suppressed 38604 times.
2024-08-28T09:26:51.037359Z ERROR transform{component_kind="transform" component_id=reduce_multiline component_type=reduce}: vector::internal_events::reduce: Event field could not be reduced. path=KeyString("kubernetes.pod_labels.app_kubernetes_io/component") error=InvalidPathSyntax { path: "kubernetes.pod_labels.app_kubernetes_io/component" } error_type="condition_failed" stage="processing" internal_log_rate_limit=true
2024-08-28T09:26:51.037389Z ERROR transform{component_kind="transform" component_id=reduce_multiline component_type=reduce}: vector::internal_events::reduce: Internal log [Event field could not be reduced.] is being suppressed to avoid flooding.

The kubernetes.pod_labels.app.kubernetes.io/component field was processed in remap, as were some others with a similar name, and the output was kubernetes.pod_labels.app_kubernetes_io/component. Apparently, reduce can't handle the / character, but why does it affect reduce like that since we use the .message field in our reduce config?

If this is the default behavior of reduce, how can this be bypassed or resolved so that does not give errors?

Thanks for help! We like Vector more and more.

[bmday]

The same errors in the console if fields like kubernetes.pod_labels.controller-revision-hash get into reduce:

2024-08-28T10:23:19.899262Z ERROR transform{component_kind="transform" component_id=reduce_multiline component_type=reduce}: vector::internal_events::reduce: Internal log [Event field could not be reduced.] has been suppressed 7995 times.
2024-08-28T10:23:19.899315Z ERROR transform{component_kind="transform" component_id=reduce_multiline component_type=reduce}: vector::internal_events::reduce: Event field could not be reduced. path=KeyString("kubernetes.pod_labels.controller-revision-hash") error=InvalidPathSyntax { path: "kubernetes.pod_labels.controller-revision-hash" } error_type="condition_failed" stage="processing" internal_log_rate_limit=true
2024-08-28T10:23:19.899345Z ERROR transform{component_kind="transform" component_id=reduce_multiline component_type=reduce}: vector::internal_events::reduce: Internal log [Event field could not be reduced.] is being suppressed to avoid flooding.

[bmday]

In Vector version 0.39.0, everything seems to work as expected, with no reports of reduce errors in the console.

[jszwedko]

Thanks for testing @bmday . I think you are running into this regression: #21165. We plan to look at it soon. For now, I'd suggest running 0.39.0 until we resolve it.