diff --git a/cmd/kube-vip.go b/cmd/kube-vip.go
index 0c3d1385..bb5981ab 100644
--- a/cmd/kube-vip.go
+++ b/cmd/kube-vip.go
@@ -4,17 +4,14 @@ import (
"context"
"fmt"
"net"
- "net/http"
"os"
"slices"
"strconv"
"strings"
"sync"
- "time"
log "log/slog"
- "github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/spf13/cobra"
"github.com/vishvananda/netlink"
"golang.org/x/sys/unix"
@@ -350,12 +347,17 @@ var kubeVipManager = &cobra.Command{
ctx, cancel := context.WithCancel(cmd.Context())
defer cancel()
+ metrics.RegisterPrometheusMetrics()
+
// start prometheus server
if initConfig.PrometheusHTTPServer != "" {
wg.Go(func() {
- servePrometheusHTTPServer(ctx, PrometheusHTTPServerConfig{
+ if err := metrics.Serve(ctx, metrics.ServerConfig{
Addr: initConfig.PrometheusHTTPServer,
- })
+ }); err != nil {
+ // Continue even if metrics server fails
+ log.Error("prometheus HTTP server", "err", err)
+ }
})
}
@@ -473,7 +475,8 @@ var kubeVipManager = &cobra.Command{
return fmt.Errorf("new manager: %w", err)
}
- metrics.RegisterPrometheusMetrics()
+ // Label metrics after the call to manager.New, as it may modify the node name
+ // if it was not set in the configuration.
metrics.BuildInfo.WithLabelValues(Release.Version, Release.Build, initConfig.NodeName)
// Start the service manager, this will watch the config Map and construct kube-vip services for it
@@ -485,65 +488,6 @@ var kubeVipManager = &cobra.Command{
},
}
-// PrometheusHTTPServerConfig defines the Prometheus server configuration.
-type PrometheusHTTPServerConfig struct {
- // Addr sets the http server address used to expose the metric endpoint
- Addr string
-}
-
-func servePrometheusHTTPServer(ctx context.Context, config PrometheusHTTPServerConfig) {
- var err error
- mux := http.NewServeMux()
- mux.Handle("/metrics", promhttp.Handler())
- mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { //nolint TODO
- _, _ = w.Write([]byte(`
- kube-vip
-
- kube-vip Metrics
- Metrics
-
- `))
- })
-
- srv := &http.Server{
- Addr: config.Addr,
- Handler: mux,
- ReadHeaderTimeout: 2 * time.Second,
- }
-
- wg := sync.WaitGroup{}
-
- wg.Go(func() {
- if err = srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
- log.Error("prometheus HTTP server", "err", err)
- return
- }
- })
-
- log.Info("prometheus HTTP server started")
-
- <-ctx.Done()
-
- // create prometheus shutdown context (independent of other contexts)
- ctxShutDown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
- defer func() {
- cancel()
- }()
-
- if err = srv.Shutdown(ctxShutDown); err != nil {
- log.Error("shutting down prometheus HTTP server", "err", err)
- return
- }
-
- if err == http.ErrServerClosed {
- err = nil
- }
-
- log.Info("prometheus HTTP server stopped")
-
- wg.Wait()
-}
-
func GenerateCidrRange(address string, dnsMode string) (string, error) {
var cidrs []string
diff --git a/pkg/manager/manager.go b/pkg/manager/manager.go
index 9802bbc0..87df7037 100644
--- a/pkg/manager/manager.go
+++ b/pkg/manager/manager.go
@@ -32,7 +32,6 @@ import (
"github.com/kube-vip/kube-vip/pkg/upnp"
"github.com/kube-vip/kube-vip/pkg/utils"
"github.com/kube-vip/kube-vip/pkg/vip"
- "github.com/prometheus/client_golang/prometheus"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/rest"
)
@@ -57,10 +56,6 @@ type Manager struct {
svcProcessor *services.Processor
- // This is a prometheus counter used to count the number of events received
- // from the service watcher
- countServiceWatchEvent *prometheus.CounterVec
-
// This mutex is to protect calls from various goroutines
mutex sync.Mutex
@@ -259,16 +254,10 @@ func New(ctx context.Context, configMap string, config *kubevip.Config) (*Manage
intfMgr, arpMgr, nodeLabelManager, electionMgr, leaseMgr, routeMgr)
return &Manager{
- clientSet: clientset,
- rwClientSet: rwClientSet,
- configMap: configMap,
- config: config,
- countServiceWatchEvent: prometheus.NewCounterVec(prometheus.CounterOpts{
- Namespace: "kube_vip",
- Subsystem: "manager",
- Name: "all_services_events",
- Help: "Count all events fired by the service watcher categorised by event type",
- }, []string{"type"}),
+ clientSet: clientset,
+ rwClientSet: rwClientSet,
+ configMap: configMap,
+ config: config,
signalChan: signalChan,
svcProcessor: svcProcessor,
intfMgr: intfMgr,
diff --git a/pkg/metrics/prometheus.go b/pkg/metrics/prometheus.go
index 8b588973..09748651 100644
--- a/pkg/metrics/prometheus.go
+++ b/pkg/metrics/prometheus.go
@@ -1,6 +1,10 @@
package metrics
-import "github.com/prometheus/client_golang/prometheus"
+import (
+ "sync"
+
+ "github.com/prometheus/client_golang/prometheus"
+)
var (
// Service / VIP Lifecycle
@@ -68,19 +72,25 @@ var (
)
)
+var registerOnce sync.Once
+
+// RegisterPrometheusMetrics registers all kube-vip metrics with the default
+// Prometheus registry.
func RegisterPrometheusMetrics() {
- // Register all metrics with Prometheus
- prometheus.MustRegister(
- ActiveServices,
- ServiceReconcileErrorsTotal,
- ServiceReconcileDuration,
- LeaderTransitionsTotal,
- IsLeader,
- ServiceElectionLoops,
- ServiceElectionAttemptsTotal,
- ServiceElectionErrorsTotal,
- BGPSessionInfoGauge,
- BuildInfo,
- CountServiceWatchEvent,
- )
+ registerOnce.Do(func() {
+ // Register all metrics with Prometheus
+ prometheus.MustRegister(
+ ActiveServices,
+ ServiceReconcileErrorsTotal,
+ ServiceReconcileDuration,
+ LeaderTransitionsTotal,
+ IsLeader,
+ ServiceElectionLoops,
+ ServiceElectionAttemptsTotal,
+ ServiceElectionErrorsTotal,
+ BGPSessionInfoGauge,
+ BuildInfo,
+ CountServiceWatchEvent,
+ )
+ })
}
diff --git a/pkg/metrics/server.go b/pkg/metrics/server.go
new file mode 100644
index 00000000..8e831e1f
--- /dev/null
+++ b/pkg/metrics/server.go
@@ -0,0 +1,94 @@
+package metrics
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "net"
+ "net/http"
+ "sync"
+ "time"
+
+ log "log/slog"
+
+ "github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+// shutdownTimeout bounds how long the server waits for in-flight requests to
+// finish once the context is cancelled.
+const shutdownTimeout = 5 * time.Second
+
+// ServerConfig defines the Prometheus server configuration.
+type ServerConfig struct {
+ // Addr sets the http server address used to expose the metric endpoint
+ Addr string
+}
+
+// Serve exposes the Prometheus metrics endpoint on the configured address.
+func Serve(ctx context.Context, config ServerConfig) error {
+ ln, err := net.Listen("tcp", config.Addr)
+ if err != nil {
+ return fmt.Errorf("listening on %q: %w", config.Addr, err)
+ }
+
+ return serve(ctx, ln)
+}
+
+// serve starts the metrics endpoint on the provided listener
+func serve(ctx context.Context, ln net.Listener) error {
+ srv := &http.Server{
+ Handler: newServeMux(),
+ ReadHeaderTimeout: 2 * time.Second,
+ }
+
+ wg := sync.WaitGroup{}
+ defer wg.Wait()
+
+ serveErr := make(chan error, 1)
+ wg.Go(func() {
+ err := srv.Serve(ln)
+ if errors.Is(err, http.ErrServerClosed) {
+ err = nil
+ }
+ serveErr <- err
+ })
+
+ log.Info("prometheus HTTP server started", "addr", ln.Addr().String())
+
+ select {
+ case err := <-serveErr:
+ if err != nil {
+ return fmt.Errorf("serving prometheus metrics: %w", err)
+ }
+ return nil
+ case <-ctx.Done():
+ }
+
+ // create prometheus shutdown context (independent of other contexts)
+ ctxShutDown, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
+ defer cancel()
+
+ if err := srv.Shutdown(ctxShutDown); err != nil {
+ return fmt.Errorf("shutting down prometheus HTTP server: %w", err)
+ }
+
+ log.Info("prometheus HTTP server stopped")
+
+ return nil
+}
+
+func newServeMux() *http.ServeMux {
+ mux := http.NewServeMux()
+ mux.Handle("/metrics", promhttp.Handler())
+ mux.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
+ _, _ = w.Write([]byte(`
+ kube-vip
+
+ kube-vip Metrics
+ Metrics
+
+ `))
+ })
+
+ return mux
+}
diff --git a/pkg/metrics/server_test.go b/pkg/metrics/server_test.go
new file mode 100644
index 00000000..aa46b5f0
--- /dev/null
+++ b/pkg/metrics/server_test.go
@@ -0,0 +1,213 @@
+package metrics
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/prometheus/client_golang/prometheus"
+)
+
+func TestServeExposesKubeVipMetrics(t *testing.T) {
+ RegisterPrometheusMetrics()
+ version, build, node := "v1.2.3", "test-build", "node-1"
+ BuildInfo.WithLabelValues(version, build, node)
+
+ base, stop := startServer(t, newTestListener(t))
+
+ body, code := get(t, base+"/metrics")
+ if code != http.StatusOK {
+ t.Fatalf("GET /metrics status = %d, want %d", code, http.StatusOK)
+ }
+
+ // Label names are exposed in alphabetical order.
+ want := fmt.Sprintf("kube_vip_build_info{build=\"%s\",node=\"%s\",version=\"%s\"}", build, node, version)
+ if !strings.Contains(body, want) {
+ t.Errorf("GET /metrics body does not contain %s, got:\n%s", want, body)
+ }
+
+ if err := stop(); err != nil {
+ t.Errorf("serve returned an error on shutdown: %v", err)
+ }
+}
+
+func TestServeRootPageLinksToMetrics(t *testing.T) {
+ base, stop := startServer(t, newTestListener(t))
+
+ body, code := get(t, base+"/")
+ if code != http.StatusOK {
+ t.Fatalf("GET / status = %d, want %d", code, http.StatusOK)
+ }
+
+ if !strings.Contains(body, `href="/metrics"`) {
+ t.Errorf("GET / body does not link to /metrics, got:\n%s", body)
+ }
+
+ if err := stop(); err != nil {
+ t.Errorf("serve returned an error on shutdown: %v", err)
+ }
+}
+
+func TestServeStopsOnContextCancellation(t *testing.T) {
+ ln := newTestListener(t)
+ addr := ln.Addr().String()
+
+ _, stop := startServer(t, ln)
+
+ // stop blocks until serve returns, and serve waits on its serving
+ // goroutine, so a clean return means nothing was left running.
+ if err := stop(); err != nil {
+ t.Fatalf("serve returned an error on shutdown: %v", err)
+ }
+
+ // Shutdown must have closed the listener, freeing the port.
+ reopened, err := net.Listen("tcp", addr)
+ if err != nil {
+ t.Fatalf("listener still bound to %s after shutdown: %v", addr, err)
+ }
+ _ = reopened.Close()
+}
+
+func TestServeWithAlreadyCancelledContext(t *testing.T) {
+ ln := newTestListener(t)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ // Shutdown can win the race against the serving goroutine here. That is
+ // safe: a server already told to shut down makes Serve return
+ // ErrServerClosed straight away, so nothing blocks.
+ done := make(chan error, 1)
+ go func() {
+ done <- serve(ctx, ln)
+ }()
+
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("serve on an already cancelled context returned: %v", err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatal("serve hung on an already cancelled context")
+ }
+}
+
+func TestServeReturnsErrorWhenAddressUnavailable(t *testing.T) {
+ ln := newTestListener(t)
+ defer ln.Close()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ // The port is already used in newTestListener, so Serve should return an error
+ if err := Serve(ctx, ServerConfig{Addr: ln.Addr().String()}); err == nil {
+ t.Fatal("Serve on an address already in use returned no error")
+ }
+}
+
+func TestRegisterPrometheusMetricsIsIdempotent(t *testing.T) {
+ RegisterPrometheusMetrics()
+
+ // Registering a collector that is already registered is an error,
+ // RegisterPrometheusMetrics should be guarded with sync.Once.
+ mustNotPanic(t, "repeated RegisterPrometheusMetrics call", RegisterPrometheusMetrics)
+
+ // Confirm the collectors were really registered.
+ err := prometheus.DefaultRegisterer.Register(ActiveServices)
+
+ var alreadyRegistered prometheus.AlreadyRegisteredError
+ if !errors.As(err, &alreadyRegistered) {
+ t.Fatalf("Register(ActiveServices) error = %v, want AlreadyRegisteredError", err)
+ }
+}
+
+// newTestListener binds a loopback listener on an arbitrary free port.
+func newTestListener(t *testing.T) net.Listener {
+ t.Helper()
+
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatalf("listening on a free loopback port: %v", err)
+ }
+
+ return ln
+}
+
+// startServer runs serve on ln and returns the base URL along with a stop
+// function that cancels the context and reports what serve returned.
+func startServer(t *testing.T, ln net.Listener) (string, func() error) {
+ t.Helper()
+
+ ctx, cancel := context.WithCancel(context.Background())
+ serveErr := make(chan error, 1)
+ go func() {
+ serveErr <- serve(ctx, ln)
+ }()
+
+ var (
+ once sync.Once
+ err error
+ )
+ stop := func() error {
+ once.Do(func() {
+ cancel()
+ select {
+ case err = <-serveErr:
+ case <-time.After(10 * time.Second):
+ err = errors.New("serve did not return after the context was cancelled")
+ }
+ })
+ return err
+ }
+ t.Cleanup(func() {
+ _ = stop()
+ })
+
+ return "http://" + ln.Addr().String(), stop
+}
+
+func get(t *testing.T, url string) (string, int) {
+ t.Helper()
+
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+ if err != nil {
+ t.Fatalf("building request for %s: %v", url, err)
+ }
+
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("GET %s: %v", url, err)
+ }
+ defer resp.Body.Close()
+
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ t.Fatalf("reading body of %s: %v", url, err)
+ }
+
+ return string(body), resp.StatusCode
+}
+
+// mustNotPanic reports a panic in fn as a test failure describing what
+// panicked, rather than letting it take down the test binary.
+func mustNotPanic(t *testing.T, what string, fn func()) {
+ t.Helper()
+
+ defer func() {
+ if r := recover(); r != nil {
+ t.Fatalf("%s panicked: %v", what, r)
+ }
+ }()
+
+ fn()
+}