batcher

Documentation

Index

• Constants
• Variables
• func NewResult(dc *dynamicconfig.Collection, params activityDeps) fxResult
• type BatchOperationStats
• type BatchParams
• type CancelParams
• type DeleteParams
• type HeartBeatDetails
  • func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDetails, error)
• type ResetParams
• type SignalParams
• type TerminateParams
• type UpdateOptionsParams

Constants

const (
	// BatchWFTypeName is the workflow type
	BatchWFTypeName   = "temporal-sys-batch-workflow"
	NamespaceDivision = "TemporalBatcher"
)

const (
	// BatchOperationTypeMemo stores batch operation type in memo
	BatchOperationTypeMemo = "batch_operation_type"
	// BatchReasonMemo stores batch operation reason in memo
	BatchReasonMemo = "batch_operation_reason"
	// BatchOperationStatsMemo stores batch operation stats in memo
	BatchOperationStatsMemo = "batch_operation_stats"
	// BatchTypeTerminate is batch type for terminating workflows
	BatchTypeTerminate = "terminate"
	// BatchTypeCancel is the batch type for canceling workflows
	BatchTypeCancel = "cancel"
	// BatchTypeSignal is batch type for signaling workflows
	BatchTypeSignal = "signal"
	// BatchTypeDelete is batch type for deleting workflows
	BatchTypeDelete = "delete"
	// BatchTypeReset is batch type for resetting workflows
	BatchTypeReset = "reset"
	// BatchTypeUpdateOptions is batch type for updating the options of workflow executions
	BatchTypeUpdateOptions = "update_options"
)

Variables

var Module = fx.Options(
	fx.Provide(NewResult),
)

var (
	OpenBatchOperationQuery = fmt.Sprintf("%s = '%s' AND %s = %d",
		searchattribute.TemporalNamespaceDivision,
		NamespaceDivision,
		searchattribute.ExecutionStatus,
		int(enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING),
	)
)

Functions

func NewResult

func NewResult(
	dc *dynamicconfig.Collection,
	params activityDeps,
) fxResult

Types

type BatchOperationStats

type BatchOperationStats struct {
	NumSuccess int
	NumFailure int
}

type BatchParams

type 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
	// 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
	// contains filtered or unexported fields
}

BatchParams is the parameters for batch operation workflow

type CancelParams

type CancelParams struct {
}

CancelParams is the parameters for canceling workflow

type DeleteParams

type DeleteParams struct {
}

DeleteParams is the parameters for deleting workflow

type HeartBeatDetails

type HeartBeatDetails struct {
	PageToken   []byte
	CurrentPage int
	// This is just an estimation for visibility
	TotalEstimate int64
	// Number of workflows processed successfully
	SuccessCount int
	// Number of workflows that give up due to errors.
	ErrorCount int
}

HeartBeatDetails is the struct for heartbeat details

func BatchWorkflow

func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDetails, error)

BatchWorkflow is the workflow that runs a batch job of resetting workflows.

type ResetParams

type 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

	// Deprecated fields:
	ResetType        enumspb.ResetType
	ResetReapplyType enumspb.ResetReapplyType
	// contains filtered or unexported fields
}

ResetParams is the parameters for reseting workflow

type SignalParams

type SignalParams struct {
	SignalName string
	Input      *commonpb.Payloads
}

SignalParams is the parameters for signaling workflow

type TerminateParams

type TerminateParams struct {
}

TerminateParams is the parameters for terminating workflow

type UpdateOptionsParams

type UpdateOptionsParams struct {
	WorkflowExecutionOptions *workflowpb.WorkflowExecutionOptions
	UpdateMask               *fieldmaskpb.FieldMask
}

UpdateOptionsParams is the parameters for updating workflow execution options