Skip to content

Commit 6af88f7

Browse files
fgogollifgogolli
authored
Merge pull request #290 from finos/fix/start-stop-machines-provider-routing
fix(machines): route provider identifier to start/stop so operations don't all-fail
2 parents 338a773 + d643358 commit 6af88f7

6 files changed

Lines changed: 501 additions & 6 deletions

File tree

‎src/orb/application/services/orchestration/start_machines.py‎

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
from orb.application.provider.commands import ExecuteProviderOperationCommand
1010
from orb.application.services.orchestration.base import OrchestratorBase
1111
from orb.application.services.orchestration.dtos import StartMachinesInput, StartMachinesOutput
12+
from orb.application.services.provider_registry_service import ProviderRegistryService
1213
from orb.domain.base.operations import (
1314
Operation as ProviderOperation,
1415
OperationType as ProviderOperationType,
@@ -20,11 +21,16 @@ class StartMachinesOrchestrator(OrchestratorBase[StartMachinesInput, StartMachin
2021
"""Orchestrator for starting machines via the provider layer."""
2122

2223
def __init__(
23-
self, command_bus: CommandBusPort, query_bus: QueryBusPort, logger: LoggingPort
24+
self,
25+
command_bus: CommandBusPort,
26+
query_bus: QueryBusPort,
27+
logger: LoggingPort,
28+
provider_registry_service: ProviderRegistryService,
2429
) -> None:
2530
self._command_bus = command_bus
2631
self._query_bus = query_bus
2732
self._logger = logger
33+
self._provider_registry_service = provider_registry_service
2834

2935
async def execute(self, input: StartMachinesInput) -> StartMachinesOutput: # type: ignore[return]
3036
self._logger.info(
@@ -57,11 +63,35 @@ async def execute(self, input: StartMachinesInput) -> StartMachinesOutput: # ty
5763
message="No machines to start",
5864
)
5965

66+
# Resolve the effective provider identifier so the command handler can
67+
# route the operation to the correct provider strategy. Preference
68+
# order: explicit name > explicit type > active provider from registry.
69+
if input.provider_name:
70+
strategy_override = input.provider_name
71+
elif input.provider_type:
72+
strategy_override = input.provider_type
73+
else:
74+
try:
75+
selection = self._provider_registry_service.select_active_provider()
76+
strategy_override = selection.provider_name
77+
except Exception as exc:
78+
self._logger.error(
79+
"StartMachinesOrchestrator: cannot resolve active provider: %s", exc
80+
)
81+
return StartMachinesOutput(
82+
started_machines=[],
83+
failed_machines=list(machine_ids),
84+
success=False,
85+
message=f"Cannot resolve active provider: {exc}",
86+
)
87+
6088
provider_op = ProviderOperation(
6189
operation_type=ProviderOperationType.START_INSTANCES,
6290
parameters={"instance_ids": machine_ids},
6391
)
64-
command = ExecuteProviderOperationCommand(operation=provider_op)
92+
command = ExecuteProviderOperationCommand(
93+
operation=provider_op, strategy_override=strategy_override
94+
)
6595
await self._command_bus.execute(command)
6696

6797
if command.result and command.result.get("success"):

‎src/orb/application/services/orchestration/stop_machines.py‎

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
from orb.application.provider.commands import ExecuteProviderOperationCommand
1010
from orb.application.services.orchestration.base import OrchestratorBase
1111
from orb.application.services.orchestration.dtos import StopMachinesInput, StopMachinesOutput
12+
from orb.application.services.provider_registry_service import ProviderRegistryService
1213
from orb.domain.base.operations import (
1314
Operation as ProviderOperation,
1415
OperationType as ProviderOperationType,
@@ -20,11 +21,16 @@ class StopMachinesOrchestrator(OrchestratorBase[StopMachinesInput, StopMachinesO
2021
"""Orchestrator for stopping machines via the provider layer."""
2122

2223
def __init__(
23-
self, command_bus: CommandBusPort, query_bus: QueryBusPort, logger: LoggingPort
24+
self,
25+
command_bus: CommandBusPort,
26+
query_bus: QueryBusPort,
27+
logger: LoggingPort,
28+
provider_registry_service: ProviderRegistryService,
2429
) -> None:
2530
self._command_bus = command_bus
2631
self._query_bus = query_bus
2732
self._logger = logger
33+
self._provider_registry_service = provider_registry_service
2834

2935
async def execute(self, input: StopMachinesInput) -> StopMachinesOutput: # type: ignore[return]
3036
self._logger.info(
@@ -58,11 +64,35 @@ async def execute(self, input: StopMachinesInput) -> StopMachinesOutput: # type
5864
message="No machines to stop",
5965
)
6066

67+
# Resolve the effective provider identifier so the command handler can
68+
# route the operation to the correct provider strategy. Preference
69+
# order: explicit name > explicit type > active provider from registry.
70+
if input.provider_name:
71+
strategy_override = input.provider_name
72+
elif input.provider_type:
73+
strategy_override = input.provider_type
74+
else:
75+
try:
76+
selection = self._provider_registry_service.select_active_provider()
77+
strategy_override = selection.provider_name
78+
except Exception as exc:
79+
self._logger.error(
80+
"StopMachinesOrchestrator: cannot resolve active provider: %s", exc
81+
)
82+
return StopMachinesOutput(
83+
stopped_machines=[],
84+
failed_machines=list(machine_ids),
85+
success=False,
86+
message=f"Cannot resolve active provider: {exc}",
87+
)
88+
6189
operation = ProviderOperation(
6290
operation_type=ProviderOperationType.STOP_INSTANCES,
6391
parameters={"instance_ids": machine_ids},
6492
)
65-
command = ExecuteProviderOperationCommand(operation=operation)
93+
command = ExecuteProviderOperationCommand(
94+
operation=operation, strategy_override=strategy_override
95+
)
6696
await self._command_bus.execute(command)
6797

6898
if command.result and command.result.get("success"):

‎src/orb/bootstrap/orchestrator_registry.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -210,21 +210,27 @@ def register_orchestrators(container: DIContainer) -> None:
210210
),
211211
)
212212
if not container.is_registered(StopMachinesOrchestrator):
213+
from orb.application.services.provider_registry_service import ProviderRegistryService
214+
213215
container.register_singleton(
214216
StopMachinesOrchestrator,
215217
lambda c: StopMachinesOrchestrator(
216218
command_bus=c.get(CommandBus),
217219
query_bus=c.get(QueryBus),
218220
logger=c.get(LoggingPort),
221+
provider_registry_service=c.get(ProviderRegistryService),
219222
),
220223
)
221224
if not container.is_registered(StartMachinesOrchestrator):
225+
from orb.application.services.provider_registry_service import ProviderRegistryService
226+
222227
container.register_singleton(
223228
StartMachinesOrchestrator,
224229
lambda c: StartMachinesOrchestrator(
225230
command_bus=c.get(CommandBus),
226231
query_bus=c.get(QueryBus),
227232
logger=c.get(LoggingPort),
233+
provider_registry_service=c.get(ProviderRegistryService),
228234
),
229235
)
230236
if not container.is_registered(GetProviderHealthOrchestrator):

‎tests/unit/application/services/orchestration/test_start_machines_orchestrator.py‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,11 +50,27 @@ def mock_logger():
5050

5151

5252
@pytest.fixture
53-
def orchestrator(mock_command_bus, mock_query_bus, mock_logger):
53+
def mock_provider_registry_service():
54+
from orb.application.services.provider_registry_service import ProviderRegistryService
55+
from orb.domain.base.results import ProviderSelectionResult
56+
57+
svc = MagicMock(spec=ProviderRegistryService)
58+
svc.select_active_provider.return_value = ProviderSelectionResult(
59+
provider_type="aws",
60+
provider_name="aws-default",
61+
selection_reason="test_fixture",
62+
confidence=1.0,
63+
)
64+
return svc
65+
66+
67+
@pytest.fixture
68+
def orchestrator(mock_command_bus, mock_query_bus, mock_logger, mock_provider_registry_service):
5469
return StartMachinesOrchestrator(
5570
command_bus=mock_command_bus,
5671
query_bus=mock_query_bus,
5772
logger=mock_logger,
73+
provider_registry_service=mock_provider_registry_service,
5874
)
5975

6076

‎tests/unit/application/services/orchestration/test_stop_machines_orchestrator.py‎

Lines changed: 17 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,11 +52,27 @@ def mock_logger():
5252

5353

5454
@pytest.fixture
55-
def orchestrator(mock_command_bus, mock_query_bus, mock_logger):
55+
def mock_provider_registry_service():
56+
from orb.application.services.provider_registry_service import ProviderRegistryService
57+
from orb.domain.base.results import ProviderSelectionResult
58+
59+
svc = MagicMock(spec=ProviderRegistryService)
60+
svc.select_active_provider.return_value = ProviderSelectionResult(
61+
provider_type="aws",
62+
provider_name="aws-default",
63+
selection_reason="test_fixture",
64+
confidence=1.0,
65+
)
66+
return svc
67+
68+
69+
@pytest.fixture
70+
def orchestrator(mock_command_bus, mock_query_bus, mock_logger, mock_provider_registry_service):
5671
return StopMachinesOrchestrator(
5772
command_bus=mock_command_bus,
5873
query_bus=mock_query_bus,
5974
logger=mock_logger,
75+
provider_registry_service=mock_provider_registry_service,
6076
)
6177

6278

0 commit comments

Comments
 (0)