Tests

config.go

@@ -219,4 +219,7 @@ var defaultConfig = Config{
 	Tracing: candlelight.Config{
 		ApplicationName: applicationName,
 	},
+	Sender: SenderConfig{
+		Linger: 3 * time.Minute,
+	},
 }

http.go

@@ -40,37 +40,37 @@
 }
 type HandlerTelemetryIn struct {
 	fx.In
-	ErrorRequests            prometheus.Counter      `name:"error_request_body_counter"`
-	EmptyRequests            prometheus.Counter      `name:"empty_request_boyd_counter"`
-	InvalidCount             prometheus.Counter      `name:"drops_due_to_invalid_payload"`
-	IncomingQueueDepthMetric prometheus.Gauge        `name:"incoming_queue_depth"`
-	ModifiedWRPCount         prometheus.CounterVec   `name:"modified_wrp_count"`
-	IncomingQueueLatency     prometheus.HistogramVec `name:"incoming_queue_latency_histogram_seconds"`
+	ErrorRequests            prometheus.Counter     `name:"error_request_body_count"`
+	EmptyRequests            prometheus.Counter     `name:"empty_request_body_count"`
+	InvalidCount             prometheus.Counter     `name:"drops_due_to_invalid_payload"`
+	IncomingQueueDepthMetric prometheus.Gauge       `name:"incoming_queue_depth"`
+	ModifiedWRPCount         *prometheus.CounterVec `name:"modified_wrp_count"`
+	IncomingQueueLatency     prometheus.ObserverVec `name:"incoming_queue_latency_histogram_seconds"`
 }
 type HandlerTelemetry struct {
 	errorRequests            prometheus.Counter
 	emptyRequests            prometheus.Counter
 	invalidCount             prometheus.Counter
 	incomingQueueDepthMetric prometheus.Gauge
-	modifiedWRPCount         prometheus.CounterVec
-	incomingQueueLatency     prometheus.HistogramVec
+	modifiedWRPCount         *prometheus.CounterVec
+	incomingQueueLatency     prometheus.ObserverVec
 }
 
 func (sh *ServerHandler) ServeHTTP(response http.ResponseWriter, request *http.Request) {
 	eventType := unknownEventType
 	log := sallust.Get(request.Context())
 	// find time difference, add to metric after function finishes
 	defer func(s time.Time) {
 		sh.recordQueueLatencyToHistogram(s, eventType)
 	}(sh.now())
 
 	log.Info("Receiving incoming request...")
 
 	if len(request.Header["Content-Type"]) != 1 || request.Header["Content-Type"][0] != "application/msgpack" {
 		//return a 415
 		response.WriteHeader(http.StatusUnsupportedMediaType)
 		response.Write([]byte("Invalid Content-Type header(s). Expected application/msgpack. \n"))
 		log.Debug("Invalid Content-Type header(s). Expected application/msgpack. \n")
 		return
 	}
 
@@ -81,7 +81,7 @@
 		// return a 503
 		response.WriteHeader(http.StatusServiceUnavailable)
 		response.Write([]byte("Incoming queue is full.\n"))
 		log.Debug("Incoming queue is full.\n")
 		return
 	}
 
@@ -91,14 +91,14 @@
 	payload, err := io.ReadAll(request.Body)
 	if err != nil {
 		sh.telemetry.errorRequests.Add(1.0)
 		log.Error("Unable to retrieve the request body.", zap.Error(err))
 		response.WriteHeader(http.StatusBadRequest)
 		return
 	}
 
 	if len(payload) == 0 {
 		sh.telemetry.emptyRequests.Add(1.0)
 		log.Error("Empty payload.")
 		response.WriteHeader(http.StatusBadRequest)
 		response.Write([]byte("Empty payload.\n"))
 		return
@@ -114,10 +114,10 @@
 		response.WriteHeader(http.StatusBadRequest)
 		if err != nil {
 			response.Write([]byte("Invalid payload format.\n"))
 			log.Debug("Invalid payload format.")
 		} else {
 			response.Write([]byte("Invalid MessageType.\n"))
 			log.Debug("Invalid MessageType.")
 		}
 		return
 	}
@@ -128,7 +128,7 @@
 		sh.telemetry.invalidCount.Add(1.0)
 		response.WriteHeader(http.StatusBadRequest)
 		response.Write([]byte("Strings must be UTF-8.\n"))
 		log.Debug("Strings must be UTF-8.")
 		return
 	}
 	eventType = msg.FindEventStringSubMatch()
@@ -139,12 +139,12 @@
 	response.WriteHeader(http.StatusAccepted)
 	response.Write([]byte("Request placed on to queue.\n"))
 
 	log.Debug("event passed to senders.", zap.Any("event", msg))
 }
 
 func (sh *ServerHandler) recordQueueLatencyToHistogram(startTime time.Time, eventType string) {
 	endTime := sh.now()
-	sh.telemetry.incomingQueueLatency.With(prometheus.Labels{"event": eventType}).Observe(float64(endTime.Sub(startTime).Seconds()))
+	sh.telemetry.incomingQueueLatency.With(prometheus.Labels{"event": eventType}).Observe(endTime.Sub(startTime).Seconds())
 }
 
 func (sh *ServerHandler) fixWrp(msg *wrp.Message) *wrp.Message {

httpClient.go

@@ -33,15 +33,15 @@
 
 type metricWrapper struct {
 	now          func() time.Time
-	queryLatency prometheus.HistogramVec
+	queryLatency prometheus.ObserverVec
 	id           string
 }
 
-func newMetricWrapper(now func() time.Time, queryLatency prometheus.HistogramVec, id string) (*metricWrapper, error) {
+func newMetricWrapper(now func() time.Time, queryLatency prometheus.ObserverVec, id string) (*metricWrapper, error) {
 	if now == nil {
 		now = time.Now
 	}
-	if queryLatency.MetricVec == nil {
+	if queryLatency == nil {
 		return nil, errNilHistogram
 	}
 	return &metricWrapper{

http_test.go

@@ -1,5 +1,5 @@
-// SPDX-FileCopyrightText: 2021 Comcast Cable Communications Management, LLC
-// SPDX-License-Identifier: Apache-2.0
+// // SPDX-FileCopyrightText: 2021 Comcast Cable Communications Management, LLC
+// // SPDX-License-Identifier: Apache-2.0
 package main
 
 // import (
@@ -13,8 +13,8 @@ package main
 
 // 	"github.com/stretchr/testify/assert"
 // 	"github.com/stretchr/testify/mock"
+// 	"go.uber.org/zap/zaptest"
 
-// 	"github.com/xmidt-org/webpa-common/v2/adapter"
 // 	"github.com/xmidt-org/wrp-go/v3"
 // )
 
@@ -88,7 +88,7 @@ package main
 // 	for _, tc := range tcs {
 // 		assert := assert.New(t)
 
-// 		logger := adapter.DefaultLogger().Logger
+// 		logger := zaptest.NewLogger(t)
 // 		fakeHandler := new(mockHandler)
 // 		if !tc.throwStatusBadRequest {
 // 			fakeHandler.On("HandleRequest", mock.AnythingOfType("int"),
@@ -112,7 +112,7 @@ package main
 // 		fakeHist.On("Observe", fakeLatency.Seconds()).Return().Once()
 
 // 		serverWrapper := &ServerHandler{
-// 			Logger:                   logger,
+// 			log:                   logger,
 // 			caduceusHandler:          fakeHandler,
 // 			errorRequests:            fakeErrorRequests,
 // 			emptyRequests:            fakeEmptyRequests,
@@ -145,7 +145,7 @@ package main
 
 // 	assert := assert.New(t)
 
-// 	logger := adapter.DefaultLogger().Logger
+// 	logger := zaptest.NewLogger(t)
 // 	fakeHandler := new(mockHandler)
 // 	fakeHandler.On("HandleRequest", mock.AnythingOfType("int"),
 // 		mock.AnythingOfType("*wrp.Message")).Return().Once()
@@ -172,7 +172,7 @@ package main
 // 	fakeHist.On("Observe", fakeLatency.Seconds()).Return().Once()
 
 // 	serverWrapper := &ServerHandler{
-// 		Logger:                   logger,
+// 		log:                   logger,
 // 		caduceusHandler:          fakeHandler,
 // 		errorRequests:            fakeErrorRequests,
 // 		emptyRequests:            fakeEmptyRequests,
@@ -206,7 +206,7 @@ package main
 
 // 	assert := assert.New(t)
 
-// 	logger := adapter.DefaultLogger().Logger
+// 	logger := zaptest.NewLogger(t)
 // 	fakeHandler := new(mockHandler)
 // 	fakeHandler.On("HandleRequest", mock.AnythingOfType("int"),
 // 		mock.AnythingOfType("*wrp.Message")).WaitUntil(time.After(time.Second)).Times(2)
@@ -221,7 +221,7 @@ package main
 // 	fakeHist.On("Observe", fakeLatency.Seconds()).Return().Once()
 
 // 	serverWrapper := &ServerHandler{
-// 		Logger:                   logger,
+// 		log:                   logger,
 // 		caduceusHandler:          fakeHandler,
 // 		incomingQueueDepthMetric: fakeQueueDepth,
 // 		maxOutstanding:           1,
@@ -259,7 +259,7 @@ package main
 // 	req := httptest.NewRequest("POST", "localhost:8080", r)
 // 	req.Header.Set("Content-Type", wrp.MimeTypeMsgpack)
 
-// 	logger := adapter.DefaultLogger().Logger
+// 	logger := zaptest.NewLogger(t)
 // 	fakeHandler := new(mockHandler)
 // 	fakeHandler.On("HandleRequest", mock.AnythingOfType("int"),
 // 		mock.AnythingOfType("*wrp.Message")).WaitUntil(time.After(time.Second)).Times(2)
@@ -276,7 +276,7 @@ package main
 // 	fakeHist.On("Observe", fakeLatency.Seconds()).Return().Once()
 
 // 	serverWrapper := &ServerHandler{
-// 		Logger:                   logger,
+// 		log:                   logger,
 // 		caduceusHandler:          fakeHandler,
 // 		emptyRequests:            fakeEmptyRequests,
 // 		incomingQueueDepthMetric: fakeQueueDepth,
@@ -314,7 +314,7 @@ package main
 // 	req := httptest.NewRequest("POST", "localhost:8080", r)
 // 	req.Header.Set("Content-Type", wrp.MimeTypeMsgpack)
 
-// 	logger := adapter.DefaultLogger().Logger
+// 	logger := zaptest.NewLogger(t)
 // 	fakeHandler := new(mockHandler)
 // 	fakeHandler.On("HandleRequest", mock.AnythingOfType("int"),
 // 		mock.AnythingOfType("*wrp.Message")).WaitUntil(time.After(time.Second)).Once()
@@ -331,7 +331,7 @@ package main
 // 	fakeHist.On("Observe", fakeLatency.Seconds()).Return().Once()
 
 // 	serverWrapper := &ServerHandler{
-// 		Logger:                   logger,
+// 		log:                   logger,
 // 		caduceusHandler:          fakeHandler,
 // 		errorRequests:            fakeErrorRequests,
 // 		incomingQueueDepthMetric: fakeQueueDepth,
@@ -368,7 +368,7 @@ package main
 // 	req := httptest.NewRequest("POST", "localhost:8080", r)
 // 	req.Header.Set("Content-Type", wrp.MimeTypeMsgpack)
 
-// 	logger := adapter.DefaultLogger().Logger
+// 	logger := zaptest.NewLogger(t)
 // 	fakeHandler := new(mockHandler)
 // 	fakeHandler.On("HandleRequest", mock.AnythingOfType("int"),
 // 		mock.AnythingOfType("*wrp.Message")).WaitUntil(time.After(time.Second)).Once()
@@ -386,7 +386,7 @@ package main
 // 	fakeHist.On("Observe", fakeLatency.Seconds()).Return().Once()
 
 // 	serverWrapper := &ServerHandler{
-// 		Logger:                   logger,
+// 		log:                   logger,
 // 		caduceusHandler:          fakeHandler,
 // 		invalidCount:             fakeInvalidCount,
 // 		incomingQueueDepthMetric: fakeQueueDepth,
@@ -420,13 +420,13 @@ package main
 
 // 	assert := assert.New(t)
 
-// 	logger := adapter.DefaultLogger().Logger
+// 	logger := zaptest.NewLogger(t)
 // 	fakeHandler := new(mockHandler)
 
 // 	fakeQueueDepth := new(mockGauge)
 
 // 	serverWrapper := &ServerHandler{
-// 		Logger:                   logger,
+// 		log:                   logger,
 // 		caduceusHandler:          fakeHandler,
 // 		incomingQueueDepthMetric: fakeQueueDepth,
 // 		maxOutstanding:           1,

main.go

@@ -140,6 +140,7 @@ func caduceus(arguments []string, run bool) error {
 		touchstone.Provide(),
 		touchhttp.Provide(),
 		ProvideMetrics(),
+		ProvideSenderMetrics(),
 		// ancla.ProvideMetrics(), //TODO: need to add back in once we fix the ancla/argus dependency issue
 
 	)