Recommended approach to add custom code in relation to concurrency

[karmingc]

Hi,

We currently use Shoryuken to process SQS messages with the following sample configuration:

groups:
  group_a:
    concurrency: 2
    queues:
      - "some_queue"
  group_b:
    concurrency: 2
    queues:
      - "some_queue_b"

As part of our processing/worker logic, we record metrics and they are then sent to Grafana. However, with the current configuration, there are certain occasions when metrics are recorded with the same labels/attributes but with a different timestamp (due to concurrency).

We also use Puma to run our Rails application server, and we were able to resolve this by appending the worker pid as an attribute through the on_worker_boot hook. We are wondering what would be the recommended approach for Shoryuken such that we can patch each thread processing a SQS message?

[mensfeld]

Can you provide me a POC? I am the new maintainer and I am learning as I go, thus I need to be able to repro this.

[karmingc]

hi @mensfeld,

please find the attached sample rails app: https://github.com/karmingc/sample-metrics-app
in parallel, you'll need to run the OTel collector, https://github.com/karmingc/playground/tree/main/dogstatsd-to-prom. For this, run docker compose up to run the Collector in a docker container.

edit: note that I'm seeing issues when it comes to processing the sqs message. It doesn't seem to be running the perform from either worker, but I can enqueue using HelloWorker.perform_async("hey"). Can you confirm? it was an issue with aws-sdk-core

[karmingc]

@mensfeld just to confirm, shoryuken doesn't create more than 1 process correct? I was able to test using my sample app and metrics were indeed aggregated assuming they were within the same process, regardless of the thread.

[mensfeld]

@karmingc yes it does not have fork capabilities (yet).