Add new metrics to aid troubleshooting tombstone convergence. (#4231)

* Add new metrics to aid troubleshooting tombstone convergence.

- `memberlist_client_kv_store_value_tombstones`
    Expose the number of tombstones currently in the value for each key
    in the store. If the tombstones are being propagated as expected,
    then each instance should have the same value for this metric.
- `memberlist_client_kv_store_value_tombstones_removed_total`
    Count the number of tombstones which have been removed because they
    aged out. This can be used to troubleshoot whether tombstones are
    being removed too soon (i.e. before the state is converged).
- `memberlist_client_messages_to_broadcast_dropped_total`
    Count occurrences of messages being quietly dropped when they were
    expecting to have been broadcast. As log messages already existed
    for these events, the logs have had the `key` field added instead
    of exposing per-key metrics, for consistency with other metrics
    of this nature in the package.

Signed-off-by: Steve Simpson <[email protected]>

* Review comments.

Signed-off-by: Steve Simpson <[email protected]>

Author: stevesg

CHANGELOG.md

@@ -33,6 +33,10 @@
 * [ENHANCEMENT] Added zone-awareness support to alertmanager for use when sharding is enabled. When zone-awareness is enabled, alerts will be replicated across availability zones. #4204
 * [ENHANCEMENT] Added `tenant_ids` tag to tracing spans #4147
 * [ENHANCEMENT] Ring, query-frontend: Avoid using automatic private IPs (APIPA) when discovering IP address from the interface during the registration of the instance in the ring, or by query-frontend when used with query-scheduler. APIPA still used as last resort with logging indicating usage. #4032
+* [ENHANCEMENT] Memberlist: introduced new metrics to aid troubleshooting tombstone convergence: #4231
+  * `memberlist_client_kv_store_value_tombstones`
+  * `memberlist_client_kv_store_value_tombstones_removed_total`
+  * `memberlist_client_messages_to_broadcast_dropped_total`
 * [BUGFIX] Purger: fix `Invalid null value in condition for column range` caused by `nil` value in range for WriteBatch query. #4128
 * [BUGFIX] Ingester: fixed infrequent panic caused by a race condition between TSDB mmap-ed head chunks truncation and queries. #4176
 * [BUGFIX] Alertmanager: fix Alertmanager status page if clustering via gossip is disabled or sharding is enabled. #4184

integration/integration_memberlist_single_binary_test.go

@@ -86,15 +86,23 @@ func testSingleBinaryEnv(t *testing.T, tlsEnabled bool) {
 	require.NoError(t, cortex2.WaitSumMetrics(e2e.Equals(3*512), "cortex_ring_tokens_total"))
 	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(3*512), "cortex_ring_tokens_total"))
 
+	// All Cortex servers should initially have no tombstones; nobody has left yet.
+	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(0), "memberlist_client_kv_store_value_tombstones"))
+	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(0), "memberlist_client_kv_store_value_tombstones"))
+	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(0), "memberlist_client_kv_store_value_tombstones"))
+
 	require.NoError(t, s.Stop(cortex1))
 	require.NoError(t, cortex2.WaitSumMetrics(e2e.Equals(2*512), "cortex_ring_tokens_total"))
 	require.NoError(t, cortex2.WaitSumMetrics(e2e.Equals(2), "memberlist_client_cluster_members_count"))
+	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(1), "memberlist_client_kv_store_value_tombstones"))
 	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(2*512), "cortex_ring_tokens_total"))
 	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(2), "memberlist_client_cluster_members_count"))
+	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(1), "memberlist_client_kv_store_value_tombstones"))
 
 	require.NoError(t, s.Stop(cortex2))
 	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(1*512), "cortex_ring_tokens_total"))
 	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(1), "memberlist_client_cluster_members_count"))
+	require.NoError(t, cortex3.WaitSumMetrics(e2e.Equals(2), "memberlist_client_kv_store_value_tombstones"))
 
 	require.NoError(t, s.Stop(cortex3))
 }

pkg/ring/kv/memberlist/memberlist_client.go

@@ -252,13 +252,16 @@ type KV struct {
 	totalSizeOfPushes                   prometheus.Counter
 	numberOfBroadcastMessagesInQueue    prometheus.GaugeFunc
 	totalSizeOfBroadcastMessagesInQueue prometheus.Gauge
+	numberOfBroadcastMessagesDropped    prometheus.Counter
 	casAttempts                         prometheus.Counter
 	casFailures                         prometheus.Counter
 	casSuccesses                        prometheus.Counter
 	watchPrefixDroppedNotifications     *prometheus.CounterVec
 
-	storeValuesDesc *prometheus.Desc
-	storeSizesDesc  *prometheus.Desc
+	storeValuesDesc        *prometheus.Desc
+	storeSizesDesc         *prometheus.Desc
+	storeTombstones        *prometheus.GaugeVec
+	storeRemovedTombstones *prometheus.CounterVec
 
 	memberlistMembersCount prometheus.GaugeFunc
 	memberlistHealthScore  prometheus.GaugeFunc
@@ -625,7 +628,7 @@ func (m *KV) get(key string, codec codec.Codec) (out interface{}, version uint,
 		if mr, ok := out.(Mergeable); ok {
 			// remove ALL tombstones before returning to client.
 			// No need for clients to see them.
-			mr.RemoveTombstones(time.Time{})
+			_, _ = mr.RemoveTombstones(time.Time{})
 		}
 	}
 
@@ -883,14 +886,16 @@ func (m *KV) trySingleCas(key string, codec codec.Codec, f func(in interface{})
 func (m *KV) broadcastNewValue(key string, change Mergeable, version uint, codec codec.Codec) {
 	data, err := codec.Encode(change)
 	if err != nil {
-		level.Error(m.logger).Log("msg", "failed to encode change", "err", err)
+		level.Error(m.logger).Log("msg", "failed to encode change", "key", key, "version", version, "err", err)
+		m.numberOfBroadcastMessagesDropped.Inc()
 		return
 	}
 
 	kvPair := KeyValuePair{Key: key, Value: data, Codec: codec.CodecID()}
 	pairData, err := kvPair.Marshal()
 	if err != nil {
-		level.Error(m.logger).Log("msg", "failed to serialize KV pair", "err", err)
+		level.Error(m.logger).Log("msg", "failed to serialize KV pair", "key", key, "version", version, "err", err)
+		m.numberOfBroadcastMessagesDropped.Inc()
 		return
 	}
 
@@ -901,7 +906,8 @@ func (m *KV) broadcastNewValue(key string, change Mergeable, version uint, codec
 		//
 		// Typically messages are smaller (when dealing with couple of updates only), but can get bigger
 		// when broadcasting result of push/pull update.
-		level.Debug(m.logger).Log("msg", "broadcast message too big, not broadcasting", "len", len(pairData))
+		level.Debug(m.logger).Log("msg", "broadcast message too big, not broadcasting", "key", key, "version", version, "len", len(pairData))
+		m.numberOfBroadcastMessagesDropped.Inc()
 		return
 	}
 
@@ -1191,7 +1197,9 @@ func (m *KV) mergeValueForKey(key string, incomingValue Mergeable, casVersion ui
 
 	if m.cfg.LeftIngestersTimeout > 0 {
 		limit := time.Now().Add(-m.cfg.LeftIngestersTimeout)
-		result.RemoveTombstones(limit)
+		total, removed := result.RemoveTombstones(limit)
+		m.storeTombstones.WithLabelValues(key).Set(float64(total))
+		m.storeRemovedTombstones.WithLabelValues(key).Add(float64(removed))
 	}
 
 	encoded, err := codec.Encode(result)

pkg/ring/kv/memberlist/memberlist_client_test.go

@@ -83,8 +83,9 @@ func (d *data) MergeContent() []string {
 	return out
 }
 
-func (d *data) RemoveTombstones(limit time.Time) {
+func (d *data) RemoveTombstones(limit time.Time) (_, _ int) {
 	// nothing to do
+	return
 }
 
 func (d *data) getAllTokens() []uint32 {
@@ -870,8 +871,9 @@ func (dc distributedCounter) MergeContent() []string {
 	return out
 }
 
-func (dc distributedCounter) RemoveTombstones(limit time.Time) {
+func (dc distributedCounter) RemoveTombstones(limit time.Time) (_, _ int) {
 	// nothing to do
+	return
 }
 
 type distributedCounterCodec struct{}

pkg/ring/kv/memberlist/mergeable.go

@@ -38,5 +38,6 @@ type Mergeable interface {
 	// Remove tombstones older than given limit from this mergeable.
 	// If limit is zero time, remove all tombstones. Memberlist client calls this method with zero limit each
 	// time when client is accessing value from the store. It can be used to hide tombstones from the clients.
-	RemoveTombstones(limit time.Time)
+	// Returns the total number of tombstones present and the number of removed tombstones by this invocation.
+	RemoveTombstones(limit time.Time) (total, removed int)
 }

pkg/ring/kv/memberlist/metrics.go

@@ -84,6 +84,13 @@ func (m *KV) createAndRegisterMetrics() {
 		Help:      "Total size of messages waiting in the broadcast queue",
 	})
 
+	m.numberOfBroadcastMessagesDropped = prometheus.NewCounter(prometheus.CounterOpts{
+		Namespace: m.cfg.MetricsNamespace,
+		Subsystem: subsystem,
+		Name:      "messages_to_broadcast_dropped_total",
+		Help:      "Number of broadcast messages intended to be sent but were dropped due to encoding errors or for being too big",
+	})
+
 	m.casAttempts = prometheus.NewCounter(prometheus.CounterOpts{
 		Namespace: m.cfg.MetricsNamespace,
 		Subsystem: subsystem,
@@ -115,6 +122,20 @@ func (m *KV) createAndRegisterMetrics() {
 		"Sizes of values in KV Store in bytes",
 		[]string{"key"}, nil)
 
+	m.storeTombstones = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: m.cfg.MetricsNamespace,
+		Subsystem: subsystem,
+		Name:      "kv_store_value_tombstones",
+		Help:      "Number of tombstones currently present in KV store values",
+	}, []string{"key"})
+
+	m.storeRemovedTombstones = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: m.cfg.MetricsNamespace,
+		Subsystem: subsystem,
+		Name:      "kv_store_value_tombstones_removed_total",
+		Help:      "Total number of tombstones which have been removed from KV store values",
+	}, []string{"key"})
+
 	m.memberlistMembersCount = prometheus.NewGaugeFunc(prometheus.GaugeOpts{
 		Namespace: m.cfg.MetricsNamespace,
 		Subsystem: subsystem,
@@ -162,10 +183,13 @@ func (m *KV) createAndRegisterMetrics() {
 		m.totalSizeOfPushes,
 		m.totalSizeOfPulls,
 		m.totalSizeOfBroadcastMessagesInQueue,
+		m.numberOfBroadcastMessagesDropped,
 		m.casAttempts,
 		m.casFailures,
 		m.casSuccesses,
 		m.watchPrefixDroppedNotifications,
+		m.storeTombstones,
+		m.storeRemovedTombstones,
 		m.memberlistMembersCount,
 		m.memberlistHealthScore,
 	}

pkg/ring/model.go

@@ -374,15 +374,19 @@ func resolveConflicts(normalizedIngesters map[string]InstanceDesc) {
 }
 
 // RemoveTombstones removes LEFT ingesters older than given time limit. If time limit is zero, remove all LEFT ingesters.
-func (d *Desc) RemoveTombstones(limit time.Time) {
-	removed := 0
+func (d *Desc) RemoveTombstones(limit time.Time) (total, removed int) {
 	for n, ing := range d.Ingesters {
-		if ing.State == LEFT && (limit.IsZero() || time.Unix(ing.Timestamp, 0).Before(limit)) {
-			// remove it
-			delete(d.Ingesters, n)
-			removed++
+		if ing.State == LEFT {
+			if limit.IsZero() || time.Unix(ing.Timestamp, 0).Before(limit) {
+				// remove it
+				delete(d.Ingesters, n)
+				removed++
+			} else {
+				total++
+			}
 		}
 	}
+	return
 }
 
 func (d *Desc) getTokensInfo() map[uint32]instanceInfo {