First pass of HTTP telemetry

go.mod

@@ -89,6 +89,7 @@ require (
 	github.com/jstemmer/go-junit-report v0.9.1 // indirect
 	github.com/klauspost/compress v1.13.5 // indirect
 	github.com/klauspost/cpuid v1.3.1 // indirect
+	github.com/kubecost/events v0.0.3 // indirect
 	github.com/magiconair/properties v1.8.5 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/minio/md5-simd v1.1.0 // indirect

go.sum

@@ -395,6 +395,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/pty v1.1.5/go.mod h1:9r2w37qlBe7rQ6e1fg1S/9xpWHSnaqNdHD3WcMdbPDA=
 github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
+github.com/kubecost/events v0.0.3 h1:q0Hn8DsovzW53T2oSRfZF92JDwlmAAQt0aktl4ccm74=
+github.com/kubecost/events v0.0.3/go.mod h1:i3DyCVatehxq6tAbvBrARuafjkX2DECPk9OWxiaRIhY=
 github.com/labstack/echo/v4 v4.1.11/go.mod h1:i541M3Fj6f76NZtHSj7TXnyM8n2gaodfvfxNnFqi74g=
 github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
 github.com/lib/pq v1.2.0 h1:LXpIM/LZ5xGFhOpXAQUIMM1HdyqzVYM13zNdjCEEcA0=

pkg/cmd/agent/agent.go

@@ -14,6 +14,7 @@ import (
 	"github.com/kubecost/cost-model/pkg/env"
 	"github.com/kubecost/cost-model/pkg/kubeconfig"
 	"github.com/kubecost/cost-model/pkg/log"
+	"github.com/kubecost/cost-model/pkg/metrics"
 	"github.com/kubecost/cost-model/pkg/prom"
 	"github.com/kubecost/cost-model/pkg/util/watcher"
 
@@ -216,7 +217,8 @@ func Execute(opts *AgentOpts) error {
 	rootMux := http.NewServeMux()
 	rootMux.HandleFunc("/healthz", Healthz)
 	rootMux.Handle("/metrics", promhttp.Handler())
-	handler := cors.AllowAll().Handler(rootMux)
+	telemetryHandler := metrics.ResponseMetricMiddleware(rootMux)
+	handler := cors.AllowAll().Handler(telemetryHandler)
 
 	return http.ListenAndServe(fmt.Sprintf(":%d", env.GetKubecostMetricsPort()), handler)
 }

pkg/cmd/costmodel/costmodel.go

@@ -6,6 +6,7 @@ import (
 	"github.com/julienschmidt/httprouter"
 	"github.com/kubecost/cost-model/pkg/costmodel"
 	"github.com/kubecost/cost-model/pkg/errors"
+	"github.com/kubecost/cost-model/pkg/metrics"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/rs/cors"
 )
@@ -29,7 +30,8 @@ func Execute(opts *CostModelOpts) error {
 	a.Router.GET("/allocation/summary", a.ComputeAllocationHandlerSummary)
 	rootMux.Handle("/", a.Router)
 	rootMux.Handle("/metrics", promhttp.Handler())
-	handler := cors.AllowAll().Handler(rootMux)
+	telemetryHandler := metrics.ResponseMetricMiddleware(rootMux)
+	handler := cors.AllowAll().Handler(telemetryHandler)
 
 	return http.ListenAndServe(":9003", errors.PanicHandlerMiddleware(handler))
 }

pkg/costmodel/metrics.go

@@ -346,6 +346,8 @@ func NewCostModelMetricsEmitter(promClient promclient.Client, clusterCache clust
 		EmitKubeStateMetricsV1Only:    env.IsEmitKsmV1MetricsOnly(),
 	})
 
+	metrics.InitKubecostTelemetry(metricsConfig)
+
 	return &CostModelMetricsEmitter{
 		PrometheusClient:              promClient,
 		KubeClusterCache:              clusterCache,

pkg/metrics/events.go

@@ -0,0 +1,12 @@
+package metrics
+
+import "time"
+
+// HttpHandlerMetricEvent contains http handler response metrics.
+type HttpHandlerMetricEvent struct {
+	Handler      string
+	Code         int
+	Method       string
+	ResponseTime time.Duration
+	ResponseSize uint64
+}

pkg/metrics/httpmetricmiddleware.go

@@ -0,0 +1,80 @@
+package metrics
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/kubecost/events"
+)
+
+// ResponseMetricMiddleware dispatches metric events for handles request and responses.
+func ResponseMetricMiddleware(handler http.Handler) http.Handler {
+	dispatcher := events.GlobalDispatcherFor[HttpHandlerMetricEvent]()
+
+	return http.HandlerFunc(func(rw http.ResponseWriter, r *http.Request) {
+		// use a ResponseWriter implementation to record telemetry for the response
+		respWriter := &responseWriterAdapter{w: rw}
+
+		// record method and path of the request
+		method := r.Method
+		path := r.URL.Path
+
+		// time and execute the handler
+		start := time.Now()
+		handler.ServeHTTP(respWriter, r)
+		duration := time.Since(start)
+
+		// record the response code and size
+		code := respWriter.StatusCode()
+		size := respWriter.TotalResponseSize()
+
+		dispatcher.Dispatch(HttpHandlerMetricEvent{
+			Handler:      path,
+			Method:       method,
+			Code:         code,
+			ResponseTime: duration,
+			ResponseSize: size,
+		})
+
+	})
+}
+
+// responseWriterAdapter implements http.ResponseWriter and extracts the statusCode.
+type responseWriterAdapter struct {
+	w          http.ResponseWriter
+	written    bool
+	statusCode int
+	size       uint64
+}
+
+func (wd *responseWriterAdapter) Header() http.Header {
+	return wd.w.Header()
+}
+
+func (wd *responseWriterAdapter) Write(bytes []byte) (int, error) {
+	numBytes, err := wd.w.Write(bytes)
+	wd.size += uint64(numBytes)
+	return numBytes, err
+}
+
+func (wd *responseWriterAdapter) WriteHeader(statusCode int) {
+	wd.written = true
+	wd.statusCode = statusCode
+	wd.w.WriteHeader(statusCode)
+}
+
+func (wd *responseWriterAdapter) StatusCode() int {
+	if !wd.written {
+		return http.StatusOK
+	}
+	return wd.statusCode
+}
+
+func (wd *responseWriterAdapter) Status() string {
+	return fmt.Sprintf("%d", wd.StatusCode())
+}
+
+func (wd *responseWriterAdapter) TotalResponseSize() uint64 {
+	return wd.size
+}

pkg/metrics/telemetry.go

@@ -0,0 +1,61 @@
+package metrics
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/kubecost/events"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	once       sync.Once
+	dispatcher events.Dispatcher[HttpHandlerMetricEvent]
+	// -- append new dispatchers here for new event types
+
+	// prometheus metrics
+	requestsCount *prometheus.CounterVec
+	responseTime  *prometheus.HistogramVec
+	responseSize  *prometheus.SummaryVec
+)
+
+// InitKubecostTelemetry registers kubecost application telemetry.
+func InitKubecostTelemetry(config *MetricsConfig) {
+	// TODO(bolt): Check MetricsConfig for disabled metrics
+
+	once.Do(func() {
+		// register prometheus metrics
+		requestsCount = prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "kubecost_http_requests_total",
+			Help: "kubecost_http_requests_total Total number of HTTP requests",
+		}, []string{"handler", "method", "code"})
+
+		var buckets = []float64{0.001, 0.01, 0.1, 0.3, 0.6, 1, 3, 6, 9, 20, 30, 60, 90, 120, 240, 360, 720}
+		responseTime = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+			Name:    "kubecost_http_response_time_seconds",
+			Help:    "kubecost_http_response_time_seconds Response time in seconds",
+			Buckets: buckets,
+		}, []string{"handler", "method", "code"})
+
+		responseSize = prometheus.NewSummaryVec(prometheus.SummaryOpts{
+			Name: "kubecost_http_response_size_bytes",
+			Help: "kubecost_http_response_size_bytes Response size in bytes",
+		}, []string{"handler", "method", "code"})
+
+		prometheus.MustRegister(requestsCount, responseTime, responseSize)
+
+		// register event listeners
+		dispatcher = events.GlobalDispatcherFor[HttpHandlerMetricEvent]()
+		dispatcher.AddEventHandler(onHttpHandlerMetricEvent)
+		// -- append new event handlers here
+	})
+}
+
+// onHttpHandlerMetricEvent handles all incoming HttpHandlerMetricEvents
+func onHttpHandlerMetricEvent(event HttpHandlerMetricEvent) {
+	code := fmt.Sprintf("%d", event.Code)
+
+	requestsCount.WithLabelValues(event.Handler, event.Method, code).Inc()
+	responseSize.WithLabelValues(event.Handler, event.Method, code).Observe(float64(event.ResponseSize))
+	responseTime.WithLabelValues(event.Handler, event.Method, code).Observe(event.ResponseTime.Seconds())
+}