How to follow, stop and time-limit long-running tasks

nr.run() returns when every host has finished. For long operations, such as a software upgrade, an image copy or a commit-confirm, that is often not enough: you want to see each host’s progress while it runs, stop the hosts that are still running when one host fails, and give each host a deadline.

This guide uses ThreadedRunner. The same calls work on AsyncioRunner with asubmit and astream, and stream works on SerialRunner. The tasks below sleep instead of talking to devices, so the guide runs anywhere.

Set up

The configuration selects ThreadedRunner with one worker, so that the output below always comes in the same order. Paths are relative to the notebook working directory, docs/howto/.

import time

from nornir import InitNornir
from nornir.core.exceptions import TaskCancelledError, TaskTimeoutError
from nornir.core.run import RunOptions
from nornir.core.task import PartialResult, Result, Task
from nornir.plugins.runners import ThreadedRunner

nr = InitNornir(config_file="long_running_tasks/config.yaml")
print(f"Runner: {type(nr.runner).__name__}, hosts: {', '.join(nr.inventory.hosts)}")

Runner: ThreadedRunner, hosts: edge-1, edge-2, edge-3, edge-4

Report progress from a task

A task reports progress by yielding PartialResult objects. Its return value is its final result, as for any other task. A partial result never marks a host as failed and is not stored in the host’s results.

nr.stream() runs the task and yields each event as it happens: the partial results, then the final Result of each host.

def upgrade(task: Task, image: str):
    yield PartialResult(host=task.host, result=f"copying {image}", progress=0.3)
    time.sleep(0.01)  # stands in for the image copy
    yield PartialResult(host=task.host, result="rebooting", progress=0.8)
    time.sleep(0.01)  # stands in for the reboot
    return Result(host=task.host, result=f"running {image}", changed=True)


for event in nr.stream(task=upgrade, image="eos-4.32.bin"):
    if isinstance(event, PartialResult):
        print(f"{event.host.name}: {event.result} ({event.progress:.0%})")
    else:
        print(f"{event.host.name}: done, {event.result}")

edge-1: copying eos-4.32.bin (30%)
edge-1: rebooting (80%)
edge-1: done, running eos-4.32.bin
edge-2: copying eos-4.32.bin (30%)
edge-2: rebooting (80%)
edge-2: done, running eos-4.32.bin
edge-3: copying eos-4.32.bin (30%)
edge-3: rebooting (80%)
edge-3: done, running eos-4.32.bin
edge-4: copying eos-4.32.bin (30%)
edge-4: rebooting (80%)
edge-4: done, running eos-4.32.bin

nr.run() runs the same task and returns the usual AggregatedResult. The partial results still reach processors that define task_instance_progress, so a progress display can live in a processor:

class PrintProgress:
    def task_instance_progress(self, task, host, partial):
        print(f"[progress] {host.name}: {partial.result}")

    def task_started(self, task): ...
    def task_completed(self, task, result): ...
    def task_instance_started(self, task, host): ...
    def task_instance_completed(self, task, host, result): ...
    def subtask_instance_started(self, task, host): ...
    def subtask_instance_completed(self, task, host, result): ...


result = nr.with_processors([PrintProgress()]).run(task=upgrade, image="eos-4.32.bin")
print(f"Results stored for edge-1: {[r.result for r in result['edge-1']]}")

[progress] edge-1: copying eos-4.32.bin
[progress] edge-1: rebooting
[progress] edge-2: copying eos-4.32.bin
[progress] edge-2: rebooting
[progress] edge-3: copying eos-4.32.bin
[progress] edge-3: rebooting
[progress] edge-4: copying eos-4.32.bin
[progress] edge-4: rebooting
Results stored for edge-1: ['running eos-4.32.bin']

Stop the remaining hosts after a failure

nr.submit() starts the run in the background and returns a handle at once. handle.results() yields the final result of each host in the order the hosts finish, and handle.cancel() stops the run.

Cancellation is delivered at the task’s next checkpoint: a call to task.checkpoint(), a yield, or the start of a subtask. Hosts that have not started yet never start.

def change(task: Task):
    for _ in range(20):
        task.checkpoint()
        time.sleep(0.01)  # one step of the change
    if task.host.name == "edge-2":
        raise ValueError("the new configuration was rejected")
    return Result(host=task.host, result="change applied", changed=True)


handle = nr.submit(task=change)
for res in handle.results():
    if res.failed:
        print(f"{res.host.name} failed, cancelling the run")
        handle.cancel()
        break

result = handle.result()
for name, host_results in result.items():
    outcome = host_results[0]
    print(f"{name}: {type(outcome.exception).__name__ if outcome.failed else outcome.result}")

edge-2 failed, cancelling the run
edge-1: change applied
edge-2: ValueError
edge-3: TaskCancelledError
edge-4: TaskCancelledError

Every selected host has exactly one final result. A cancelled host carries TaskCancelledError, and its started field says whether the task had started. Cancelled hosts are added to nr.data.failed_hosts, as any failed host is.

print(sorted(nr.data.failed_hosts))
print(result["edge-4"][0].exception)
nr.data.reset_failed_hosts()

['edge-2', 'edge-3', 'edge-4']
Task 'change' on host 'edge-4' was cancelled before it started.

TaskCancelledError subclasses BaseException, like asyncio.CancelledError, so an except Exception block in a task does not stop a cancellation.

Give each host a deadline

A runner takes a per-host deadline (per_host_timeout) and a deadline for the whole run (timeout), in seconds. They also work from the configuration file, under runner.options. A host that passes its deadline is recorded with TaskTimeoutError; the other hosts are not affected.

def slow_on_edge_3(task: Task):
    steps = 100 if task.host.name == "edge-3" else 2
    for _ in range(steps):
        task.checkpoint()
        time.sleep(0.01)
    return Result(host=task.host, result="finished")


deadline_nr = nr.with_runner(ThreadedRunner(num_workers=1, per_host_timeout=0.5))
result = deadline_nr.run(task=slow_on_edge_3)
for name, host_results in result.items():
    print(f"{name}: {host_results[0].result}")
deadline_nr.data.reset_failed_hosts()

edge-1: finished
edge-2: finished
edge-3: Task 'slow_on_edge_3' on host 'edge-3' exceeded its per-host timeout of 0.5 seconds.
edge-4: finished

submit and stream also take run_options, which overrides the runner’s deadlines for one run. It is a separate argument, so it never takes a name from the task’s own arguments:

handle = nr.submit(task=slow_on_edge_3, run_options=RunOptions(per_host_timeout=0.3))
error = handle.result()["edge-3"][0].exception
print(isinstance(error, TaskTimeoutError), error.scope, error.timeout)
nr.data.reset_failed_hosts()

True host 0.3

On ThreadedRunner, a deadline is enforced even when a task is blocked in a call that never returns: the run records the host as timed out and returns on time. The blocked thread keeps its worker slot until the call returns, so tasks built on blocking libraries should call task.checkpoint() between commands. AsyncioRunner interrupts a host at the call it is awaiting, without checkpoints.

See the execution model for what each runner supports.