Skip to content

GÉANT Automation Platform

Modify trunk interface

geant-service-orchestrator

Modify trunk interface

A modification workflow that updates the LAG interfaces that are part of an existing IP trunk.

Modifies LAG interfaces and members. This is used to increase capacity or to change SID/interface descriptions.

The strategy is to re-apply the necessary template to the configuration construct: using a "replace" strategy only the necessary modifications will be applied.

`initialize_ae_members(subscription, initial_user_input, side_index)`

Initialize the list of AE members.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

def initialize_ae_members(
    subscription: Iptrunk, initial_user_input: dict, side_index: int
) -> type[LAGMemberList[LAGMember]]:
    """Initialize the list of AE members."""
    router = subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_node
    router_vendor = get_router_vendor(router.owner_subscription_id)
    iptrunk_number_of_members = initial_user_input["iptrunk_number_of_members"]
    if router_vendor == Vendor.NOKIA:
        iptrunk_speed = initial_user_input["iptrunk_speed"]

        class NokiaLAGMember(LAGMember):
            interface_name: (  # type: ignore[valid-type]
                available_interfaces_choices_including_current_members(
                    router.owner_subscription_id,
                    iptrunk_speed,
                    subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_ae_members,
                )
                if iptrunk_speed == subscription.iptrunk.iptrunk_speed
                else (
                    available_interfaces_choices(
                        router.owner_subscription_id,
                        initial_user_input["iptrunk_speed"],
                    )
                )
            )

        return Annotated[  # type: ignore[return-value]
            LAGMemberList[NokiaLAGMember],
            Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
        ]

    return Annotated[  # type: ignore[return-value]
        LAGMemberList[JuniperLAGMember],
        Len(min_length=iptrunk_number_of_members, max_length=iptrunk_number_of_members),
    ]

`initial_input_form_generator(subscription_id)`

Gather input from the operator on the interfaces that should be modified.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Gather input from the operator on the interfaces that should be modified."""
    subscription = Iptrunk.from_subscription(subscription_id)

    class ModifyIptrunkForm(FormPage):
        tt_number: TTNumber
        gs_id: (
            Annotated[
                str, AfterValidator(partial(validate_field_is_unique, subscription_id)), Field(pattern=r"^GS-\d{5}$")
            ]
            | None
        ) = subscription.iptrunk.gs_id
        iptrunk_description: str | None = subscription.iptrunk.iptrunk_description
        iptrunk_type: IptrunkType = subscription.iptrunk.iptrunk_type
        warning_label: Label = (
            "Changing the PhyPortCapacity will result in the deletion of all AE members. "
            "You will need to add the new AE members in the next steps."
        )
        iptrunk_speed: PhysicalPortCapacity = subscription.iptrunk.iptrunk_speed
        iptrunk_number_of_members: int = len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members)
        iptrunk_isis_metric: ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric, default_type=int)  # type: ignore[valid-type]
        iptrunk_ipv4_network: ReadOnlyField(  # type: ignore[valid-type]
            str(subscription.iptrunk.iptrunk_ipv4_network), default_type=IPv4AddressType
        )
        iptrunk_ipv6_network: ReadOnlyField(  # type: ignore[valid-type]
            str(subscription.iptrunk.iptrunk_ipv6_network), default_type=IPv6AddressType
        )

    initial_user_input = yield ModifyIptrunkForm

    recommended_minimum_links = calculate_recommended_minimum_links(
        initial_user_input.iptrunk_number_of_members, initial_user_input.iptrunk_speed
    )

    class VerifyMinimumLinksForm(FormPage):
        info_label: Label = f"Current value of minimum-links : {subscription.iptrunk.iptrunk_minimum_links}"
        info_label1: Label = f"Recommended minimum-links for this LAG: {recommended_minimum_links}"
        iptrunk_minimum_links: int = recommended_minimum_links
        info_label2: Label = "Please review the recommended value and adjust if necessary."

    verify_minimum_links = yield VerifyMinimumLinksForm
    ae_members_side_a = initialize_ae_members(subscription, initial_user_input.model_dump(), 0)

    class ModifyIptrunkSideAForm(FormPage):
        model_config = ConfigDict(title="Provide subscription details for side A of the trunk.")

        side_a_node: ReadOnlyField(  # type: ignore[valid-type]
            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str
        )
        side_a_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface, default_type=str)  # type: ignore[valid-type]
        side_a_ga_id: str | None = subscription.iptrunk.iptrunk_sides[0].ga_id
        side_a_ae_members: ae_members_side_a = (  # type: ignore[valid-type]
            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members
            if initial_user_input.iptrunk_speed == subscription.iptrunk.iptrunk_speed
            else []
        )

    user_input_side_a = yield ModifyIptrunkSideAForm
    ae_members_side_b = initialize_ae_members(subscription, initial_user_input.model_dump(), 1)

    class ModifyIptrunkSideBForm(SubmitFormPage):
        model_config = ConfigDict(title="Provide subscription details for side B of the trunk.")

        side_b_node: ReadOnlyField(  # type: ignore[valid-type]
            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str
        )
        side_b_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface, default_type=str)  # type: ignore[valid-type]
        side_b_ga_id: str | None = subscription.iptrunk.iptrunk_sides[1].ga_id
        side_b_ae_members: ae_members_side_b = (  # type: ignore[valid-type]
            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members
            if initial_user_input.iptrunk_speed == subscription.iptrunk.iptrunk_speed
            else []
        )

    user_input_side_b = yield ModifyIptrunkSideBForm

    return (
        initial_user_input.model_dump()
        | user_input_side_a.model_dump()
        | user_input_side_b.model_dump()
        | verify_minimum_links.model_dump()
    )

`determine_change_in_capacity(subscription, iptrunk_speed, side_a_ae_members, side_b_ae_members)`

Determine whether we should run pre- and post-checks on the IP trunk.

This can be caused by the following conditions

The total capacity of the trunk changes
The amount of interfaces changes
One or more interface names have changed on side A
One or more interface names have changed on side B

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Determine whether we should be running interface checks")
def determine_change_in_capacity(
    subscription: Iptrunk, iptrunk_speed: str, side_a_ae_members: list[dict], side_b_ae_members: list[dict]
) -> State:
    """Determine whether we should run pre- and post-checks on the IP trunk.

    This can be caused by the following conditions:
     * The total capacity of the trunk changes
     * The amount of interfaces changes
     * One or more interface names have changed on side A
     * One or more interface names have changed on side B
    """
    capacity_has_changed = (
        iptrunk_speed != subscription.iptrunk.iptrunk_speed
        or len(side_a_ae_members) != len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members)
        or any(
            old_interface.interface_name != new_interface["interface_name"]
            for old_interface, new_interface in zip(
                subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members, side_a_ae_members, strict=False
            )
        )
        or any(
            old_interface.interface_name != new_interface["interface_name"]
            for old_interface, new_interface in zip(
                subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members, side_b_ae_members, strict=False
            )
        )
    )

    return {"capacity_has_changed": capacity_has_changed}

`check_ip_trunk_connectivity(subscription)`

Check successful connectivity across a trunk.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Check IP connectivity of the trunk")
def check_ip_trunk_connectivity(subscription: Iptrunk) -> LSOState:
    """Check successful connectivity across a trunk."""
    extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "ping"}

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks_checks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

`check_ip_trunk_lldp(subscription)`

Check LLDP on trunk endpoints.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Check LLDP on the trunk endpoints")
def check_ip_trunk_lldp(subscription: Iptrunk) -> LSOState:
    """Check LLDP on trunk endpoints."""
    extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "lldp"}

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks_checks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

`update_side_members(subscription, side_index, new_members)`

Update the AE members for a given side without removing unchanged members.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

def update_side_members(subscription: Iptrunk, side_index: int, new_members: list[dict]) -> None:
    """Update the AE members for a given side without removing unchanged members."""
    # Prepare a dictionary for quick lookup of existing members by name
    current_members = subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_ae_members
    existing_members_dict = {member.interface_name: member for member in current_members}

    # Iterate over new members and update or add them
    for new_member in new_members:
        interface_name = new_member["interface_name"]
        if interface_name in existing_members_dict:
            # Member exists, update details but keep the same subscription ID
            existing_member = existing_members_dict[interface_name]
            existing_member.interface_description = new_member["interface_description"]
        else:
            # New member, create a new subscription ID
            current_members.append(IptrunkInterfaceBlock.new(subscription_id=uuid4(), **new_member))

    # Remove members that are no longer in the new members list
    subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_ae_members = [
        member for member in current_members if member.interface_name in [m["interface_name"] for m in new_members]
    ]

`modify_iptrunk_subscription(subscription, gs_id, iptrunk_type, iptrunk_description, iptrunk_speed, iptrunk_minimum_links, side_a_ga_id, side_a_ae_members, side_b_ga_id, side_b_ae_members)`

Modify the subscription in the service database, reflecting the changes to the newly selected interfaces.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Update subscription")
def modify_iptrunk_subscription(
    subscription: Iptrunk,
    gs_id: str | None,
    iptrunk_type: IptrunkType,
    iptrunk_description: str | None,
    iptrunk_speed: PhysicalPortCapacity,
    iptrunk_minimum_links: int,
    side_a_ga_id: str | None,
    side_a_ae_members: list[dict],
    side_b_ga_id: str | None,
    side_b_ae_members: list[dict],
) -> State:
    """Modify the subscription in the service database, reflecting the changes to the newly selected interfaces."""
    # Prepare the list of removed AE members
    previous_ae_members = [
        [
            {
                "interface_name": member.interface_name,
                "interface_description": member.interface_description,
            }
            for member in side.iptrunk_side_ae_members
        ]
        for side in subscription.iptrunk.iptrunk_sides
    ]
    removed_ae_members = []
    # Compare previous and current members to determine which ones were removed
    for side_index in range(2):
        previous_members = previous_ae_members[side_index]
        current_members = side_a_ae_members if side_index == 0 else side_b_ae_members
        removed_ae_members.append([
            ae_member
            for ae_member in previous_members
            if ae_member["interface_name"] not in [m["interface_name"] for m in current_members]
        ])
    # Update the subscription
    subscription.iptrunk.gs_id = gs_id
    subscription.iptrunk.iptrunk_description = iptrunk_description
    subscription.iptrunk.iptrunk_type = iptrunk_type
    subscription.iptrunk.iptrunk_speed = iptrunk_speed
    subscription.iptrunk.iptrunk_minimum_links = iptrunk_minimum_links

    subscription.iptrunk.iptrunk_sides[0].ga_id = side_a_ga_id
    update_side_members(subscription, 0, side_a_ae_members)
    subscription.iptrunk.iptrunk_sides[1].ga_id = side_b_ga_id
    update_side_members(subscription, 1, side_b_ae_members)

    side_names = sorted([
        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
    ])
    subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, {gs_id}"

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

`provision_ip_trunk_iface_dry(subscription, process_id, tt_number, removed_ae_members)`

Perform a dry run of deploying the updated IP trunk.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("[DRY RUN] Provision IP trunk interface")
def provision_ip_trunk_iface_dry(
    subscription: Iptrunk, process_id: UUIDstr, tt_number: str, removed_ae_members: list[str]
) -> LSOState:
    """Perform a dry run of deploying the updated IP trunk."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": True,
        "verb": "deploy",
        "config_object": "trunk_interface",
        "removed_ae_members": removed_ae_members,
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
        f"{subscription.iptrunk.gs_id}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

`provision_ip_trunk_iface_real(subscription, process_id, tt_number, removed_ae_members)`

Provision the new IP trunk with updated interfaces.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("[FOR REAL] Provision IP trunk interface")
def provision_ip_trunk_iface_real(
    subscription: Iptrunk, process_id: UUIDstr, tt_number: str, removed_ae_members: list[str]
) -> LSOState:
    """Provision the new IP trunk with updated interfaces."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": False,
        "verb": "deploy",
        "config_object": "trunk_interface",
        "removed_ae_members": removed_ae_members,
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - Deploy config for "
        f"{subscription.iptrunk.gs_id}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

`netbox_update_interfaces_side_a(subscription, removed_ae_members, previous_ae_members)`

Update Netbox such that it contains the new interfaces on side A.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Netbox: Reserve side A interfaces")
def netbox_update_interfaces_side_a(
    subscription: Iptrunk, removed_ae_members: list[list[dict]], previous_ae_members: list[list[dict]]
) -> None:
    """Update Netbox such that it contains the new interfaces on side A."""
    _netbox_update_interfaces(
        subscription.subscription_id,
        subscription.iptrunk.iptrunk_sides[0],
        removed_ae_members[0],
        previous_ae_members[0],
    )

`netbox_update_interfaces_side_b(subscription, removed_ae_members, previous_ae_members)`

Update Netbox such that it contains the new interfaces on side B.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Netbox: Reserve side B interfaces")
def netbox_update_interfaces_side_b(
    subscription: Iptrunk, removed_ae_members: list[list[dict]], previous_ae_members: list[list[dict]]
) -> None:
    """Update Netbox such that it contains the new interfaces on side B."""
    _netbox_update_interfaces(
        subscription.subscription_id,
        subscription.iptrunk.iptrunk_sides[1],
        removed_ae_members[1],
        previous_ae_members[1],
    )

`allocate_interfaces_in_netbox_side_a(subscription, previous_ae_members)`

Allocate the LAG interfaces on side A in Netbox.

Attach the LAG interface to the physical interface detach old one from the LAG.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Netbox: Allocate side A interfaces")
def allocate_interfaces_in_netbox_side_a(subscription: Iptrunk, previous_ae_members: list[list[dict]]) -> None:
    """Allocate the LAG interfaces on side A in Netbox.

    Attach the LAG interface to the physical interface detach old one from the LAG.
    """
    _netbox_allocate_interfaces(subscription.iptrunk.iptrunk_sides[0], previous_ae_members[0])

`allocate_interfaces_in_netbox_side_b(subscription, previous_ae_members)`

Allocate the LAG interface on side B in Netbox.

Attach the LAG interface to the physical interface detach old one from the LAG.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Netbox: Allocate side B interfaces")
def allocate_interfaces_in_netbox_side_b(subscription: Iptrunk, previous_ae_members: list[list[dict]]) -> None:
    """Allocate the LAG interface on side B in Netbox.

    Attach the LAG interface to the physical interface detach old one from the LAG.
    """
    _netbox_allocate_interfaces(subscription.iptrunk.iptrunk_sides[1], previous_ae_members[1])

`check_ip_trunk_optical_levels_post(subscription)`

Check Optical POST levels on the trunk.

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@step("Check Optical POST levels on the trunk endpoint")
def check_ip_trunk_optical_levels_post(subscription: Iptrunk) -> LSOState:
    """Check Optical POST levels on the trunk."""
    extra_vars = {"wfo_ip_trunk_json": json.loads(json_dumps(subscription)), "check": "optical_post"}

    return {
        "playbook_name": "gap_ansible/playbooks/iptrunks_checks.yaml",
        "inventory": {
            "all": {
                "hosts": {
                    subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn: None,
                    subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn: None,
                }
            }
        },
        "extra_vars": extra_vars,
    }

`modify_trunk_interface()`

Modify the interfaces that are part of an IP trunk.

Update the subscription in the database
Reserve new interfaces in Netbox
Provision the updated version of the IP trunk, first as a dry run
Allocate the reserved interfaces in Netbox

Source code in gso/workflows/iptrunk/modify_trunk_interface.py

@workflow(
    "Modify IP Trunk interface",
    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
    target=Target.MODIFY,
)
def modify_trunk_interface() -> StepList:
    """Modify the interfaces that are part of an IP trunk.

    * Update the subscription in the database
    * Reserve new interfaces in Netbox
    * Provision the updated version of the IP trunk, first as a dry run
    * Allocate the reserved interfaces in Netbox
    """
    side_a_is_nokia = conditional(
        lambda state: get_router_vendor(
            state["subscription"]["iptrunk"]["iptrunk_sides"][0]["iptrunk_side_node"]["owner_subscription_id"]
        )
        == Vendor.NOKIA
    )
    side_b_is_nokia = conditional(
        lambda state: get_router_vendor(
            state["subscription"]["iptrunk"]["iptrunk_sides"][1]["iptrunk_side_node"]["owner_subscription_id"]
        )
        == Vendor.NOKIA
    )
    capacity_has_changed = conditional(lambda state: state["capacity_has_changed"])

    return (
        begin
        >> store_process_subscription(Target.MODIFY)
        >> unsync
        >> determine_change_in_capacity
        >> capacity_has_changed(lso_interaction(check_ip_trunk_lldp))
        >> capacity_has_changed(lso_interaction(check_ip_trunk_optical_levels_pre))
        >> capacity_has_changed(lso_interaction(check_ip_trunk_connectivity))
        >> capacity_has_changed(lso_interaction(check_ip_trunk_isis))
        >> modify_iptrunk_subscription
        >> side_a_is_nokia(netbox_update_interfaces_side_a)
        >> side_b_is_nokia(netbox_update_interfaces_side_b)
        >> lso_interaction(provision_ip_trunk_iface_dry)
        >> lso_interaction(provision_ip_trunk_iface_real)
        >> side_a_is_nokia(allocate_interfaces_in_netbox_side_a)
        >> side_b_is_nokia(allocate_interfaces_in_netbox_side_b)
        >> capacity_has_changed(lso_interaction(check_ip_trunk_lldp))
        >> capacity_has_changed(lso_interaction(check_ip_trunk_optical_levels_post))
        >> capacity_has_changed(lso_interaction(check_ip_trunk_connectivity))
        >> capacity_has_changed(lso_interaction(check_ip_trunk_isis))
        >> resync
        >> done
    )