Create lan switch interconnect

A creation workflow for creating a new interconnect between a switch and a router.

create_subscription(product, partner)

Create a new subscription object in the database.

Source code in gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@step("Create subscription")
def create_subscription(product: UUIDstr, partner: str) -> State:
    """Create a new subscription object in the database."""
    subscription = LanSwitchInterconnectInactive.from_product_id(product, get_partner_by_name(partner).partner_id)

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

initialize_subscription(subscription, description, minimum_link_count, router_side, router_side_iface, router_side_ae_members, switch_side, switch_side_iface, switch_side_ae_members)

Update the product model with all input from the operator.

Source code in gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@step("Initialize subscription")
def initialize_subscription(
    subscription: LanSwitchInterconnectInactive,
    description: str,
    minimum_link_count: int,
    router_side: UUIDstr,
    router_side_iface: JuniperPhyInterface,
    router_side_ae_members: list[dict],
    switch_side: UUIDstr,
    switch_side_iface: JuniperPhyInterface,
    switch_side_ae_members: list[dict],
) -> State:
    """Update the product model with all input from the operator."""
    subscription.lan_switch_interconnect.lan_switch_interconnect_description = description
    subscription.lan_switch_interconnect.minimum_links = minimum_link_count
    subscription.lan_switch_interconnect.switch_management_vlan_id = DEFAULT_SWITCH_MANAGEMENT_VLAN_ID
    subscription.lan_switch_interconnect.router_side.node = Router.from_subscription(router_side).router
    subscription.lan_switch_interconnect.router_side.ae_iface = router_side_iface
    for member in router_side_ae_members:
        subscription.lan_switch_interconnect.router_side.ae_members.append(
            LanSwitchInterconnectInterfaceBlockInactive.new(subscription_id=uuid4(), **member)
        )
    subscription.lan_switch_interconnect.switch_side.switch = Switch.from_subscription(switch_side).switch
    subscription.lan_switch_interconnect.switch_side.ae_iface = switch_side_iface
    for member in switch_side_ae_members:
        subscription.lan_switch_interconnect.switch_side.ae_members.append(
            LanSwitchInterconnectInterfaceBlockInactive.new(subscription_id=uuid4(), **member)
        )
    if subscription.lan_switch_interconnect.router_side.node.router_site.site_contains_optical_equipment:
        subscription.lan_switch_interconnect.dcn_management_vlan_id = DEFAULT_DCN_MANAGEMENT_VLAN_ID

    return {"subscription": subscription}

register_dns_records_v4_network(subscription)

Add DNS records in IPAM.

Source code in gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@step("Register IPv4 network in IPAM")
def register_dns_records_v4_network(subscription: LanSwitchInterconnectInactive) -> State:
    """Add DNS records in IPAM."""
    router_site = subscription.lan_switch_interconnect.router_side.node.router_site
    if not router_site or not router_site.site_internal_id:
        msg = "Site internal ID not set. Cannot continue."
        raise ProcessFailureError(msg, details=router_site)

    new_network = generate_lan_switch_interconnect_subnet_v4(router_site.site_internal_id)
    ipam_oss_params = load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
    create_v4_network_by_ip(
        ipam_oss_params.dns_view, ipam_oss_params.network_view, new_network, str(subscription.subscription_id)
    )

    return {"ipam_registrations": {"v4": {"network": new_network}}}

register_dns_records_v4_devices(subscription, subscription_id, ipam_registrations)

Register DNS records for both switch and router side in IPAM.

Source code in gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@step("Register IPv4 devices in IPAM")
def register_dns_records_v4_devices(
    subscription: LanSwitchInterconnectInactive, subscription_id: UUIDstr, ipam_registrations: dict[str, dict[str, str]]
) -> State:
    """Register DNS records for both switch and router side in IPAM."""
    switch_hostname = subscription.lan_switch_interconnect.switch_side.switch.fqdn
    router_hostname = (
        f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
        f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
    )
    if not (switch_hostname and router_hostname):
        msg = "Missing switch or router hostname, cannot continue."
        raise ProcessFailureError(msg, details=subscription.lan_switch_interconnect)

    ip_network = IPv4Network(ipam_registrations["v4"]["network"])

    create_host_by_ip(switch_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv4_address=ip_network[10])
    create_host_by_ip(router_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv4_address=ip_network[1])

    return {"ipam_registrations": {"v4": {switch_hostname: ip_network[10], router_hostname: ip_network[1]}}}

register_dns_records_v6_network(subscription)

Add DNS records in IPAM.

Source code in gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@step("Register IPv6 network in IPAM")
def register_dns_records_v6_network(subscription: LanSwitchInterconnectInactive) -> State:
    """Add DNS records in IPAM."""
    router_site = subscription.lan_switch_interconnect.router_side.node.router_site
    if not router_site or not router_site.site_internal_id:
        msg = "Site internal ID not set. Cannot continue."
        raise ProcessFailureError(msg, details=router_site)

    new_network = generate_lan_switch_interconnect_subnet_v6(router_site.site_internal_id)
    ipam_oss_params = load_oss_params().IPAM.LAN_SWITCH_INTERCONNECT
    create_v6_network_by_ip(
        ipam_oss_params.dns_view, ipam_oss_params.network_view, new_network, str(subscription.subscription_id)
    )

    return {"ipam_registrations": {"v6": {"network": new_network}}}

register_dns_records_v6_devices(subscription, subscription_id, ipam_registrations)

Register DNS records for both switch and router side in IPAM.

Source code in gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@step("Register IPv6 devices in IPAM")
def register_dns_records_v6_devices(
    subscription: LanSwitchInterconnectInactive, subscription_id: UUIDstr, ipam_registrations: dict[str, dict[str, str]]
) -> State:
    """Register DNS records for both switch and router side in IPAM."""
    switch_hostname = subscription.lan_switch_interconnect.switch_side.switch.fqdn
    router_hostname = (
        f"{subscription.lan_switch_interconnect.router_side.ae_iface}."
        f"{subscription.lan_switch_interconnect.router_side.node.router_fqdn}"
    )
    if not (switch_hostname and router_hostname):
        msg = "Missing switch or router hostname, cannot continue."
        raise ProcessFailureError(msg, details=subscription.lan_switch_interconnect)

    ip_network = IPv6Network(ipam_registrations["v6"]["network"])

    create_host_by_ip(switch_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv6_address=ip_network[10])
    create_host_by_ip(router_hostname, "LAN_SWITCH_INTERCONNECT", subscription_id, ipv6_address=ip_network[1])

    return {"ipam_registrations": {"v6": {switch_hostname: ip_network[10], router_hostname: ip_network[1]}}}

create_lan_switch_interconnect()

Create a new LAN interconnect between a Switch and a Router.

Source code in gso/workflows/lan_switch_interconnect/create_lan_switch_interconnect.py

@workflow(
    "Create LAN Switch Interconnect",
    initial_input_form=wrap_create_initial_input_form(_initial_input_form),
    target=Target.CREATE,
)
def create_lan_switch_interconnect() -> StepList:
    """Create a new LAN interconnect between a Switch and a Router."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription(Target.CREATE)
        >> initialize_subscription
        >> register_dns_records_v4_network
        >> register_dns_records_v4_devices
        >> register_dns_records_v6_network
        >> register_dns_records_v6_devices
        >> set_status(SubscriptionLifecycle.ACTIVE)
        >> resync
        >> done
    )