Add Workflow.get_logger()

Author: firdaus

cadence/decision_loop.py

@@ -515,6 +515,12 @@ def record_marker(self, marker_name: str, header: Header, details: bytes):
     def get_version(self, change_id: str, min_supported: int, max_supported: int) -> int:
         return self.workflow_clock.get_version(change_id, min_supported, max_supported)
 
+    def get_logger(self, name) -> logging.Logger:
+        replay_aware_logger = logging.getLogger(name)
+        make_replay_aware(replay_aware_logger)
+        return replay_aware_logger
+
+
 @dataclass
 class ReplayDecider:
     execution_id: str
@@ -939,3 +945,4 @@ def respond_decisions(self, task_token: bytes, decisions: List[Decision]):
 
 
 from cadence.clock_decision_context import ClockDecisionContext, TimerCancellationHandler, LOCAL_ACTIVITY_MARKER_NAME
+from cadence.replay_interceptor import make_replay_aware

cadence/replay_interceptor.py

@@ -0,0 +1,23 @@
+import inspect
+from typing import Callable
+
+
+def get_replay_aware_interceptor(fn: Callable):
+    def interceptor(*args, **kwargs):
+        from cadence.decision_loop import ITask
+        task: ITask = ITask.current()
+        if not task.decider.decision_context.is_replaying():
+            return fn(*args, **kwargs)
+
+    return interceptor
+
+
+def make_replay_aware(target: object):
+    # TODO: Consider using metaclasses instead
+    if hasattr(target, "_cadence_python_intercepted"):
+        return target
+    for name, fn in inspect.getmembers(target):
+        if inspect.ismethod(fn):
+            setattr(target, name, get_replay_aware_interceptor(fn))
+    target._cadence_python_intercepted = True
+    return target

cadence/tests/interceptor_testing_utils.py

@@ -0,0 +1,43 @@
+import logging
+import logging.config
+
+counter_filter_counter = 0
+
+
+class CounterFilter(logging.Filter):
+
+    def filter(self, record):
+        global counter_filter_counter
+        counter_filter_counter += 1
+        return True
+
+
+def reset_counter_filter_counter():
+    global counter_filter_counter
+    counter_filter_counter = 0
+
+
+def get_counter_filter_counter():
+    return counter_filter_counter
+
+
+LOGGING = {
+    'version': 1,
+    'filters': {
+        'counter-filter': {
+            '()': CounterFilter,
+        }
+    },
+    'handlers': {
+        'console': {
+            'class': 'logging.StreamHandler',
+            'filters': ['counter-filter']
+        }
+    },
+    'loggers': {
+        'test-logger': {
+            'level': 'DEBUG',
+            'handlers': ['console']
+        }
+    },
+}

cadence/tests/test_interceptor.py

@@ -0,0 +1,61 @@
+import inspect
+import logging
+import logging.config
+
+from cadence.tests.interceptor_testing_utils import LOGGING, get_counter_filter_counter, reset_counter_filter_counter
+
+a_captured = None
+b_captured = None
+args_captured = None
+kwargs_captured = None
+
+
+class Target:
+    def do_stuff(self, a, b=2):
+        global a_captured, b_captured
+        a_captured = a
+        b_captured = b
+
+
+def interceptor(fn):
+    def intercept(*args, **kwargs):
+        global args_captured, kwargs_captured
+        args_captured = args
+        kwargs_captured = kwargs
+        fn(*args, **kwargs)
+
+    return intercept
+
+
+def test_interceptor():
+    target = Target()
+    for name, fn in inspect.getmembers(target):
+        if inspect.ismethod(fn):
+            setattr(target, name, interceptor(fn))
+    target.do_stuff(1, b=20)
+    assert args_captured == (1,)
+    assert kwargs_captured == {"b": 20}
+    assert a_captured == 1
+    assert b_captured == 20
+
+    target2 = Target()
+    target2.do_stuff(99, b=100)
+    assert args_captured == (1,)
+    assert kwargs_captured == {"b": 20}
+
+
+def test_logger():
+    reset_counter_filter_counter()
+
+    logging.config.dictConfig(LOGGING)
+    logger = logging.getLogger("test-logger")
+    logger.info("1")
+    logger.info("2")
+    logger.info("3")
+
+    logger = logging.getLogger("something-else")
+    logger.info("1")
+    logger.info("2")
+    logger.info("3")
+
+    assert get_counter_filter_counter() == 3

cadence/tests/test_replay_interceptor.py

@@ -0,0 +1,70 @@
+from unittest.mock import MagicMock, Mock
+
+import pytest
+
+from cadence.decision_loop import DecisionContext, ITask
+from cadence.replay_interceptor import make_replay_aware
+import cadence.decision_loop
+
+a_captured = None
+b_captured = None
+
+
+@pytest.fixture()
+def task_decision_context_replaying():
+    task: ITask = Mock()
+    task.decider = Mock()
+    decision_context = MagicMock()
+    decision_context.is_replaying = Mock(return_value=True)
+    task.decider.decision_context = decision_context
+    return task
+
+
+@pytest.fixture()
+def task_decision_context_not_replaying():
+    task: ITask = Mock()
+    task.decider = Mock()
+    decision_context = MagicMock()
+    decision_context.is_replaying = Mock(return_value=False)
+    task.decider.decision_context = decision_context
+    return task
+
+
+@pytest.fixture()
+def target():
+    return Target()
+
+class Target:
+    def do_stuff(self, a, b=1):
+        global a_captured, b_captured
+        a_captured = a
+        b_captured = b
+
+
+@pytest.mark.asyncio
+async def test_get_replay_aware_interceptor_not_replaying(task_decision_context_not_replaying, target: Target):
+    cadence.decision_loop.current_task.set(task_decision_context_not_replaying)
+    global a_captured, b_captured
+    a_captured = None
+    b_captured = None
+    target = Target()
+    original_fn = target.do_stuff
+    make_replay_aware(target)
+    assert target.do_stuff != original_fn
+    target.do_stuff(20, b=30)
+    assert a_captured == 20
+    assert b_captured == 30
+
+
+@pytest.mark.asyncio
+async def test_get_replay_aware_interceptor_replaying(task_decision_context_replaying, target: Target):
+    cadence.decision_loop.current_task.set(task_decision_context_replaying)
+    global a_captured, b_captured
+    a_captured = None
+    b_captured = None
+    original_fn = target.do_stuff
+    make_replay_aware(target)
+    assert target.do_stuff != original_fn
+    target.do_stuff(20, b=30)
+    assert a_captured is None
+    assert b_captured is None

cadence/tests/test_workflow_logger.py

@@ -0,0 +1,51 @@
+import logging
+
+from cadence.tests.interceptor_testing_utils import reset_counter_filter_counter, LOGGING, get_counter_filter_counter
+from cadence.workerfactory import WorkerFactory
+from cadence.workflow import workflow_method, Workflow, WorkflowClient
+
+TASK_LIST = "TestWorkflowLogger"
+DOMAIN = "sample"
+
+
+class TestWorkflowLogger:
+    @workflow_method(task_list=TASK_LIST)
+    async def get_greetings(self) -> list:
+        raise NotImplementedError
+
+
+class TestWorkflowLoggerImpl(TestWorkflowLogger):
+
+    def __init__(self):
+        pass
+
+    async def get_greetings(self):
+        logger = Workflow.get_logger("test-logger")
+        logger.info("********Test %d", 1)
+        await Workflow.sleep(10)
+        logger.info("********Test %d", 2)
+        await Workflow.sleep(10)
+        logger.info("********Test %d", 3)
+        await Workflow.sleep(10)
+        logger.info("********Test %d", 4)
+        await Workflow.sleep(10)
+        logger.info("********Test %d", 5)
+
+
+def test_workflow_logger():
+    reset_counter_filter_counter()
+    logging.config.dictConfig(LOGGING)
+
+    factory = WorkerFactory("localhost", 7933, DOMAIN)
+    worker = factory.new_worker(TASK_LIST)
+    worker.register_workflow_implementation_type(TestWorkflowLoggerImpl)
+    factory.start()
+
+    client = WorkflowClient.new_client(domain=DOMAIN)
+    workflow: TestWorkflowLogger = client.new_workflow_stub(TestWorkflowLogger)
+
+    workflow.get_greetings()
+    assert get_counter_filter_counter() == 5
+
+    print("Stopping workers")
+    worker.stop()

cadence/workflow.py

@@ -82,6 +82,12 @@ def get_version(change_id: str, min_supported: int, max_supported: int):
         decision_context: DecisionContext = task.decider.decision_context
         return decision_context.get_version(change_id, min_supported, max_supported)
 
+    @staticmethod
+    def get_logger(name):
+        from cadence.decision_loop import ITask
+        task: ITask = ITask.current()
+        return task.decider.decision_context.get_logger(name)
+
 
 class WorkflowStub:
     pass