replication

type DLQHandler ¶

type DLQHandler interface {
	common.Daemon

	GetMessageCount(
		ctx context.Context,
		forceFetch bool,
	) (map[string]int64, error)
	ReadMessages(
		ctx context.Context,
		sourceCluster string,
		lastMessageID int64,
		pageSize int,
		pageToken []byte,
	) ([]*types.ReplicationTask, []*types.ReplicationTaskInfo, []byte, error)
	PurgeMessages(
		ctx context.Context,
		sourceCluster string,
		lastMessageID int64,
	) error
	MergeMessages(
		ctx context.Context,
		sourceCluster string,
		lastMessageID int64,
		pageSize int,
		pageToken []byte,
	) ([]byte, error)
}

DLQHandler is the interface handles replication DLQ messages

func NewDLQHandler(
	shard shard.Context,
	taskExecutors map[string]TaskExecutor,
) DLQHandler

type DynamicTaskBatchSizer ¶ added in v1.3.0

type DynamicTaskBatchSizer interface {
	// contains filtered or unexported methods
}

DynamicTaskBatchSizer is responsible for the batch size used to retrieve ReplicationTasks by TaskAckManager It adjusts the task batch size based on the error and the getTasksResult, and use the following rules:

1. If there is an error, decrease the task batch size. In case of an increased load to the database, we should reduce the load to the database Emitted metric with tag reason:"error"

2. If the task batch size is shrunk, decrease the task batch size. The payload size of messages is too big, so we should decrease the number of messages to be sure that future messages will not be shrunk Emitted metric with tag reason:"shrunk"

3. If the read level of a passive cluster has not been changed and there are no fetched tasks, not change the task batch size. There is no need to change because the replication is not stuck, there just are no new tasks Metric is not emitted

4. If the read level of a passive cluster has not been changed and if there are fetched tasks, and number of previously fetched tasks is not zero, decrease the task batch size. The replication is stuck on the passive side Emitted metric with tag reason:"possible_stuck"

5. If the read level of a passive cluster has not been changed and if there are fetched tasks, and number of previously fetched tasks is zero, not change the task batch size. The replication is not stuck, and there are new tasks to be replicated Metric is not emitted

6. If the read level of a passive cluster has been changed and if there are more tasks in db, increase the task batch size. We should retrieve the maximum possible value at the next time, as there are more tasks to be replicated Emitted metric with tag reason:"more_tasks"

7. If the read level of a passive cluster has been changed and if there are no more tasks in db, not change the size. The existing size is already enough, and there are no more tasks to be replicated Metric is not emitted

func NewDynamicTaskBatchSizer(shardID int, logger log.Logger, config *config.Config, metricsClient metrics.Client) DynamicTaskBatchSizer

type MetricsEmitterImpl ¶ added in v0.25.0

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

MetricsEmitterImpl is responsible for emitting source side replication metrics occasionally.

func NewMetricsEmitter(
	shardID int,
	shardData metricsEmitterShardData,
	reader taskReader,
	metricsClient metrics.Client,
) *MetricsEmitterImpl

func (m *MetricsEmitterImpl) Start()

func (m *MetricsEmitterImpl) Stop()

type MockTaskExecutor ¶

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

MockTaskExecutor is a mock of TaskExecutor interface.

func NewMockTaskExecutor(ctrl *gomock.Controller) *MockTaskExecutor

func (m *MockTaskExecutor) EXPECT() *MockTaskExecutorMockRecorder

type MockTaskExecutorMockRecorder ¶

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

MockTaskExecutorMockRecorder is the mock recorder for MockTaskExecutor.

type MockTaskFetcher ¶

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

MockTaskFetcher is a mock of TaskFetcher interface.

func NewMockTaskFetcher(ctrl *gomock.Controller) *MockTaskFetcher

func (m *MockTaskFetcher) EXPECT() *MockTaskFetcherMockRecorder

func (m *MockTaskFetcher) GetRateLimiter() quotas.Limiter

func (m *MockTaskFetcher) GetRequestChan(shardID int) chan<- *request

func (m *MockTaskFetcher) GetSourceCluster() string

func (m *MockTaskFetcher) Start()

func (m *MockTaskFetcher) Stop()

type MockTaskFetcherMockRecorder ¶

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

MockTaskFetcherMockRecorder is the mock recorder for MockTaskFetcher.

func (mr *MockTaskFetcherMockRecorder) GetRateLimiter() *gomock.Call

func (mr *MockTaskFetcherMockRecorder) GetRequestChan(shardID any) *gomock.Call

func (mr *MockTaskFetcherMockRecorder) GetSourceCluster() *gomock.Call

func (mr *MockTaskFetcherMockRecorder) Start() *gomock.Call

func (mr *MockTaskFetcherMockRecorder) Stop() *gomock.Call

type MockTaskFetchers ¶

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

MockTaskFetchers is a mock of TaskFetchers interface.

func NewMockTaskFetchers(ctrl *gomock.Controller) *MockTaskFetchers

func (m *MockTaskFetchers) EXPECT() *MockTaskFetchersMockRecorder

func (m *MockTaskFetchers) GetFetchers() []TaskFetcher

func (m *MockTaskFetchers) Start()

func (m *MockTaskFetchers) Stop()

type MockTaskFetchersMockRecorder ¶

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

MockTaskFetchersMockRecorder is the mock recorder for MockTaskFetchers.

func (mr *MockTaskFetchersMockRecorder) GetFetchers() *gomock.Call

func (mr *MockTaskFetchersMockRecorder) Start() *gomock.Call

func (mr *MockTaskFetchersMockRecorder) Stop() *gomock.Call

type TaskAckManager ¶ added in v0.15.0

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

TaskAckManager is the ack manager for replication tasks

func NewTaskAckManager(
	shardID int,
	ackLevels ackLevelStore,
	metricsClient metrics.Client,
	logger log.Logger,
	reader taskReader,
	store *TaskStore,
	timeSource clock.TimeSource,
	config *config.Config,
	replicationMessagesSizeFn types.ReplicationMessagesSizeFn,
	dynamicTaskBatchSizer DynamicTaskBatchSizer,
) TaskAckManager

func (t *TaskAckManager) GetTasks(ctx context.Context, pollingCluster string, lastReadTaskID int64) (_ *types.ReplicationMessages, err error)

type TaskExecutor ¶

type TaskExecutor interface {
	// contains filtered or unexported methods
}

TaskExecutor is the executor for replication task

func NewTaskExecutor(
	sourceCluster string,
	shard shard.Context,
	domainCache cache.DomainCache,
	historyResender ndc.HistoryResender,
	historyEngine engine.Engine,
	metricsClient metrics.Client,
	logger log.Logger,
) TaskExecutor

type TaskFetcher ¶

type TaskFetcher interface {
	common.Daemon

	GetSourceCluster() string
	GetRequestChan(shardID int) chan<- *request
	GetRateLimiter() quotas.Limiter
}

TaskFetcher is responsible for fetching replication messages from remote DC.

type TaskFetchers ¶

type TaskFetchers interface {
	common.Daemon

	GetFetchers() []TaskFetcher
}

TaskFetchers is a group of fetchers, one per source DC.

func NewTaskFetchers(
	logger log.Logger,
	config *config.Config,
	clusterMetadata cluster.Metadata,
	clientBean client.Bean,
	metricsClient metrics.Client,
) (TaskFetchers, error)

type TaskHydrator ¶ added in v0.25.0

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

TaskHydrator will enrich replication task with additional information from mutable state and history events. Mutable state and history providers can be either in-memory or persistence based implementations; depending whether we have available data already or need to load it.

func NewDeferredTaskHydrator(shardID int, historyManager persistence.HistoryManager, executionCache execution.Cache, domains domainCache) TaskHydrator

func NewImmediateTaskHydrator(isRunning bool, vh *persistence.VersionHistories, activities map[int64]*persistence.ActivityInfo, blob, nextBlob *persistence.DataBlob) TaskHydrator

func (h TaskHydrator) Hydrate(ctx context.Context, task persistence.Task) (retTask *types.ReplicationTask, retErr error)

type TaskProcessor ¶

type TaskProcessor interface {
	common.Daemon
}

TaskProcessor is responsible for processing replication tasks for a shard.

func NewTaskProcessor(
	shard shard.Context,
	historyEngine engine.Engine,
	config *config.Config,
	metricsClient metrics.Client,
	taskFetcher TaskFetcher,
	taskExecutor TaskExecutor,
	clock clock.TimeSource,
) TaskProcessor

type TaskReader ¶ added in v1.3.0

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

TaskReader will read replication tasks from database

func NewTaskReader(shardID int, executionManager persistence.ExecutionManager) *TaskReader

func (r *TaskReader) Read(ctx context.Context, readLevel int64, maxReadLevel int64, batchSize int) ([]persistence.Task, bool, error)

type TaskStore ¶ added in v0.25.0

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

TaskStore is a component that hydrates and caches replication messages so that they can be reused across several polling source clusters. It also exposes public Put method. This allows pre-store already hydrated messages at the end of successful transaction, saving a DB call to fetch history events.

TaskStore uses a separate cache per each source cluster allowing messages to be fetched at different rates. Once a cache becomes full it will not accept further messages for that cluster. Later those messages be fetched from DB and hydrated again. A cache stores only a pointer to the message. It is hydrates once and shared across caches. Cluster acknowledging the message will remove it from that corresponding cache. Once all clusters acknowledge it, no more references will be held, and GC will eventually pick it up.

func NewTaskStore(
	config *config.Config,
	clusterMetadata cluster.Metadata,
	domains domainCache,
	metricsClient metrics.Client,
	logger log.Logger,
	hydrator taskHydrator,
	budgetManager cache.Manager,
	shardID int,
) *TaskStore

func (m *TaskStore) Ack(cluster string, lastTaskID int64) error

func (m *TaskStore) Get(ctx context.Context, cluster string, info persistence.Task) (*types.ReplicationTask, error)

func (m *TaskStore) Put(task *types.ReplicationTask)