[pkg/ottl] setting higher contexts in OTTL can result in unexpected transformations

[mut3]

Component(s)

pkg/ottl, pkg/stanza, processor/transform, receiver/filelog

What happened?

Description

We have a pipeline that uses a filelog input with some operators that set certain fields in attributes. We then have a transform processor that reads those fields to set() some Resource.attributes. The values that get set in the resource attributes of the log appear to be from a different log as if the transform is reading from one log and writing another.

Steps to Reproduce

I do not currently have a MVC for this issue, but I will include a stripped and sanitized config that contains the core logic/entities around the bug we are seeing.

Configuration

receivers:
      filelog:
        start_at: beginning
        operators:
          # route and parse log based on formatting
          - id: get-format
            type: router
            routes:
              - expr: body matches "^\\{"
                output: parser-docker
              - expr: body matches "^[^ Z]+ "
                output: parser-crio
              - expr: body matches "^[^ Z]+Z"
                output: parser-containerd
          - id: parser-crio
            type: regex_parser
            regex: ^(?P<time>[^ Z]+) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
            timestamp:
              layout: 2006-01-02T15:04:05.999999999Z07:00
              layout_type: gotime
              parse_from: attributes.time
          - id: crio-recombine
            type: recombine
            combine_field: attributes.log
            combine_with: ""
            is_last_entry: attributes.logtag == 'F'
            output: extract_metadata_from_filepath
            source_identifier: attributes["log.file.path"]
          - id: parser-containerd
            type: regex_parser
            regex: ^(?P<time>[^ ^Z]+Z) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
            timestamp:
              layout: "%Y-%m-%dT%H:%M:%S.%LZ"
              parse_from: attributes.time
          - id: containerd-recombine
            type: recombine
            combine_field: attributes.log
            combine_with: ""
            is_last_entry: attributes.logtag == 'F'
            output: extract_metadata_from_filepath
            source_identifier: attributes["log.file.path"]
          - id: parser-docker
            type: json_parser
            output: extract_metadata_from_filepath
            timestamp:
              layout: "%Y-%m-%dT%H:%M:%S.%LZ"
              parse_from: attributes.time
          - id: extract_metadata_from_filepath
            parse_from: attributes["log.file.path"]
            regex: ^.*\/(?P<namespace>[^_]+)_(?P<pod_name>[^_]+)_(?P<uid>[a-f0-9\-]+)\/(?P<container_name>[^\._]+)\/(?P<restart_count>\d+)\.log$
            type: regex_parser
          # move log to body, parse to attributes
          - type: move
            from: attributes.log
            to: body
          - type: json_parser
            parse_from: body
            parse_to: attributes

processors:
      transform/attribute_moves:
        error_mode: ignore
        log_statements:
          - context: log
            statements:
              - set(attributes["log.iostream"], attributes["stream"])
              - set(resource.attributes["k8s.container.name"], attributes["container_name"])
              - set(resource.attributes["k8s.namespace.name"], attributes["namespace"])
              - set(resource.attributes["k8s.pod.name"], attributes["pod_name"])
              - set(resource.attributes["k8s.container.restart_count"], attributes["restart_count"])
              - set(resource.attributes["k8s.pod.uid"], attributes["uid"])

Expected Result

Logs where attributes["pod_name"] == resource.attributes["k8s.pod.name"] and resource.attributes["k8s.namespace.name"] == attributes["namespace"]

This is the case for most of the logs emitted from our pipeline

Actual Result

For some small percentage of our logs, the values of

• resource.attributes["k8s.container.name"]
• resource.attributes["k8s.namespace.name"]
• resource.attributes["k8s.pod.name"]
• resource.attributes["k8s.container.restart_count"]
• resource.attributes["k8s.pod.uid"]

Do not match their attribute counterparts, but instead appear to come from a different log event. The fields appear to be internally consistent, as if they all came from the same different log event. For example if resource.attributes["k8s.pod.name"] is wrong and holds the name of some other pod, the resource.attributes["k8s.namespace.name"] will have the namespace of that other pod.

Here is a sanitized example log json pulled from our elasticsearch

Sanitized incorrect log

{
  "_index": ".ds-otel-logs-stg-my-app-2024.03.07-000022",
  "_id": "Pz1km44B8ceE1n7Gbbpq",
  "_version": 1,
  "_score": 0,
  "_source": {
    "@timestamp": "2024-04-01T20:40:10.983479227Z",
    "Attributes": {
      "OTEL_collector_type": "daemonset",
      "canonical": {
        "owner": {
          "team_name": "unknown"
        }
      },
      "cloud": "aws",
      "cluster": "ci-stg",
      "container_name": "cilium-agent",
      "elasticsearch": {
        "index": {
          "suffix": "my-app"
        }
      },
      "environment": "stg",
      "log": {
        "file": {
          "path": "/var/log/pods/kube-system_cilium-hlkmw_fb7fae7c-d3ac-4308-885d-5d1426c96bcc/cilium-agent/0.log"
        },
        "iostream": "stderr"
      },
      "logtag": "F",
      "namespace": "kube-system",
      "pod_name": "cilium-hlkmw",
      "restart_count": "0",
      "stream": "stderr",
      "time": "2024-04-01T20:40:10.983479227Z",
      "uid": "fb7fae7c-d3ac-4308-885d-5d1426c96bcc"
    },
    "Body": "level=info msg=\"Delete endpoint request\" containerID=593b7f5fe7 subsys=daemon",
    "Resource": {
      "k8s": {
        "container": {
          "name": "my-app",
          "restart_count": "0"
        },
        "namespace": {
          "name": "my-app-namespace"
        },
        "pod": {
          "name": "my-app-65b4cf8769-bb66g",
          "uid": "33b38675-15a7-4f64-a3f7-f38a97455ce6"
        }
      }
    },
    "Scope": {
      "name": "",
      "version": ""
    },
    "SeverityNumber": 0,
    "TraceFlags": 0
  },
  "fields": {
    "Attributes.uid": [
      "fb7fae7c-d3ac-4308-885d-5d1426c96bcc"
    ],
    "Attributes.stream": [
      "stderr"
    ],
    "Attributes.environment": [
      "stg"
    ],
    "Attributes.canonical.owner.team_name": [
      "unknown"
    ],
    "Resource.k8s.container.restart_count": [
      "0"
    ],
    "Attributes.log.iostream": [
      "stderr"
    ],
    "Attributes.restart_count": [
      "0"
    ],
    "Attributes.pod_name": [
      "cilium-hlkmw"
    ],
    "Resource.k8s.namespace.name": [
      "my-app-namespace"
    ],
    "TraceFlags": [
      0
    ],
    "Resource.k8s.pod.name": [
      "my-app-65b4cf8769-bb66g"
    ],
    "Scope.name": [
      ""
    ],
    "Attributes.logtag": [
      "F"
    ],
    "Resource.k8s.pod.uid": [
      "33b38675-15a7-4f64-a3f7-f38a97455ce6"
    ],
    "Attributes.elasticsearch.index.suffix": [
      "my-app"
    ],
    "Attributes.cloud": [
      "aws"
    ],
    "SeverityNumber": [
      0
    ],
    "Attributes.log.file.path": [
      "/var/log/pods/kube-system_cilium-hlkmw_fb7fae7c-d3ac-4308-885d-5d1426c96bcc/cilium-agent/0.log"
    ],
    "Body": [
      "level=info msg=\"Delete endpoint request\" containerID=593b7f5fe7 subsys=daemon"
    ],
    "Attributes.time": [
      "2024-04-01T20:40:10.983Z"
    ],
    "Attributes.namespace": [
      "kube-system"
    ],
    "Resource.k8s.container.name": [
      "my-app"
    ],
    "@timestamp": [
      "2024-04-01T20:40:10.983Z"
    ],
    "Attributes.container_name": [
      "cilium-agent"
    ]
  }
}

Collector version

v0.96.0

Environment information

Environment

OS: Amazon Linux 2023 (Amazon's v1.25.16-eks ami)
Compiler(if manually compiled): Collector Helm Chart v0.84.0

OpenTelemetry Collector configuration

See Steps to Reproduce

Log output

No response

Additional context

We never experience this issue with these same moves to resources.attributes done via operators on the filelog receiver:

All operator solution

receivers:
      filelog:
        start_at: beginning
        operators:
          - id: get-format
            routes:
              - expr: body matches "^\\{"
                output: parser-docker
              - expr: body matches "^[^ Z]+ "
                output: parser-crio
              - expr: body matches "^[^ Z]+Z"
                output: parser-containerd
            type: router
          - id: parser-crio
            regex: ^(?P<time>[^ Z]+) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
            timestamp:
              layout: 2006-01-02T15:04:05.999999999Z07:00
              layout_type: gotime
              parse_from: attributes.time
            type: regex_parser
          - combine_field: attributes.log
            combine_with: ""
            id: crio-recombine
            is_last_entry: attributes.logtag == 'F'
            output: extract_metadata_from_filepath
            source_identifier: attributes["log.file.path"]
            type: recombine
          - id: parser-containerd
            regex: ^(?P<time>[^ ^Z]+Z) (?P<stream>stdout|stderr) (?P<logtag>[^ ]*) ?(?P<log>.*)$
            timestamp:
              layout: "%Y-%m-%dT%H:%M:%S.%LZ"
              parse_from: attributes.time
            type: regex_parser
          - combine_field: attributes.log
            combine_with: ""
            id: containerd-recombine
            is_last_entry: attributes.logtag == 'F'
            output: extract_metadata_from_filepath
            source_identifier: attributes["log.file.path"]
            type: recombine
          - id: parser-docker
            output: extract_metadata_from_filepath
            timestamp:
              layout: "%Y-%m-%dT%H:%M:%S.%LZ"
              parse_from: attributes.time
            type: json_parser
          - id: extract_metadata_from_filepath
            parse_from: attributes["log.file.path"]
            regex: ^.*\/(?P<namespace>[^_]+)_(?P<pod_name>[^_]+)_(?P<uid>[a-f0-9\-]+)\/(?P<container_name>[^\._]+)\/(?P<restart_count>\d+)\.log$
            type: regex_parser
          - from: attributes.stream
            to: attributes["log.iostream"]
            type: move
          - from: attributes.container_name
            to: resource["k8s.container.name"]
            type: move
          - from: attributes.namespace
            to: resource["k8s.namespace.name"]
            type: move
          - from: attributes.pod_name
            to: resource["k8s.pod.name"]
            type: move
          - from: attributes.restart_count
            to: resource["k8s.container.restart_count"]
            type: move
          - from: attributes.uid
            to: resource["k8s.pod.uid"]
            type: move
          - from: attributes.log
            to: body
            type: move
          - type: json_parser
            parse_from: body
            parse_to: attributes

[github-actions]

Pinging code owners:

• pkg/ottl: @TylerHelmuth @kentquirk @bogdandrutu @evan-bradley
• pkg/stanza: @djaglowski
• processor/transform: @TylerHelmuth @kentquirk @bogdandrutu @evan-bradley
• receiver/filelog: @djaglowski

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[djaglowski]

At a glance, I believe the issue is caused because ottl operates on a hierarchical data model which groups multiple log records by their shared resource attributes. In my opinion, when setting a value on the resource that comes from an individual log, this should produce a new resource with the updated value and move the log record into that resource.

An example with much unnecessary detail removed:

# input
logs:
  - resource:
    attributes:
      foo: bar
    records:
      - attributes:
          some.key: left
        body: Hello one
      - attributes:
          some.key: right
        body: Hello two

# Current result if we run "set(resource.attributes["num"], attributes["some.key"])
logs:
  - resource:
    attributes:
      foo: bar
      num: right # was "left" for one iteration then was overwritten
    records:
      - attributes:
          some.key: left
        body: Hello one
      - attributes:
          some.key: right
        body: Hello two

# Correct result (in my opinion)
logs:
  - resource:
    attributes:
      foo: bar
      num: left
    records:
      - attributes:
          some.key: left
        body: Hello one
  - resource:
    attributes:
      foo: bar
      num: right
    records:
      - attributes:
          some.key: right
        body: Hello two

In other words, the fact that our pdata packages represents data in a hierarchical manner should be an implementation detail rather than a guiding principle. User directed modifications should generally work as if the data model is completely flat (one log record per scope and resource) and split hierarchical structures accordingly.

[mut3]

Thank your for the explanation!

Based on my understanding then, we should see many fewer issues such as these if we do not try to use ottl to modify resource attributes in processors. Would you expect that to workaround this behavior caused by pData structure grouping logs by resources?

[TylerHelmuth]

@djaglowski is correct.

When using the logs context in the transformprocessor it will cycles through each log, executing each statement in order. When you use a value from a log to set a property in that log's scope or resource all the logs for that scope/resource will try to do the same unless you add some clever where clauses.

As @djaglowski showed, if two logs in a resource have different values, the last (or first depending on Where clauses) will win - you cant get both values in the same resource attribute key unless you concat.

resource that comes from an individual log, this should produce a new resource with the updated value and move the log record into that resource.

I'm not sure we want this to be the default behavior either. It is totally valid for all the spans/datapoints/log records to share an attribute and the user wants that attribute set on the resource. If we always created a new resource each time we set from a "lower" context then we can end up with duplicate resources across the slice.

I also believe that if set had the capability of creating new Resource*Slice entries in my payload it would be overloaded. It would no longer only be setting an attribute, but also making the decision to create a new Resource*Slice. Having the function make that decision unilaterally feels incorrect for an OTTL function.

This issue has been known, but confusing/limiting, behavior since OTTL's inception. I think we'll need a more complex solution than to always split it out into its own Resource/Scope/Log_record combo. I see a couple options:

• Use the groupbyattributes processor to organize the data before sending the to the transformprocessor. I think the processor would need a new feature where the original resource attributes may be included in the new resource.
• Add an optional param to set that allows creating new resources in this situation.
• Add a function that allows creating new ResourceLogsSlice from existing ResourceLogsSlice.
  • Something like this may also require some new language feature that allows running statements on newly created objects, not sure. Some of the features in Proposal for language additions to OTTL #30800 would be applicable.
• A new language feature that let's you "break" out of a list of statements. Something like "if this statement executed, then don't execute anymore statements for this transform context".

@mut3 since you're using the filelog receiver to collect this data we highly recommend doing as much work during the collection as you can. I recommend using the move operator to set the resource attributes instead of the transformprocessor since you have the opportunity to do it via the receiver. In general, the less components in your pipeline the better.

If you must use the transformprocessor for this data, I believe you could hack together a solution like this if you know the exact keys of all your resource attributes:

1. Use a transformprocessor to move all the resource attributes to the logs:
2. use a groupbyattributesor to rearrange the resource/scope/log relationship using ALL the resource keys you merged in step 1 and ALL the attribute keys you were trying to set.
3. Use another transformprocessor to delete the resource keys of the log records.

I highly recommend using the filelog receiver instead of doing that approach lol

[TylerHelmuth]

/cc @evan-bradley

[mut3]

Thank you!

We moved all our resource.attributes setters back to filelog operators and that appears to be working as expected.

Maybe there should be some warnings about modifying resource.attributes from a log context included in the docs for transformprocessor. As a new user I was not at all aware of this danger from my reading of the transformprocessor and ottl context documentation.