@@ -178,7 +178,7 @@ def compute_queue_load_factor():
178178 """
179179 field = models .PipelineSchedule ._meta .get_field ("run_priority" )
180180 label_to_value = {label : value for value , label in field .choices }
181- total_compute_seconds_per_queue = {}
181+ total_compute_seconds_per_queue_in_24hr_cycle = {}
182182 worker_per_queue = {}
183183 load_per_queue = {}
184184 seconds_in_24_hr = 86400
@@ -191,26 +191,39 @@ def compute_queue_load_factor():
191191 worker_per_queue = dict (Counter (queue_names ))
192192
193193 for queue in RQ_QUEUES .keys ():
194- total_compute_seconds_per_queue [queue ] = sum (
195- (p .latest_successful_run .runtime / (p .run_interval / (24 * 60 )))
196- for p in models .PipelineSchedule .objects .filter (
197- is_active = True , run_priority = label_to_value [queue ]
198- )
199- if p .latest_successful_run
194+ total_compute_seconds = 0
195+ active_schedules = models .PipelineSchedule .objects .filter (
196+ is_active = True ,
197+ run_priority = label_to_value [queue ],
200198 )
199+ for schedule in active_schedules :
200+ if not schedule .latest_successful_run :
201+ continue
202+
203+ runs_per_day = (24 * 60 ) / schedule .run_interval
204+ compute_seconds_per_day = schedule .latest_successful_run .runtime * runs_per_day
205+ total_compute_seconds += compute_seconds_per_day
206+
207+ total_compute_seconds_per_queue_in_24hr_cycle [queue ] = total_compute_seconds
208+
201209 if queue not in worker_per_queue :
202210 worker_per_queue [queue ] = 0
203211
204212 for queue_name , worker_count in worker_per_queue .items ():
205213 net_load_on_queue = "no_worker"
206- total_compute = total_compute_seconds_per_queue .get (queue_name , 0 )
214+ total_compute = total_compute_seconds_per_queue_in_24hr_cycle .get (queue_name , 0 )
215+ total_pipeline_for_queue = models .PipelineSchedule .objects .filter (
216+ is_active = True , run_priority = label_to_value [queue_name ]
217+ ).count ()
207218 if total_compute == 0 :
208219 continue
209220
210221 unit_load_on_queue = total_compute / seconds_in_24_hr
211222
212223 num_of_worker_for_balanced_queue = round (unit_load_on_queue )
213- addition_worker_needed = max (num_of_worker_for_balanced_queue - worker_count , 0 )
224+ addition_worker_needed = min (
225+ max (num_of_worker_for_balanced_queue - worker_count , 0 ), total_pipeline_for_queue
226+ )
214227
215228 if worker_count > 0 :
216229 net_load_on_queue = unit_load_on_queue / worker_count
0 commit comments