Skip to content

Create switch management

A creation workflow for adding a new Switch Management to the subscription database.

create_subscription(product, partner_id)

Create a new subscription object.

Source code in gso/workflows/switch_management/create_switch_management.py 
55
56
57
58
59
60
@step("Create subscription")
def create_subscription(product: UUID, partner_id: UUIDstr) -> State:
    """Create a new subscription object."""
    subscription = SwitchManagementInactive.from_product_id(product, partner_id)

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

initialize_subscription(subscription, edge_port, dcn_vlan_id, mgmt_vlan_id)

Initialize the subscription with user input.

This includes calculating the IP spaces that the dependant Layer 3 interfaces will sit in.

Source code in gso/workflows/switch_management/create_switch_management.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@step("Initialize subscription")
def initialize_subscription(
    subscription: SwitchManagementInactive,
    edge_port: UUID,
    dcn_vlan_id: VLAN_ID,
    mgmt_vlan_id: VLAN_ID,
) -> State:
    """Initialize the subscription with user input.

    This includes calculating the IP spaces that the dependant Layer 3 interfaces will sit in.
    """
    ipam_params = load_oss_params().IPAM
    edge_port_pb = EdgePort.from_subscription(subscription_id=edge_port).edge_port
    site_internal_id = edge_port_pb.node.router_site.site_internal_id

    switch_management_networks = {
        "v4": generate_management_subnet_v4(ipam_params.SWITCH_MANAGEMENT, site_internal_id),
        "v6": generate_management_subnet_v6(ipam_params.SWITCH_MANAGEMENT, site_internal_id),
    }
    subscription.switch_management.management_interface = L3InterfacePortBlockInactive.new(
        subscription_id=uuid4(),
        is_tagged=True,
        vlan_id=mgmt_vlan_id,
        ipv4_address=switch_management_networks["v4"][1],  # type: ignore[index]
        ipv4_mask=ipam_params.SWITCH_MANAGEMENT.V4.mask,
        ipv6_address=switch_management_networks["v6"][1],  # type: ignore[index]
        ipv6_mask=ipam_params.SWITCH_MANAGEMENT.V6.mask,
        custom_firewall_filters=False,
        gs_id=generate_unique_id(prefix="GS"),
        edge_port=edge_port_pb,
    )

    if edge_port_pb.node.router_site.site_contains_optical_equipment:
        dcn_management_networks = {
            "v4": generate_management_subnet_v4(ipam_params.OPTICAL_MANAGEMENT, site_internal_id),
            "v6": generate_management_subnet_v6(ipam_params.OPTICAL_MANAGEMENT, site_internal_id),
        }
        subscription.switch_management.dcn_interface = L3InterfacePortBlockInactive.new(
            subscription_id=uuid4(),
            is_tagged=True,
            vlan_id=dcn_vlan_id,
            ipv4_address=dcn_management_networks["v4"][1],  # type: ignore[index]
            ipv4_mask=ipam_params.OPTICAL_MANAGEMENT.V4.mask,
            ipv6_address=dcn_management_networks["v6"][1],  # type: ignore[index]
            ipv6_mask=ipam_params.OPTICAL_MANAGEMENT.V6.mask,
            custom_firewall_filters=False,
            gs_id=generate_unique_id(prefix="GS"),
            edge_port=edge_port_pb,
        )
    else:
        dcn_management_networks = {}

    return {
        "subscription": subscription,
        "ipam_registrations": {
            "switch_management": switch_management_networks,
            "optical_management": dcn_management_networks,
        },
    }

register_dns_records_v4_network(subscription, ipam_registrations)

Register an IPv4 network for Switch and Optical Management in IPAM.

Source code in gso/workflows/switch_management/create_switch_management.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
@step("Register IPv4 network in IPAM")
def register_dns_records_v4_network(subscription: SwitchManagementInactive, ipam_registrations: dict) -> None:
    """Register an IPv4 network for Switch and Optical Management in IPAM."""
    ipam_oss_params = load_oss_params().IPAM
    create_v4_network_by_ip(
        ipam_oss_params.SWITCH_MANAGEMENT.dns_view,
        ipam_oss_params.SWITCH_MANAGEMENT.network_view,
        ipam_registrations["switch_management"]["v4"],
        str(subscription.subscription_id),
    )

    if ipam_registrations["optical_management"]:
        create_v4_network_by_ip(
            ipam_oss_params.OPTICAL_MANAGEMENT.dns_view,
            ipam_oss_params.OPTICAL_MANAGEMENT.network_view,
            ipam_registrations["optical_management"]["v4"],
            str(subscription.subscription_id),
        )

register_dns_records_v6_network(subscription, ipam_registrations)

Register an IPv6 network for Switch and Optical Management in IPAM.

Source code in gso/workflows/switch_management/create_switch_management.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
@step("Register IPv6 network in IPAM")
def register_dns_records_v6_network(subscription: SwitchManagementInactive, ipam_registrations: dict) -> None:
    """Register an IPv6 network for Switch and Optical Management in IPAM."""
    ipam_oss_params = load_oss_params().IPAM
    create_v6_network_by_ip(
        ipam_oss_params.SWITCH_MANAGEMENT.dns_view,
        ipam_oss_params.SWITCH_MANAGEMENT.network_view,
        ipam_registrations["switch_management"]["v6"],
        str(subscription.subscription_id),
    )

    if ipam_registrations["optical_management"]:
        create_v6_network_by_ip(
            ipam_oss_params.OPTICAL_MANAGEMENT.dns_view,
            ipam_oss_params.OPTICAL_MANAGEMENT.network_view,
            ipam_registrations["optical_management"]["v6"],
            str(subscription.subscription_id),
        )

register_dns_records_devices(subscription)

Register DNS records for both switch and optical management interfaces in IPAM.

Source code in gso/workflows/switch_management/create_switch_management.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
@step("Register devices in IPAM")
def register_dns_records_devices(subscription: SwitchManagementInactive) -> None:
    """Register DNS records for both switch and optical management interfaces in IPAM."""
    switch_management_fqdn = calculate_switch_management_fqdn(subscription)
    create_host_by_ip(
        switch_management_fqdn,
        "SWITCH_MANAGEMENT",
        str(subscription.subscription_id),
        ipv4_address=subscription.switch_management.management_interface.ipv4_address,
        ipv6_address=subscription.switch_management.management_interface.ipv6_address,
    )

    if (dcn_interface := subscription.switch_management.dcn_interface) and (
        dcn_management_fqdn := calculate_dcn_management_fqdn(subscription)
    ):
        create_host_by_ip(
            dcn_management_fqdn,
            "OPTICAL_MANAGEMENT",
            str(subscription.subscription_id),
            ipv4_address=dcn_interface.ipv4_address,
            ipv6_address=dcn_interface.ipv6_address,
        )

create_switch_management()

Create a new Switch Management.

Source code in gso/workflows/switch_management/create_switch_management.py 
188
189
190
191
192
193
194
195
196
197
198
199
@create_workflow("Create Switch Management", _initial_input_form_generator)
def create_switch_management() -> StepList:
    """Create a new Switch Management."""
    return (
        begin
        >> create_subscription
        >> store_process_subscription()
        >> initialize_subscription
        >> register_dns_records_v4_network
        >> register_dns_records_v6_network
        >> register_dns_records_devices
    )