Validate router

Router validation workflow. Used in a nightly schedule.

prepare_state(subscription_id)

Add required keys to the state for the workflow to run successfully.

Source code in gso/workflows/router/validate_router.py

@step("Prepare required keys in state")
def prepare_state(subscription_id: UUIDstr) -> State:
    """Add required keys to the state for the workflow to run successfully."""
    router = Router.from_subscription(subscription_id)

    return {"subscription": router}

verify_ipam_loopback(subscription)

Validate the IPAM resources for the loopback interface.

Raises:

Type	Description
ProcessFailureError	If IPAM is configured incorrectly.

Source code in gso/workflows/router/validate_router.py

@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: Router) -> None:
    """Validate the IPAM resources for the loopback interface.

    Raises:
        ProcessFailureError: If IPAM is configured incorrectly.
    """
    host_record = infoblox.find_host_by_fqdn(f"lo0.{subscription.router.router_fqdn}")
    if not host_record or str(subscription.subscription_id) not in host_record.comment:
        msg = "Loopback record is incorrectly configured in IPAM, please investigate this manually!"
        raise ProcessFailureError(msg)

check_netbox_entry_exists(subscription)

Validate the Netbox entry for a Router.

This will only ensure existence of the node itself in Netbox. Validation of separate interfaces takes places in other subscriptions' validation workflows.

Source code in gso/workflows/router/validate_router.py

@step("Verify correct Netbox entry")
def check_netbox_entry_exists(subscription: Router) -> None:
    """Validate the Netbox entry for a Router.

    This will only ensure existence of the node itself in Netbox. Validation of separate interfaces takes places in
    other subscriptions' validation workflows.
    """
    client = NetboxClient()
    #  Try and fetch the host, which will raise an exception on failure.
    client.get_device_by_name(subscription.router.router_fqdn)

verify_p_ibgp(subscription)

Verify PE neighbors in P-ONLY group on a P router.

Source code in gso/workflows/router/validate_router.py

@step("Verify P BGP P-ONLY neighbors")
def verify_p_ibgp(subscription: dict[str, Any]) -> LSOState:
    """Verify PE neighbors in `P-ONLY` group on a P router."""
    extra_vars = {
        "dry_run": True,
        "subscription": subscription,
        "pe_router_list": generate_inventory_for_routers(RouterRole.PE)["all"]["hosts"],
        "verb": "add_pe_to_p",
        "is_verification_workflow": "true",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/update_ibgp_mesh.yaml",
        "inventory": {"all": {"hosts": {subscription["router"]["router_fqdn"]: None}}},
        "extra_vars": extra_vars,
    }

verify_pe_mesh_in_pe(subscription)

Verify PE internal mesh neighbors on a PE router.

Source code in gso/workflows/router/validate_router.py

@step("Verify PE BGP internal mesh neighbors")
def verify_pe_mesh_in_pe(subscription: dict[str, Any]) -> LSOState:
    """Verify PE internal mesh neighbors on a PE router."""
    extra_vars = {
        "dry_run": True,
        "subscription": subscription,
        "verb": "add_pe_mesh_to_pe",
        "pe_router_list": generate_inventory_for_routers(
            router_role=RouterRole.PE, exclude_routers=[subscription["router"]["router_fqdn"]]
        )["all"]["hosts"],
        "is_verification_workflow": "true",
    }

    if not extra_vars["pe_router_list"]:
        return {
            "playbook_name": "",
            "inventory": {"all": {"hosts": {}}},
            "extra_vars": {},
        }

    return {
        "playbook_name": "gap_ansible/playbooks/update_ibgp_mesh.yaml",
        "inventory": {"all": {"hosts": {subscription["router"]["router_fqdn"]: None}}},
        "extra_vars": extra_vars,
    }

verify_all_p_in_pe(subscription)

Verify P neighbors in P-ONLY group on a PE router.

Source code in gso/workflows/router/validate_router.py

@step("Verify PE BGP P-ONLY neighbors")
def verify_all_p_in_pe(subscription: dict[str, Any]) -> LSOState:
    """Verify P neighbors in `P-ONLY` group on a PE router."""
    extra_vars = {
        "dry_run": True,
        "subscription": subscription,
        "verb": "add_all_p_to_pe",
        "p_router_list": generate_inventory_for_routers(
            router_role=RouterRole.P, exclude_routers=[subscription["router"]["router_fqdn"]]
        )["all"]["hosts"],
        "is_verification_workflow": "true",
    }

    if not extra_vars["p_router_list"]:
        return {
            "playbook_name": "",
            "inventory": {"all": {"hosts": {}}},
            "extra_vars": {},
        }

    return {
        "playbook_name": "gap_ansible/playbooks/update_ibgp_mesh.yaml",
        "inventory": {"all": {"hosts": {subscription["router"]["router_fqdn"]: None}}},
        "extra_vars": extra_vars,
    }

check_librenms_entry_exists(subscription)

Validate the LibreNMS entry for a Router.

Raises an HTTP error 404 when the device is not present in LibreNMS.

Source code in gso/workflows/router/validate_router.py

@step("Verify correct LibreNMS entry")
def check_librenms_entry_exists(subscription: Router) -> None:
    """Validate the LibreNMS entry for a Router.

    Raises an HTTP error 404 when the device is not present in LibreNMS.
    """
    client = LibreNMSClient()
    errors = client.validate_device(subscription.router.router_fqdn)
    if errors:
        raise ProcessFailureError(message="LibreNMS configuration error", details=errors)

check_kentik_entry_exists(subscription)

Validate the Kentik entry for a PE Router.

If a router has at least one layer 3 service insisting on it, there should be a valid Kentik license applied to this device. The only thing we can check for reliably, is whether this device does not have an archiving or placeholder license on it. This is because there can be multiple, valid, non-archiving licenses for devices.

Source code in gso/workflows/router/validate_router.py

@step("Verify Kentik entry for PE router")
def check_kentik_entry_exists(subscription: Router) -> None:
    """Validate the Kentik entry for a PE Router.

    If a router has at least one layer 3 service insisting on it, there should be a valid Kentik license applied to this
    device. The only thing we can check for reliably, is whether this device does not have an archiving or placeholder
    license on it. This is because there can be multiple, valid, non-archiving licenses for devices.

    Raises:
        ProcessFailureError when a Kentik device is missing, or configured incorrectly.
    """
    client = KentikClient()

    #  Check if the device exists in Kentik.
    device = client.get_device_by_name(subscription.router.router_fqdn)
    if not device:
        raise ProcessFailureError(
            message="Device not found in Kentik", details={"device": subscription.router.router_fqdn}
        )

    #  If there are active layer 3 services, check the license type. It may not be the placeholder or archiving license.
    if bool(get_active_layer_3_services_on_router(subscription.subscription_id)):
        kentik_params = load_oss_params().KENTIK
        archive_plan = client.get_plan_by_name(kentik_params.archive_license_key)
        if any(device["device_name"] == subscription.router.router_fqdn for device in archive_plan["devices"]):
            raise ProcessFailureError(
                message="Device in Kentik incorrectly configured",
                details=f"Kentik device {subscription.router.router_fqdn} has the archiving license "
                f"{archive_plan["name"]} applied to it, despite the existence of active layer 3 services.",
            )

        placeholder_plan = client.get_plan_by_name(kentik_params.placeholder_license_key)
        if any(device["device_name"] == subscription.router.router_fqdn for device in placeholder_plan["devices"]):
            raise ProcessFailureError(
                message="Device in Kentik incorrectly configured",
                details=f"Kentik device {subscription.router.router_fqdn} has the placeholder license "
                f"{placeholder_plan["name"]} applied to it, despite the existence of active layer 3 services.",
            )

verify_base_config(subscription)

Workflow step for running a playbook that checks whether base config has drifted.

Source code in gso/workflows/router/validate_router.py

@step("Check base config for drift")
def verify_base_config(subscription: dict[str, Any]) -> LSOState:
    """Workflow step for running a playbook that checks whether base config has drifted."""
    vrf_list = get_active_vrfs_linked_to_router(str(subscription["subscription_id"]))
    return {
        "playbook_name": "gap_ansible/playbooks/base_config.yaml",
        "inventory": {"all": {"hosts": {subscription["router"]["router_fqdn"]: None}}},
        "extra_vars": {
            "wfo_router_json": subscription,
            "vrf_list": vrf_list,
            "verb": "deploy",
            "dry_run": "true",
            "is_verification_workflow": "true",
        },
    }

validate_router()

Validate an existing, active Router subscription.

• Verify that the loopback interface is correctly configured in IPAM.
• Verify that the router is correctly configured in Netbox.
• Verify that the router is correctly configured in LibreNMS.
• Redeploy base config to verify the configuration is intact.
• Validate configuration of the iBGP mesh

Source code in gso/workflows/router/validate_router.py

@workflow(
    "Validate router configuration", target=Target.SYSTEM, initial_input_form=wrap_modify_initial_input_form(None)
)
def validate_router() -> StepList:
    """Validate an existing, active Router subscription.

    * Verify that the loopback interface is correctly configured in IPAM.
    * Verify that the router is correctly configured in Netbox.
    * Verify that the router is correctly configured in LibreNMS.
    * Redeploy base config to verify the configuration is intact.
    * Validate configuration of the iBGP mesh
    """
    is_juniper_router = conditional(lambda state: state["subscription"]["router"]["vendor"] == Vendor.JUNIPER)
    is_pe_router = conditional(lambda state: state["subscription"]["router"]["router_role"] == RouterRole.PE)
    is_p_router = conditional(lambda state: state["subscription"]["router"]["router_role"] == RouterRole.P)

    return (
        begin
        >> store_process_subscription(Target.SYSTEM)
        >> prepare_state
        >> is_juniper_router(done)
        >> unsync
        >> verify_ipam_loopback
        >> check_netbox_entry_exists
        >> check_librenms_entry_exists
        >> is_pe_router(check_kentik_entry_exists)
        >> anonymous_lso_interaction(verify_base_config)
        >> is_p_router(anonymous_lso_interaction(verify_p_ibgp))
        >> is_pe_router(anonymous_lso_interaction(verify_pe_mesh_in_pe))
        >> is_pe_router(anonymous_lso_interaction(verify_all_p_in_pe))
        >> resync
        >> done
    )