From 85147522a9324419d3393e55ab40ac7fce4037f2 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 14 Sep 2026 21:19:30 -0700 Subject: [PATCH] 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 | 2 +- weed/admin/dash/admin_server.go | 6 ++ weed/admin/dash/metrics_scraper.go | 153 +++++++++++++++++++++++++++++ 3 files changed, 160 insertions(+), 1 deletion(-) create mode 100644 weed/admin/dash/metrics_scraper.go diff --git a/go.mod b/go.mod index 0fe155f66..430178dc1 100644 --- a/go.mod +++ b/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 diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index d015d571c..34ca3be8c 100644 --- a/weed/admin/dash/admin_server.go +++ b/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 } diff --git a/weed/admin/dash/metrics_scraper.go b/weed/admin/dash/metrics_scraper.go new file mode 100644 index 000000000..f50ff1852 --- /dev/null +++ b/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) + } + } +}