create_router.py 10.72 KiB
"""A creation workflow for adding a new router to the network."""
from typing import Self
from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Choice, Label
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
from orchestrator.utils.errors import ProcessFailureError
from orchestrator.workflow import StepList, conditional, done, init, inputstep, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from pydantic import ConfigDict, model_validator
from pydantic_forms.validators import ReadOnlyField
from gso.products.product_blocks.router import RouterRole
from gso.products.product_types.router import RouterInactive, RouterProvisioning
from gso.products.product_types.site import Site
from gso.services import infoblox, subscriptions
from gso.services.lso_client import lso_interaction
from gso.services.netbox_client import NetboxClient
from gso.services.partners import get_partner_by_name
from gso.services.sharepoint import SharePointClient
from gso.settings import load_oss_params
from gso.utils.helpers import generate_fqdn, iso_from_ipv4
from gso.utils.shared_enums import PortNumber, Vendor
from gso.utils.workflow_steps import (
deploy_base_config_dry,
deploy_base_config_real,
prompt_sharepoint_checklist_url,
run_checks_after_base_config,
)
def _site_selector() -> Choice:
site_subscriptions = {}
for site in subscriptions.get_active_site_subscriptions(includes=["subscription_id", "description"]):
site_subscriptions[str(site["subscription_id"])] = site["description"]
# noinspection PyTypeChecker
return Choice("Select a site", zip(site_subscriptions.keys(), site_subscriptions.items(), strict=True)) # type: ignore[arg-type]
def initial_input_form_generator(product_name: str) -> FormGenerator:
"""Gather information about the new router from the operator."""
class CreateRouterForm(FormPage):
model_config = ConfigDict(title=product_name)
tt_number: str
partner: ReadOnlyField("GEANT", default_type=str) # type: ignore[valid-type]
vendor: Vendor
router_site: _site_selector() # type: ignore[valid-type]
hostname: str
ts_port: PortNumber
router_role: RouterRole
@model_validator(mode="after")
def hostname_must_be_available(self) -> Self:
router_site = self.router_site
if not router_site:
msg = "Please select a site before setting the hostname."
raise ValueError(msg)
selected_site = Site.from_subscription(router_site).site
input_fqdn = generate_fqdn(self.hostname, selected_site.site_name, selected_site.site_country_code)
if not infoblox.hostname_available(f"lo0.{input_fqdn}"):
msg = f'FQDN "{input_fqdn}" is not available.'
raise ValueError(msg)
return self
user_input = yield CreateRouterForm
return user_input.dict()
@step("Create subscription")
def create_subscription(product: UUIDstr, partner: str) -> State:
"""Create a new subscription object."""
subscription = RouterInactive.from_product_id(product, get_partner_by_name(partner)["partner_id"])
return {
"subscription": subscription,
"subscription_id": subscription.subscription_id,
}
@step("Initialize subscription")
def initialize_subscription(
subscription: RouterInactive,
hostname: str,
ts_port: PortNumber,
router_site: str,
router_role: RouterRole,
vendor: Vendor,
) -> State:
"""Initialise the subscription object in the service database."""
subscription.router.router_ts_port = ts_port
subscription.router.router_site = Site.from_subscription(router_site).site
fqdn = generate_fqdn(
hostname,
subscription.router.router_site.site_name,
subscription.router.router_site.site_country_code,
)
subscription.router.router_fqdn = fqdn
subscription.router.router_role = router_role
subscription.router.router_access_via_ts = True
subscription.router.vendor = vendor
subscription.description = f"Router {fqdn}"
return {"subscription": subscription}
@step("Allocate loopback interfaces in IPAM")
def ipam_allocate_loopback(subscription: RouterInactive) -> State:
"""Allocate :term:`IPAM` resources for the loopback interface."""
fqdn = subscription.router.router_fqdn
if not fqdn:
msg = f"Router fqdn for subscription id {subscription.subscription_id} is missing!"
raise ValueError(msg)
loopback_v4, loopback_v6 = infoblox.allocate_host(f"lo0.{fqdn}", "LO", [fqdn], str(subscription.subscription_id))
subscription.router.router_lo_ipv4_address = loopback_v4
subscription.router.router_lo_ipv6_address = loopback_v6
subscription.router.router_lo_iso_address = iso_from_ipv4(subscription.router.router_lo_ipv4_address)
return {"subscription": subscription}
@step("Create NetBox Device")
def create_netbox_device(subscription: RouterInactive) -> State:
"""Create a new NOKIA device in Netbox."""
fqdn = subscription.router.router_fqdn
site_tier = subscription.router.router_site.site_tier if subscription.router.router_site else None
if not fqdn or not site_tier:
msg = f"FQDN and/or Site tier missing in router subscription {subscription.subscription_id}!"
raise ValueError(msg)
NetboxClient().create_device(fqdn, site_tier)
return {"subscription": subscription}
@step("Verify IPAM resources for loopback interface")
def verify_ipam_loopback(subscription: RouterInactive) -> None:
"""Validate the :term:`IPAM` resources for the loopback interface.
Raises an :class:`orchestrator.utils.errors.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)
@inputstep("Prompt to reboot", assignee=Assignee.SYSTEM)
def prompt_reboot_router(subscription: RouterInactive) -> FormGenerator:
"""Wait for confirmation from an operator that the router has been rebooted."""
class RebootPrompt(FormPage):
model_config = ConfigDict(title="Please reboot before continuing")
if subscription.router.router_site and subscription.router.router_site.site_ts_address:
info_label_1: Label = (
f"Base config has been deployed. Please log in via the console using https://"
f"{subscription.router.router_site.site_ts_address}."
)
else:
info_label_1 = "Base config has been deployed. Please log in via the console."
info_label_2: Label = "Reboot the router, and once it is up again, press submit to continue the workflow."
yield RebootPrompt
return {}
@inputstep("Prompt to test the console", assignee=Assignee.SYSTEM)
def prompt_console_login() -> FormGenerator:
"""Wait for confirmation from an operator that the router can be logged into."""
class ConsolePrompt(FormPage):
model_config = ConfigDict(title="Verify local authentication")
info_label_1: Label = (
"Verify that you are able to log in to the router via the console using the admin account."
)
info_label_2: Label = "Once this is done, press submit to continue the workflow."
yield ConsolePrompt
return {}
@inputstep("Prompt IMS insertion", assignee=Assignee.SYSTEM)
def prompt_insert_in_ims() -> FormGenerator:
"""Wait for confirmation from an operator that the router has been inserted in IMS."""
class IMSPrompt(FormPage):
model_config = ConfigDict(title="Update IMS mediation server")
info_label_1: Label = "Insert the router into IMS."
info_label_2: Label = "Once this is done, press submit to continue the workflow."
yield IMSPrompt
return {}
@inputstep("Prompt RADIUS insertion", assignee=Assignee.SYSTEM)
def prompt_insert_in_radius(subscription: RouterInactive) -> FormGenerator:
"""Wait for confirmation from an operator that the router has been inserted in RADIUS."""
class RadiusPrompt(FormPage):
model_config = ConfigDict(title="Update RADIUS clients")
info_label_1: Label = (
f"Please go to https://kratos.geant.org/add_radius_client and add the {subscription.router.router_fqdn}"
f" - {subscription.router.router_lo_ipv4_address} to radius authentication"
)
info_label_2: Label = "This will be functionally checked later during verification work."
yield RadiusPrompt
return {}
@step("Create a new SharePoint checklist")
def create_new_sharepoint_checklist(subscription: RouterProvisioning, tt_number: str, process_id: UUIDstr) -> State:
"""Create a new checklist in SharePoint for approving this router."""
new_list_item_url = SharePointClient().add_list_item(
list_name="p_router",
fields={
"Title": subscription.router.router_fqdn,
"TT_NUMBER": tt_number,
"GAP_PROCESS_URL": f"{load_oss_params().GENERAL.public_hostname}/workflows/{process_id}",
},
)
return {"checklist_url": new_list_item_url}
@workflow(
"Create router",
initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
target=Target.CREATE,
)
def create_router() -> StepList:
"""Create a new router in the service database.
* Create and initialise the subscription object in the service database
* Allocate :term:`IPAM` resources for the loopback interface
* Deploy configuration on the new router, first as a dry run
* Validate :term:`IPAM` resources
* Create a new device in Netbox
"""
router_is_nokia = conditional(lambda state: state["vendor"] == Vendor.NOKIA)
return (
init
>> create_subscription
>> store_process_subscription(Target.CREATE)
>> initialize_subscription
>> ipam_allocate_loopback
>> lso_interaction(deploy_base_config_dry)
>> lso_interaction(deploy_base_config_real)
>> verify_ipam_loopback
>> prompt_reboot_router
>> prompt_console_login
>> prompt_insert_in_ims
>> prompt_insert_in_radius
>> router_is_nokia(create_netbox_device)
>> lso_interaction(run_checks_after_base_config)
>> set_status(SubscriptionLifecycle.PROVISIONING)
>> create_new_sharepoint_checklist
>> prompt_sharepoint_checklist_url
>> resync
>> done
)