Workflow Service Run Task

task api name: run

Run workflow defined using YAML file.

Workflow Sample Usage

Workflow service run task uses YAML formatted files to execute workflow steps:

workflow-1.yaml

name: workflow_1
description: Sample workflow with two steps.

step1:
  service: nornir
  task: cli
  kwargs:
    FC: spine
    commands:
      - show version
      - show ip int brief

step2:
  service: nornir
  task: cli
  kwargs:
    FC: leaf
    commands:
      - show hostname
      - show ntp status

File workflow-1.yaml stored on broker and downloaded by Workflow service prior to running steps, below is an example of how to run the the workflow.

Example

CLIPython

C:\nf>nfcli
Welcome to NorFab Interactive Shell.
nf#
nf#workflow run workflow nf://workflow/workflow-1.yaml
--------------------------------------------- Job Events -----------------------------------------------
05-Apr-2025 21:34:53.846 d1634ce3dc764f56ac00971950a033cc job started
05-Apr-2025 21:34:53.883 INFO workflow-worker-1 running workflow.run  - Starting workflow 'workflow_1'
05-Apr-2025 21:34:53.883 INFO workflow-worker-1 running workflow.run  - Doing workflow step 'step1'
05-Apr-2025 21:34:55.008 INFO workflow-worker-1 running workflow.run  - Doing workflow step 'step2'
05-Apr-2025 21:34:56.557 d1634ce3dc764f56ac00971950a033cc job completed in 2.711 seconds

--------------------------------------------- Job Results --------------------------------------------

workflow-worker-1:
    ----------
    workflow_1:
        ----------
        step1:
            ----------
            nornir-worker-1:
                ----------
                task:
                    nornir-worker-1:cli
                failed:
                    False
                errors:
                result:
                    ----------
                    ceos-spine-2:
                        ----------
                        show version:
                            Arista cEOSLab
                            Hardware version:
                            Serial number: 8B7EBC67A4FA6C48F1D1BCC5438866A7
                            Hardware MAC address: 001c.73ab.5167
                            System MAC address: 001c.73ab.5167

                            Software image version: 4.30.0F-31408673.4300F (engineering build)
                            Architecture: x86_64
                            Internal build version: 4.30.0F-31408673.4300F
                            Internal build ID: a35f0dc7-2d65-4f2a-a010-279cf445fd8c
                            Image format version: 1.0
                            Image optimization: None

                            cEOS tools version: (unknown)
                            Kernel version: 5.15.0-136-generic

                            Uptime: 1 hour and 17 minutes
                            Total memory: 32827264 kB
                            Free memory: 19525528 kB
                        show ip int brief:
                                                                                                              Address
                            Interface         IP Address              Status       Protocol            MTU    Owner
                            ----------------- ----------------------- ------------ -------------- ----------- -------
                            Loopback0         unassigned              up           up                65535
                            Loopback123       unassigned              up           up                65535
                            Management0       172.100.100.11/24       up           up                 1500
                    ceos-spine-1:
                        ----------
                        show version:
                            Arista cEOSLab
                            Hardware version:
                            Serial number: 7B5E3CF8CB9A6DE53FB8411896DE476F
                            Hardware MAC address: 001c.730a.5369
                            System MAC address: 001c.730a.5369

                            Software image version: 4.30.0F-31408673.4300F (engineering build)
                            Architecture: x86_64
                            Internal build version: 4.30.0F-31408673.4300F
                            Internal build ID: a35f0dc7-2d65-4f2a-a010-279cf445fd8c
                            Image format version: 1.0
                            Image optimization: None

                            cEOS tools version: (unknown)
                            Kernel version: 5.15.0-136-generic

                            Uptime: 1 hour and 17 minutes
                            Total memory: 32827264 kB
                            Free memory: 19525528 kB
                        show ip int brief:
                                                                                                              Address
                            Interface         IP Address              Status       Protocol            MTU    Owner
                            ----------------- ----------------------- ------------ -------------- ----------- -------
                            Loopback0         unassigned              up           up                65535
                            Loopback123       unassigned              up           up                65535
                            Management0       172.100.100.10/24       up           up                 1500
                messages:
                juuid:
                    0bef61db66cf46318735e02bdb0389c0
                status:
                    completed
        step2:
            ----------
            nornir-worker-2:
                ----------
                task:
                    nornir-worker-2:cli
                failed:
                    False
                errors:
                result:
                    ----------
                    ceos-leaf-2:
                        ----------
                        show hostname:
                            Hostname: ceos-leaf-2
                            FQDN:     ceos-leaf-2
                            unsynchronised
                            poll interval unknown
                    ceos-leaf-1:
                        ----------
                        show hostname:
                            Hostname: ceos-leaf-1
                            FQDN:     ceos-leaf-1
                        show ntp status:
                            unsynchronised
                            poll interval unknown
                    ceos-leaf-3:
                        ----------
                        show hostname:
                            Hostname: ceos-leaf-3
                            FQDN:     ceos-leaf-3
                        show ntp status:
                            unsynchronised
                            poll interval unknown
                messages:
                juuid:
                    972287e3abc94e86acfff99b54940ef9
                status:
                    completed
nf#

In this example:

• nfcli command starts the NorFab Interactive Shell.
• workflow run command runs workflow-1.yaml workflow.

This code is complete and can run as is

import pprint

from norfab.core.nfapi import NorFab

if __name__ == '__main__':
    nf = NorFab(inventory="inventory.yaml")
    nf.start()

    client = nf.make_client()

    res = client.run_job(
        service="workflow",
        task="run",
        kwargs={
            "workflow": "nf://workflow/workflow-1.yaml",
        }
    )

    pprint.pprint(res)

    nf.destroy()

NORFAB Workflow Test Shell Reference

NorFab shell supports these command options for workflow run task:

nf#man tree workflow
root
└── workflow:    Workflow service
    └── run:    Run workflows
        ├── timeout:    Job timeout
        ├── workers:    Filter worker to target, default 'all'
        ├── workflow:    Workflow to run
        └── progress:    Display progress events, default 'True'
nf#

* - mandatory/required command argument

Python API Reference

Executes a workflow defined by a dictionary.

Parameters:

Name	Type	Description	Default
workflow	Union[str, Dict]	The workflow to execute. This can be a URL to a YAML file.	required
remove_no_match_results	bool	Whether to remove empty results from the final output. Defaults to True.	required

Returns:

Name	Type	Description
Dict	Result	A dictionary containing the results of the workflow execution.

Raises:

Type	Description
ValueError	If the workflow is not a valid URL or dictionary.

Source code in norfab\workers\workflow_worker.py

@Task(fastapi={"methods": ["POST"]})
def run(self, job: Job, workflow: Union[str, Dict]) -> Result:
    """
    Executes a workflow defined by a dictionary.

    Args:
        workflow (Union[str, Dict]): The workflow to execute. This can be a URL to a YAML file.
        remove_no_match_results (bool, optional): Whether to remove empty results from the final output. Defaults to True.

    Returns:
        Dict: A dictionary containing the results of the workflow execution.

    Raises:
        ValueError: If the workflow is not a valid URL or dictionary.
    """
    ret = Result(task=f"{self.name}:run", result={})

    # load workflow from URL
    if self.is_url(workflow):
        workflow_name = (
            os.path.split(workflow)[-1].replace(".yaml", "").replace(".yml", "")
        )
        workflow = self.jinja2_render_templates([workflow])
        workflow = yaml.safe_load(workflow)

    # extract workflow parameters
    workflow_name = workflow.pop("name", "workflow")
    workflow_description = workflow.pop("description", "")
    remove_no_match_results = workflow.pop("remove_no_match_results", True)

    job.event(f"Starting workflow '{workflow_name}'")
    log.info(f"Starting workflow '{workflow_name}': {workflow_description}")

    ret.result[workflow_name] = {}

    # run each step in the workflow
    for step, data in workflow.items():
        # check if need to skip step based on run_if_x flags
        skip_status, message = self.skip_step_check(
            ret.result[workflow_name], step, data
        )
        if skip_status is True:
            ret.result[workflow_name][step] = {
                "all-workers": {
                    "failed": False,
                    "result": None,
                    "status": "skipped",
                    "task": data["task"],
                    "errors": [],
                    "messages": [message],
                    "juuid": None,
                }
            }
            job.event(
                f"Skipping workflow step '{step}', one of run_if_x conditions not satisfied"
            )
            continue
        # stop workflow execution on error
        elif skip_status == "error":
            ret.result[workflow_name][step] = {
                "all-workers": {
                    "failed": True,
                    "result": None,
                    "status": "error",
                    "task": data["task"],
                    "errors": [message],
                    "messages": [],
                    "juuid": None,
                }
            }
            job.event(message)
            log.error(message)
            break

        job.event(f"Doing workflow step '{step}'")

        ret.result[workflow_name][step] = self.client.run_job(
            service=data["service"],
            task=data["task"],
            workers=data.get("workers", "all"),
            kwargs=data.get("kwargs", {}),
            args=data.get("args", []),
            timeout=data.get("timeout", 600),
        )

        # check if need to stop workflow based on stop_if_fail flag
        if (
            self.stop_workflow_check(ret.result[workflow_name][step], step, data)
            is True
        ):
            break

    if remove_no_match_results:
        ret.result[workflow_name] = self.remove_no_match_results(
            ret.result[workflow_name]
        )

    return ret