Workflow Implementation - Continues...

[macromania]

Description

This PR continues the Workflow implementation on top of #857. This PR will be rebased once #857 is merged to main. So, about 20 of the files in this PR would be already reviewed and merged to main 👀

☢️ Due to missing commit sign offs in our setup, I had to rebase and sign-off the commits again. The commit history seems bigger than expected due to that. No work is lost and only commit messages are amended. 🙏🏼

In the meantime, please feel free to review and provide feedback 😅

As seen from the implementation, it involves abstraction of Durable Task Framework in most places. This means some of the functionality are not provided in this PR as they also lack implementation support at Durable Task Framework.

Management API

• raiseEvent
• getInstanceMetadata
• waitForInstanceStart
• waitForInstanceCompletion
• createTaskHub
• deleteTaskHub
• purgeInstance

As mentioned in microsoft/durabletask-java#142, queryInstances and purgeInstances are not supported due to limitations at Dapr datastore.

Authoring API

• getName
• getInput
• getInstanceId
• getCurrentInstant
• getIsReplaying
• allOf
• anyOf
• createTimer
• complete
• callActivity
• continueAsNew
• sendEvent
• callSubOrchestrator
• waitForExternalEvent

Testing

We have provided unit test coverage with mocks for all the SDK methods. Also, we have expanded the example provided in #857 by adding all of the Authoring and Management API calls. We used the example as a mean to test the integrity of the code and also understand if the developer experience is satisfactory.

When the example is run, following output is expected from the DemoWorkflow.

INFO[0006] 834a10a5-6fca-4615-98f2-457b9b29a27a: starting new 'io.dapr.examples.workflows.DemoWorkflow' instance with ID = '834a10a5-6fca-4615-98f2-457b9b29a27a'.  app_id=demoworkflowworker instance=Macbook scope=dapr.runtime.wfengine type=log ver=1.11.0
== APP == 2023-06-20 10:27:01,258 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Starting Workflow: io.dapr.examples.workflows.DemoWorkflow
== APP == 2023-06-20 10:27:01,260 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Instance ID: 834a10a5-6fca-4615-98f2-457b9b29a27a
== APP == 2023-06-20 10:27:01,260 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Current Orchestration Time: 2023-06-20T09:27:01.232Z
== APP == 2023-06-20 10:27:01,260 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Waiting for event: 'TimedOutEvent'...
== APP == 2023-06-20 10:27:11,010 {HH:mm:ss.SSS} [main] WARN  i.d.w.runtime.WorkflowContext - Timed out
== APP == 2023-06-20 10:27:11,010 {HH:mm:ss.SSS} [main] WARN  i.d.w.runtime.WorkflowContext - Timeout of PT10S expired while waiting for an event named 'TimedOutEvent' (ID = 0).
== APP == 2023-06-20 10:27:11,010 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Waiting for event: 'TestEvent'...
== APP == 2023-06-20 10:27:11,010 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Received TestEvent
== APP == 2023-06-20 10:27:11,010 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Parallel Execution - Waiting for all tasks to finish...
== APP == 2023-06-20 10:27:11,024 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - finished task: TestEvent 1 Payload
== APP == 2023-06-20 10:27:11,024 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - finished task: TestEvent 2 Payload
== APP == 2023-06-20 10:27:11,024 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - finished task: TestEvent 3 Payload
== APP == 2023-06-20 10:27:11,024 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - All tasks finished!
== APP == 2023-06-20 10:27:11,024 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Parallel Execution - Waiting for any task to finish...
== APP == 2023-06-20 10:27:11,025 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - One of the tasks finished!
== APP == 2023-06-20 10:27:11,025 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Calling Activity...
== APP == 2023-06-20 10:27:11,086 {HH:mm:ss.SSS} [main] INFO  i.d.e.workflows.DemoWorkflowActivity - Starting Activity: io.dapr.examples.workflows.DemoWorkflowActivity
== APP == 2023-06-20 10:27:11,103 {HH:mm:ss.SSS} [main] INFO  i.d.e.workflows.DemoWorkflowActivity - Message Received from input: Hello Activity!
== APP == 2023-06-20 10:27:11,103 {HH:mm:ss.SSS} [main] INFO  i.d.e.workflows.DemoWorkflowActivity - Sending message to output: Hello Activity! World!, from Activity
== APP == 2023-06-20 10:27:11,103 {HH:mm:ss.SSS} [main] INFO  i.d.e.workflows.DemoWorkflowActivity - Sleeping for 5 seconds to simulate long running operation...
== APP == 2023-06-20 10:27:16,108 {HH:mm:ss.SSS} [main] INFO  i.d.e.workflows.DemoWorkflowActivity - Activity finished
== APP == 2023-06-20 10:27:16,109 {HH:mm:ss.SSS} [main] INFO  i.d.e.workflows.DemoWorkflowActivity - Activity returned: io.dapr.examples.workflows.DemoActivityOutput@69e153c5
== APP == 2023-06-20 10:27:16,162 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Activity returned: io.dapr.examples.workflows.DemoActivityOutput@1ca3b418
== APP == 2023-06-20 10:27:16,162 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Activity returned: Hello Activity! World!, from Activity
== APP == 2023-06-20 10:27:16,162 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Activity returned: Hello Activity!
== APP == 2023-06-20 10:27:16,162 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Waiting for event: 'RestartEvent'...
== APP == 2023-06-20 10:27:26,010 {HH:mm:ss.SSS} [main] WARN  i.d.w.runtime.WorkflowContext - Restart Timed out
== APP == 2023-06-20 10:27:26,011 {HH:mm:ss.SSS} [main] WARN  i.d.w.runtime.WorkflowContext - Timeout of PT10S expired while waiting for an event named 'RestartEvent' (ID = 13).
== APP == 2023-06-20 10:27:26,011 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Child-Workflow> Calling ChildWorkflow...
INFO[0031] a2ff610e-009f-4bcb-af52-2b35241e980e: starting new 'io.dapr.examples.workflows.DemoSubWorkflow' instance with ID = 'a2ff610e-009f-4bcb-af52-2b35241e980e'.  app_id=demoworkflowworker instance=Macbook scope=dapr.runtime.wfengine type=log ver=1.11.0
== APP == 2023-06-20 10:27:26,046 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Child-Workflow> Started: io.dapr.examples.workflows.DemoSubWorkflow
== APP == 2023-06-20 10:27:26,047 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Child-Workflow> Instance ID: a2ff610e-009f-4bcb-af52-2b35241e980e
== APP == 2023-06-20 10:27:26,047 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Child-Workflow> Current Time: 2023-06-20T09:27:26.044Z
== APP == 2023-06-20 10:27:26,047 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Child-Workflow> Input: Hello ChildWorkflow!
== APP == 2023-06-20 10:27:26,047 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Child-Workflow> Completed
INFO[0031] a2ff610e-009f-4bcb-af52-2b35241e980e: 'io.dapr.examples.workflows.DemoSubWorkflow' completed with a COMPLETED status.  app_id=demoworkflowworker instance=MSFT-Macbook scope=dapr.runtime.wfengine type=log ver=1.11.0
== APP == 2023-06-20 10:27:26,063 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Child-Workflow> returned: result: Hello ChildWorkflow!
== APP == 2023-06-20 10:27:26,064 {HH:mm:ss.SSS} [main] INFO  i.d.w.runtime.WorkflowContext - Workflow finished
INFO[0031] 834a10a5-6fca-4615-98f2-457b9b29a27a: 'io.dapr.examples.workflows.DemoWorkflow' completed with a COMPLETED status.  app_id=demoworkflowworker instance=Macbook scope=dapr.runtime.wfengine type=log ver=1.11.0

The above logs show that:

1. Workflow starts
2. Wait for an event called TimedOutEvent and eventually time out
3. Wait for an event called TestEvent and show the event is received
4. Parallel execution of tasks using allOf
5. Parallel execution of tasks using anyOf
6. Starting an Activity
7. Executing the Activity
8. Parsing the result of the Activity
9. Starting a ChildWorkflow

Also, when you run the example to the end, you will be showcased:

1. purgeInstance
2. terminate
3. restart

Issue reference

#839

Checklist

Please make sure you've completed the relevant tasks for this PR, out of the following list:

• Code compiles correctly
• Created/updated tests
• Extended the documentation

[macromania]

@skyao I believe I have covered all the comments in my latest commits.

[macromania]

Another review comment, let me write down it here so that we can discuss it.

Let's design the sub-packages in workflow.

I found that in current implementation of workflow, we have introduced a lot of concepts and their implementation code for :

• workflow
• workflow client
• workflow runtime / workflow runtime status
• workflow instance / workflow instance status
• workflow failure
• workflow activity /workflow activity context
• workflow context
• workflow stub

Now we have two sub packages:

1. client: includes workflow client / workflow failure / workflow instance status
2. runtime: includes workflow runtime and workflow activity /workflow activity context

I suggest that we should review these concepts and the sub-packages.

Another question is that: for the new comers who is trying to use dapr workflow at the first time, how can we let him understand above concepts quickly?

Fair points @skyao 👍🏼

The concept of workflow and running activities as a workflow is not a common paradigm that we could see in most applications. However, when we consider the choreography vs orchestration discussion in microservices architecture, workflows are closer to orchestration.

An orchestrator by design should be able to provide an instance of its workflow running. Also, it should provide a client for an external entities (another program, the developer etc.) to operate on the orchestration. These two concerns are covered by workflow client, workflow instance and workflow runtime.

On the other hand, an orchestrator needs tasks to complete the business objective. In our case, these are covered by workflow activity and workflow activity context.

In this context, someone looking at dapr workflows for the first time needs a clear understanding of which path they are choosing orchestration over choreography. The example project provides the implementation details but the documentation should touch the above concepts in detail.

[skyao]

Great work! Thank for your contribution @macromania !

I have competed my review and confirmed that all my comments have been updated.

@artursouza @mukundansundar please help to have a look at this PR and if OK please help to merge it.

[skyao]

@artursouza @mukundansundar please help to merge this PR.

[skyao]

I need to confirm the design of dapr workflow java sdk:

our dapr workflow java sdk will access backend of durable task directly and bypass dapr's sidecar (including dapr api layer and all workflow components).

To confirm this design, I list a lot of our implementation:

• in dapr/dapr repo, workflow api is added in dapr/proto/runtime/v1/dapr.proto

service Dapr {
  // Starts a new instance of a workflow
  rpc StartWorkflowAlpha1 (StartWorkflowRequest) returns (StartWorkflowResponse) {}

  // Gets details about a started workflow instance
  rpc GetWorkflowAlpha1 (GetWorkflowRequest) returns (GetWorkflowResponse) {}

  // Purge Workflow
  rpc PurgeWorkflowAlpha1 (PurgeWorkflowRequest) returns (google.protobuf.Empty) {}

  // Terminates a running workflow instance
  rpc TerminateWorkflowAlpha1 (TerminateWorkflowRequest) returns (google.protobuf.Empty) {}

  // Pauses a running workflow instance
  rpc PauseWorkflowAlpha1 (PauseWorkflowRequest) returns (google.protobuf.Empty) {}

  // Resumes a paused workflow instance
  rpc ResumeWorkflowAlpha1 (ResumeWorkflowRequest) returns (google.protobuf.Empty) {}

  // Raise an event to a running workflow instance
  rpc RaiseEventWorkflowAlpha1 (RaiseEventWorkflowRequest) returns (google.protobuf.Empty) {}
}

• and also in dapr/dapr repo, we have updated HTTP API and gRPC API to support above workflow API

• in dapr/components-contrib rep, we also added the support for workflow and provided a workflow component implementation for temporal.

• BUT, in dapr/java sdk of current two workflow PRs, we used the java sdk of DurableTask directly:

import com.microsoft.durabletask.DurableTaskClient;

public class DaprWorkflowClient implements AutoCloseable {

  DurableTaskClient innerClient;
  ManagedChannel grpcChannel;

 public <T extends Workflow> String scheduleNewWorkflow(Class<T> clazz) {
    return this.innerClient.scheduleNewOrchestrationInstance(clazz.getCanonicalName());
  }

And the method of dapr workflow api like StartWorkflowAlpha1/GetWorkflowAlpha1 are never called in the two PR.

• I also found similar design in dotnet sdk and python sdk:

• https://github.com/dapr/dotnet-sdk/blob/master/src/Dapr.Workflow/DaprWorkflowClient.cs

• https://github.com/dapr/python-sdk/blob/master/ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_client.py

class DaprWorkflowClient:
    """Defines client operations for managing Dapr Workflow instances.

       This is an alternative to the general purpose Dapr client. It uses a gRPC connection to send
       commands directly to the workflow engine, bypassing the Dapr API layer.

       This client is intended to be used by workflow application, not by general purpose
       application.
    """

So, with current design of java workflow sdk, we only provides support for microsoft durabletask and the implementation is hard code.

[DeepanshuA]

And the method of dapr workflow api like StartWorkflowAlpha1/GetWorkflowAlpha1 are never called in the two PR.

Please refer this PR for python-sdk, where management APIs are used: dapr/python-sdk#554

This PR for dotnet: dapr/dotnet-sdk#1003

[artursouza]

Use generic and the serializer interface to return the actual object. See how it is done for DaprClient

[skyao]

@artursouza this run() method defined in WorkflowActivity doesn't have a input parameter, we need to call ctx.getInput() to get the input, like this:

    public Object run(WorkflowActivityContext ctx) {
        OrderPayload order = ctx.getInput(OrderPayload.class);
        logger.info("Requesting approval for order: {}", order);

        return ApprovalResult.Approved;
    }

I suggest to move this input parameter from ctx to run() method:

public interface WorkflowActivity<Input, Output>{
  Output run(WorkflowActivityContext ctx, Input input);
}

[artursouza]

Exactly. Also, use TypeRef from the main SDK. See how we do it for Actors and state store.

[artursouza]

Please, make this as a separate PR. I will make the other changes and merge this.

[artursouza]

I don't have permissions to edit files in this PR. Please, address the other comments so I can merge this.

[macromania]

just pushed the changes, thanks for your review @artursouza 👍🏼

[codecov]

Codecov Report

Patch coverage: 73.38% and project coverage change: -0.05% ⚠️

Comparison is base (cf8040d) 76.94% compared to head (65b3354) 76.90%.
Report is 1 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #880      +/-   ##
============================================
- Coverage     76.94%   76.90%   -0.05%     
- Complexity     1326     1395      +69     
============================================
  Files           126      132       +6     
  Lines          4091     4230     +139     
  Branches        484      493       +9     
============================================
+ Hits           3148     3253     +105     
- Misses          696      720      +24     
- Partials        247      257      +10     

Files Changed	Coverage Δ
...c/main/java/io/dapr/workflows/WorkflowContext.java	41.17% <41.17%> (ø)
...ava/io/dapr/workflows/runtime/ActivityWrapper.java	64.70% <64.70%> (ø)
...a/io/dapr/workflows/client/DaprWorkflowClient.java	80.95% <68.42%> (-10.36%)	⬇️
...apr/workflows/runtime/WorkflowActivityContext.java	71.42% <71.42%> (ø)
...ava/io/dapr/workflows/DaprWorkflowContextImpl.java	87.50% <78.57%> (-12.50%)	⬇️
.../dapr/workflows/runtime/WorkflowRuntimeStatus.java	79.41% <79.41%> (ø)
.../dapr/workflows/client/WorkflowFailureDetails.java	85.71% <85.71%> (ø)
.../dapr/workflows/client/WorkflowInstanceStatus.java	90.90% <90.90%> (ø)
...dapr/workflows/runtime/WorkflowRuntimeBuilder.java	84.61% <100.00%> (+39.16%)	⬆️

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.