mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
batch: register proto workflow for proto-qualified and non-proto-qualified workflow type. (#8511)
## What changed? Register the `BatchWorkflowProtobuf` as both `BatchWFTypeName` and `BatchWFTypeNameProtobuf`, picking up the legacy batch operation in addition to the new one. ## Why? Now in the version after 1.30 the frontend can be changed to call the new workflow function. ## How did you test it? - [ ] built - [ ] run locally and tested manually - [ ] covered by existing tests - [ ] added new unit test(s) - [ ] added new functional test(s) ## Potential risks None in this iteration.
This commit is contained in:
Sean Kane
@@ -9,12 +9,10 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/pborman/uuid"
|
||||
activitypb "go.temporal.io/api/activity/v1"
|
||||
batchpb "go.temporal.io/api/batch/v1"
|
||||
commonpb "go.temporal.io/api/common/v1"
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
"go.temporal.io/api/serviceerror"
|
||||
workflowpb "go.temporal.io/api/workflow/v1"
|
||||
"go.temporal.io/api/workflowservice/v1"
|
||||
"go.temporal.io/sdk/activity"
|
||||
sdkclient "go.temporal.io/sdk/client"
|
||||
@@ -25,10 +23,8 @@ import (
|
||||
"go.temporal.io/server/common/log/tag"
|
||||
"go.temporal.io/server/common/metrics"
|
||||
"go.temporal.io/server/common/namespace"
|
||||
"go.temporal.io/server/common/primitives"
|
||||
"go.temporal.io/server/common/sdk"
|
||||
"golang.org/x/time/rate"
|
||||
"google.golang.org/protobuf/types/known/durationpb"
|
||||
"google.golang.org/protobuf/types/known/fieldmaskpb"
|
||||
)
|
||||
|
||||
@@ -255,15 +251,6 @@ type activities struct {
|
||||
concurrency dynamicconfig.IntPropertyFnWithNamespaceFilter
|
||||
}
|
||||
|
||||
func (a *activities) checkNamespace(ns string) error {
|
||||
// Ignore system namespace for backward compatibility.
|
||||
// TODO: Remove the system namespace special handling after 1.19+
|
||||
if ns != a.namespace.String() && a.namespace.String() != primitives.SystemLocalNamespace {
|
||||
return errNamespaceMismatch
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (a *activities) checkNamespaceID(namespaceID string) error {
|
||||
if namespaceID != a.namespaceID.String() {
|
||||
return errNamespaceMismatch
|
||||
@@ -271,98 +258,6 @@ func (a *activities) checkNamespaceID(namespaceID string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// BatchActivity is an activity for processing batch operation.
|
||||
func (a *activities) BatchActivity(ctx context.Context, batchParams BatchParams) (HeartBeatDetails, error) {
|
||||
logger := a.getActivityLogger(ctx)
|
||||
hbd := HeartBeatDetails{}
|
||||
metricsHandler := a.MetricsHandler.WithTags(metrics.OperationTag(metrics.BatcherScope), metrics.NamespaceTag(batchParams.Namespace))
|
||||
|
||||
if err := a.checkNamespace(batchParams.Namespace); err != nil {
|
||||
metrics.BatcherOperationFailures.With(metricsHandler).Record(1)
|
||||
logger.Error("Failed to run batch operation due to namespace mismatch", tag.Error(err))
|
||||
return hbd, err
|
||||
}
|
||||
|
||||
// Deserialize batch reset options if set
|
||||
if b := batchParams.ResetParams.ResetOptions; b != nil {
|
||||
batchParams.ResetParams.resetOptions = &commonpb.ResetOptions{}
|
||||
if err := batchParams.ResetParams.resetOptions.Unmarshal(b); err != nil {
|
||||
logger.Error("Failed to deserialize batch reset options", tag.Error(err))
|
||||
return hbd, err
|
||||
}
|
||||
}
|
||||
|
||||
// Deserialize batch post reset operations if set
|
||||
if postOps := batchParams.ResetParams.PostResetOperations; postOps != nil {
|
||||
batchParams.ResetParams.postResetOperations = make([]*workflowpb.PostResetOperation, len(postOps))
|
||||
for i, serializedOp := range postOps {
|
||||
op := &workflowpb.PostResetOperation{}
|
||||
if err := op.Unmarshal(serializedOp); err != nil {
|
||||
logger.Error("Failed to deserialize batch post reset operation", tag.Error(err))
|
||||
return hbd, err
|
||||
}
|
||||
batchParams.ResetParams.postResetOperations[i] = op
|
||||
}
|
||||
}
|
||||
|
||||
sdkClient := a.ClientFactory.NewClient(sdkclient.Options{
|
||||
Namespace: batchParams.Namespace,
|
||||
DataConverter: sdk.PreferProtoDataConverter,
|
||||
})
|
||||
startOver := true
|
||||
if activity.HasHeartbeatDetails(ctx) {
|
||||
if err := activity.GetHeartbeatDetails(ctx, &hbd); err == nil {
|
||||
startOver = false
|
||||
} else {
|
||||
logger.Error("Failed to recover from last heartbeat, start over from beginning", tag.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
adjustedQuery := a.adjustQuery(batchParams.Query, batchParams.BatchType)
|
||||
|
||||
if startOver {
|
||||
estimateCount := int64(len(batchParams.Executions))
|
||||
if len(adjustedQuery) > 0 {
|
||||
resp, err := sdkClient.CountWorkflow(ctx, &workflowservice.CountWorkflowExecutionsRequest{
|
||||
Query: adjustedQuery,
|
||||
})
|
||||
if err != nil {
|
||||
metrics.BatcherOperationFailures.With(metricsHandler).Record(1)
|
||||
logger.Error("Failed to get estimate workflow count", tag.Error(err))
|
||||
return HeartBeatDetails{}, err
|
||||
}
|
||||
estimateCount = resp.GetCount()
|
||||
}
|
||||
hbd.TotalEstimate = estimateCount
|
||||
}
|
||||
|
||||
// Prepare configuration for shared processing function
|
||||
config := batchProcessorConfig{
|
||||
namespace: batchParams.Namespace,
|
||||
adjustedQuery: adjustedQuery,
|
||||
rps: a.getOperationRPS(batchParams.RPS),
|
||||
concurrency: a.getOperationConcurrency(batchParams.Concurrency),
|
||||
initialPageToken: hbd.PageToken,
|
||||
initialExecutions: batchParams.Executions,
|
||||
}
|
||||
|
||||
// Create a wrapper for the batch params specific worker processor
|
||||
workerProcessor := func(
|
||||
ctx context.Context,
|
||||
taskCh chan task,
|
||||
respCh chan taskResponse,
|
||||
rateLimiter *rate.Limiter,
|
||||
sdkClient sdkclient.Client,
|
||||
frontendClient workflowservice.WorkflowServiceClient,
|
||||
metricsHandler metrics.Handler,
|
||||
logger log.Logger,
|
||||
) {
|
||||
startTaskProcessor(ctx, batchParams, taskCh, respCh, rateLimiter, sdkClient, a.FrontendClient, metricsHandler, logger)
|
||||
}
|
||||
|
||||
return a.processWorkflowsWithProactiveFetching(ctx, config, workerProcessor, sdkClient, metricsHandler, logger, hbd)
|
||||
}
|
||||
|
||||
// BatchActivityWithProtobuf is an activity for processing batch operations using protobuf as the input type.
|
||||
// nolint:revive,cognitive-complexity
|
||||
func (a *activities) BatchActivityWithProtobuf(ctx context.Context, batchParams *batchspb.BatchOperationInput) (HeartBeatDetails, error) {
|
||||
@@ -444,20 +339,6 @@ func (a *activities) getActivityLogger(ctx context.Context) log.Logger {
|
||||
)
|
||||
}
|
||||
|
||||
func (a *activities) adjustQuery(query, batchType string) string {
|
||||
if len(query) == 0 {
|
||||
// don't add anything if query is empty
|
||||
return query
|
||||
}
|
||||
|
||||
switch batchType {
|
||||
case BatchTypeTerminate, BatchTypeSignal, BatchTypeCancel, BatchTypeUpdateOptions, BatchTypeUnpauseActivities:
|
||||
return fmt.Sprintf("(%s) AND (%s)", query, statusRunningQueryFilter)
|
||||
default:
|
||||
return query
|
||||
}
|
||||
}
|
||||
|
||||
func (a *activities) adjustQueryBatchTypeEnum(query string, batchType enumspb.BatchOperationType) string {
|
||||
if len(query) == 0 {
|
||||
// don't add anything if query is empty
|
||||
@@ -472,14 +353,6 @@ func (a *activities) adjustQueryBatchTypeEnum(query string, batchType enumspb.Ba
|
||||
}
|
||||
}
|
||||
|
||||
func (a *activities) getOperationRPS(requestedRPS float64) float64 {
|
||||
maxRPS := float64(a.rps(a.namespace.String()))
|
||||
if requestedRPS <= 0 || requestedRPS > maxRPS {
|
||||
return maxRPS
|
||||
}
|
||||
return requestedRPS
|
||||
}
|
||||
|
||||
func (a *activities) getOperationConcurrency(concurrency int) int {
|
||||
if concurrency <= 0 {
|
||||
return a.concurrency(a.namespace.String())
|
||||
@@ -487,216 +360,6 @@ func (a *activities) getOperationConcurrency(concurrency int) int {
|
||||
return concurrency
|
||||
}
|
||||
|
||||
func startTaskProcessor(
|
||||
ctx context.Context,
|
||||
batchParams BatchParams,
|
||||
taskCh chan task,
|
||||
respCh chan taskResponse,
|
||||
limiter *rate.Limiter,
|
||||
sdkClient sdkclient.Client,
|
||||
frontendClient workflowservice.WorkflowServiceClient,
|
||||
metricsHandler metrics.Handler,
|
||||
logger log.Logger,
|
||||
) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case task := <-taskCh:
|
||||
if isDone(ctx) {
|
||||
return
|
||||
}
|
||||
|
||||
if task.execution == nil {
|
||||
continue
|
||||
}
|
||||
var err error
|
||||
|
||||
switch batchParams.BatchType {
|
||||
case BatchTypeTerminate:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
return sdkClient.TerminateWorkflow(ctx, execution.WorkflowId, execution.RunId, batchParams.Reason)
|
||||
})
|
||||
case BatchTypeCancel:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
return sdkClient.CancelWorkflow(ctx, execution.WorkflowId, execution.RunId)
|
||||
})
|
||||
case BatchTypeSignal:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
_, err := frontendClient.SignalWorkflowExecution(ctx, &workflowservice.SignalWorkflowExecutionRequest{
|
||||
Namespace: batchParams.Namespace,
|
||||
WorkflowExecution: execution,
|
||||
SignalName: batchParams.SignalParams.SignalName,
|
||||
Input: batchParams.SignalParams.Input,
|
||||
})
|
||||
return err
|
||||
})
|
||||
case BatchTypeDelete:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
_, err := frontendClient.DeleteWorkflowExecution(ctx, &workflowservice.DeleteWorkflowExecutionRequest{
|
||||
Namespace: batchParams.Namespace,
|
||||
WorkflowExecution: execution,
|
||||
})
|
||||
return err
|
||||
})
|
||||
case BatchTypeReset:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
var eventID int64
|
||||
var err error
|
||||
var resetReapplyType enumspb.ResetReapplyType //nolint:staticcheck // SA1019 deprecated but still supported
|
||||
var resetReapplyExcludeTypes []enumspb.ResetReapplyExcludeType
|
||||
if batchParams.ResetParams.resetOptions != nil {
|
||||
// Using ResetOptions
|
||||
// Note: getResetEventIDByOptions may modify workflowExecution.RunId, if reset should be to a prior run
|
||||
eventID, err = getResetEventIDByOptions(ctx, batchParams.ResetParams.resetOptions, batchParams.Namespace, execution, frontendClient, logger)
|
||||
resetReapplyType = batchParams.ResetParams.resetOptions.ResetReapplyType //nolint:staticcheck // SA1019 deprecated but still supported
|
||||
resetReapplyExcludeTypes = batchParams.ResetParams.resetOptions.ResetReapplyExcludeTypes
|
||||
} else {
|
||||
// Old fields
|
||||
eventID, err = getResetEventIDByType(ctx, batchParams.ResetParams.ResetType, batchParams.Namespace, execution, frontendClient, logger)
|
||||
resetReapplyType = batchParams.ResetParams.ResetReapplyType
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = frontendClient.ResetWorkflowExecution(ctx, &workflowservice.ResetWorkflowExecutionRequest{
|
||||
Namespace: batchParams.Namespace,
|
||||
WorkflowExecution: execution,
|
||||
Reason: batchParams.Reason,
|
||||
RequestId: uuid.New(),
|
||||
WorkflowTaskFinishEventId: eventID,
|
||||
ResetReapplyType: resetReapplyType,
|
||||
ResetReapplyExcludeTypes: resetReapplyExcludeTypes,
|
||||
PostResetOperations: batchParams.ResetParams.postResetOperations,
|
||||
})
|
||||
return err
|
||||
})
|
||||
case BatchTypeUnpauseActivities:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
unpauseRequest := &workflowservice.UnpauseActivityRequest{
|
||||
Namespace: batchParams.Namespace,
|
||||
Execution: execution,
|
||||
Identity: batchParams.UnpauseActivitiesParams.Identity,
|
||||
Activity: &workflowservice.UnpauseActivityRequest_Type{Type: batchParams.UnpauseActivitiesParams.ActivityType},
|
||||
ResetAttempts: !batchParams.UnpauseActivitiesParams.ResetAttempts,
|
||||
ResetHeartbeat: batchParams.UnpauseActivitiesParams.ResetHeartbeat,
|
||||
Jitter: durationpb.New(batchParams.UnpauseActivitiesParams.Jitter),
|
||||
}
|
||||
|
||||
if batchParams.UnpauseActivitiesParams.MatchAll {
|
||||
unpauseRequest.Activity = &workflowservice.UnpauseActivityRequest_UnpauseAll{UnpauseAll: true}
|
||||
} else {
|
||||
unpauseRequest.Activity = &workflowservice.UnpauseActivityRequest_Type{Type: batchParams.UnpauseActivitiesParams.ActivityType}
|
||||
}
|
||||
_, err = frontendClient.UnpauseActivity(ctx, unpauseRequest)
|
||||
return err
|
||||
})
|
||||
|
||||
case BatchTypeUpdateOptions:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
var err error
|
||||
_, err = frontendClient.UpdateWorkflowExecutionOptions(ctx, &workflowservice.UpdateWorkflowExecutionOptionsRequest{
|
||||
Namespace: batchParams.Namespace,
|
||||
WorkflowExecution: execution,
|
||||
WorkflowExecutionOptions: batchParams.UpdateOptionsParams.WorkflowExecutionOptions,
|
||||
UpdateMask: &fieldmaskpb.FieldMask{Paths: batchParams.UpdateOptionsParams.UpdateMask.Paths},
|
||||
})
|
||||
return err
|
||||
})
|
||||
|
||||
case BatchTypeResetActivities:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
resetRequest := &workflowservice.ResetActivityRequest{
|
||||
Namespace: batchParams.Namespace,
|
||||
Execution: execution,
|
||||
Identity: batchParams.ResetActivitiesParams.Identity,
|
||||
Activity: &workflowservice.ResetActivityRequest_Type{Type: batchParams.ResetActivitiesParams.ActivityType},
|
||||
ResetHeartbeat: batchParams.ResetActivitiesParams.ResetHeartbeat,
|
||||
Jitter: durationpb.New(batchParams.ResetActivitiesParams.Jitter),
|
||||
KeepPaused: batchParams.ResetActivitiesParams.KeepPaused,
|
||||
RestoreOriginalOptions: batchParams.ResetActivitiesParams.RestoreOriginalOptions,
|
||||
}
|
||||
|
||||
if batchParams.ResetActivitiesParams.MatchAll {
|
||||
resetRequest.Activity = &workflowservice.ResetActivityRequest_MatchAll{MatchAll: true}
|
||||
} else {
|
||||
resetRequest.Activity = &workflowservice.ResetActivityRequest_Type{Type: batchParams.ResetActivitiesParams.ActivityType}
|
||||
}
|
||||
|
||||
_, err = frontendClient.ResetActivity(ctx, resetRequest)
|
||||
return err
|
||||
})
|
||||
case BatchTypeUpdateActivitiesOptions:
|
||||
err = processTask(ctx, limiter, task,
|
||||
func(execution *commonpb.WorkflowExecution) error {
|
||||
updateRequest := &workflowservice.UpdateActivityOptionsRequest{
|
||||
Namespace: batchParams.Namespace,
|
||||
Execution: execution,
|
||||
Activity: &workflowservice.UpdateActivityOptionsRequest_Type{Type: batchParams.UpdateActivitiesOptionsParams.ActivityType},
|
||||
UpdateMask: &fieldmaskpb.FieldMask{Paths: batchParams.UpdateActivitiesOptionsParams.UpdateMask.Paths},
|
||||
RestoreOriginal: batchParams.UpdateActivitiesOptionsParams.RestoreOriginal,
|
||||
Identity: batchParams.UpdateActivitiesOptionsParams.Identity,
|
||||
}
|
||||
|
||||
if ao := batchParams.UpdateActivitiesOptionsParams.ActivityOptions; ao != nil {
|
||||
updateRequest.ActivityOptions = &activitypb.ActivityOptions{
|
||||
ScheduleToStartTimeout: durationpb.New(ao.ScheduleToStartTimeout),
|
||||
ScheduleToCloseTimeout: durationpb.New(ao.ScheduleToCloseTime),
|
||||
StartToCloseTimeout: durationpb.New(ao.StartToCloseTimeout),
|
||||
HeartbeatTimeout: durationpb.New(ao.HeartbeatTimeout),
|
||||
}
|
||||
|
||||
if rp := ao.RetryPolicy; rp != nil {
|
||||
updateRequest.ActivityOptions.RetryPolicy = &commonpb.RetryPolicy{
|
||||
InitialInterval: durationpb.New(rp.InitialInterval),
|
||||
BackoffCoefficient: rp.BackoffCoefficient,
|
||||
MaximumInterval: durationpb.New(rp.MaximumInterval),
|
||||
MaximumAttempts: rp.MaximumAttempts,
|
||||
NonRetryableErrorTypes: rp.NonRetryableErrorTypes,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if batchParams.UpdateActivitiesOptionsParams.MatchAll {
|
||||
updateRequest.Activity = &workflowservice.UpdateActivityOptionsRequest_MatchAll{MatchAll: true}
|
||||
} else {
|
||||
updateRequest.Activity = &workflowservice.UpdateActivityOptionsRequest_Type{Type: batchParams.UpdateActivitiesOptionsParams.ActivityType}
|
||||
}
|
||||
|
||||
_, err = frontendClient.UpdateActivityOptions(ctx, updateRequest)
|
||||
return err
|
||||
})
|
||||
default:
|
||||
err = errors.New("unknown batch type: " + batchParams.BatchType)
|
||||
}
|
||||
if err != nil {
|
||||
metrics.BatcherProcessorFailures.With(metricsHandler).Record(1)
|
||||
logger.Error("Failed to process batch operation task", tag.Error(err))
|
||||
|
||||
_, ok := batchParams._nonRetryableErrors[err.Error()]
|
||||
if ok || task.attempts > batchParams.AttemptsOnRetryableError {
|
||||
respCh <- taskResponse{err: err, page: task.page}
|
||||
} else {
|
||||
// put back to the channel if less than attemptsOnError
|
||||
task.attempts++
|
||||
taskCh <- task
|
||||
}
|
||||
} else {
|
||||
metrics.BatcherProcessorSuccess.With(metricsHandler).Record(1)
|
||||
respCh <- taskResponse{err: nil, page: task.page}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// nolint:revive,cognitive-complexity
|
||||
func startTaskProcessorProtobuf(
|
||||
ctx context.Context,
|
||||
|
||||
@@ -289,66 +289,6 @@ func (s *activitiesSuite) TestGetResetPoint() {
|
||||
}
|
||||
}
|
||||
|
||||
func (s *activitiesSuite) TestAdjustQuery() {
|
||||
tests := []struct {
|
||||
name string
|
||||
query string
|
||||
expectedResult string
|
||||
batchType string
|
||||
}{
|
||||
{
|
||||
name: "Empty query",
|
||||
query: "",
|
||||
expectedResult: "",
|
||||
batchType: BatchTypeTerminate,
|
||||
},
|
||||
{
|
||||
name: "Acceptance",
|
||||
query: "A=B",
|
||||
expectedResult: fmt.Sprintf("(A=B) AND (%s)", statusRunningQueryFilter),
|
||||
batchType: BatchTypeTerminate,
|
||||
},
|
||||
{
|
||||
name: "Acceptance with parenthesis",
|
||||
query: "(A=B)",
|
||||
expectedResult: fmt.Sprintf("((A=B)) AND (%s)", statusRunningQueryFilter),
|
||||
batchType: BatchTypeTerminate,
|
||||
},
|
||||
{
|
||||
name: "Acceptance with multiple conditions",
|
||||
query: "(A=B) OR C=D",
|
||||
expectedResult: fmt.Sprintf("((A=B) OR C=D) AND (%s)", statusRunningQueryFilter),
|
||||
batchType: BatchTypeTerminate,
|
||||
},
|
||||
{
|
||||
name: "Contains status - 1",
|
||||
query: "ExecutionStatus=Completed",
|
||||
expectedResult: fmt.Sprintf("(ExecutionStatus=Completed) AND (%s)", statusRunningQueryFilter),
|
||||
batchType: BatchTypeTerminate,
|
||||
},
|
||||
{
|
||||
name: "Contains status - 2",
|
||||
query: "A=B OR ExecutionStatus='Completed'",
|
||||
expectedResult: fmt.Sprintf("(A=B OR ExecutionStatus='Completed') AND (%s)", statusRunningQueryFilter),
|
||||
batchType: BatchTypeTerminate,
|
||||
},
|
||||
{
|
||||
name: "Not supported batch type",
|
||||
query: "A=B",
|
||||
expectedResult: "A=B",
|
||||
batchType: "NotSupported",
|
||||
},
|
||||
}
|
||||
for _, testRun := range tests {
|
||||
s.Run(testRun.name, func() {
|
||||
a := activities{}
|
||||
batchParams := BatchParams{Query: testRun.query, BatchType: testRun.batchType}
|
||||
adjustedQuery := a.adjustQuery(batchParams.Query, batchParams.BatchType)
|
||||
s.Equal(testRun.expectedResult, adjustedQuery)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (s *activitiesSuite) TestAdjustQueryBatchTypeEnum() {
|
||||
tests := []struct {
|
||||
name string
|
||||
|
||||
@@ -67,7 +67,9 @@ func (s *workerComponent) DedicatedWorkerOptions(ns *namespace.Namespace) *worke
|
||||
}
|
||||
|
||||
func (s *workerComponent) Register(registry sdkworker.Registry, ns *namespace.Namespace, _ workercommon.RegistrationDetails) func() {
|
||||
registry.RegisterWorkflowWithOptions(BatchWorkflow, workflow.RegisterOptions{Name: BatchWFTypeName})
|
||||
// Register the batch workflow with both the proto-qualified and unqualified types.
|
||||
// TODO(spkane31): Remove the proto-qualified type and call the unqualified type from the frontend after the 1.30 release.
|
||||
registry.RegisterWorkflowWithOptions(BatchWorkflowProtobuf, workflow.RegisterOptions{Name: BatchWFTypeName})
|
||||
// Newer version of the batch workflow which was rewritten to accept a proto struct as input.
|
||||
registry.RegisterWorkflowWithOptions(BatchWorkflowProtobuf, workflow.RegisterOptions{Name: BatchWFTypeProtobufName})
|
||||
registry.RegisterActivity(s.activities(ns.Name(), ns.ID()))
|
||||
|
||||
@@ -9,7 +9,6 @@ import (
|
||||
commonpb "go.temporal.io/api/common/v1"
|
||||
enumspb "go.temporal.io/api/enums/v1"
|
||||
"go.temporal.io/api/serviceerror"
|
||||
workflowpb "go.temporal.io/api/workflow/v1"
|
||||
"go.temporal.io/api/workflowservice/v1"
|
||||
"go.temporal.io/sdk/temporal"
|
||||
"go.temporal.io/sdk/workflow"
|
||||
@@ -63,152 +62,6 @@ var (
|
||||
)
|
||||
|
||||
type (
|
||||
// TerminateParams is the parameters for terminating workflow
|
||||
TerminateParams struct {
|
||||
}
|
||||
|
||||
// CancelParams is the parameters for canceling workflow
|
||||
CancelParams struct {
|
||||
}
|
||||
|
||||
// SignalParams is the parameters for signaling workflow
|
||||
SignalParams struct {
|
||||
SignalName string
|
||||
Input *commonpb.Payloads
|
||||
}
|
||||
|
||||
// DeleteParams is the parameters for deleting workflow
|
||||
DeleteParams struct {
|
||||
}
|
||||
|
||||
// ResetParams is the parameters for reseting workflow
|
||||
ResetParams struct {
|
||||
// This is a serialized commonpb.ResetOptions. We can't include it with the
|
||||
// correct type because workflow/activity arguments are going to be serialized with the
|
||||
// json dataconverter, which doesn't support the "oneof" field in ResetOptions.
|
||||
ResetOptions []byte
|
||||
resetOptions *commonpb.ResetOptions // deserialized version
|
||||
|
||||
// This is a slice of serialized workflowpb.PostResetOperation.
|
||||
// For the same reason as ResetOptions, we manually serialize/deserialize it.
|
||||
PostResetOperations [][]byte
|
||||
postResetOperations []*workflowpb.PostResetOperation
|
||||
// Deprecated fields:
|
||||
ResetType enumspb.ResetType
|
||||
ResetReapplyType enumspb.ResetReapplyType //nolint:staticcheck // SA1019 deprecated but still supported
|
||||
}
|
||||
|
||||
// UpdateOptionsParams is the parameters for updating workflow execution options
|
||||
UpdateOptionsParams struct {
|
||||
WorkflowExecutionOptions *workflowpb.WorkflowExecutionOptions
|
||||
UpdateMask *FieldMask
|
||||
}
|
||||
|
||||
UnpauseActivitiesParams struct {
|
||||
Identity string
|
||||
ActivityType string
|
||||
MatchAll bool
|
||||
ResetAttempts bool
|
||||
ResetHeartbeat bool
|
||||
Jitter time.Duration
|
||||
}
|
||||
|
||||
UpdateActivitiesOptionsParams struct {
|
||||
Identity string
|
||||
ActivityType string
|
||||
MatchAll bool
|
||||
ActivityOptions *ActivityOptions
|
||||
UpdateMask *FieldMask
|
||||
RestoreOriginal bool
|
||||
}
|
||||
|
||||
ActivityOptions struct {
|
||||
TaskQueue *TaskQueue
|
||||
ScheduleToCloseTime time.Duration
|
||||
ScheduleToStartTimeout time.Duration
|
||||
StartToCloseTimeout time.Duration
|
||||
HeartbeatTimeout time.Duration
|
||||
RetryPolicy *RetryPolicy
|
||||
}
|
||||
|
||||
TaskQueue struct {
|
||||
Name string
|
||||
Kind int32
|
||||
}
|
||||
|
||||
RetryPolicy struct {
|
||||
InitialInterval time.Duration
|
||||
BackoffCoefficient float64
|
||||
MaximumInterval time.Duration
|
||||
MaximumAttempts int32
|
||||
NonRetryableErrorTypes []string
|
||||
}
|
||||
|
||||
ResetActivitiesParams struct {
|
||||
Identity string
|
||||
ActivityType string
|
||||
MatchAll bool
|
||||
ResetAttempts bool
|
||||
ResetHeartbeat bool
|
||||
KeepPaused bool
|
||||
Jitter time.Duration
|
||||
RestoreOriginalOptions bool
|
||||
}
|
||||
|
||||
FieldMask struct {
|
||||
Paths []string
|
||||
}
|
||||
|
||||
// BatchParams is the parameters for batch operation workflow
|
||||
BatchParams struct {
|
||||
// Target namespace to execute batch operation
|
||||
Namespace string
|
||||
// To get the target workflows for processing
|
||||
Query string
|
||||
// Target workflows for processing
|
||||
Executions []*commonpb.WorkflowExecution
|
||||
// Reason for the operation
|
||||
Reason string
|
||||
// Supporting: signal,cancel,terminate,delete,reset
|
||||
BatchType string
|
||||
|
||||
// Below are all optional
|
||||
// TerminateParams is params only for BatchTypeTerminate
|
||||
TerminateParams TerminateParams
|
||||
// CancelParams is params only for BatchTypeCancel
|
||||
CancelParams CancelParams
|
||||
// SignalParams is params only for BatchTypeSignal
|
||||
SignalParams SignalParams
|
||||
// DeleteParams is params only for BatchTypeDelete
|
||||
DeleteParams DeleteParams
|
||||
// ResetParams is params only for BatchTypeReset
|
||||
ResetParams ResetParams
|
||||
// UpdateOptionsParams is params only for BatchTypeUpdateOptions
|
||||
UpdateOptionsParams UpdateOptionsParams
|
||||
// UnpauseActivitiesParams is params only for BatchTypeUnpauseActivities
|
||||
UnpauseActivitiesParams UnpauseActivitiesParams
|
||||
// UpdateActivitiesOptionsParams is params only for BatchTypeUpdateActivitiesOptions
|
||||
UpdateActivitiesOptionsParams UpdateActivitiesOptionsParams
|
||||
// ResetActivitiesParams is params only for BatchTypeResetActivities
|
||||
ResetActivitiesParams ResetActivitiesParams
|
||||
|
||||
// RPS sets the requests-per-second limit for the batch.
|
||||
// The default (and max) is defined by `worker.BatcherRPS` in the dynamic config.
|
||||
RPS float64
|
||||
// Number of goroutines running in parallel to process
|
||||
// This is moving to dynamic config.
|
||||
// TODO: Remove it from BatchParams after 1.19+
|
||||
Concurrency int
|
||||
// Number of attempts for each workflow to process in case of retryable error before giving up
|
||||
AttemptsOnRetryableError int
|
||||
// timeout for activity heartbeat
|
||||
ActivityHeartBeatTimeout time.Duration
|
||||
// errors that will not retry which consumes AttemptsOnRetryableError. Default to empty
|
||||
NonRetryableErrors []string
|
||||
// internal conversion for NonRetryableErrors
|
||||
_nonRetryableErrors map[string]struct{}
|
||||
}
|
||||
|
||||
// HeartBeatDetails is the struct for heartbeat details
|
||||
HeartBeatDetails struct {
|
||||
PageToken []byte
|
||||
@@ -252,37 +105,13 @@ var (
|
||||
}
|
||||
)
|
||||
|
||||
// BatchWorkflow is the workflow that runs a batch job of resetting workflows.
|
||||
func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDetails, error) {
|
||||
batchParams = setDefaultParams(batchParams)
|
||||
err := validateParams(batchParams)
|
||||
if err != nil {
|
||||
return HeartBeatDetails{}, err
|
||||
}
|
||||
|
||||
batchActivityOptions.HeartbeatTimeout = batchParams.ActivityHeartBeatTimeout
|
||||
opt := workflow.WithActivityOptions(ctx, batchActivityOptions)
|
||||
var result HeartBeatDetails
|
||||
var ac *activities
|
||||
err = workflow.ExecuteActivity(opt, ac.BatchActivity, batchParams).Get(ctx, &result)
|
||||
if err != nil {
|
||||
return HeartBeatDetails{}, err
|
||||
}
|
||||
|
||||
err = attachBatchOperationStats(ctx, result)
|
||||
if err != nil {
|
||||
return HeartBeatDetails{}, err
|
||||
}
|
||||
return result, err
|
||||
}
|
||||
|
||||
// BatchWorkflowProtobuf is the workflow that runs a batch job of resetting workflows.
|
||||
func BatchWorkflowProtobuf(ctx workflow.Context, batchParams *batchspb.BatchOperationInput) (HeartBeatDetails, error) {
|
||||
if batchParams == nil {
|
||||
return HeartBeatDetails{}, errors.New("batchParams is nil")
|
||||
}
|
||||
|
||||
batchParams = setDefaultParamsProtobuf(batchParams)
|
||||
batchParams = setDefaultParams(batchParams)
|
||||
err := ValidateBatchOperation(batchParams.Request)
|
||||
if err != nil {
|
||||
return HeartBeatDetails{}, err
|
||||
@@ -321,54 +150,6 @@ func attachBatchOperationStats(ctx workflow.Context, result HeartBeatDetails) er
|
||||
return workflow.UpsertMemo(ctx, memo)
|
||||
}
|
||||
|
||||
func validateParams(params BatchParams) error {
|
||||
if params.BatchType == "" ||
|
||||
params.Reason == "" ||
|
||||
params.Namespace == "" ||
|
||||
(params.Query == "" && len(params.Executions) == 0) {
|
||||
return errors.New("must provide required parameters: BatchType/Reason/Namespace/Query/Executions")
|
||||
}
|
||||
|
||||
if len(params.Query) > 0 && len(params.Executions) > 0 {
|
||||
return errors.New("batch query and executions are mutually exclusive")
|
||||
}
|
||||
|
||||
switch params.BatchType {
|
||||
case BatchTypeSignal:
|
||||
if params.SignalParams.SignalName == "" {
|
||||
return errors.New("must provide signal name")
|
||||
}
|
||||
return nil
|
||||
case BatchTypeUpdateOptions:
|
||||
if params.UpdateOptionsParams.WorkflowExecutionOptions == nil {
|
||||
return errors.New("must provide UpdateOptions")
|
||||
}
|
||||
if params.UpdateOptionsParams.UpdateMask == nil {
|
||||
return errors.New("must provide UpdateMask")
|
||||
}
|
||||
return worker_versioning.ValidateVersioningOverride(params.UpdateOptionsParams.WorkflowExecutionOptions.VersioningOverride)
|
||||
case BatchTypeCancel, BatchTypeTerminate, BatchTypeDelete, BatchTypeReset:
|
||||
return nil
|
||||
case BatchTypeUnpauseActivities:
|
||||
if params.UnpauseActivitiesParams.ActivityType == "" && !params.UnpauseActivitiesParams.MatchAll {
|
||||
return errors.New("must provide ActivityType or MatchAll")
|
||||
}
|
||||
return nil
|
||||
case BatchTypeResetActivities:
|
||||
if params.ResetActivitiesParams.ActivityType == "" && !params.ResetActivitiesParams.MatchAll {
|
||||
return errors.New("must provide ActivityType or MatchAll")
|
||||
}
|
||||
return nil
|
||||
case BatchTypeUpdateActivitiesOptions:
|
||||
if params.UpdateActivitiesOptionsParams.ActivityType == "" && !params.UpdateActivitiesOptionsParams.MatchAll {
|
||||
return errors.New("must provide ActivityType or MatchAll")
|
||||
}
|
||||
return nil
|
||||
default:
|
||||
return fmt.Errorf("not supported batch type: %v", params.BatchType)
|
||||
}
|
||||
}
|
||||
|
||||
// nolint:revive,cognitive-complexity
|
||||
func ValidateBatchOperation(params *workflowservice.StartBatchOperationRequest) error {
|
||||
if params.GetOperation() == nil ||
|
||||
@@ -495,23 +276,7 @@ func ValidateBatchOperation(params *workflowservice.StartBatchOperationRequest)
|
||||
return nil
|
||||
}
|
||||
|
||||
func setDefaultParams(params BatchParams) BatchParams {
|
||||
if params.AttemptsOnRetryableError <= 1 {
|
||||
params.AttemptsOnRetryableError = defaultAttemptsOnRetryableError
|
||||
}
|
||||
if params.ActivityHeartBeatTimeout <= 0 {
|
||||
params.ActivityHeartBeatTimeout = defaultActivityHeartBeatTimeout
|
||||
}
|
||||
if len(params.NonRetryableErrors) > 0 {
|
||||
params._nonRetryableErrors = make(map[string]struct{}, len(params.NonRetryableErrors))
|
||||
for _, estr := range params.NonRetryableErrors {
|
||||
params._nonRetryableErrors[estr] = struct{}{}
|
||||
}
|
||||
}
|
||||
return params
|
||||
}
|
||||
|
||||
func setDefaultParamsProtobuf(params *batchspb.BatchOperationInput) *batchspb.BatchOperationInput {
|
||||
func setDefaultParams(params *batchspb.BatchOperationInput) *batchspb.BatchOperationInput {
|
||||
if params.GetAttemptsOnRetryableError() <= 1 {
|
||||
params.AttemptsOnRetryableError = defaultAttemptsOnRetryableError
|
||||
}
|
||||
|
||||
@@ -29,7 +29,6 @@ func TestBatcherSuite(t *testing.T) {
|
||||
func (s *batcherSuite) SetupTest() {
|
||||
s.controller = gomock.NewController(s.T())
|
||||
s.env = s.WorkflowTestSuite.NewTestWorkflowEnvironment()
|
||||
s.env.RegisterWorkflow(BatchWorkflow)
|
||||
s.env.RegisterWorkflow(BatchWorkflowProtobuf)
|
||||
}
|
||||
|
||||
@@ -38,13 +37,6 @@ func (s *batcherSuite) TearDownTest() {
|
||||
s.env.AssertExpectations(s.T())
|
||||
}
|
||||
|
||||
func (s *batcherSuite) TestBatchWorkflow_MissingParams() {
|
||||
s.env.ExecuteWorkflow(BatchWorkflow, BatchParams{})
|
||||
err := s.env.GetWorkflowError()
|
||||
s.Require().Error(err)
|
||||
s.Contains(err.Error(), "must provide required parameters")
|
||||
}
|
||||
|
||||
func (s *batcherSuite) TestBatchWorkflow_MissingParams_Protobuf() {
|
||||
s.env.ExecuteWorkflow(BatchWorkflowProtobuf, &batchspb.BatchOperationInput{})
|
||||
err := s.env.GetWorkflowError()
|
||||
@@ -52,32 +44,6 @@ func (s *batcherSuite) TestBatchWorkflow_MissingParams_Protobuf() {
|
||||
s.Contains(err.Error(), "must provide required parameters")
|
||||
}
|
||||
|
||||
func (s *batcherSuite) TestBatchWorkflow_ValidParams_Query() {
|
||||
var ac *activities
|
||||
s.env.OnActivity(ac.BatchActivity, mock.Anything, mock.Anything).Return(HeartBeatDetails{
|
||||
SuccessCount: 42,
|
||||
ErrorCount: 27,
|
||||
}, nil)
|
||||
s.env.OnUpsertMemo(mock.Anything).Run(func(args mock.Arguments) {
|
||||
memo, ok := args.Get(0).(map[string]interface{})
|
||||
s.Require().True(ok)
|
||||
s.Equal(map[string]interface{}{
|
||||
"batch_operation_stats": BatchOperationStats{
|
||||
NumSuccess: 42,
|
||||
NumFailure: 27,
|
||||
},
|
||||
}, memo)
|
||||
}).Once()
|
||||
s.env.ExecuteWorkflow(BatchWorkflow, BatchParams{
|
||||
BatchType: BatchTypeTerminate,
|
||||
Reason: "test-reason",
|
||||
Namespace: "test-namespace",
|
||||
Query: "test-query",
|
||||
})
|
||||
err := s.env.GetWorkflowError()
|
||||
s.Require().NoError(err)
|
||||
}
|
||||
|
||||
func (s *batcherSuite) TestBatchWorkflow_ValidParams_Query_Protobuf() {
|
||||
var ac *activities
|
||||
s.env.OnActivity(ac.BatchActivityWithProtobuf, mock.Anything, mock.Anything).Return(HeartBeatDetails{
|
||||
@@ -110,37 +76,6 @@ func (s *batcherSuite) TestBatchWorkflow_ValidParams_Query_Protobuf() {
|
||||
s.Require().NoError(err)
|
||||
}
|
||||
|
||||
func (s *batcherSuite) TestBatchWorkflow_ValidParams_Executions() {
|
||||
var ac *activities
|
||||
s.env.OnActivity(ac.BatchActivity, mock.Anything, mock.Anything).Return(HeartBeatDetails{
|
||||
SuccessCount: 42,
|
||||
ErrorCount: 27,
|
||||
}, nil)
|
||||
s.env.OnUpsertMemo(mock.Anything).Run(func(args mock.Arguments) {
|
||||
memo, ok := args.Get(0).(map[string]interface{})
|
||||
s.Require().True(ok)
|
||||
s.Equal(map[string]interface{}{
|
||||
"batch_operation_stats": BatchOperationStats{
|
||||
NumSuccess: 42,
|
||||
NumFailure: 27,
|
||||
},
|
||||
}, memo)
|
||||
}).Once()
|
||||
s.env.ExecuteWorkflow(BatchWorkflow, BatchParams{
|
||||
BatchType: BatchTypeTerminate,
|
||||
Reason: "test-reason",
|
||||
Namespace: "test-namespace",
|
||||
Executions: []*commonpb.WorkflowExecution{
|
||||
{
|
||||
WorkflowId: uuid.New(),
|
||||
RunId: uuid.New(),
|
||||
},
|
||||
},
|
||||
})
|
||||
err := s.env.GetWorkflowError()
|
||||
s.Require().NoError(err)
|
||||
}
|
||||
|
||||
func (s *batcherSuite) TestBatchWorkflow_ValidParams_Executions_Protobuf() {
|
||||
var ac *activities
|
||||
s.env.OnActivity(ac.BatchActivityWithProtobuf, mock.Anything, mock.Anything).Return(HeartBeatDetails{
|
||||
|
||||
Reference in New Issue
Block a user