From 24b81af43cec4d83cd8b32e6ac271d7e429d7391 Mon Sep 17 00:00:00 2001 From: Miek Gieben Date: Fri, 29 May 2026 10:23:07 +0200 Subject: [PATCH] feat: add :51090/metrics endpoint with Prometheus metrics listening on machine IP (#304) Signed-off-by: Miek Gieben Co-authored-by: Pasha Sviderski --- cmd/uncloudd/main.go | 3 ++ go.mod | 2 +- internal/machine/cluster.go | 13 +++++++ internal/machine/dns/server.go | 3 ++ internal/machine/machine.go | 4 +++ internal/machine/metrics/server.go | 58 ++++++++++++++++++++++++++++++ internal/metrics/metrics.go | 35 ++++++++++++++++++ test/e2e/service_test.go | 52 +++++++++++++++++++++++++++ 8 files changed, 169 insertions(+), 1 deletion(-) create mode 100644 internal/machine/metrics/server.go create mode 100644 internal/metrics/metrics.go diff --git a/cmd/uncloudd/main.go b/cmd/uncloudd/main.go index 5b6e2b37..3ce48fdf 100644 --- a/cmd/uncloudd/main.go +++ b/cmd/uncloudd/main.go @@ -11,6 +11,7 @@ import ( "github.com/psviderski/uncloud/internal/daemon" "github.com/psviderski/uncloud/internal/log" "github.com/psviderski/uncloud/internal/machine" + "github.com/psviderski/uncloud/internal/metrics" "github.com/psviderski/uncloud/internal/version" "github.com/spf13/cobra" ) @@ -46,6 +47,8 @@ func main() { cmd.AddCommand(newDialStdioCommand()) cmd.AddCommand(newVersionCommand()) + metrics.Version.WithLabelValues(version.String()).Set(1) + // ctx is canceled when the daemon command is interrupted. ctx, cancel := context.WithCancel(context.Background()) diff --git a/go.mod b/go.mod index 3914714d..9b717a28 100644 --- a/go.mod +++ b/go.mod @@ -37,6 +37,7 @@ require ( github.com/moby/term v0.5.2 github.com/opencontainers/go-digest v1.0.0 github.com/opencontainers/image-spec v1.1.1 + github.com/prometheus/client_golang v1.22.0 github.com/psviderski/unregistry v0.4.1 github.com/siderolabs/grpc-proxy v0.5.1 github.com/spf13/cobra v1.10.1 @@ -247,7 +248,6 @@ require ( github.com/pkg/errors v0.9.1 // 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_golang v1.22.0 // indirect github.com/prometheus/client_model v0.6.1 // indirect github.com/prometheus/common v0.62.0 // indirect github.com/prometheus/procfs v0.15.1 // indirect diff --git a/internal/machine/cluster.go b/internal/machine/cluster.go index b82edc70..e6a04ca4 100644 --- a/internal/machine/cluster.go +++ b/internal/machine/cluster.go @@ -20,6 +20,7 @@ import ( "github.com/psviderski/uncloud/internal/machine/dns" "github.com/psviderski/uncloud/internal/machine/docker" "github.com/psviderski/uncloud/internal/machine/firewall" + "github.com/psviderski/uncloud/internal/machine/metrics" "github.com/psviderski/uncloud/internal/machine/network" "github.com/psviderski/uncloud/internal/machine/store" "github.com/psviderski/unregistry" @@ -55,6 +56,8 @@ type clusterController struct { // unregistry is the embedded container registry that uses the local Docker (containerd) image store as its backend. unregistry *unregistry.Registry + metricsServer *metrics.Server + // stopped is a channel that is closed when the controller is stopped. stopped chan struct{} } @@ -72,6 +75,7 @@ func newClusterController( dnsServer *dns.Server, dnsResolver *dns.ClusterResolver, unregistry *unregistry.Registry, + metricsServer *metrics.Server, ) (*clusterController, error) { slog.Info("Starting WireGuard network.") wgnet, err := network.NewWireGuardNetwork() @@ -95,6 +99,7 @@ func newClusterController( dnsServer: dnsServer, dnsResolver: dnsResolver, unregistry: unregistry, + metricsServer: metricsServer, stopped: make(chan struct{}), }, nil } @@ -186,6 +191,14 @@ func (cc *clusterController) Run(ctx context.Context) error { return err } + errGroup.Go(func() error { + slog.Info("Starting metrics server.") + if err := cc.metricsServer.Run(ctx); err != nil { + return fmt.Errorf("metrics server failed: %w", err) + } + return nil + }) + errGroup.Go(func() error { slog.Info("Starting embedded DNS resolver.") if err := cc.dnsResolver.Run(ctx); err != nil { diff --git a/internal/machine/dns/server.go b/internal/machine/dns/server.go index 31018ac9..c93bba30 100644 --- a/internal/machine/dns/server.go +++ b/internal/machine/dns/server.go @@ -15,6 +15,7 @@ import ( "time" "github.com/miekg/dns" + "github.com/psviderski/uncloud/internal/metrics" ) const ( @@ -196,6 +197,7 @@ func (s *Server) handleRequest(w dns.ResponseWriter, req *dns.Msg) { log.Error("Failed to forward DNS query.", "err", err) resp = new(dns.Msg).SetRcode(req, dns.RcodeServerFailure) } + metrics.DNSQuery.WithLabelValues("false", metrics.Status(err)).Inc() s.reply(w, req, resp) return @@ -232,6 +234,7 @@ func (s *Server) handleRequest(w dns.ResponseWriter, req *dns.Msg) { } } resp.Truncate(maxSize) + metrics.DNSQuery.WithLabelValues("true", metrics.Ok).Inc() // NameError is not an error s.reply(w, req, resp) } diff --git a/internal/machine/machine.go b/internal/machine/machine.go index 5cd05aac..ff274681 100644 --- a/internal/machine/machine.go +++ b/internal/machine/machine.go @@ -33,6 +33,7 @@ import ( "github.com/psviderski/uncloud/internal/machine/corroservice" "github.com/psviderski/uncloud/internal/machine/dns" machinedocker "github.com/psviderski/uncloud/internal/machine/docker" + "github.com/psviderski/uncloud/internal/machine/metrics" "github.com/psviderski/uncloud/internal/machine/network" "github.com/psviderski/uncloud/internal/machine/store" "github.com/psviderski/uncloud/internal/secret" @@ -483,6 +484,8 @@ func (m *Machine) Run(ctx context.Context) error { return fmt.Errorf("create embedded DNS server: %w", err) } + metricsServer := metrics.New(m.IP()) + var unreg *unregistry.Registry if containerdSock := m.ContainerdSock(); containerdSock != "" { isContainerdStore, err := m.dockerService.IsContainerdImageStoreEnabled(ctx) @@ -524,6 +527,7 @@ func (m *Machine) Run(ctx context.Context) error { dnsServer, dnsResolver, unreg, + metricsServer, ) m.mu.Unlock() if err != nil { diff --git a/internal/machine/metrics/server.go b/internal/machine/metrics/server.go new file mode 100644 index 00000000..b52583a2 --- /dev/null +++ b/internal/machine/metrics/server.go @@ -0,0 +1,58 @@ +package metrics + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "net/netip" + "strconv" + "time" + + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +const Port = 51090 + +type Server struct { + *http.Server + listenAddr netip.Addr + log *slog.Logger +} + +func New(listenAddr netip.Addr) *Server { + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.Handler()) + server := &http.Server{Handler: mux, ReadTimeout: 5 * time.Second} + return &Server{server, listenAddr, slog.With("component", "metrics")} +} + +func (s *Server) Run(ctx context.Context) error { + addr := net.JoinHostPort(s.listenAddr.String(), strconv.Itoa(Port)) + l, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("listen metrics server: %w", err) + } + + errCh := make(chan error, 1) + + go func() { + if err := s.Serve(l); err != nil && !errors.Is(err, http.ErrServerClosed) { + errCh <- fmt.Errorf("metrics server failed: %w", err) + } + }() + + select { + case err := <-errCh: + // The server failed on its own so there are no connections to drain. + return err + case <-ctx.Done(): + s.log.Info("Stopping metrics server.") + // ctx is already cancelled so a graceful drain needs a new one with a reasonable timeout. + stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return s.Shutdown(stopCtx) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 00000000..205e04ab --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,35 @@ +// Package metrics declares all metrics Uncloud uses. +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var ( + Version = promauto.NewGaugeVec(prometheus.GaugeOpts{ + Namespace: Namespace, Subsystem: "uncloudd", + Name: "build_info", + Help: "Build information.", + }, []string{"version"}) + DNSQuery = promauto.NewCounterVec(prometheus.CounterOpts{ + Namespace: Namespace, Subsystem: "dns", + Name: "query_total", + Help: "Counter of DNS queries.", + }, []string{"internal", "status"}) +) + +const ( + Err = "err" + Ok = "ok" +) + +// Status returns "ok" if err is nil, otherwise "err". +func Status(err error) string { + if err != nil { + return Err + } + return Ok +} + +const Namespace = "uncloud" diff --git a/test/e2e/service_test.go b/test/e2e/service_test.go index 685411eb..b2ff175a 100644 --- a/test/e2e/service_test.go +++ b/test/e2e/service_test.go @@ -4,9 +4,11 @@ import ( "context" "errors" "fmt" + "net" "net/netip" "regexp" "slices" + "strconv" "strings" "testing" "time" @@ -16,6 +18,8 @@ import ( "github.com/docker/docker/api/types/volume" "github.com/docker/go-units" "github.com/psviderski/uncloud/internal/machine/api/pb" + "github.com/psviderski/uncloud/internal/machine/metrics" + "github.com/psviderski/uncloud/internal/machine/network" "github.com/psviderski/uncloud/internal/secret" "github.com/psviderski/uncloud/internal/ucind" "github.com/psviderski/uncloud/pkg/api" @@ -2223,4 +2227,52 @@ func TestServiceLifecycle(t *testing.T) { } }) }) + + t.Run("metrics endpoint on machine returns prometheus metrics", func(t *testing.T) { + t.Parallel() + + // Deploy a test service to run wget from. + wgetServiceName := "test-metrics-service" + t.Cleanup(func() { + // this races with the cluster cleanup, ignore error + cli.RemoveService(ctx, wgetServiceName) + }) + wgetSvcSpec := api.ServiceSpec{ + Name: wgetServiceName, + Mode: api.ServiceModeReplicated, + Replicas: 1, + Placement: api.Placement{ + Machines: []string{c.Machines[0].Name}, + }, + Container: api.ContainerSpec{ + Image: "busybox:1.37.0-musl", + Command: []string{"sleep", "infinity"}, + }, + } + _, err = cli.RunService(ctx, wgetSvcSpec) + require.NoError(t, err) + + wgetSvc, err := cli.InspectService(ctx, wgetServiceName) + require.NoError(t, err) + wgetContainer := wgetSvc.Containers[0] + + runWget := func(t *testing.T, url string) string { + metricsOutput, err := execInContainerAndReadOutput( + t, ctx, cli, wgetServiceName, wgetContainer.Container.ID, + []string{"wget", "-q", "-O", "-", url}, + ) + require.NoError(t, err) + return metricsOutput + } + + t.Run("metrics version is available", func(t *testing.T) { + // sub-sub-test to reuse the current deployment + metricsIP := network.MachineIP(netip.PrefixFrom(wgetContainer.Container.UncloudNetworkIP(), 24)).String() + endpoint := net.JoinHostPort(metricsIP, strconv.Itoa(metrics.Port)) + metricsOutput := runWget(t, "http://"+endpoint+"/metrics") + t.Logf("metrics output from %s:\n%s", endpoint, metricsOutput) + + assert.Contains(t, metricsOutput, "uncloud_uncloudd_build_info") + }) + }) }