feat: merge two metric servers

cmd/bbr/main.go

@@ -18,26 +18,22 @@ package main
 
 import (
 	"flag"
-	"net"
-	"net/http"
 	"os"
-	"strconv"
 
 	"github.com/go-logr/logr"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	uberzap "go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"google.golang.org/grpc"
 	healthPb "google.golang.org/grpc/health/grpc_health_v1"
-	"k8s.io/client-go/rest"
-	"k8s.io/component-base/metrics/legacyregistry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
 	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
+	"sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/metrics"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/server"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
@@ -50,9 +46,8 @@ var (
 		"grpcHealthPort",
 		9005,
 		"The port used for gRPC liveness and readiness probes")
-	metricsPort = flag.Int(
-		"metricsPort", 9090, "The metrics port")
-	streaming = flag.Bool(
+	metricsAddr = flag.String("metrics-bind-address", ":9090", "The address the metric endpoint binds to.")
+	streaming   = flag.Bool(
 		"streaming", false, "Enables streaming support for Envoy full-duplex streaming mode")
 	logVerbosity = flag.Int("v", logging.DEFAULT, "number for the log level verbosity")
 
@@ -85,7 +80,18 @@ func run() error {
 		return err
 	}
 
-	mgr, err := ctrl.NewManager(cfg, ctrl.Options{})
+	metrics.Register()
+
+	// Register metrics handler.
+	// Metrics endpoint is enabled in 'config/default/kustomization.yaml'. The Metrics options configure the server.
+	// More info:
+	// - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/metrics/server
+	// - https://book.kubebuilder.io/reference/metrics.html
+	metricsServerOptions := metricsserver.Options{
+		BindAddress:    *metricsAddr,
+		FilterProvider: filters.WithAuthenticationAndAuthorization,
+	}
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{Metrics: metricsServerOptions})
 	if err != nil {
 		setupLog.Error(err, "Failed to create manager", "config", cfg)
 		return err
@@ -107,11 +113,6 @@ func run() error {
 		return err
 	}
 
-	// Register metrics handler.
-	if err := registerMetricsHandler(mgr, *metricsPort, cfg); err != nil {
-		return err
-	}
-
 	// Start the manager. This blocks until a signal is received.
 	setupLog.Info("Manager starting")
 	if err := mgr.Start(ctx); err != nil {
@@ -152,58 +153,3 @@ func initLogging(opts *zap.Options) {
 	logger := zap.New(zap.UseFlagOptions(opts), zap.RawZapOpts(uberzap.AddCaller()))
 	ctrl.SetLogger(logger)
 }
-
-const metricsEndpoint = "/metrics"
-
-// registerMetricsHandler adds the metrics HTTP handler as a Runnable to the given manager.
-func registerMetricsHandler(mgr manager.Manager, port int, cfg *rest.Config) error {
-	metrics.Register()
-
-	// Init HTTP server.
-	h, err := metricsHandlerWithAuthenticationAndAuthorization(cfg)
-	if err != nil {
-		return err
-	}
-
-	mux := http.NewServeMux()
-	mux.Handle(metricsEndpoint, h)
-
-	srv := &http.Server{
-		Addr:    net.JoinHostPort("", strconv.Itoa(port)),
-		Handler: mux,
-	}
-
-	if err := mgr.Add(&manager.Server{
-		Name:   "metrics",
-		Server: srv,
-	}); err != nil {
-		setupLog.Error(err, "Failed to register metrics HTTP handler")
-		return err
-	}
-	return nil
-}
-
-func metricsHandlerWithAuthenticationAndAuthorization(cfg *rest.Config) (http.Handler, error) {
-	h := promhttp.HandlerFor(
-		legacyregistry.DefaultGatherer,
-		promhttp.HandlerOpts{},
-	)
-	httpClient, err := rest.HTTPClientFor(cfg)
-	if err != nil {
-		setupLog.Error(err, "Failed to create http client for metrics auth")
-		return nil, err
-	}
-
-	filter, err := filters.WithAuthenticationAndAuthorization(cfg, httpClient)
-	if err != nil {
-		setupLog.Error(err, "Failed to create metrics filter for auth")
-		return nil, err
-	}
-	metricsLogger := ctrl.Log.WithName("metrics").WithValues("path", metricsEndpoint)
-	metricsAuthHandler, err := filter(metricsLogger, h)
-	if err != nil {
-		setupLog.Error(err, "Failed to create metrics auth handler")
-		return nil, err
-	}
-	return metricsAuthHandler, nil
-}

cmd/epp/main.go

@@ -19,38 +19,29 @@ package main
 import (
 	"flag"
 	"fmt"
-	"net"
-	"net/http"
 	"os"
-	"strconv"
 
 	"github.com/go-logr/logr"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	uberzap "go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"google.golang.org/grpc"
 	healthPb "google.golang.org/grpc/health/grpc_health_v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/rest"
-	"k8s.io/component-base/metrics/legacyregistry"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
 	"sigs.k8s.io/gateway-api-inference-extension/internal/runnable"
 	backendmetrics "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/backend/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/datastore"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics"
-	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/metrics/collectors"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/scheduling"
 	runserver "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/server"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
 
-const (
-	defaultMetricsEndpoint = "/metrics"
-)
-
 var (
 	grpcPort = flag.Int(
 		"grpcPort",
@@ -60,8 +51,7 @@ var (
 		"grpcHealthPort",
 		9003,
 		"The port used for gRPC liveness and readiness probes")
-	metricsPort = flag.Int(
-		"metricsPort", 9090, "The metrics port")
+	metricsAddr                = flag.String("metrics-bind-address", ":9090", "The address the metric endpoint binds to.")
 	destinationEndpointHintKey = flag.String(
 		"destinationEndpointHintKey",
 		runserver.DefaultDestinationEndpointHintKey,
@@ -143,11 +133,22 @@ func run() error {
 		return err
 	}
 
+	metrics.Register()
+	// Register metrics handler.
+	// Metrics endpoint is enabled in 'config/default/kustomization.yaml'. The Metrics options configure the server.
+	// More info:
+	// - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/metrics/server
+	// - https://book.kubebuilder.io/reference/metrics.html
+	metricsServerOptions := metricsserver.Options{
+		BindAddress:    *metricsAddr,
+		FilterProvider: filters.WithAuthenticationAndAuthorization,
+	}
+
 	poolNamespacedName := types.NamespacedName{
 		Name:      *poolName,
 		Namespace: *poolNamespace,
 	}
-	mgr, err := runserver.NewDefaultManager(poolNamespacedName, cfg)
+	mgr, err := runserver.NewDefaultManager(poolNamespacedName, cfg, metricsServerOptions)
 	if err != nil {
 		setupLog.Error(err, "Failed to create controller manager")
 		return err
@@ -199,11 +200,6 @@ func run() error {
 		return err
 	}
 
-	// Register metrics handler.
-	if err := registerMetricsHandler(mgr, *metricsPort, cfg, datastore); err != nil {
-		return err
-	}
-
 	// Start the manager. This blocks until a signal is received.
 	setupLog.Info("Controller manager starting")
 	if err := mgr.Start(ctx); err != nil {
@@ -247,62 +243,6 @@ func registerHealthServer(mgr manager.Manager, logger logr.Logger, ds datastore.
 	return nil
 }
 
-// registerMetricsHandler adds the metrics HTTP handler as a Runnable to the given manager.
-func registerMetricsHandler(mgr manager.Manager, port int, cfg *rest.Config, ds datastore.Datastore) error {
-	metrics.Register()
-	legacyregistry.CustomMustRegister(collectors.NewInferencePoolMetricsCollector(ds))
-
-	metrics.RecordInferenceExtensionInfo()
-
-	// Init HTTP server.
-	h, err := metricsHandlerWithAuthenticationAndAuthorization(cfg)
-	if err != nil {
-		return err
-	}
-
-	mux := http.NewServeMux()
-	mux.Handle(defaultMetricsEndpoint, h)
-
-	srv := &http.Server{
-		Addr:    net.JoinHostPort("", strconv.Itoa(port)),
-		Handler: mux,
-	}
-
-	if err := mgr.Add(&manager.Server{
-		Name:   "metrics",
-		Server: srv,
-	}); err != nil {
-		setupLog.Error(err, "Failed to register metrics HTTP handler")
-		return err
-	}
-	return nil
-}
-
-func metricsHandlerWithAuthenticationAndAuthorization(cfg *rest.Config) (http.Handler, error) {
-	h := promhttp.HandlerFor(
-		legacyregistry.DefaultGatherer,
-		promhttp.HandlerOpts{},
-	)
-	httpClient, err := rest.HTTPClientFor(cfg)
-	if err != nil {
-		setupLog.Error(err, "Failed to create http client for metrics auth")
-		return nil, err
-	}
-
-	filter, err := filters.WithAuthenticationAndAuthorization(cfg, httpClient)
-	if err != nil {
-		setupLog.Error(err, "Failed to create metrics filter for auth")
-		return nil, err
-	}
-	metricsLogger := ctrl.Log.WithName("metrics").WithValues("path", defaultMetricsEndpoint)
-	metricsAuthHandler, err := filter(metricsLogger, h)
-	if err != nil {
-		setupLog.Error(err, "Failed to create metrics auth handler")
-		return nil, err
-	}
-	return metricsAuthHandler, nil
-}
-
 func validateFlags() error {
 	if *poolName == "" {
 		return fmt.Errorf("required %q flag not set", "poolName")

pkg/bbr/handlers/request_test.go

@@ -26,8 +26,8 @@ import (
 	extProcPb "github.com/envoyproxy/go-control-plane/envoy/service/ext_proc/v3"
 	"github.com/google/go-cmp/cmp"
 	"google.golang.org/protobuf/testing/protocmp"
-	"k8s.io/component-base/metrics/legacyregistry"
 	metricsutils "k8s.io/component-base/metrics/testutil"
+	crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 	"sigs.k8s.io/gateway-api-inference-extension/pkg/bbr/metrics"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
 )
@@ -204,7 +204,7 @@ func TestHandleRequestBody(t *testing.T) {
 	bbr_success_total{} 1
 	`
 
-	if err := metricsutils.GatherAndCompare(legacyregistry.DefaultGatherer, strings.NewReader(wantMetrics), "inference_model_request_total"); err != nil {
+	if err := metricsutils.GatherAndCompare(crmetrics.Registry, strings.NewReader(wantMetrics), "inference_model_request_total"); err != nil {
 		t.Error(err)
 	}
 }

pkg/bbr/metrics/metrics.go

@@ -17,51 +17,49 @@ limitations under the License.
 package metrics
 
 import (
+	"fmt"
 	"sync"
 
+	"github.com/prometheus/client_golang/prometheus"
 	compbasemetrics "k8s.io/component-base/metrics"
-	"k8s.io/component-base/metrics/legacyregistry"
+	"sigs.k8s.io/controller-runtime/pkg/metrics"
 )
 
 const component = "bbr"
 
 var (
-	successCounter = compbasemetrics.NewCounterVec(
-		&compbasemetrics.CounterOpts{
-			Subsystem:      component,
-			Name:           "success_total",
-			Help:           "Count of successes pulling model name from body and injecting it in the request headers.",
-			StabilityLevel: compbasemetrics.ALPHA,
+	successCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: component,
+			Name:      "success_total",
+			Help:      fmt.Sprintf("[%v] %v", compbasemetrics.ALPHA, "Count of successes pulling model name from body and injecting it in the request headers."),
 		},
 		[]string{},
 	)
-	modelNotInBodyCounter = compbasemetrics.NewCounterVec(
-		&compbasemetrics.CounterOpts{
-			Subsystem:      component,
-			Name:           "model_not_in_body_total",
-			Help:           "Count of times the model was not present in the request body.",
-			StabilityLevel: compbasemetrics.ALPHA,
+	modelNotInBodyCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: component,
+			Name:      "model_not_in_body_total",
+			Help:      fmt.Sprintf("[%v] %v", compbasemetrics.ALPHA, "Count of times the model was not present in the request body."),
 		},
 		[]string{},
 	)
-	modelNotParsedCounter = compbasemetrics.NewCounterVec(
-		&compbasemetrics.CounterOpts{
-			Subsystem:      component,
-			Name:           "model_not_parsed_total",
-			Help:           "Count of times the model was in the request body but we could not parse it.",
-			StabilityLevel: compbasemetrics.ALPHA,
+	modelNotParsedCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Subsystem: component,
+			Name:      "model_not_parsed_total",
+			Help:      fmt.Sprintf("[%v] %v", compbasemetrics.ALPHA, "Count of times the model was in the request body but we could not parse it."),
 		},
 		[]string{},
 	)
 
 	// TODO: Uncomment and use this metrics once the core server implementation has handling to skip body parsing if header exists.
 	/*
-		modelAlreadyPresentInHeaderCounter = compbasemetrics.NewCounterVec(
-			&compbasemetrics.CounterOpts{
+		modelAlreadyPresentInHeaderCounter = prometheus.NewCounterVec(
+			prometheus.CounterOpts{
 				Subsystem:      component,
 				Name:           "model_already_present_in_header_total",
 				Help:           "Count of times the model was already present in request headers.",
-				StabilityLevel: compbasemetrics.ALPHA,
 			},
 			[]string{},
 		)
@@ -73,10 +71,10 @@ var registerMetrics sync.Once
 // Register all metrics.
 func Register() {
 	registerMetrics.Do(func() {
-		legacyregistry.MustRegister(successCounter)
-		legacyregistry.MustRegister(modelNotInBodyCounter)
-		legacyregistry.MustRegister(modelNotParsedCounter)
-		// legacyregistry.MustRegister(modelAlreadyPresentInHeaderCounter)
+		metrics.Registry.MustRegister(successCounter)
+		metrics.Registry.MustRegister(modelNotInBodyCounter)
+		metrics.Registry.MustRegister(modelNotParsedCounter)
+		// metrics.Registry.MustRegister(modelAlreadyPresentInHeaderCounter)
 	})
 }