import ipaddress from orchestrator.forms import FormPage from orchestrator.forms.validators import Label from orchestrator.targets import Target from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr from orchestrator.workflow import StepList, conditional, done, init, step, workflow from orchestrator.workflows.steps import resync, set_status, store_process_subscription, unsync from orchestrator.workflows.utils import wrap_modify_initial_input_form from gso.products.product_types.iptrunk import Iptrunk from gso.services import infoblox, provisioning_proxy from gso.services.provisioning_proxy import pp_interaction from gso.workflows.iptrunk.utils import set_isis_to_90000 def initial_input_form_generator() -> FormGenerator: class TerminateForm(FormPage): termination_label: Label = ( "Please confirm whether configuration should get removed from the A and B sides of the trunk, and whether " "IPAM resources should be released." # type: ignore ) tt_number: str remove_configuration: bool = True clean_up_ipam: bool = True user_input = yield TerminateForm return user_input.dict() @step("Drain traffic from trunk") def drain_traffic_from_ip_trunk(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> State: provisioning_proxy.provision_ip_trunk(subscription, process_id, tt_number, "isis_interface", False) return { "subscription": subscription, "label_text": "This is setting the ISIS metric of the trunk to 9000. Press refresh to get the results." "When traffic is drained, confirm to continue.", } @step("Deprovision IP trunk [DRY RUN]") def deprovision_ip_trunk_dry(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> State: provisioning_proxy.deprovision_ip_trunk(subscription, process_id, tt_number, True) return { "subscription": subscription, "label_text": "[DRY RUN] Terminating IP trunk, please refresh to get the results of the playbook.", } @step("Deprovision IP trunk [FOR REAL]") def deprovision_ip_trunk_real(subscription: Iptrunk, process_id: UUIDstr, tt_number: str) -> State: provisioning_proxy.deprovision_ip_trunk(subscription, process_id, tt_number, False) return { "subscription": subscription, "label_text": "[COMMIT] Terminating IP trunk, please refresh to get the results of the playbook.", } @step("Deprovision IPv4 networks") def deprovision_ip_trunk_ipv4(subscription: Iptrunk) -> dict: infoblox.delete_network(ipaddress.IPv4Network(subscription.iptrunk.iptrunk_ipv4_network)) return {"subscription": subscription} @step("Deprovision IPv6 networks") def deprovision_ip_trunk_ipv6(subscription: Iptrunk) -> dict: infoblox.delete_network(ipaddress.IPv6Network(subscription.iptrunk.iptrunk_ipv6_network)) return {"subscription": subscription} @workflow( "Terminate IPtrunk", initial_input_form=wrap_modify_initial_input_form(initial_input_form_generator), target=Target.TERMINATE, ) def terminate_iptrunk() -> StepList: run_config_steps = conditional(lambda state: state["remove_configuration"]) run_ipam_steps = conditional(lambda state: state["clean_up_ipam"]) config_steps = ( init >> pp_interaction(set_isis_to_90000, 3) >> pp_interaction(deprovision_ip_trunk_dry, 3) >> pp_interaction(deprovision_ip_trunk_real, 3) ) ipam_steps = init >> deprovision_ip_trunk_ipv4 >> deprovision_ip_trunk_ipv6 return ( init >> store_process_subscription(Target.TERMINATE) >> unsync >> run_config_steps(config_steps) >> run_ipam_steps(ipam_steps) >> set_status(SubscriptionLifecycle.TERMINATED) >> resync >> done )