Add JitterStart support to client

uber-go · Jul 14, 2022 · 7a8ab81 · 7a8ab81
1 parent 08bc387
commit 7a8ab81
Show file tree

Hide file tree

Showing 13 changed files with 355 additions and 14 deletions.
diff --git a/.gen/go/shared/shared.go b/.gen/go/shared/shared.go
diff --git a/go.mod b/go.mod
@@ -16,7 +16,7 @@ require (
 	github.com/stretchr/testify v1.4.0
 	github.com/twmb/murmur3 v1.1.6 // indirect
 	github.com/uber-go/tally/v4 v4.1.1
-	github.com/uber/cadence-idl v0.0.0-20220223020740-f2f5b7fc2bbd
+	github.com/uber/cadence-idl v0.0.0-20220713235846-fda89e95df1e
 	github.com/uber/jaeger-client-go v2.22.1+incompatible
 	github.com/uber/tchannel-go v1.16.0
 	go.uber.org/atomic v1.9.0

diff --git a/go.sum b/go.sum
@@ -194,8 +194,8 @@ github.com/uber-go/tally v3.3.15+incompatible h1:9hLSgNBP28CjIaDmAuRTq9qV+UZY+9P
 github.com/uber-go/tally v3.3.15+incompatible/go.mod h1:YDTIBxdXyOU/sCWilKB4bgyufu1cEi0jdVnRdxvjnmU=
 github.com/uber-go/tally/v4 v4.1.1 h1:jhy6WOZp4nHyCqeV43x3Wz370LXUGBhgW2JmzOIHCWI=
 github.com/uber-go/tally/v4 v4.1.1/go.mod h1:aXeSTDMl4tNosyf6rdU8jlgScHyjEGGtfJ/uwCIf/vM=
-github.com/uber/cadence-idl v0.0.0-20220223020740-f2f5b7fc2bbd h1:KGLWWhlM4I1++coccaIqGJyC4x8Lyof+42lPZP05aaI=
-github.com/uber/cadence-idl v0.0.0-20220223020740-f2f5b7fc2bbd/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
+github.com/uber/cadence-idl v0.0.0-20220713235846-fda89e95df1e h1:N5yzwaoEzjQ7YlXU0UDJeOBrTt/EGLsrqmtu1YSbvdk=
+github.com/uber/cadence-idl v0.0.0-20220713235846-fda89e95df1e/go.mod h1:oyUK7GCNCRHCCyWyzifSzXpVrRYVBbAMHAzF5dXiKws=
 github.com/uber/jaeger-client-go v2.22.1+incompatible h1:NHcubEkVbahf9t3p75TOCR83gdUHXjRJvjoBh1yACsM=
 github.com/uber/jaeger-client-go v2.22.1+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
 github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=

diff --git a/idls b/idls
diff --git a/internal/client.go b/internal/client.go
@@ -413,6 +413,11 @@ type (
 		// The resolution is seconds.
 		// Optional: defaulted to 0 seconds
 		DelayStart time.Duration
+
+		// JitterStart - Seconds to jitter the workflow start. For example, if set to 10, the workflow will start some time between 0-10 seconds.
+		// This works with CronSchedule and with DelayStart.
+		// Optional: defaulted to 0 seconds
+		JitterStart time.Duration
 	}
 
 	// RetryPolicy defines the retry policy.

diff --git a/internal/compatibility/proto/decision.go b/internal/compatibility/proto/decision.go
@@ -140,6 +140,7 @@ func Decision(d *shared.Decision) *apiv1.Decision {
 				Header:                       Header(attr.Header),
 				Memo:                         Memo(attr.Memo),
 				SearchAttributes:             SearchAttributes(attr.SearchAttributes),
+				JitterStart:                  secondsToDuration(attr.JitterStartSeconds),
 			},
 		}
 	case shared.DecisionTypeStartChildWorkflowExecution:

diff --git a/internal/compatibility/proto/history.go b/internal/compatibility/proto/history.go
@@ -540,6 +540,7 @@ func StartChildWorkflowExecutionInitiatedEventAttributes(t *shared.StartChildWor
 		Memo:                         Memo(t.Memo),
 		SearchAttributes:             SearchAttributes(t.SearchAttributes),
 		DelayStart:                   secondsToDuration(t.DelayStartSeconds),
+		JitterStart:                  secondsToDuration(t.JitterStartSeconds),
 	}
 }
 

diff --git a/internal/compatibility/proto/request.go b/internal/compatibility/proto/request.go
@@ -408,6 +408,7 @@ func SignalWithStartWorkflowExecutionRequest(t *shared.SignalWithStartWorkflowEx
 			SearchAttributes:             SearchAttributes(t.SearchAttributes),
 			Header:                       Header(t.Header),
 			DelayStart:                   secondsToDuration(t.DelayStartSeconds),
+			JitterStart:                  secondsToDuration(t.JitterStartSeconds),
 		},
 		SignalName:  t.GetSignalName(),
 		SignalInput: Payload(t.SignalInput),
@@ -451,6 +452,7 @@ func StartWorkflowExecutionRequest(t *shared.StartWorkflowExecutionRequest) *api
 		SearchAttributes:             SearchAttributes(t.SearchAttributes),
 		Header:                       Header(t.Header),
 		DelayStart:                   secondsToDuration(t.DelayStartSeconds),
+		JitterStart:                  secondsToDuration(t.JitterStartSeconds),
 	}
 }
 

diff --git a/internal/compatibility/testdata/decision.go b/internal/compatibility/testdata/decision.go
@@ -105,6 +105,7 @@ var (
 		Header:                       &Header,
 		Memo:                         &Memo,
 		SearchAttributes:             &SearchAttributes,
+		JitterStart:                  Duration4,
 	}
 	FailWorkflowExecutionDecisionAttributes = apiv1.FailWorkflowExecutionDecisionAttributes{
 		Failure: &Failure,

diff --git a/internal/compatibility/thrift/decision.go b/internal/compatibility/thrift/decision.go
@@ -134,6 +134,7 @@ func Decision(d *apiv1.Decision) *shared.Decision {
 			Header:                              Header(a.Header),
 			Memo:                                Memo(a.Memo),
 			SearchAttributes:                    SearchAttributes(a.SearchAttributes),
+			JitterStartSeconds:                  durationToSeconds(a.JitterStart),
 		}
 	case *apiv1.Decision_StartChildWorkflowExecutionDecisionAttributes:
 		decision.DecisionType = shared.DecisionTypeStartChildWorkflowExecution.Ptr()

diff --git a/internal/compatibility/thrift/history.go b/internal/compatibility/thrift/history.go
@@ -589,6 +589,7 @@ func StartChildWorkflowExecutionInitiatedEventAttributes(t *apiv1.StartChildWork
 		Memo:                                Memo(t.Memo),
 		SearchAttributes:                    SearchAttributes(t.SearchAttributes),
 		DelayStartSeconds:                   durationToSeconds(t.DelayStart),
+		JitterStartSeconds:                  durationToSeconds(t.JitterStart),
 	}
 }
 

diff --git a/internal/compatibility/thrift/request.go b/internal/compatibility/thrift/request.go
@@ -461,6 +461,7 @@ func StartWorkflowExecutionRequest(t *apiv1.StartWorkflowExecutionRequest) *shar
 		SearchAttributes:                    SearchAttributes(t.SearchAttributes),
 		Header:                              Header(t.Header),
 		DelayStartSeconds:                   durationToSeconds(t.DelayStart),
+		JitterStartSeconds:                  durationToSeconds(t.JitterStart),
 	}
 }
 

diff --git a/internal/internal_workflow_client.go b/internal/internal_workflow_client.go
@@ -196,6 +196,11 @@ func (wc *workflowClient) StartWorkflow(
 		return nil, errors.New("Invalid DelayStart option")
 	}
 
+	jitterStartSeconds := common.Int32Ceil(options.JitterStart.Seconds())
+	if jitterStartSeconds < 0 {
+		return nil, errors.New("Invalid JitterStart option")
+	}
+
 	// create a workflow start span and attach it to the context object.
 	// N.B. we need to finish this immediately as jaeger does not give us a way
 	// to recreate a span given a span context - which means we will run into
@@ -227,6 +232,7 @@ func (wc *workflowClient) StartWorkflow(
 		SearchAttributes:                    searchAttr,
 		Header:                              header,
 		DelayStartSeconds:                   common.Int32Ptr(delayStartSeconds),
+		JitterStartSeconds:                  common.Int32Ptr(jitterStartSeconds),
 	}
 
 	var response *s.StartWorkflowExecutionResponse
@@ -383,6 +389,11 @@ func (wc *workflowClient) SignalWithStartWorkflow(ctx context.Context, workflowI
 		return nil, errors.New("Invalid DelayStart option")
 	}
 
+	jitterStartSeconds := common.Int32Ceil(options.JitterStart.Seconds())
+	if jitterStartSeconds < 0 {
+		return nil, errors.New("Invalid JitterStart option")
+	}
+
 	// create a workflow start span and attach it to the context object. finish it immediately
 	ctx, span := createOpenTracingWorkflowSpan(ctx, wc.tracer, time.Now(), fmt.Sprintf("SignalWithStartWorkflow-%s", workflowType.Name), workflowID)
 	span.Finish()
@@ -409,6 +420,7 @@ func (wc *workflowClient) SignalWithStartWorkflow(ctx context.Context, workflowI
 		WorkflowIdReusePolicy:               options.WorkflowIDReusePolicy.toThriftPtr(),
 		Header:                              header,
 		DelayStartSeconds:                   common.Int32Ptr(delayStartSeconds),
+		JitterStartSeconds:                  common.Int32Ptr(jitterStartSeconds),
 	}
 
 	var response *s.StartWorkflowExecutionResponse