Add/increase execution logging

durabletask/worker.py

@@ -1233,13 +1233,21 @@ def execute(
             old_events: Sequence[pb.HistoryEvent],
             new_events: Sequence[pb.HistoryEvent],
     ) -> ExecutionResults:
+        orchestration_name = "<unknown>"
+        orchestration_started_events = [e for e in old_events if e.HasField("executionStarted")]
+        if len(orchestration_started_events) > 1:
+            orchestration_name = orchestration_started_events[0].executionStarted.name
+
+        self._logger.info(
+            f"{instance_id}: Beginning replay for orchestrator {orchestration_name}..."
+        )
+
         self._entity_state = OrchestrationEntityContext(instance_id)
 
         if not new_events:
             raise task.OrchestrationStateError(
                 "The new history event list must have at least one event in it."
             )
-
         ctx = _RuntimeOrchestrationContext(instance_id, self._registry, self._entity_state)
         try:
             # Rebuild local state by replaying old history into the orchestrator function
@@ -1271,13 +1279,15 @@ def execute(
 
         except Exception as ex:
             # Unhandled exceptions fail the orchestration
+            self._logger.info(f"{instance_id}: Orchestration {orchestration_name} failed")
             ctx.set_failed(ex)
 
         if not ctx._is_complete:
             task_count = len(ctx._pending_tasks)
             event_count = len(ctx._pending_events)
             self._logger.info(
-                f"{instance_id}: Orchestrator yielded with {task_count} task(s) and {event_count} event(s) outstanding."
+                f"{instance_id}: Orchestrator {orchestration_name} yielded with {task_count} task(s) "
+                f"and {event_count} event(s) outstanding."
             )
         elif (
                 ctx._completion_status and ctx._completion_status is not pb.ORCHESTRATION_STATUS_CONTINUED_AS_NEW
@@ -1286,7 +1296,7 @@ def execute(
                 ctx._completion_status
             )
             self._logger.info(
-                f"{instance_id}: Orchestration completed with status: {completion_status_str}"
+                f"{instance_id}: Orchestration {orchestration_name} completed with status: {completion_status_str}"
             )
 
         actions = ctx.get_actions()
@@ -1754,7 +1764,7 @@ def execute(
             encoded_input: Optional[str],
     ) -> Optional[str]:
         """Executes an activity function and returns the serialized result, if any."""
-        self._logger.debug(
+        self._logger.info(
             f"{orchestration_id}/{task_id}: Executing activity '{name}'..."
         )
         fn = self._registry.get_activity(name)
@@ -1773,7 +1783,7 @@ def execute(
             shared.to_json(activity_output) if activity_output is not None else None
         )
         chars = len(encoded_output) if encoded_output else 0
-        self._logger.debug(
+        self._logger.info(
             f"{orchestration_id}/{task_id}: Activity '{name}' completed successfully with {chars} char(s) of encoded output."
         )
         return encoded_output
@@ -1793,7 +1803,7 @@ def execute(
             encoded_input: Optional[str],
     ) -> Optional[str]:
         """Executes an entity function and returns the serialized result, if any."""
-        self._logger.debug(
+        self._logger.info(
             f"{orchestration_id}: Executing entity '{entity_id}'..."
         )
         fn = self._registry.get_entity(entity_id.entity)
@@ -1827,7 +1837,7 @@ def execute(
             shared.to_json(entity_output) if entity_output is not None else None
         )
         chars = len(encoded_output) if encoded_output else 0
-        self._logger.debug(
+        self._logger.info(
             f"{orchestration_id}: Entity '{entity_id}' completed successfully with {chars} char(s) of encoded output."
         )
         return encoded_output