sysworkflow

Documentation

Index

• Constants
• func ArchivalActivity(ctx context.Context, request ArchiveRequest) error
• func BackfillActivity(_ context.Context, _ BackfillRequest) error
• func ConvertHeaderToTags(header *HistoryBlobHeader) (map[string]string, error)
• func NewHistoryBlobKey(domainID, workflowID, runID, pageToken, failoverVersion string) (blob.Key, error)
• func SystemWorkflow(ctx workflow.Context) error
• type ArchivalClient
  • func NewArchivalClient(publicClient public.Client, numSWFn dynamicconfig.IntPropertyFn) ArchivalClient
• type ArchiveRequest
• type BackfillRequest
• type Config
• type HistoryBlob
• type HistoryBlobHeader
• type RequestType
• type SysWorker
  • func NewSysWorker(publicClient public.Client, scope tally.Scope, ...) *SysWorker
  • func (w *SysWorker) Start() error
  • func (w *SysWorker) Stop()

Constants

const (
	// Domain is the cadence system workflows domain
	Domain = "cadence-system"
	// DecisionTaskList is the task list that all system workflows share
	DecisionTaskList = "cadsys-decision-tl"
	// SignalName is the name of the cadence signal that system tasks are sent on
	SignalName = "cadsys-signal-sig"
	// WorkflowIDPrefix is the prefix of all system workflow ids
	WorkflowIDPrefix = "cadsys-wf"
	// SignalsUntilContinueAsNew is the number of signals system workflow must receive before continuing as new
	SignalsUntilContinueAsNew = 1000
	// SystemWorkflowScope scope for all metrics emitted by system workflow
	SystemWorkflowScope = "system-workflow"
	// SystemWorkflowIDTag tag for system workflowID
	SystemWorkflowIDTag = "system-workflow-id"
	// HandledSignalCount counter of number of signals processed by system workflow
	HandledSignalCount = "handled-signal"
	// UnknownSignalTypeErr counter of number of unknown signals received by system workflow
	UnknownSignalTypeErr = "unknown-signal-err"
	// ArchivalFailureErr counter of number of archival activity failures
	ArchivalFailureErr = "archival-failure"
	// BackfillFailureErr counter of number of backfill activity failures
	BackfillFailureErr = "backfill-failure"
	// ChannelClosedUnexpectedlyError counter of number of unexpected channel closes in system workflow
	ChannelClosedUnexpectedlyError = "channel-closed-unexpectedly-err"
	// ArchivalActivityFnName name of archival activity function
	ArchivalActivityFnName = "ArchivalActivity"
	// BackfillActivityFnName name of backfill activity function
	BackfillActivityFnName = "BackfillActivity"
	// SystemWorkflowFnName name of system workflow function
	SystemWorkflowFnName = "SystemWorkflow"
	// WorkflowStartToCloseTimeout is the time for the workflow to finish
	WorkflowStartToCloseTimeout = time.Hour * 24 * 30
	// DecisionTaskStartToCloseTimeout is the time for decision to finish
	DecisionTaskStartToCloseTimeout = time.Minute
	// DomainIDTag tag which identifies the domainID of an archived history
	DomainIDTag = "domainID"
	// WorkflowIDTag tag which identifies the workflowID of an archived history
	WorkflowIDTag = "workflowID"
	//RunIDTag tag which identifies the runID of an archived history
	RunIDTag = "runID"
	// BucketNameTag tag which identifies the bucket name
	BucketNameTag = "bucket-name"
)

const (
	// HistoryBlobKeyExt is the blob key extension on all history blobs
	HistoryBlobKeyExt = "history"
)

Functions

func ArchivalActivity

func ArchivalActivity(ctx context.Context, request ArchiveRequest) error

ArchivalActivity is the archival activity code

func BackfillActivity

func BackfillActivity(_ context.Context, _ BackfillRequest) error

BackfillActivity is the backfill activity code

func ConvertHeaderToTags

func ConvertHeaderToTags(header *HistoryBlobHeader) (map[string]string, error)

ConvertHeaderToTags converts header into metadata tags for blob

func NewHistoryBlobKey

func NewHistoryBlobKey(domainID, workflowID, runID, pageToken, failoverVersion string) (blob.Key, error)

NewHistoryBlobKey returns a key for history blob

func SystemWorkflow

func SystemWorkflow(ctx workflow.Context) error

SystemWorkflow is the system workflow code

Types

type ArchivalClient

type ArchivalClient interface {
	Archive(*ArchiveRequest) error
	Backfill(*BackfillRequest) error
}

ArchivalClient is used to archive workflow histories

func NewArchivalClient

func NewArchivalClient(publicClient public.Client, numSWFn dynamicconfig.IntPropertyFn) ArchivalClient

NewArchivalClient creates a new ArchivalClient

type ArchiveRequest

type ArchiveRequest struct {
	DomainID         string
	WorkflowID       string
	RunID            string
	LastWriteVersion int64
}

ArchiveRequest is request to Archive

type BackfillRequest

type BackfillRequest struct {
}

BackfillRequest is request to Backfill

type Config

type Config struct{}

Config for SysWorker

type HistoryBlob

type HistoryBlob struct {
	Header *HistoryBlobHeader `json:"header"`
	Body   *shared.History    `json:"body"`
}

HistoryBlob is the serializable data that forms the body of a blob

type HistoryBlobHeader

type HistoryBlobHeader struct {
	DomainName           *string `json:"domain_name,omitempty"`
	DomainID             *string `json:"domain_id,omitempty"`
	WorkflowID           *string `json:"workflow_id,omitempty"`
	RunID                *string `json:"run_id,omitempty"`
	CurrentPageToken     *string `json:"current_page_token,omitempty"`
	NextPageToken        *string `json:"next_page_token,omitempty"`
	FirstFailoverVersion *string `json:"first_failover_version,omitempty"`
	LastFailoverVersion  *string `json:"last_failover_version,omitempty"`
	FirstEventID         *int64  `json:"first_event_id,omitempty"`
	LastEventID          *int64  `json:"last_event_id,omitempty"`
	UploadDateTime       *string `json:"upload_date_time,omitempty"`
	UploadCluster        *string `json:"upload_cluster,omitempty"`
	EventCount           *int64  `json:"event_count,omitempty"`
}

HistoryBlobHeader is the header attached to all history blobs

type RequestType

type RequestType int

RequestType is the type for signals that can be sent to system workflows

type SysWorker

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

SysWorker is the cadence client worker responsible for running system workflows

func NewSysWorker

func NewSysWorker(publicClient public.Client, scope tally.Scope, blobstoreClient blobstore.Client) *SysWorker

NewSysWorker returns a new SysWorker

func (*SysWorker) Start

func (w *SysWorker) Start() error

Start the SysWorker

func (*SysWorker) Stop

func (w *SysWorker) Stop()

Stop the SysWorker