Delete cached shuffle-shard subring when user is inactive.

CHANGELOG.md

@@ -67,6 +67,7 @@
 * [ENHANCEMENT] Query-frontend, query-scheduler: cleanup metrics for inactive tenants. #3826
 * [ENHANCEMENT] Distributor: Prevent failed ingestion from affecting rate limiting. #3825
 * [ENHANCEMENT] Blocks storage: added `-blocks-storage.s3.region` support to S3 client configuration. #3811
+* [ENHANCEMENT] Distributor: Remove cached subrings for inactive users when using shuffle sharding. #3849
 * [BUGFIX] Cortex: Fixed issue where fatal errors and various log messages where not logged. #3778
 * [BUGFIX] HA Tracker: don't track as error in the `cortex_kv_request_duration_seconds` metric a CAS operation intentionally aborted. #3745
 * [BUGFIX] Querier / ruler: do not log "error removing stale clients" if the ring is empty. #3761

pkg/distributor/distributor.go

@@ -273,7 +273,7 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 		}, []string{"user"}),
 	}
 	d.replicationFactor.Set(float64(ingestersRing.ReplicationFactor()))
-	d.activeUsers = util.NewActiveUsersCleanupWithDefaultValues(d.cleanupMetricsForUser)
+	d.activeUsers = util.NewActiveUsersCleanupWithDefaultValues(d.cleanupInactiveUser)
 
 	subservices = append(subservices, d.ingesterPool, d.activeUsers)
 	d.subservices, err = services.NewManager(subservices...)
@@ -304,7 +304,9 @@ func (d *Distributor) running(ctx context.Context) error {
 	}
 }
 
-func (d *Distributor) cleanupMetricsForUser(userID string) {
+func (d *Distributor) cleanupInactiveUser(userID string) {
+	d.ingestersRing.CleanupShuffleShardCache(userID)
+
 	d.HATracker.cleanupHATrackerMetricsForUser(userID)
 
 	d.receivedSamples.DeleteLabelValues(userID)

pkg/distributor/distributor_test.go

@@ -323,7 +323,7 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 		cortex_distributor_samples_in_total{user="userA"} 5
 `), metrics...))
 
-	d.cleanupMetricsForUser("userA")
+	d.cleanupInactiveUser("userA")
 
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# HELP cortex_distributor_deduped_samples_total The total number of deduplicated samples.

pkg/ring/ring.go

@@ -77,6 +77,9 @@ type ReadRing interface {
 
 	// HasInstance returns whether the ring contains an instance matching the provided instanceID.
 	HasInstance(instanceID string) bool
+
+	// CleanupShuffleShardCache should delete cached shuffle-shard subrings for given identifier.
+	CleanupShuffleShardCache(identifier string)
 }
 
 var (
@@ -807,6 +810,21 @@ func (r *Ring) setCachedShuffledSubring(identifier string, size int, subring *Ri
 	}
 }
 
+func (r *Ring) CleanupShuffleShardCache(identifier string) {
+	if r.cfg.SubringCacheDisabled {
+		return
+	}
+
+	r.mtx.Lock()
+	defer r.mtx.Unlock()
+
+	for k := range r.shuffledSubringCache {
+		if k.identifier == identifier {
+			delete(r.shuffledSubringCache, k)
+		}
+	}
+}
+
 // Operation describes which instances can be included in the replica set, based on their state.
 //
 // Implemented as bitmap, with upper 16-bits used for encoding extendReplicaSet, and lower 16-bits used for encoding healthy states.

pkg/ring/ring_test.go

@@ -2014,6 +2014,16 @@ func TestShuffleShardWithCaching(t *testing.T) {
 	require.False(t, subring == newSubring)
 	// Zone-aware shuffle-shard gives all zones the same number of instances (at least one).
 	require.Equal(t, zones, newSubring.InstancesCount())
+
+	// Verify that getting the same subring uses cached instance.
+	subring = newSubring
+	newSubring = ring.ShuffleShard("user", 1)
+	require.True(t, subring == newSubring)
+
+	// But after cleanup, it doesn't.
+	ring.CleanupShuffleShardCache("user")
+	newSubring = ring.ShuffleShard("user", 1)
+	require.False(t, subring == newSubring)
 }
 
 // User shuffle shard token.