Add logs to Dapr Workflows

dapr/logger/options.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright 2023 The Dapr Authors
+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.
+"""
+
+from typing import Union
+import logging
+
+
+class LoggerOptions:
+    def __init__(
+            self,
+            log_level: Union[str, None] = None,
+            log_handler: Union[logging.Handler, None] = None,
+            log_formatter: Union[logging.Formatter, None] = None,
+    ):
+        # Set default log level to INFO if none is provided
+        if log_level is None:
+            log_level = logging.INFO
+        # Add a default log handler if none is provided
+        if log_handler is None:
+            log_handler = logging.StreamHandler()
+        # Set a default log formatter if none is provided
+        if log_formatter is None:
+            log_formatter = logging.Formatter(
+                fmt="%(asctime)s.%(msecs)03d %(name)s %(levelname)s: %(message)s",
+                datefmt='%Y-%m-%d %H:%M:%S')
+        self.log_level = log_level
+        self.log_handler = log_handler
+        self.log_formatter = log_formatter
+
+    def get_logger(
+            self,
+            name: str) -> logging.Logger:
+        logger = logging.Logger(name)
+        log_handler = self.log_handler
+        log_handler.setLevel(self.log_level)
+        log_handler.setFormatter(self.log_formatter)
+        logger.handlers.append(log_handler)
+        return logger

ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_client.py

@@ -27,6 +27,7 @@
 from dapr.clients.http.client import DAPR_API_TOKEN_HEADER
 from dapr.conf import settings
 from dapr.conf.helpers import GrpcEndpoint
+from dapr.logger.options import LoggerOptions
 
 T = TypeVar('T')
 TInput = TypeVar('TInput')
@@ -43,19 +44,30 @@ class DaprWorkflowClient:
        application.
     """
 
-    def __init__(self, host: Optional[str] = None, port: Optional[str] = None):
+    def __init__(
+            self,
+            host: Optional[str] = None,
+            port: Optional[str] = None,
+            logger_options: Optional[LoggerOptions] = None):
         address = getAddress(host, port)
 
         try:
             uri = GrpcEndpoint(address)
         except ValueError as error:
             raise DaprInternalError(f'{error}') from error
 
+        if logger_options is None:
+            logger_options = LoggerOptions()
+        self._logger = logger_options.get_logger("DaprWorkflowClient")
+
         metadata = tuple()
         if settings.DAPR_API_TOKEN:
             metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
-        self.__obj = client.TaskHubGrpcClient(host_address=uri.endpoint, metadata=metadata,
-                                              secure_channel=uri.tls)
+        self.__obj = client.TaskHubGrpcClient(host_address=uri.endpoint,
+                                              metadata=metadata,
+                                              secure_channel=uri.tls,
+                                              log_handler=logger_options.log_handler,
+                                              log_formatter=logger_options.log_formatter)
 
     def schedule_new_workflow(self, workflow: Workflow, *, input: Optional[TInput] = None,
                               instance_id: Optional[str] = None,

ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_context.py

@@ -20,6 +20,7 @@
 
 from dapr.ext.workflow.workflow_context import WorkflowContext, Workflow
 from dapr.ext.workflow.workflow_activity_context import WorkflowActivityContext
+from dapr.logger.options import LoggerOptions
 
 T = TypeVar('T')
 TInput = TypeVar('TInput')
@@ -29,8 +30,15 @@
 class DaprWorkflowContext(WorkflowContext):
     """DaprWorkflowContext that provides proxy access to internal OrchestrationContext instance."""
 
-    def __init__(self, ctx: task.OrchestrationContext):
+    def __init__(
+            self,
+            ctx: task.OrchestrationContext,
+            logger_options: Optional[LoggerOptions] = None):
         self.__obj = ctx
+        if logger_options is None:
+            logger_options = LoggerOptions()
+        self._logger_options = logger_options
+        self._logger = logger_options.get_logger("DaprWorkflowContext")
 
     # provide proxy access to regular attributes of wrapped object
     def __getattr__(self, name):
@@ -49,26 +57,33 @@
         return self.__obj.is_replaying
 
     def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
+        self._logger.debug(f'instance_id {self.instance_id} create_timer: {fire_at}')
         return self.__obj.create_timer(fire_at)
 
     def call_activity(self, activity: Callable[[WorkflowActivityContext, TInput], TOutput], *,
                       input: TInput = None) -> task.Task[TOutput]:
+        self._logger.debug(f'instance_id {self.instance_id} call_activity: {activity.__name__}')
         return self.__obj.call_activity(activity=activity.__name__, input=input)
 
     def call_child_workflow(self, workflow: Workflow, *,
                             input: Optional[TInput],
                             instance_id: Optional[str]) -> task.Task[TOutput]:
+        self._logger.debug(f'instance_id {self.instance_id} \
+                           call_child_workflow: {workflow.__name__}')
+
         def wf(ctx: task.OrchestrationContext, inp: TInput):
-            daprWfContext = DaprWorkflowContext(ctx)
+            daprWfContext = DaprWorkflowContext(ctx, self._logger_options)
             return workflow(daprWfContext, inp)
         # copy workflow name so durabletask.worker can find the orchestrator in its registry
         wf.__name__ = workflow.__name__
         return self.__obj.call_sub_orchestrator(wf, input=input, instance_id=instance_id)
 
     def wait_for_external_event(self, name: str) -> task.Task:
+        self._logger.debug(f'instance_id {self.instance_id} wait_for_external_event: {name}')
         return self.__obj.wait_for_external_event(name)
 
     def continue_as_new(self, new_input: Any, *, save_events: bool = False) -> None:
+        self._logger.debug(f'instance_id {self.instance_id} continue_as_new')
         self.__obj.continue_as_new(new_input, save_events=save_events)
 
 

ext/dapr-ext-workflow/dapr/ext/workflow/workflow_runtime.py

@@ -26,6 +26,7 @@
 from dapr.clients.http.client import DAPR_API_TOKEN_HEADER
 from dapr.conf import settings
 from dapr.conf.helpers import GrpcEndpoint
+from dapr.logger.options import LoggerOptions
 
 T = TypeVar('T')
 TInput = TypeVar('TInput')
@@ -36,7 +37,15 @@
     """WorkflowRuntime is the entry point for registering workflows and activities.
     """
 
-    def __init__(self, host: Optional[str] = None, port: Optional[str] = None):
+    def __init__(
+            self,
+            host: Optional[str] = None,
+            port: Optional[str] = None,
+            logger_options: Optional[LoggerOptions] = None):
+        if logger_options is None:
+            logger_options = LoggerOptions()
+        self._logger_options = logger_options
+        self._logger = logger_options.get_logger("WorkflowRuntime")
         metadata = tuple()
         if settings.DAPR_API_TOKEN:
             metadata = ((DAPR_API_TOKEN_HEADER, settings.DAPR_API_TOKEN),)
@@ -47,13 +56,18 @@
         except ValueError as error:
             raise DaprInternalError(f'{error}') from error
 
-        self.__worker = worker.TaskHubGrpcWorker(host_address=uri.endpoint, metadata=metadata,
-                                                 secure_channel=uri.tls)
+        self.__worker = worker.TaskHubGrpcWorker(host_address=uri.endpoint,
+                                                 metadata=metadata,
+                                                 secure_channel=uri.tls,
+                                                 log_handler=logger_options.log_handler,
+                                                 log_formatter=logger_options.log_formatter)
 
     def register_workflow(self, fn: Workflow):
+        self._logger.info(f"Registering workflow '{fn.__name__}' with runtime")
+
         def orchestrationWrapper(ctx: task.OrchestrationContext, inp: Optional[TInput] = None):
             """Responsible to call Workflow function in orchestrationWrapper"""
-            daprWfContext = DaprWorkflowContext(ctx)
+            daprWfContext = DaprWorkflowContext(ctx, self._logger_options)
             if inp is None:
                 return fn(daprWfContext)
             return fn(daprWfContext, inp)
@@ -64,6 +78,8 @@
         """Registers a workflow activity as a function that takes
            a specified input type and returns a specified output type.
         """
+        self._logger.info(f"Registering activity '{fn.__name__}' with runtime")
+
         def activityWrapper(ctx: task.ActivityContext, inp: Optional[TInput] = None):
             """Responsible to call Activity function in activityWrapper"""
             wfActivityContext = WorkflowActivityContext(ctx)

ext/dapr-ext-workflow/tests/test_dapr_workflow_context.py

@@ -28,6 +28,8 @@
 
 
 class FakeOrchestrationContext:
+    def __init__(self):
+        self.instance_id = mock_instance_id
 
     def create_timer(self, fire_at):
         return mock_create_timer