Skip to content
Snippets Groups Projects
Verified Commit ef07e6b1 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

Add flag to router termination workflow that allows for skipping IPAM cleanup

parent 8fe1d93a
Branches
Tags
1 merge request!443Add flag to router termination workflow that allows for skipping IPAM cleanup
Pipeline #95144 passed
...@@ -15,6 +15,7 @@ The workflow consists of the following steps: ...@@ -15,6 +15,7 @@ The workflow consists of the following steps:
- Set the subscription status to `TERMINATED`. - Set the subscription status to `TERMINATED`.
""" """
import datetime
import ipaddress import ipaddress
import json import json
import logging import logging
...@@ -34,6 +35,7 @@ from orchestrator.workflows.steps import ( ...@@ -34,6 +35,7 @@ from orchestrator.workflows.steps import (
unsync, unsync,
) )
from orchestrator.workflows.utils import wrap_modify_initial_input_form from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic import Field
from pydantic_forms.types import FormGenerator, State, UUIDstr from pydantic_forms.types import FormGenerator, State, UUIDstr
from requests import HTTPError from requests import HTTPError
...@@ -63,19 +65,28 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: ...@@ -63,19 +65,28 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
class TerminateForm(SubmitFormPage): class TerminateForm(SubmitFormPage):
if router.status == SubscriptionLifecycle.INITIAL: if router.status == SubscriptionLifecycle.INITIAL:
info_label_2: Label = ( info_label_2: Label = Field(
"This will immediately mark the subscription as terminated, preventing any other workflows from " "This will immediately mark the subscription as terminated, preventing any other workflows from "
"interacting with this product subscription." "interacting with this product subscription.",
exclude=True,
)
info_label_3: Label = Field(
"ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING.", exclude=True
) )
info_label_3: Label = "ONLY EXECUTE THIS WORKFLOW WHEN YOU ARE ABSOLUTELY SURE WHAT YOU ARE DOING."
tt_number: TTNumber tt_number: TTNumber
termination_label: Label = "Please confirm whether configuration should get removed from the router." termination_label: Label = Field(
"Please confirm whether configuration should get removed from the router.", exclude=True
)
remove_configuration: bool = False remove_configuration: bool = False
update_ibgp_mesh_label: Label = "Please confirm whether the iBGP mesh should get updated." update_ibgp_mesh_label: Label = Field("Please confirm whether the iBGP mesh should get updated.", exclude=True)
update_ibgp_mesh: bool = True update_ibgp_mesh: bool = True
update_sdp_mesh_label: Label = "Please confirm whether the SDP mesh should get updated." update_sdp_mesh_label: Label = Field("Please confirm whether the SDP mesh should get updated.", exclude=True)
update_sdp_mesh: bool = True update_sdp_mesh: bool = True
remove_loopback_from_ipam_label: Label = Field(
"Please confirm whether the loopback address should be released in IPAM.", exclude=True
)
remove_loopback_from_ipam: bool = False
user_input = yield TerminateForm user_input = yield TerminateForm
return user_input.model_dump() | { return user_input.model_dump() | {
...@@ -85,11 +96,20 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator: ...@@ -85,11 +96,20 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@step("Deprovision loopback IPs from IPAM") @step("Deprovision loopback IPs from IPAM")
def deprovision_loopback_ips(subscription: Router) -> dict: def deprovision_loopback_ips(subscription: Router, remove_loopback_from_ipam: bool, process_id: UUIDstr) -> None: # noqa: FBT001
"""Clear up the loopback addresses from IPAM.""" """Clear up the loopback addresses from IPAM."""
infoblox.delete_host_by_ip(ipaddress.IPv4Address(subscription.router.router_lo_ipv4_address)) if remove_loopback_from_ipam:
infoblox.delete_host_by_ip(ipaddress.IPv4Address(subscription.router.router_lo_ipv4_address))
return {"subscription": subscription} else:
record = infoblox.find_host_by_fqdn(subscription.router.router_fqdn)
if record:
# We keep the record in IPAM but add a comment stating that this router is terminated.
# This is done to prevent an address from being re-used.
record.comment = (
f"This router was terminated by GAP process {process_id} on "
f"{datetime.datetime.now(tz=datetime.UTC).strftime("%d/%m/%Y")}."
)
record.update()
@step("[DRY RUN] Remove configuration from router") @step("[DRY RUN] Remove configuration from router")
......
...@@ -37,6 +37,7 @@ def test_terminate_pe_router_full_success( ...@@ -37,6 +37,7 @@ def test_terminate_pe_router_full_success(
"remove_configuration": remove_configuration, "remove_configuration": remove_configuration,
"update_ibgp_mesh": update_ibgp_mesh, "update_ibgp_mesh": update_ibgp_mesh,
"update_sdp_mesh": update_sdp_mesh, "update_sdp_mesh": update_sdp_mesh,
"remove_loopback_from_ipam": True,
} }
lso_interaction_count = 0 lso_interaction_count = 0
if remove_configuration: if remove_configuration:
...@@ -89,6 +90,7 @@ def test_terminate_p_router_full_success( ...@@ -89,6 +90,7 @@ def test_terminate_p_router_full_success(
"tt_number": faker.tt_number(), "tt_number": faker.tt_number(),
"remove_configuration": remove_configuration, "remove_configuration": remove_configuration,
"update_ibgp_mesh": update_ibgp_mesh, "update_ibgp_mesh": update_ibgp_mesh,
"remove_loopback_from_ipam": True,
} }
lso_interaction_count = 0 lso_interaction_count = 0
if remove_configuration: if remove_configuration:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment