Merge pull request #312 from xmidt-org/trackLatency

Track latency

CHANGELOG.md

@@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+- Added latency metric, which Tracks the time spent waiting on outbound client URLs to respond. [#312](https://github.com/xmidt-org/caduceus/pull/312)
 
 ## [v0.6.6]
 - Fix a missing return after an invalid utf8 string is handled.  [#315](https://github.com/xmidt-org/caduceus/pull/315)

caduceus_type.go

@@ -57,6 +57,7 @@ type SenderConfig struct {
 type CaduceusMetricsRegistry interface {
 	NewCounter(name string) metrics.Counter
 	NewGauge(name string) metrics.Gauge
+	NewHistogram(name string, buckets int) metrics.Histogram
 }
 
 type RequestHandler interface {

httpClient.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"errors"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/go-kit/kit/metrics"
+)
+
+var (
+	errNilHistogram = errors.New("histogram cannot be nil")
+)
+
+type httpClient interface {
+	Do(*http.Request) (*http.Response, error)
+}
+
+func nopHTTPClient(next httpClient) httpClient {
+	return next
+}
+
+// DoerFunc implements HTTPClient
+type doerFunc func(*http.Request) (*http.Response, error)
+
+func (d doerFunc) Do(req *http.Request) (*http.Response, error) {
+	return d(req)
+}
+
+type metricWrapper struct {
+	now          func() time.Time
+	queryLatency metrics.Histogram
+}
+
+func newMetricWrapper(now func() time.Time, queryLatency metrics.Histogram) (*metricWrapper, error) {
+	if now == nil {
+		now = time.Now
+	}
+	if queryLatency == nil {
+		return nil, errNilHistogram
+	}
+	return &metricWrapper{
+		now:          now,
+		queryLatency: queryLatency,
+	}, nil
+}
+
+func (m *metricWrapper) roundTripper(next httpClient) httpClient {
+	return doerFunc(func(req *http.Request) (*http.Response, error) {
+		startTime := m.now()
+		resp, err := next.Do(req)
+		endTime := m.now()
+		code := networkError
+
+		if err == nil {
+			code = strconv.Itoa(resp.StatusCode)
+		}
+
+		// find time difference, add to metric
+		var latency = endTime.Sub(startTime)
+		m.queryLatency.With("code", code).Observe(latency.Seconds())
+
+		return resp, err
+	})
+}

httpClient_test.go

@@ -0,0 +1,178 @@
+/**
+ * Copyright 2022 Comcast Cable Communications Management, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package main
+
+import (
+	"errors"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/go-kit/kit/metrics"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRoundTripper(t *testing.T) {
+	errTest := errors.New("test error")
+	date1 := time.Date(2021, time.Month(2), 21, 1, 10, 30, 0, time.UTC)
+	date2 := time.Date(2021, time.Month(2), 21, 1, 10, 30, 45, time.UTC)
+
+	tests := []struct {
+		description      string
+		startTime        time.Time
+		endTime          time.Time
+		expectedCode     string
+		request          *http.Request
+		expectedErr      error
+		expectedResponse *http.Response
+	}{
+		{
+			description:  "Success",
+			startTime:    date1,
+			endTime:      date2,
+			expectedCode: "200",
+			request:      exampleRequest(1),
+			expectedErr:  nil,
+			expectedResponse: &http.Response{
+				StatusCode: 200,
+			},
+		},
+		{
+			description:  "503 Service Unavailable",
+			startTime:    date1,
+			endTime:      date2,
+			expectedCode: "503",
+			request:      exampleRequest(1),
+			expectedErr:  nil,
+			expectedResponse: &http.Response{
+				StatusCode: 503,
+			},
+		},
+		{
+			description:      "Network Error",
+			startTime:        date1,
+			endTime:          date2,
+			expectedCode:     "network_err",
+			request:          exampleRequest(1),
+			expectedErr:      errTest,
+			expectedResponse: nil,
+		},
+	}
+
+	for _, tc := range tests {
+
+		t.Run(tc.description, func(t *testing.T) {
+
+			fakeTime := mockTime(tc.startTime, tc.endTime)
+			fakeHandler := new(mockHandler)
+			fakeHist := new(mockHistogram)
+			histogramFunctionCall := []string{"code", tc.expectedCode}
+			fakeLatency := date2.Sub(date1)
+			fakeHist.On("With", histogramFunctionCall).Return().Once()
+			fakeHist.On("Observe", fakeLatency.Seconds()).Return().Once()
+
+			// Create a roundtripper with mock time and mock histogram
+			m, err := newMetricWrapper(fakeTime, fakeHist)
+			require.NoError(t, err)
+			require.NotNil(t, m)
+
+			client := doerFunc(func(*http.Request) (*http.Response, error) {
+
+				return tc.expectedResponse, tc.expectedErr
+			})
+
+			c := m.roundTripper(client)
+			resp, err := c.Do(tc.request)
+
+			if tc.expectedErr == nil {
+				// Read and close response body
+				if resp.Body != nil {
+					io.Copy(ioutil.Discard, resp.Body)
+					resp.Body.Close()
+				}
+				assert.NoError(t, err)
+			} else {
+				assert.ErrorIs(t, tc.expectedErr, err)
+			}
+
+			// Check the histogram and expectations
+			fakeHandler.AssertExpectations(t)
+			fakeHist.AssertExpectations(t)
+
+		})
+	}
+
+}
+
+func TestNewMetricWrapper(t *testing.T) {
+
+	tests := []struct {
+		description   string
+		expectedErr   error
+		fakeTime      func() time.Time
+		fakeHistogram metrics.Histogram
+	}{
+		{
+			description:   "Success",
+			expectedErr:   nil,
+			fakeTime:      time.Now,
+			fakeHistogram: &mockHistogram{},
+		},
+		{
+			description:   "Nil Histogram",
+			expectedErr:   errNilHistogram,
+			fakeTime:      time.Now,
+			fakeHistogram: nil,
+		},
+		{
+			description:   "Nil Time",
+			expectedErr:   nil,
+			fakeTime:      nil,
+			fakeHistogram: &mockHistogram{},
+		},
+	}
+
+	for _, tc := range tests {
+
+		t.Run(tc.description, func(t *testing.T) {
+
+			// Make function call
+			mw, err := newMetricWrapper(tc.fakeTime, tc.fakeHistogram)
+
+			if tc.expectedErr == nil {
+				// Check for no errors
+				assert.NoError(t, err)
+				require.NotNil(t, mw)
+
+				// Check that the time and histogram aren't nil
+				assert.NotNil(t, mw.now)
+				assert.NotNil(t, mw.queryLatency)
+				return
+			}
+
+			// with error checks
+			assert.Nil(t, mw)
+			assert.ErrorIs(t, err, tc.expectedErr)
+
+		})
+	}
+
+}

main.go

@@ -133,10 +133,10 @@ func caduceus(arguments []string) int {
 		DeliveryInterval:    caduceusConfig.Sender.DeliveryInterval,
 		MetricsRegistry:     metricsRegistry,
 		Logger:              logger,
-		Sender: (&http.Client{
+		Sender: doerFunc((&http.Client{
 			Transport: tr,
 			Timeout:   caduceusConfig.Sender.ClientTimeout,
-		}).Do,
+		}).Do),
 		CustomPIDs:        caduceusConfig.Sender.CustomPIDs,
 		DisablePartnerIDs: caduceusConfig.Sender.DisablePartnerIDs,
 	}.New()

metrics.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"github.com/go-kit/kit/metrics"
 	"github.com/xmidt-org/webpa-common/v2/xmetrics"
 )
 
@@ -23,12 +24,14 @@ const (
 	ConsumerDropUntilGauge          = "consumer_drop_until"
 	ConsumerDeliveryWorkersGauge    = "consumer_delivery_workers"
 	ConsumerMaxDeliveryWorkersGauge = "consumer_delivery_workers_max"
+	QueryDurationSecondsHistogram   = "query_duration_seconds_histogram"
 )
 
 const (
 	emptyContentTypeReason = "empty_content_type"
 	emptyUUIDReason        = "empty_uuid"
 	bothEmptyReason        = "empty_uuid_and_content_type"
+	networkError           = "network_err"
 )
 
 func Metrics() []xmetrics.Metric {
@@ -137,6 +140,13 @@ func Metrics() []xmetrics.Metric {
 			Type:       "gauge",
 			LabelNames: []string{"url"},
 		},
+		{
+			Name:       QueryDurationSecondsHistogram,
+			Help:       "A histogram of latencies for queries.",
+			Type:       "histogram",
+			LabelNames: []string{"url", "code"},
+			Buckets:    []float64{0.0625, 0.125, .25, .5, 1, 5, 10, 20, 40, 80, 160},
+		},
 	}
 }
 
@@ -151,7 +161,7 @@ func CreateOutbounderMetrics(m CaduceusMetricsRegistry, c *CaduceusOutboundSende
 
 	c.droppedCutoffCounter = m.NewCounter(SlowConsumerDroppedMsgCounter).With("url", c.id, "reason", "cut_off")
 	c.droppedInvalidConfig = m.NewCounter(SlowConsumerDroppedMsgCounter).With("url", c.id, "reason", "invalid_config")
-	c.droppedNetworkErrCounter = m.NewCounter(SlowConsumerDroppedMsgCounter).With("url", c.id, "reason", "network_err")
+	c.droppedNetworkErrCounter = m.NewCounter(SlowConsumerDroppedMsgCounter).With("url", c.id, "reason", networkError)
 	c.droppedPanic = m.NewCounter(DropsDueToPanic).With("url", c.id)
 	c.queueDepthGauge = m.NewGauge(OutgoingQueueDepth).With("url", c.id)
 	c.renewalTimeGauge = m.NewGauge(ConsumerRenewalTimeGauge).With("url", c.id)
@@ -160,3 +170,7 @@ func CreateOutbounderMetrics(m CaduceusMetricsRegistry, c *CaduceusOutboundSende
 	c.currentWorkersGauge = m.NewGauge(ConsumerDeliveryWorkersGauge).With("url", c.id)
 	c.maxWorkersGauge = m.NewGauge(ConsumerMaxDeliveryWorkersGauge).With("url", c.id)
 }
+
+func NewMetricWrapperMeasures(m CaduceusMetricsRegistry) metrics.Histogram {
+	return m.NewHistogram(QueryDurationSecondsHistogram, 11)
+}

mocks_test.go

@@ -17,6 +17,7 @@
 package main
 
 import (
+	"time"
 	"unicode/utf8"
 
 	"github.com/go-kit/kit/metrics"
@@ -95,6 +96,25 @@ func (m *mockGauge) With(labelValues ...string) metrics.Gauge {
 	return args.Get(0).(metrics.Gauge)
 }
 
+// mockHistogram provides the mock implementation of the metrics.Histogram object
+type mockHistogram struct {
+	mock.Mock
+}
+
+func (m *mockHistogram) Observe(value float64) {
+	m.Called(value)
+}
+
+func (m *mockHistogram) With(labelValues ...string) metrics.Histogram {
+	for _, v := range labelValues {
+		if !utf8.ValidString(v) {
+			panic("not UTF-8")
+		}
+	}
+	m.Called(labelValues)
+	return m
+}
+
 // mockCaduceusMetricsRegistry provides the mock implementation of the
 // CaduceusMetricsRegistry  object
 type mockCaduceusMetricsRegistry struct {
@@ -110,3 +130,20 @@ func (m *mockCaduceusMetricsRegistry) NewGauge(name string) metrics.Gauge {
 	args := m.Called(name)
 	return args.Get(0).(metrics.Gauge)
 }
+
+func (m *mockCaduceusMetricsRegistry) NewHistogram(name string, buckets int) metrics.Histogram {
+	args := m.Called(name)
+	return args.Get(0).(metrics.Histogram)
+}
+
+// mockTime provides two mock time values
+func mockTime(one, two time.Time) func() time.Time {
+	var called bool
+	return func() time.Time {
+		if called {
+			return two
+		}
+		called = true
+		return one
+	}
+}