Просмотр исходного кода

perf: distribute concurrent plugin auto upgrade tasks (#26282)

tags/1.9.1
Junyan Qin (Chin) 1 месяц назад
Родитель
Сommit
043ec46c33
Аккаунт пользователя с таким Email не найден
1 измененных файлов: 24 добавлений и 9 удалений
  1. 24
    9
      api/schedule/check_upgradable_plugin_task.py

+ 24
- 9
api/schedule/check_upgradable_plugin_task.py Просмотреть файл

@@ -1,3 +1,4 @@
import math
import time

import click
@@ -8,6 +9,7 @@ from models.account import TenantPluginAutoUpgradeStrategy
from tasks.process_tenant_plugin_autoupgrade_check_task import process_tenant_plugin_autoupgrade_check_task

AUTO_UPGRADE_MINIMAL_CHECKING_INTERVAL = 15 * 60 # 15 minutes
MAX_CONCURRENT_CHECK_TASKS = 20


@app.celery.task(queue="plugin")
@@ -30,15 +32,28 @@ def check_upgradable_plugin_task():
.all()
)

for strategy in strategies:
process_tenant_plugin_autoupgrade_check_task.delay(
strategy.tenant_id,
strategy.strategy_setting,
strategy.upgrade_time_of_day,
strategy.upgrade_mode,
strategy.exclude_plugins,
strategy.include_plugins,
)
total_strategies = len(strategies)
click.echo(click.style(f"Total strategies: {total_strategies}", fg="green"))

batch_chunk_count = math.ceil(
total_strategies / MAX_CONCURRENT_CHECK_TASKS
) # make sure all strategies are checked in this interval
batch_interval_time = (AUTO_UPGRADE_MINIMAL_CHECKING_INTERVAL / batch_chunk_count) if batch_chunk_count > 0 else 0

for i in range(0, total_strategies, MAX_CONCURRENT_CHECK_TASKS):
batch_strategies = strategies[i : i + MAX_CONCURRENT_CHECK_TASKS]
for strategy in batch_strategies:
process_tenant_plugin_autoupgrade_check_task.delay(
strategy.tenant_id,
strategy.strategy_setting,
strategy.upgrade_time_of_day,
strategy.upgrade_mode,
strategy.exclude_plugins,
strategy.include_plugins,
)

if batch_interval_time > 0.0001: # if lower than 1ms, skip
time.sleep(batch_interval_time)

end_at = time.perf_counter()
click.echo(

Загрузка…
Отмена
Сохранить