Allow num_consumers to be greater than 1 if batch enabled and partitioner defined

[lahsivjar]

Description

As per #12473, the number of consumers was set to 1 to prevent contention due to single-threaded batchers. However, with the introduction of multiBatcher the batching will be done in multiple goroutines.

Optimally a dynamic configuration to update the number of consumers to the number of partitions would be ideal, however, this could also cause contention as the queue is not partitioned and we can have hotspots where one partioner messages are more than other partitions. This PR makes it possible to configure the number of consumers to be greater than 1 statically (based on otel config) which should be better than what we have right now which bottlenecks the consumption even with partitioned batchers.

Link to tracking issue

Related to #12473

Testing

N/A

Documentation

N/A

[codecov]

Codecov Report

❌ Patch coverage is 33.33333% with 2 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (main@aef8b1c). Learn more about missing BASE report.
⚠️ Report is 82 commits behind head on main.

Files with missing lines	Patch %	Lines
.../exporterhelper/internal/queuebatch/queue_batch.go	33.33%	2 Missing ⚠️

❌ Your patch status has failed because the patch coverage (33.33%) is below the target coverage (95.00%). You can increase the patch coverage or adjust the target coverage.

Additional details and impacted files

@@           Coverage Diff           @@
##             main   #13607   +/-   ##
=======================================
  Coverage        ?   91.52%           
=======================================
  Files           ?      639           
  Lines           ?    41547           
  Branches        ?        0           
=======================================
  Hits            ?    38024           
  Misses          ?     2704           
  Partials        ?      819           

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[jmacd]

I'm afraid that my understanding of the exporterhelper's batcher has diverged from what's really happening. It would help if there were a document explaining how partitioning and num_consumers relate, or how much concurrency I can get without a partitioner configured.

As a thought experiment, I am interested in "simple concurrent batching", let's say, where there is not a meaningful partition function, all data goes to the same logical destination. I understand the idea that only one active thread can take the mutex and locate a full batch, but the way I've thought about the num_consumers setting, it determines how many concurrent exporters there are. I would not expect to be limited to num_consumers==1 for any reason by the batcher.

Can you explain the current situation or refer to an explanation of how I can get simple concurrent batching?

[lahsivjar]

(Updated for posterity following the clarification in this comment)

Can you explain the current situation or refer to an explanation of how I can get simple concurrent batching?

👋 @jmacd this is an excellent question. I have also spent quite some time to wrap my head around the current situation so let me take a stab at explaining this based on what I have understood so far.

TL;DR if you configure batching using the exporterhelper's QueueBatch and without partitioning then, AFAIU, the Consume call of partition batcher's will be called one by one, but it could create multiple goroutines for flushing/exporting data with the max concurrency as num_consumers. If you configure only sending queue without any batching then, as expected, there will be no batching but you will get concurrency equal to the configured num_consumers.

Case 1: Configuring exporterhelper with queue and batch but without partitioner

1. On configuring QueueBatch config a new QueueBatch is created with the export function as the consume function for the mature batch. The export function simply calls the configured sender in the exporter.
2. The QueueBatch is a pipeline of queuing and batching. The queue is a simple async queue (ref) and the batcher, without partitioner defined, is a single instance of partition batcher (ref).
3. The partition batcher, on a high level, attempts to split the batch and calls the consume function in sync in a goroutine for all the batches with size greater than the min size (ref). The flush happens in a separate goroutine - bounded by the configured num_consumers.
4. If there is a batch that is not mature enough to be consumed, it is cached by the batcher and then merged in the following Consume call to the partition batcher. At this stage the flush happens somewhat similar to the above case - a catch here being that the context metadata (and other values) are modified (you are definitely aware of the decision here related to this).

The number of consumers for the queue, which basically refers to the number of gorutines consuming from the queue and pushing it to the partition batcher, in this case is forced to be 1 to avoid contention due to the non-concurrent batcher (ref).

Case 2: Configuring exporterhelper without batching

In this case, a batcher is basically disabled (ref). The disabled batcher directly calls the consume function in sync (ref).

The num_consumers in this directly relates to the number of concurrent flushes.

[jmacd]

@sfc-gh-sili
would you comment? I would expect for num_consumers to work, without partitioning as a requirement. The change here asks for much less (and I want more). While one export is happening (a single-batch consumer) the other num_consumers-1 should be busy forming batches and exporting.

[sfc-gh-sili]

@jmacd Reading through the code, my understanding is that when both batch and queue are enabled:

• The numConsumer passed to NewBatcher defines the maximum number of goroutines responsible for exporting
• The numConsumer passed to NewQueue defines the number of goroutines responsible for reading from the queue and adding items into batches.

With the current setting "Reading from the queue and adding the item to the batch" is bounded by 1 goroutine at a time, and I guess we kept things that way for single batch because adding to the batch is guarded by a lock anyway.

While one export is happening (a single-batch consumer) the other num_consumers-1 should be busy forming batches and exporting.

I believe the current implementation allows numConsumer goroutines exporting at the same time.

[lahsivjar]

I believe the current implementation allows numConsumer goroutines exporting at the same time.

Thanks for the clarification @sfc-gh-sili, I can see where my understanding went wrong. I missed the point that flush happens via a worker pool and is not in sync with the Consume call of the partition batcher - even if the asyncQueue calls the partition batcher's Consume func with one goroutine when partitioner is defined, it will create a separate goroutine for each flush. (Updated my initial explanation too with this).

With the current setting "Reading from the queue and adding the item to the batch" is bounded by 1 goroutine at a time, and I guess we kept things that way for single batch because adding to the batch is guarded by a lock anyway.

Right, however, with the introduction of partition batcher, I think this PR would still make sense as the locks would be for each instance of partition batcher, and we could achieve true concurrency when the queue is reading for more than one partition.

I guess the ideal solution here would be to have a partitioned queue in addition to a partitioned batcher so that we can eliminate the contention due to a heavily loaded partition. This could also deal with any head-of-line issues if we assigned atleast one worker from the pool for each partition.

[JakeDern]

I'm late to the party, but this conversation about num_consumers and how that affects the queuing and batching in exporterhelper is exactly what I'm interested in and just spent time understanding.

The behavior I'm interested in achieving is the following:

• Keep the collector stateless and avoid dependence on a persistent queue
• Enable batching somewhere in the pipeline
• Hold acknowledgements back to the clients until a batch has been successfully flushed
• Have exports of multiple batches happening in parallel

It seems like this is almost possible using exporterhelper, except for the fact that the batching and queueing are fused together in QueueBatch and are enabled/disabled together under the QueueSender. So currently we have to have a queue in front of the batcher.

Does it seem reasonable to allow for separate Batching and Queueing senders? Maybe @lahsivjar or @sfc-gh-sili could comment. (And thanks for all of the recent work in this area, I've been very excited to learn about it!)

[JakeDern]

I'm late to the party, but this conversation about num_consumers and how that affects the queuing and batching in exporterhelper is exactly what I'm interested in and just spent time understanding.

The behavior I'm interested in achieving is the following:

• Keep the collector stateless and avoid dependence on a persistent queue
• Enable batching somewhere in the pipeline
• Hold acknowledgements back to the clients until a batch has been successfully flushed
• Have exports of multiple batches happening in parallel

It seems like this is almost possible using exporterhelper, except for the fact that the batching and queueing are fused together in QueueBatch and are enabled/disabled together under the QueueSender. So currently we have to have a queue in front of the batcher.

Does it seem reasonable to allow for separate Batching and Queueing senders? Maybe @lahsivjar or @sfc-gh-sili could comment. (And thanks for all of the recent work in this area, I've been very excited to learn about it!)

I think I see now that this is possible. Something that I missed initially was the wiring for wait_for_result in the memory queue offer implementation.

I think this makes my scenario possible by configuring the following:

• Move batching settings to the exporter helper
• Enable wait_for_result: true, block_on_overflow: true, and num_consumers > 1 in exporter helper