0bfa018ec1
This commit enables dcmanager orchestrator's worker service for System
Controller and implements its communication to the orchestrator manager
process.
Test plan:
1. PASS: Build an ISO containing the changes for the implementation of
the orchestrator worker and successfully deploy a new DC system.
2. PASS: Verify that the deployed system has the new service running as
expected and logging to the orchestrator.log file.
3. PASS: Verify that the orchestrator-worker service can be managed by
SM and that restarting dcmanager-manager also restarts the
orchestrator worker.
4. PASS: Create a strategy with a max parallel subclouds smaller than
the total number of subclouds being orchestrated and verify that the
subclouds deletion is properly handled by multiple workers,
respecting the parameter. Once the steps are removed, verify that the
strategy is deleted by the manager process.
5. PASS: Create a strategy with a max parallel subclouds smaller than
the total number of subclouds being orchestrated, apply it and
verify that only the amount of subclouds specified by the parameter
are processed at a time.
6. PASS: Start applying a strategy, restart dcmanager-manager service,
verify that the orchestrator's manager identify that there was a
strategy being applied and, once the orchestration interval has
passed, it retrieves the steps that were being processed and send
them to the workers.
7. PASS: Start applying a strategy, delete one of the worker processes
that were handling steps, wait for the orchestration interval and
verify that the manager sends the steps that were pending to a new
worker.
8. PASS: When applying a strategy, verify that a step that is in the
applying state for more than a minute has its updated_at field
updated to the current time to avoid it being detected by the manager
as a step that is not in process.
9. PASS: Create a strategy and verify that, when a step fails, the
strategy's state is also set to failed.
10. PASS: Create a strategy with stop-on-failure set and verify that once
a step reaches a failed state, all workers stop processing new steps.
11. PASS: Abort an applying strategy, verify that all steps in initial
state are updated to aborted, the ones that were processing
continue until they reach a finished state.
12. PASS: Verify that the manager's strategy monitoring thread only
executes when a strategy is being applied, until it reaches a
complete state, e.g. complete, failed or aborted, or when the
strategy's deletion is requested.
Story: 2011311
Task: 51695
Change-Id: I3ad086fa3222eda6edd1d3a314b13cb0bde161ef
Signed-off-by: Raphael Lima <Raphael.Lima@windriver.com>
123 lines
3.8 KiB
Python
123 lines
3.8 KiB
Python
# Copyright (c) 2020-2021, 2024-2025 Wind River Systems, Inc.
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
|
|
"""
|
|
Client side of the DC Manager Orchestrator RPC API.
|
|
"""
|
|
|
|
from dcmanager.common import consts
|
|
from dcmanager.common import messaging
|
|
|
|
|
|
class ManagerOrchestratorClient(object):
|
|
"""Client side of the DC Manager Orchestrator RPC API.
|
|
|
|
Version History:
|
|
1.0 - Initial version
|
|
"""
|
|
|
|
BASE_RPC_API_VERSION = "1.0"
|
|
|
|
def __init__(self, timeout=None):
|
|
self._client = messaging.get_rpc_client(
|
|
timeout=timeout,
|
|
topic=consts.TOPIC_DC_MANAGER_ORCHESTRATOR,
|
|
version=self.BASE_RPC_API_VERSION,
|
|
)
|
|
|
|
@staticmethod
|
|
def make_msg(method, **kwargs):
|
|
return method, kwargs
|
|
|
|
def call(self, ctxt, msg, version=None):
|
|
method, kwargs = msg
|
|
if version is not None:
|
|
client = self._client.prepare(version=version)
|
|
else:
|
|
client = self._client
|
|
return client.call(ctxt, method, **kwargs)
|
|
|
|
def cast(self, ctxt, msg, version=None):
|
|
method, kwargs = msg
|
|
if version is not None:
|
|
client = self._client.prepare(version=version)
|
|
else:
|
|
client = self._client
|
|
return client.cast(ctxt, method, **kwargs)
|
|
|
|
def create_sw_update_strategy(self, ctxt, payload):
|
|
return self.call(
|
|
ctxt, self.make_msg("create_sw_update_strategy", payload=payload)
|
|
)
|
|
|
|
def delete_sw_update_strategy(self, ctxt, update_type=None):
|
|
return self.call(
|
|
ctxt, self.make_msg("delete_sw_update_strategy", update_type=update_type)
|
|
)
|
|
|
|
def apply_sw_update_strategy(self, ctxt, update_type=None):
|
|
return self.call(
|
|
ctxt, self.make_msg("apply_sw_update_strategy", update_type=update_type)
|
|
)
|
|
|
|
def abort_sw_update_strategy(self, ctxt, update_type=None):
|
|
return self.call(
|
|
ctxt, self.make_msg("abort_sw_update_strategy", update_type=update_type)
|
|
)
|
|
|
|
def stop_strategy(self, ctxt, strategy_type):
|
|
return self.call(
|
|
ctxt, self.make_msg("stop_strategy", strategy_type=strategy_type)
|
|
)
|
|
|
|
|
|
class ManagerOrchestratorWorkerClient(object):
|
|
"""Client side of the DC Manager Orchestrator Worker RPC API.
|
|
|
|
Version History:
|
|
1.0 - Initial version
|
|
"""
|
|
|
|
BASE_RPC_API_VERSION = "1.0"
|
|
|
|
def __init__(self, timeout=None):
|
|
self._client = messaging.get_rpc_client(
|
|
timeout=timeout,
|
|
topic=consts.TOPIC_DC_MANAGER_ORCHESTRATOR_WORKER,
|
|
version=self.BASE_RPC_API_VERSION,
|
|
)
|
|
|
|
@staticmethod
|
|
def make_msg(method, **kwargs):
|
|
return method, kwargs
|
|
|
|
def call(self, ctxt, msg, fanout=None, version=None):
|
|
method, kwargs = msg
|
|
if fanout or version:
|
|
client = self._client.prepare(fanout=fanout, version=version)
|
|
else:
|
|
client = self._client
|
|
return client.call(ctxt, method, **kwargs)
|
|
|
|
def orchestrate(self, ctxt, steps_id, strategy_type):
|
|
return self.call(
|
|
ctxt,
|
|
self.make_msg(
|
|
"orchestrate", steps_id=steps_id, strategy_type=strategy_type
|
|
),
|
|
)
|
|
|
|
def stop_processing(self, ctxt):
|
|
return self.call(ctxt, self.make_msg("stop_processing"), fanout=True)
|