Modify edge port

Modify an existing edge port subscription.

initial_input_form_generator(subscription_id)

Gather input from the operator on what to change about the selected edge port subscription.

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

def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Gather input from the operator on what to change about the selected edge port subscription."""
    subscription = EdgePort.from_subscription(subscription_id)

    class ModifyEdgePortForm(FormPage):
        model_config = ConfigDict(title="Modify Edge Port")

        tt_number: TTNumber
        enable_lacp: bool = subscription.edge_port.enable_lacp
        member_speed: PhysicalPortCapacity = subscription.edge_port.member_speed
        encapsulation: EncapsulationType = subscription.edge_port.encapsulation
        number_of_members: int = len(subscription.edge_port.edge_port_ae_members)
        minimum_links: int | None = subscription.edge_port.minimum_links or None
        mac_address: str | None = subscription.edge_port.mac_address or None
        ignore_if_down: bool = subscription.edge_port.ignore_if_down
        custom_service_name: str | None = subscription.edge_port.custom_service_name
        ga_id: (
            Annotated[
                str, AfterValidator(partial(validate_field_is_unique, subscription_id)), Field(pattern=r"^GA-\d{5}$")
            ]
            | None
        ) = subscription.edge_port.ga_id or 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

    user_input = yield ModifyEdgePortForm

    class EdgePortLAGMember(LAGMember):
        interface_name: (  # type: ignore[valid-type]
            available_interfaces_choices_including_current_members(
                subscription.edge_port.node.owner_subscription_id,
                user_input.member_speed,
                subscription.edge_port.edge_port_ae_members,
            )
            if user_input.member_speed == subscription.edge_port.member_speed
            else (
                available_interfaces_choices(subscription.edge_port.node.owner_subscription_id, user_input.member_speed)
            )
        )

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

    current_lag_ae_members = (
        [
            EdgePortLAGMember(
                interface_name=iface.interface_name,
                interface_description=iface.interface_description,
            )
            for iface in subscription.edge_port.edge_port_ae_members
        ]
        if user_input.member_speed == subscription.edge_port.member_speed
        else []
    )

    class ModifyEdgePortInterfaceForm(SubmitFormPage):
        model_config = ConfigDict(title="Modify Edge Port Interface")

        name: ReadOnlyField(subscription.edge_port.edge_port_name, default_type=str)  # type: ignore[valid-type]
        description: str | None = subscription.edge_port.edge_port_description or None
        ae_members: lag_ae_members = current_lag_ae_members

    interface_form_input = yield ModifyEdgePortInterfaceForm

    capacity_has_changed = (
        user_input.member_speed != subscription.edge_port.member_speed
        or user_input.number_of_members != len(subscription.edge_port.edge_port_ae_members)
        or any(
            old_interface.interface_name
            not in [new_interface.interface_name for new_interface in interface_form_input.ae_members]
            for old_interface in subscription.edge_port.edge_port_ae_members
        )
        or len(subscription.edge_port.edge_port_ae_members) != len(interface_form_input.ae_members)
    )
    return user_input.model_dump() | interface_form_input.model_dump() | {"capacity_has_changed": capacity_has_changed}

modify_edge_port_subscription(subscription, member_speed, encapsulation, minimum_links, mac_address, ga_id, enable_lacp, ae_members, ignore_if_down, description=None, custom_service_name=None)

Modify the edge port subscription with the given parameters.

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

@step("Modify edge port subscription.")
def modify_edge_port_subscription(
    subscription: EdgePort,
    member_speed: PhysicalPortCapacity,
    encapsulation: EncapsulationType,
    minimum_links: int,
    mac_address: str | None,
    ga_id: str | None,
    enable_lacp: bool,  # noqa: FBT001
    ae_members: list[dict[str, str]],
    ignore_if_down: bool,  # noqa: FBT001
    description: str | None = None,
    custom_service_name: str | None = None,
) -> State:
    """Modify the edge port subscription with the given parameters."""
    previous_ae_members = [
        {
            "interface_name": member.interface_name,
            "interface_description": member.interface_description,
        }
        for member in subscription.edge_port.edge_port_ae_members
    ]
    removed_ae_members = [member for member in previous_ae_members if member not in ae_members]
    subscription.edge_port.enable_lacp = enable_lacp
    subscription.edge_port.member_speed = member_speed
    subscription.edge_port.encapsulation = encapsulation
    subscription.edge_port.minimum_links = minimum_links
    subscription.edge_port.mac_address = mac_address
    subscription.edge_port.ignore_if_down = ignore_if_down
    subscription.edge_port.ga_id = ga_id
    subscription.edge_port.edge_port_description = description
    subscription.edge_port.custom_service_name = custom_service_name
    subscription.description = (
        f"Edge Port {subscription.edge_port.edge_port_name} on"
        f" {subscription.edge_port.node.router_fqdn},"
        f" {get_partner_by_id(subscription.customer_id).name}, {ga_id or ""}"
    )
    subscription.edge_port.edge_port_ae_members.clear()
    partner_name = get_partner_by_id(subscription.customer_id).name
    for member in ae_members:
        subscription.edge_port.edge_port_ae_members.append(EdgePortAEMemberBlock.new(subscription_id=uuid4(), **member))

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

update_interfaces_in_netbox(subscription, removed_ae_members, previous_ae_members)

Update the interfaces in NetBox.

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

@step("Update interfaces in NetBox")
def update_interfaces_in_netbox(
    subscription: EdgePort, removed_ae_members: list[dict], previous_ae_members: list[dict]
) -> State:
    """Update the interfaces in NetBox."""
    nbclient = NetboxClient()
    # Free removed interfaces
    for removed_member in removed_ae_members:
        nbclient.free_interface(subscription.edge_port.node.router_fqdn, removed_member["interface_name"])
    # Attach physical interfaces to LAG
    # Update interface description to subscription ID
    # Reserve interfaces
    for member in subscription.edge_port.edge_port_ae_members:
        if any(prev_member["interface_name"] == member.interface_name for prev_member in previous_ae_members):
            continue
        nbclient.attach_interface_to_lag(
            device_name=subscription.edge_port.node.router_fqdn,
            lag_name=subscription.edge_port.edge_port_name,
            iface_name=member.interface_name,
            description=str(subscription.subscription_id),
        )
        nbclient.reserve_interface(subscription.edge_port.node.router_fqdn, member.interface_name)

    return {"subscription": subscription}

update_edge_port_dry(subscription, process_id, tt_number, removed_ae_members, partner_name)

Perform a dry run of updating the edge port configuration.

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

@step("[DRY RUN] Update edge port configuration.")
def update_edge_port_dry(
    subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, removed_ae_members: list[dict], partner_name: str
) -> LSOState:
    """Perform a dry run of updating the edge port configuration."""
    extra_vars = {
        "subscription": subscription,
        "partner_name": partner_name,
        "dry_run": True,
        "verb": "update",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} "
        f"- Update Edge Port {subscription["edge_port"]["edge_port_name"]}"
        f" on {subscription["edge_port"]["node"]["router_fqdn"]}",
        "removed_ae_members": removed_ae_members,
    }

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

update_edge_port_real(subscription, process_id, tt_number, removed_ae_members, partner_name)

Update the edge port configuration.

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

@step("[FOR REAL] Update edge port configuration.")
def update_edge_port_real(
    subscription: dict[str, Any], process_id: UUIDstr, tt_number: str, removed_ae_members: list[str], partner_name: str
) -> LSOState:
    """Update the edge port configuration."""
    extra_vars = {
        "subscription": subscription,
        "partner_name": partner_name,
        "dry_run": False,
        "verb": "update",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} "
        f"- Update Edge Port {subscription["edge_port"]["edge_port_name"]}"
        f" on {subscription["edge_port"]["node"]["router_fqdn"]}",
        "removed_ae_members": removed_ae_members,
    }

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

allocate_interfaces_in_netbox(subscription, previous_ae_members)

Allocate the new interfaces in NetBox and detach the old ones from the LAG.

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

@step("Allocate/Deallocate interfaces in NetBox")
def allocate_interfaces_in_netbox(subscription: EdgePort, previous_ae_members: list[dict]) -> None:
    """Allocate the new interfaces in NetBox and detach the old ones from the LAG."""
    nbclient = NetboxClient()
    for member in subscription.edge_port.edge_port_ae_members:
        if any(member.interface_name == prev_member["interface_name"] for prev_member in previous_ae_members):
            continue
        nbclient.allocate_interface(
            device_name=subscription.edge_port.node.router_fqdn,
            iface_name=member.interface_name,
        )

    # detach the old interfaces from LAG
    nbclient.detach_interfaces_from_lag(
        device_name=subscription.edge_port.node.router_fqdn, lag_name=subscription.edge_port.edge_port_name
    )

modify_edge_port()

Modify a new edge port in the network.

• Modify the subscription object in the service database
• Modify configuration on the new edge port, first as a dry run
• Change LAG and LAG members in the Netbox.

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

@workflow(
    "Modify Edge Port",
    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
    target=Target.MODIFY,
)
def modify_edge_port() -> StepList:
    """Modify a new edge port in the network.

    * Modify the subscription object in the service database
    * Modify configuration on the new edge port, first as a dry run
    * Change LAG and LAG members in the Netbox.
    """
    capacity_has_changed = conditional(lambda state: state["capacity_has_changed"])
    return (
        begin
        >> store_process_subscription(Target.MODIFY)
        >> unsync
        >> modify_edge_port_subscription
        >> capacity_has_changed(update_interfaces_in_netbox)
        >> capacity_has_changed(lso_interaction(update_edge_port_dry))
        >> capacity_has_changed(lso_interaction(update_edge_port_real))
        >> capacity_has_changed(allocate_interfaces_in_netbox)
        >> resync
        >> done
    )