Distributors rings status (cortexproject#4151)

* Distributors rings status

Now we're able to see the ring status for the distributor when the ingestion-rate-limit-strategy
setting is set to global

Signed-off-by: Mario de Frutos <mario@defrutos.org>

* Fix lint error

Signed-off-by: Mario de Frutos <mario@defrutos.org>

* Use the proper prometheus registerer

Signed-off-by: Mario de Frutos <mario@defrutos.org>

* Improve error message

Signed-off-by: Mario de Frutos <mario@defrutos.org>

* Create util.WriteHTMLResponse and use it

Signed-off-by: Mario de Frutos <mario@defrutos.org>

Author: ethervoid

CHANGELOG.md

@@ -6,6 +6,7 @@
 * [CHANGE] Alertmanager: allowed to configure the experimental receivers firewall on a per-tenant basis. The following CLI flags (and their respective YAML config options) have been changed and moved to the limits config section: #4143
   - `-alertmanager.receivers-firewall.block.cidr-networks` renamed to `-alertmanager.receivers-firewall-block-cidr-networks`
   - `-alertmanager.receivers-firewall.block.private-addresses` renamed to `-alertmanager.receivers-firewall-block-private-addresses`
+* [CHANGE] Distributor: Added ring status section in the admin page #4151
 * [ENHANCEMENT] Alertmanager: introduced new metrics to monitor operation when using `-alertmanager.sharding-enabled`: #4149
   * `cortex_alertmanager_state_fetch_replica_state_total`
   * `cortex_alertmanager_state_fetch_replica_state_failed_total`

docs/api/_index.md

@@ -213,6 +213,14 @@ _For more information, please check out Prometheus [Remote storage integrations]
 
 _Requires [authentication](#authentication)._
 
+### Distributor ring status
+
+```
+GET /distributor/ring
+```
+
+Displays a web page with the distributor hash ring status, including the state, healthy and last heartbeat time of each distributor.
+
 ### Tenants stats
 
 ```

pkg/api/api.go

@@ -217,9 +217,11 @@ func (a *API) RegisterDistributor(d *distributor.Distributor, pushConfig distrib
 
 	a.RegisterRoute("/api/v1/push", push.Handler(pushConfig.MaxRecvMsgSize, a.sourceIPs, a.cfg.wrapDistributorPush(d)), true, "POST")
 
+	a.indexPage.AddLink(SectionAdminEndpoints, "/distributor/ring", "Distributor Ring Status")
 	a.indexPage.AddLink(SectionAdminEndpoints, "/distributor/all_user_stats", "Usage Statistics")
 	a.indexPage.AddLink(SectionAdminEndpoints, "/distributor/ha_tracker", "HA Tracking Status")
 
+	a.RegisterRoute("/distributor/ring", d, false, "GET", "POST")
 	a.RegisterRoute("/distributor/all_user_stats", http.HandlerFunc(d.AllUserStatsHandler), false, "GET")
 	a.RegisterRoute("/distributor/ha_tracker", d.HATracker, false, "GET")
 

pkg/distributor/distributor.go

@@ -73,7 +73,8 @@ type Distributor struct {
 
 	// The global rate limiter requires a distributors ring to count
 	// the number of healthy instances
-	distributorsRing *ring.Lifecycler
+	distributorsLifeCycler *ring.Lifecycler
+	distributorsRing       *ring.Ring
 
 	// For handling HA replicas.
 	HATracker *haTracker
@@ -202,33 +203,39 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 	// it's an internal dependency and can't join the distributors ring, we skip rate
 	// limiting.
 	var ingestionRateStrategy limiter.RateLimiterStrategy
-	var distributorsRing *ring.Lifecycler
+	var distributorsLifeCycler *ring.Lifecycler
+	var distributorsRing *ring.Ring
 
 	if !canJoinDistributorsRing {
 		ingestionRateStrategy = newInfiniteIngestionRateStrategy()
 	} else if limits.IngestionRateStrategy() == validation.GlobalIngestionRateStrategy {
-		distributorsRing, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ring.DistributorRingKey, true, reg)
+		distributorsLifeCycler, err = ring.NewLifecycler(cfg.DistributorRing.ToLifecyclerConfig(), nil, "distributor", ring.DistributorRingKey, true, reg)
 		if err != nil {
 			return nil, err
 		}
 
-		subservices = append(subservices, distributorsRing)
+		distributorsRing, err = ring.New(cfg.DistributorRing.ToRingConfig(), "distributor", ring.DistributorRingKey, reg)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to initialize distributors' ring client")
+		}
+		subservices = append(subservices, distributorsLifeCycler, distributorsRing)
 
-		ingestionRateStrategy = newGlobalIngestionRateStrategy(limits, distributorsRing)
+		ingestionRateStrategy = newGlobalIngestionRateStrategy(limits, distributorsLifeCycler)
 	} else {
 		ingestionRateStrategy = newLocalIngestionRateStrategy(limits)
 	}
 
 	d := &Distributor{
-		cfg:                  cfg,
-		log:                  log,
-		ingestersRing:        ingestersRing,
-		ingesterPool:         NewPool(cfg.PoolConfig, ingestersRing, cfg.IngesterClientFactory, log),
-		distributorsRing:     distributorsRing,
-		limits:               limits,
-		ingestionRateLimiter: limiter.NewRateLimiter(ingestionRateStrategy, 10*time.Second),
-		HATracker:            haTracker,
-		ingestionRate:        util_math.NewEWMARate(0.2, instanceIngestionRateTickInterval),
+		cfg:                    cfg,
+		log:                    log,
+		ingestersRing:          ingestersRing,
+		ingesterPool:           NewPool(cfg.PoolConfig, ingestersRing, cfg.IngesterClientFactory, log),
+		distributorsLifeCycler: distributorsLifeCycler,
+		distributorsRing:       distributorsRing,
+		limits:                 limits,
+		ingestionRateLimiter:   limiter.NewRateLimiter(ingestionRateStrategy, 10*time.Second),
+		HATracker:              haTracker,
+		ingestionRate:          util_math.NewEWMARate(0.2, instanceIngestionRateTickInterval),
 
 		queryDuration: instrument.NewHistogramCollector(promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: "cortex",
@@ -1029,3 +1036,23 @@ func (d *Distributor) AllUserStats(ctx context.Context) ([]UserIDStats, error) {
 
 	return response, nil
 }
+
+func (d *Distributor) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	if d.distributorsRing != nil {
+		d.distributorsRing.ServeHTTP(w, req)
+	} else {
+		var ringNotEnabledPage = `
+			<!DOCTYPE html>
+			<html>
+				<head>
+					<meta charset="UTF-8">
+					<title>Cortex Distributor Status</title>
+				</head>
+				<body>
+					<h1>Cortex Distributor Status</h1>
+					<p>Distributor is not running with global limits enabled</p>
+				</body>
+			</html>`
+		util.WriteHTMLResponse(w, ringNotEnabledPage)
+	}
+}

pkg/distributor/distributor_ring.go

@@ -86,3 +86,14 @@ func (cfg *RingConfig) ToLifecyclerConfig() ring.LifecyclerConfig {
 
 	return lc
 }
+
+func (cfg *RingConfig) ToRingConfig() ring.Config {
+	rc := ring.Config{}
+	flagext.DefaultValues(&rc)
+
+	rc.KVStore = cfg.KVStore
+	rc.HeartbeatTimeout = cfg.HeartbeatTimeout
+	rc.ReplicationFactor = 1
+
+	return rc
+}

pkg/distributor/distributor_test.go

@@ -1789,7 +1789,7 @@ func prepare(t *testing.T, cfg prepConfig) ([]*Distributor, []mockIngester, *rin
 	// updates to the expected size
 	if distributors[0].distributorsRing != nil {
 		test.Poll(t, time.Second, cfg.numDistributors, func() interface{} {
-			return distributors[0].distributorsRing.HealthyInstancesCount()
+			return distributors[0].distributorsLifeCycler.HealthyInstancesCount()
 		})
 	}
 

pkg/ruler/ruler.go

@@ -426,11 +426,7 @@ func (r *Ruler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 					<p>Ruler running with shards disabled</p>
 				</body>
 			</html>`
-		w.WriteHeader(http.StatusOK)
-		_, err := w.Write([]byte(unshardedPage))
-		if err != nil {
-			level.Error(r.logger).Log("msg", "unable to serve status page", "err", err)
-		}
+		util.WriteHTMLResponse(w, unshardedPage)
 	}
 }
 

pkg/util/http.go

@@ -85,6 +85,14 @@ func WriteTextResponse(w http.ResponseWriter, message string) {
 	_, _ = w.Write([]byte(message))
 }
 
+// Sends message as text/html response with 200 status code.
+func WriteHTMLResponse(w http.ResponseWriter, message string) {
+	w.Header().Set("Content-Type", "text/html")
+
+	// Ignore inactionable errors.
+	_, _ = w.Write([]byte(message))
+}
+
 // RenderHTTPResponse either responds with json or a rendered html page using the passed in template
 // by checking the Accepts header
 func RenderHTTPResponse(w http.ResponseWriter, v interface{}, t *template.Template, r *http.Request) {