Using ingester ID on the distributor metrics

Signed-off-by: alanprot <[email protected]>

Author: alanprot

pkg/distributor/distributor.go

@@ -734,10 +734,12 @@ func (d *Distributor) cleanStaleIngesterMetrics() {
 		return
 	}
 
-	ipsMap := map[string]struct{}{}
+	idsMap := map[string]struct{}{}
 
 	for _, ing := range append(healthy, unhealthy...) {
-		ipsMap[ing.Addr] = struct{}{}
+		if id, err := d.ingestersRing.GetInstanceIdByAddr(ing.Addr); err == nil {
+			idsMap[id] = struct{}{}
+		}
 	}
 
 	ingesterMetrics := []*prometheus.CounterVec{d.ingesterAppends, d.ingesterAppendFailures, d.ingesterQueries, d.ingesterQueryFailures}
@@ -751,7 +753,7 @@ func (d *Distributor) cleanStaleIngesterMetrics() {
 		}
 
 		for _, lbls := range metrics {
-			if _, ok := ipsMap[lbls.Get("ingester")]; !ok {
+			if _, ok := idsMap[lbls.Get("ingester")]; !ok {
 				err := util.DeleteMatchingLabels(m, map[string]string{"ingester": lbls.Get("ingester")})
 				if err != nil {
 					level.Warn(d.log).Log("msg", "error cleaning metrics: DeleteMatchingLabels", "err", err)
@@ -956,6 +958,12 @@ func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, time
 	if err != nil {
 		return err
 	}
+
+	id, err := d.ingestersRing.GetInstanceIdByAddr(ingester.Addr)
+	if err != nil {
+		level.Warn(d.log).Log("msg", "instance not found in the ring", "addr", ingester.Addr, "err", err)
+	}
+
 	c := h.(ingester_client.HealthAndIngesterClient)
 
 	req := cortexpb.PreallocWriteRequestFromPool()
@@ -972,15 +980,15 @@ func (d *Distributor) send(ctx context.Context, ingester ring.InstanceDesc, time
 	}
 
 	if len(metadata) > 0 {
-		d.ingesterAppends.WithLabelValues(ingester.Addr, typeMetadata).Inc()
+		d.ingesterAppends.WithLabelValues(id, typeMetadata).Inc()
 		if err != nil {
-			d.ingesterAppendFailures.WithLabelValues(ingester.Addr, typeMetadata, getErrorStatus(err)).Inc()
+			d.ingesterAppendFailures.WithLabelValues(id, typeMetadata, getErrorStatus(err)).Inc()
 		}
 	}
 	if len(timeseries) > 0 {
-		d.ingesterAppends.WithLabelValues(ingester.Addr, typeSamples).Inc()
+		d.ingesterAppends.WithLabelValues(id, typeSamples).Inc()
 		if err != nil {
-			d.ingesterAppendFailures.WithLabelValues(ingester.Addr, typeSamples, getErrorStatus(err)).Inc()
+			d.ingesterAppendFailures.WithLabelValues(id, typeSamples, getErrorStatus(err)).Inc()
 		}
 	}
 

pkg/distributor/distributor_test.go

@@ -232,12 +232,12 @@ func TestDistributor_Push(t *testing.T) {
 			expectedMetrics: `
 				# HELP cortex_distributor_ingester_append_failures_total The total number of failed batch appends sent to ingesters.
 				# TYPE cortex_distributor_ingester_append_failures_total counter
-				cortex_distributor_ingester_append_failures_total{ingester="2",status="5xx",type="samples"} 1
+				cortex_distributor_ingester_append_failures_total{ingester="ingester-2",status="5xx",type="samples"} 1
 				# HELP cortex_distributor_ingester_appends_total The total number of batch appends sent to ingesters.
 				# TYPE cortex_distributor_ingester_appends_total counter
-				cortex_distributor_ingester_appends_total{ingester="0",type="samples"} 1
-				cortex_distributor_ingester_appends_total{ingester="1",type="samples"} 1
-				cortex_distributor_ingester_appends_total{ingester="2",type="samples"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-0",type="samples"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-1",type="samples"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-2",type="samples"} 1
 			`,
 		},
 		"A push to ingesters should report the correct metrics with no samples": {
@@ -251,12 +251,12 @@ func TestDistributor_Push(t *testing.T) {
 			expectedMetrics: `
 				# HELP cortex_distributor_ingester_append_failures_total The total number of failed batch appends sent to ingesters.
 				# TYPE cortex_distributor_ingester_append_failures_total counter
-				cortex_distributor_ingester_append_failures_total{ingester="2",status="5xx",type="metadata"} 1
+				cortex_distributor_ingester_append_failures_total{ingester="ingester-2",status="5xx",type="metadata"} 1
 				# HELP cortex_distributor_ingester_appends_total The total number of batch appends sent to ingesters.
 				# TYPE cortex_distributor_ingester_appends_total counter
-				cortex_distributor_ingester_appends_total{ingester="0",type="metadata"} 1
-				cortex_distributor_ingester_appends_total{ingester="1",type="metadata"} 1
-				cortex_distributor_ingester_appends_total{ingester="2",type="metadata"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-0",type="metadata"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-1",type="metadata"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-2",type="metadata"} 1
 			`,
 		},
 		"A push to overloaded ingesters should report the correct metrics": {
@@ -268,14 +268,14 @@ func TestDistributor_Push(t *testing.T) {
 			expectedResponse: emptyResponse,
 			ingesterError:    httpgrpc.Errorf(http.StatusTooManyRequests, "Fail"),
 			expectedMetrics: `
-				# HELP cortex_distributor_ingester_append_failures_total The total number of failed batch appends sent to ingesters.
-				# TYPE cortex_distributor_ingester_append_failures_total counter
-				cortex_distributor_ingester_append_failures_total{ingester="2",status="4xx",type="metadata"} 1
 				# HELP cortex_distributor_ingester_appends_total The total number of batch appends sent to ingesters.
 				# TYPE cortex_distributor_ingester_appends_total counter
-				cortex_distributor_ingester_appends_total{ingester="0",type="metadata"} 1
-				cortex_distributor_ingester_appends_total{ingester="1",type="metadata"} 1
-				cortex_distributor_ingester_appends_total{ingester="2",type="metadata"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-0",type="metadata"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-1",type="metadata"} 1
+				cortex_distributor_ingester_appends_total{ingester="ingester-2",type="metadata"} 1
+				# HELP cortex_distributor_ingester_append_failures_total The total number of failed batch appends sent to ingesters.
+				# TYPE cortex_distributor_ingester_append_failures_total counter
+				cortex_distributor_ingester_append_failures_total{ingester="ingester-2",status="4xx",type="metadata"} 1
 			`,
 		},
 		"A push to 3 happy ingesters should succeed, histograms": {
@@ -436,14 +436,16 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 	d.latestSeenSampleTimestampPerUser.WithLabelValues("userA").Set(1111)
 
 	h, _, _ := r.GetAllInstanceDescs(ring.WriteNoExtend)
-	d.ingesterAppends.WithLabelValues(h[0].Addr, typeMetadata).Inc()
-	d.ingesterAppendFailures.WithLabelValues(h[0].Addr, typeMetadata, "2xx").Inc()
-	d.ingesterAppends.WithLabelValues(h[1].Addr, typeMetadata).Inc()
-	d.ingesterAppendFailures.WithLabelValues(h[1].Addr, typeMetadata, "2xx").Inc()
-	d.ingesterQueries.WithLabelValues(h[0].Addr).Inc()
-	d.ingesterQueries.WithLabelValues(h[1].Addr).Inc()
-	d.ingesterQueryFailures.WithLabelValues(h[0].Addr).Inc()
-	d.ingesterQueryFailures.WithLabelValues(h[1].Addr).Inc()
+	ingId0, _ := r.GetInstanceIdByAddr(h[0].Addr)
+	ingId1, _ := r.GetInstanceIdByAddr(h[1].Addr)
+	d.ingesterAppends.WithLabelValues(ingId0, typeMetadata).Inc()
+	d.ingesterAppendFailures.WithLabelValues(ingId0, typeMetadata, "2xx").Inc()
+	d.ingesterAppends.WithLabelValues(ingId1, typeMetadata).Inc()
+	d.ingesterAppendFailures.WithLabelValues(ingId1, typeMetadata, "2xx").Inc()
+	d.ingesterQueries.WithLabelValues(ingId0).Inc()
+	d.ingesterQueries.WithLabelValues(ingId1).Inc()
+	d.ingesterQueryFailures.WithLabelValues(ingId0).Inc()
+	d.ingesterQueryFailures.WithLabelValues(ingId1).Inc()
 
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# HELP cortex_distributor_deduped_samples_total The total number of deduplicated samples.
@@ -489,27 +491,27 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 		
 		# HELP cortex_distributor_ingester_append_failures_total The total number of failed batch appends sent to ingesters.
 		# TYPE cortex_distributor_ingester_append_failures_total counter
-		cortex_distributor_ingester_append_failures_total{ingester="0",status="2xx",type="metadata"} 1
-		cortex_distributor_ingester_append_failures_total{ingester="1",status="2xx",type="metadata"} 1
+		cortex_distributor_ingester_append_failures_total{ingester="ingester-0",status="2xx",type="metadata"} 1
+		cortex_distributor_ingester_append_failures_total{ingester="ingester-1",status="2xx",type="metadata"} 1
 		# HELP cortex_distributor_ingester_appends_total The total number of batch appends sent to ingesters.
 		# TYPE cortex_distributor_ingester_appends_total counter
-		cortex_distributor_ingester_appends_total{ingester="0",type="metadata"} 1
-		cortex_distributor_ingester_appends_total{ingester="1",type="metadata"} 1
+		cortex_distributor_ingester_appends_total{ingester="ingester-0",type="metadata"} 1
+		cortex_distributor_ingester_appends_total{ingester="ingester-1",type="metadata"} 1
 		# HELP cortex_distributor_ingester_queries_total The total number of queries sent to ingesters.
 		# TYPE cortex_distributor_ingester_queries_total counter
-		cortex_distributor_ingester_queries_total{ingester="0"} 1
-		cortex_distributor_ingester_queries_total{ingester="1"} 1
+		cortex_distributor_ingester_queries_total{ingester="ingester-0"} 1
+		cortex_distributor_ingester_queries_total{ingester="ingester-1"} 1
 		# HELP cortex_distributor_ingester_query_failures_total The total number of failed queries sent to ingesters.
 		# TYPE cortex_distributor_ingester_query_failures_total counter
-		cortex_distributor_ingester_query_failures_total{ingester="0"} 1
-		cortex_distributor_ingester_query_failures_total{ingester="1"} 1
+		cortex_distributor_ingester_query_failures_total{ingester="ingester-0"} 1
+		cortex_distributor_ingester_query_failures_total{ingester="ingester-1"} 1
 		`), metrics...))
 
 	d.cleanupInactiveUser("userA")
 
 	err := r.KVClient.CAS(context.Background(), ingester.RingKey, func(in interface{}) (interface{}, bool, error) {
 		r := in.(*ring.Desc)
-		delete(r.Ingesters, "0")
+		delete(r.Ingesters, "ingester-0")
 		return in, true, nil
 	})
 
@@ -556,16 +558,16 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 
 		# HELP cortex_distributor_ingester_append_failures_total The total number of failed batch appends sent to ingesters.
 		# TYPE cortex_distributor_ingester_append_failures_total counter
-		cortex_distributor_ingester_append_failures_total{ingester="1",status="2xx",type="metadata"} 1
+		cortex_distributor_ingester_append_failures_total{ingester="ingester-1",status="2xx",type="metadata"} 1
 		# HELP cortex_distributor_ingester_appends_total The total number of batch appends sent to ingesters.
 		# TYPE cortex_distributor_ingester_appends_total counter
-		cortex_distributor_ingester_appends_total{ingester="1",type="metadata"} 1
+		cortex_distributor_ingester_appends_total{ingester="ingester-1",type="metadata"} 1
 		# HELP cortex_distributor_ingester_queries_total The total number of queries sent to ingesters.
 		# TYPE cortex_distributor_ingester_queries_total counter
-		cortex_distributor_ingester_queries_total{ingester="1"} 1
+		cortex_distributor_ingester_queries_total{ingester="ingester-1"} 1
 		# HELP cortex_distributor_ingester_query_failures_total The total number of failed queries sent to ingesters.
 		# TYPE cortex_distributor_ingester_query_failures_total counter
-		cortex_distributor_ingester_query_failures_total{ingester="1"} 1
+		cortex_distributor_ingester_query_failures_total{ingester="ingester-1"} 1
 		`), metrics...))
 }
 
@@ -2717,8 +2719,9 @@ func prepare(tb testing.TB, cfg prepConfig) ([]*Distributor, []*mockIngester, []
 		} else {
 			tokens = []uint32{uint32((math.MaxUint32 / cfg.numIngesters) * i)}
 		}
-		addr := fmt.Sprintf("%d", i)
-		ingesterDescs[addr] = ring.InstanceDesc{
+		ingester := fmt.Sprintf("ingester-%d", i)
+		addr := fmt.Sprintf("ip-ingester-%d", i)
+		ingesterDescs[ingester] = ring.InstanceDesc{
 			Addr:                addr,
 			Zone:                "",
 			State:               ring.ACTIVE,

pkg/distributor/query.go

@@ -2,6 +2,7 @@ package distributor
 
 import (
 	"context"
+	"github.com/go-kit/log/level"
 	"io"
 	"sort"
 	"time"
@@ -166,9 +167,14 @@ func (d *Distributor) queryIngestersExemplars(ctx context.Context, replicationSe
 		}
 
 		resp, err := client.(ingester_client.IngesterClient).QueryExemplars(ctx, req)
-		d.ingesterQueries.WithLabelValues(ing.Addr).Inc()
+		ingesterId, err := d.ingestersRing.GetInstanceIdByAddr(ing.Addr)
 		if err != nil {
-			d.ingesterQueryFailures.WithLabelValues(ing.Addr).Inc()
+			level.Warn(d.log).Log("msg", "instance not found in the ring", "addr", ing.Addr, "err", err)
+		}
+
+		d.ingesterQueries.WithLabelValues(ingesterId).Inc()
+		if err != nil {
+			d.ingesterQueryFailures.WithLabelValues(ingesterId).Inc()
 			return nil, err
 		}
 
@@ -225,11 +231,17 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 		if err != nil {
 			return nil, err
 		}
-		d.ingesterQueries.WithLabelValues(ing.Addr).Inc()
+
+		ingesterId, err := d.ingestersRing.GetInstanceIdByAddr(ing.Addr)
+		if err != nil {
+			level.Warn(d.log).Log("msg", "instance not found in the ring", "addr", ing.Addr, "err", err)
+		}
+
+		d.ingesterQueries.WithLabelValues(ingesterId).Inc()
 
 		stream, err := client.(ingester_client.IngesterClient).QueryStream(ctx, req)
 		if err != nil {
-			d.ingesterQueryFailures.WithLabelValues(ing.Addr).Inc()
+			d.ingesterQueryFailures.WithLabelValues(ingesterId).Inc()
 			return nil, err
 		}
 		defer stream.CloseSend() //nolint:errcheck
@@ -242,7 +254,7 @@ func (d *Distributor) queryIngesterStream(ctx context.Context, replicationSet ri
 			} else if err != nil {
 				// Do not track a failure if the context was canceled.
 				if !grpcutil.IsGRPCContextCanceled(err) {
-					d.ingesterQueryFailures.WithLabelValues(ing.Addr).Inc()
+					d.ingesterQueryFailures.WithLabelValues(ingesterId).Inc()
 				}
 
 				return nil, err

pkg/ring/model.go

@@ -527,6 +527,16 @@ func (d *Desc) getTokensByZone() map[string][]uint32 {
 	return MergeTokensByZone(zones)
 }
 
+// getInstancesByAddr returns instances id by its address
+func (d *Desc) getInstancesByAddr() map[string]string {
+	instancesByAddMap := make(map[string]string, len(d.Ingesters))
+	for id, instance := range d.Ingesters {
+		instancesByAddMap[instance.Addr] = id
+	}
+
+	return instancesByAddMap
+}
+
 type CompareResult int
 
 // CompareResult responses

pkg/ring/ring.go

@@ -80,6 +80,10 @@ type ReadRing interface {
 	// instance does not exist in the ring.
 	GetInstanceState(instanceID string) (InstanceState, error)
 
+	// GetInstanceIdByAddr returns the instance id from its address or an error if the
+	//	// instance does not exist in the ring.
+	GetInstanceIdByAddr(addr string) (string, error)
+
 	// ShuffleShardWithLookback is like ShuffleShard() but the returned subring includes
 	// all instances that have been part of the identifier's shard since "now - lookbackPeriod".
 	ShuffleShardWithLookback(identifier string, size int, lookbackPeriod time.Duration, now time.Time) ReadRing
@@ -186,6 +190,8 @@ type Ring struct {
 	// change it is to create a new one and replace it).
 	ringInstanceByToken map[uint32]instanceInfo
 
+	ringInstanceIdByAddr map[string]string
+
 	// When did a set of instances change the last time (instance changing state or heartbeat is ignored for this timestamp).
 	lastTopologyChange time.Time
 
@@ -338,6 +344,7 @@ func (r *Ring) updateRingState(ringDesc *Desc) {
 	ringTokens := ringDesc.GetTokens()
 	ringTokensByZone := ringDesc.getTokensByZone()
 	ringInstanceByToken := ringDesc.getTokensInfo()
+	ringInstanceByAddr := ringDesc.getInstancesByAddr()
 	ringZones := getZones(ringTokensByZone)
 
 	r.mtx.Lock()
@@ -346,6 +353,7 @@ func (r *Ring) updateRingState(ringDesc *Desc) {
 	r.ringTokens = ringTokens
 	r.ringTokensByZone = ringTokensByZone
 	r.ringInstanceByToken = ringInstanceByToken
+	r.ringInstanceIdByAddr = ringInstanceByAddr
 	r.ringZones = ringZones
 	r.lastTopologyChange = now
 	if r.shuffledSubringCache != nil {
@@ -895,6 +903,15 @@ func (r *Ring) GetInstanceState(instanceID string) (InstanceState, error) {
 	return instance.GetState(), nil
 }
 
+// GetInstanceIdByAddr implements ReadRing.
+func (r *Ring) GetInstanceIdByAddr(addr string) (string, error) {
+	if i, ok := r.ringInstanceIdByAddr[addr]; ok {
+		return i, nil
+	}
+
+	return "notFound", ErrInstanceNotFound
+}
+
 // HasInstance returns whether the ring contains an instance matching the provided instanceID.
 func (r *Ring) HasInstance(instanceID string) bool {
 	r.mtx.RLock()