Terminate iptrunk

A termination workflow for an active IP trunk.

This workflow deletes all the configuration related with an IP trunk from the network and brings the subscription from ACTIVE to TERMINATED. The steps are the following:

• Modify the ISIS metric of the trunks so to evacuate traffic - and await confirmation from an operator.
• Delete all the configuration (first dry then actual deletion):
  • LAG and members of the LAG
  • reference in LLDP protocol (if Juniper)
  • reference in ISIS protocol
• Delete the IPv4/IPv6 networks from IPAM

initial_input_form_generator(subscription_id)

Ask the operator to confirm whether router configuration and IPAM resources should be deleted.

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

def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Ask the operator to confirm whether router configuration and IPAM resources should be deleted."""
    iptrunk = Iptrunk.from_subscription(subscription_id)

    class TerminateForm(SubmitFormPage):
        if iptrunk.status == SubscriptionLifecycle.INITIAL:
            info_label_2: Label = (
                "This will immediately mark the subscription as terminated, preventing any other workflows from "
                "interacting with this product subscription."
            )
            info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."

        tt_number: TTNumber
        termination_label: Label = (
            "Please confirm whether configuration should get removed from the A and B sides of the trunk."
        )
        remove_configuration: bool = True

    user_input = yield TerminateForm
    return user_input.model_dump()

deprovision_ip_trunk_dry(subscription, process_id, tt_number)

Perform a dry run of deleting configuration from the routers.

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

@step("[DRY RUN] Deprovision IP trunk")
def deprovision_ip_trunk_dry(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> LSOState:
    """Perform a dry run of deleting configuration from the routers."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": True,
        "verb": "terminate",
        "config_object": "trunk_deprovision",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} "
        f"- Remove config for {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,
    }

deprovision_ip_trunk_real(subscription, process_id, tt_number)

Delete configuration from the routers.

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

@step("[FOR REAL] Deprovision IP trunk")
def deprovision_ip_trunk_real(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> LSOState:
    """Delete configuration from the routers."""
    extra_vars = {
        "wfo_trunk_json": json.loads(json_dumps(subscription)),
        "dry_run": False,
        "verb": "terminate",
        "config_object": "trunk_deprovision",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} "
        f"- Remove config for {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_clean_up_side_a(subscription)

Mark used interfaces on side A as free in Netbox.

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

@step("Netbox: Remove interfaces on side A")
def netbox_clean_up_side_a(subscription: Iptrunk) -> State:
    """Mark used interfaces on side A as free in Netbox."""
    _free_up_interfaces_from_netbox(subscription.iptrunk.iptrunk_sides[0])

    return {"subscription": subscription}

netbox_clean_up_side_b(subscription)

Mark used interfaces on side B as free in Netbox.

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

@step("Netbox: Remove interfaces on side B")
def netbox_clean_up_side_b(subscription: Iptrunk) -> State:
    """Mark used interfaces on side B as free in Netbox."""
    _free_up_interfaces_from_netbox(subscription.iptrunk.iptrunk_sides[1])

    return {"subscription": subscription}

deprovision_ip_trunk_ipv4(subscription)

Clear up IPv4 resources in IPAM.

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

@step("Deprovision IPv4 networks")
def deprovision_ip_trunk_ipv4(subscription: Iptrunk) -> dict:
    """Clear up IPv4 resources in IPAM."""
    infoblox.delete_network(ipaddress.IPv4Network(subscription.iptrunk.iptrunk_ipv4_network))

    return {"subscription": subscription}

deprovision_ip_trunk_ipv6(subscription)

Clear up IPv6 resources in IPAM.

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

@step("Deprovision IPv6 networks")
def deprovision_ip_trunk_ipv6(subscription: Iptrunk) -> dict:
    """Clear up IPv6 resources in IPAM."""
    infoblox.delete_network(ipaddress.IPv6Network(subscription.iptrunk.iptrunk_ipv6_network))

    return {"subscription": subscription}

terminate_iptrunk()

Terminate an IP trunk.

• Let the operator decide whether to remove configuration from the routers, if so:
  • Set the ISIS metric of the IP trunk to an arbitrarily high value
  • Disable and remove configuration from the routers, first as a dry run
• Mark the IP trunk interfaces as free in Netbox
• Clear IPAM resources
• Terminate the subscription in the service database

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

@workflow(
    "Terminate IPtrunk",
    initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator),
    target=Target.TERMINATE,
)
def terminate_iptrunk() -> StepList:
    """Terminate an IP trunk.

    * Let the operator decide whether to remove configuration from the routers, if so:
        * Set the ISIS metric of the IP trunk to an arbitrarily high value
        * Disable and remove configuration from the routers, first as a dry run
    * Mark the IP trunk interfaces as free in Netbox
    * Clear IPAM resources
    * Terminate the subscription in the service database
    """
    run_config_steps = conditional(lambda state: state["remove_configuration"])
    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
    )

    config_steps = (
        begin
        >> lso_interaction(set_isis_to_max)
        >> lso_interaction(deprovision_ip_trunk_dry)
        >> lso_interaction(deprovision_ip_trunk_real)
    )

    return (
        begin
        >> store_process_subscription(Target.TERMINATE)
        >> unsync
        >> run_config_steps(config_steps)
        >> side_a_is_nokia(netbox_clean_up_side_a)
        >> side_b_is_nokia(netbox_clean_up_side_b)
        >> deprovision_ip_trunk_ipv4
        >> deprovision_ip_trunk_ipv6
        >> set_status(SubscriptionLifecycle.TERMINATED)
        >> resync
        >> done
    )