batcher

Documentation

Index

• Constants
• Variables
• func NewResult(dc *dynamicconfig.Collection, params activityDeps) fxResult
• type BatchOperationStats
• type BatchParams
• type Batcher
  • func New(metricsHandler metrics.Handler, logger log.Logger, ...) *Batcher
  • func (s *Batcher) Start() error
• type CancelParams
• type DeleteParams
• type HeartBeatDetails
  • func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDetails, error)
• type SignalParams
• type TerminateParams

Constants

const (
	// BatchWFTypeName is the workflow type
	BatchWFTypeName   = "temporal-sys-batch-workflow"
	NamespaceDivision = "TemporalBatcher"
	// DefaultRPS is the default RPS
	DefaultRPS = 50
	// DefaultConcurrency is the default concurrency
	DefaultConcurrency = 5
)

const (
	// InfiniteDuration is a long duration(20 yrs) we used for infinite workflow running
	InfiniteDuration = 20 * 365 * 24 * time.Hour

	// DefaultAttemptsOnRetryableError is the default value for AttemptsOnRetryableError
	DefaultAttemptsOnRetryableError = 50
	// DefaultActivityHeartBeatTimeout is the default value for ActivityHeartBeatTimeout
	DefaultActivityHeartBeatTimeout = time.Second * 10
)

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"
)

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
	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
	// RPS of processing. Default to DefaultRPS
	// This is moving to dynamic config.
	// TODO: Remove it from BatchParams after 1.19+
	RPS int
	// 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 Batcher

type Batcher struct {
	// contains filtered or unexported fields
}

Batcher is the background sub-system that execute workflow for batch operations It is also the context object that get's passed around within the scanner workflows / activities

func New

func New(
	metricsHandler metrics.Handler,
	logger log.Logger,
	sdkClientFactory sdk.ClientFactory,
	rps dynamicconfig.IntPropertyFnWithNamespaceFilter,
	concurrency dynamicconfig.IntPropertyFnWithNamespaceFilter,
) *Batcher

New returns a new instance of batcher daemon Batcher

func (*Batcher) Start

func (s *Batcher) Start() error

Start starts the scanner

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 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