Skip to content

Base migrate placement port

Base functionality for migrating Placement Port services.

initial_input_form_generator(subscription_id)

Generate the initial input form for migrating a Placement Port service.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
    """Generate the initial input form for migrating a Placement Port service."""
    subscription = SubscriptionModel.from_subscription(subscription_id)
    product_name = subscription.product.name
    geant_partner_id = get_partner_by_name("GEANT").partner_id
    l3_interface = subscription.l3_interface  # type: ignore[attr-defined]

    def new_edge_port_selector() -> TypeAlias:
        """Generate a dropdown selector for choosing an active Edge Port in an input form except for the current one."""
        edge_ports = get_active_edge_port_subscriptions(
            partner_id=subscription.customer_id if subscription.customer_id != geant_partner_id else None,
            exclude_subscription_ids=[str(subscription.l3_interface.edge_port.owner_subscription_id)],  # type: ignore[attr-defined]
        )

        options = {str(edge_port.subscription_id): edge_port.description for edge_port in edge_ports}

        return cast(
            type[Choice],
            Choice.__call__(
                "Select an Edge Port",
                zip(options.keys(), options.items(), strict=True),
            ),
        )

    class ModifyPlacementPortForm(SubmitFormPage):
        model_config = ConfigDict(title=f"Migrate {product_name}")

        tt_number: TTNumber
        current_edge_port: read_only_field(  # type: ignore[valid-type]
            EdgePort.from_subscription(l3_interface.edge_port.owner_subscription_id).description
        )
        new_edge_port: new_edge_port_selector() = l3_interface.edge_port  # type: ignore[valid-type]

    modified_input = yield ModifyPlacementPortForm

    return modified_input.model_dump() | {"partner_name": get_partner_by_id(subscription.customer_id).name}

deactivate_old_l3_port_dry(subscription, process_id, tt_number, partner_name)

Perform a dry run of deactivating the old L3 interface.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@step("[DRY RUN] Deactivate L3 port on the old router")
def deactivate_old_l3_port_dry(
    subscription: SubscriptionModel, process_id: UUIDstr, tt_number: str, partner_name: str
) -> LSOState:
    """Perform a dry run of deactivating the old L3 interface."""
    extra_vars = {
        "subscription": json.loads(json_dumps(subscription)),
        "partner_name": partner_name,
        "dry_run": True,
        "verb": "deactivate",
        "object": "l3_interface",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
        f"Deactivate config for {subscription.description}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/manage_commercial_l3_interface.yaml",
        "inventory": {"all": {"hosts": {subscription.l3_interface.edge_port.node.router_fqdn: None}}},  # type: ignore[attr-defined]
        "extra_vars": extra_vars,
    }

deactivate_old_l3_port_real(subscription, process_id, tt_number, partner_name)

Perform a real run of deactivating the old L3 interface.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@step("[REAL RUN] Deactivate L3 port on the old router")
def deactivate_old_l3_port_real(
    subscription: SubscriptionModel, process_id: UUIDstr, tt_number: str, partner_name: str
) -> LSOState:
    """Perform a real run of deactivating the old L3 interface."""
    extra_vars = {
        "subscription": json.loads(json_dumps(subscription)),
        "partner_name": partner_name,
        "dry_run": False,
        "verb": "deactivate",
        "object": "l3_interface",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
        f"Deactivate config for {subscription.description}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/manage_commercial_l3_interface.yaml",
        "inventory": {"all": {"hosts": {subscription.l3_interface.edge_port.node.router_fqdn: None}}},  # type: ignore[attr-defined]
        "extra_vars": extra_vars,
    }

confirm_remove_old_bgp_peers()

Wait for confirmation from an operator that the old BGP peers are removed manually.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
113
114
115
116
117
118
119
120
121
122
123
124
@inputstep("Wait for confirmation", assignee=Assignee.SYSTEM)
def confirm_remove_old_bgp_peers() -> FormGenerator:
    """Wait for confirmation from an operator that the old BGP peers are removed manually."""

    class ProvisioningResultPage(SubmitFormPage):
        model_config = ConfigDict(title="Please confirm before continuing")

        info_label: Label = "Please remove/deactivate the old BGP peers manually from the old router before continuing."

    yield ProvisioningResultPage

    return {}

update_subscription_model(subscription, new_edge_port)

Update the subscription model with new edge port information.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
127
128
129
130
131
132
133
134
@step("Update subscription with new edge port")
def update_subscription_model(
    subscription: SubscriptionModel,
    new_edge_port: UUIDstr,
) -> State:
    """Update the subscription model with new edge port information."""
    subscription.l3_interface.edge_port = EdgePort.from_subscription(new_edge_port).edge_port  # type: ignore[attr-defined]
    return {"subscription": subscription}

generate_commercial_peers_linked_to_placement_port(subscription)

Prepare a list of commercial peers linked to the current Placement Port.

Filters each commercial peers' peering connection list to only include those with the current placement port subscription ID.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
@step("Generate list of commercial peers linked to the current Placement Port")
def generate_commercial_peers_linked_to_placement_port(
    subscription: SubscriptionModel,
) -> State:
    """Prepare a list of commercial peers linked to the current Placement Port.

    Filters each commercial peers' peering connection list to only include those
    with the current placement port subscription ID.
    """
    current_sub_id = str(subscription.subscription_id)
    raw_commercial_peers = get_active_commercial_peers_linked_to_placement_port(current_sub_id)

    scoped_peers = []

    for peer in raw_commercial_peers:
        peer_dict = peer.model_dump()
        commercial_peer_data = peer_dict.get("commercial_peer", {})
        peer_dict["partner"] = get_partner_by_id(peer_dict["customer_id"]).name
        original_peering_connections = commercial_peer_data.get("peering_connection_list", [])
        filtered_connections = [
            pc
            for pc in original_peering_connections
            if str(pc["placement_port"]["owner_subscription_id"]) == current_sub_id
        ]

        commercial_peer_data["peering_connection_list"] = filtered_connections
        scoped_peers.append(peer_dict)

    return {_MONITORED_OBJECTS_KEY: scoped_peers}

create_bgp_peers_on_new_edge_port_dry(subscription, process_id, tt_number, partner_name, scoped_commercial_peers)

Perform a dry run of creating a list of existing BGP peers on the new placement port.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
@step("[DRY RUN] Create/Place BGP peers on the new placement port")
def create_bgp_peers_on_new_edge_port_dry(
    subscription: SubscriptionModel,
    process_id: UUIDstr,
    tt_number: str,
    partner_name: str,
    scoped_commercial_peers: list[dict],
) -> LSOState:
    """Perform a dry run of creating a list of existing BGP peers on the new placement port."""
    extra_vars = {
        "subscription": json.loads(json_dumps(subscription)),
        "scoped_commercial_peers": scoped_commercial_peers,
        "partner_name": partner_name,
        "dry_run": True,
        "verb": "deploy",
        "object": "bgp_peer",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
        f"Deploy config for {subscription.description}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/manage_commercial_bgp_peers_list.yaml",
        "inventory": {"all": {"hosts": {subscription.l3_interface.edge_port.node.router_fqdn: None}}},  # type: ignore[attr-defined]
        "extra_vars": extra_vars,
    }

create_bgp_peers_on_new_edge_port_real(subscription, process_id, tt_number, partner_name, scoped_commercial_peers)

Perform a real run of creating a list of existing BGP peers on the new placement port.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
@step("[REAL RUN] Create/Place BGP peers on the new placement port")
def create_bgp_peers_on_new_edge_port_real(
    subscription: SubscriptionModel,
    process_id: UUIDstr,
    tt_number: str,
    partner_name: str,
    scoped_commercial_peers: list[dict],
) -> LSOState:
    """Perform a real run of creating a list of existing BGP peers on the new placement port."""
    extra_vars = {
        "subscription": json.loads(json_dumps(subscription)),
        "scoped_commercial_peers": scoped_commercial_peers,
        "partner_name": partner_name,
        "dry_run": False,
        "verb": "deploy",
        "object": "bgp_peer",
        "commit_comment": f"GSO_PROCESS_ID: {process_id} - TT_NUMBER: {tt_number} - "
        f"Deploy config for {subscription.description}",
    }

    return {
        "playbook_name": "gap_ansible/playbooks/manage_commercial_bgp_peers_list.yaml",
        "inventory": {"all": {"hosts": {subscription.l3_interface.edge_port.node.router_fqdn: None}}},  # type: ignore[attr-defined]
        "extra_vars": extra_vars,
    }

migrate_placement_port_steps()

Steps for migrating a Placement Port service.

Source code in gso/workflows/placement_port/base_migrate_placement_port.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def migrate_placement_port_steps() -> StepList:
    """Steps for migrating a Placement Port service."""
    return (
        begin
        >> lso_interaction(deactivate_old_l3_port_dry)
        >> lso_interaction(deactivate_old_l3_port_real)
        >> confirm_remove_old_bgp_peers
        >> update_subscription_model
        >> generate_commercial_peers_linked_to_placement_port
        >> start_moodi(monitored_objects_key=_MONITORED_OBJECTS_KEY)
        >> lso_interaction(deploy_l3_port_dry)
        >> lso_interaction(deploy_l3_port_real)
        >> lso_interaction(create_bgp_peers_on_new_edge_port_dry)
        >> lso_interaction(create_bgp_peers_on_new_edge_port_real)
        >> stop_moodi()
    )