Support orchestration id reuse policy #7308

kaibocai · 2023-12-15T17:41:01Z

Description

Issue reference

This PR tries to update the logic to support reuse orchestration ID, more details can be found microsoft/durabletask-go#42, #7101
Corresponding protobuf updates can be found microsoft/durabletask-protobuf#19

This PR requires Orchestration ID reuse policies microsoft/durabletask-go#51 in durabletask-go
This PR upgrade the OTEL dependency to be compatible with durabletask-go as a newer version is used in durabletask-go
This PR update the purge instance logic a bit so that only instance in [Completed, Failed, Terminated] will be purged.

Please reference the issue this PR will close: #7101

Checklist

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

Code compiles correctly
Created/updated tests
Unit tests passing
End-to-end tests passing
Extended the documentation / Created issue in the https://github.com/dapr/docs/ repo: dapr/docs#[3909]
Specification has been updated / Created issue in the https://github.com/dapr/docs/ repo: dapr/docs#[3909]
Provided sample for the feature / Created issue in the https://github.com/dapr/docs/ repo: dapr/docs#[3909]

Signed-off-by: kaibocai <kaibocai@microsoft.com>

go.mod

pkg/runtime/wfengine/wfengine_test.go

pkg/runtime/wfengine/workflow.go

kaibocai · 2023-12-15T18:04:24Z

tag @cgillum , @ItalyPaleAle , @RyanLettieri for review.

Signed-off-by: kaibocai <kaibocai@microsoft.com>

cgillum · 2023-12-15T20:05:01Z

pkg/runtime/wfengine/wfengine_test.go

 	for _, opt := range GetTestOptions() {
+		suffix := opt(engine)
 		t.Run(opt(engine), func(t *testing.T) {


Should we be using suffix here?

Suggested change

t.Run(opt(engine), func(t *testing.T) {

t.Run(suffix, func(t *testing.T) {

cgillum · 2023-12-15T20:06:47Z

pkg/runtime/wfengine/workflow.go

@@ -62,6 +62,11 @@ type recoverableError struct {
 	cause error
 }

+type PolicyAndEventData struct {


Let's give this a more specific name, like CreateWorkflowInstanceRequest

cgillum · 2023-12-15T20:09:22Z

pkg/runtime/wfengine/workflow.go

@@ -62,6 +62,11 @@ type recoverableError struct {
 	cause error
 }

+type PolicyAndEventData struct {
+	Policy    *api.OrchestrationIdReusePolicy `json:"policy"`
+	EventData []byte                          `json:"eventData"`


For consistency with existing code, I suggest StartEventData.

Suggested change

EventData []byte `json:"eventData"`

StartEventBytes []byte `json:"startEventBytes"`

cgillum · 2023-12-15T20:14:25Z

pkg/runtime/wfengine/workflow.go

+	// orchestration already exists, apply reuse id policy
+	runtimeState := getRuntimeState(actorID, state)
+	runtimeStatus := runtimeState.RuntimeStatus()
+	targetStatusValues := backend.BuildStatusSet(reuseIDPolicy.GetOperationStatus())


This goes back to my comment on the durabletask-go repo, but I don't think we need to expose a BuildStatusSet utility function for this simple purpose. I think a simple for-loop over reuseIDPolicy.GetOperationStatus() and comparing to runtimeStatus on each loop iteration is good enough since we only have a small number of possible status values. This might even be more performant than creating a map and doing hashing operations for the key lookup.

cgillum · 2023-12-15T20:17:29Z

pkg/runtime/wfengine/workflow.go


+func (wf *workflowActor) createInstance(ctx context.Context, actorID string, startEvent *backend.HistoryEvent, state *workflowState) error {


I'm a little confused about the function name here. It's called createInstance but the only thing it does is create a reminder? Can we give this a more specific name, like scheduleWorkflowStart?

cgillum · 2023-12-15T20:19:10Z

pkg/runtime/wfengine/workflow.go

@@ -209,6 +255,36 @@ func (wf *workflowActor) createWorkflowInstance(ctx context.Context, actorID str
 	return wf.saveInternalState(ctx, actorID, state)
 }

+// This method cleans up a workflow associated with the given actorID
+func (wf *workflowActor) cleanupOrchestrationStateInternal(ctx context.Context, actorID string, state *workflowState, requireCompleted bool) error {
+	runtimeState := getRuntimeState(actorID, state)


We've already loaded the runtime state in createWorkflowInstance. Should we reuse that and pass it as a parameter rather than getting the runtime state again?

cgillum · 2023-12-15T20:20:03Z

pkg/runtime/wfengine/workflow.go

+func (wf *workflowActor) cleanupOrchestrationStateInternal(ctx context.Context, actorID string, state *workflowState, requireCompleted bool) error {
+	runtimeState := getRuntimeState(actorID, state)
+	runtimeStatus := runtimeState.RuntimeStatus()
+	isCompleted := runtimeStatus == api.COMPLETED || runtimeStatus == api.TERMINATED || runtimeStatus == api.FAILED


This requires a change in durabletask-go, but I wonder if we should prefix these constants with RUNTIME_STATUS_ so that it's clear what these values are for.

cgillum · 2023-12-15T20:21:14Z

pkg/runtime/wfengine/workflow.go

+func (wf *workflowActor) cleanupOrchestrationStateInternal(ctx context.Context, actorID string, state *workflowState, requireCompleted bool) error {
+	runtimeState := getRuntimeState(actorID, state)
+	runtimeStatus := runtimeState.RuntimeStatus()
+	isCompleted := runtimeStatus == api.COMPLETED || runtimeStatus == api.TERMINATED || runtimeStatus == api.FAILED


Can't we just use the existing runtimeStatus.IsCompleted() for this instead of checking specific status values? That would be more maintainable.

The reason I am not using runtimeStatus.IsCompleted() is that I am not quite sure if it equals [COMPLETED , TERMINATED , FAILED]. From the code, the completedEvent is assigned when the orchestration completed at https://github.com/microsoft/durabletask-go/blob/aa335e20a192395649eb9ede97b57c9f647238ea/backend/runtimestate.go#L134, but when orchestration is terminated we don't assign a completedEvent, instead we create a ExecutionTerminatedEvent at https://github.com/microsoft/durabletask-go/blob/aa335e20a192395649eb9ede97b57c9f647238ea/backend/runtimestate.go#L197, so it seems runtimeStatus.IsCompleted() only check if orchestration is completed but not check if it's terminated/failed.

According to this integration test code, a terminated orchestration results in runtimeStatus.IsCompleted() being true.

The second code block you're referencing is specifically for handling cascading terminate. It's not directly related to how we represent a currently terminated orchestration.

Actually, hang on - I need to double check. The code I pointed to is for orchestration metadata, not the orchestration runtime state object...

Okay, so confirming this wasn't as straightforward as I thought. But the short answer is that a terminated orchestration should always be completed. When an orchestration processes an ExecutionTerminated event, the SDK will generate a completedEvent with a status of "Terminated". This can be seen in the SDK code here.

Got it. Updated!

Signed-off-by: kaibocai <kaibocai@microsoft.com>

codecov · 2023-12-15T20:39:06Z

Codecov Report

Attention: 17 lines in your changes are missing coverage. Please review.

Comparison is base (ef925c9) 64.59% compared to head (0205b9c) 64.60%.
Report is 1 commits behind head on master.

❗ Current head 0205b9c differs from pull request most recent head 33f1fd7. Consider uploading reports for the commit 33f1fd7 to get more accurate results

Files	Patch %	Lines
pkg/runtime/wfengine/workflow.go	65.90%	8 Missing and 7 partials ⚠️
pkg/runtime/wfengine/backend.go	80.00%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #7308      +/-   ##
==========================================
+ Coverage   64.59%   64.60%   +0.01%     
==========================================
  Files         226      226              
  Lines       21191    21223      +32     
==========================================
+ Hits        13689    13712      +23     
- Misses       6324     6329       +5     
- Partials     1178     1182       +4

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

Signed-off-by: kaibocai <kaibocai@microsoft.com> simplify complete check Signed-off-by: kaibocai <kaibocai@microsoft.com>

cgillum

Just some minor things.

pkg/runtime/wfengine/backend.go

cgillum · 2023-12-15T23:37:24Z

pkg/runtime/wfengine/workflow.go

-	}
-	wf.states.Delete(actorID)
-	return nil
+	return wf.cleanupOrchestrationStateInternal(ctx, actorID, state, true && !runtimeState.IsCompleted())


Suggested change

return wf.cleanupOrchestrationStateInternal(ctx, actorID, state, true && !runtimeState.IsCompleted())

return wf.cleanupOrchestrationStateInternal(ctx, actorID, state, !runtimeState.IsCompleted())

cgillum · 2023-12-15T23:40:20Z

pkg/runtime/wfengine/workflow.go

+	switch reuseIDPolicy.GetAction() {
+	case api.REUSE_ID_ACTION_IGNORE:
+		// Log an warning message and ignore creating new instance
+		wfLogger.Warnf("An instance with ID '%s' already exists; dropping duplicate create request", actorID)


For consistency, we should follow the logging format that's used elsewhere in this file. For example:

Suggested change

wfLogger.Warnf("An instance with ID '%s' already exists; dropping duplicate create request", actorID)

wfLogger.Warnf("Workflow actor '%s': ignoring request to recreate the current workflow instance", actorID)

cgillum · 2023-12-15T23:42:03Z

pkg/runtime/wfengine/workflow.go

@@ -209,6 +263,33 @@ func (wf *workflowActor) createWorkflowInstance(ctx context.Context, actorID str
 	return wf.saveInternalState(ctx, actorID, state)
 }

+// This method cleans up a workflow associated with the given actorID
+func (wf *workflowActor) cleanupOrchestrationStateInternal(ctx context.Context, actorID string, state *workflowState, requiredAndNotCompleted bool) error {
+	// Only purge orchestration in the ['COMPLETED', 'FAILED', 'TERMINATED'] statuses,


This comment should probably be updated.

Signed-off-by: kaibocai <kaibocai@microsoft.com> update method name Signed-off-by: kaibocai <kaibocai@microsoft.com>

cgillum

LGTM! I'll go ahead and merge the durabletask-go changes and create a release so that this PR can be unblocked.

kaibocai · 2023-12-18T21:44:34Z

LGTM! I'll go ahead and merge the durabletask-go changes and create a release so that this PR can be unblocked.

sorry that I didn't see this one and created the release myself.

kaibocai · 2023-12-18T21:47:57Z

Hi @cgillum, @ItalyPaleAle and @RyanLettieri, the duraletask-go is updated to version v0.4.0 and I think this PR should be ready for final review and merge. Please help to take another look. Thanks!

Signed-off-by: kaibocai <kaibocai@microsoft.com>

Signed-off-by: kaibocai <kaibocai@microsoft.com> update test time compare Signed-off-by: kaibocai <kaibocai@microsoft.com> fix assert format Signed-off-by: kaibocai <kaibocai@microsoft.com>

kaibocai · 2023-12-18T22:40:23Z

tag @mukundansundar for review as well.

Signed-off-by: Alessandro (Ale) Segala <43508+ItalyPaleAle@users.noreply.github.com>

DeepanshuA · 2023-12-19T14:12:29Z

pkg/runtime/wfengine/workflow.go

+		return wf.createIfCompleted(ctx, runtimeState, actorID, state, startEvent)
+	}
+
+	switch reuseIDPolicy.GetAction() {


The PR LGTM, I just wanted to understand how are we skipping the creation of a new orchestration here, (i.e. the option SkipIfExists from microsoft/durabletask-go#42 (comment))

case api.REUSE_ID_ACTION_IGNORE: // Log an warning message and ignore creating new instance wfLogger.Warnf("Workflow actor '%s': ignoring request to recreate the current workflow instance", actorID) return nil

So if users choose REUSE_ID_ACTION_IGNORE and if there is an existing instance with the same instance ID, then we will enter the above block and just log a message and directly return.

my bad, I missed that Ignore is Skip, thanks for clarification

cgillum · 2023-12-19T21:12:47Z

Hi @artursouza, @yaron2, @mukundansundar. This workflow PR has been signed off and is ready to be merged.

kaibocai requested review from a team as code owners December 15, 2023 17:41

kaibocai added 2 commits December 15, 2023 11:43

support orchestration id reuse policy

6417134

Signed-off-by: kaibocai <kaibocai@microsoft.com>

clean up

e927c1b

Signed-off-by: kaibocai <kaibocai@microsoft.com>

kaibocai force-pushed the kaibocai/reuse-id branch from 8d85b62 to e927c1b Compare December 15, 2023 17:44

kaibocai changed the title ~~support orchestration id reuse policy~~ Support orchestration id reuse policy Dec 15, 2023

lint fix

6f26f68

Signed-off-by: kaibocai <kaibocai@microsoft.com>

ItalyPaleAle reviewed Dec 15, 2023

View reviewed changes

fix nit

d4c1276

Signed-off-by: kaibocai <kaibocai@microsoft.com>

cgillum reviewed Dec 15, 2023

View reviewed changes

minor updates

8e89c1f

Signed-off-by: kaibocai <kaibocai@microsoft.com>

kaibocai force-pushed the kaibocai/reuse-id branch from 0cecc57 to ee6e8f7 Compare December 15, 2023 22:08

minor updates

2517e86

Signed-off-by: kaibocai <kaibocai@microsoft.com> simplify complete check Signed-off-by: kaibocai <kaibocai@microsoft.com>

kaibocai force-pushed the kaibocai/reuse-id branch from ee6e8f7 to 2517e86 Compare December 15, 2023 22:39

cgillum reviewed Dec 15, 2023

View reviewed changes

minor updates on log message, comments, and request type

ab1d3f1

Signed-off-by: kaibocai <kaibocai@microsoft.com> update method name Signed-off-by: kaibocai <kaibocai@microsoft.com>

kaibocai force-pushed the kaibocai/reuse-id branch from e6ecaf9 to ab1d3f1 Compare December 18, 2023 19:49

cgillum approved these changes Dec 18, 2023

View reviewed changes

kaibocai force-pushed the kaibocai/reuse-id branch from 53ca9db to 88df935 Compare December 18, 2023 21:42

update durabletask-go to v0.4.0

12e9e2b

Signed-off-by: kaibocai <kaibocai@microsoft.com>

kaibocai force-pushed the kaibocai/reuse-id branch 2 times, most recently from 914f7ee to 0280640 Compare December 18, 2023 22:18

fix time compare in test

57bf384

Signed-off-by: kaibocai <kaibocai@microsoft.com> update test time compare Signed-off-by: kaibocai <kaibocai@microsoft.com> fix assert format Signed-off-by: kaibocai <kaibocai@microsoft.com>

kaibocai force-pushed the kaibocai/reuse-id branch from 0280640 to 57bf384 Compare December 18, 2023 22:27

💄

630b14f

Signed-off-by: Alessandro (Ale) Segala <43508+ItalyPaleAle@users.noreply.github.com>

Merge branch 'master' into kaibocai/reuse-id

0205b9c

ItalyPaleAle approved these changes Dec 19, 2023

View reviewed changes

DeepanshuA reviewed Dec 19, 2023

View reviewed changes

DeepanshuA approved these changes Dec 19, 2023

View reviewed changes

kaibocai mentioned this pull request Dec 19, 2023

Add example for workflow ID reuse policy dapr/docs#3909

Open

Merge branch 'master' into kaibocai/reuse-id

33f1fd7

yaron2 merged commit b118deb into dapr:master Dec 20, 2023
16 of 20 checks passed

This was referenced Jan 12, 2024

Cascade terminate/Purge Workflow Support #7340

Merged

Fix workflow subOrchestration creation #7385

Merged

ItalyPaleAle added this to the v1.13 milestone Jan 15, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support orchestration id reuse policy #7308

Support orchestration id reuse policy #7308

kaibocai commented Dec 15, 2023 •

edited

kaibocai commented Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

kaibocai Dec 15, 2023

cgillum Dec 15, 2023 •

edited

cgillum Dec 15, 2023

cgillum Dec 15, 2023

kaibocai Dec 15, 2023

codecov bot commented Dec 15, 2023 •

edited

cgillum left a comment

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum Dec 15, 2023

cgillum left a comment

kaibocai commented Dec 18, 2023

kaibocai commented Dec 18, 2023

kaibocai commented Dec 18, 2023

DeepanshuA Dec 19, 2023

kaibocai Dec 19, 2023

DeepanshuA Dec 19, 2023

cgillum commented Dec 19, 2023

	t.Run(opt(engine), func(t *testing.T) {
	t.Run(suffix, func(t *testing.T) {

	EventData []byte `json:"eventData"`
	StartEventBytes []byte `json:"startEventBytes"`


		func (wf workflowActor) createInstance(ctx context.Context, actorID string, startEvent backend.HistoryEvent, state *workflowState) error {

	return wf.cleanupOrchestrationStateInternal(ctx, actorID, state, true && !runtimeState.IsCompleted())
	return wf.cleanupOrchestrationStateInternal(ctx, actorID, state, !runtimeState.IsCompleted())

	wfLogger.Warnf("An instance with ID '%s' already exists; dropping duplicate create request", actorID)
	wfLogger.Warnf("Workflow actor '%s': ignoring request to recreate the current workflow instance", actorID)

Support orchestration id reuse policy #7308

Support orchestration id reuse policy #7308

Conversation

kaibocai commented Dec 15, 2023 • edited

Description

Issue reference

Checklist

kaibocai commented Dec 15, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgillum Dec 15, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Dec 15, 2023 • edited

Codecov Report

cgillum left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgillum left a comment

Choose a reason for hiding this comment

kaibocai commented Dec 18, 2023

kaibocai commented Dec 18, 2023

kaibocai commented Dec 18, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cgillum commented Dec 19, 2023

kaibocai commented Dec 15, 2023 •

edited

cgillum Dec 15, 2023 •

edited

codecov bot commented Dec 15, 2023 •

edited