distributor.distributor_service

Attributes

logger

Classes

DistributorService

Module Contents

distributor.distributor_service.logger

class distributor.distributor_service.DistributorService(cluster_interface: jobmon.core.cluster_protocol.ClusterDistributor, requester: jobmon.core.requester.Requester | None = None, workflow_run_heartbeat_interval: int | None = None, task_instance_heartbeat_interval: int | None = None, heartbeat_report_by_buffer: float | None = None, distributor_poll_interval: int | None = None, raise_on_error: bool = False)

config

raise_on_error

_task_instances: Dict[int, jobmon.distributor.distributor_task_instance.DistributorTaskInstance]

_task_instance_batches: Dict[Tuple[int, int], jobmon.distributor.task_instance_batch.TaskInstanceBatch]

_distributor_commands: Iterator[jobmon.distributor.distributor_command.DistributorCommand]

_task_instance_status_map: Dict[str, Set[jobmon.distributor.distributor_task_instance.DistributorTaskInstance]]

gen_map: Dict[str, Callable[Ellipsis, Generator[jobmon.distributor.distributor_command.DistributorCommand, None, None]]]

_command_generator_map

_last_heartbeat_time

cluster_interface

property _next_report_increment: float

set_workflow_run(workflow_run_id: int) → None

run() → None

process_status(status: str, timeout: int | float = -1) → None

Processes commands until all work is done or timeout is reached.

Parameters:

• status – which status to process work for.

• timeout – time until we stop processing. -1 means process till no more work

instantiate_task_instances(task_instances: List[jobmon.distributor.distributor_task_instance.DistributorTaskInstance]) → None

launch_task_instance_batch(task_instance_batch: jobmon.distributor.task_instance_batch.TaskInstanceBatch) → None

launch_task_instance(task_instance: jobmon.distributor.distributor_task_instance.DistributorTaskInstance) → None

Submits a task instance on a given distributor.

Adds the new task instance to self.submitted_or_running_task_instances.

triage_error(task_instance: jobmon.distributor.distributor_task_instance.DistributorTaskInstance) → None

Triage a running task instance that has missed a heartbeat.

Allowed transitions are (R, U, Z, F)

kill_self(task_instance: jobmon.distributor.distributor_task_instance.DistributorTaskInstance) → None

Terminate a task instance that has received a Kill Self signal.

This signal is sent from a cold workflow resume, and transitions the task instance to an ERROR_FATAL state with no retries.

no_heartbeat_error(task_instance: jobmon.distributor.distributor_task_instance.DistributorTaskInstance) → None

Move a task instance in NO_HEARTBEAT state to a recoverable error state.

This signal is sent from the swarm in the event a task instance in LAUNCHED state fails to log a heartbeat, either due to the distributor failing to log a heartbeat batch or due to the worker node failing to start up properly.

ERROR state allows for a retry, so that a new task instance can attempt to run.

log_task_instance_report_by_date() → None

Log the heartbeat to show that the task instance is still alive.

async _log_heartbeats(task_instance_batches: List[List[int]]) → None

Create a task for each batch of task instances to send heartbeat.

async _log_heartbeat_by_batch(session: aiohttp.ClientSession, task_instance_ids_to_heartbeat: List[int]) → None

Send heartbeat for a batch of task instances.

_initialize_signal_handlers() → None

refresh_status_from_db(status: str) → None

Got to DB to check the list tis status.

_check_queued_for_work() → Generator[jobmon.distributor.distributor_command.DistributorCommand, None, None]

_check_instantiated_for_work() → Generator[jobmon.distributor.distributor_command.DistributorCommand, None, None]

_check_triaging_for_work() → Generator[jobmon.distributor.distributor_command.DistributorCommand, None, None]

Handle TIs in TRIAGING state.

For TaskInstances with TRIAGING status, check the nature of no heartbeat, and change the statuses accordingly.

_check_kill_self_for_work() → Generator[jobmon.distributor.distributor_command.DistributorCommand, None, None]

Handle TIs in KILL_SELF state.

For TaskInstances with KILL_SELF status, terminate it and transition it to error accordingly.

_check_no_heartbeat_for_work() → Generator[jobmon.distributor.distributor_command.DistributorCommand, None, None]

Handle TIs in NO_HEARTBEAT state.

For TaskInstances with NO_HEARTBEAT status, move to an error recoverable state