Throttling with Sidekiq Pro's super_fetch

[Blokh]

Greetings, we've been using Sidekiq-Throttle gem for a while and it seems to work perfectly. Lately we've implemented sidekiq pros's super_fetch.

For some reason, after this change sidekiq-throttling seems to be broken.

Do you know by any chance to fix this issue? Could you provide me some guidance of how to implement sidekiq-throttling in a way that would not affect the fetch strategy?

Thanks in advance, Daniel

[holstvoogd]

We are using super_fetch icm sidekiq_throttled and it seems to work as expected. Perhaps the initializer order matters? This how I init relevant parts of the server:

initializer 'sidekiq-server' do |app|
  Sidekiq.configure_server do |config|
    config.super_fetch!
  end
  Sidekiq::Throttled.setup!
end

We are on sidekiq 6.2.1.

[exAspArk]

From our observation, you can either use sidekiq_throttled or super_fetch, but not both, unfortunately.

To prevent any problems, add the .setup! call to the bottom of your init file.

^ If you follow this instruction from the readme or essentially do what @holstvoogd suggested, sidekiq_throttled is going to work. However, it'll override Sidekiq's custom fetch from Sidekiq::Pro::SuperFetch to Sidekiq::Throttled::Fetch.

[holstvoogd]

However, it'll override Sidekiq's custom fetch from Sidekiq::Pro::SuperFetch to Sidekiq::Throttled::Fetch.

We've since learned this too, but at first glance it looks like it should be possible to 'wrap' the original fetcher, albeit the normal or super one, with the throttling. I'm hoping to look into that at some point and make a PR, but that might take a while...

[ixti]

I'll be happy to accept the PR. I was thinking to make fetch more composable to make it possible to easily hook into any other one.

[holstvoogd]

@ixti I see you've setup a new repo for sidekick-throttled? Shall I make a PR there? It might also be a good idea to put a little notice on this repo of the move? I hadn't noticed until I thought I was done 😅

I think I have a pretty decent solution that wraps the orignal fetcher, works with both SuperFetch or BasicFetch. It does require a few patches to BasicFetch, depending on the version of sidekiq, to make everything compatible.

Setup is quite straight forward, it just picks up the current fetcher by default. I think that leads to less suprises then having to pass it to #setup! manually, especially as under normal circumstances Sidekiq.options[:fetch] is nil init. I wouldn't want to break the default flow.

I've also made it possible to disable pausing/resuming queues & the enhanched queues tab; as this features is also provided by sk pro and having both active seems like trouble waiting to happen. Although it should still work if you really want to pause a queue twice ;)

I am not sure what to do for testing with Pro though. I have confirmed it seems to work properly with sidekiq pro in a project, but can't really figure out a way to run spec agains pro. I can add apraisals for it, but you'd have to have a compatible version of sidekiq-pro installed already. And I have not been able to get SuperFetch to properly start in that context.

I could make some tests that use a mock and not actually hit sidekiq, but is that testing anything not covered by the existing tests? Main thing is that the SuperFetch 'api' needs to be stable, can't really test that with out pulling it in..

[ixti]

@holstvoogd I've lost any control over this repo, so I can't change README as well. :D Please open PR in the new repo – I'll be happy to accept the changes!

Re configurability. Thanks for doing that. I was considering extracting the pausing feature away from this gem, as it's unrelated. As well as Communicator class, I think I made a mistake by introducing it in a first place. :D

On testing, I think we can mark tests as "require sidekiq-pro" and have extra appraisal for those. Thus, it will be possible to run test-suite against sidekiq-pro locally.

[holstvoogd]

On testing, I think we can mark tests as "require sidekiq-pro" and have extra appraisal for those.

I'll see if I can get that to work 👍

[legendetm]

What is the status of the fix and is there any estimates when this PR may merged and released?

[legendetm]

@ixti or @holstvoogd any progress with this?

[ixti]

I'm finishing preparation of v1.0.0, which should make this possible. The refactored fetcher class should become compatible with super-fetch out of the box (but I will need volounteers to check that). :D

[ixti]

Just released v1.0.0.alpha that should make sidekiq-throttled compatible with super fetch. But as I don't know hte details of sidekiq-pro, I can't test it myself.

[Parth-Rewind]

@ixti What do you think of this? Am I making a mistake by configuring super_fetch! explicitly like so along with Sidekiq::Throttled.setup! Here's a sample config that I have sidekiq.rb

require 'sidekiq/throttled'

Sidekiq.configure_server do |config|
  config.super_fetch!
end

Sidekiq::Throttled.setup!

Sidekiq::Throttled::Registry.add(....)

And here's the error I am getting when deploying

48 | on(:heartbeat) do
49 |   f.register_myself
50 | end

NoMethodError: undefined method `register_myself' for #<Sidekiq::Throttled::BasicFetch:0x00007fd087736a90...
/gems/sidekiq-pro-7.1.2/lib/sidekiq-pro.rb:49:in `block (3 levels) in super_fetch!'
/gems/sidekiq-7.1.1/lib/sidekiq/component.rb:60:in `block in fire_event'
/gems/sidekiq-7.1.1/lib/sidekiq/component.rb:59:in `each'
/gems/sidekiq-7.1.1/lib/sidekiq/component.rb:59:in `fire_event'
/gems/sidekiq-7.1.1/lib/sidekiq/launcher.rb:181:in `❤'
/gems/sidekiq-7.1.1/lib/sidekiq/launcher.rb:98:in `beat'

[ixti]

Hmm. I don't really know what is going inside of the register_myself. Without knowing what that method should do, the way I see, would be to provide a mixin for original basic fetch class that would simply overload some methods that I know to exist in the original class

[Parth-Rewind]

I don't really know what is going inside of the register_myself.

For each capsules, the register_myself will register the capsules as super_processes

[legendetm]

This seems to work with Sidekiq 6 Pro. With the v1.0.0.alpha version pause/unpause works and concurrency is also handled correctly.

The capsules where introduced in Sidekiq 7.

[ixti]

For each capsules, the register_myself will register the capsules as super_processes

Hmm. So, is should be something like?:

def register_myself
  @config.capsules.each_value do |capsule|
    capsule.super_process(self)
  end
end

FWIW I think the best way forward will be to simply provide something like:

Sidekiq::Throttled::Patches::BasicFetch.patch!(Sidekiq::BasicFetch)

That would prepend needed fixes to retrieve_work. Will add such one this week.

[ixti]

I'm gonna prepare a fix over the next couple of days, but I will need somebody to test it :D

[ixti]

I've changed approach of inheriting from Sidekiq::BasicFetch to directly patching its #retrieve_work that should make throttling work on both Sidekiq and Sidekiq::Pro seamingly.

[ixti]

Please give 1.0.0.alpha.1 a try.

[Parth-Rewind]

For 1.0.0.alpha.1, I can enable the super_fetch without sidekiq erroring out, however, the throttling that I had in place with 1.0.0.alpha doesn't seem to work. Workers make calls as fast as possible without respecting the concurrency limit.

Update: If i disable super_fetch! the throttling works fine with 1.0.0.alpha.1. Once super_fetch! is enabled with 1.0.0.alpha.1, the throttling won't work (There would be no 'throttled' key in redis)

[ixti]

@Parth-Rewind Can you check which fetch class is being used once super_fetch is required and what are the ancestors of the class?

[Parth-Rewind]

@Parth-Rewind Can you check which fetch class is being used once super_fetch is required and what are the ancestors of the class?

With 1.0.0.alpha.1, the fetch class will be set to :fetch_class => Sidekiq::Pro::SuperFetch < Object

Doing

Sidekiq::BasicFetch.ancestors

Gives me this

[
  Sidekiq::Throttled::Patches::BasicFetch,
  Sidekiq::BasicFetch,
  Sidekiq::Component,
  ActiveSupport::Dependencies::RequireDependency,
  ActiveSupport::ToJsonWithActiveSupportEncoder,
  Object, ActiveSupport::Tryable,
  JSON::Ext::Generator::GeneratorMethods::Object,
  PP::ObjectMixin,
  Debase::PrivateMultiProcess,
  Debase::MultiProcess,
  Kernel,
  BasicObject
]

---

My config file looks like so:

require 'sidekiq/throttled'

Sidekiq.configure_server do |config|
  config.super_fetch!
end

Sidekiq::Throttled.setup!

Sidekiq::Throttled::Registry.add(....)

[ixti]

Oh. That explains. Okay will prepare a fix shortly.

[Parth-Rewind]

Oh. That explains. Okay will prepare a fix shortly.

Thank You!!

When you have time, could you please tell me what was causing this? I am curious. Is that the fetch_class is not being set to Sidekiq::Throttled::Patches::BasicFetch

[ixti]

The fetch class after config.super_fetch! is set to Sidekiq::Pro::SuperFetch. Which does not inherit the behaviour of Sidekiq::BasicFetch. I'm gonna change how fetch class is being patched, to correctly patch super fetch.

[srinu-adari]

Thank you very much @ixti so far for actively working on fixing this issue. Just wanted to know, when is it planned to move these fixes in alpha to a stable release branch?

[ixti]

After trying to get my head around Sidekiq-Pro, I realize I need somebody with sidekiq-pro to answer a couple of questions. Otherwise, I'm acting in a dark place with my eyes shut, trying to guess. And I have a hunch that my guess will lead to some breakage without knowing the expected API contract:

• How to notify super-fetch that job was pushed back?
• How does retrieve_work gets list of queues to poll?
  • Is it calling queues_cmd?
  • If so, is the expected response is just queues list?

[mnovelo]

I can try to answer your questions about Sidekiq-Pro @ixti! We really appreciate all the hard work you're putting into this

• How to notify super-fetch that job was pushed back?

Sidekiq::Pro::SuperFetch defines Sidekiq::Pro::SuperFetch::UnitOfWork which is similar to Sidekiq::BasicFetch::UnitOfWork and both have a requeue method. The SuperFetch version uses Lua to requeue the job. The current thread's unit of work is made available to Sidekiq::Batch as Thread.current["sfuow"]

• How does retrieve_work gets list of queues to poll?
  • Is it calling queues_cmd?

In Sidekiq-Pro 7.1.3,

• Sidekiq::Pro::BasicFetch
  • Is prepended onto Sidekiq::BasicFetch
  • Overrides queues_cmd to check if any queues are paused (so they can be skipped) before calling super
• Sidekiq::Pro::SuperFetch
  • Completely independent from either BasicFetch implementation
  • Does not implement queues_cmd and instead has its own implementation of retrieve_work, but it relies on a public method called queues which just gets the queues from the capsule

[legendetm]

Is there any progress with this?

[legendetm]

Can you link the pull request with this issue? It easier to give feedback or contribute if can see the current implementation.