Changes

- Set level to debug for execution logs
- Allow logging config in durabletask-python
- Update docs for logging

Author: andystaples

docs/features.md

@@ -148,4 +148,21 @@ Orchestrations can be suspended using the `suspend_orchestration` client API and
 
 ### Retry policies
 
-Orchestrations can specify retry policies for activities and sub-orchestrations. These policies control how many times and how frequently an activity or sub-orchestration will be retried in the event of a transient error.
+Orchestrations can specify retry policies for activities and sub-orchestrations. These policies control how many times and how frequently an activity or sub-orchestration will be retried in the event of a transient error.
+
+### Logging configuration
+
+Both the TaskHubGrpcWorker and TaskHubGrpcClient (as well as DurableTaskSchedulerWorker and DurableTaskSchedulerClient for durabletask-azuremanaged) accept a log_handler and log_formatter object from `logging`. These can be used to customize verbosity, output location, and format of logs emitted by these sources.
+
+For example, to output logs to a file called `worker.log` at level `DEBUG`, the following syntax might apply:
+
+```python
+log_handler = logging.FileHandler('durable.log', encoding='utf-8')
+log_handler.setLevel(logging.DEBUG)
+
+with DurableTaskSchedulerWorker(host_address=endpoint, secure_channel=secure_channel,
+                                taskhub=taskhub_name, token_credential=credential, log_handler=log_handler) as w:
+```
+
+**NOTE**
+The worker and client output many logs at the `DEBUG` level that will be useful when understanding orchestration flow and diagnosing issues with Durable applications. Before submitting issues, please attempt a repro of the issue with debug logging enabled.

durabletask-azuremanaged/durabletask/azuremanaged/client.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import logging
+
 from typing import Optional
 
 from azure.core.credentials import TokenCredential
@@ -18,7 +20,9 @@ def __init__(self, *,
                  taskhub: str,
                  token_credential: Optional[TokenCredential],
                  secure_channel: bool = True,
-                 default_version: Optional[str] = None):
+                 default_version: Optional[str] = None,
+                 log_handler=None,
+                 log_formatter: Optional[logging.Formatter] = None):
 
         if not taskhub:
             raise ValueError("Taskhub value cannot be empty. Please provide a value for your taskhub")
@@ -31,5 +35,7 @@ def __init__(self, *,
             host_address=host_address,
             secure_channel=secure_channel,
             metadata=None,
+            log_handler=log_handler,
+            log_formatter=log_formatter,
             interceptors=interceptors,
             default_version=default_version)

durabletask-azuremanaged/durabletask/azuremanaged/worker.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+import logging
+
 from typing import Optional
 
 from azure.core.credentials import TokenCredential
@@ -52,12 +54,15 @@ class DurableTaskSchedulerWorker(TaskHubGrpcWorker):
         parameter is set to None since authentication is handled by the
         DTS interceptor.
     """
+
     def __init__(self, *,
                  host_address: str,
                  taskhub: str,
                  token_credential: Optional[TokenCredential],
                  secure_channel: bool = True,
-                 concurrency_options: Optional[ConcurrencyOptions] = None):
+                 concurrency_options: Optional[ConcurrencyOptions] = None,
+                 log_handler=None,
+                 log_formatter: Optional[logging.Formatter] = None):
 
         if not taskhub:
             raise ValueError("The taskhub value cannot be empty.")
@@ -70,5 +75,7 @@ def __init__(self, *,
             host_address=host_address,
             secure_channel=secure_channel,
             metadata=None,
+            log_handler=log_handler,
+            log_formatter=log_formatter,
             interceptors=interceptors,
             concurrency_options=concurrency_options)

durabletask/worker.py

@@ -1238,7 +1238,7 @@ def execute(
         if len(orchestration_started_events) >= 1:
             orchestration_name = orchestration_started_events[0].executionStarted.name
 
-        self._logger.info(
+        self._logger.debug(
             f"{instance_id}: Beginning replay for orchestrator {orchestration_name}..."
         )
 
@@ -1279,7 +1279,7 @@ def execute(
 
         except Exception as ex:
             # Unhandled exceptions fail the orchestration
-            self._logger.info(f"{instance_id}: Orchestration {orchestration_name} failed")
+            self._logger.debug(f"{instance_id}: Orchestration {orchestration_name} failed")
             ctx.set_failed(ex)
 
         if not ctx._is_complete:
@@ -1295,7 +1295,7 @@ def execute(
             completion_status_str = ph.get_orchestration_status_str(
                 ctx._completion_status
             )
-            self._logger.info(
+            self._logger.debug(
                 f"{instance_id}: Orchestration {orchestration_name} completed with status: {completion_status_str}"
             )
 
@@ -1764,7 +1764,7 @@ def execute(
             encoded_input: Optional[str],
     ) -> Optional[str]:
         """Executes an activity function and returns the serialized result, if any."""
-        self._logger.info(
+        self._logger.debug(
             f"{orchestration_id}/{task_id}: Executing activity '{name}'..."
         )
         fn = self._registry.get_activity(name)
@@ -1783,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.info(
+        self._logger.debug(
             f"{orchestration_id}/{task_id}: Activity '{name}' completed successfully with {chars} char(s) of encoded output."
         )
         return encoded_output
@@ -1803,7 +1803,7 @@ def execute(
             encoded_input: Optional[str],
     ) -> Optional[str]:
         """Executes an entity function and returns the serialized result, if any."""
-        self._logger.info(
+        self._logger.debug(
             f"{orchestration_id}: Executing entity '{entity_id}'..."
         )
         fn = self._registry.get_entity(entity_id.entity)
@@ -1837,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.info(
+        self._logger.debug(
             f"{orchestration_id}: Entity '{entity_id}' completed successfully with {chars} char(s) of encoded output."
         )
         return encoded_output