Added gRPC worker, client, and e2e test (#3)

Author: cgillum

README.md

@@ -17,6 +17,13 @@ If the gRPC proto definitions need to be updated, the corresponding source code
 python3 -m grpc_tools.protoc --proto_path=./submodules/durabletask-protobuf/protos  --python_out=./durabletask/protos --pyi_out=./durabletask/protos --grpc_python_out=./durabletask/protos orchestrator_service.proto
 ```
 
+### Running E2E tests
+The E2E (end-to-end) tests require a sidecar process to be running. You can run a sidecar process using the following `docker` command (assumes you have Docker installed on your local system.)
+
+```sh
+docker run --name durabletask-sidecar -p 4001:4001 --env 'DURABLETASK_SIDECAR_LOGLEVEL=Debug' --rm cgillum/durabletask-sidecar:latest start --backend Emulator
+```
+
 ## Contributing
 
 This project welcomes contributions and suggestions.  Most contributions require you to agree to a

durabletask/api/state.py

@@ -0,0 +1,35 @@
+from dataclasses import dataclass
+from datetime import datetime
+
+import durabletask.protos.orchestrator_service_pb2 as pb
+from durabletask.protos.orchestrator_service_pb2 import TaskFailureDetails
+
+
+@dataclass
+class OrchestrationState:
+    instance_id: str
+    name: str
+    runtime_status: pb.OrchestrationStatus
+    created_at: datetime
+    last_updated_at: datetime
+    serialized_input: str | None
+    serialized_output: str | None
+    serialized_custom_status: str | None
+    failure_details: pb.TaskFailureDetails | None
+
+
+def new_orchestration_state(instance_id: str, res: pb.GetInstanceResponse) -> OrchestrationState | None:
+    if not res.exists:
+        return None
+
+    state = res.orchestrationState
+    return OrchestrationState(
+        instance_id,
+        state.name,
+        state.orchestrationStatus,
+        state.createdTimestamp.ToDatetime(),
+        state.lastUpdatedTimestamp.ToDatetime(),
+        state.input.value if state.input is not None else None,
+        state.output.value if state.output is not None else None,
+        state.customStatus.value if state.customStatus is not None else None,
+        state.failureDetails if state.failureDetails.errorMessage != '' or state.failureDetails.errorType != '' else None)

durabletask/api/task_hub_client.py

@@ -0,0 +1,92 @@
+from dataclasses import dataclass
+from datetime import datetime
+import logging
+from typing import Any
+import uuid
+import grpc
+from durabletask.api.state import OrchestrationState, new_orchestration_state
+import durabletask.protos.orchestrator_service_pb2 as pb
+import durabletask.internal.shared as shared
+import simplejson as json
+
+from google.protobuf import timestamp_pb2, wrappers_pb2
+
+from durabletask.protos.orchestrator_service_pb2_grpc import TaskHubSidecarServiceStub
+
+
+class TaskHubGrpcClient:
+
+    def __init__(self, *,
+                 host_address: str | None = None,
+                 log_handler=None,
+                 log_formatter: logging.Formatter | None = None):
+        channel = shared.get_grpc_channel(host_address)
+        self._stub = TaskHubSidecarServiceStub(channel)
+        self._logger = shared.get_logger(log_handler, log_formatter)
+
+    def schedule_new_orchestration(self, name: str, *,
+                                   input: Any = None,
+                                   instance_id: str | None = None,
+                                   start_at: datetime | None = None) -> str:
+        req = pb.CreateInstanceRequest(name=name)
+        if instance_id is None:
+            instance_id = uuid.uuid4().hex
+        req.instanceId = instance_id
+
+        if input is not None:
+            json_input = json.dumps(input)
+            req.input = wrappers_pb2.StringValue(value=json_input)
+
+        if start_at is not None:
+            req.scheduledStartTimestamp = timestamp_pb2.Timestamp()
+            req.scheduledStartTimestamp.FromDatetime(start_at)
+        self._logger.info(f"Starting new '{name}' instance with ID = '{instance_id}'.")
+        res: pb.CreateInstanceResponse = self._stub.StartInstance(req)
+        return res.instanceId
+
+    def get_orchestration_state(self, instance_id: str, *, fetch_payloads: bool = True) -> OrchestrationState | None:
+        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        res: pb.GetInstanceResponse = self._stub.GetInstance(req)
+        return new_orchestration_state(req.instanceId, res)
+
+    def wait_for_orchestration_start(self, instance_id: str, *,
+                                     fetch_payloads: bool = False,
+                                     timeout: int = 60) -> OrchestrationState | None:
+        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        try:
+            self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to start.")
+            res: pb.GetInstanceResponse = self._stub.WaitForInstanceStart(req, timeout=timeout)
+            return new_orchestration_state(req.instanceId, res)
+        except grpc.RpcError as rpc_error:
+            if rpc_error.code() == grpc.StatusCode.DEADLINE_EXCEEDED:  # type: ignore
+                # Replace gRPC error with the built-in TimeoutError
+                raise TimeoutError("Timed-out waiting for the orchestration to start")
+            else:
+                raise
+
+    def wait_for_orchestration_completion(self, instance_id: str, *,
+                                          fetch_payloads: bool = True,
+                                          timeout: int = 60) -> OrchestrationState | None:
+        req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads)
+        try:
+            self._logger.info(f"Waiting {timeout}s for instance '{instance_id}' to complete.")
+            res: pb.GetInstanceResponse = self._stub.WaitForInstanceCompletion(req, timeout=timeout)
+            return new_orchestration_state(req.instanceId, res)
+        except grpc.RpcError as rpc_error:
+            if rpc_error.code() == grpc.StatusCode.DEADLINE_EXCEEDED:  # type: ignore
+                # Replace gRPC error with the built-in TimeoutError
+                raise TimeoutError("Timed-out waiting for the orchestration to complete")
+            else:
+                raise
+
+    def terminate_orchestration(self):
+        pass
+
+    def suspend_orchestration(self):
+        pass
+
+    def resume_orchestration(self):
+        pass
+
+    def raise_orchestration_event(self):
+        pass

durabletask/internal/shared.py

@@ -0,0 +1,32 @@
+import logging
+
+import grpc
+
+
+def get_default_host_address() -> str:
+    return "localhost:4001"
+
+
+def get_grpc_channel(host_address: str | None) -> grpc.Channel:
+    if host_address is None:
+        host_address = get_default_host_address()
+    channel = grpc.insecure_channel(host_address)
+    return channel
+
+
+def get_logger(log_handler: logging.Handler | None, log_formatter: logging.Formatter | None) -> logging.Logger:
+    logger = logging.Logger("durabletask")
+
+    # Add a default log handler if none is provided
+    if log_handler is None:
+        log_handler = logging.StreamHandler()
+        log_handler.setLevel(logging.INFO)
+    logger.handlers.append(log_handler)
+
+    # Set a default log formatter to our handler 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')
+    log_handler.setFormatter(log_formatter)
+    return logger

durabletask/protos/helpers.py

@@ -57,19 +57,23 @@ def new_complete_orchestration_action(
         id: int,
         status: OrchestrationStatus,
         result: str | None = None,
-        failureDetails: TaskFailureDetails | None = None) -> OrchestratorAction:
-    resultWrapper = None
+        failure_details: TaskFailureDetails | None = None) -> OrchestratorAction:
+
+    result_pb: wrappers_pb2.StringValue | None = None
     if result is not None:
-        resultWrapper = wrappers_pb2.StringValue(value=result)
-    return OrchestratorAction(id=id, completeOrchestration=CompleteOrchestrationAction(
+        result_pb = wrappers_pb2.StringValue(value=result)
+
+    completeOrchestrationAction = CompleteOrchestrationAction(
         orchestrationStatus=status,
-        result=resultWrapper,
-        carryoverEvents=None,  # TODO: Populate carryoverEvents
-        failureDetails=failureDetails,
-    ))
+        result=result_pb,
+        failureDetails=failure_details)
+
+    # TODO: CarryoverEvents
+
+    return OrchestratorAction(id=id, completeOrchestration=completeOrchestrationAction)
 
 
-def create_timer_action(id: int, fire_at: datetime) -> OrchestratorAction:
+def new_create_timer_action(id: int, fire_at: datetime) -> OrchestratorAction:
     timestamp = timestamp_pb2.Timestamp()
     timestamp.FromDatetime(fire_at)
     return OrchestratorAction(id=id, createTimer=CreateTimerAction(fireAt=timestamp))