batcher

Documentation

Index

• Constants
• Variables
• type BatchParams
• type Batcher
  • func New(cfg *Config, sdkSystemClient sdkclient.Client, metricsClient metrics.Client, ...) *Batcher
  • func (s *Batcher) Start() error
• type CancelParams
• type Config
• type HeartBeatDetails
  • func BatchActivity(ctx context.Context, batchParams BatchParams) (HeartBeatDetails, error)
  • func BatchWorkflow(ctx workflow.Context, batchParams BatchParams) (HeartBeatDetails, error)
• type SignalParams
• type TerminateParams

Constants

const (

	// BatcherTaskQueueName is the taskqueue name
	BatcherTaskQueueName = "temporal-sys-batcher-taskqueue"
	// BatchWFTypeName is the workflow type
	BatchWFTypeName = "temporal-sys-batch-workflow"

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

	// DefaultRPS is the default RPS
	DefaultRPS = 50
	// DefaultConcurrency is the default concurrency
	DefaultConcurrency = 5
	// DefaultAttemptsOnRetryableError is the default value for AttemptsOnRetryableError
	DefaultAttemptsOnRetryableError = 50
	// DefaultActivityHeartBeatTimeout is the default value for ActivityHeartBeatTimeout
	DefaultActivityHeartBeatTimeout = time.Second * 10
)

const (
	// 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"
)

Variables

var AllBatchTypes = []string{BatchTypeTerminate, BatchTypeCancel, BatchTypeSignal}

AllBatchTypes is the batch types we supported

Types

type BatchParams

type BatchParams struct {
	// Target namespace to execute batch operation
	Namespace string
	// To get the target workflows for processing
	Query string
	// Reason for the operation
	Reason string
	// Supporting: signal,cancel,terminate
	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
	// RPS of processing. Default to DefaultRPS
	// TODO we will implement smarter way than this static rate limiter: https://go.temporal.io/server/issues/2138
	RPS int
	// Number of goroutines running in parallel to process
	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(
	cfg *Config,
	sdkSystemClient sdkclient.Client,
	metricsClient metrics.Client,
	logger log.Logger,
	sdkClientFactory sdk.ClientFactory,
) *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 {
	// this indicates whether to cancel children workflow. Default to true.
	// TODO https://github.com/uber/cadence/issues/2159
	// Ideally default should be childPolicy of the workflow. But it's currently totally broken.
	CancelChildren *bool
}

CancelParams is the parameters for canceling workflow

type Config

type Config struct {
	MaxConcurrentActivityExecutionSize     dynamicconfig.IntPropertyFn
	MaxConcurrentWorkflowTaskExecutionSize dynamicconfig.IntPropertyFn
	MaxConcurrentActivityTaskPollers       dynamicconfig.IntPropertyFn
	MaxConcurrentWorkflowTaskPollers       dynamicconfig.IntPropertyFn
}

Config defines the configuration for batcher

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 BatchActivity

func BatchActivity(ctx context.Context, batchParams BatchParams) (HeartBeatDetails, error)

BatchActivity is activity for processing batch operation

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 {
	// this indicates whether to terminate children workflow. Default to true.
	// TODO https://github.com/uber/cadence/issues/2159
	// Ideally default should be childPolicy of the workflow. But it's currently totally broken.
	TerminateChildren *bool
}

TerminateParams is the parameters for terminating workflow