Add task for redeploying base config on multiple routers

gso/migrations/versions/2025-05-15_9a7bae1f6438_add_mass_router_redeploy_task.py

+"""Add mass router redeploy task.
+
+Revision ID: 9a7bae1f6438
+Revises: 465008ed496e
+Create Date: 2025-05-15 12:01:54.469229
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = '9a7bae1f6438'
+down_revision = '465008ed496e'
+branch_labels = None
+depends_on = None
+
+
+from orchestrator.migrations.helpers import create_task, delete_workflow
+
+new_tasks = [
+    {
+        "name": "task_redeploy_base_config",
+        "description": "Redeploy base config on multiple routers"
+    }
+]
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    for task in new_tasks:
+        create_task(conn, task)
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    for task in new_tasks:
+        delete_workflow(conn, task["name"])

gso/translations/en-GB.json

@@ -150,6 +150,7 @@
         "task_create_partners": "Create partner task",
         "task_delete_partners": "Delete partner task",
         "task_modify_partners": "Modify partner task",
+        "task_redeploy_base_config": "Redeploy base config on multiple routers",
         "task_send_email_notifications": "Send email notifications for failed tasks",
         "task_validate_geant_products": "Validation task for GEANT products",
         "terminate_edge_port": "Terminate Edge Port",

gso/workflows/__init__.py

@@ -110,6 +110,7 @@ LazyWorkflowInstance("gso.workflows.tasks.modify_partners", "task_modify_partner
 LazyWorkflowInstance("gso.workflows.tasks.delete_partners", "task_delete_partners")
 LazyWorkflowInstance("gso.workflows.tasks.clean_old_tasks", "task_clean_old_tasks")
 LazyWorkflowInstance("gso.workflows.tasks.check_site_connectivity", "task_check_site_connectivity")
+LazyWorkflowInstance("gso.workflows.tasks.redeploy_base_config", "task_redeploy_base_config")
 
 #  Edge port workflows
 LazyWorkflowInstance("gso.workflows.edge_port.create_edge_port", "create_edge_port")

gso/workflows/tasks/redeploy_base_config.py

+"""Task for redeploying base config on multiple routers at one.
+
+This task spawns multiple instances of the ``redeploy_base_config`` workflow, based on a list of Nokia routers given as
+input by the operator. The operator can then
+"""
+
+import json
+from typing import Annotated
+
+import requests
+from annotated_types import Len
+from orchestrator.config.assignee import Assignee
+from orchestrator.forms import SubmitFormPage
+from orchestrator.targets import Target
+from orchestrator.workflow import StepList, conditional, done, init, inputstep, step, workflow
+from pydantic import AfterValidator, ConfigDict
+from pydantic_forms.types import FormGenerator, State, UUIDstr
+from pydantic_forms.validators import LongText, validate_unique_list
+from requests import HTTPError
+
+from gso.products.product_types.router import Router
+from gso.services.subscriptions import get_active_router_subscriptions
+from gso.utils.helpers import active_nokia_router_selector
+from gso.utils.shared_enums import Vendor
+from gso.utils.types.tt_number import TTNumber
+
+
+def _input_form_generator() -> FormGenerator:
+    router_selection_list = Annotated[  # type: ignore[valid-type]
+        list[active_nokia_router_selector()],  # type: ignore[misc]
+        AfterValidator(validate_unique_list),
+        Len(min_length=1),
+    ]
+
+    all_active_nokia_routers = [
+        router["subscription_id"]
+        for router in get_active_router_subscriptions()
+        if Router.from_subscription(router["subscription_id"]).router.vendor == Vendor.NOKIA
+    ]
+
+    class RedeployBaseConfigForm(SubmitFormPage):
+        model_config = ConfigDict(title="Redeploy base config on multiple routers")
+
+        tt_number: TTNumber
+        selected_routers: router_selection_list = all_active_nokia_routers  # type: ignore[valid-type]
+
+    user_input = yield RedeployBaseConfigForm
+    return user_input.model_dump()
+
+
+@step("Start running redeploy workflows on selected routers")
+def start_redeploy_workflows(tt_number: TTNumber, selected_routers: list[UUIDstr]) -> State:
+    """Loop over all selected routers, and try to start a new workflow for each of them."""
+    wf_errors = {}
+    workflow_url = "http://localhost:8080/api/processes/redeploy_base_config"
+
+    for selected_router in selected_routers:
+        try:
+            result = requests.post(
+                workflow_url, json=[{"subscription_id": selected_router}, {"tt_number": tt_number}], timeout=10
+            )
+            result.raise_for_status()
+        except HTTPError as e:
+            if e.response.json()["validation_errors"]:
+                error_message = e.response.json()["validation_errors"][0]["msg"]
+            else:
+                error_message = e.response.json()
+            wf_errors[Router.from_subscription(selected_router).router.router_fqdn] = error_message
+
+    return {"wf_errors": wf_errors}
+
+
+@inputstep("Some workflows have failed to start", assignee=Assignee.SYSTEM)
+def workflows_failed_to_start_prompt(wf_errors: list) -> FormGenerator:
+    """Prompt the operator that some workflows have failed to start."""
+
+    class WFFailurePrompt(SubmitFormPage):
+        model_config = ConfigDict(title="Some redeploy workflows have failed to start, please inspect the list below")
+        failed_to_start: LongText = json.dumps(wf_errors, indent=4)
+
+    yield WFFailurePrompt
+    return {}
+
+
+@workflow("Redeploy base config on multiple routers", initial_input_form=_input_form_generator, target=Target.SYSTEM)
+def task_redeploy_base_config() -> StepList:
+    """Gather a list of routers from the operator to redeploy base config onto."""
+    some_failed_to_start = conditional(lambda state: state["wf_errors"])
+
+    return init >> start_redeploy_workflows >> some_failed_to_start(workflows_failed_to_start_prompt) >> done