feat(metrics): add exposing metrics functionality

cmd/serf/command/agent/agent.go

@@ -4,15 +4,20 @@
 package agent
 
 import (
+	"context"
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"log"
+	"net/http"
 	"os"
 	"strings"
 	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
 
 	"github.com/hashicorp/memberlist"
 	"github.com/hashicorp/serf/serf"
@@ -42,6 +47,9 @@ type Agent struct {
 	// This is the underlying Serf we are wrapping
 	serf *serf.Serf
 
+	// Prometheus server instance
+	promServer *http.Server
+
 	// shutdownCh is used for shutdowns
 	shutdown     bool
 	shutdownCh   chan struct{}
@@ -104,6 +112,19 @@ func (a *Agent) Start() error {
 	}
 	a.serf = serf
 
+	if a.agentConf.EnableMetrics {
+		a.logger.Printf("[INFO] agent: Starting Prometheus HTTP server on %s", a.agentConf.MetricsBindAddr)
+		mux := http.NewServeMux()
+		mux.Handle("/metrics", promhttp.Handler())
+		server := &http.Server{Addr: a.agentConf.MetricsBindAddr, Handler: mux}
+		go func() {
+			if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+				a.logger.Printf("[ERROR] agent: Prometheus HTTP server error: %v", err)
+			}
+		}()
+		a.promServer = server
+	}
+
 	// Start event loop
 	go a.eventLoop()
 	return nil
@@ -138,6 +159,16 @@ func (a *Agent) Shutdown() error {
 		return err
 	}
 
+	// Shutdown Prometheus HTTP server
+	if a.promServer != nil {
+		a.logger.Println("[INFO] agent: shutting down Prometheus HTTP server")
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := a.promServer.Shutdown(ctx); err != nil {
+			a.logger.Printf("[ERROR] agent: Prometheus HTTP server shutdown error: %v", err)
+		}
+	}
+
 EXIT:
 	a.logger.Println("[INFO] agent: shutdown complete")
 	a.shutdown = true
@@ -252,13 +283,18 @@ func (a *Agent) DeregisterEventHandler(eh EventHandler) {
 // eventLoop listens to events from Serf and fans out to event handlers
 func (a *Agent) eventLoop() {
 	serfShutdownCh := a.serf.ShutdownCh()
+	var extraHandlers []EventHandler
+	if a.agentConf.EnableMetrics {
+		extraHandlers = append(extraHandlers, NewMetricsEventHandler(a))
+	}
 	for {
 		select {
 		case e := <-a.eventCh:
 			a.logger.Printf("[INFO] agent: Received event: %s", e.String())
 			a.eventHandlersLock.Lock()
 			handlers := a.eventHandlerList
 			a.eventHandlersLock.Unlock()
+			handlers = append(handlers, extraHandlers...)
 			for _, eh := range handlers {
 				eh.HandleEvent(e)
 			}

cmd/serf/command/agent/command.go

@@ -100,6 +100,8 @@ func (c *Command) readConfig() *Config {
 	cmdFlags.StringVar(&retryInterval, "retry-interval", "", "retry join interval")
 	cmdFlags.BoolVar(&cmdConfig.RejoinAfterLeave, "rejoin", false,
 		"enable re-joining after a previous leave")
+	cmdFlags.BoolVar(&cmdConfig.EnableMetrics, "metrics", false, "expose Prometheus formatted metrics")
+	cmdFlags.StringVar(&cmdConfig.MetricsBindAddr, "metrics-addr", "0.0.0.0:10073", "address to bind metrics listener to")
 
 	cmdFlags.BoolVar(
 		&disableCompression,
@@ -197,6 +199,13 @@ func (c *Command) readConfig() *Config {
 		}
 	}
 
+	if config.EnableMetrics {
+		if _, _, err := net.SplitHostPort(config.MetricsBindAddr); err != nil {
+			c.Ui.Error(fmt.Sprintf("Invalid metrics bind address %q: %s", config.MetricsBindAddr, err))
+			return nil
+		}
+	}
+
 	// Backward compatibility hack for 'Role'
 	if config.Role != "" {
 		c.Ui.Output("Deprecation warning: 'Role' has been replaced with 'Tags'")
@@ -492,6 +501,7 @@ func (c *Command) startAgent(config *Config, agent *Agent,
 	c.Ui.Info(fmt.Sprintf("                   Snapshot: %v", config.SnapshotPath != ""))
 	c.Ui.Info(fmt.Sprintf("                    Profile: %s", config.Profile))
 	c.Ui.Info(fmt.Sprintf("Message Compression Enabled: %v", config.EnableCompression))
+	c.Ui.Info(fmt.Sprintf("         Prometheus Metrics: %v", config.EnableMetrics))
 
 	if config.Discover != "" {
 		c.Ui.Info(fmt.Sprintf("               mDNS cluster: %s", config.Discover))

cmd/serf/command/agent/config.go

@@ -235,6 +235,12 @@ type Config struct {
 	// metrics will be sent to that instance.
 	StatsdAddr string `mapstructure:"statsd_addr"`
 
+	// EnableMetrics is used to expose prometheus formated metrics
+	EnableMetrics bool `mapstructure:"enable_metrics"`
+
+	// MetricsBindAddr is the address to bind the metrics server to
+	MetricsBindAddr string `mapstructure:"metrics_bind_addr"`
+
 	// BroadcastTimeoutRaw is the string retry interval. This interval
 	// controls the timeout for broadcast events. This defaults to
 	// 5 seconds.
@@ -517,6 +523,15 @@ func MergeConfig(a, b *Config) *Config {
 	if b.UserEventSizeLimit != 0 {
 		result.UserEventSizeLimit = b.UserEventSizeLimit
 	}
+
+	if b.EnableMetrics == true {
+		result.EnableMetrics = true
+	}
+
+	if b.MetricsBindAddr != "" {
+		result.MetricsBindAddr = b.MetricsBindAddr
+	}
+
 	if b.BroadcastTimeout != 0 {
 		result.BroadcastTimeout = b.BroadcastTimeout
 	}

cmd/serf/command/agent/metrics.go

@@ -0,0 +1,184 @@
+package agent
+
+import (
+	"fmt"
+	"slices"
+	"sort"
+	"sync/atomic"
+
+	"github.com/hashicorp/serf/serf"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+const (
+	EXPORTER_ROLE = "metrics_exporter"
+	ROLE_TAG      = "role"
+)
+
+type MetricsEventHandler struct {
+	agent                   *Agent
+	serfEventReceived       *prometheus.CounterVec
+	serfNodeStatus          *prometheus.GaugeVec
+	serfMetricsExporterRole prometheus.Gauge
+	// metricExporter is true when this node is the elected exporter.
+	// Written by the event loop goroutine, read by the Prometheus scrape goroutine.
+	metricExporter atomic.Bool
+}
+
+func NewMetricsEventHandler(a *Agent) *MetricsEventHandler {
+	handler := &MetricsEventHandler{
+		agent: a,
+		serfEventReceived: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "serf_event_received",
+			Help: "Number of serf events received, by event type",
+		}, []string{"event_type"}),
+		serfNodeStatus: prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "serf_node_status",
+			Help: fmt.Sprintf("Status of each serf cluster member as seen by the elected exporter: %d=alive, %d=leaving, %d=left, %d=failed",
+				serf.StatusAlive, serf.StatusLeaving, serf.StatusLeft, serf.StatusFailed),
+		}, []string{"node"}),
+		serfMetricsExporterRole: prometheus.NewGauge(prometheus.GaugeOpts{
+			Name: "serf_metrics_exporter",
+			Help: "1 if this node is the elected metrics exporter for the cluster, 0 otherwise",
+		}),
+	}
+
+	prometheus.MustRegister(handler)
+	return handler
+}
+
+// HandleEvent implements EventHandler. Election is re-evaluated only on
+// membership events. User events and queries do not change cluster membership
+// and should not trigger tag gossip.
+func (m *MetricsEventHandler) HandleEvent(e serf.Event) {
+	m.serfEventReceived.WithLabelValues(e.EventType().String()).Inc()
+
+	switch e.EventType() {
+	case serf.EventMemberJoin,
+		serf.EventMemberLeave,
+		serf.EventMemberFailed,
+		serf.EventMemberUpdate,
+		serf.EventMemberReap:
+		m.evaluateExporterRole()
+	}
+}
+
+// evaluateExporterRole checks whether this node should be the metrics exporter
+// and updates the serf role tag only when the role actually changes, to avoid
+// unnecessary gossip churn.
+//
+// A node that sees fewer than 2 alive peers is considered potentially isolated
+// (split-brain) and must not claim to be the authoritative cluster exporter.
+func (m *MetricsEventHandler) evaluateExporterRole() {
+	exporters := m.exporters()
+	aliveNodes := m.aliveNodes()
+	localName := m.agent.serf.LocalMember().Name
+	isExporter := slices.Contains(exporters, localName)
+
+	// Step down if isolated: seeing only yourself means you may be partitioned
+	// from the rest of the cluster and cannot speak for it.
+	if len(aliveNodes) < 2 {
+		if isExporter {
+			m.setExporterRole(false)
+		}
+		return
+	}
+
+	switch {
+	case len(exporters) == 0:
+		// No exporter in the cluster: elect the alphabetically first alive node.
+		m.electMetricsExporter(aliveNodes)
+
+	case len(exporters) > 1 && isExporter:
+		// Multiple exporters (transient during gossip convergence):
+		// all but the highest-priority one step down.
+		if !m.hasPriority(exporters) {
+			m.setExporterRole(false)
+		}
+	}
+}
+
+func (m *MetricsEventHandler) electMetricsExporter(aliveNodes []string) {
+	sort.Strings(aliveNodes)
+	if aliveNodes[0] == m.agent.serf.LocalMember().Name {
+		m.setExporterRole(true)
+	}
+}
+
+// setExporterRole updates the serf role tag and the local flag only when the
+// role actually changes, to avoid gossip churn on every membership event.
+func (m *MetricsEventHandler) setExporterRole(exporter bool) {
+	if exporter {
+		if m.metricExporter.Load() {
+			return // already exporter, no change
+		}
+		m.agent.serf.SetTags(map[string]string{ROLE_TAG: EXPORTER_ROLE})
+		m.metricExporter.Store(true)
+	} else {
+		if !m.metricExporter.Load() {
+			return // already not exporter, no change
+		}
+		// Copy the tags map before mutating to avoid modifying serf's internal state.
+		tags := make(map[string]string)
+		for k, v := range m.agent.serf.LocalMember().Tags {
+			tags[k] = v
+		}
+		delete(tags, ROLE_TAG)
+		m.agent.serf.SetTags(tags)
+		m.metricExporter.Store(false)
+	}
+}
+
+func (m *MetricsEventHandler) aliveNodes() []string {
+	var aliveNodes []string
+	for _, member := range m.agent.serf.Members() {
+		if member.Status == serf.StatusAlive {
+			aliveNodes = append(aliveNodes, member.Name)
+		}
+	}
+	return aliveNodes
+}
+
+func (m *MetricsEventHandler) exporters() []string {
+	var exporters []string
+	for _, member := range m.agent.serf.Members() {
+		if role, ok := member.Tags[ROLE_TAG]; ok && role == EXPORTER_ROLE && member.Status == serf.StatusAlive {
+			exporters = append(exporters, member.Name)
+		}
+	}
+	return exporters
+}
+
+func (m *MetricsEventHandler) hasPriority(exporters []string) bool {
+	sort.Strings(exporters)
+	return exporters[0] == m.agent.serf.LocalMember().Name
+}
+
+func (m *MetricsEventHandler) Collect(ch chan<- prometheus.Metric) {
+	if m.metricExporter.Load() {
+		m.serfMetricsExporterRole.Set(1)
+		// Reset before re-populating so nodes that left the cluster don't
+		// leave behind stale time series.
+		m.serfNodeStatus.Reset()
+		for _, member := range m.agent.serf.Members() {
+			if member.Status == serf.StatusNone {
+				continue // transient/unknown omit rather than emit a misleading value
+			}
+			m.serfNodeStatus.WithLabelValues(member.Name).Set(float64(member.Status))
+		}
+		m.serfNodeStatus.Collect(ch)
+	} else {
+		m.serfMetricsExporterRole.Set(0)
+	}
+
+	m.serfEventReceived.Collect(ch)
+	m.serfMetricsExporterRole.Collect(ch)
+}
+
+func (m *MetricsEventHandler) Describe(ch chan<- *prometheus.Desc) {
+	if m.metricExporter.Load() {
+		m.serfNodeStatus.Describe(ch)
+	}
+	m.serfEventReceived.Describe(ch)
+	m.serfMetricsExporterRole.Describe(ch)
+}

go.mod

@@ -1,6 +1,6 @@
 module github.com/hashicorp/serf
 
-go 1.19
+go 1.21
 
 require (
 	github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e
@@ -12,6 +12,7 @@ require (
 	github.com/hashicorp/memberlist v0.5.2
 	github.com/mitchellh/cli v1.1.5
 	github.com/mitchellh/mapstructure v1.5.0
+	github.com/prometheus/client_golang v1.11.1
 	github.com/ryanuber/columnize v2.1.2+incompatible
 )
 
@@ -21,8 +22,11 @@ require (
 	github.com/Masterminds/sprig/v3 v3.2.1 // indirect
 	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/armon/go-radix v1.0.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bgentry/speakeasy v0.1.0 // indirect
+	github.com/cespare/xxhash/v2 v2.1.1 // indirect
 	github.com/fatih/color v1.9.0 // indirect
+	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/btree v1.1.2 // indirect
 	github.com/google/uuid v1.1.2 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
@@ -35,10 +39,14 @@ require (
 	github.com/imdario/mergo v0.3.11 // indirect
 	github.com/mattn/go-colorable v0.1.6 // indirect
 	github.com/mattn/go-isatty v0.0.12 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/miekg/dns v1.1.56 // indirect
 	github.com/mitchellh/copystructure v1.0.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.0 // indirect
 	github.com/posener/complete v1.2.3 // indirect
+	github.com/prometheus/client_model v0.2.0 // indirect
+	github.com/prometheus/common v0.26.0 // indirect
+	github.com/prometheus/procfs v0.6.0 // indirect
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
 	github.com/shopspring/decimal v1.2.0 // indirect
 	github.com/spf13/cast v1.3.1 // indirect
@@ -47,4 +55,5 @@ require (
 	golang.org/x/net v0.35.0 // indirect
 	golang.org/x/sys v0.30.0 // indirect
 	golang.org/x/tools v0.14.0 // indirect
+	google.golang.org/protobuf v1.28.1 // indirect
 )