[processor/transform] Concat function concatenates too many times

[edenkoveshi]

Description

I'm trying to change the service name to service.namespace:service.name (delimiter doesn't really matter)
To do that, I've tried the following query:

set(attributes[service.name],Concat(": ",attributes[service.namespace],attributes[service.name]))

The result I'm getting is <namespace>:<namespace>:<namespace>:....<namespace>:<name>
The namespace is concatenated an arbitrary number of times. Name exactly once.

I've also tried the following queries, they result in the same behaviour:

set(attributes[temp],Concat(": ",attributes[service.namespace],attributes[service.name]))
set(attributes[service.name],attributes[temp])
delete_key(attributes[temp]) #If I omit this line, temp is also assigned <namespace>:<namespace>:<namespace>:....<namespace>:<name>

set(attributes[service.name],Concat("",attributes[service.namespace],":",attributes[service.name]))

Looking at the source code (there is no documentation for Concat) the queries seem ok.

Collector version

0.60.0

Environment information

OpenTelemetry Collector Contrib 0.60.0 deployed on Openshift using OpenTelemetry Operator 0.60.0
Traces backend is Jaeger 1.35 deployed on Openshift with Jaeger Operator
I am using 2 pipelines, both use the same traces receivers (otlp,jaeger) and send to different Jaeger instances. Transform processor is only used on one of them. The other one uses spanmetrics processor (which is configured correctly and works)

OpenTelemetry Collector configuration

receivers:
  otlp:
  jaeger:
processors:
  batch:
  spanmetrics: #Configured correctly, full configuration is irrelevant
  transform:
    traces:
      queries:
        # The queries described above
exporters:
  jaeger:
    endpoint: <jaeger-collector>
  jaeger/2:
    endpoint: <jaeger-2-collector>
pipelines:
  traces:
    receivers: [otlp,jaeger]
    processors: [spanmetrics,batch]
    exporters: [jaeger]
  traces/2:
    receivers: [otlp,jaeger]
    processors: [transform,batch]
    exporters: [jaeger/2]

[github-actions]

Pinging code owners: @TylerHelmuth @kentquirk @bogdandrutu. See Adding Labels via Comments if you do not have permissions to add labels yourself.

[TylerHelmuth]

@edenkoveshi can you provide your entire transformprocessor config? Are there any queries that modify attributes[service.namespace]?

[TylerHelmuth]

@edenkoveshi also, attributes[service.namespace] is incorrect syntax. It should be attributes["service.namespace"]. The processor should crash without the "", so I suspect that in your actual config you have them?

[TylerHelmuth]

@edenkoveshi I haven't been able to reproduce this yet

func Test_Bug(t *testing.T) {
	td := ptrace.NewTraces()
	rs0 := td.ResourceSpans().AppendEmpty()
	rs0.Resource().Attributes().PutString("host.name", "localhost")
	rs0ils0 := rs0.ScopeSpans().AppendEmpty()
	span := rs0ils0.Spans().AppendEmpty()
	span.SetName("operationA")
	span.SetKind(1)
	span.Attributes().PutString("service.namespace", "namespace")
	span.Attributes().PutString("service.name", "name")

	statement := `set(attributes["service.name"], Concat(": ",attributes["service.namespace"],attributes["service.name"]))`
	processor, err := NewProcessor([]string{statement}, Functions(), component.TelemetrySettings{})

	assert.NoError(t, err)

	temp, err := processor.ProcessTraces(context.Background(), td)
	assert.NoError(t, err)
	attr, _ := temp.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes().Get("service.name")
	fmt.Println(attr.AsString())
}

output:

=== RUN   Test_Bug
namespace: name
--- PASS: Test_Bug (0.00s)
PASS

Process finished with the exit code 0

I suspect that the issue is some combination of statements, not only the one using concat.

[edenkoveshi]

@edenkoveshi also, attributes[service.namespace] is incorrect syntax. It should be attributes["service.namespace"]. The processor should crash without the "", so I suspect that in your actual config you have them?

It is indeed quoted

@edenkoveshi I haven't been able to reproduce this yet

func Test_Bug(t *testing.T) {
	td := ptrace.NewTraces()
	rs0 := td.ResourceSpans().AppendEmpty()
	rs0.Resource().Attributes().PutString("host.name", "localhost")
	rs0ils0 := rs0.ScopeSpans().AppendEmpty()
	span := rs0ils0.Spans().AppendEmpty()
	span.SetName("operationA")
	span.SetKind(1)
	span.Attributes().PutString("service.namespace", "namespace")
	span.Attributes().PutString("service.name", "name")

	statement := `set(attributes["service.name"], Concat(": ",attributes["service.namespace"],attributes["service.name"]))`
	processor, err := NewProcessor([]string{statement}, Functions(), component.TelemetrySettings{})

	assert.NoError(t, err)

	temp, err := processor.ProcessTraces(context.Background(), td)
	assert.NoError(t, err)
	attr, _ := temp.ResourceSpans().At(0).ScopeSpans().At(0).Spans().At(0).Attributes().Get("service.name")
	fmt.Println(attr.AsString())
}

output:

=== RUN   Test_Bug
namespace: name
--- PASS: Test_Bug (0.00s)
PASS

Process finished with the exit code 0

I suspect that the issue is some combination of statements, not only the one using concat.

The query is actually set(resource.attributes["service.name"],Concat(":",resource.attributes["k8s.namespace.name"],resource.attributes["service.name"]))
and it actually changes the trace name. This is probably the cause for it.

Edit:
Using

set(attributes["temp"],Concat(":",resource.attributes["k8s.namespace.name"],resource.attributes["service.name"])) 

Sets the temp attribute correctly. The problem occurs when I'm trying to set the name attribute.
However, when adding

set(resource.attributes["service.name"],attributes["temp"])

it concatenates the namespace an arbitrary number of times both in temp and service.name (for example, concatenates 2 times in temp and 5 in service.name)

My exact configuration is:

receivers:
  jaeger:
  otlp:
  otlp/dummy:
    protocols:
      grpc:
        endpoint: ":12345" #For service metrics processor

processors:
  batch:
  spanmetrics:
    metrics_exporter: prometheus
  transform:
    traces:
      queries:
         -set(resource.attributes["service.name"],Concat(":",resource.attributes["k8s.namespace.name"],resource.attributes["service.name"])) 

exporters:
  jaeger:
    endpoint: <my-jaeger-endpoint>
  jaeger/2:
    endpoint: <my-jaeger-endpoint-2>
  logging:
  prometheus:
    endpoint: "0.0.0.0:8889"

extensions:
   health_check:

service:
  extensions:
    - health_check
   pipelines:
     traces:
       receivers:
         - otlp
         -  jaeger
        processors:
          - spanmetrics
          - batch
        exporters:
          - logging
          -  jaeger
      traces/2:
       receivers:
         - otlp
         -  jaeger
        processors:
          - transform
          - batch
        exporters:
          -  jaeger/2
      metrics:
         receivers: [otlp/dummy]
         exporters: [prometheus]
  telemetry:
    metrics:
       address: "0.0.0.0:8888"
    logs:
      level: "info"

[TylerHelmuth]

I see what is happening. The issue is that the statement is setting resource.attributes["service.name"] but also depends on the value of resource.attributes["service.name"] to do that set.

For Spans this isn't a problem because a span would be set once and then the processor moves on. But we are setting a resource, which within the Traces context is currently a "virtual". So the resource is shared for all spans of that resource.

They way the transform processor works for spans is that for every resource, for every scope spans and for every span, execute the list of functions. That means that each group of Spans in a Scope Spans all have the spam InstrumentionScope and Resource. Since the processor executes all the functions on all the spans (conditions apply of course), every time we process a span with the same resource, we update the attribute another time.

Assuming resource.attributes["service.name"] starts as A and resource.attributes["k8s.namespace.name"] starts as B, the processor will do the following:

• resource.attributes["service.name"] is B:A after processing span 1.
• Now that resource.attributes["service.name"] is B:A, when used in the next Concat function, resource.attributes["service.name"] is B:B:A after processing span 2.
• resource.attributes["service.name"] is B:B:B:A after processing span 2.
  .
  .
  .
• resource.attributes["service.name"] is B:...:B:A after processing span N.

This is an existing limitation of the fact that the transform processor works on the "lowest" unit of telemetry (the span in this case). There have been talks of add more contexts, specifically for Instrumentation Scope processing and Resource processing.

In the meantime, you can handle the situation by adding a where clause:

transform:
  traces:
    queries:
      - set(resource.attributes["service.name"],Concat(":",resource.attributes["k8s.namespace.name"],resource.attributes["service.name"])) where IsMatch(resource.attributes["service.name"], ".*:.*") != true

or something like that. It's not super performant because the function's condition will run for every span in the resource, but until the processor can act on a resource context, we're stuck with it.

Proof of that statement working:

func Test_Bug(t *testing.T) {
	td := ptrace.NewTraces()
	rs0 := td.ResourceSpans().AppendEmpty()
	rs0.Resource().Attributes().PutString("k8s.namespace.name", "B")
	rs0.Resource().Attributes().PutString("service.name", "A")
	rs0ils0 := rs0.ScopeSpans().AppendEmpty()
	span1 := rs0ils0.Spans().AppendEmpty()
	span1.SetName("operationA")
	span2 := rs0ils0.Spans().AppendEmpty()
	span2.SetName("operationB")

	statement := `set(resource.attributes["service.name"],Concat(":",resource.attributes["k8s.namespace.name"],resource.attributes["service.name"])) where IsMatch(resource.attributes["service.name"], ".*:.*") != true`
	processor, err := NewProcessor([]string{statement}, Functions(), component.TelemetrySettings{})

	assert.NoError(t, err)

	temp, err := processor.ProcessTraces(context.Background(), td)
	assert.NoError(t, err)
	attr, _ := temp.ResourceSpans().At(0).Resource().Attributes().Get("service.name")
	fmt.Println(attr.AsString())
}

=== RUN   Test_Bug
B:A
--- PASS: Test_Bug (0.00s)
PASS

[edenkoveshi]

It seems to work, thanks
Let me know if you need any help implementing the resource scope attributes concat