Add latency metrics for each visibility message to ES

common/metrics/defs.go

@@ -1330,7 +1330,9 @@ const (
 	ESProcessorRetries
 	ESProcessorFailures
 	ESProcessorCorruptedData
+	ESProcessorProcessMsgLatency
 	IndexProcessorCorruptedData
+	IndexProcessorProcessMsgLatency
 	ArchiverNonRetryableErrorCount
 	ArchiverSkipUploadCount
 	ArchiverHistoryMutatedCount
@@ -1576,7 +1578,9 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		ESProcessorRetries:                                     {metricName: "es_processor_retries"},
 		ESProcessorFailures:                                    {metricName: "es_processor_errors"},
 		ESProcessorCorruptedData:                               {metricName: "es_processor_corrupted_data"},
+		ESProcessorProcessMsgLatency:                           {metricName: "es_processor_process_msg_latency", metricType: Timer},
 		IndexProcessorCorruptedData:                            {metricName: "index_processor_corrupted_data"},
+		IndexProcessorProcessMsgLatency:                        {metricName: "index_processor_process_msg_latency", metricType: Timer},
 		ArchiverNonRetryableErrorCount:                         {metricName: "archiver_non_retryable_error"},
 		ArchiverSkipUploadCount:                                {metricName: "archiver_skip_upload"},
 		ArchiverHistoryMutatedCount:                            {metricName: "archiver_history_mutated"},

common/metrics/fake.go

@@ -0,0 +1,36 @@
+// Copyright (c) 2017 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package metrics
+
+import (
+	"github.com/uber-go/tally"
+	"time"
+)
+
+type nopStopwatchRecorder struct{}
+
+// RecordStopwatch is a nop impl for replay mode
+func (n *nopStopwatchRecorder) RecordStopwatch(stopwatchStart time.Time) {}
+
+// NopStopwatch return a fake tally stop watch
+func NopStopwatch() tally.Stopwatch {
+	return tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})
+}

service/worker/archiver/archiver_test.go

@@ -26,11 +26,9 @@ import (
 	"fmt"
 	"math/rand"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
-	"github.com/uber-go/tally"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/metrics"
 	mmocks "github.com/uber/cadence/common/metrics/mocks"
@@ -61,7 +59,7 @@ func (s *archiverSuite) SetupSuite() {
 
 func (s *archiverSuite) SetupTest() {
 	archiverTestMetrics = &mmocks.Client{}
-	archiverTestMetrics.On("StartTimer", mock.Anything, mock.Anything).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{}))
+	archiverTestMetrics.On("StartTimer", mock.Anything, mock.Anything).Return(metrics.NopStopwatch())
 	archiverTestLogger = &log.MockLogger{}
 	archiverTestLogger.On("WithTags", mock.Anything).Return(archiverTestLogger)
 }

service/worker/archiver/pump_test.go

@@ -27,7 +27,6 @@ import (
 
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/suite"
-	"github.com/uber-go/tally"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/metrics"
 	mmocks "github.com/uber/cadence/common/metrics/mocks"
@@ -58,7 +57,7 @@ func (s *pumpSuite) SetupSuite() {
 
 func (s *pumpSuite) SetupTest() {
 	pumpTestMetrics = &mmocks.Client{}
-	pumpTestMetrics.On("StartTimer", mock.Anything, mock.Anything).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})).Once()
+	pumpTestMetrics.On("StartTimer", mock.Anything, mock.Anything).Return(metrics.NopStopwatch()).Once()
 	pumpTestLogger = &log.MockLogger{}
 }
 

service/worker/archiver/replay_metrics_client.go

@@ -67,7 +67,7 @@ func (r *replayMetricsClient) AddCounter(scope int, counter int, delta int64) {
 // StartTimer starts a timer for the given metric name. Time will be recorded when stopwatch is stopped.
 func (r *replayMetricsClient) StartTimer(scope int, timer int) tally.Stopwatch {
 	if workflow.IsReplaying(r.ctx) {
-		return nopStopwatch()
+		return metrics.NopStopwatch()
 	}
 	return r.client.StartTimer(scope, timer)
 }
@@ -145,12 +145,3 @@ func (r *replayMetricsScope) UpdateGauge(gauge int, value float64) {
 func (r *replayMetricsScope) Tagged(tags ...metrics.Tag) metrics.Scope {
 	return NewReplayMetricsScope(r.scope.Tagged(tags...), r.ctx)
 }
-
-type nopStopwatchRecorder struct{}
-
-// RecordStopwatch is a nop impl for replay mode
-func (n *nopStopwatchRecorder) RecordStopwatch(stopwatchStart time.Time) {}
-
-func nopStopwatch() tally.Stopwatch {
-	return tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})
-}

service/worker/archiver/workflow_test.go

@@ -22,10 +22,8 @@ package archiver
 
 import (
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/suite"
-	"github.com/uber-go/tally"
 	"github.com/uber/cadence/common/log"
 	"github.com/uber/cadence/common/log/loggerimpl"
 	"github.com/uber/cadence/common/metrics"
@@ -71,8 +69,8 @@ func (s *workflowSuite) SetupTest() {
 
 func (s *workflowSuite) TestArchivalWorkflow_Fail_HashesDoNotEqual() {
 	workflowTestMetrics.On("IncCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverWorkflowStartedCount).Once()
-	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.CadenceLatency).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})).Once()
-	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverHandleAllRequestsLatency).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})).Once()
+	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.CadenceLatency).Return(metrics.NopStopwatch()).Once()
+	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverHandleAllRequestsLatency).Return(metrics.NopStopwatch()).Once()
 	workflowTestMetrics.On("AddCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverNumPumpedRequestsCount, int64(3)).Once()
 	workflowTestMetrics.On("AddCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverNumHandledRequestsCount, int64(3)).Once()
 	workflowTestMetrics.On("IncCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverPumpedNotEqualHandledCount).Once()
@@ -93,8 +91,8 @@ func (s *workflowSuite) TestArchivalWorkflow_Fail_HashesDoNotEqual() {
 
 func (s *workflowSuite) TestArchivalWorkflow_Exit_TimeoutWithoutSignals() {
 	workflowTestMetrics.On("IncCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverWorkflowStartedCount).Once()
-	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.CadenceLatency).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})).Once()
-	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverHandleAllRequestsLatency).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})).Once()
+	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.CadenceLatency).Return(metrics.NopStopwatch()).Once()
+	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverHandleAllRequestsLatency).Return(metrics.NopStopwatch()).Once()
 	workflowTestMetrics.On("AddCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverNumPumpedRequestsCount, int64(0)).Once()
 	workflowTestMetrics.On("AddCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverNumHandledRequestsCount, int64(0)).Once()
 	workflowTestMetrics.On("IncCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverWorkflowStoppingCount).Once()
@@ -115,8 +113,8 @@ func (s *workflowSuite) TestArchivalWorkflow_Exit_TimeoutWithoutSignals() {
 
 func (s *workflowSuite) TestArchivalWorkflow_Success() {
 	workflowTestMetrics.On("IncCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverWorkflowStartedCount).Once()
-	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.CadenceLatency).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})).Once()
-	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverHandleAllRequestsLatency).Return(tally.NewStopwatch(time.Now(), &nopStopwatchRecorder{})).Once()
+	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.CadenceLatency).Return(metrics.NopStopwatch()).Once()
+	workflowTestMetrics.On("StartTimer", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverHandleAllRequestsLatency).Return(metrics.NopStopwatch()).Once()
 	workflowTestMetrics.On("AddCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverNumPumpedRequestsCount, int64(5)).Once()
 	workflowTestMetrics.On("AddCounter", metrics.ArchiverArchivalWorkflowScope, metrics.ArchiverNumHandledRequestsCount, int64(5)).Once()
 	workflowTestArchiver.On("Start").Once()

service/worker/indexer/esProcessor.go

@@ -26,6 +26,7 @@ import (
 	"time"
 
 	"github.com/olivere/elastic"
+	"github.com/uber-go/tally"
 	"github.com/uber/cadence/common"
 	"github.com/uber/cadence/common/collection"
 	es "github.com/uber/cadence/common/elasticsearch"
@@ -64,6 +65,11 @@ type (
 		logger        log.Logger
 		metricsClient metrics.Client
 	}
+
+	kafkaMessageWithMetrics struct { // value of esProcessorImpl.mapToKafkaMsg
+		message        messaging.Message
+		swFromAddToAck *tally.Stopwatch // metric from message add to process, to message ack/nack
+	}
 )
 
 var _ ESProcessor = (*esProcessorImpl)(nil)
@@ -115,7 +121,12 @@ func (p *esProcessorImpl) Add(request elastic.BulkableRequest, key string, kafka
 		kafkaMsg.Ack()
 		return nil
 	}
-	_, isDup, _ := p.mapToKafkaMsg.PutOrDo(key, kafkaMsg, actionWhenFoundDuplicates)
+	sw := p.metricsClient.StartTimer(metrics.ESProcessorScope, metrics.ESProcessorProcessMsgLatency)
+	mapVal := &kafkaMessageWithMetrics{
+		message:        kafkaMsg,
+		swFromAddToAck: &sw,
+	}
+	_, isDup, _ := p.mapToKafkaMsg.PutOrDo(key, mapVal, actionWhenFoundDuplicates)
 	if isDup {
 		return
 	}
@@ -177,16 +188,21 @@ func (p *esProcessorImpl) ackKafkaMsgHelper(key string, nack bool) {
 	if !ok {
 		return // duplicate kafka message
 	}
-	kafkaMsg, ok := msg.(messaging.Message)
+	kafkaMsg, ok := msg.(*kafkaMessageWithMetrics)
 	if !ok { // must be bug in code and bad deployment
 		p.logger.Fatal("Message is not kafka message.", tag.ESKey(key))
 	}
 
 	if nack {
-		kafkaMsg.Nack()
+		kafkaMsg.message.Nack()
 	} else {
-		kafkaMsg.Ack()
+		kafkaMsg.message.Ack()
 	}
+
+	if kafkaMsg.swFromAddToAck != nil {
+		kafkaMsg.swFromAddToAck.Stop()
+	}
+
 	p.mapToKafkaMsg.Remove(key)
 }
 

service/worker/indexer/esProcessor_test.go

@@ -51,9 +51,12 @@ type esProcessorSuite struct {
 }
 
 var (
-	testIndex = "test-index"
-	testType  = esDocType
-	testID    = "test-doc-id"
+	testIndex     = "test-index"
+	testType      = esDocType
+	testID        = "test-doc-id"
+	testStopWatch = metrics.NopStopwatch()
+	testScope     = metrics.ESProcessorScope
+	testMetric    = metrics.ESProcessorProcessMsgLatency
 )
 
 func TestESProcessorSuite(t *testing.T) {
@@ -139,12 +142,14 @@ func (s *esProcessorSuite) TestAdd() {
 	s.Equal(0, s.esProcessor.mapToKafkaMsg.Len())
 
 	s.mockBulkProcessor.On("Add", request).Return().Once()
+	s.mockMetricClient.On("StartTimer", testScope, testMetric).Return(testStopWatch).Once()
 	s.esProcessor.Add(request, key, mockKafkaMsg)
 	s.Equal(1, s.esProcessor.mapToKafkaMsg.Len())
 	mockKafkaMsg.AssertExpectations(s.T())
 
 	// handle duplicate
 	mockKafkaMsg.On("Ack").Return(nil).Once()
+	s.mockMetricClient.On("StartTimer", testScope, testMetric).Return(testStopWatch).Once()
 	s.esProcessor.Add(request, key, mockKafkaMsg)
 	s.Equal(1, s.esProcessor.mapToKafkaMsg.Len())
 	mockKafkaMsg.AssertExpectations(s.T())
@@ -156,6 +161,7 @@ func (s *esProcessorSuite) TestAdd_ConcurrentAdd() {
 	key := "test-key"
 
 	addFunc := func(wg *sync.WaitGroup) {
+		s.mockMetricClient.On("StartTimer", testScope, testMetric).Return(testStopWatch).Once()
 		s.esProcessor.Add(request, key, mockKafkaMsg)
 		wg.Done()
 	}
@@ -171,7 +177,7 @@ func (s *esProcessorSuite) TestAdd_ConcurrentAdd() {
 	mockKafkaMsg.AssertExpectations(s.T())
 }
 
-func (s *esProcessorSuite) TestBulkAfterAction() {
+func (s *esProcessorSuite) TestBulkAfterActionX() {
 	version := int64(3)
 	testKey := "testKey"
 	request := elastic.NewBulkIndexRequest().
@@ -199,7 +205,11 @@ func (s *esProcessorSuite) TestBulkAfterAction() {
 	}
 
 	mockKafkaMsg := &msgMocks.Message{}
-	s.esProcessor.mapToKafkaMsg.Put(testKey, mockKafkaMsg)
+	mapVal := &kafkaMessageWithMetrics{
+		message:        mockKafkaMsg,
+		swFromAddToAck: &testStopWatch,
+	}
+	s.esProcessor.mapToKafkaMsg.Put(testKey, mapVal)
 	mockKafkaMsg.On("Ack").Return(nil).Once()
 	s.esProcessor.bulkAfterAction(0, requests, response, nil)
 	mockKafkaMsg.AssertExpectations(s.T())
@@ -233,7 +243,11 @@ func (s *esProcessorSuite) TestBulkAfterAction_Nack() {
 	}
 
 	mockKafkaMsg := &msgMocks.Message{}
-	s.esProcessor.mapToKafkaMsg.Put(testKey, mockKafkaMsg)
+	mapVal := &kafkaMessageWithMetrics{
+		message:        mockKafkaMsg,
+		swFromAddToAck: &testStopWatch,
+	}
+	s.esProcessor.mapToKafkaMsg.Put(testKey, mapVal)
 	mockKafkaMsg.On("Nack").Return(nil).Once()
 	s.esProcessor.bulkAfterAction(0, requests, response, nil)
 	mockKafkaMsg.AssertExpectations(s.T())
@@ -275,6 +289,7 @@ func (s *esProcessorSuite) TestAckKafkaMsg() {
 
 	request := elastic.NewBulkIndexRequest()
 	mockKafkaMsg := &msgMocks.Message{}
+	s.mockMetricClient.On("StartTimer", testScope, testMetric).Return(testStopWatch).Once()
 	s.mockBulkProcessor.On("Add", request).Return().Once()
 	s.esProcessor.Add(request, key, mockKafkaMsg)
 	s.Equal(1, s.esProcessor.mapToKafkaMsg.Len())
@@ -293,6 +308,7 @@ func (s *esProcessorSuite) TestNackKafkaMsg() {
 	request := elastic.NewBulkIndexRequest()
 	mockKafkaMsg := &msgMocks.Message{}
 	s.mockBulkProcessor.On("Add", request).Return().Once()
+	s.mockMetricClient.On("StartTimer", testScope, testMetric).Return(testStopWatch).Once()
 	s.esProcessor.Add(request, key, mockKafkaMsg)
 	s.Equal(1, s.esProcessor.mapToKafkaMsg.Len())
 

service/worker/indexer/processor.go

@@ -167,7 +167,9 @@ func (p *indexProcessor) messageProcessLoop(workerWG *sync.WaitGroup, workerID i
 				p.logger.Info("Worker for index processor shutting down.")
 				return // channel closed
 			}
+			sw := p.metricsClient.StartTimer(metrics.IndexProcessorScope, metrics.IndexProcessorProcessMsgLatency)
 			err := p.process(msg)
+			sw.Stop()
 			if err != nil {
 				msg.Nack()
 			}