Add missing metrics to replication task (cadence-workflow#1682)

* Add missing metrics to replication task
zshni · Apr 9, 2019 · b6f9684 · b6f9684
1 parent 2dc8575
commit b6f9684
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 5 deletions.
diff --git a/service/worker/replicator/replicationTask.go b/service/worker/replicator/replicationTask.go
@@ -41,11 +41,13 @@ import (
 
 type (
 	workflowReplicationTask struct {
-		partitionID definition.WorkflowIdentifier
-		taskID      int64
-		attempt     int
-		kafkaMsg    messaging.Message
-		logger      bark.Logger
+		metricsScope int
+		startTime    time.Time
+		partitionID  definition.WorkflowIdentifier
+		taskID       int64
+		attempt      int
+		kafkaMsg     messaging.Message
+		logger       bark.Logger
 
 		config              *Config
 		historyClient       history.Client
@@ -76,6 +78,8 @@ func newActivityReplicationTask(task *replicator.ReplicationTask, msg messaging.
 	attr := task.SyncActicvityTaskAttributes
 	return &activityReplicationTask{
 		workflowReplicationTask: workflowReplicationTask{
+			metricsScope: metrics.SyncActivityTaskScope,
+			startTime:    time.Now(),
 			partitionID: definition.NewWorkflowIdentifier(
 				attr.GetDomainId(), attr.GetWorkflowId(), attr.GetRunId(),
 			),
@@ -117,6 +121,8 @@ func newHistoryReplicationTask(task *replicator.ReplicationTask, msg messaging.M
 	attr := task.HistoryTaskAttributes
 	return &historyReplicationTask{
 		workflowReplicationTask: workflowReplicationTask{
+			metricsScope: metrics.HistoryReplicationTaskScope,
+			startTime:    time.Now(),
 			partitionID: definition.NewWorkflowIdentifier(
 				attr.GetDomainId(), attr.GetWorkflowId(), attr.GetRunId(),
 			),
@@ -262,6 +268,9 @@ func (t *workflowReplicationTask) HashCode() uint32 {
 }
 
 func (t *workflowReplicationTask) Ack() {
+	t.metricsClient.IncCounter(t.metricsScope, metrics.ReplicatorMessages)
+	t.metricsClient.RecordTimer(t.metricsScope, metrics.ReplicatorLatency, time.Now().Sub(t.startTime))
+
 	// the underlying implementation will not return anything other than nil
 	// do logging just in case
 	err := t.kafkaMsg.Ack()
@@ -271,6 +280,9 @@ func (t *workflowReplicationTask) Ack() {
 }
 
 func (t *workflowReplicationTask) Nack() {
+	t.metricsClient.IncCounter(t.metricsScope, metrics.ReplicatorMessages)
+	t.metricsClient.RecordTimer(t.metricsScope, metrics.ReplicatorLatency, time.Now().Sub(t.startTime))
+
 	// the underlying implementation will not return anything other than nil
 	// do logging just in case
 	err := t.kafkaMsg.Nack()

diff --git a/service/worker/replicator/replicationTask_test.go b/service/worker/replicator/replicationTask_test.go
@@ -152,6 +152,8 @@ func (s *activityReplicationTaskSuite) TestNewActivityReplicationTask() {
 	s.Equal(
 		&activityReplicationTask{
 			workflowReplicationTask: workflowReplicationTask{
+				metricsScope: metrics.SyncActivityTaskScope,
+				startTime:    task.startTime,
 				partitionID: definition.NewWorkflowIdentifier(
 					replicationAttr.GetDomainId(),
 					replicationAttr.GetWorkflowId(),
@@ -304,6 +306,8 @@ func (s *historyReplicationTaskSuite) TestNewHistoryReplicationTask() {
 	s.Equal(
 		&historyReplicationTask{
 			workflowReplicationTask: workflowReplicationTask{
+				metricsScope: metrics.HistoryReplicationTaskScope,
+				startTime:    task.startTime,
 				partitionID: definition.NewWorkflowIdentifier(
 					replicationAttr.GetDomainId(),
 					replicationAttr.GetWorkflowId(),