Create CLI admin command to describe Transfer Task (#412)

feedmeapples · web-flow · commit 1e7c2809e0f1 · 2020-06-05T10:54:16.000-07:00
diff --git a/common/metrics/defs.go b/common/metrics/defs.go
@@ -154,6 +154,8 @@ const (
 	PersistenceGetCurrentExecutionScope
 	// PersistenceListConcreteExecutionsScope tracks ListConcreteExecutions calls made by service to persistence layer
 	PersistenceListConcreteExecutionsScope
+	// PersistenceGetTransferTaskScope tracks GetTransferTask calls made by service to persistence layer
+	PersistenceGetTransferTaskScope
 	// PersistenceGetTransferTasksScope tracks GetTransferTasks calls made by service to persistence layer
 	PersistenceGetTransferTasksScope
 	// PersistenceCompleteTransferTaskScope tracks CompleteTransferTasks calls made by service to persistence layer
@@ -1073,6 +1075,7 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
 		PersistenceDeleteCurrentWorkflowExecutionScope:           {operation: "DeleteCurrentWorkflowExecution"},
 		PersistenceGetCurrentExecutionScope:                      {operation: "GetCurrentExecution"},
 		PersistenceListConcreteExecutionsScope:                   {operation: "ListConcreteExecutions"},
+		PersistenceGetTransferTaskScope:                          {operation: "GetTransferTask"},
 		PersistenceGetTransferTasksScope:                         {operation: "GetTransferTasks"},
 		PersistenceCompleteTransferTaskScope:                     {operation: "CompleteTransferTask"},
 		PersistenceRangeCompleteTransferTaskScope:                {operation: "RangeCompleteTransferTask"},
diff --git a/common/mocks/ExecutionManager.go b/common/mocks/ExecutionManager.go
@@ -234,6 +234,29 @@ func (_m *ExecutionManager) ListConcreteExecutions(request *persistence.ListConc
 	return r0, r1
 }
 
+// GetTransferTask provides a mock function with given fields: request
+func (_m *ExecutionManager) GetTransferTask(request *persistence.GetTransferTaskRequest) (*persistence.GetTransferTaskResponse, error) {
+	ret := _m.Called(request)
+
+	var r0 *persistence.GetTransferTaskResponse
+	if rf, ok := ret.Get(0).(func(*persistence.GetTransferTaskRequest) *persistence.GetTransferTaskResponse); ok {
+		r0 = rf(request)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*persistence.GetTransferTaskResponse)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(*persistence.GetTransferTaskRequest) error); ok {
+		r1 = rf(request)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetTransferTasks provides a mock function with given fields: request
 func (_m *ExecutionManager) GetTransferTasks(request *persistence.GetTransferTasksRequest) (*persistence.GetTransferTasksResponse, error) {
 	ret := _m.Called(request)
diff --git a/common/persistence/cassandra/cassandraPersistence.go b/common/persistence/cassandra/cassandraPersistence.go
@@ -506,6 +506,16 @@ workflow_state = ? ` +
 		`and visibility_ts = ? ` +
 		`and task_id = ? `
 
+	templateGetTransferTaskQuery = `SELECT transfer, transfer_encoding ` +
+		`FROM executions ` +
+		`WHERE shard_id = ? ` +
+		`and type = ? ` +
+		`and namespace_id = ? ` +
+		`and workflow_id = ? ` +
+		`and run_id = ? ` +
+		`and visibility_ts = ? ` +
+		`and task_id = ? `
+
 	templateGetTransferTasksQuery = `SELECT transfer, transfer_encoding ` +
 		`FROM executions ` +
 		`WHERE shard_id = ? ` +
@@ -1932,6 +1942,33 @@ func (d *cassandraPersistence) ListConcreteExecutions(
 	return response, nil
 }
 
+func (d *cassandraPersistence) GetTransferTask(request *p.GetTransferTaskRequest) (*p.GetTransferTaskResponse, error) {
+	shardID := d.shardID
+	taskID := request.TaskID
+	query := d.session.Query(templateGetTransferTaskQuery,
+		shardID,
+		rowTypeTransferTask,
+		rowTypeTransferNamespaceID,
+		rowTypeTransferWorkflowID,
+		rowTypeTransferRunID,
+		defaultVisibilityTimestamp,
+		taskID)
+
+	var data []byte
+	var encoding string
+	if err := query.Scan(&data, &encoding); err != nil {
+		return nil, convertCommonErrors("GetTransferTask", err)
+	}
+
+	info, err := serialization.TransferTaskInfoFromBlob(data, encoding)
+
+	if err != nil {
+		return nil, convertCommonErrors("GetTransferTask", err)
+	}
+
+	return &p.GetTransferTaskResponse{TransferTaskInfo: info}, nil
+}
+
 func (d *cassandraPersistence) GetTransferTasks(request *p.GetTransferTasksRequest) (*p.GetTransferTasksResponse, error) {
 
 	// Reading transfer tasks need to be quorum level consistent, otherwise we could loose task
diff --git a/common/persistence/dataInterfaces.go b/common/persistence/dataInterfaces.go
@@ -776,6 +776,17 @@ type (
 		RunID       string
 	}
 
+	// GetTransferTaskRequest is the request for GetTransferTask
+	GetTransferTaskRequest struct {
+		ShardID int32
+		TaskID  int64
+	}
+
+	// GetTransferTaskResponse is the response to GetTransferTask
+	GetTransferTaskResponse struct {
+		TransferTaskInfo *persistenceblobs.TransferTaskInfo
+	}
+
 	// GetTransferTasksRequest is used to read tasks from the transfer task queue
 	GetTransferTasksRequest struct {
 		ReadLevel     int64
@@ -1351,6 +1362,7 @@ type (
 		GetCurrentExecution(request *GetCurrentExecutionRequest) (*GetCurrentExecutionResponse, error)
 
 		// Transfer task related methods
+		GetTransferTask(request *GetTransferTaskRequest) (*GetTransferTaskResponse, error)
 		GetTransferTasks(request *GetTransferTasksRequest) (*GetTransferTasksResponse, error)
 		CompleteTransferTask(request *CompleteTransferTaskRequest) error
 		RangeCompleteTransferTask(request *RangeCompleteTransferTaskRequest) error
diff --git a/common/persistence/executionStore.go b/common/persistence/executionStore.go
@@ -802,6 +802,12 @@ func (m *executionManagerImpl) ListConcreteExecutions(
 }
 
 // Transfer task related methods
+func (m *executionManagerImpl) GetTransferTask(
+	request *GetTransferTaskRequest,
+) (*GetTransferTaskResponse, error) {
+	return m.persistence.GetTransferTask(request)
+}
+
 func (m *executionManagerImpl) GetTransferTasks(
 	request *GetTransferTasksRequest,
 ) (*GetTransferTasksResponse, error) {
diff --git a/common/persistence/persistenceInterface.go b/common/persistence/persistenceInterface.go
@@ -98,6 +98,7 @@ type (
 		GetCurrentExecution(request *GetCurrentExecutionRequest) (*GetCurrentExecutionResponse, error)
 
 		// Transfer task related methods
+		GetTransferTask(request *GetTransferTaskRequest) (*GetTransferTaskResponse, error)
 		GetTransferTasks(request *GetTransferTasksRequest) (*GetTransferTasksResponse, error)
 		CompleteTransferTask(request *CompleteTransferTaskRequest) error
 		RangeCompleteTransferTask(request *RangeCompleteTransferTaskRequest) error
diff --git a/common/persistence/persistenceMetricClients.go b/common/persistence/persistenceMetricClients.go
@@ -365,6 +365,20 @@ func (p *workflowExecutionPersistenceClient) ListConcreteExecutions(request *Lis
 	return response, err
 }
 
+func (p *workflowExecutionPersistenceClient) GetTransferTask(request *GetTransferTaskRequest) (*GetTransferTaskResponse, error) {
+	p.metricClient.IncCounter(metrics.PersistenceGetTransferTaskScope, metrics.PersistenceRequests)
+
+	sw := p.metricClient.StartTimer(metrics.PersistenceGetTransferTaskScope, metrics.PersistenceLatency)
+	response, err := p.persistence.GetTransferTask(request)
+	sw.Stop()
+
+	if err != nil {
+		p.updateErrorMetric(metrics.PersistenceGetTransferTaskScope, err)
+	}
+
+	return response, err
+}
+
 func (p *workflowExecutionPersistenceClient) GetTransferTasks(request *GetTransferTasksRequest) (*GetTransferTasksResponse, error) {
 	p.metricClient.IncCounter(metrics.PersistenceGetTransferTasksScope, metrics.PersistenceRequests)
 
diff --git a/common/persistence/persistenceRateLimitedClients.go b/common/persistence/persistenceRateLimitedClients.go
@@ -293,6 +293,15 @@ func (p *workflowExecutionRateLimitedPersistenceClient) ListConcreteExecutions(r
 	return response, err
 }
 
+func (p *workflowExecutionRateLimitedPersistenceClient) GetTransferTask(request *GetTransferTaskRequest) (*GetTransferTaskResponse, error) {
+	if ok := p.rateLimiter.Allow(); !ok {
+		return nil, ErrPersistenceLimitExceeded
+	}
+
+	response, err := p.persistence.GetTransferTask(request)
+	return response, err
+}
+
 func (p *workflowExecutionRateLimitedPersistenceClient) GetTransferTasks(request *GetTransferTasksRequest) (*GetTransferTasksResponse, error) {
 	if ok := p.rateLimiter.Allow(); !ok {
 		return nil, ErrPersistenceLimitExceeded
diff --git a/common/persistence/sql/sqlExecutionManager.go b/common/persistence/sql/sqlExecutionManager.go
@@ -728,6 +728,30 @@ func (m *sqlExecutionManager) ListConcreteExecutions(
 	return nil, serviceerror.NewUnimplemented("ListConcreteExecutions is not implemented")
 }
 
+func (m *sqlExecutionManager) GetTransferTask(request *persistence.GetTransferTaskRequest) (*persistence.GetTransferTaskResponse, error) {
+	rows, err := m.db.SelectFromTransferTasks(&sqlplugin.TransferTasksFilter{ShardID: int(request.ShardID), TaskID: &request.TaskID})
+	if err != nil {
+		if err == sql.ErrNoRows {
+			return nil, serviceerror.NewNotFound(fmt.Sprintf("GetTransferTask operation failed. Task with ID %v not found. Error: %v", request.TaskID, err))
+		}
+		return nil, serviceerror.NewInternal(fmt.Sprintf("GetTransferTask operation failed. Failed to get record. TaskId: %v. Error: %v", request.TaskID, err))
+	}
+
+	if len(rows) == 0 {
+		return nil, serviceerror.NewInternal(fmt.Sprintf("GetTransferTask operation failed. Failed to get record. TaskId: %v", request.TaskID))
+	}
+
+	transferRow := rows[0]
+	transferInfo, err := serialization.TransferTaskInfoFromBlob(transferRow.Data, transferRow.DataEncoding)
+	if err != nil {
+		return nil, err
+	}
+
+	resp := &persistence.GetTransferTaskResponse{TransferTaskInfo: transferInfo}
+
+	return resp, nil
+}
+
 func (m *sqlExecutionManager) GetTransferTasks(
 	request *p.GetTransferTasksRequest,
 ) (*p.GetTransferTasksResponse, error) {
diff --git a/tools/cli/adminCommands.go b/tools/cli/adminCommands.go
@@ -373,7 +373,6 @@ func AdminGetShardID(c *cli.Context) {
 func AdminDescribeTask(c *cli.Context) {
 	sid := getRequiredIntOption(c, FlagShardID)
 	tid := getRequiredIntOption(c, FlagTaskID)
-	vis := getRequiredInt64Option(c, FlagTaskVisibilityTimestamp)
 	category := commongenpb.TaskCategory(c.Int(FlagTaskType))
 
 	pFactory := CreatePersistenceFactory(c)
@@ -383,6 +382,7 @@ func AdminDescribeTask(c *cli.Context) {
 	}
 
 	if category == commongenpb.TaskCategory_TaskCategory_Timer {
+		vis := getRequiredInt64Option(c, FlagTaskVisibilityTimestamp)
 		req := &persistence.GetTimerTaskRequest{ShardID: int32(sid), TaskID: int64(tid), VisibilityTimestamp: time.Unix(0, vis)}
 		task, err := executionManager.GetTimerTask(req)
 		if err != nil {
@@ -396,6 +396,13 @@ func AdminDescribeTask(c *cli.Context) {
 			ErrorAndExit("Failed to get Replication Task", err)
 		}
 		prettyPrintJSONObject(task)
+	} else if category == commongenpb.TaskCategory_TaskCategory_Transfer {
+		req := &persistence.GetTransferTaskRequest{ShardID: int32(sid), TaskID: int64(tid)}
+		task, err := executionManager.GetTransferTask(req)
+		if err != nil {
+			ErrorAndExit("Failed to get Transfer Task", err)
+		}
+		prettyPrintJSONObject(task)
 	} else {
 		ErrorAndExit("Failed to describe task", fmt.Errorf("Unrecognized task type, task_type=%v", category))
 	}
