Rate Limiting

[alexcarruthers]

Are there any plans to introduce rate limiting of jobs?

[behrad]

Have you read https://github.com/LearnBoost/kue#processing-concurrency for throttling jobs?

[behrad]

in a distributed work pulling with done acknowledgement, there's no easy way to implement "n jobs done per t seconds" since you don't know when jobs will be done.
However you could use a rate limiting module in node.js (like q) to limit the injection of new jobs into your system :)

[alexcarruthers]

The current method for processing concurrency only limits the number on jobs running at the same time, it doesn't really rate limit it. What I was looking for is rate limiting the start of jobs. I could use some external module but there doesn't seem to be anything decent out there (can't see where in q it can be done) and it would be much easier if kue itself was able to provide this functionality.

[behrad]

pordon, my mistake!
You can use node-rate-limiter, here it is https://github.com/jhurliman/node-rate-limiter
I don't see a benefit in throttling at producer side, but introducing a rate-limit-able worker is something exciting :)

[alexcarruthers]

The problem with things like node-rate-limiter is that it's not distributed. If I have multiple instances of my code base running, the rate limiter is on a per-instance basis. I have yet to see one that is backed by something like redis in order to preserve the limit across all instances. I would write my own, but having it directly in Kue makes it much easier as there isn't a separate layer to configure/maintain

[behrad]

https://github.com/chaoran/node-throttle

https://github.com/brycebaril/node-tokenthrottle-redis

[jes-carr]

https://github.com/jesucarr/tokenbucket

[mansona]

Hi All,

I just want to restart this conversation about rate limiting with Kue, I am currently looking to do something quite complicated using Kue and @jesucarr's token bucket implementation and I'm wondering if anyone else has an idea of what exactly they are looking for so that we might actually collaborate and maybe get a PR together?

Is this the best place to have an architectural discussion about this?

[behrad]

you can share here what you are thinking about @mansona

[mansona]

I'll just give a quick overview of the use case for now and maybe go into more detail if it is required. There is a whole structure as to why I want to do things this way so if you feel like it is useful for me to give more context just let me know. (It might be a very long post so be warned 😉 )

I am currently working with the Twitter API and I want to push as much throughput with a single job processor as possible. You can see on this page that you are limited to 180 users/lookup calls every 15 mins. Each one of those calls has the ability of batching 100 lookups so theoretically you can query up to 18000 every 15 mins.

The way that my jobs are going to be laid out is that I am going to be requesting between 12 and 20 profiles at a time (maximum) which means that I'm only going to be able to batch 12 profiles together (worst case) if I limit my "batching" to 1 batch per job. This reduces my potential throughput to 2160, which is a meagre 12% of the theoretical maximum.

The idea that I'm coming up with is that I want my original job, lets call it getTwelveProfiles, doesn't actually do the profile request directly. Instead it in turn batches the jobs in some way and this is where the TokenBucket algorithm comes in.

For the example below to work i am assuming that each getTwelveProfiles call actually creates 12 new lookupProfile jobs each with the profileId that it needs to lookup

My original thought would be for us to integrate @jesucarr's token bucket repo directly into Kue. I know this would be quite a difficult thing to do but I think it would provide such an interesting API that it might just be worth it. It might even offer a different solution to #493 as a side affect.

The idea would be to provide an api something like this for the worker:

var tokenBucket = new TokenBucket({
  size: 18000,
  tokensLeft: 0,
  tokensToAddPerInterval: 1200,
  interval: 'minute'
});

queue.tokenBucketProcess('lookupProfile', tokenBucket, 100,  function(job, done){
  // job.data would be aggregated in some way so that individual profileIds are aggregated together into an array
  // Object.prototype.toString.call(job.data.profileId) === "[object array]"
  batchLookupProfiles(job.data.profileId, done);
});

Internally queue.tokenBucktProcess() would call tokenBucket.removeTokens(100) and only actually call the process function once there are enough tokens available.

This implementation would also allow for you to start multiple processors with different tokenBuckets, in my use case that makes sense because Twitter limits request on a per user basis. So if I had more users and access tokens I could get more throughput through the system by just adding another worker with a new access token.

I hope that explains the use case, please let me know if you have any questions.

I have a need to implement something along these lines in the coming days so if it is something that you are interested in adding to the project I could work on a fork with the hope of creating a PR to add the functionality. I would probably need some guidance a) to know if this is just a stupid idea or not 😖 and b) if I were to implement this what the best strategy might be.

[behrad]

At very first I think,

1. The rate-limiter/token-bucket implementation should work in distributed environment.

2. How a rate-limited job worker/puller can be introduced into the current implementation?

Optimistic method is: after worker is BLPOPed, and before it fetches new job, Then worker should return swallowed item into helper list, and prevent from fetching the job...

another method is to always check if tokens are available and then wait on BLPOP and fetch job ignoring a double check on token bucket. After current job is finished again check for tokens...

[Niels-Be]

I have a similar problem. Limiting API requests per second.
What I do is, use kue and just repeat the job until following redis script code gives me an OK

local i = redis.call('LLEN', KEYS[1])
if i < tonumber(ARGV[1]) then
    if redis.call('EXISTS', KEYS[1]) == 0 then
        redis.call('RPUSH', KEYS[1], '1') 
        redis.call('EXPIRE', KEYS[1],  tonumber(ARGV[2]) )
    else
        redis.call('RPUSHX', KEYS[1], '1') 
    end
    return 'OK'
else
    return 'FULL' 
end

Call this in nodejs with:

redisClient.eval([theAboveCode, 1, "activeApiConnections", maxApiConnections, timeintervalInSec], function(err, res) {
    if(res === 'OK') {
        //make the request
    } else {
        //restart the job
    }
});

This works in distributed environments, because redis script is guaranteed to be executed atomically.
But kind of sucks because I always have to 'ask' redis and then restart the job.

I am not familiar with the kue implementation but you should be able to call this just before a worker pulls a job.

[joshbedo]

So has anyone found out a way to limit x jobs until completed and then run the next set of jobs? For example you have 6 orders go through and nobody else can process anymore orders until one of the 6 are done.

[mansona]

Hi Folks 👋

I just wanted to give you an indication on how i got this working in the end. This is not ideal but it gives a good indication of what I'm trying to achieve.

var jobTokenBucket = new TokenBucket({
    size: 180,
    tokensLeft: 0,
    tokensToAddPerInterval: 12,
    interval: 'minute'
});

queue.process('myTwitterRateLimitedJob', function(job, done) {
    return jobTokenBucket.removeTokens(1).then(function() {
        return helpers.getStuffFromTwitter();
    }).nodeify(done);
});

So how this works is it doesn't request anything from twitter until there are the required tokens in the bucket, the problem with this is that the job becomes active before it has the right number of tokes it is requesting.

The ideal solution to integrate these two methods would see the job not becoming active until there are enough tokens in the bucket.

Does that make sense or do you need me to explain it any better?

[tarraq]

Some thoughts on an old thread:
How about a simpler dual queue setup?
If you have one queue (backlog) where you dump all the work you need done, and then a simple consumer taking jobs from the backlog and move them into a processing queue where they are then processed by whatever you need. That way you can easily rate limit the processing:

// pseudo code presuming a 120 request per minute limit on external service

//Worker 1 processing 2 per second with setInterval as per rate limit:
backlog.process('genericJob', function({jobType: "twitterCall", data: {actualData: "here"}}, done){
  sendToQueue2(job.data.jobType, job.data.data, done); // creates job in processing queue
});

//Worker 2 Processing the processing queue as normal, but will often get empty queue, because of the rate limiting, but is otherwise oblivious to it.
processQueue.process("twitterCall",function(job, done){
  callToTwitter(job.data.actualData, done);
});

This solution will work in large distributed setups and can be scaled to any amount of processing workers (but there should only be one backlog worker, or carefully calculating shared rate among more of them.

Just my 2 cents.

/ Michael

EDIT: Updated the pseudo-code to be a little less pseudo.

[behrad]

Nice trick @tarraq , Layers always do the job ;)

[tarraq]

Thanks @behrad. I haven't tried it out in practice, it's just a mind trick as it were :)
But it should work quite reliably, providing that the source of the calls are not flooding the queue and the rate limited workers can't keep up.

On that note, been wondering, what's the maximum queue length in kue? 2 to the power of 32 - 1, the Redis list limit?

[behrad]

On that note, been wondering, what's the maximum queue length in kue? 2 to the power of 32 - 1, the Redis list limit?

Actually Redis ZSETs limit, and in real the MEMORY before that :)

[tarraq]

On that note, been wondering, what's the maximum queue length in kue? 2 to the power of 32 - 1, the >>Redis list limit?
Actually Redis ZSETs limit, and in real the MEMORY before that :)

That should be alright then ;)

[mansona]

Hmm... the problem that I have with this approach is that it doesn't really deal with the fact that things like Twitter allow 180 requests per 15 minutes per set of credentials.

The way that works with the Token Bucket example above is that you would create instances of processors that have unique credentials and their own token bucket. That way you are not limiting throughput of an "interim queue" you are more closely limiting throughput of how many jobs a processor can actually process per minute.

I hope this makes sense. I can probably explain it better with a graph if needs be?

[snig-b]

Looking forward to a rate limit feature in kue. I really like all the examples and discussion going on here. Just hoping something would be integrated into kue for a consistent solution.. Any updates on this?

[behrad]

I believe best place to solve API/request rate limiting are API gateways. When we say rate-limiting, it means overflow requests are denied/rejected. So any (distributed) rate-limiter can be placed in front of Kue.
Kue is a job queue with a little, efficient priority task queueing + per node concurrency control.
For whom they need a time-based job processing control, this issue may leave opened, so that we can argue if this is a valuable feature or not!

[snig-b]

That's correct, I'm anyways looking for a time based option only - like celery's rate limit. I am moving from a django-celery combination to a node-kue one.
So far going good, but found nothing on rate-limit here. Would be great if we can come up with something like that.

[markstos]

I'm also interested in this feature. My use-case is sending emails to the Amazon SES API from dozens of Node.js processes across several servers. I would like to pass all the email sending through a single job queue which make farmers out the sending to workers and makes sure that we don't exceed the prescribed number of messages/second, while still sending as fast as possible.

[sssahas]

+1. Github APIs have rate limiting (5000/hr) + abuse rate limits on top of this. Having a rate_limit option will be helpful.

[bobmoff]

+1

[karthikrock]

+1

[bobmoff]

I added https://www.npmjs.com/package/limiter and it works great

[Niels-Be]

For an example implementation you could have a look at Josiah Carlson post about Rate limiting with Redis. He describes a very robust implementation that also supports multiple buckets.

[maxloginov]

I've been having a problem recently with limiting jobs rate globally, depending on the memory usage. It was an urgent issue, so, I've implemented a really simple patch that does the trick, and we're using it now on our production server: max-loginov@112b094

I went through this discussion and realized that my solution can probably fit here, as well. It allows rate-limiting of jobs either globally of job type-wise, using any imaginable condition. What do you think guys?

[maxloginov]

Created a pull request for this feature. #1103