How are threads configurations applied in streams mode?

[ryanpodonnell1]

We're currently piloting benthos and keep getting OOM'd on ECS tasks with 8GB memory per container (1-2 containers per host at 8 vcpus).

We are currently using the following processors (just listing them not a config file)

input:
    - aws_s3
processors:
  - mutation
  - mapping
output:
   - custom plugin

Is there a best practice around using streams mode?

Were really only using streams mode to have it load multiple config files (right now were at about 5 config files or so with 1input/1output) and not leveraging the API at this point. I'm thinking its prob better to have a dedicated container per SQS queue? Any suggestions to pinpoint culprits or debug the issue?

we've applied GOMEMLIMITas an env var to about 250MB less than the hard limit of the container and that has limited some of the bleeding but still getting random crashes

my current theory is that its attempting to load everything and not "balancing" the thread count across all the streams (i.e. each stream is using -1 as its value and defaulting to the cpu count to the task for ALL of the streams) instead of using a shared pool. Could be totally wrong on this though without any proof to back it up

[mihaitodor]

Hey @ryanpodonnell1 👋

Is there a best practice around using streams mode?

Test each individual stream to see what is its max memory usage can be and then reserve enough memory for Benthos to accommodate the sum of the max memory usage of each stream, because, worst case, you may have all the streams reach their peak memory consumption simultaneously. For memory intensive workloads, it may be easier to stick with one pipeline per Benthos instance instead of running it in Streams Mode.

we've applied GOMEMLIMIT as an env var to about 250MB less than the hard limit of the container and that has limited some of the bleeding but still getting random crashes

I think it's good practice to set that variable in production like you did, but if your streams need to allocate more memory than what's available, you'll get an OOM.

my current theory is that its attempting to load everything and not "balancing" the thread count across all the streams (i.e. each stream is using -1 as its value and defaulting to the cpu count to the task for ALL of the streams) instead of using a shared pool.

The threads field from the pipeline section of the config is not shared across streams. Think of streams as fully independent from each other (although they can share resources such as caches). If you leave the threads field to its default value, which is -1, then Benthos will automatically match the number of logical CPUs available to run the pipeline section in parallel for each stream (I haven't checked the code, but I'm fairly sure this is the case).

I'm thinking its prob better to have a dedicated container per SQS queue?

Possibly. Benthos processes are cheap. Also, note that you can also use the broker input to read messages from multiple inputs and you can set a metadata field for each of them to indicate the source of the message if you need to process messages differently for each input. In this case, the pipeline section of the config will be shared across all inputs.

Maybe start by setting threads: 1 and, depending on your custom output, if it's a batched output, then set max_in_flight: 1 to specify that only one batch is allowed to be in flight at any given time. The output will apply back pressure to the input, so it won't consume more messages if it's still busy writing data. Now you can gradually increase these values (including the batch size) and watch the memory usage and other process-specific and Go-specific stuff via metrics (see add_process_metrics and add_go_metrics) in a synthetic test. It won't be trivial to find the optimal settings, especially if you have multiple streams running at the same time with different workloads.

[ryanpodonnell1]

Awesome thank you for the detailed guidance!