Total running Jobs must not exceed maxScale - Running jobs

[balchua]

This PR is to honor the queueLength taking into account the number of running Jobs.
In short, total running Jobs will be queueLength - runningJobs.

Running jobs are those jobs which are not in Completed or Failed Status.
This should address issue #525

[zroubalik]

LGTM, thanks @balchua !

[Cottonglow]

Just tested this with a Kafka scaler and seems to work fine!

One thing I've noticed is that the scaler will create a job per message even if there are jobs running already. This isn't something this PR introduced but it is something this PR can help fix.

It might be worth adding to the code, to only create the necessary number of jobs as there are messages in the queue, otherwise you might end up with 5 jobs for 1 message.

[balchua]

@Cottonglow by setting queueLength to 1 prevents it from having more running jobs than the number of messages.

[balchua]

I realized that you are using kafka scaler, i wonder how it works for kafka, since the scaler is only checking for metrics with queueLength and kafka scaler doesn't use queueLength.

[stgricci]

Looks good! Thanks @balchua!

[Cottonglow]

@balchua
Kafka uses lagThreshold which from my understanding is the threshold for the number of messages on the queue that hasn't been consumed yet before kicking off the scaling.
Looking at the code, it looks like handleScaleJob treats lagThreshold the same as queueLength. (Which might need to be considered in the future for Kafka)

I've experimented by putting the value of lagThreshold to 1 and it seems to only create 1 job regardless of the number of messages on my queue. I'm not sure what behaviour you're seeing with your scaler?

If the idea is to have 1 job per message (which is what I understand the plan is from the docs), then this wouldn't work. Unless this is a problem with the Kafka scaler only.

This can be a consideration for a future PR though if it is too much work as this does fix the problem of the number of jobs exceeding the max value which is very useful now.

[balchua]

@Cottonglow i was using rabbitmq scaler.

The code in

keda/pkg/handler/scale_loop.go

Lines 82 to 85 in 8658859

	if m.MetricName == "queueLength" {
	metricValue, _ = m.Value.AsInt64()
	queueLength += metricValue
	}

Only considers metricName of queuelength. So i was just wondering how it worked for kafka scaler.

[balchua]

Forget about my previous comment 😁. I got it figured out. handleScaleJob sort of "forces" the metric name to be queueLength as @Cottonglow mentioned.

[balchua]

Well it is still going to start x number of jobs per available message in the "queue" limited by the trigger's metric queueLength or lagThreshold. This helps to avoid spawning new jobs when previous jobs are long running and before the messages are dequeued.

[jeffhollan]

@ahmelsayed is this one ready to merge?

[balchua]

@jeffhollan, i have yet to work on the review given above. 😁

[balchua]

@stgricci i have updated the code to make sure maxScale is not negative. thanks for the review.

[balchua]

@ahmelsayed i think this is good to go. @stgricci gave his thumbs up. 😁