extract undersync-device workflow into understack_workflows

argo-workflows/nautobot-update-cf/code/helpers.py

@@ -31,41 +31,6 @@ def arg_parser(name):
 
     return parser
 
-
-def undersync_parser(name):
-    parser = argparse.ArgumentParser(
-        prog=os.path.basename(name), description="Trigger undersync run for a device"
-    )
-    parser.add_argument("--device_uuid", required=True, help="Nautobot device UUID")
-    parser.add_argument("--network-name", required=True)
-    parser.add_argument("--nautobot_url", required=False)
-    parser.add_argument("--nautobot_token", required=False)
-    parser.add_argument(
-        "--force",
-        type=__boolean_args,
-        help="Call Undersync's force endpoint",
-        required=False,
-    )
-    parser.add_argument(
-        "--dry-run",
-        type=__boolean_args,
-        help="Call Undersync's dry-run endpoint",
-        required=False,
-    )
-
-    return parser
-
-
-def __boolean_args(val):
-    normalised = str(val).upper()
-    if normalised in ["YES", "TRUE", "T", "1"]:
-        return True
-    elif normalised in ["NO", "FALSE", "F", "N", "0"]:
-        return False
-    else:
-        raise argparse.ArgumentTypeError("boolean expected")
-
-
 def exit_with_error(error):
     logger.error(error)
     sys.exit(1)

argo-workflows/trigger-undersync/workflowtemplates/undersync-device.yaml

@@ -3,13 +3,14 @@ metadata:
   name: undersync-device
 kind: WorkflowTemplate
 spec:
+  entrypoint: trigger-undersync
+  serviceAccountName: workflow
   templates:
     - name: trigger-undersync
       container:
-        image: ghcr.io/rackerlabs/understack/nautobot-update-cf:latest
+        image: ghcr.io/rackerlabs/understack/ironic-nautobot-client:latest
         command:
-          - python
-          - /app/with_ifaces.py
+          - undersync-device
         args:
           - --device_uuid
           - "{{workflow.parameters.device_uuid}}"

python/understack-workflows/pyproject.toml

@@ -44,6 +44,7 @@ synchronize-obm-creds = "understack_workflows.main.synchronize_obm_creds:main"
 synchronize-server = "understack_workflows.main.synchronize_server:main"
 sync-nautobot-interfaces = "understack_workflows.main.sync_nautobot_interfaces:main"
 undersync-switch = "understack_workflows.main.undersync_switch:main"
+undersync-device = "understack_workflows.main.undersync_device:main"
 
 [tool.setuptools.packages.find]
 # avoid packaging up our tests
@@ -60,6 +61,9 @@ testpaths = [
 target-version = "py311"
 fix = true
 
+[tool.isort]
+profile = "open_stack"
+
 [tool.ruff.lint]
 select = [
   "D",       # pydocstyle

python/understack-workflows/understack_workflows/main/undersync_device.py

@@ -0,0 +1,86 @@
+import argparse
+import os
+import sys
+
+from understack_workflows.helpers import boolean_args
+from understack_workflows.helpers import credential
+from understack_workflows.helpers import setup_logger
+from understack_workflows.nautobot import Nautobot
+from understack_workflows.undersync.client import Undersync
+
+logger = setup_logger(__name__)
+
+def update_nautobot(args) -> list[str]:
+    default_nb_url = "http://nautobot-default.nautobot.svc.cluster.local"
+    device_uuid = args.device_uuid
+    field_name = "connected_to_network"
+    field_value = args.network_name
+    nb_url = args.nautobot_url or default_nb_url
+
+    nb_token = args.nautobot_token or credential("nb-token", "token")
+    nautobot = Nautobot(nb_url, nb_token, logger=logger)
+    logger.info(
+        f"Updating Device {device_uuid} and moving it to '{field_value}' network."
+    )
+    nautobot.update_cf(device_uuid, field_name, field_value)
+    logger.debug(f"Updated Device.{field_name} to {field_value}")
+    switches = nautobot.uplink_switches(device_uuid)
+    logger.info(f"Obtained switch IDs: {switches}")
+    return switches
+
+
+def call_undersync(args, switches):
+    undersync_token = credential("undersync", "token")
+    if not undersync_token:
+        logger.error("Please provide auth token for Undersync.")
+        sys.exit(1)
+    undersync = Undersync(undersync_token)
+
+    try:
+        return undersync.sync_devices(switches, dry_run=args.dry_run, force=args.force)
+    except Exception as error:
+        logger.error(error)
+        sys.exit(2)
+
+
+def argument_parser():
+    parser = argparse.ArgumentParser(
+        prog=os.path.basename(__file__),
+        description="Trigger undersync run for a device",
+    )
+    parser.add_argument("--device_uuid", required=True, help="Nautobot device UUID")
+    parser.add_argument("--network-name", required=True)
+    parser.add_argument("--nautobot_url", required=False)
+    parser.add_argument("--nautobot_token", required=False)
+    parser.add_argument(
+        "--force",
+        type=boolean_args,
+        help="Call Undersync's force endpoint",
+        required=False,
+    )
+    parser.add_argument(
+        "--dry-run",
+        type=boolean_args,
+        help="Call Undersync's dry-run endpoint",
+        required=False,
+    )
+
+    return parser
+
+
+def main():
+    """Updates connected_to_network and triggers Undersync.
+
+    Updates Nautobot Device's 'connected_to_network' field and follows with
+    request to Undersync service, requesting sync for all of the
+    uplink_switches that the device is connected to.
+    """
+    args = argument_parser().parse_args()
+
+    switches = update_nautobot(args)
+    response = call_undersync(args, switches)
+    logger.info(f"Undersync returned: {response.json()}")
+
+
+if __name__ == "__main__":
+    main()

python/understack-workflows/understack_workflows/nautobot.py

@@ -110,3 +110,38 @@ def bulk_create_interfaces(
                 self.logger.info(f"{interface.name} successfully created")
 
             return req
+
+    def device_by_id(self, device_id: str) -> NautobotDevice:
+        device = self.session.dcim.devices.get(device_id)
+        if not device:
+            self.exit_with_error(f"Device {device_id} not found in Nautobot")
+        return device
+
+    def device_interfaces(self, device_id: str):
+        return self.session.dcim.interfaces.filter(device_id=device_id)
+
+    def update_cf(self, device_id, field_name: str, field_value: str):
+        device = self.device_by_id(device_id)
+        device.custom_fields[field_name] = field_value
+        response = device.save()
+        self.logger.info(f"save result: {response}")
+        return response
+
+    def uplink_switches(self, device_id) -> list[str]:
+        interfaces = self.device_interfaces(device_id)
+        ids = set()
+        for iface in interfaces:
+            endpoint = iface.connected_endpoint
+            if not endpoint:
+                continue
+            endpoint.full_details()
+            self.logger.debug(
+                f"{iface} connected device {iface.connected_endpoint.device} "
+            )
+            remote_switch = endpoint.device
+            if not remote_switch:
+                continue
+
+            ids.add(remote_switch.id)
+
+        return list(ids)