mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Alex ShtinandGitHub
8e8f1b0f7c
## What changed?
<!-- Describe what has changed in this PR -->
Remove license header from every file. Because it is really hard to
follow in this PR here is the summary:
1. License header is removed from all `.go` and `.proto` files
:fireworks::fireworks:🎆.
2. `LICENSE` file in the root directory has only Temporal and Uber
copyrights.
3. 5 other `LICENSE` files added to the packages which have copyrights
different from Temporal and Uber: Datadog, Xargin, "Mat Ryer, Tyler
Bunnell and contributors".
4. `license_file` flag is removed from all code generation tools.
5. `copyright_file` flag is removed from `go:generate mockgen`
directive.
6. All copyright related targets are removed from `Makefile`.
7. Updated Temporal copyright year to 2025 everywhere.
## Why?
<!-- Tell your future self why have you made these changes -->
I double checked with legal department that it is not needed to have
license header in every file. One file per repo is enough. I put all
copyrights to the root `LICENSE` file and removed header from all other
files. Also updated tools and `Makefile`.
371 lines
10 KiB
Go
371 lines
10 KiB
Go
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination dlq_handler_mock.go
|
|
|
|
package replication
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
|
|
historypb "go.temporal.io/api/history/v1"
|
|
"go.temporal.io/server/api/adminservice/v1"
|
|
enumsspb "go.temporal.io/server/api/enums/v1"
|
|
historyspb "go.temporal.io/server/api/history/v1"
|
|
replicationspb "go.temporal.io/server/api/replication/v1"
|
|
"go.temporal.io/server/client"
|
|
"go.temporal.io/server/common/definition"
|
|
"go.temporal.io/server/common/log"
|
|
"go.temporal.io/server/common/log/tag"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/persistence"
|
|
"go.temporal.io/server/common/xdc"
|
|
"go.temporal.io/server/service/history/consts"
|
|
"go.temporal.io/server/service/history/deletemanager"
|
|
historyi "go.temporal.io/server/service/history/interfaces"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
wcache "go.temporal.io/server/service/history/workflow/cache"
|
|
)
|
|
|
|
type (
|
|
// DLQHandler is the interface handles replication DLQ messages
|
|
DLQHandler interface {
|
|
GetMessages(
|
|
ctx context.Context,
|
|
sourceCluster string,
|
|
lastMessageID int64,
|
|
pageSize int,
|
|
pageToken []byte,
|
|
) ([]*replicationspb.ReplicationTask, []*replicationspb.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)
|
|
}
|
|
|
|
dlqHandlerImpl struct {
|
|
taskExecutorsLock sync.Mutex
|
|
taskExecutors map[string]TaskExecutor
|
|
shard historyi.ShardContext
|
|
deleteManager deletemanager.DeleteManager
|
|
workflowCache wcache.Cache
|
|
resender xdc.NDCHistoryResender
|
|
taskExecutorProvider TaskExecutorProvider
|
|
logger log.Logger
|
|
}
|
|
)
|
|
|
|
func NewLazyDLQHandler(
|
|
shard historyi.ShardContext,
|
|
deleteManager deletemanager.DeleteManager,
|
|
workflowCache wcache.Cache,
|
|
clientBean client.Bean,
|
|
taskExecutorProvider TaskExecutorProvider,
|
|
) DLQHandler {
|
|
return newDLQHandler(
|
|
shard,
|
|
deleteManager,
|
|
workflowCache,
|
|
clientBean,
|
|
make(map[string]TaskExecutor),
|
|
taskExecutorProvider,
|
|
)
|
|
}
|
|
|
|
func newDLQHandler(
|
|
shard historyi.ShardContext,
|
|
deleteManager deletemanager.DeleteManager,
|
|
workflowCache wcache.Cache,
|
|
clientBean client.Bean,
|
|
taskExecutors map[string]TaskExecutor,
|
|
taskExecutorProvider TaskExecutorProvider,
|
|
) *dlqHandlerImpl {
|
|
|
|
if taskExecutors == nil {
|
|
panic("Failed to initialize replication DLQ handler due to nil task executors")
|
|
}
|
|
|
|
return &dlqHandlerImpl{
|
|
shard: shard,
|
|
deleteManager: deleteManager,
|
|
workflowCache: workflowCache,
|
|
resender: xdc.NewNDCHistoryResender(
|
|
shard.GetNamespaceRegistry(),
|
|
clientBean,
|
|
func(
|
|
ctx context.Context,
|
|
sourceClusterName string,
|
|
namespaceId namespace.ID,
|
|
workflowId string,
|
|
runId string,
|
|
events [][]*historypb.HistoryEvent,
|
|
versionHistory []*historyspb.VersionHistoryItem,
|
|
) error {
|
|
engine, err := shard.GetEngine(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = engine.ReplicateHistoryEvents(
|
|
ctx,
|
|
definition.WorkflowKey{
|
|
NamespaceID: namespaceId.String(),
|
|
WorkflowID: workflowId,
|
|
RunID: runId,
|
|
},
|
|
nil,
|
|
versionHistory,
|
|
events,
|
|
nil,
|
|
"",
|
|
)
|
|
if errors.Is(err, consts.ErrDuplicate) {
|
|
return nil
|
|
}
|
|
return err
|
|
|
|
},
|
|
shard.GetPayloadSerializer(),
|
|
shard.GetConfig().StandbyTaskReReplicationContextTimeout,
|
|
shard.GetLogger(),
|
|
nil,
|
|
),
|
|
taskExecutors: taskExecutors,
|
|
taskExecutorProvider: taskExecutorProvider,
|
|
logger: shard.GetLogger(),
|
|
}
|
|
}
|
|
|
|
func (r *dlqHandlerImpl) GetMessages(
|
|
ctx context.Context,
|
|
sourceCluster string,
|
|
lastMessageID int64,
|
|
pageSize int,
|
|
pageToken []byte,
|
|
) ([]*replicationspb.ReplicationTask, []*replicationspb.ReplicationTaskInfo, []byte, error) {
|
|
|
|
taskList, taskInfoList, _, token, err := r.readMessagesWithAckLevel(
|
|
ctx,
|
|
sourceCluster,
|
|
lastMessageID,
|
|
pageSize,
|
|
pageToken,
|
|
)
|
|
return taskList, taskInfoList, token, err
|
|
}
|
|
|
|
func (r *dlqHandlerImpl) PurgeMessages(
|
|
ctx context.Context,
|
|
sourceCluster string,
|
|
lastMessageID int64,
|
|
) error {
|
|
|
|
ackLevel := r.shard.GetReplicatorDLQAckLevel(sourceCluster)
|
|
err := r.shard.GetExecutionManager().RangeDeleteReplicationTaskFromDLQ(
|
|
ctx,
|
|
&persistence.RangeDeleteReplicationTaskFromDLQRequest{
|
|
RangeCompleteHistoryTasksRequest: persistence.RangeCompleteHistoryTasksRequest{
|
|
ShardID: r.shard.GetShardID(),
|
|
TaskCategory: tasks.CategoryReplication,
|
|
InclusiveMinTaskKey: tasks.NewImmediateKey(ackLevel + 1),
|
|
ExclusiveMaxTaskKey: tasks.NewImmediateKey(lastMessageID + 1),
|
|
},
|
|
SourceClusterName: sourceCluster,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err = r.shard.UpdateReplicatorDLQAckLevel(
|
|
sourceCluster,
|
|
lastMessageID,
|
|
); err != nil {
|
|
r.logger.Error("Failed to purge history replication message", tag.Error(err))
|
|
// The update ack level should not block the call. Ignore the error.
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *dlqHandlerImpl) MergeMessages(
|
|
ctx context.Context,
|
|
sourceCluster string,
|
|
lastMessageID int64,
|
|
pageSize int,
|
|
pageToken []byte,
|
|
) ([]byte, error) {
|
|
|
|
replicationTasks, _, ackLevel, token, err := r.readMessagesWithAckLevel(
|
|
ctx,
|
|
sourceCluster,
|
|
lastMessageID,
|
|
pageSize,
|
|
pageToken,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
taskExecutor, err := r.getOrCreateTaskExecutor(sourceCluster)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, task := range replicationTasks {
|
|
if err := taskExecutor.Execute(
|
|
ctx,
|
|
task,
|
|
true,
|
|
); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
err = r.shard.GetExecutionManager().RangeDeleteReplicationTaskFromDLQ(
|
|
ctx,
|
|
&persistence.RangeDeleteReplicationTaskFromDLQRequest{
|
|
RangeCompleteHistoryTasksRequest: persistence.RangeCompleteHistoryTasksRequest{
|
|
ShardID: r.shard.GetShardID(),
|
|
TaskCategory: tasks.CategoryReplication,
|
|
InclusiveMinTaskKey: tasks.NewImmediateKey(ackLevel + 1),
|
|
ExclusiveMaxTaskKey: tasks.NewImmediateKey(lastMessageID + 1),
|
|
},
|
|
SourceClusterName: sourceCluster,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if err = r.shard.UpdateReplicatorDLQAckLevel(
|
|
sourceCluster,
|
|
lastMessageID,
|
|
); err != nil {
|
|
r.logger.Error("Failed to purge history replication message", tag.Error(err))
|
|
// The update ack level should not block the call. Ignore the error.
|
|
}
|
|
return token, nil
|
|
}
|
|
|
|
func (r *dlqHandlerImpl) readMessagesWithAckLevel(
|
|
ctx context.Context,
|
|
sourceCluster string,
|
|
lastMessageID int64,
|
|
pageSize int,
|
|
pageToken []byte,
|
|
) ([]*replicationspb.ReplicationTask, []*replicationspb.ReplicationTaskInfo, int64, []byte, error) {
|
|
|
|
ackLevel := r.shard.GetReplicatorDLQAckLevel(sourceCluster)
|
|
resp, err := r.shard.GetExecutionManager().GetReplicationTasksFromDLQ(ctx, &persistence.GetReplicationTasksFromDLQRequest{
|
|
GetHistoryTasksRequest: persistence.GetHistoryTasksRequest{
|
|
ShardID: r.shard.GetShardID(),
|
|
TaskCategory: tasks.CategoryReplication,
|
|
InclusiveMinTaskKey: tasks.NewImmediateKey(ackLevel + 1),
|
|
ExclusiveMaxTaskKey: tasks.NewImmediateKey(lastMessageID + 1),
|
|
BatchSize: pageSize,
|
|
NextPageToken: pageToken,
|
|
},
|
|
SourceClusterName: sourceCluster,
|
|
})
|
|
if err != nil {
|
|
return nil, nil, ackLevel, nil, err
|
|
}
|
|
pageToken = resp.NextPageToken
|
|
|
|
remoteAdminClient, err := r.shard.GetRemoteAdminClient(sourceCluster)
|
|
if err != nil {
|
|
return nil, nil, ackLevel, nil, err
|
|
}
|
|
taskInfo := make([]*replicationspb.ReplicationTaskInfo, 0, len(resp.Tasks))
|
|
for _, task := range resp.Tasks {
|
|
switch task := task.(type) {
|
|
case *tasks.SyncActivityTask:
|
|
taskInfo = append(taskInfo, &replicationspb.ReplicationTaskInfo{
|
|
NamespaceId: task.NamespaceID,
|
|
WorkflowId: task.WorkflowID,
|
|
RunId: task.RunID,
|
|
TaskType: enumsspb.TASK_TYPE_REPLICATION_SYNC_ACTIVITY,
|
|
TaskId: task.TaskID,
|
|
Version: task.GetVersion(),
|
|
FirstEventId: 0,
|
|
NextEventId: 0,
|
|
ScheduledEventId: task.ScheduledEventID,
|
|
})
|
|
case *tasks.HistoryReplicationTask:
|
|
taskInfo = append(taskInfo, &replicationspb.ReplicationTaskInfo{
|
|
NamespaceId: task.NamespaceID,
|
|
WorkflowId: task.WorkflowID,
|
|
RunId: task.RunID,
|
|
TaskType: enumsspb.TASK_TYPE_REPLICATION_HISTORY,
|
|
TaskId: task.TaskID,
|
|
Version: task.Version,
|
|
FirstEventId: task.FirstEventID,
|
|
NextEventId: task.NextEventID,
|
|
ScheduledEventId: 0,
|
|
})
|
|
case *tasks.SyncWorkflowStateTask:
|
|
taskInfo = append(taskInfo, &replicationspb.ReplicationTaskInfo{
|
|
NamespaceId: task.NamespaceID,
|
|
WorkflowId: task.WorkflowID,
|
|
RunId: task.RunID,
|
|
TaskType: enumsspb.TASK_TYPE_REPLICATION_SYNC_WORKFLOW_STATE,
|
|
TaskId: task.TaskID,
|
|
Version: task.Version,
|
|
FirstEventId: 0,
|
|
NextEventId: 0,
|
|
ScheduledEventId: 0,
|
|
})
|
|
case *tasks.SyncHSMTask:
|
|
taskInfo = append(taskInfo, &replicationspb.ReplicationTaskInfo{
|
|
NamespaceId: task.NamespaceID,
|
|
WorkflowId: task.WorkflowID,
|
|
RunId: task.RunID,
|
|
TaskType: enumsspb.TASK_TYPE_REPLICATION_SYNC_HSM,
|
|
TaskId: task.TaskID,
|
|
})
|
|
default:
|
|
panic(fmt.Sprintf("Unknown repication task type: %v", task))
|
|
}
|
|
}
|
|
|
|
if len(taskInfo) == 0 {
|
|
return nil, nil, ackLevel, pageToken, nil
|
|
}
|
|
|
|
dlqResponse, err := remoteAdminClient.GetDLQReplicationMessages(
|
|
ctx,
|
|
&adminservice.GetDLQReplicationMessagesRequest{
|
|
TaskInfos: taskInfo,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return nil, nil, ackLevel, nil, err
|
|
}
|
|
|
|
return dlqResponse.ReplicationTasks, taskInfo, ackLevel, pageToken, nil
|
|
}
|
|
|
|
func (r *dlqHandlerImpl) getOrCreateTaskExecutor(clusterName string) (TaskExecutor, error) {
|
|
r.taskExecutorsLock.Lock()
|
|
defer r.taskExecutorsLock.Unlock()
|
|
if executor, ok := r.taskExecutors[clusterName]; ok {
|
|
return executor, nil
|
|
}
|
|
taskExecutor := r.taskExecutorProvider(TaskExecutorParams{
|
|
RemoteCluster: clusterName,
|
|
Shard: r.shard,
|
|
HistoryResender: r.resender,
|
|
DeleteManager: r.deleteManager,
|
|
WorkflowCache: r.workflowCache,
|
|
})
|
|
r.taskExecutors[clusterName] = taskExecutor
|
|
return taskExecutor, nil
|
|
}
|