Differential D5818 Diff 20792 swh/scheduler/cli/origin.py

Changeset View

Standalone View

swh/scheduler/cli/origin.py

Show First 20 Lines • Show All 149 Lines • ▼ Show 20 Lines	@click.option(
"--policy", "-p", default="oldest_scheduled_first", help="Scheduling policy"		"--policy", "-p", default="oldest_scheduled_first", help="Scheduling policy"
)		)
@click.option(		@click.option(
"--queue", "-q", help="Target celery queue", type=str,		"--queue", "-q", help="Target celery queue", type=str,
)		)
@click.option(		@click.option(
"--tablesample", help="Table sampling percentage", type=float,		"--tablesample", help="Table sampling percentage", type=float,
)		)
		@click.option(
		douarddaUnsubmitted Not Done Inline Actions the semantics of this flag option is not clear to me. What does `--with-enabled` means when I use this send-to-celery command? And what `--without-enabeld`? it looks to me the naming/semantics for this option is very close to the implementation, but does not make much sense for the user. douardda: the semantics of this flag option is not clear to me. What does `--with-enabled` means when I…
		ardumontAuthorUnsubmitted Done Inline Actions ok, let's go with '--only-enabled/--only-disabled'. I'll add the help message for that option which will clarify the meaning. Thanks for the heads up. ardumont: ok, let's go with '--only-enabled/--only-disabled'. I'll add the help message for that option…
		"--with-enabled/--without-enabled", "enabled", is_flag=True, default=True,
		)
		@click.option(
		"--lister-uuid",
		default=None,
		help="Limit origins to those listed from such lister",
		)
@click.argument("type", type=str)		@click.argument("type", type=str)
@click.pass_context		@click.pass_context
def send_to_celery(		def send_to_celery(
ctx, policy: str, queue: Optional[str], tablesample: Optional[float], type: str		ctx,
		policy: str,
		queue: Optional[str],
		tablesample: Optional[float],
		type: str,
		enabled: bool,
		lister_uuid: Optional[str] = None,
):		):
"""Send the next origin visits of the TYPE loader to celery, filling the queue."""		"""Send the next origin visits of the TYPE loader to celery, filling the queue."""
from kombu.utils.uuid import uuid		from kombu.utils.uuid import uuid

from swh.scheduler.celery_backend.config import app, get_available_slots		from swh.scheduler.celery_backend.config import app, get_available_slots

scheduler = ctx.obj["scheduler"]		scheduler = ctx.obj["scheduler"]

task_type = scheduler.get_task_type(f"load-{type}")		task_type = scheduler.get_task_type(f"load-{type}")

task_name = task_type["backend_name"]		task_name = task_type["backend_name"]
queue_name = queue or task_name		queue_name = queue or task_name

num_tasks = get_available_slots(app, queue_name, task_type["max_queue_length"])		num_tasks = get_available_slots(app, queue_name, task_type["max_queue_length"])

print(num_tasks, "slots available in celery queue")		print(num_tasks, "slots available in celery queue")
origins = scheduler.grab_next_visits(		origins = scheduler.grab_next_visits(
type, num_tasks, policy=policy, tablesample=tablesample		type,
		num_tasks,
		policy=policy,
		tablesample=tablesample,
		enabled=enabled,
		lister_uuid=lister_uuid,
)		)

print(len(origins), "visits to send to celery")		print(len(origins), "visits to send to celery")
for origin in origins:		for origin in origins:
		print("####", origin)
task_dict = origin.as_task_dict()		task_dict = origin.as_task_dict()
app.send_task(		app.send_task(
task_name,		task_name,
task_id=uuid(),		task_id=uuid(),
args=task_dict["arguments"]["args"],		args=task_dict["arguments"]["args"],
kwargs=task_dict["arguments"]["kwargs"],		kwargs=task_dict["arguments"]["kwargs"],
queue=queue_name,		queue=queue_name,
)		)
Show All 37 Lines