executions

Documentation

Index

• Constants
• Variables
• func FixShardActivity(activityCtx context.Context, params FixShardActivityParams) ([]shard.FixReport, error)
• func FixerWorkflow(ctx workflow.Context, params FixerWorkflowParams) error
• func ScanShardActivity(activityCtx context.Context, params ScanShardActivityParams) ([]shard.ScanReport, error)
• func ScannerEmitMetricsActivity(activityCtx context.Context, params ScannerEmitMetricsActivityParams) error
• func ScannerWorkflow(ctx workflow.Context, params ScannerWorkflowParams) error
• type AggregateFixReportResult
• type AggregateScanReportResult
• type ContextKey
• type CorruptedKeysEntry
• type DynamicConfigInvariantCollections
• type FixReportError
• type FixShardActivityParams
• type FixShardHeartbeatDetails
• type FixerContext
• type FixerCorruptedKeysActivityParams
• type FixerCorruptedKeysActivityResult
  • func FixerCorruptedKeysActivity(activityCtx context.Context, params FixerCorruptedKeysActivityParams) (*FixerCorruptedKeysActivityResult, error)
• type FixerWorkflowConfigOverwrites
• type FixerWorkflowParams
• type InvariantCollections
• type PaginatedShardQueryRequest
• type ResolvedFixerWorkflowConfig
• type ResolvedScannerWorkflowConfig
  • func ScannerConfigActivity(activityCtx context.Context, params ScannerConfigActivityParams) (ResolvedScannerWorkflowConfig, error)
• type ScanReportError
• type ScanShardActivityParams
• type ScanShardHeartbeatDetails
• type ScannerConfigActivityParams
• type ScannerContext
• type ScannerEmitMetricsActivityParams
• type ScannerWorkflowConfigOverwrites
• type ScannerWorkflowDynamicConfig
• type ScannerWorkflowParams
• type ShardCorruptKeysQueryResult
• type ShardCorruptKeysResult
• type ShardDistributionStats
• type ShardQueryPaginationToken
• type ShardRange
• type ShardSizeQueryRequest
• type ShardSizeQueryResult
• type ShardSizeTuple
• type ShardStatus
• type ShardStatusQueryResult
• type ShardStatusResult
• type ShardStatusSummaryResult
• type Shards
  • func (s Shards) Flatten() ([]int, int, int)
  • func (s Shards) Validate() error

Constants

const (
	// ScannerConfigActivityName is the activity name ScannerConfigActivity
	ScannerConfigActivityName = "cadence-sys-executions-scanner-config-activity"
	// ScannerScanShardActivityName is the activity name for ScanShardActivity
	ScannerScanShardActivityName = "cadence-sys-executions-scanner-scan-shard-activity"
	// ScannerEmitMetricsActivityName is the activity name for ScannerEmitMetricsActivity
	ScannerEmitMetricsActivityName = "cadence-sys-executions-scanner-emit-metrics-activity"
	// FixerCorruptedKeysActivityName is the activity name for FixerCorruptedKeysActivity
	FixerCorruptedKeysActivityName = "cadence-sys-executions-fixer-corrupted-keys-activity"
	// FixerFixShardActivityName is the activity name for FixShardActivity
	FixerFixShardActivityName = "cadence-sys-executions-fixer-fix-shard-activity"

	// ErrScanWorkflowNotClosed indicates fix was attempted on scan workflow which was not finished
	ErrScanWorkflowNotClosed = "scan workflow is not closed, only can run fix on output of finished scan workflow"
	// ErrSerialization indicates a serialization or deserialization error occurred
	ErrSerialization = "encountered serialization error"
)

const (
	// ConcreteScannerContextKey is the key used to access ScannerContext in activities for concrete executions
	ConcreteScannerContextKey = ContextKey(0)
	// ConcreteFixerContextKey is the key used to access FixerContext in activities for concrete executions
	ConcreteFixerContextKey = ContextKey(1)
	// CurrentScannerContextKey is the key used to access ScannerContext in activities for current executions
	CurrentScannerContextKey = ContextKey(2)
	// CurrentFixerContextKey is the key used to access FixerContext in activities for current executions
	CurrentFixerContextKey = ContextKey(3)

	// ShardReportQuery is the query name for the query used to get a single shard's report
	ShardReportQuery = "shard_report"
	// ShardStatusQuery is the query name for the query used to get the status of all shards
	ShardStatusQuery = "shard_status"
	// ShardStatusSummaryQuery is the query name for the query used to get the shard status -> counts map
	ShardStatusSummaryQuery = "shard_status_summary"
	// AggregateReportQuery is the query name for the query used to get the aggregate result of all finished shards
	AggregateReportQuery = "aggregate_report"
	// ShardCorruptKeysQuery is the query name for the query used to get all completed shards with at least one corruption
	ShardCorruptKeysQuery = "shard_corrupt_keys"
	// ShardSizeQuery is the query name for the query used to get the number of executions per shard in sorted order
	ShardSizeQuery = "shard_size"

	// ShardStatusRunning indicates the shard has not completed yet
	ShardStatusRunning ShardStatus = "running"
	// ShardStatusSuccess indicates the scan on the shard ran successfully
	ShardStatusSuccess ShardStatus = "success"
	// ShardStatusControlFlowFailure indicates the scan on the shard failed
	ShardStatusControlFlowFailure ShardStatus = "control_flow_failure"
)

Variables

var ScanTypeFixerContextKeyMap = map[shard.ScanType]interface{}{
	shard.ConcreteExecutionType: ConcreteFixerContextKey,
	shard.CurrentExecutionType:  CurrentFixerContextKey,
}

ScanTypeFixerContextKeyMap maps execution type to the context key used by fixer

var ScanTypeScannerContextKeyMap = map[shard.ScanType]interface{}{
	shard.ConcreteExecutionType: ConcreteScannerContextKey,
	shard.CurrentExecutionType:  CurrentScannerContextKey,
}

ScanTypeScannerContextKeyMap maps execution type to the context key used by scanner

Functions

func FixShardActivity

func FixShardActivity(
	activityCtx context.Context,
	params FixShardActivityParams,
) ([]shard.FixReport, error)

FixShardActivity will fix a collection of shards.

func FixerWorkflow

func FixerWorkflow(
	ctx workflow.Context,
	params FixerWorkflowParams,
) error

FixerWorkflow is the workflow that fixes all concrete executions from a scan output.

func ScanShardActivity

func ScanShardActivity(
	activityCtx context.Context,
	params ScanShardActivityParams,
) ([]shard.ScanReport, error)

ScanShardActivity will scan a collection of shards for invariant violations.

func ScannerEmitMetricsActivity

func ScannerEmitMetricsActivity(
	activityCtx context.Context,
	params ScannerEmitMetricsActivityParams,
) error

ScannerEmitMetricsActivity will emit metrics for a complete run of scanner

func ScannerWorkflow

func ScannerWorkflow(
	ctx workflow.Context,
	params ScannerWorkflowParams,
) error

ScannerWorkflow is the workflow that scans over all executions for the given scan type

Types

type AggregateFixReportResult

type AggregateFixReportResult shard.FixStats

AggregateFixReportResult indicates the result of summing together all shard reports that have finished for fix.

type AggregateScanReportResult

type AggregateScanReportResult shard.ScanStats

AggregateScanReportResult indicates the result of summing together all shard reports which have finished scan.

type ContextKey

type ContextKey int

ContextKey is the type which identifies context keys

type CorruptedKeysEntry

type CorruptedKeysEntry struct {
	ShardID       int
	CorruptedKeys store.Keys
}

CorruptedKeysEntry is a pair of shardID and corrupted keys

type DynamicConfigInvariantCollections

type DynamicConfigInvariantCollections struct {
	InvariantCollectionMutableState dynamicconfig.BoolPropertyFn
	InvariantCollectionHistory      dynamicconfig.BoolPropertyFn
}

DynamicConfigInvariantCollections is the portion of ScannerWorkflowDynamicConfig which indicates which collections of invariants should be run

type FixReportError

type FixReportError struct {
	Reports  []shard.FixReport
	ErrorStr *string
}

FixReportError is a type that is used to send either error or report on a channel. Exactly one of Report and ErrorStr should be non-nil.

type FixShardActivityParams

type FixShardActivityParams struct {
	CorruptedKeysEntries        []CorruptedKeysEntry
	ResolvedFixerWorkflowConfig ResolvedFixerWorkflowConfig
	ScanType                    shard.ScanType
}

FixShardActivityParams is the parameter for FixShardActivity

type FixShardHeartbeatDetails

type FixShardHeartbeatDetails struct {
	LastShardIndexHandled int
	Reports               []shard.FixReport
}

FixShardHeartbeatDetails is the heartbeat details for the fix shard

type FixerContext

type FixerContext struct {
	Resource resource.Resource
	Scope    metrics.Scope
}

FixerContext is the resource that is available to activities under ConcreteFixerContextKey

type FixerCorruptedKeysActivityParams

type FixerCorruptedKeysActivityParams struct {
	ScannerWorkflowWorkflowID string
	ScannerWorkflowRunID      string
	StartingShardID           *int
	ScanType                  shard.ScanType
}

FixerCorruptedKeysActivityParams is the parameter for FixerCorruptedKeysActivity

type FixerCorruptedKeysActivityResult

type FixerCorruptedKeysActivityResult struct {
	CorruptedKeys             []CorruptedKeysEntry
	MinShard                  *int
	MaxShard                  *int
	ShardQueryPaginationToken ShardQueryPaginationToken
}

FixerCorruptedKeysActivityResult is the result of FixerCorruptedKeysActivity

func FixerCorruptedKeysActivity

func FixerCorruptedKeysActivity(
	activityCtx context.Context,
	params FixerCorruptedKeysActivityParams,
) (*FixerCorruptedKeysActivityResult, error)

FixerCorruptedKeysActivity will fetch the keys of blobs from shards with corruptions from a completed scan workflow. If scan workflow is not closed or if query fails activity will return an error. Accepts as input the shard to start query at and returns a next page token, therefore this activity can be used to do pagination.

type FixerWorkflowConfigOverwrites

type FixerWorkflowConfigOverwrites struct {
	Concurrency             *int
	BlobstoreFlushThreshold *int
	ActivityBatchSize       *int
	InvariantCollections    *InvariantCollections
}

FixerWorkflowConfigOverwrites enables overwriting the default values. If provided workflow will favor overwrites over defaults. Any overwrites that are left as nil will fall back to defaults.

type FixerWorkflowParams

type FixerWorkflowParams struct {
	ScannerWorkflowWorkflowID     string
	ScannerWorkflowRunID          string
	FixerWorkflowConfigOverwrites FixerWorkflowConfigOverwrites
	ScanType                      shard.ScanType
}

FixerWorkflowParams are the parameters to the fix workflow

type InvariantCollections

type InvariantCollections struct {
	InvariantCollectionMutableState bool
	InvariantCollectionHistory      bool
}

InvariantCollections represents the resolved set of invariant collections that scanner workflow should run

type PaginatedShardQueryRequest

type PaginatedShardQueryRequest struct {
	// StartingShardID is the first shard to start iteration from.
	// Setting to nil will start iteration from the beginning of the shards.
	StartingShardID *int
	// LimitShards indicates the maximum number of results that can be returned.
	// If nil or larger than allowed maximum, will default to maximum allowed.
	LimitShards *int
}

PaginatedShardQueryRequest is the request used for queries which return results over all shards

type ResolvedFixerWorkflowConfig

type ResolvedFixerWorkflowConfig struct {
	Concurrency             int
	BlobstoreFlushThreshold int
	ActivityBatchSize       int
	InvariantCollections    InvariantCollections
}

ResolvedFixerWorkflowConfig is the resolved config after reading defaults and applying overwrites.

type ResolvedScannerWorkflowConfig

type ResolvedScannerWorkflowConfig struct {
	Enabled                 bool
	Concurrency             int
	ExecutionsPageSize      int
	BlobstoreFlushThreshold int
	ActivityBatchSize       int
	InvariantCollections    InvariantCollections
}

ResolvedScannerWorkflowConfig is the resolved config after reading dynamic config and applying overwrites.

func ScannerConfigActivity

func ScannerConfigActivity(
	activityCtx context.Context,
	params ScannerConfigActivityParams,
) (ResolvedScannerWorkflowConfig, error)

ScannerConfigActivity will read dynamic config, apply overwrites and return a resolved config.

type ScanReportError

type ScanReportError struct {
	Reports  []shard.ScanReport
	ErrorStr *string
}

ScanReportError is a type that is used to send either error or report on a channel. Exactly one of Report and ErrorStr should be non-nil.

type ScanShardActivityParams

type ScanShardActivityParams struct {
	Shards                  []int
	ExecutionsPageSize      int
	BlobstoreFlushThreshold int
	InvariantCollections    InvariantCollections
	ScanType                shard.ScanType
}

ScanShardActivityParams is the parameter for ScanShardActivity

type ScanShardHeartbeatDetails

type ScanShardHeartbeatDetails struct {
	LastShardIndexHandled int
	Reports               []shard.ScanReport
}

ScanShardHeartbeatDetails is the heartbeat details for scan shard

type ScannerConfigActivityParams

type ScannerConfigActivityParams struct {
	Overwrites ScannerWorkflowConfigOverwrites
	ScanType   shard.ScanType
}

ScannerConfigActivityParams is the parameter for ScannerConfigActivity

type ScannerContext

type ScannerContext struct {
	Resource                     resource.Resource
	Scope                        metrics.Scope
	ScannerWorkflowDynamicConfig *ScannerWorkflowDynamicConfig
}

ScannerContext is the resource that is available in activities under ConcreteScannerContextKey context key

type ScannerEmitMetricsActivityParams

type ScannerEmitMetricsActivityParams struct {
	ShardSuccessCount            int
	ShardControlFlowFailureCount int
	AggregateReportResult        AggregateScanReportResult
	ShardDistributionStats       ShardDistributionStats
	ScanType                     shard.ScanType
}

ScannerEmitMetricsActivityParams is the parameter for ScannerEmitMetricsActivity

type ScannerWorkflowConfigOverwrites

type ScannerWorkflowConfigOverwrites struct {
	Enabled                 *bool
	Concurrency             *int
	ExecutionsPageSize      *int
	BlobstoreFlushThreshold *int
	ActivityBatchSize       *int
	InvariantCollections    *InvariantCollections
}

ScannerWorkflowConfigOverwrites enables overwriting the values in dynamic config. If provided workflow will favor overwrites over dynamic config. Any overwrites that are left as nil will fall back to using dynamic config.

type ScannerWorkflowDynamicConfig

type ScannerWorkflowDynamicConfig struct {
	Enabled                           dynamicconfig.BoolPropertyFn
	Concurrency                       dynamicconfig.IntPropertyFn
	ExecutionsPageSize                dynamicconfig.IntPropertyFn
	BlobstoreFlushThreshold           dynamicconfig.IntPropertyFn
	ActivityBatchSize                 dynamicconfig.IntPropertyFn
	DynamicConfigInvariantCollections DynamicConfigInvariantCollections
}

ScannerWorkflowDynamicConfig is the dynamic config for scanner workflow

type ScannerWorkflowParams

type ScannerWorkflowParams struct {
	Shards                          Shards
	ScannerWorkflowConfigOverwrites ScannerWorkflowConfigOverwrites
	ScanType                        shard.ScanType
}

ScannerWorkflowParams are the parameters to the scan workflow

type ShardCorruptKeysQueryResult

type ShardCorruptKeysQueryResult struct {
	Result                    ShardCorruptKeysResult
	ShardQueryPaginationToken ShardQueryPaginationToken
}

ShardCorruptKeysQueryResult is the query result for ShardCorruptKeysQuery

type ShardCorruptKeysResult

type ShardCorruptKeysResult map[int]store.Keys

ShardCorruptKeysResult is a map of all shards which have finished scan successfully and have at least one corruption

type ShardDistributionStats

type ShardDistributionStats struct {
	Max    int64
	Median int64
	Min    int64
	P90    int64
	P75    int64
	P25    int64
	P10    int64
}

ShardDistributionStats contains stats on the distribution of executions in shards. It is used by the ScannerEmitMetricsActivityParams.

type ShardQueryPaginationToken

type ShardQueryPaginationToken struct {
	// NextShardID is one greater than the highest shard returned in the current query.
	// NextShardID is nil if IsDone is true.
	// It is possible to get NextShardID != nil and on the next call to get an empty result with IsDone = true.
	NextShardID *int
	IsDone      bool
}

ShardQueryPaginationToken is used to return information used to make the next query

type ShardRange

type ShardRange struct {
	Min int
	Max int
}

ShardRange identifies a set of shards based on min (inclusive) and max (exclusive)

type ShardSizeQueryRequest

type ShardSizeQueryRequest struct {
	StartIndex int
	EndIndex   int
}

ShardSizeQueryRequest is the request used for ShardSizeQuery. The following must be true: 0 <= StartIndex < EndIndex <= len(shards successfully finished) The following must be true: EndIndex - StartIndex <= maxShardQueryResult. StartIndex is inclusive, EndIndex is exclusive.

type ShardSizeQueryResult

type ShardSizeQueryResult []ShardSizeTuple

ShardSizeQueryResult is the result from ShardSizeQuery. Contains sorted list of shards, sorted by the number of executions per shard.

type ShardSizeTuple

type ShardSizeTuple struct {
	ShardID         int
	ExecutionsCount int64
}

ShardSizeTuple indicates the size and sorted index of a single shard

type ShardStatus

type ShardStatus string

ShardStatus is the type which indicates the status of a shard scan.

type ShardStatusQueryResult

type ShardStatusQueryResult struct {
	Result                    ShardStatusResult
	ShardQueryPaginationToken ShardQueryPaginationToken
}

ShardStatusQueryResult is the query result for ShardStatusQuery

type ShardStatusResult

type ShardStatusResult map[int]ShardStatus

ShardStatusResult indicates the status for all shards

type ShardStatusSummaryResult

type ShardStatusSummaryResult map[ShardStatus]int

ShardStatusSummaryResult indicates the counts of shards in each status

type Shards

type Shards struct {
	List  []int
	Range *ShardRange
}

Shards identify the shards that should be scanned. Exactly one of List or Range should be non-nil.

func (Shards) Flatten

func (s Shards) Flatten() ([]int, int, int)

Flatten flattens Shards to a list of shard IDs and finds the min/max shardID

func (Shards) Validate

func (s Shards) Validate() error

Validate validates shard list or range