admin: scrape per-server /metrics into the store

Adds a 15s scrape loop that fetches /metrics from every discovered
master, volume, filer, and S3 server (addresses come from the
existing topology + ListClusterNodes helpers) and records each
series into the in-memory store. Parses Prometheus text exposition
via prometheus/common/expfmt.

go.mod

@@ -60,7 +60,7 @@ require (
 	github.com/pquerna/cachecontrol v0.2.0
 	github.com/prometheus/client_golang v1.24.1
 	github.com/prometheus/client_model v0.6.3
-	github.com/prometheus/common v0.70.1 // indirect
+	github.com/prometheus/common v0.70.1
 	github.com/prometheus/procfs v0.22.0
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect

weed/admin/dash/admin_server.go

@@ -126,6 +126,10 @@ type AdminServer struct {
 	dashSamples   []dashSample
 	dashSamplesMu sync.Mutex
 
+	// metricsStore holds scraped per-server Prometheus series for the
+	// monitoring pages. Filled by the scrape loop in startMetricsScraper.
+	metricsStore *metricsStore
+
 	// Filer discovery and caching
 	cachedFilers         []string
 	lastFilerUpdate      time.Time
@@ -210,6 +214,7 @@ func NewAdminServer(masters string, filerGroup string, templateFS http.FileSyste
 		pluginLock:                    lockManager,
 		adminPresenceLock:             presenceLock,
 		bgCancel:                      bgCancel,
+		metricsStore:                  newMetricsStore(),
 	}
 
 	// Initialize topic retention purger
@@ -321,6 +326,7 @@ func NewAdminServer(masters string, filerGroup string, templateFS http.FileSyste
 	}
 
 	go server.publishMaintenanceMetrics(bgCtx)
+	go server.startMetricsScraper(bgCtx)
 
 	return server
 }

weed/admin/dash/metrics_scraper.go

@@ -0,0 +1,153 @@
+package dash
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	dto "github.com/prometheus/client_model/go"
+	"github.com/prometheus/common/expfmt"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
+)
+
+type scrapedMetric struct {
+	name   string
+	labels map[string]string
+	value  float64
+}
+
+func scrapeMetrics(ctx context.Context, target string) ([]scrapedMetric, error) {
+	if !strings.HasPrefix(target, "http://") && !strings.HasPrefix(target, "https://") {
+		target = "http://" + target
+	}
+	target = strings.TrimRight(target, "/") + "/metrics"
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Accept", string(expfmt.TextVersion))
+
+	client := util_http.GetGlobalHttpClient()
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("scrape %s: status %d", target, resp.StatusCode)
+	}
+
+	return parsePrometheusText(resp.Body)
+}
+
+func parsePrometheusText(r io.Reader) ([]scrapedMetric, error) {
+	dec := expfmt.NewDecoder(r, expfmt.NewFormat(expfmt.TypeTextPlain))
+	var out []scrapedMetric
+	for {
+		var fam dto.MetricFamily
+		if err := dec.Decode(&fam); err != nil {
+			if err == io.EOF {
+				break
+			}
+			return nil, err
+		}
+		for _, m := range fam.Metric {
+			labels := map[string]string{}
+			for _, l := range m.Label {
+				labels[l.GetName()] = l.GetValue()
+			}
+			var v float64
+			switch {
+			case m.Gauge != nil:
+				v = m.Gauge.GetValue()
+			case m.Counter != nil:
+				v = m.Counter.GetValue()
+			case m.Untyped != nil:
+				v = m.Untyped.GetValue()
+			case m.Histogram != nil:
+				v = m.Histogram.GetSampleSum()
+			case m.Summary != nil:
+				v = m.Summary.GetSampleSum()
+			}
+			out = append(out, scrapedMetric{name: fam.GetName(), labels: labels, value: v})
+		}
+	}
+	return out, nil
+}
+
+func (s *AdminServer) scrapeAllServers(ctx context.Context) {
+	targets := s.scrapeTargets()
+	if len(targets) == 0 {
+		return
+	}
+	now := time.Now()
+	type result struct {
+		source  string
+		metrics []scrapedMetric
+		err     error
+	}
+	results := make(chan result, len(targets))
+	scrapeCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	for _, t := range targets {
+		go func(t scrapeTarget) {
+			ms, err := scrapeMetrics(scrapeCtx, t.address)
+			results <- result{source: t.source, metrics: ms, err: err}
+		}(t)
+	}
+	for i := 0; i < len(targets); i++ {
+		r := <-results
+		if r.err != nil {
+			glog.V(1).Infof("metrics scrape %s: %v", r.source, r.err)
+			continue
+		}
+		for _, m := range r.metrics {
+			s.metricsStore.recordLabeled(r.source, m.name, m.labels, m.value, now)
+		}
+	}
+}
+
+type scrapeTarget struct {
+	source  string
+	address string
+}
+
+func (s *AdminServer) scrapeTargets() []scrapeTarget {
+	var out []scrapeTarget
+	if topo, err := s.GetClusterTopology(); err == nil && topo != nil {
+		for _, m := range topo.Masters {
+			out = append(out, scrapeTarget{source: "master/" + m.Address, address: m.Address})
+		}
+		for _, vs := range topo.VolumeServers {
+			out = append(out, scrapeTarget{source: "volume/" + vs.Address, address: vs.Address})
+		}
+	}
+	for _, f := range s.getFilerNodesStatus() {
+		out = append(out, scrapeTarget{source: "filer/" + f.Address, address: f.Address})
+	}
+	for _, n := range s.getS3NodesStatus() {
+		out = append(out, scrapeTarget{source: "s3/" + n.Address, address: n.Address})
+	}
+	return out
+}
+
+func (s *AdminServer) startMetricsScraper(ctx context.Context) {
+	const interval = 15 * time.Second
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	s.scrapeAllServers(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			s.scrapeAllServers(ctx)
+		}
+	}
+}