fanhuadesenlinnn/kube-vip
1
0
Fork 0
mirror of https://hubproxy.babadafafafafa.cn/https://github.com/kube-vip/kube-vip.git synced 2026-09-20 08:03:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactored and fixed metrics code

Browse Source 
Signed-off-by: Patryk Strusiewicz-Surmacki <patryk-pawel.strusiewicz-surmacki@external.telekom.de>
This commit is contained in:
Patryk Strusiewicz-Surmacki
2026-09-14 19:10:57 +02:00
committed by Marcel Fest
parent d99186480c
commit a60f80b1b2
5 changed files with 345 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

74
cmd/kube-vip.go
View File

@@ -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(`<html>
<head><title>kube-vip</title></head>
<body>
<h1>kube-vip Metrics</h1>
<p><a href="` + "/metrics" + `">Metrics</a></p>
</body>
</html>`))
})
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

11
pkg/manager/manager.go
View File

@@ -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
@@ -263,12 +258,6 @@ func New(ctx context.Context, configMap string, config *kubevip.Config) (*Manage
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"}),
signalChan: signalChan,
svcProcessor: svcProcessor,
intfMgr: intfMgr,

12
pkg/metrics/prometheus.go
View File

@@ -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,7 +72,12 @@ var (
)
)
var registerOnce sync.Once
// RegisterPrometheusMetrics registers all kube-vip metrics with the default
// Prometheus registry.
func RegisterPrometheusMetrics() {
registerOnce.Do(func() {
// Register all metrics with Prometheus
prometheus.MustRegister(
ActiveServices,
@@ -83,4 +92,5 @@ func RegisterPrometheusMetrics() {
BuildInfo,
CountServiceWatchEvent,
)
})
}

94
pkg/metrics/server.go Normal file
View File

@@ -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(`<html>
<head><title>kube-vip</title></head>
<body>
<h1>kube-vip Metrics</h1>
<p><a href="/metrics">Metrics</a></p>
</body>
</html>`))
})
return mux
}

213
pkg/metrics/server_test.go Normal file
View File

@@ -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()
}