New component: OTTL-based request partitioning processor

[axw]

The purpose and use-cases of the new component

I would like to propose a new processor for partitioning batches using OTTL. The processor would be configured with a mapping of OTTL expressions to partitioning keys, and:

1. Infer the relevant context from the expressions (e.g. request, resource, log, datapoint, etc.)
2. Evaluate the expressions against each batch
3. Partition the batch according to the partitioning keys
4. Send each batch partition to the next consumer, add the partitioning keys & values to request metadata

Some use cases below:

Topic routing and message key/partitioning for message broker exporters

The idea for this came from a PoC I implemented for routing and partitioning data before exporting to Kafka -- see #38888 (comment). This is intended to address several issues:

• At the moment the Kafka exporter supports configuration for determining a topic from a resource attribute, but this does not behave correctly with multiple resources - see topic_from_attribute does not work as expected #37470
• the exporter supports partitioning traces by trace_id, but does not support the same for logs
• the exporter does not support partitioning by arbitrary attributes -- see [exporter/kafkaexporter] Enable Partitioning by Specific Attribute for Logs & Metrics  #38484

It's just a matter of time before users ask for the same functionality in other exporters, e.g. Pulsar.

Batching by arbitrary data and metadata attributes

In open-telemetry/opentelemetry-collector#10825 it has been suggested that the exporterhelper batch sender should be enhanced with support for batching by request metadata or by pdata attribute. The former case is simple, while the latter tends towards OTTL, so I think this would be best covered by a separate processor like this one.

General purpose "groupby" processor

Finally, this processor would provide arbitrary "group by" functionality, which could enable replacing existing processors:

• we could replace the https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/groupbyattrsprocessor processor by using the new partitioning processor followed by the transform processor to set resource attributes from the partitioning key
• we could replace the https://github.com/open-telemetry/opentelemetry-collector-contrib/tree/main/processor/groupbytraceprocessor processor by using the new partitioning processor followed by a new "buffering" processor, i.e. one that buffers batches by some request metadata keys and emits them after a configurable duration

(Whether we want to replace them is another matter, just saying it's a possible use case.)

Example configuration for the component

processors:
  partitioner:
    keys:
      logs_topic: Concat(request.metadata["X-Tenant-Id"]).otlp_logs
      logs_message_key: log.trace_id

In this example, requests would be partitioned by the X-Tenant-Id client request header, and data would further be partitioned by trace ID. Each partitioned batch would include the request metadata logs_topic and logs_message_key.

Telemetry data types supported

Logs, metrics, traces. When OTTL supports profiles, profiles too.

Code Owner(s)

axw

Sponsor (optional)

No response

Additional context

No response

[dehaansa]

I really like the general concept here, but question if this should be another component.

This sounds like it could be satisfied by the ability to control metadata within the transform processor, and following open-telemetry/opentelemetry-collector#10825 and supporting metadata-aware batching configuration.

As there is inertia towards deprecating the batchprocessor, defining a new more complex batch processor doesn't feel like a

[axw]

This sounds like it could be satisfied by the ability to control metadata within the transform processor

I thought about it, and we could do that, but I think it will be more complex to configure and will require some pretty major changes to the transform processor.

If we were to extend the transform processor, we would need to:

• make it capable of splitting a batch into multiple batches
• efficiently copy data, e.g. if you're operating on the log context and you want to group by trace_id, in order for that to end up in request metadata you would need to create a new plog.Logs for each trace ID, with the matching log record(s) and Resource & Scope copied

Given that they will not be relevant to all users, and that not all transform capabilities will be relevant to those doing partitioning, I think it's simpler overall to separate the responsibilities.

and following open-telemetry/opentelemetry-collector#10825 and supporting metadata-aware batching configuration.

As there is inertia towards open-telemetry/opentelemetry-collector#12022 the batchprocessor, defining a new more complex batch processor doesn't feel like a

To clarify, we would still be relying on open-telemetry/opentelemetry-collector#10825. The new component I'm proposing would just be for splitting batches and setting metadata. That metadata could then be used for metadata-aware batching, among other use cases, such as Kafka multi-tenant topic routing and message keying.

[mattsains]

I think this would be quite useful for a rate limiting use case as well - this processor could be used as a precursor to set up "per-tenant" rate limiting processor where there is a separate bucket based on the resource, instrumentation scope, or other attribute of the incoming data. For example, if you want to process all the Error logs your application emits, but rate limit any other logs (eg., lots of useless Info logs) to avoid overloading the downstream ingestion endpoint. Or maybe you have two applications publishing to the same OTLP endpoint and want to prioritize one application's logs over another.

I was wondering whether you were planning on synchronously grouping data and sending it onward, or combining multiple ConsumeLogs calls like the batch processor does. If the latter, it would be less useful for the rate limiting case, because downstream rate limiter processors wouldn't be able to throw an error back up to the receiver nor pause the processing (depending on how rate limiting is implemented)

[axw]

@mattsains agreed!

I have in mind that the processor will be stateless and synchronous. Each incoming ConsumeX call to the processor will result in one or more outgoing ConsumeX calls to the next consumer. Each of those calls will be in a separate goroutine, and the processor will block until all are done, probably using errgroup.

[mattsains]

@axw should I open another issue for the creation of a rate limiter processor that can limit by request metadata? I think it's implied by the need for partitioned rate limiting and this groupby processor

[axw]

@mattsains there's #35204 which mentions the need for limiting by metadata (e.g. see #35204 (comment)). That issue has gone stale. It would be good to have your voice there to add some weight.