feat(go): implement metrics and tracing for http and grpc servers

Signed-off-by: Luis Azofra Begara <luis.azofra@cabify.com>

Author: luisazofracabify

go.mod

@@ -11,11 +11,13 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.29.14
 	github.com/aws/aws-sdk-go-v2/service/dynamodb v1.43.3
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.79.3
+	github.com/cabify/gotoprom v1.1.0
 	github.com/ghodss/yaml v1.0.0
 	github.com/golang/protobuf v1.5.4
 	github.com/google/uuid v1.6.0
 	github.com/mattn/go-sqlite3 v1.14.23
 	github.com/pkg/errors v0.9.1
+	github.com/prometheus/client_golang v1.23.2
 	github.com/redis/go-redis/v9 v9.6.1
 	github.com/roberson-io/mmh3 v0.0.0-20190729202758-fdfce3ba6225
 	github.com/rs/zerolog v1.33.0
@@ -62,6 +64,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.30.1 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.33.19 // indirect
 	github.com/aws/smithy-go v1.22.2 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cncf/xds/go v0.0.0-20250501225837-2ac532fd4443 // indirect
@@ -81,15 +84,19 @@ require (
 	github.com/googleapis/gax-go/v2 v2.15.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect
 	github.com/klauspost/asmfmt v1.3.2 // indirect
-	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.8 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/minio/asm2plan9s v0.0.0-20200509001527-cdd76441f9d8 // indirect
 	github.com/minio/c2goasm v0.0.0-20190812172519-36a3d3bbc4f3 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/pierrec/lz4/v4 v4.1.21 // indirect
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/prometheus/client_model v0.6.2 // indirect
+	github.com/prometheus/common v0.66.1 // indirect
+	github.com/prometheus/procfs v0.16.1 // indirect
 	github.com/spiffe/go-spiffe/v2 v2.5.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/zeebo/errs v1.4.0 // indirect
@@ -101,6 +108,7 @@ require (
 	go.opentelemetry.io/otel/metric v1.38.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.38.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.7.1 // indirect
+	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/crypto v0.45.0 // indirect
 	golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
 	golang.org/x/mod v0.29.0 // indirect

go.sum

@@ -0,0 +1,84 @@
+package metrics
+
+import (
+	"reflect"
+	"time"
+
+	"github.com/cabify/gotoprom"
+	"github.com/cabify/gotoprom/prometheusvanilla"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var Metrics struct {
+	HttpDuration func(HttpLabels) TimeHistogram `name:"feast_http_request_duration_seconds" help:"Time taken to serve HTTP requests" buckets:".005,.01,.025,.05,.1,.25,.5,1,2.5,5,10"`
+
+	GrpcDuration func(GrpcLabels) TimeHistogram `name:"feast_grpc_request_duration_seconds" help:"Time taken to serve gRPC requests" buckets:".005,.01,.025,.05,.1,.25,.5,1,2.5,5,10"`
+
+	HttpRequestsTotal func(HttpLabels) prometheus.Counter `name:"feast_http_requests_total" help:"Total number of HTTP requests"`
+
+	GrpcRequestsTotal func(GrpcLabels) prometheus.Counter `name:"feast_grpc_requests_total" help:"Total number of gRPC requests"`
+}
+
+type HttpLabels struct {
+	Method string `label:"method"`
+	Status int    `label:"status"`
+	Path   string `label:"path"`
+}
+
+type GrpcLabels struct {
+	Service string `label:"service"`
+	Method  string `label:"method"`
+	Code    string `label:"code"`
+}
+
+func InitMetrics() {
+	gotoprom.MustInit(&Metrics, "feast")
+}
+
+// TimeHistogram boilerplate from gotoprom README
+
+var (
+	// TimeHistogramType is the reflect.Type of the TimeHistogram interface
+	TimeHistogramType = reflect.TypeOf((*TimeHistogram)(nil)).Elem()
+)
+
+func init() {
+	gotoprom.MustAddBuilder(TimeHistogramType, RegisterTimeHistogram)
+}
+
+// RegisterTimeHistogram registers a TimeHistogram after registering the underlying prometheus.Histogram in the prometheus.Registerer provided
+// The function it returns returns a TimeHistogram type as an interface{}
+func RegisterTimeHistogram(name, help, namespace string, labelNames []string, tag reflect.StructTag) (func(prometheus.Labels) interface{}, prometheus.Collector, error) {
+	f, collector, err := prometheusvanilla.BuildHistogram(name, help, namespace, labelNames, tag)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return func(labels prometheus.Labels) interface{} {
+		return timeHistogramAdapter{Histogram: f(labels).(prometheus.Histogram)}
+	}, collector, nil
+}
+
+// TimeHistogram offers the basic prometheus.Histogram functionality
+// with additional time-observing functions
+type TimeHistogram interface {
+	prometheus.Histogram
+	// Duration observes the duration in seconds
+	Duration(duration time.Duration)
+	// Since observes the duration in seconds since the time point provided
+	Since(time.Time)
+}
+
+type timeHistogramAdapter struct {
+	prometheus.Histogram
+}
+
+// Duration observes the duration in seconds
+func (to timeHistogramAdapter) Duration(duration time.Duration) {
+	to.Observe(duration.Seconds())
+}
+
+// Since observes the duration in seconds since the time point provided
+func (to timeHistogramAdapter) Since(duration time.Time) {
+	to.Duration(time.Since(duration))
+}

go/feast-server

@@ -0,0 +1,21 @@
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMetricsInit(t *testing.T) {
+	InitMetrics()
+
+	// We can't easily check the global registry without possibly conflicting with other tests or init order,
+	// but we can check if the struct fields are populated (not nil)
+
+	assert.NotNil(t, Metrics.HttpDuration)
+	assert.NotNil(t, Metrics.GrpcDuration)
+	assert.NotNil(t, Metrics.HttpRequestsTotal)
+
+	// Call them to ensure no panic
+	Metrics.HttpRequestsTotal(HttpLabels{Method: "GET", Status: 200, Path: "/"}).Inc()
+}

go/internal/feast/metrics/metrics.go

@@ -9,7 +9,6 @@ import (
 	prototypes "github.com/feast-dev/feast/go/protos/feast/types"
 	"github.com/feast-dev/feast/go/types"
 	"github.com/google/uuid"
-
 )
 
 const feastServerVersion = "0.0.1"

go/internal/feast/metrics/metrics_test.go

@@ -17,6 +17,9 @@ import (
 	prototypes "github.com/feast-dev/feast/go/protos/feast/types"
 	"github.com/feast-dev/feast/go/types"
 	"github.com/rs/zerolog/log"
+
+	"github.com/feast-dev/feast/go/internal/feast/metrics"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 type httpServer struct {
@@ -335,9 +338,51 @@ func recoverMiddleware(next http.Handler) http.Handler {
 	})
 }
 
+type statusWriter struct {
+	http.ResponseWriter
+	status int
+	length int
+}
+
+func (w *statusWriter) WriteHeader(status int) {
+	w.status = status
+	w.ResponseWriter.WriteHeader(status)
+}
+
+func (w *statusWriter) Write(b []byte) (int, error) {
+	if w.status == 0 {
+		w.status = 200
+	}
+	n, err := w.ResponseWriter.Write(b)
+	w.length += n
+	return n, err
+}
+
+func metricsMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		t0 := time.Now()
+		sw := &statusWriter{ResponseWriter: w}
+		next.ServeHTTP(sw, r)
+		duration := time.Since(t0)
+
+		metrics.Metrics.HttpDuration(metrics.HttpLabels{
+			Method: r.Method,
+			Status: sw.status,
+			Path:   r.URL.Path,
+		}).Duration(duration)
+
+		metrics.Metrics.HttpRequestsTotal(metrics.HttpLabels{
+			Method: r.Method,
+			Status: sw.status,
+			Path:   r.URL.Path,
+		}).Inc()
+	})
+}
+
 func (s *httpServer) Serve(host string, port int) error {
 	mux := http.NewServeMux()
-	mux.Handle("/get-online-features", recoverMiddleware(http.HandlerFunc(s.getOnlineFeatures)))
+	mux.Handle("/get-online-features", metricsMiddleware(recoverMiddleware(http.HandlerFunc(s.getOnlineFeatures))))
+	mux.Handle("/metrics", promhttp.Handler())
 	mux.HandleFunc("/health", healthCheckHandler)
 	s.server = &http.Server{Addr: fmt.Sprintf("%s:%d", host, port), Handler: mux, ReadTimeout: 5 * time.Second, WriteTimeout: 10 * time.Second, IdleTimeout: 15 * time.Second}
 	err := s.server.ListenAndServe()

go/internal/feast/server/grpc_server.go

@@ -4,8 +4,8 @@ import (
 	"os"
 
 	"github.com/rs/zerolog"
-	"go.opentelemetry.io/otel/trace"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/trace"
 )
 
 var tracer = otel.Tracer("github.com/feast-dev/feast/go/server")

go/internal/feast/server/http_server.go

@@ -5,10 +5,12 @@ import (
 	"flag"
 	"fmt"
 	"net"
+	"net/http"
 	"os"
 	"os/signal"
 	"strings"
 	"syscall"
+	"time"
 
 	"github.com/feast-dev/feast/go/internal/feast"
 	"github.com/feast-dev/feast/go/internal/feast/registry"
@@ -19,7 +21,10 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/health"
 	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/status"
 
+	"github.com/feast-dev/feast/go/internal/feast/metrics"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
 	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
@@ -65,6 +70,8 @@ func main() {
 	flag.IntVar(&port, "port", port, "Specify a port for the server")
 	flag.Parse()
 
+	metrics.InitMetrics()
+
 	// Initialize tracer
 	if OTELTracingEnabled() {
 		ctx := context.Background()
@@ -156,11 +163,50 @@ func StartGrpcServer(fs *feast.FeatureStore, host string, port int, writeLoggedF
 		return err
 	}
 
-	grpcServer := grpc.NewServer()
+	grpcServer := grpc.NewServer(
+		grpc.UnaryInterceptor(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+			t0 := time.Now()
+			resp, err := handler(ctx, req)
+			duration := time.Since(t0)
+			code := status.Code(err).String()
+
+			parts := strings.Split(info.FullMethod, "/")
+			service := "unknown"
+			method := "unknown"
+			if len(parts) >= 3 {
+				service = parts[1]
+				method = parts[2]
+			}
+
+			metrics.Metrics.GrpcDuration(metrics.GrpcLabels{
+				Service: service,
+				Method:  method,
+				Code:    code,
+			}).Duration(duration)
+
+			metrics.Metrics.GrpcRequestsTotal(metrics.GrpcLabels{
+				Service: service,
+				Method:  method,
+				Code:    code,
+			}).Inc()
+
+			return resp, err
+		}),
+	)
 	serving.RegisterServingServiceServer(grpcServer, ser)
 	healthService := health.NewServer()
 	grpc_health_v1.RegisterHealthServer(grpcServer, healthService)
 
+	metricsServer := &http.Server{Addr: fmt.Sprintf(":%d", 9090)}
+	go func() {
+		log.Info().Msgf("Starting metrics server on port %d", 9090)
+		http.Handle("/metrics", promhttp.Handler())
+		metricsServer.Handler = http.DefaultServeMux
+		if err := metricsServer.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			log.Error().Err(err).Msg("Failed to start metrics server")
+		}
+	}()
+
 	stop := make(chan os.Signal, 1)
 	signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
 
@@ -172,6 +218,12 @@ func StartGrpcServer(fs *feast.FeatureStore, host string, port int, writeLoggedF
 		if loggingService != nil {
 			loggingService.Stop()
 		}
+		
+		log.Info().Msg("Stopping metrics server...")
+		if err := metricsServer.Shutdown(context.Background()); err != nil {
+			log.Error().Err(err).Msg("Error stopping metrics server")
+		}
+		
 		log.Info().Msg("gRPC server terminated")
 	}()