Skip to content

Commit d06e307

Browse files
feedmeapplesfeedmeapples
authored
Create CLI command to describe Replication Task (#416)
* Create CLI command to describe Replication Task
1 parent b16833c commit d06e307

File tree

10 files changed

+147
-4
lines changed

10 files changed

+147
-4
lines changed

‎common/metrics/defs.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,8 @@ const (
160160
PersistenceCompleteTransferTaskScope
161161
// PersistenceRangeCompleteTransferTaskScope tracks CompleteTransferTasks calls made by service to persistence layer
162162
PersistenceRangeCompleteTransferTaskScope
163+
// PersistenceGetReplicationTaskScope tracks GetReplicationTask calls made by service to persistence layer
164+
PersistenceGetReplicationTaskScope
163165
// PersistenceGetReplicationTasksScope tracks GetReplicationTasks calls made by service to persistence layer
164166
PersistenceGetReplicationTasksScope
165167
// PersistenceCompleteReplicationTaskScope tracks CompleteReplicationTasks calls made by service to persistence layer
@@ -1074,6 +1076,7 @@ var ScopeDefs = map[ServiceIdx]map[int]scopeDefinition{
10741076
PersistenceGetTransferTasksScope: {operation: "GetTransferTasks"},
10751077
PersistenceCompleteTransferTaskScope: {operation: "CompleteTransferTask"},
10761078
PersistenceRangeCompleteTransferTaskScope: {operation: "RangeCompleteTransferTask"},
1079+
PersistenceGetReplicationTaskScope: {operation: "GetReplicationTask"},
10771080
PersistenceGetReplicationTasksScope: {operation: "GetReplicationTasks"},
10781081
PersistenceCompleteReplicationTaskScope: {operation: "CompleteReplicationTask"},
10791082
PersistenceRangeCompleteReplicationTaskScope: {operation: "RangeCompleteReplicationTask"},

‎common/mocks/ExecutionManager.go‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -285,6 +285,29 @@ func (_m *ExecutionManager) RangeCompleteTransferTask(request *persistence.Range
285285
return r0
286286
}
287287

288+
// GetReplicationTask provides a mock function with given fields: request
289+
func (_m *ExecutionManager) GetReplicationTask(request *persistence.GetReplicationTaskRequest) (*persistence.GetReplicationTaskResponse, error) {
290+
ret := _m.Called(request)
291+
292+
var r0 *persistence.GetReplicationTaskResponse
293+
if rf, ok := ret.Get(0).(func(*persistence.GetReplicationTaskRequest) *persistence.GetReplicationTaskResponse); ok {
294+
r0 = rf(request)
295+
} else {
296+
if ret.Get(0) != nil {
297+
r0 = ret.Get(0).(*persistence.GetReplicationTaskResponse)
298+
}
299+
}
300+
301+
var r1 error
302+
if rf, ok := ret.Get(1).(func(*persistence.GetReplicationTaskRequest) error); ok {
303+
r1 = rf(request)
304+
} else {
305+
r1 = ret.Error(1)
306+
}
307+
308+
return r0, r1
309+
}
310+
288311
// GetReplicationTasks provides a mock function with given fields: request
289312
func (_m *ExecutionManager) GetReplicationTasks(request *persistence.GetReplicationTasksRequest) (*persistence.GetReplicationTasksResponse, error) {
290313
ret := _m.Called(request)

‎common/persistence/cassandra/cassandraPersistence.go‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -517,6 +517,16 @@ workflow_state = ? ` +
517517
`and task_id > ? ` +
518518
`and task_id <= ?`
519519

520+
templateGetReplicationTaskQuery = `SELECT replication, replication_encoding ` +
521+
`FROM executions ` +
522+
`WHERE shard_id = ? ` +
523+
`and type = ? ` +
524+
`and namespace_id = ? ` +
525+
`and workflow_id = ? ` +
526+
`and run_id = ? ` +
527+
`and visibility_ts = ? ` +
528+
`and task_id = ? `
529+
520530
templateGetReplicationTasksQuery = `SELECT replication, replication_encoding ` +
521531
`FROM executions ` +
522532
`WHERE shard_id = ? ` +
@@ -1964,6 +1974,33 @@ func (d *cassandraPersistence) GetTransferTasks(request *p.GetTransferTasksReque
19641974
return response, nil
19651975
}
19661976

1977+
func (d *cassandraPersistence) GetReplicationTask(request *p.GetReplicationTaskRequest) (*p.GetReplicationTaskResponse, error) {
1978+
shardID := d.shardID
1979+
taskID := request.TaskID
1980+
query := d.session.Query(templateGetReplicationTaskQuery,
1981+
shardID,
1982+
rowTypeReplicationTask,
1983+
rowTypeReplicationNamespaceID,
1984+
rowTypeReplicationWorkflowID,
1985+
rowTypeReplicationRunID,
1986+
defaultVisibilityTimestamp,
1987+
taskID)
1988+
1989+
var data []byte
1990+
var encoding string
1991+
if err := query.Scan(&data, &encoding); err != nil {
1992+
return nil, convertCommonErrors("GetReplicationTask", err)
1993+
}
1994+
1995+
info, err := serialization.ReplicationTaskInfoFromBlob(data, encoding)
1996+
1997+
if err != nil {
1998+
return nil, convertCommonErrors("GetReplicationTask", err)
1999+
}
2000+
2001+
return &p.GetReplicationTaskResponse{ReplicationTaskInfo: info}, nil
2002+
}
2003+
19672004
func (d *cassandraPersistence) GetReplicationTasks(
19682005
request *p.GetReplicationTasksRequest,
19692006
) (*p.GetReplicationTasksResponse, error) {

‎common/persistence/dataInterfaces.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -790,6 +790,17 @@ type (
790790
NextPageToken []byte
791791
}
792792

793+
// GetReplicationTaskRequest is the request for GetReplicationTask
794+
GetReplicationTaskRequest struct {
795+
ShardID int32
796+
TaskID int64
797+
}
798+
799+
// GetReplicationTaskResponse is the response to GetReplicationTask
800+
GetReplicationTaskResponse struct {
801+
ReplicationTaskInfo *persistenceblobs.ReplicationTaskInfo
802+
}
803+
793804
// GetReplicationTasksRequest is used to read tasks from the replication task queue
794805
GetReplicationTasksRequest struct {
795806
ReadLevel int64
@@ -1345,6 +1356,7 @@ type (
13451356
RangeCompleteTransferTask(request *RangeCompleteTransferTaskRequest) error
13461357

13471358
// Replication task related methods
1359+
GetReplicationTask(request *GetReplicationTaskRequest) (*GetReplicationTaskResponse, error)
13481360
GetReplicationTasks(request *GetReplicationTasksRequest) (*GetReplicationTasksResponse, error)
13491361
CompleteReplicationTask(request *CompleteReplicationTaskRequest) error
13501362
RangeCompleteReplicationTask(request *RangeCompleteReplicationTaskRequest) error

‎common/persistence/executionStore.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -821,6 +821,12 @@ func (m *executionManagerImpl) RangeCompleteTransferTask(
821821
}
822822

823823
// Replication task related methods
824+
func (m *executionManagerImpl) GetReplicationTask(
825+
request *GetReplicationTaskRequest,
826+
) (*GetReplicationTaskResponse, error) {
827+
return m.persistence.GetReplicationTask(request)
828+
}
829+
824830
func (m *executionManagerImpl) GetReplicationTasks(
825831
request *GetReplicationTasksRequest,
826832
) (*GetReplicationTasksResponse, error) {

‎common/persistence/persistenceInterface.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,7 @@ type (
103103
RangeCompleteTransferTask(request *RangeCompleteTransferTaskRequest) error
104104

105105
// Replication task related methods
106+
GetReplicationTask(request *GetReplicationTaskRequest) (*GetReplicationTaskResponse, error)
106107
GetReplicationTasks(request *GetReplicationTasksRequest) (*GetReplicationTasksResponse, error)
107108
CompleteReplicationTask(request *CompleteReplicationTaskRequest) error
108109
RangeCompleteReplicationTask(request *RangeCompleteReplicationTaskRequest) error

‎common/persistence/persistenceMetricClients.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -379,6 +379,20 @@ func (p *workflowExecutionPersistenceClient) GetTransferTasks(request *GetTransf
379379
return response, err
380380
}
381381

382+
func (p *workflowExecutionPersistenceClient) GetReplicationTask(request *GetReplicationTaskRequest) (*GetReplicationTaskResponse, error) {
383+
p.metricClient.IncCounter(metrics.PersistenceGetReplicationTaskScope, metrics.PersistenceRequests)
384+
385+
sw := p.metricClient.StartTimer(metrics.PersistenceGetReplicationTaskScope, metrics.PersistenceLatency)
386+
response, err := p.persistence.GetReplicationTask(request)
387+
sw.Stop()
388+
389+
if err != nil {
390+
p.updateErrorMetric(metrics.PersistenceGetReplicationTaskScope, err)
391+
}
392+
393+
return response, err
394+
}
395+
382396
func (p *workflowExecutionPersistenceClient) GetReplicationTasks(request *GetReplicationTasksRequest) (*GetReplicationTasksResponse, error) {
383397
p.metricClient.IncCounter(metrics.PersistenceGetReplicationTasksScope, metrics.PersistenceRequests)
384398

‎common/persistence/persistenceRateLimitedClients.go‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -302,6 +302,15 @@ func (p *workflowExecutionRateLimitedPersistenceClient) GetTransferTasks(request
302302
return response, err
303303
}
304304

305+
func (p *workflowExecutionRateLimitedPersistenceClient) GetReplicationTask(request *GetReplicationTaskRequest) (*GetReplicationTaskResponse, error) {
306+
if ok := p.rateLimiter.Allow(); !ok {
307+
return nil, ErrPersistenceLimitExceeded
308+
}
309+
310+
response, err := p.persistence.GetReplicationTask(request)
311+
return response, err
312+
}
313+
305314
func (p *workflowExecutionRateLimitedPersistenceClient) GetReplicationTasks(request *GetReplicationTasksRequest) (*GetReplicationTasksResponse, error) {
306315
if ok := p.rateLimiter.Allow(); !ok {
307316
return nil, ErrPersistenceLimitExceeded

‎common/persistence/sql/sqlExecutionManager.go‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -777,6 +777,30 @@ func (m *sqlExecutionManager) RangeCompleteTransferTask(
777777
return nil
778778
}
779779

780+
func (m *sqlExecutionManager) GetReplicationTask(request *persistence.GetReplicationTaskRequest) (*persistence.GetReplicationTaskResponse, error) {
781+
rows, err := m.db.SelectFromReplicationTasks(&sqlplugin.ReplicationTasksFilter{ShardID: int(request.ShardID), TaskID: request.TaskID})
782+
if err != nil {
783+
if err == sql.ErrNoRows {
784+
return nil, serviceerror.NewNotFound(fmt.Sprintf("GetReplicationTask operation failed. Task with ID %v not found. Error: %v", request.TaskID, err))
785+
}
786+
return nil, serviceerror.NewInternal(fmt.Sprintf("GetReplicationTask operation failed. Failed to get record. TaskId: %v. Error: %v", request.TaskID, err))
787+
}
788+
789+
if len(rows) == 0 {
790+
return nil, serviceerror.NewInternal(fmt.Sprintf("GetReplicationTask operation failed. Failed to get record. TaskId: %v", request.TaskID))
791+
}
792+
793+
replicationRow := rows[0]
794+
replicationInfo, err := serialization.ReplicationTaskInfoFromBlob(replicationRow.Data, replicationRow.DataEncoding)
795+
if err != nil {
796+
return nil, err
797+
}
798+
799+
resp := &persistence.GetReplicationTaskResponse{ReplicationTaskInfo: replicationInfo}
800+
801+
return resp, nil
802+
}
803+
780804
func (m *sqlExecutionManager) GetReplicationTasks(
781805
request *p.GetReplicationTasksRequest,
782806
) (*p.GetReplicationTasksResponse, error) {

‎tools/cli/adminCommands.go‎

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -374,17 +374,31 @@ func AdminDescribeTask(c *cli.Context) {
374374
sid := getRequiredIntOption(c, FlagShardID)
375375
tid := getRequiredIntOption(c, FlagTaskID)
376376
vis := getRequiredInt64Option(c, FlagTaskVisibilityTimestamp)
377+
category := commongenpb.TaskCategory(c.Int(FlagTaskType))
377378

378379
pFactory := CreatePersistenceFactory(c)
379380
executionManager, err := pFactory.NewExecutionManager(sid)
380381
if err != nil {
381382
ErrorAndExit("Failed to initialize execution manager", err)
382383
}
383384

384-
req := &persistence.GetTimerTaskRequest{ShardID: int32(sid), TaskID: int64(tid), VisibilityTimestamp: time.Unix(0, vis)}
385-
timerTask, err := executionManager.GetTimerTask(req)
386-
387-
prettyPrintJSONObject(timerTask)
385+
if category == commongenpb.TaskCategory_TaskCategory_Timer {
386+
req := &persistence.GetTimerTaskRequest{ShardID: int32(sid), TaskID: int64(tid), VisibilityTimestamp: time.Unix(0, vis)}
387+
task, err := executionManager.GetTimerTask(req)
388+
if err != nil {
389+
ErrorAndExit("Failed to get Timer Task", err)
390+
}
391+
prettyPrintJSONObject(task)
392+
} else if category == commongenpb.TaskCategory_TaskCategory_Replication {
393+
req := &persistence.GetReplicationTaskRequest{ShardID: int32(sid), TaskID: int64(tid)}
394+
task, err := executionManager.GetReplicationTask(req)
395+
if err != nil {
396+
ErrorAndExit("Failed to get Replication Task", err)
397+
}
398+
prettyPrintJSONObject(task)
399+
} else {
400+
ErrorAndExit("Failed to describe task", fmt.Errorf("Unrecognized task type, task_type=%v", category))
401+
}
388402
}
389403

390404
// AdminRemoveTask describes history host

0 commit comments

Comments
 (0)