Create edge port

A creation workflow for adding a new edge port to the network.

initial_input_form_generator(product_name)

Gather information to create a new Edge Port.

Source code in gso/workflows/edge_port/create_edge_port.py

def initial_input_form_generator(product_name: str) -> FormGenerator:
    """Gather information to create a new Edge Port."""

    class CreateEdgePortForm(FormPage):
        model_config = ConfigDict(title=product_name)

        tt_number: TTNumber
        node: active_pe_router_selector()  # type: ignore[valid-type]
        partner: partner_choice()  # type: ignore[valid-type]
        service_type: EdgePortType
        enable_lacp: bool = False
        speed: PhysicalPortCapacity
        encapsulation: EncapsulationType = EncapsulationType.DOT1Q
        number_of_members: int
        minimum_links: int
        mac_address: str | None = None
        ignore_if_down: bool = False
        custom_service_name: str | None = None
        generate_ga_id: bool = True
        ga_id: IMPORTED_GA_ID | None = None

        @model_validator(mode="after")
        def validate_number_of_members(self) -> Self:
            validate_edge_port_number_of_members_based_on_lacp(
                enable_lacp=self.enable_lacp, number_of_members=self.number_of_members
            )
            return self

        @model_validator(mode="before")
        def validate_ga_id(cls, input_data: dict[str, Any]) -> dict[str, Any]:
            ga_id = input_data.get("ga_id")
            generate_ga_id = input_data.get("generate_ga_id", True)

            if generate_ga_id and ga_id:
                error_message = (
                    "You cannot provide a GA ID manually while the 'Auto-generate GA ID' option is enabled."
                    "Please either uncheck 'Auto-generate GA ID' or remove the manual GA ID."
                )
                raise ValueError(error_message)
            return input_data

    initial_user_input = yield CreateEdgePortForm

    class EdgePortLAGMember(LAGMember):
        interface_name: available_interfaces_choices(  # type: ignore[valid-type]
            initial_user_input.node, initial_user_input.speed
        )

    lag_ae_members = Annotated[
        list[EdgePortLAGMember],
        AfterValidator(validate_unique_list),
        Len(
            min_length=initial_user_input.number_of_members,
            max_length=initial_user_input.number_of_members,
        ),
    ]

    class SelectInterfaceForm(FormPage):
        model_config = ConfigDict(title="Select Interfaces")

        name: available_service_lags_choices(initial_user_input.node)  # type: ignore[valid-type]
        description: str | None = None
        ae_members: lag_ae_members

    interface_form_input_data = yield SelectInterfaceForm

    input_forms_data = initial_user_input.model_dump() | interface_form_input_data.model_dump()
    summary_form_data = input_forms_data | {
        "node": Router.from_subscription(initial_user_input.node).router.router_fqdn,
        "partner": get_partner_by_id(initial_user_input.partner).name,
        "edge_port_ae_members": input_forms_data["ae_members"],
        "edge_port_name": input_forms_data["name"],
        "edge_port_description": input_forms_data["description"],
        "edge_port_type": input_forms_data["service_type"],
        "custom_service_name": input_forms_data["custom_service_name"],
    }
    summary_fields = [
        "node",
        "partner",
        "edge_port_type",
        "speed",
        "encapsulation",
        "minimum_links",
        "mac_address",
        "ignore_if_down",
        "enable_lacp",
        "edge_port_name",
        "edge_port_description",
        "edge_port_ae_members",
        "custom_service_name",
    ]
    yield from create_summary_form(summary_form_data, product_name, summary_fields)
    return input_forms_data

create_subscription(product, partner)

Create a new subscription object.

Source code in gso/workflows/edge_port/create_edge_port.py

@step("Create subscription")
def create_subscription(product: UUIDstr, partner: UUIDstr) -> State:
    """Create a new subscription object."""
    subscription = EdgePortInactive.from_product_id(product, partner)

    return {
        "subscription": subscription,
        "subscription_id": subscription.subscription_id,
    }

initialize_subscription(subscription, node, service_type, speed, encapsulation, name, minimum_links, mac_address, partner, enable_lacp, ignore_if_down, generate_ga_id, ae_members, description=None, custom_service_name=None, ga_id=None)

Initialise the subscription object in the service database.

Source code in gso/workflows/edge_port/create_edge_port.py

@step("Initialize subscription")
def initialize_subscription(
    subscription: EdgePortInactive,
    node: UUIDstr,
    service_type: EdgePortType,
    speed: PhysicalPortCapacity,
    encapsulation: EncapsulationType,
    name: str,
    minimum_links: int,
    mac_address: str | None,
    partner: str,
    enable_lacp: bool,  # noqa: FBT001
    ignore_if_down: bool,  # noqa: FBT001
    generate_ga_id: bool,  # noqa: FBT001
    ae_members: list[dict[str, Any]],
    description: str | None = None,
    custom_service_name: str | None = None,
    ga_id: str | None = None,
) -> State:
    """Initialise the subscription object in the service database."""
    router = Router.from_subscription(node).router
    subscription.edge_port.node = router
    subscription.edge_port.edge_port_type = service_type
    subscription.edge_port.enable_lacp = enable_lacp
    subscription.edge_port.member_speed = speed
    subscription.edge_port.encapsulation = encapsulation
    subscription.edge_port.edge_port_name = name
    subscription.edge_port.minimum_links = minimum_links
    subscription.edge_port.ignore_if_down = ignore_if_down
    ga_id = generate_unique_id(prefix="GA") if generate_ga_id else ga_id
    subscription.edge_port.ga_id = ga_id
    subscription.edge_port.mac_address = mac_address
    partner_name = get_partner_by_id(partner).name
    subscription.description = f"Edge Port {name} on {router.router_fqdn}, {partner_name}, {ga_id or ""}"
    subscription.edge_port.edge_port_description = description
    subscription.edge_port.custom_service_name = custom_service_name
    for member in ae_members:
        subscription.edge_port.edge_port_ae_members.append(
            EdgePortAEMemberBlockInactive.new(subscription_id=uuid4(), **member)
        )
    subscription = EdgePortProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)

    return {"subscription": subscription, "partner_name": partner_name}

reserve_interfaces_in_netbox(subscription)

Create the LAG interfaces in NetBox and attach the LAG interfaces to the physical interfaces.

Source code in gso/workflows/edge_port/create_edge_port.py

@step("Reserve interfaces in NetBox")
def reserve_interfaces_in_netbox(subscription: EdgePortProvisioning) -> State:
    """Create the LAG interfaces in NetBox and attach the LAG interfaces to the physical interfaces."""
    nbclient = NetboxClient()
    edge_port = subscription.edge_port
    # Create LAG interfaces
    lag_interface: Interfaces = nbclient.create_interface(
        iface_name=edge_port.edge_port_name,
        interface_type="lag",
        device_name=edge_port.node.router_fqdn,
        description=str(subscription.subscription_id),
        enabled=True,
    )
    # Attach physical interfaces to LAG
    # Update interface description to subscription ID
    # Reserve interfaces
    for interface in edge_port.edge_port_ae_members:
        nbclient.attach_interface_to_lag(
            device_name=edge_port.node.router_fqdn,
            lag_name=lag_interface.name,
            iface_name=interface.interface_name,
            description=str(subscription.subscription_id),
        )
        nbclient.reserve_interface(
            device_name=edge_port.node.router_fqdn,
            iface_name=interface.interface_name,
        )
    return {
        "subscription": subscription,
    }

allocate_interfaces_in_netbox(subscription)

Allocate the interfaces in NetBox.

Source code in gso/workflows/edge_port/create_edge_port.py

@step("Allocate interfaces in NetBox")
def allocate_interfaces_in_netbox(subscription: EdgePortProvisioning) -> None:
    """Allocate the interfaces in NetBox."""
    for interface in subscription.edge_port.edge_port_ae_members:
        fqdn = subscription.edge_port.node.router_fqdn
        iface_name = interface.interface_name
        if not fqdn or not iface_name:
            msg = "FQDN and/or interface name missing in subscription"
            raise ProcessFailureError(msg, details=subscription.subscription_id)

        NetboxClient().allocate_interface(device_name=fqdn, iface_name=iface_name)

create_edge_port_dry(subscription, tt_number, process_id, partner_name)

Create a new edge port in the network as a dry run.

Source code in gso/workflows/edge_port/create_edge_port.py

@step("[DRY RUN] Create edge port")
def create_edge_port_dry(
    subscription: dict[str, Any], tt_number: str, process_id: UUIDstr, partner_name: str
) -> LSOState:
    """Create a new edge port in the network as a dry run."""
    extra_vars = {
        "dry_run": True,
        "subscription": subscription,
        "partner_name": partner_name,
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Create Edge Port",
        "verb": "create",
    }

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

create_edge_port_real(subscription, tt_number, process_id, partner_name)

Create a new edge port in the network for real.

Source code in gso/workflows/edge_port/create_edge_port.py

@step("[FOR REAL] Create edge port")
def create_edge_port_real(
    subscription: dict[str, Any], tt_number: str, process_id: UUIDstr, partner_name: str
) -> LSOState:
    """Create a new edge port in the network for real."""
    extra_vars = {
        "dry_run": False,
        "subscription": subscription,
        "partner_name": partner_name,
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Create Edge Port",
        "verb": "create",
    }

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

create_edge_port()

Create a new edge port in the network.

• Create and initialise the subscription object in the service database
• Deploy configuration on the new edge port, first as a dry run
• allocate LAG and LAG members in the Netbox.

Source code in gso/workflows/edge_port/create_edge_port.py

@workflow(
    "Create Edge Port",
    initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
    target=Target.CREATE,
)
def create_edge_port() -> StepList:
    """Create a new edge port in the network.

    * Create and initialise the subscription object in the service database
    * Deploy configuration on the new edge port, first as a dry run
    * allocate LAG and LAG members in the Netbox.
    """
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> start_moodi()
        >> reserve_interfaces_in_netbox
        >> lso_interaction(create_edge_port_dry)
        >> lso_interaction(create_edge_port_real)
        >> allocate_interfaces_in_netbox
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> stop_moodi()
        >> done
    )