Fix admin remove task command for timer queue (#3157)

* Remove DeleteTask method from persistence interface * Remove related metrics * Use CompleteTask method of executionManger to delete task * Add visibility timestamp tag when initializing logger for task * Add visibility timestamp flag to admin remove task command
uber · Apr 10, 2020 · 19c32ad · 19c32ad
1 parent 3560992
commit 19c32ad
Show file tree

Hide file tree

Showing 20 changed files with 265 additions and 159 deletions.
diff --git a/.gen/go/shared/shared.go b/.gen/go/shared/shared.go
diff --git a/common/constants.go b/common/constants.go
@@ -139,3 +139,20 @@ const (
 	// AdvancedVisibilityWritingModeDual means write to both normal visibility and advanced visibility store
 	AdvancedVisibilityWritingModeDual = "dual"
 )
+
+// DomainDataKeyForManagedFailover is key of DomainData for managed failover
+const DomainDataKeyForManagedFailover = "IsManagedByCadence"
+
+type (
+	// TaskType is the enum for representing different task types
+	TaskType int
+)
+
+const (
+	// TaskTypeTransfer is the task type for transfer task
+	TaskTypeTransfer TaskType = iota + 2 // starting from 2 here to be consistent with the row type define for cassandra
+	// TaskTypeTimer is the task type for timer task
+	TaskTypeTimer
+	// TaskTypeReplication is the task type for replication task
+	TaskTypeReplication
+)
diff --git a/common/log/tag/tags.go b/common/log/tag/tags.go
@@ -465,6 +465,11 @@ func TaskType(taskType int) Tag {
 	return newInt("queue-task-type", taskType)
 }
 
+// TaskVisibilityTimestamp returns tag for task visibilityTimestamp
+func TaskVisibilityTimestamp(timestamp int64) Tag {
+	return newInt64("queue-task-visibility-timestamp", timestamp)
+}
+
 // NumberProcessed returns tag for NumberProcessed
 func NumberProcessed(n int) Tag {
 	return newInt("number-processed", n)

diff --git a/common/metrics/defs.go b/common/metrics/defs.go
@@ -148,8 +148,6 @@ const (
 	PersistenceDeleteWorkflowExecutionScope
 	// PersistenceDeleteCurrentWorkflowExecutionScope tracks DeleteCurrentWorkflowExecution calls made by service to persistence layer
 	PersistenceDeleteCurrentWorkflowExecutionScope
-	// PersistenceDeleteTaskScope tracks RemoveTask calls made by service to persistence layer
-	PersistenceDeleteTaskScope
 	// PersistenceGetCurrentExecutionScope tracks GetCurrentExecution calls made by service to persistence layer
 	PersistenceGetCurrentExecutionScope
 	// PersistenceGetTransferTasksScope tracks GetTransferTasks calls made by service to persistence layer
@@ -1053,7 +1051,6 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
 		PersistenceResetWorkflowExecutionScope:                   {operation: "ResetWorkflowExecution"},
 		PersistenceDeleteWorkflowExecutionScope:                  {operation: "DeleteWorkflowExecution"},
 		PersistenceDeleteCurrentWorkflowExecutionScope:           {operation: "DeleteCurrentWorkflowExecution"},
-		PersistenceDeleteTaskScope:                               {operation: "PersistenceDelete"},
 		PersistenceGetCurrentExecutionScope:                      {operation: "GetCurrentExecution"},
 		PersistenceGetTransferTasksScope:                         {operation: "GetTransferTasks"},
 		PersistenceCompleteTransferTaskScope:                     {operation: "CompleteTransferTask"},

diff --git a/common/mocks/ExecutionManager.go b/common/mocks/ExecutionManager.go
@@ -20,8 +20,10 @@
 
 package mocks
 
-import "github.com/uber/cadence/common/persistence"
-import "github.com/stretchr/testify/mock"
+import (
+	"github.com/stretchr/testify/mock"
+	"github.com/uber/cadence/common/persistence"
+)
 
 // ExecutionManager mock implementation
 type ExecutionManager struct {
@@ -153,20 +155,6 @@ func (_m *ExecutionManager) ResetWorkflowExecution(request *persistence.ResetWor
 	return r0
 }
 
-// DeleteTask provides a mock function with given fields: request
-func (_m *ExecutionManager) DeleteTask(request *persistence.DeleteTaskRequest) error {
-	ret := _m.Called(request)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(executionRequest *persistence.DeleteTaskRequest) error); ok {
-		r0 = rf(request)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // DeleteWorkflowExecution provides a mock function with given fields: request
 func (_m *ExecutionManager) DeleteWorkflowExecution(request *persistence.DeleteWorkflowExecutionRequest) error {
 	ret := _m.Called(request)

diff --git a/common/persistence/cassandra/cassandraPersistence.go b/common/persistence/cassandra/cassandraPersistence.go
@@ -25,12 +25,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/uber/cadence/common/cassandra"
-
 	"github.com/gocql/gocql"
 
 	workflow "github.com/uber/cadence/.gen/go/shared"
 	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/cassandra"
 	"github.com/uber/cadence/common/log"
 	p "github.com/uber/cadence/common/persistence"
 	"github.com/uber/cadence/common/service/config"
@@ -1942,53 +1941,6 @@ func (d *cassandraPersistence) assertNotCurrentExecution(
 	return nil
 }
 
-func (d *cassandraPersistence) DeleteTask(request *p.DeleteTaskRequest) error {
-	var domainID, workflowID, runID string
-	switch request.Type {
-	case rowTypeTransferTask:
-		domainID = rowTypeTransferDomainID
-		workflowID = rowTypeTransferWorkflowID
-		runID = rowTypeTransferRunID
-
-	case rowTypeTimerTask:
-		domainID = rowTypeTimerDomainID
-		workflowID = rowTypeTimerWorkflowID
-		runID = rowTypeTimerRunID
-
-	case rowTypeReplicationTask:
-		domainID = rowTypeReplicationDomainID
-		workflowID = rowTypeReplicationWorkflowID
-		runID = rowTypeReplicationRunID
-
-	default:
-		return fmt.Errorf("DeleteTask type id is not one of 2 (transfer task), 3 (timer task), 4 (replication task) ")
-
-	}
-
-	query := d.session.Query(templateDeleteWorkflowExecutionMutableStateQuery,
-		request.ShardID,
-		request.Type,
-		domainID,
-		workflowID,
-		runID,
-		defaultVisibilityTimestamp,
-		request.TaskID)
-
-	err := query.Exec()
-	if err != nil {
-		if isThrottlingError(err) {
-			return &workflow.ServiceBusyError{
-				Message: fmt.Sprintf("DeleteTask operation failed. Error: %v", err),
-			}
-		}
-		return &workflow.InternalServiceError{
-			Message: fmt.Sprintf("DeleteTask operation failed. Error: %v", err),
-		}
-	}
-
-	return nil
-}
-
 func (d *cassandraPersistence) DeleteWorkflowExecution(request *p.DeleteWorkflowExecutionRequest) error {
 	query := d.session.Query(templateDeleteWorkflowExecutionMutableStateQuery,
 		d.shardID,

diff --git a/common/persistence/dataInterfaces.go b/common/persistence/dataInterfaces.go
@@ -25,12 +25,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/uber/cadence/common/checksum"
-
 	"github.com/pborman/uuid"
 
 	workflow "github.com/uber/cadence/.gen/go/shared"
 	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/checksum"
 	"github.com/uber/cadence/common/codec"
 )
 
@@ -926,14 +925,6 @@ type (
 		RunID      string
 	}
 
-	// DeleteTaskRequest is used to detele a task that corrupted and need to be removed
-	// 	e.g. corrupted history event batch, eventID is not continouous
-	DeleteTaskRequest struct {
-		TaskID  int64
-		Type    int
-		ShardID int
-	}
-
 	// DeleteCurrentWorkflowExecutionRequest is used to delete the current workflow execution
 	DeleteCurrentWorkflowExecutionRequest struct {
 		DomainID   string
@@ -1500,9 +1491,6 @@ type (
 		GetTimerIndexTasks(request *GetTimerIndexTasksRequest) (*GetTimerIndexTasksResponse, error)
 		CompleteTimerTask(request *CompleteTimerTaskRequest) error
 		RangeCompleteTimerTask(request *RangeCompleteTimerTaskRequest) error
-
-		// Remove Task due to corrupted data
-		DeleteTask(request *DeleteTaskRequest) error
 	}
 
 	// ExecutionManagerFactory creates an instance of ExecutionManager for a given shard

diff --git a/common/persistence/executionStore.go b/common/persistence/executionStore.go
@@ -760,12 +760,6 @@ func (m *executionManagerImpl) DeserializeVersionHistories(
 	return NewVersionHistoriesFromThrift(versionHistories), nil
 }
 
-func (m *executionManagerImpl) DeleteTask(
-	request *DeleteTaskRequest,
-) error {
-	return m.persistence.DeleteTask(request)
-}
-
 func (m *executionManagerImpl) DeleteWorkflowExecution(
 	request *DeleteWorkflowExecutionRequest,
 ) error {

diff --git a/common/persistence/persistenceInterface.go b/common/persistence/persistenceInterface.go
@@ -88,9 +88,6 @@ type (
 		GetTimerIndexTasks(request *GetTimerIndexTasksRequest) (*GetTimerIndexTasksResponse, error)
 		CompleteTimerTask(request *CompleteTimerTaskRequest) error
 		RangeCompleteTimerTask(request *RangeCompleteTimerTaskRequest) error
-
-		// Remove corrupted task
-		DeleteTask(request *DeleteTaskRequest) error
 	}
 
 	// HistoryStore is to manager workflow history events

diff --git a/common/persistence/persistenceMetricClients.go b/common/persistence/persistenceMetricClients.go
@@ -520,19 +520,6 @@ func (p *workflowExecutionPersistenceClient) RangeCompleteTimerTask(request *Ran
 	return err
 }
 
-func (p *workflowExecutionPersistenceClient) DeleteTask(request *DeleteTaskRequest) error {
-	p.metricClient.IncCounter(metrics.PersistenceDeleteTaskScope, metrics.PersistenceRequests)
-	sw := p.metricClient.StartTimer(metrics.PersistenceDeleteTaskScope, metrics.PersistenceLatency)
-	err := p.persistence.DeleteTask(request)
-	sw.Stop()
-
-	if err != nil {
-		p.updateErrorMetric(metrics.PersistenceRangeCompleteTimerTaskScope, err)
-	}
-
-	return err
-}
-
 func (p *workflowExecutionPersistenceClient) updateErrorMetric(scope int, err error) {
 	switch err.(type) {
 	case *WorkflowExecutionAlreadyStartedError: