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.

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

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.

    Raises an HTTP error 404 when the device is not present in Kentik.
    """
    client = KentikClient()
    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}
        )

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
    )