Scaffold GetTaskListSize method at persistence layer

cadence-workflow · Nov 7, 2023 · 2e97487 · 2e97487
1 parent d30ab5f
commit 2e97487
Show file tree

Hide file tree

Showing 9 changed files with 163 additions and 11 deletions.
diff --git a/common/log/tag/values.go b/common/log/tag/values.go
@@ -222,6 +222,7 @@ var (
 	StoreOperationUpdateTaskList        = storeOperation("update-task-list")
 	StoreOperationListTaskList          = storeOperation("list-task-list")
 	StoreOperationDeleteTaskList        = storeOperation("delete-task-list")
+	StoreOperationGetTaskListSize       = storeOperation("get-task-list-size")
 	StoreOperationStopTaskList          = storeOperation("stop-task-list")
 
 	StoreOperationCreateDomain       = storeOperation("create-domain")

diff --git a/common/metrics/defs.go b/common/metrics/defs.go
@@ -219,6 +219,8 @@ const (
 	PersistenceListTaskListScope
 	// PersistenceDeleteTaskListScope is the metric scope for persistence.TaskManager.DeleteTaskList API
 	PersistenceDeleteTaskListScope
+	// PersistenceGetTaskListSizeScope is the metric scope for persistence.TaskManager.GetTaskListSize API
+	PersistenceGetTaskListSizeScope
 	// PersistenceAppendHistoryEventsScope tracks AppendHistoryEvents calls made by service to persistence layer
 	PersistenceAppendHistoryEventsScope
 	// PersistenceGetWorkflowExecutionHistoryScope tracks GetWorkflowExecutionHistory calls made by service to persistence layer
@@ -1339,6 +1341,7 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
 		PersistenceUpdateTaskListScope:                                 {operation: "UpdateTaskList"},
 		PersistenceListTaskListScope:                                   {operation: "ListTaskList"},
 		PersistenceDeleteTaskListScope:                                 {operation: "DeleteTaskList"},
+		PersistenceGetTaskListSizeScope:                                {operation: "GetTaskListSize"},
 		PersistenceAppendHistoryEventsScope:                            {operation: "AppendHistoryEvents"},
 		PersistenceGetWorkflowExecutionHistoryScope:                    {operation: "GetWorkflowExecutionHistory"},
 		PersistenceDeleteWorkflowExecutionHistoryScope:                 {operation: "DeleteWorkflowExecutionHistory"},

diff --git a/common/mocks/TaskManager.go b/common/mocks/TaskManager.go
diff --git a/common/persistence/dataManagerInterfaces.go b/common/persistence/dataManagerInterfaces.go
@@ -1270,6 +1270,18 @@ type (
 		RangeID      int64
 	}
 
+	GetTaskListSizeRequest struct {
+		DomainID     string
+		DomainName   string
+		TaskListName string
+		TaskListType int
+		AckLevel     int64
+	}
+
+	GetTaskListSizeResponse struct {
+		Size int
+	}
+
 	// CreateTasksRequest is used to create a new task for a workflow exectution
 	CreateTasksRequest struct {
 		TaskListInfo *TaskListInfo
@@ -1793,6 +1805,7 @@ type (
 		UpdateTaskList(ctx context.Context, request *UpdateTaskListRequest) (*UpdateTaskListResponse, error)
 		ListTaskList(ctx context.Context, request *ListTaskListRequest) (*ListTaskListResponse, error)
 		DeleteTaskList(ctx context.Context, request *DeleteTaskListRequest) error
+		GetTaskListSize(ctx context.Context, request *GetTaskListSizeRequest) (*GetTaskListSizeResponse, error)
 		CreateTasks(ctx context.Context, request *CreateTasksRequest) (*CreateTasksResponse, error)
 		GetTasks(ctx context.Context, request *GetTasksRequest) (*GetTasksResponse, error)
 		CompleteTask(ctx context.Context, request *CompleteTaskRequest) error

diff --git a/common/persistence/dataManagerInterfaces_mock.go b/common/persistence/dataManagerInterfaces_mock.go
diff --git a/common/persistence/persistenceErrorInjectionClients.go b/common/persistence/persistenceErrorInjectionClients.go
@@ -1206,6 +1206,31 @@ func (p *taskErrorInjectionPersistenceClient) DeleteTaskList(
 	return persistenceErr
 }
 
+func (p *taskErrorInjectionPersistenceClient) GetTaskListSize(
+	ctx context.Context,
+	request *GetTaskListSizeRequest,
+) (*GetTaskListSizeResponse, error) {
+	fakeErr := generateFakeError(p.errorRate)
+
+	var resp *GetTaskListSizeResponse
+	var persistenceErr error
+	var forwardCall bool
+	if forwardCall = shouldForwardCallToPersistence(fakeErr); forwardCall {
+		resp, persistenceErr = p.persistence.GetTaskListSize(ctx, request)
+	}
+
+	if fakeErr != nil {
+		p.logger.Error(msgInjectedFakeErr,
+			tag.StoreOperationGetTaskListSize,
+			tag.Error(fakeErr),
+			tag.Bool(forwardCall),
+			tag.StoreError(persistenceErr),
+		)
+		return nil, fakeErr
+	}
+	return resp, persistenceErr
+}
+
 func (p *taskErrorInjectionPersistenceClient) Close() {
 	p.persistence.Close()
 }

diff --git a/common/persistence/persistenceMetricClients.go b/common/persistence/persistenceMetricClients.go
@@ -1029,6 +1029,23 @@ func (p *taskPersistenceClient) DeleteTaskList(
 	return p.call(metrics.PersistenceDeleteTaskListScope, op, metrics.DomainTag(request.DomainName))
 }
 
+func (p *taskPersistenceClient) GetTaskListSize(
+	ctx context.Context,
+	request *GetTaskListSizeRequest,
+) (*GetTaskListSizeResponse, error) {
+	var resp *GetTaskListSizeResponse
+	op := func() error {
+		var err error
+		resp, err = p.persistence.GetTaskListSize(ctx, request)
+		return err
+	}
+	err := p.call(metrics.PersistenceGetTaskListSizeScope, op)
+	if err != nil {
+		return nil, err
+	}
+	return resp, nil
+}
+
 func (p *taskPersistenceClient) UpdateTaskList(
 	ctx context.Context,
 	request *UpdateTaskListRequest,

diff --git a/common/persistence/persistenceRateLimitedClients.go b/common/persistence/persistenceRateLimitedClients.go
@@ -680,6 +680,16 @@ func (p *taskRateLimitedPersistenceClient) DeleteTaskList(
 	return p.persistence.DeleteTaskList(ctx, request)
 }
 
+func (p *taskRateLimitedPersistenceClient) GetTaskListSize(
+	ctx context.Context,
+	request *GetTaskListSizeRequest,
+) (*GetTaskListSizeResponse, error) {
+	if ok := p.rateLimiter.Allow(); !ok {
+		return nil, ErrPersistenceLimitExceeded
+	}
+	return p.persistence.GetTaskListSize(ctx, request)
+}
+
 func (p *taskRateLimitedPersistenceClient) Close() {
 	p.persistence.Close()
 }

diff --git a/common/persistence/taskManager.go b/common/persistence/taskManager.go
@@ -26,6 +26,7 @@ import (
 	"context"
 
 	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/types"
 )
 
 type (
@@ -69,6 +70,10 @@ func (t *taskManager) DeleteTaskList(ctx context.Context, request *DeleteTaskLis
 	return t.persistence.DeleteTaskList(ctx, request)
 }
 
+func (t *taskManager) GetTaskListSize(ctx context.Context, request *GetTaskListSizeRequest) (*GetTaskListSizeResponse, error) {
+	return nil, &types.InternalServiceError{Message: "Not yet implemented"}
+}
+
 func (t *taskManager) CreateTasks(ctx context.Context, request *CreateTasksRequest) (*CreateTasksResponse, error) {
 	var internalCreateTasks []*InternalCreateTasksInfo
 	for _, task := range request.Tasks {