Skip to content

Redeploy base config

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

start_redeploy_workflows(tt_number, selected_routers, callback_route)

Start the massive redeploy base config task with the selected routers.

Source code in gso/workflows/tasks/redeploy_base_config.py 
50
51
52
53
54
55
56
@step("Start worker to redeploy base config on selected routers")
def start_redeploy_workflows(tt_number: TTNumber, selected_routers: list[UUIDstr], callback_route: str) -> State:
    """Start the massive redeploy base config task with the selected routers."""
    # TODO if in the future you changed UUIDstr to UUID, you need to convert them to string when passing to the task
    massive_redeploy_base_config_task.apply_async(args=[selected_routers, tt_number, callback_route], countdown=5)  # type: ignore[attr-defined]

    return {"failed_wfs": {}, "successful_wfs": {}}

evaluate_results(callback_result)

Evaluate the result of the provisioning proxy callback.

Source code in gso/workflows/tasks/redeploy_base_config.py 
59
60
61
62
63
64
@step("Evaluate provisioning proxy result")
def evaluate_results(callback_result: dict) -> State:
    """Evaluate the result of the provisioning proxy callback."""
    failed_wfs = callback_result.pop("failed_wfs", {})
    successful_wfs = callback_result.pop("successful_wfs", {})
    return {"callback_result": callback_result, "failed_wfs": failed_wfs, "successful_wfs": successful_wfs}

workflows_failed_to_prompt(failed_wfs, successful_wfs)

Prompt the operator that some workflows have failed to start.

Source code in gso/workflows/tasks/redeploy_base_config.py 
67
68
69
70
71
72
73
74
75
76
77
@inputstep("Some workflows have failed", assignee=Assignee.SYSTEM)
def workflows_failed_to_prompt(failed_wfs: dict, successful_wfs: dict) -> FormGenerator:
    """Prompt the operator that some workflows have failed to start."""

    class WFFailurePrompt(SubmitFormPage):
        model_config = ConfigDict(title="Some redeploy workflows have failed, please inspect the list below")
        failed_workflows: LongText = json.dumps(failed_wfs, indent=4)
        successful_workflows: LongText = json.dumps(successful_wfs, indent=4)

    yield WFFailurePrompt
    return {}

task_redeploy_base_config()

Gather a list of routers from the operator to redeploy base config onto.

Source code in gso/workflows/tasks/redeploy_base_config.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@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: len(state.get("failed_wfs", {})) > 0)

    return (
        init
        >> callback_step(
            name="Start running redeploy workflows on selected routers",
            action_step=start_redeploy_workflows,
            validate_step=evaluate_results,
        )
        >> some_failed_to_start(workflows_failed_to_prompt)
        >> done
    )