shardscanner

Jan 23, 2024

Changes in this version

+ const AllResultsQuery

type FixResult

+ func (f *FixResult) Empty() bool

type ScanResult

+ func (s *ScanResult) Empty() bool

type ShardFixResultAggregator

+ func (a *ShardFixResultAggregator) GetAllFixResults() (map[int]FixResult, error)

type ShardScanResultAggregator

+ func (a *ShardScanResultAggregator) GetAllScanResults() (map[int]ScanResult, error)

Oct 19, 2023

Changes in this version

+ const ActivityFixerConfig

type FixShardActivityParams

+ EnabledInvariants CustomScannerConfig

+ type FixShardConfigParams struct

+ type FixShardConfigResults struct

+ EnabledInvariants CustomScannerConfig

type FixerHooks

+ GetFixerConfig func(fixer FixerContext) CustomScannerConfig

+ type ScannerContext struct

+ Config *ScannerConfig

+ Hooks *ScannerHooks

+ Logger log.Logger

+ Resource resource.Resource

+ Scope metrics.Scope

May 24, 2022

Changes in this version

type Context

+ Logger log.Logger

type FixerContext

+ Logger log.Logger

May 6, 2021

Changes in this version

+ const DomainReportQuery

+ type DomainFixReportQueryResult struct

+ Reports []DomainFixStats

+ type DomainFixStats struct

+ DomainID string

+ Stats FixStats

+ type DomainReportQueryRequest struct

+ DomainID *string

+ type DomainScanReportQueryResult struct

+ Reports []DomainScanStats

+ type DomainScanStats struct

+ DomainID string

+ Stats ScanStats

type FixReport

+ DomainStats map[string]*FixStats

type ScanReport

+ DomainStats map[string]*ScanStats

type ShardFixResultAggregator

+ func (a *ShardFixResultAggregator) GetDomainStatus(req DomainReportQueryRequest) (*DomainFixReportQueryResult, error)

type ShardScanResultAggregator

+ func (a *ShardScanResultAggregator) GetDomainStatus(req DomainReportQueryRequest) (*DomainScanReportQueryResult, error)

Mar 1, 2021

Changes in this version

+ func NewFixerContext(ctx context.Context, workflowName string, fixerContext FixerContext) context.Context

+ func NewScannerContext(ctx context.Context, workflowName string, scannerContext Context) context.Context

type Context

+ func GetScannerContext(ctx context.Context) (Context, error)

+ func NewShardScannerContext(res resource.Resource, config *ScannerConfig) Context

type FixerContext

+ func GetFixerContext(ctx context.Context) (FixerContext, error)

+ func NewShardFixerContext(res resource.Resource, config *ScannerConfig) FixerContext

Jan 22, 2021

Changes in this version

type DynamicParams

+ AllowDomain dynamicconfig.BoolPropertyFnWithDomainFilter

+ FixerEnabled dynamicconfig.BoolPropertyFn

+ ScannerEnabled dynamicconfig.BoolPropertyFn

type ScannerConfig

+ StartFixerOptions cclient.StartWorkflowOptions

Jan 13, 2021

Changes in this version

+ const ActivityFixShard

+ const ActivityFixerCorruptedKeys

+ const ActivityScanShard

+ const ActivityScannerConfig

+ const ActivityScannerEmitMetrics

+ const AggregateReportQuery

+ const ErrMissingHooks

+ const ErrScanWorkflowNotClosed

+ const ErrSerialization

+ const ShardCorruptKeysQuery

+ const ShardReportQuery

+ const ShardSizeQuery

+ const ShardStatusQuery

+ const ShardStatusSummaryQuery

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

+ type AggregateFixReportResult FixStats

+ type AggregateScanReportResult ScanStats

+ type Context struct

+ Config *ScannerConfig

+ ContextKey ScannerContextKey

+ Hooks *ScannerHooks

+ Resource resource.Resource

+ Scope metrics.Scope

+ type ControlFlowFailure struct

+ Info string

+ InfoDetails string

+ type CorruptedKeysEntry struct

+ CorruptedKeys store.Keys

+ ShardID int

+ type CustomScannerConfig map[string]string

+ type DynamicParams struct

+ ActivityBatchSize dynamicconfig.IntPropertyFn

+ BlobstoreFlushThreshold dynamicconfig.IntPropertyFn

+ Concurrency dynamicconfig.IntPropertyFn

+ Enabled dynamicconfig.BoolPropertyFn

+ PageSize dynamicconfig.IntPropertyFn

+ type FixKeys struct

+ Failed *store.Keys

+ Fixed *store.Keys

+ Skipped *store.Keys

+ type FixReport struct

+ Result FixResult

+ ShardID int

+ Stats FixStats

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

+ type FixReportError struct

+ ErrorStr *string

+ Reports []FixReport

+ type FixResult struct

+ ControlFlowFailure *ControlFlowFailure

+ ShardFixKeys *FixKeys

+ type FixShardActivityParams struct

+ ContextKey ScannerContextKey

+ CorruptedKeysEntries []CorruptedKeysEntry

+ ResolvedFixerWorkflowConfig ResolvedFixerWorkflowConfig

+ type FixShardHeartbeatDetails struct

+ LastShardIndexHandled int

+ Reports []FixReport

+ type FixStats struct

+ EntitiesCount int64

+ FailedCount int64

+ FixedCount int64

+ SkippedCount int64

+ type Fixer interface

+ Fix func() FixReport

+ type FixerContext struct

+ Config *ScannerConfig

+ ContextKey ScannerContextKey

+ Hooks *FixerHooks

+ Resource resource.Resource

+ Scope metrics.Scope

+ type FixerCorruptedKeysActivityParams struct

+ ContextKey ScannerContextKey

+ ScannerWorkflowRunID string

+ ScannerWorkflowWorkflowID string

+ StartingShardID *int

+ type FixerCorruptedKeysActivityResult struct

+ CorruptedKeys []CorruptedKeysEntry

+ MaxShard *int

+ MinShard *int

+ ShardQueryPaginationToken ShardQueryPaginationToken

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

+ func GetCorruptedKeys(ctx workflow.Context, params FixerWorkflowParams) (*FixerCorruptedKeysActivityResult, error)

+ type FixerHooks struct

+ InvariantManager FixerManagerCB

+ Iterator FixerIteratorCB

+ func NewFixerHooks(manager FixerManagerCB, iterator FixerIteratorCB) (*FixerHooks, error)

+ type FixerIteratorCB func(blobstore.Client, store.Keys, FixShardActivityParams, ScannerConfig) store.ScanOutputIterator

+ type FixerManagerCB func(persistence.Retryer, FixShardActivityParams, ScannerConfig) invariant.Manager

+ type FixerWorkflow struct

+ Aggregator *ShardFixResultAggregator

+ Keys *FixerCorruptedKeysActivityResult

+ Params FixerWorkflowParams

+ func NewFixerWorkflow(ctx workflow.Context, name string, params FixerWorkflowParams) (*FixerWorkflow, error)

+ func (fx *FixerWorkflow) Start(ctx workflow.Context) error

+ type FixerWorkflowConfigOverwrites struct

+ ActivityBatchSize *int

+ BlobstoreFlushThreshold *int

+ Concurrency *int

+ type FixerWorkflowParams struct

+ ContextKey ScannerContextKey

+ FixerWorkflowConfigOverwrites FixerWorkflowConfigOverwrites

+ ScannerWorkflowRunID string

+ ScannerWorkflowWorkflowID string

+ type GenericScannerConfig struct

+ ActivityBatchSize int

+ BlobstoreFlushThreshold int

+ Concurrency int

+ Enabled bool

+ PageSize int

+ type GenericScannerConfigOverwrites struct

+ ActivityBatchSize *int

+ BlobstoreFlushThreshold *int

+ Concurrency *int

+ Enabled *bool

+ PageSize *int

+ type IteratorCB func(persistence.Retryer, ScanShardActivityParams, ScannerConfig) pagination.Iterator

+ type ManagerCB func(persistence.Retryer, ScanShardActivityParams, ScannerConfig) invariant.Manager

+ type PaginatedShardQueryRequest struct

+ LimitShards *int

+ StartingShardID *int

+ type ResolvedFixerWorkflowConfig struct

+ ActivityBatchSize int

+ BlobstoreFlushThreshold int

+ Concurrency int

+ type ResolvedScannerWorkflowConfig struct

+ CustomScannerConfig CustomScannerConfig

+ GenericScannerConfig GenericScannerConfig

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

+ type ScanKeys struct

+ Corrupt *store.Keys

+ Failed *store.Keys

+ type ScanReport struct

+ Result ScanResult

+ ShardID int

+ Stats ScanStats

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

+ type ScanReportError struct

+ ErrorStr *string

+ Reports []ScanReport

+ type ScanResult struct

+ ControlFlowFailure *ControlFlowFailure

+ ShardScanKeys *ScanKeys

+ type ScanShardActivityParams struct

+ BlobstoreFlushThreshold int

+ ContextKey ScannerContextKey

+ PageSize int

+ ScannerConfig CustomScannerConfig

+ Shards []int

+ type ScanShardHeartbeatDetails struct

+ LastShardIndexHandled int

+ Reports []ScanReport

+ type ScanStats struct

+ CheckFailedCount int64

+ CorruptedCount int64

+ CorruptionByType map[invariant.Name]int64

+ EntitiesCount int64

+ type Scanner interface

+ Scan func() ScanReport

+ type ScannerConfig struct

+ DynamicCollection *dynamicconfig.Collection

+ DynamicParams DynamicParams

+ FixerHooks func() *FixerHooks

+ FixerTLName string

+ FixerWFTypeName string

+ ScannerHooks func() *ScannerHooks

+ ScannerWFTypeName string

+ StartWorkflowOptions cclient.StartWorkflowOptions

+ type ScannerConfigActivityParams struct

+ ContextKey ScannerContextKey

+ Overwrites ScannerWorkflowConfigOverwrites

+ type ScannerContextKey string

+ func (c ScannerContextKey) String() string

+ type ScannerEmitMetricsActivityParams struct

+ AggregateReportResult AggregateScanReportResult

+ ContextKey ScannerContextKey

+ ShardControlFlowFailureCount int

+ ShardDistributionStats ShardDistributionStats

+ ShardSuccessCount int

+ type ScannerHooks struct

+ GetScannerConfig func(scanner Context) CustomScannerConfig

+ Iterator IteratorCB

+ Manager ManagerCB

+ func NewScannerHooks(manager ManagerCB, iterator IteratorCB) (*ScannerHooks, error)

+ func (sh *ScannerHooks) SetConfig(config func(scanner Context) CustomScannerConfig)

+ type ScannerWorkflow struct

+ Aggregator *ShardScanResultAggregator

+ Name string

+ Params ScannerWorkflowParams

+ Shards []int

+ func NewScannerWorkflow(ctx workflow.Context, name string, params ScannerWorkflowParams) (*ScannerWorkflow, error)

+ func (wf *ScannerWorkflow) Start(ctx workflow.Context) error

+ type ScannerWorkflowConfigOverwrites struct

+ CustomScannerConfig *CustomScannerConfig

+ GenericScannerConfig GenericScannerConfigOverwrites

+ type ScannerWorkflowParams struct

+ ScannerWorkflowConfigOverwrites ScannerWorkflowConfigOverwrites

+ Shards Shards

+ type ShardCorruptKeysQueryResult struct

+ Result ShardCorruptKeysResult

+ ShardQueryPaginationToken ShardQueryPaginationToken

+ type ShardCorruptKeysResult map[int]store.Keys

+ type ShardDistributionStats struct

+ Max int64

+ Median int64

+ Min int64

+ P10 int64

+ P25 int64

+ P75 int64

+ P90 int64

+ type ShardFixResultAggregator struct

+ func NewShardFixResultAggregator(corruptKeys []CorruptedKeysEntry, minShard int, maxShard int) *ShardFixResultAggregator

+ func (a *ShardFixResultAggregator) AddReport(report FixReport)

+ func (a *ShardFixResultAggregator) GetAggregation() AggregateFixReportResult

+ func (a *ShardFixResultAggregator) GetReport(shardID int) (*FixReport, error)

+ func (a *ShardFixResultAggregator) GetStatusResult(req PaginatedShardQueryRequest) (*ShardStatusQueryResult, error)

+ func (a *ShardFixResultAggregator) GetStatusSummary() ShardStatusSummaryResult

+ type ShardFixer struct

+ func NewFixer(shardID int, manager invariant.Manager, iterator store.ScanOutputIterator, ...) *ShardFixer

+ func (f *ShardFixer) Fix(ctx context.Context) FixReport

+ type ShardQueryPaginationToken struct

+ IsDone bool

+ NextShardID *int

+ type ShardRange struct

+ Max int

+ Min int

+ type ShardScanResultAggregator struct

+ func NewShardScanResultAggregator(shards []int, minShard int, maxShard int) *ShardScanResultAggregator

+ func (a *ShardScanResultAggregator) AddReport(report ScanReport)

+ func (a *ShardScanResultAggregator) GetAggregateReport() AggregateScanReportResult

+ func (a *ShardScanResultAggregator) GetCorruptionKeys(req PaginatedShardQueryRequest) (*ShardCorruptKeysQueryResult, error)

+ func (a *ShardScanResultAggregator) GetReport(shardID int) (*ScanReport, error)

+ func (a *ShardScanResultAggregator) GetShardDistributionStats() ShardDistributionStats

+ func (a *ShardScanResultAggregator) GetShardSizeQueryResult(req ShardSizeQueryRequest) (ShardSizeQueryResult, error)

+ func (a *ShardScanResultAggregator) GetStatusResult(req PaginatedShardQueryRequest) (*ShardStatusQueryResult, error)

+ func (a *ShardScanResultAggregator) GetStatusSummary() ShardStatusSummaryResult

+ type ShardScanner struct

+ func NewScanner(shardID int, iterator pagination.Iterator, blobstoreClient blobstore.Client, ...) *ShardScanner

+ func (s *ShardScanner) Scan(ctx context.Context) ScanReport

+ type ShardSizeQueryRequest struct

+ EndIndex int

+ StartIndex int

+ type ShardSizeQueryResult []ShardSizeTuple

+ type ShardSizeTuple struct

+ EntitiesCount int64

+ ShardID int

+ type ShardStatus string

+ const ShardStatusControlFlowFailure

+ const ShardStatusRunning

+ const ShardStatusSuccess

+ type ShardStatusQueryResult struct

+ Result ShardStatusResult

+ ShardQueryPaginationToken ShardQueryPaginationToken

+ type ShardStatusResult map[int]ShardStatus

+ type ShardStatusSummaryResult map[ShardStatus]int

+ type Shards struct

+ List []int

+ Range *ShardRange

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

+ func (s Shards) Validate() error