Task Plugins¶
The only task plugin currently is the "dispatcher" plugin. This plugin dispatches to the more specific OS specific functions. To demonstrate the primary components of the code:
Dispatcher Sender¶
- If exists check
custom_dispatcher
, for network_driver, if a custom_dispatcher is used but not found, fail immediately - Check for framework & driver
f"nornir_nautobot.plugins.tasks.dispatcher.{network_driver}.{framework.title()}{network_driver_title}"
- Check for default, e.g.
f"nornir_nautobot.plugins.tasks.dispatcher.default.{framework.title()}Default"
Info
Where framework
is a library like netmiko
or napalm
and network_driver
is the platform like cisco_ios
or arista_eos
.
This may seem like a lot, but it essentially can be broken down to:
- If there is a custom_dispatcher, only use that
- Check for the
framework
andnetwork_driver
- Check for the
framework
's default
For completeness here is the referenced code as of October 2023.
custom_dispatcher = ""
if kwargs.get("custom_dispatcher"):
custom_dispatcher = kwargs["custom_dispatcher"]
del kwargs["custom_dispatcher"]
logger.debug(f"Dispatcher process started for {task.host.name} ({task.host.platform})")
network_driver = task.host.platform
network_driver_title = snake_to_title_case(network_driver)
framework_path = (
f"nornir_nautobot.plugins.tasks.dispatcher.{network_driver}.{framework.title()}{network_driver_title}"
)
framework_default_path = f"nornir_nautobot.plugins.tasks.dispatcher.default.{framework.title()}Default"
if custom_dispatcher:
driver_class = import_string(custom_dispatcher)
checked_path = [custom_dispatcher]
elif import_string(framework_path):
driver_class = import_string(framework_path)
checked_path = [framework_path]
else:
driver_class = import_string(framework_default_path)
checked_path = [framework_path, framework_default_path]
result = task.run(task=driver_task, *args, **kwargs)
Dispatcher Receiver¶
class NautobotNornirDriver:
"""Default collection of Nornir Tasks based on Napalm."""
@classmethod
def get_config(cls, task: Task, backup_file: str, *args, **kwargs) -> Result:
Calling Dispatcher¶
task.run(
task=dispatcher,
obj=obj,
logger=logger,
method="get_config",
framework="netmiko",
name="SAVE BACKUP CONFIGURATION TO FILE",
backup_file=backup_file,
remove_lines=remove_regex_dict.get(obj.platform.network_driver, []),
substitute_lines=replace_regex_dict.get(obj.platform.network_driver, []),
)
The dispatcher expects the two primary objects, the obj
and logger
objects. The obj
object should be a Device model instance. The logger must conform to the standard Python logger, in that it should take is message
as the first arg and allow a dictionary called extra
.
Each task will raise a NornirNautobotException
for known issues. Using a custom processor, the user can predict when it was an well known error.
Check Connectivity Configuration¶
The check connectivity receiver will send attempt to tcp ping the port based on the following order or precedence.
- Prefer
obj.cf["tcp_port"]
if is a valid integer - Prefer
obj.get_config_context()["tcp_port"]
if is a valid integer - Prefer cls.tcp_port, which by default is defined in
DispatcherMixin
as 22
In this code you can see how it is set.
class DispatcherMixin:
tcp_port = 22
@classmethod
def _get_tcp_port(cls, obj) -> str:
custom_field = obj.cf.get("tcp_port")
if isinstance(custom_field, int):
return custom_field
config_context = obj.get_config_context().get("tcp_port")
if isinstance(config_context, int):
return config_context
return cls.tcp_port