[Node] improve error logging

Author: GuillaumeDSM

packages/flow/octobot_flow/entities/actions/actions_dependencies.py

@@ -1,6 +1,8 @@
 import typing
+import dataclasses
 
 import octobot_commons.dsl_interpreter
+import octobot_commons.logging
 
 import octobot_flow.entities.actions.action_details as action_details
 import octobot_flow.enums
@@ -93,7 +95,21 @@ def read_dependency_result(
             )
         value = dependency_action.result
         if dependency.result_path:
-            value = _navigate_dict_path(value, dependency.result_path)
+            try:
+                value = _navigate_dict_path(value, dependency.result_path)
+            except octobot_flow.errors.ActionDependencyError as err:
+                dependency_dict = dataclasses.asdict(dependency)
+                result_keys = (
+                    list(value) if isinstance(value, dict)
+                    else octobot_commons.logging.get_private_minimized_message_if_necessary(value)
+                )
+                error_message = (
+                    f"Impossible to resolve dependency {dependency_dict} "
+                    f"result path {dependency.result_path} from action {dependency_action.id} "
+                    f"with action result shape: {result_keys}"
+                ) + f": {err}"
+                # add as much info as possible to the error message
+                raise octobot_flow.errors.ActionDependencyError(error_message) from err
         return value
 
     def _dependency_values_for_dynamic_dependencies(

packages/flow/tests/entities/actions/test_actions_dependencies.py

@@ -13,9 +13,11 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
+import mock
 import pytest
 
 import octobot_commons.constants as commons_constants
+import octobot_commons.logging
 
 import octobot_flow.entities.actions.action_details as action_details
 import octobot_flow.entities.actions.actions_dependencies as actions_dependencies
@@ -188,6 +190,70 @@ def test_raises_when_dependency_action_errored(self):
         with pytest.raises(flow_errors.ActionDependencyError):
             resolver.read_dependency_result(dependency)
 
+    def test_raises_extended_error_when_result_path_key_missing(self):
+        dependency_action = action_details.DSLScriptActionDetails(
+            id="action_source",
+            dsl_script="exchange_action()",
+            dependencies=[],
+        )
+        dependency_action.complete(result={"nested": {"order_id": "abc-123"}})
+        dependency = action_details.ActionDependency(
+            action_id="action_source",
+            parameter="exchange_order_id",
+            result_path=["nested", "missing_key"],
+        )
+        resolver = actions_dependencies.ActionsDependenciesResolver(_actions_by_id(dependency_action))
+
+        with mock.patch.object(
+            octobot_commons.logging,
+            "get_private_minimized_message_if_necessary",
+        ) as minimize_message_mock:
+            with pytest.raises(flow_errors.ActionDependencyError) as error_info:
+                resolver.read_dependency_result(dependency)
+
+        minimize_message_mock.assert_not_called()
+        error_message = str(error_info.value)
+        assert "Impossible to resolve dependency" in error_message
+        assert "action_id': 'action_source'" in error_message
+        assert "result_path': ['nested', 'missing_key']" in error_message
+        assert "from action action_source" in error_message
+        assert "with action result shape: ['nested']" in error_message
+        assert "missing key: missing_key" in error_message
+        assert isinstance(error_info.value.__cause__, flow_errors.ActionDependencyError)
+
+    def test_raises_extended_error_when_result_path_list_index_out_of_range(self):
+        dependency_action = action_details.DSLScriptActionDetails(
+            id="action_source",
+            dsl_script="exchange_action()",
+            dependencies=[],
+        )
+        dependency_action.complete(result=[{"id": "a"}])
+        list_result = dependency_action.result
+        dependency = action_details.ActionDependency(
+            action_id="action_source",
+            parameter="exchange_order_id",
+            result_path=["5"],
+        )
+        resolver = actions_dependencies.ActionsDependenciesResolver(_actions_by_id(dependency_action))
+
+        with mock.patch.object(
+            octobot_commons.logging,
+            "get_private_minimized_message_if_necessary",
+            return_value="minimized-list-shape",
+        ) as minimize_message_mock:
+            with pytest.raises(flow_errors.ActionDependencyError) as error_info:
+                resolver.read_dependency_result(dependency)
+
+        minimize_message_mock.assert_called_once_with(list_result)
+        error_message = str(error_info.value)
+        assert "Impossible to resolve dependency" in error_message
+        assert "action_id': 'action_source'" in error_message
+        assert "result_path': ['5']" in error_message
+        assert "from action action_source" in error_message
+        assert "with action result shape: minimized-list-shape" in error_message
+        assert "list index 5 out of range" in error_message
+        assert isinstance(error_info.value.__cause__, flow_errors.ActionDependencyError)
+
 
 class TestActionsDependenciesResolverFilledAllDependencies:
     def test_returns_true_when_all_dependencies_completed(self):

packages/node/octobot_node/scheduler/octobot_flow_client.py

@@ -16,9 +16,9 @@
 import typing
 import dataclasses
 import json
-import logging
 
 import octobot_commons.dataclasses
+import octobot_commons.logging
 
 import octobot_node.errors as errors
 import octobot_node.scheduler.workflows_util as workflows_util
@@ -149,7 +149,7 @@ async def run(self) -> None:
                 self.priority_user_actions
                 or automation_job.automation_state.automation.actions_dag.get_executable_actions()
             )
-            logging.getLogger(self.__class__.__name__).info(f"Running automation actions: {selected_actions}")
+            octobot_commons.logging.get_logger(self.__class__.__name__).info(f"Running automation actions: {selected_actions}")
             executed_actions = await automation_job.run()
             self.after_execution_state = automation_job.automation_state
             post_execution_state_dump = automation_job.dump()

packages/node/octobot_node/scheduler/workflows/automation_workflow.py

@@ -199,6 +199,12 @@ async def execute_iteration(inputs: dict, actions_update: typing.Optional[dict])
                     has_next_actions_override = True
                     next_iteration_description_override = parsed_inputs.task.content
                     next_iteration_description_metadata_override = parsed_inputs.task.content_metadata
+                except Exception as err:
+                    # log propagated errors to also associate them to the automation's error tracking
+                    AutomationWorkflow.get_logger(parsed_inputs).exception(
+                        err, True, f"Error while running automation job: {err}"
+                    )
+                    raise
                 if result.processed_actions:
                     if latest_step := AutomationWorkflow._get_actions_summary(result.processed_actions, minimal=True):
                         executed_step = latest_step

…cheduler/test_octobot_flow_client_lib.py …ts/scheduler/test_octobot_flow_client.pypackages/node/tests/scheduler/test_octobot_flow_client_lib.py renamed to packages/node/tests/scheduler/test_octobot_flow_client.py packages/node/tests/scheduler/test_octobot_flow_client_lib.py renamed to packages/node/tests/scheduler/test_octobot_flow_client.py

@@ -128,6 +128,27 @@ def _apply_octobot_actions_job_result_template(
     target.should_stop = template.should_stop
 
 
+def _assert_iteration_job_errors_logged(
+    mock_logger: mock.Mock,
+    raised_exception: BaseException,
+    *,
+    iteration_failure_count: int,
+    expect_workflow_interrupted_log: bool = False,
+) -> None:
+    expected_call_count = iteration_failure_count + (1 if expect_workflow_interrupted_log else 0)
+    assert mock_logger.exception.call_count == expected_call_count
+    for call_index in range(iteration_failure_count):
+        logged_exception, publish_error, error_message = mock_logger.exception.call_args_list[call_index][0]
+        assert isinstance(logged_exception, type(raised_exception))
+        assert str(logged_exception) == str(raised_exception)
+        assert publish_error is True
+        assert error_message == f"Error while running automation job: {logged_exception}"
+    if expect_workflow_interrupted_log:
+        assert "Interrupted workflow: unexpected critical error: " in str(
+            mock_logger.exception.call_args_list[-1][0][2]
+        )
+
+
 def _octobot_actions_job_mock_class(
     *,
     run_on_result: typing.Callable[[octobot_flow_client.OctoBotActionsJobResult], typing.Any] | None = None,
@@ -452,7 +473,20 @@ async def test_execute_automation(
                 assert parsed_output.error == expected_error_status
                 assert parsed_output.error_message == expected_error_message
                 assert run_mock.await_count == expected_run_await_count
-                mock_logger.exception.assert_called_once()
+                if expected_run_await_count > 1:
+                    _assert_iteration_job_errors_logged(
+                        mock_logger,
+                        raised_exception,
+                        iteration_failure_count=expected_run_await_count,
+                        expect_workflow_interrupted_log=True,
+                    )
+                else:
+                    _assert_iteration_job_errors_logged(
+                        mock_logger,
+                        raised_exception,
+                        iteration_failure_count=1,
+                        expect_workflow_interrupted_log=True,
+                    )
                 mock_process.assert_not_called()
 
 
@@ -567,6 +601,40 @@ async def test_execute_iteration_missing_trading_signal_sets_no_trading_signal_e
         assert parsed_progress_status.error == octobot_flow.enums.ActionErrorStatus.NO_TRADING_SIGNAL.value
         assert parsed_progress_status.error_message == error_message
 
+    @pytest.mark.asyncio
+    @required_imports
+    async def test_execute_iteration_logs_and_reraises_when_octobot_actions_job_fails(
+        self, import_automation_workflow, task
+    ):
+        task.content = json.dumps({"params": {"ACTIONS": "trade", "EXCHANGE_FROM": "binance",
+            "ORDER_SYMBOL": "ETH/BTC", "ORDER_AMOUNT": 1, "ORDER_TYPE": "market",
+            "ORDER_SIDE": "BUY", "SIMULATED_PORTFOLIO": {"BTC": 1}}})
+        inputs = params.AutomationWorkflowInputs(task=task, execution_time=0).to_dict(include_default_values=False)
+        run_error = RuntimeError("automation failed")
+        mock_octobot_actions_job_class, _ = _octobot_actions_job_mock_class(
+            run_side_effect=run_error,
+        )
+        mock_logger = mock.Mock()
+        automation_workflow = octobot_node.scheduler.workflows.automation_workflow.AutomationWorkflow
+
+        with mock.patch.object(
+            octobot_flow_client,
+            "OctoBotActionsJob",
+            mock_octobot_actions_job_class,
+        ), mock.patch.object(
+            automation_workflow,
+            "get_logger",
+            return_value=mock_logger,
+        ):
+            with pytest.raises(RuntimeError, match="automation failed"):
+                await automation_workflow.execute_iteration(inputs, None)
+
+        mock_logger.exception.assert_called_once_with(
+            run_error,
+            True,
+            f"Error while running automation job: {run_error}",
+        )
+
     @pytest.mark.asyncio
     @required_imports
     async def test_execute_iteration_authentication_error_sets_postponed_iteration(
@@ -1680,7 +1748,11 @@ async def run_with_retries_then_apply_success(*args, **kwargs) -> None:
             assert parsed_output == _parse_automation_workflow_output(wf_status.output)
 
         assert run_mock.await_count == max_attempts
-        mock_logger.exception.assert_not_called()
+        _assert_iteration_job_errors_logged(
+            mock_logger,
+            RuntimeError("simulated transient failure"),
+            iteration_failure_count=max_attempts - 1,
+        )
 
     @pytest.mark.asyncio
     @required_imports
@@ -1745,8 +1817,12 @@ async def test_execute_automation_execute_iteration_exhausts_retries_when_octobo
             assert wf_status.output == workflow_result
 
         assert run_mock.await_count == max_attempts
-        mock_logger.exception.assert_called_once()
-        assert "Interrupted workflow: unexpected critical error: " in str(mock_logger.exception.call_args[0][2])
+        _assert_iteration_job_errors_logged(
+            mock_logger,
+            RuntimeError("persistent failure"),
+            iteration_failure_count=max_attempts,
+            expect_workflow_interrupted_log=True,
+        )
 
     @pytest.mark.asyncio
     @required_imports