Is there an easy way to glean the concurrency pattern of a vector component?

[nzxwang]

For some components like datadog_logs, we found that the component was single-threaded after hitting performance limits so we had to retroactively replace it with a router that would round-robin assign the logs to n datadog_logs components.

It would be great to know the concurrency or "threadedness" of a component beforehand though. For example, is the sample component also single-threaded? Should I expect to have to fanout my logs to several sample components before merging them for downstream processing? What about other components like remap, throttle, etc.?

[jszwedko]

Hi @nzxwang !

The datadog_logs sink isn't single threaded, it will batch and issue requests concurrently. This can be configured via the request.concurrency option. By default "adaptive" concurrency is used where it'll try to optimize the concurrency based on latency and responses from upstream, but you can also set request.concurrency = <some number> to use a static amount of request concurrency.

To answer your other question about how how to determine whether a component executes concurrently:

• Stateful transforms are single-threaded (those with the state: stateful badge in the docs). Examples are the reduce and lua transforms. state: stateless transforms (like sample) execute concurrently
• Unfortunately for sources and sinks it can be harder to tell without looking at the code. Sinks generally are concurrent but there are some exceptions (like the redis sink iirc). Sources are generally not concurrent, but there are also some exceptions like the http_server source. Happy to answer questions about specific components.

[nzxwang]

Hi Jesse. As always, thanks for the detailed explanation (:

Regarding datadog_logs's concurrency however, could you comment on whether what we're doing is an anti-pattern or not? Instead of using a single datadog_logs component and tuning the request.* options, we use a route that randomly forwards logs to 1 of 5 datadog_logs components:

[transforms.remap_fanout_add_partition]
type = "remap"
inputs = ["logs_from_upstream"]
source = '''
  .tags.partition = to_string(random_int(1,6))

[transforms.route_fanout]
type = "route"
inputs = ["remap_fanout_add_partition"]

    [transforms.route_fanout.route]
    output1 = '.tags.partition == "1"'
    output2 = '.tags.partition == "2"'
    output3 = '.tags.partition == "3"'
    output4 = '.tags.partition == "4"'
    output5 = '.tags.partition == "5"'

[sinks.datadog_metrics_sink_1]
type = "datadog_metrics"
inputs = ["route_fanout.output1"]
default_api_key = "${DD_API_KEY}"
batch.timeout_secs = 5
batch.max_events = 2000
buffer.max_events = 200000
buffer.when_full = "block"
request.concurrency = "adaptive"

[sinks.datadog_metrics_sink_2]
type = "datadog_metrics"
inputs = ["route_fanout.output2"]
default_api_key = "${DD_API_KEY}"
batch.timeout_secs = 5
batch.max_events = 2000
buffer.max_events = 200000
buffer.when_full = "block"
request.concurrency = "adaptive"

# ETC
'''

If it helps, this is the comment justifying this block:

# This part is a bit awesome. The issue we're trying to overcome is that Vector sinks work on a single thread only.
# It makes usage of large instances inefficient, and using smaller instances is inefficient because NLB redistributes traffic unevenly.

[jszwedko]

I think it is a bit of an anti-pattern. A single datadog_logs sink should be able to scale up to handle the full volume. It's possible that the adaptive request concurrency isn't finding the optimal concurrency (and thus causing backpressure). I'd suggest trying a single sink with statically configured concurrency. To determine a good static value you could look at the value of https://vector.dev/docs/reference/configuration/sources/internal_metrics/#adaptive_concurrency_limit to see what limit the adaptive concurrency algorithm is picking for each sink, multiply that by 5, and maybe add a little headroom.