Serialization in grouped tasks breaks when using click

[lookinthebest]

Issue: Click output is not properly using serialization in a grouped task when it is in the subtask and num_worker is set to 1.   If I switch click.confirm with a custom-made input() confirmation it will properly separate the two outputs between threads.

Is there a work around for this? It is quite convenient to be able to use click for prompting like this.   Below is the result I get when utilizing the click package for prompting the user. It puts both confirmations for both switches in the filter on one line. If I answer 'n' it will go to a blank new line expecting the answer for the second thread and then proceed like normal. The same thing happens for all click related prompts. It's not locking them to the thread.

The code used is summarized below:

def create_port_lists(task: Task, data: Dict[str, Any]) -> Result:
    result = defaultdict(list)
    interfaces = data["data_collection"][task.host.name]["ttp_interfaces"]
    update = []
    ommitPorts = click.confirm(
        f"Does {task.host} have any ports to ommit?", default=False
    )
    if ommitPorts == True:
        click.echo(f"\nPorts:")
        click.echo(f"{interfaces}\n")
        ports = click.prompt(
            "Enter list of ports separated by space (Example: FiveGigabgitEthernet1/0/10 FiveGigabgitEthernet1/0/42)"
        )
        ports = ports.split()
        click.echo(f"\nOmmited Ports:")
        click.echo(f"{ports}\n")
        click.confirm(
            "Is the above list of ommited ports correct?", default=False, abort=True
        )
        for interface in ports:
            result["ommited"].append(interface)
            update.append(interface)
    interfaces = [x for x in interfaces if x not in update]
    task.host.data["ports"] = result
    return Result(host=task.host, result=result)

def examine_ports(task: Task, data: Dict[str, Any], num_worker: int) -> Result:
    task.nornir.runner.num_worker = num_worker
    task.run(task=create_port_lists, data=data)
    return Result(
        host=task.host,
        result=f"{task.host}'s configuration collected and parsed.",
    )
kwargs = {"num_workers": 1}
nr = InitNornir(config_file=NORNIR_PATH, dry_run=dryrun)
data = {}
nr.run(name="data_collection", task=data_collection)
nr.run(name="examine_ports", task=examine_ports, data=data, **kwargs)

[ktbyers]

You need to include your Nornir config file i.e. NORNIR_PATH???

You should not be setting num_workers=1, you should be changing to serial runner in your Nornir config file.

runner:
  plugin: serial

[lookinthebest]

This code is summarized. I want it to be parallel except for this task.

I thought the documentation said that by setting num_worker=1 it defaults the grouped tasks to be serialized in its execution flow?

If I still have to set the plugin to serial can that be done in the subtask?

inventory:
  plugin: SimpleInventory
  options:
    host_file: "C:/Users/hosts.yaml"
    group_file: "C:/Users/groups.yaml"
    defaults_file: "C:/Users/defaults.yaml"

runner:
  plugin: threaded
  options:
    num_workers: 50

logging:
  enabled: true
  log_file: "./nornir.log"
  to_console: false

[ktbyers]

I thought the documentation said that by setting num_worker=1 it defaults the grouped tasks 
to be serialized in its execution flow?

Do you have a link for where the docs say this?

Anyways you should set it to serial for no threading (i.e. serial behavior)

[lookinthebest]

Anyways you should set it to serial for no threading (i.e. serial behavior)

The whole script from globally at the config.yaml file level? Or can I update just the runner plugin object from within the tasks themselves?

https://nornir.readthedocs.io/en/latest/plugins/execution_model.html?highlight=execution%20flow

[lookinthebest]

[ubaumann]

I personally would not do a "prompt" inside of a task and try to do it outside. Would something like this work for you?

<.... your basic imports and normal code >

result = nr.run(name="data_collection", task=data_collection)

for r in result:
   # logic to ask the prompt and store the info for the host

# run the rest of the tasks
nr.run(name="examine_ports", task=examine_ports, data=data, **kwargs)

The other option could be using LiveDisplay by rich. Rich also as prompts, but I did not use it inside a task https://rich.readthedocs.io/en/stable/live.html https://rich.readthedocs.io/en/stable/prompt.html

[lookinthebest]

Oh, so you guys are just bringing prompts out of the tasks and updating the object in the main thread.

That is what I was about to do but I thought I would ask since the documentation seemed like it should work.

        r = nr.run(name="data_collection", task=data_collection)
        r = nr.run(name="examine_ports", task=examine_ports, data=data)
        ommit_ports(data=data)

[ubaumann]

Anyways you should set it to serial for no threading (i.e. serial behavior)

The whole script from globally at the config.yaml file level? Or can I update just the runner plugin object from within the tasks themselves?

https://nornir.readthedocs.io/en/latest/plugins/execution_model.html?highlight=execution%20flow

Nornir object has a with_runner function to run a task with another runner plugin, but it has not been used for a long time. It could have been even before the last major release, so you must try it. https://nornir.readthedocs.io/en/latest/api/nornir/core/__init__.html#nornir.core.__init__.Nornir.with_runner

[lookinthebest]

Thank you for your help. I'll try one of these two options.

[ubaumann]

Oh, so you guys are just bringing prompts out of the tasks and updating the object in the main thread.

That is what I was about to do but I thought I would ask since the documentation seemed like it should work.

        r = nr.run(name="data_collection", task=data_collection)
        r = nr.run(name="examine_ports", task=examine_ports, data=data)
        ommit_ports(data=data)

I can't talk for others, but my functions typically run without inputs from outside. If I create a CLI tool, I first ask for the prompt, if possible. But it always depends

[ktbyers]

@lookinthebest Okay, thanks for the doc link...those docs are crazy old though and I think need to be fixed (reference Brigade i.e. the name of this project before version 1...so something like 5+ years old).

[lookinthebest]

@lookinthebest Okay, thanks for the doc link...those docs are crazy old though and I think need to be fixed (reference Brigade i.e. the name of this project before version 1...so something like 5+ years old).

Oh wow. Well yea that might need deleted then it is quite confusing.

Oh, so you guys are just bringing prompts out of the tasks and updating the object in the main thread. That is what I was about to do but I thought I would ask since the documentation seemed like it should work.

        r = nr.run(name="data_collection", task=data_collection)
        r = nr.run(name="examine_ports", task=examine_ports, data=data)
        ommit_ports(data=data)

I can't talk for others, but my functions typically run without inputs from outside. If I create a CLI tool, I first ask for the prompt, if possible. But it always depends

Yea in this case I'm making the selection a lot easier buy parsing some of the information first. Then asking the user for more.

[ktbyers]

Similar logic to what @ubaumann posted earlier, but this also should work:

<.... your basic imports and normal code >

# I assume relevant interface data is bound to the host as part of this task
result = nr.run(name="data_collection", task=data_collection)

for host_obj in nr.inventory.hosts.values():
    omit_ports(host_obj)

# Additional nr.run calls

Looping over the results should also work.

[ubaumann]

Yea in this case I'm making the selection a lot easier buy parsing some of the information first. Then asking the user for more.

I am not sure if I followed, but you still can make the data collection a task,

You could also use threading.Lock()

import threading

LOCK = threadin.Lock()

def create_port_lists(task: Task, data: Dict[str, Any]) -> Result:
    result = defaultdict(list)
    interfaces = data["data_collection"][task.host.name]["ttp_interfaces"]
    update = []
    LOCK.acquire()
    ommitPorts = click.confirm(
        f"Does {task.host} have any ports to ommit?", default=False
    )
    if ommitPorts == True:
        click.echo(f"\nPorts:")
        click.echo(f"{interfaces}\n")
        ports = click.prompt(
            "Enter list of ports separated by space (Example: FiveGigabgitEthernet1/0/10 FiveGigabgitEthernet1/0/42)"
        )
        ports = ports.split()
        click.echo(f"\nOmmited Ports:")
        click.echo(f"{ports}\n")
        click.confirm(
            "Is the above list of ommited ports correct?", default=False, abort=True
        )
        for interface in ports:
            result["ommited"].append(interface)
            update.append(interface)
    LOCK.release()    
    interfaces = [x for x in interfaces if x not in update]
    task.host.data["ports"] = result
    return Result(host=task.host, result=result)

But personally, I don't like this approach in this situation. I would try the with_runner solution.

from nornir.plugins.runner import SerialRunner

nr = InitNornir(config_file=NORNIR_PATH, dry_run=dryrun)
data = {}
nr.with_runner(SerialRunner()).run(name="data_collection", task=data_collection)
nr.run(name="examine_ports", task=examine_ports, data=data, **kwargs)

[lookinthebest]

Both methods work. The threading lock and the with_runner. No num_worker argument was needed thanks for the clarification on all of this.