Move replication APIs to api package

service/history/api/replication/generate_task.go

@@ -0,0 +1,86 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package replication
+
+import (
+	"context"
+
+	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/common/definition"
+	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/persistence"
+	"go.temporal.io/server/service/history/api"
+	"go.temporal.io/server/service/history/shard"
+	"go.temporal.io/server/service/history/tasks"
+)
+
+func GenerateTask(
+	ctx context.Context,
+	request *historyservice.GenerateLastHistoryReplicationTasksRequest,
+	shard shard.Context,
+	workflowConsistencyChecker api.WorkflowConsistencyChecker,
+) (_ *historyservice.GenerateLastHistoryReplicationTasksResponse, retError error) {
+	namespaceEntry, err := api.GetActiveNamespace(shard, namespace.ID(request.GetNamespaceId()))
+	if err != nil {
+		return nil, err
+	}
+	namespaceID := namespaceEntry.ID()
+
+	wfContext, err := workflowConsistencyChecker.GetWorkflowContext(
+		ctx,
+		nil,
+		api.BypassMutableStateConsistencyPredicate,
+		definition.NewWorkflowKey(
+			namespaceID.String(),
+			request.Execution.WorkflowId,
+			request.Execution.RunId,
+		),
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { wfContext.GetReleaseFn()(retError) }()
+
+	now := shard.GetTimeSource().Now()
+	task, err := wfContext.GetMutableState().GenerateMigrationTasks(now)
+	if err != nil {
+		return nil, err
+	}
+
+	err = shard.AddTasks(ctx, &persistence.AddHistoryTasksRequest{
+		ShardID: shard.GetShardID(),
+		// RangeID is set by shard
+		NamespaceID: string(namespaceID),
+		WorkflowID:  request.Execution.WorkflowId,
+		RunID:       request.Execution.RunId,
+		Tasks: map[tasks.Category][]tasks.Task{
+			tasks.CategoryReplication: {task},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &historyservice.GenerateLastHistoryReplicationTasksResponse{}, nil
+}

service/history/api/replication/get_dlq_tasks.go

@@ -0,0 +1,52 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package replication
+
+import (
+	"context"
+
+	replicationspb "go.temporal.io/server/api/replication/v1"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/service/history/replication"
+	"go.temporal.io/server/service/history/shard"
+)
+
+func GetDLQTasks(
+	ctx context.Context,
+	shard shard.Context,
+	replicationAckMgr replication.AckManager,
+	taskInfos []*replicationspb.ReplicationTaskInfo,
+) ([]*replicationspb.ReplicationTask, error) {
+	tasks := make([]*replicationspb.ReplicationTask, 0, len(taskInfos))
+	for _, taskInfo := range taskInfos {
+		task, err := replicationAckMgr.GetTask(ctx, taskInfo)
+		if err != nil {
+			shard.GetLogger().Error("Failed to fetch DLQ replication messages.", tag.Error(err))
+			return nil, err
+		}
+		tasks = append(tasks, task)
+	}
+	return tasks, nil
+}

service/history/api/replication/get_tasks.go

@@ -0,0 +1,72 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package replication
+
+import (
+	"context"
+	"time"
+
+	replicationspb "go.temporal.io/server/api/replication/v1"
+	"go.temporal.io/server/common/log/tag"
+	"go.temporal.io/server/common/persistence"
+	"go.temporal.io/server/service/history/replication"
+	"go.temporal.io/server/service/history/shard"
+	"go.temporal.io/server/service/history/tasks"
+)
+
+func GetTasks(
+	ctx context.Context,
+	shard shard.Context,
+	replicationAckMgr replication.AckManager,
+	pollingCluster string,
+	ackMessageID int64,
+	ackTimestamp time.Time,
+	queryMessageID int64,
+) (*replicationspb.ReplicationMessages, error) {
+
+	if ackMessageID != persistence.EmptyQueueMessageID {
+		if err := shard.UpdateQueueClusterAckLevel(
+			tasks.CategoryReplication,
+			pollingCluster,
+			tasks.NewImmediateKey(ackMessageID),
+		); err != nil {
+			shard.GetLogger().Error("error updating replication level for shard", tag.Error(err), tag.OperationFailed)
+		}
+		shard.UpdateRemoteClusterInfo(pollingCluster, ackMessageID, ackTimestamp)
+	}
+
+	replicationMessages, err := replicationAckMgr.GetTasks(
+		ctx,
+		pollingCluster,
+		queryMessageID,
+	)
+	if err != nil {
+		shard.GetLogger().Error("Failed to retrieve replication messages.", tag.Error(err))
+		return nil, err
+	}
+
+	shard.GetLogger().Debug("Successfully fetched replication messages.", tag.Counter(len(replicationMessages.ReplicationTasks)))
+	return replicationMessages, nil
+}

service/history/api/replication/status.go

@@ -0,0 +1,58 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package replication
+
+import (
+	"context"
+
+	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/common/primitives/timestamp"
+	"go.temporal.io/server/service/history/replication"
+	"go.temporal.io/server/service/history/shard"
+)
+
+func GetStatus(
+	ctx context.Context,
+	request *historyservice.GetReplicationStatusRequest,
+	shard shard.Context,
+	replicationAckMgr replication.AckManager,
+) (_ *historyservice.ShardReplicationStatus, retError error) {
+	resp := &historyservice.ShardReplicationStatus{
+		ShardId:        shard.GetShardID(),
+		ShardLocalTime: timestamp.TimePtr(shard.GetTimeSource().Now()),
+	}
+
+	maxReplicationTaskId, maxTaskVisibilityTimeStamp := replicationAckMgr.GetMaxTaskInfo()
+	resp.MaxReplicationTaskId = maxReplicationTaskId
+	resp.MaxReplicationTaskVisibilityTime = timestamp.TimePtr(maxTaskVisibilityTimeStamp)
+
+	remoteClusters, handoverNamespaces, err := shard.GetReplicationStatus(request.RemoteClusters)
+	if err != nil {
+		return nil, err
+	}
+	resp.RemoteClusters = remoteClusters
+	resp.HandoverNamespaces = handoverNamespaces
+	return resp, nil
+}

service/history/api/replicationadmin/get_dlq.go

@@ -0,0 +1,62 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package replicationadmin
+
+import (
+	"context"
+
+	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/service/history/consts"
+	"go.temporal.io/server/service/history/replication"
+	"go.temporal.io/server/service/history/shard"
+)
+
+func GetDLQ(
+	ctx context.Context,
+	request *historyservice.GetDLQMessagesRequest,
+	shard shard.Context,
+	replicationDLQHandler replication.DLQHandler,
+) (*historyservice.GetDLQMessagesResponse, error) {
+	_, ok := shard.GetClusterMetadata().GetAllClusterInfo()[request.GetSourceCluster()]
+	if !ok {
+		return nil, consts.ErrUnknownCluster
+	}
+
+	tasks, token, err := replicationDLQHandler.GetMessages(
+		ctx,
+		request.GetSourceCluster(),
+		request.GetInclusiveEndMessageId(),
+		int(request.GetMaximumPageSize()),
+		request.GetNextPageToken(),
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &historyservice.GetDLQMessagesResponse{
+		Type:             request.GetType(),
+		ReplicationTasks: tasks,
+		NextPageToken:    token,
+	}, nil
+}

service/history/api/replicationadmin/merge_dlq.go

@@ -0,0 +1,60 @@
+// The MIT License
+//
+// Copyright (c) 2020 Temporal Technologies Inc.  All rights reserved.
+//
+// Copyright (c) 2020 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package replicationadmin
+
+import (
+	"context"
+
+	"go.temporal.io/server/api/historyservice/v1"
+	"go.temporal.io/server/service/history/consts"
+	"go.temporal.io/server/service/history/replication"
+	"go.temporal.io/server/service/history/shard"
+)
+
+func MergeDLQ(
+	ctx context.Context,
+	request *historyservice.MergeDLQMessagesRequest,
+	shard shard.Context,
+	replicationDLQHandler replication.DLQHandler,
+) (*historyservice.MergeDLQMessagesResponse, error) {
+	_, ok := shard.GetClusterMetadata().GetAllClusterInfo()[request.GetSourceCluster()]
+	if !ok {
+		return nil, consts.ErrUnknownCluster
+	}
+
+	token, err := replicationDLQHandler.MergeMessages(
+		ctx,
+		request.GetSourceCluster(),
+		request.GetInclusiveEndMessageId(),
+		int(request.GetMaximumPageSize()),
+		request.GetNextPageToken(),
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &historyservice.MergeDLQMessagesResponse{
+		NextPageToken: token,
+	}, nil
+}