Add JitterStart support to client (#1178)

ZackLK · web-flow · commit e66e2d4da8de · 2022-07-15T19:45:01.000-07:00
* Add JitterStart support to client

* Dummy commit to see if it unblocks buildkite...
diff --git a/.gen/go/shared/shared.go b/.gen/go/shared/shared.go
diff --git a/README.md b/README.md
@@ -28,4 +28,4 @@ We'd love your help in making the Cadence Go client great. Please review our [co
 
 ## License
 MIT License, please see [LICENSE](LICENSE) for details.
-
+ 
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
@@ -1 +1 @@
-Subproject commit f2f5b7fc2bbd9849ac49f6338f8ed81d9c4c577e
+Subproject commit fda89e95df1e90544644a9dfa47a6d0867b532a7
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

Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,7 @@ func Decision(d shared.Decision) apiv1.Decision {`
`140`	`140`	`Header: Header(attr.Header),`
`141`	`141`	`Memo: Memo(attr.Memo),`
`142`	`142`	`SearchAttributes: SearchAttributes(attr.SearchAttributes),`
	`143`	`+ JitterStart: secondsToDuration(attr.JitterStartSeconds),`
`143`	`144`	`},`
`144`	`145`	`}`
`145`	`146`	`case shared.DecisionTypeStartChildWorkflowExecution:`
Original file line number	Diff line number	Diff line change
`@@ -540,6 +540,7 @@ func StartChildWorkflowExecutionInitiatedEventAttributes(t *shared.StartChildWor`
`540`	`540`	`Memo: Memo(t.Memo),`
`541`	`541`	`SearchAttributes: SearchAttributes(t.SearchAttributes),`
`542`	`542`	`DelayStart: secondsToDuration(t.DelayStartSeconds),`
	`543`	`+ JitterStart: secondsToDuration(t.JitterStartSeconds),`
`543`	`544`	`}`
`544`	`545`	`}`
`545`	`546`
Original file line number	Diff line number	Diff line change
`@@ -408,6 +408,7 @@ func SignalWithStartWorkflowExecutionRequest(t *shared.SignalWithStartWorkflowEx`
`408`	`408`	`SearchAttributes: SearchAttributes(t.SearchAttributes),`
`409`	`409`	`Header: Header(t.Header),`
`410`	`410`	`DelayStart: secondsToDuration(t.DelayStartSeconds),`
	`411`	`+ JitterStart: secondsToDuration(t.JitterStartSeconds),`
`411`	`412`	`},`
`412`	`413`	`SignalName: t.GetSignalName(),`
`413`	`414`	`SignalInput: Payload(t.SignalInput),`
`@@ -451,6 +452,7 @@ func StartWorkflowExecutionRequest(t shared.StartWorkflowExecutionRequest) api`
`451`	`452`	`SearchAttributes: SearchAttributes(t.SearchAttributes),`
`452`	`453`	`Header: Header(t.Header),`
`453`	`454`	`DelayStart: secondsToDuration(t.DelayStartSeconds),`
	`455`	`+ JitterStart: secondsToDuration(t.JitterStartSeconds),`
`454`	`456`	`}`
`455`	`457`	`}`
`456`	`458`
Original file line number	Diff line number	Diff line change
`@@ -105,6 +105,7 @@ var (`
`105`	`105`	`Header: &Header,`
`106`	`106`	`Memo: &Memo,`
`107`	`107`	`SearchAttributes: &SearchAttributes,`
	`108`	`+ JitterStart: Duration4,`
`108`	`109`	`}`
`109`	`110`	`FailWorkflowExecutionDecisionAttributes = apiv1.FailWorkflowExecutionDecisionAttributes{`
`110`	`111`	`Failure: &Failure,`