-
Karel van Klink authoredKarel van Klink authored
validate_router.py 3.53 KiB
"""Router validation workflow. Used in a nightly schedule."""
from orchestrator.targets import Target
from orchestrator.utils.errors import ProcessFailureError
from orchestrator.workflow import StepList, begin, done, step, workflow
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic_forms.core import FormPage
from pydantic_forms.types import FormGenerator, UUIDstr
from gso.products.product_types.router import Router
from gso.services import infoblox
from gso.services.librenms_client import LibreNMSClient
from gso.services.lso_client import anonymous_lso_interaction
from gso.services.netbox_client import NetboxClient
from gso.utils.workflow_steps import deploy_base_config_dry, detect_configuration_drift
from gso.workflows.router.update_ibgp_mesh import add_p_to_mesh_dry
TT_NUMBER_ZERO = "TT#0000000000000000"
def _seed_initial_state() -> FormGenerator:
class InputForm(FormPage):
subscription_id: UUIDstr
user_input = yield InputForm
router = Router.from_subscription(user_input["subscription_id"])
return {"subscription": router, "tt_number": TT_NUMBER_ZERO}
@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: Router) -> None:
"""Validate the :term:`IPAM` resources for the loopback interface.
Raises an :class:`orchestrator.utils.errors.ProcessFailureError` if :term:`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)
@step("Verify correct Netbox entry")
def verify_netbox_entry(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)
@step("Verify correct LibreNMS entry")
def verify_librenms_entry(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()
client.get_device(subscription.router.router_fqdn)
@workflow(
"Validate router configuration",
target=Target.SYSTEM,
initial_input_form=wrap_modify_initial_input_form(_seed_initial_state),
)
def validate_router() -> StepList:
"""Validate an existing, active Router subscription.
* Verify that the loopback interface is correctly configured in :term:`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
"""
return (
begin
>> store_process_subscription(Target.SYSTEM)
>> unsync
>> verify_ipam_loopback
>> verify_netbox_entry
>> verify_librenms_entry
>> anonymous_lso_interaction(deploy_base_config_dry, detect_configuration_drift)
>> anonymous_lso_interaction(add_p_to_mesh_dry, detect_configuration_drift)
>> resync
>> done
)