Add global labels to federation

cmd/prometheus/main.go

@@ -93,7 +93,9 @@ func Main() int {
 
 	webHandler := web.New(memStorage, queryEngine, ruleManager, status, &cfg.web)
 
-	if !reloadConfig(cfg.configFile, status, targetManager, ruleManager) {
+	reloadables := []Reloadable{status, targetManager, ruleManager, webHandler}
+
+	if !reloadConfig(cfg.configFile, reloadables...) {
 		return 1
 	}
 
@@ -110,7 +112,7 @@ func Main() int {
 			case <-hup:
 			case <-webHandler.Reload():
 			}
-			reloadConfig(cfg.configFile, status, targetManager, ruleManager)
+			reloadConfig(cfg.configFile, reloadables...)
 		}
 	}()
 

web/federate.go

@@ -19,20 +19,18 @@ import (
 	"github.com/golang/protobuf/proto"
 
 	"github.com/prometheus/prometheus/promql"
-	"github.com/prometheus/prometheus/storage/local"
 	"github.com/prometheus/prometheus/storage/metric"
 
-	dto "github.com/prometheus/client_model/go"
 	"github.com/prometheus/common/expfmt"
 	"github.com/prometheus/common/model"
+
+	dto "github.com/prometheus/client_model/go"
 )
 
-// Federation implements a web handler to serve scrape federation requests.
+func (h *Handler) federation(w http.ResponseWriter, req *http.Request) {
-type Federation struct {
+	h.mtx.RLock()
-	Storage local.Storage
+	defer h.mtx.RUnlock()
-}
 
-func (fed *Federation) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	req.ParseForm()
 
 	metrics := map[model.Fingerprint]metric.Metric{}
@@ -43,7 +41,7 @@ func (fed *Federation) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 			http.Error(w, err.Error(), http.StatusBadRequest)
 			return
 		}
-		for fp, met := range fed.Storage.MetricsForLabelMatchers(matchers...) {
+		for fp, met := range h.storage.MetricsForLabelMatchers(matchers...) {
 			metrics[fp] = met
 		}
 	}
@@ -63,7 +61,9 @@ func (fed *Federation) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	}
 
 	for fp, met := range metrics {
-		sp := fed.Storage.LastSamplePairForFingerprint(fp)
+		globalUsed := map[model.LabelName]struct{}{}
+
+		sp := h.storage.LastSamplePairForFingerprint(fp)
 		if sp == nil {
 			continue
 		}
@@ -80,14 +80,27 @@ func (fed *Federation) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 				Name:  proto.String(string(ln)),
 				Value: proto.String(string(lv)),
 			})
+			if _, ok := h.globalLabels[ln]; ok {
+				globalUsed[ln] = struct{}{}
 			}
+		}
+
+		// Attach global labels if they do not exist yet.
+		for ln, lv := range h.globalLabels {
+			if _, ok := globalUsed[ln]; !ok {
+				protMetric.Label = append(protMetric.Label, &dto.LabelPair{
+					Name:  proto.String(string(ln)),
+					Value: proto.String(string(lv)),
+				})
+			}
+		}
+
 		protMetric.TimestampMs = (*int64)(&sp.Timestamp)
 		protMetric.Untyped.Value = (*float64)(&sp.Value)
 
 		if err := enc.Encode(protMetricFam); err != nil {
 			http.Error(w, err.Error(), http.StatusInternalServerError)
 			return
-
 		}
 	}
 }

web/web.go

@@ -54,10 +54,10 @@ var localhostRepresentations = []string{"127.0.0.1", "localhost"}
 type Handler struct {
 	ruleManager *rules.Manager
 	queryEngine *promql.Engine
+	storage     local.Storage
 
 	apiV1     *v1.API
 	apiLegacy *legacy.API
-	federation *Federation
 
 	router      *route.Router
 	listenErrCh chan error
@@ -66,7 +66,19 @@ type Handler struct {
 	options     *Options
 	statusInfo  *PrometheusStatus
 
-	muAlerts sync.Mutex
+	globalLabels model.LabelSet
+	mtx          sync.RWMutex
+}
+
+// ApplyConfig updates the status state as the new config requires.
+// Returns true on success.
+func (h *Handler) ApplyConfig(conf *config.Config) bool {
+	h.mtx.Lock()
+	defer h.mtx.Unlock()
+
+	h.globalLabels = conf.GlobalConfig.Labels
+
+	return true
 }
 
 // PrometheusStatus contains various information about the status
@@ -89,8 +101,10 @@ type PrometheusStatus struct {
 // Returns true on success.
 func (s *PrometheusStatus) ApplyConfig(conf *config.Config) bool {
 	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	s.Config = conf.String()
-	s.mu.Unlock()
+
 	return true
 }
 
@@ -120,6 +134,7 @@ func New(st local.Storage, qe *promql.Engine, rm *rules.Manager, status *Prometh
 
 		ruleManager: rm,
 		queryEngine: qe,
+		storage:     st,
 
 		apiV1: &v1.API{
 			QueryEngine: qe,
@@ -130,9 +145,6 @@ func New(st local.Storage, qe *promql.Engine, rm *rules.Manager, status *Prometh
 			Storage:     st,
 			Now:         model.Now,
 		},
-		federation: &Federation{
-			Storage: st,
-		},
 	}
 
 	if o.ExternalURL.Path != "" {
@@ -153,7 +165,7 @@ func New(st local.Storage, qe *promql.Engine, rm *rules.Manager, status *Prometh
 
 	router.Get("/heap", instrf("heap", dumpHeap))
 
-	router.Get("/federate", instrh("federate", h.federation))
+	router.Get("/federate", instrf("federate", h.federation))
 	router.Get(o.MetricsPath, prometheus.Handler().ServeHTTP)
 
 	h.apiLegacy.Register(router.WithPrefix("/api"))
@@ -205,9 +217,6 @@ func (h *Handler) Run() {
 }
 
 func (h *Handler) alerts(w http.ResponseWriter, r *http.Request) {
-	h.muAlerts.Lock()
-	defer h.muAlerts.Unlock()
-
 	alerts := h.ruleManager.AlertingRules()
 	alertsSorter := byAlertStateSorter{alerts: alerts}
 	sort.Sort(alertsSorter)