Switch ingester limiter to use ingesterRing rather than lifecycler

pkg/ingester/ingester.go

@@ -408,9 +408,10 @@ func New(cfg Config, limits *validation.Overrides, ingestersRing ring.ReadRing,
 	// Init the limter and instantiate the user states which depend on it
 	i.limiter = NewLimiter(
 		limits,
-		i.lifecycler,
+		ingestersRing,
 		cfg.IngesterRing.ReplicationFactor,
 		cfg.IngesterRing.ZoneAwarenessEnabled,
+		cfg.IngesterRing.InstanceZone,
 	)
 
 	if cfg.ReadPathCPUUtilizationLimit > 0 || cfg.ReadPathMemoryUtilizationLimit > 0 {

pkg/ingester/ingester_ingest_storage_test.go

@@ -408,7 +408,15 @@ func createTestIngesterWithIngestStorage(t testing.TB, ingesterCfg *Config, over
 	// Disable TSDB head compaction jitter to have predictable tests.
 	ingesterCfg.BlocksStorageConfig.TSDB.HeadCompactionIntervalJitterEnabled = false
 
-	ingester, err := New(*ingesterCfg, overrides, nil, nil, reg, util_test.NewTestingLogger(t))
+	// Start the ingester ring
+	rng, err := ring.New(ingesterCfg.IngesterRing.ToRingConfig(), "ingester", IngesterRingKey, log.NewNopLogger(), nil)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), rng))
+	t.Cleanup(func() {
+		require.NoError(t, services.StopAndAwaitTerminated(context.Background(), rng))
+	})
+
+	ingester, err := New(*ingesterCfg, overrides, rng, nil, reg, util_test.NewTestingLogger(t))
 	require.NoError(t, err)
 
 	return ingester, kafkaCluster

pkg/ingester/ingester_test.go

@@ -4596,7 +4596,15 @@ func prepareIngesterWithBlockStorageAndOverrides(t testing.TB, ingesterCfg Confi
 	// Disable TSDB head compaction jitter to have predictable tests.
 	ingesterCfg.BlocksStorageConfig.TSDB.HeadCompactionIntervalJitterEnabled = false
 
-	ingester, err := New(ingesterCfg, overrides, nil, nil, registerer, noDebugNoopLogger{})
+	// Start the ingester ring
+	rng, err := ring.New(ingesterCfg.IngesterRing.ToRingConfig(), "ingester", IngesterRingKey, log.NewNopLogger(), nil)
+	require.NoError(t, err)
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), rng))
+	t.Cleanup(func() {
+		require.NoError(t, services.StopAndAwaitTerminated(context.Background(), rng))
+	})
+
+	ingester, err := New(ingesterCfg, overrides, rng, nil, registerer, noDebugNoopLogger{})
 	if err != nil {
 		return nil, err
 	}

pkg/ingester/limiter.go

@@ -17,7 +17,7 @@ import (
 // to count members
 type RingCount interface {
 	InstancesCount() int
-	InstancesInZoneCount() int
+	InstancesInZoneCount(zone string) int
 	ZonesCount() int
 }
 
@@ -28,6 +28,7 @@ type Limiter struct {
 	ring                 RingCount
 	replicationFactor    int
 	zoneAwarenessEnabled bool
+	ingesterZone         string
 }
 
 // NewLimiter makes a new in-memory series limiter
@@ -36,12 +37,14 @@ func NewLimiter(
 	ring RingCount,
 	replicationFactor int,
 	zoneAwarenessEnabled bool,
+	ingesterZone string,
 ) *Limiter {
 	return &Limiter{
 		limits:               limits,
 		ring:                 ring,
 		replicationFactor:    replicationFactor,
 		zoneAwarenessEnabled: zoneAwarenessEnabled,
+		ingesterZone:         ingesterZone,
 	}
 }
 
@@ -112,7 +115,7 @@ func (l *Limiter) convertGlobalToLocalLimit(userShardSize int, globalLimit int)
 	if zonesCount > 1 {
 		// In this case zone-aware replication is enabled, and ingestersInZoneCount is initially set to
 		// the total number of ingesters in the corresponding zone
-		ingestersInZoneCount = l.ring.InstancesInZoneCount()
+		ingestersInZoneCount = l.ring.InstancesInZoneCount(l.ingesterZone)
 	} else {
 		// In this case zone-aware replication is disabled, and ingestersInZoneCount is initially set to
 		// the total number of ingesters

pkg/ingester/limiter_test.go

@@ -10,7 +10,6 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	"github.com/grafana/mimir/pkg/util/validation"
@@ -287,10 +286,7 @@ func runLimiterMaxFunctionTest(
 
 		t.Run(testName, func(t *testing.T) {
 			// Mock the ring
-			ring := &ringCountMock{}
-			ring.On("InstancesCount").Return(testData.ringIngesterCount)
-			ring.On("InstancesInZoneCount").Return(testData.ingestersInZoneCount)
-			ring.On("ZonesCount").Return(testData.ringZonesCount)
+			ring := &ringCountMock{instancesCount: testData.ringIngesterCount, zonesCount: testData.ringZonesCount, instancesInZoneCount: testData.ingestersInZoneCount}
 
 			// Mock limits
 			limits := validation.Limits{IngestionTenantShardSize: testData.shardSize}
@@ -299,7 +295,7 @@ func runLimiterMaxFunctionTest(
 			overrides, err := validation.NewOverrides(limits, nil)
 			require.NoError(t, err)
 
-			limiter := NewLimiter(overrides, ring, testData.ringReplicationFactor, testData.ringZoneAwarenessEnabled)
+			limiter := NewLimiter(overrides, ring, testData.ringReplicationFactor, testData.ringZoneAwarenessEnabled, "zone")
 			actual := runMaxFn(limiter)
 			assert.Equal(t, testData.expectedValue, actual)
 		})
@@ -342,17 +338,15 @@ func TestLimiter_AssertMaxSeriesPerMetric(t *testing.T) {
 
 		t.Run(testName, func(t *testing.T) {
 			// Mock the ring
-			ring := &ringCountMock{}
-			ring.On("InstancesCount").Return(testData.ringIngesterCount)
-			ring.On("ZonesCount").Return(1)
+			ring := &ringCountMock{instancesCount: testData.ringIngesterCount, zonesCount: 1}
 
 			// Mock limits
 			limits, err := validation.NewOverrides(validation.Limits{
 				MaxGlobalSeriesPerMetric: testData.maxGlobalSeriesPerMetric,
 			}, nil)
 			require.NoError(t, err)
 
-			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false)
+			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false, "")
 			actual := limiter.IsWithinMaxSeriesPerMetric("test", testData.series)
 
 			assert.Equal(t, testData.expected, actual)
@@ -395,17 +389,15 @@ func TestLimiter_AssertMaxMetadataPerMetric(t *testing.T) {
 
 		t.Run(testName, func(t *testing.T) {
 			// Mock the ring
-			ring := &ringCountMock{}
-			ring.On("InstancesCount").Return(testData.ringIngesterCount)
-			ring.On("ZonesCount").Return(1)
+			ring := &ringCountMock{instancesCount: testData.ringIngesterCount, zonesCount: 1}
 
 			// Mock limits
 			limits, err := validation.NewOverrides(validation.Limits{
 				MaxGlobalMetadataPerMetric: testData.maxGlobalMetadataPerMetric,
 			}, nil)
 			require.NoError(t, err)
 
-			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false)
+			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false, "")
 			actual := limiter.IsWithinMaxMetadataPerMetric("test", testData.metadata)
 
 			assert.Equal(t, testData.expected, actual)
@@ -449,17 +441,15 @@ func TestLimiter_AssertMaxSeriesPerUser(t *testing.T) {
 
 		t.Run(testName, func(t *testing.T) {
 			// Mock the ring
-			ring := &ringCountMock{}
-			ring.On("InstancesCount").Return(testData.ringIngesterCount)
-			ring.On("ZonesCount").Return(1)
+			ring := &ringCountMock{instancesCount: testData.ringIngesterCount, zonesCount: 1}
 
 			// Mock limits
 			limits, err := validation.NewOverrides(validation.Limits{
 				MaxGlobalSeriesPerUser: testData.maxGlobalSeriesPerUser,
 			}, nil)
 			require.NoError(t, err)
 
-			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false)
+			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false, "")
 			actual := limiter.IsWithinMaxSeriesPerUser("test", testData.series, limiter.getShardSize("test"))
 
 			assert.Equal(t, testData.expected, actual)
@@ -503,17 +493,15 @@ func TestLimiter_AssertMaxMetricsWithMetadataPerUser(t *testing.T) {
 
 		t.Run(testName, func(t *testing.T) {
 			// Mock the ring
-			ring := &ringCountMock{}
-			ring.On("InstancesCount").Return(testData.ringIngesterCount)
-			ring.On("ZonesCount").Return(1)
+			ring := &ringCountMock{instancesCount: testData.ringIngesterCount, zonesCount: 1}
 
 			// Mock limits
 			limits, err := validation.NewOverrides(validation.Limits{
 				MaxGlobalMetricsWithMetadataPerUser: testData.maxGlobalMetadataPerUser,
 			}, nil)
 			require.NoError(t, err)
 
-			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false)
+			limiter := NewLimiter(limits, ring, testData.ringReplicationFactor, false, "")
 			actual := limiter.IsWithinMaxMetricsWithMetadataPerUser("test", testData.metadata)
 
 			assert.Equal(t, testData.expected, actual)
@@ -522,20 +510,19 @@ func TestLimiter_AssertMaxMetricsWithMetadataPerUser(t *testing.T) {
 }
 
 type ringCountMock struct {
-	mock.Mock
+	instancesCount       int
+	instancesInZoneCount int
+	zonesCount           int
 }
 
 func (m *ringCountMock) InstancesCount() int {
-	args := m.Called()
-	return args.Int(0)
+	return m.instancesCount
 }
 
-func (m *ringCountMock) InstancesInZoneCount() int {
-	args := m.Called()
-	return args.Int(0)
+func (m *ringCountMock) InstancesInZoneCount(_ string) int {
+	return m.instancesInZoneCount
 }
 
 func (m *ringCountMock) ZonesCount() int {
-	args := m.Called()
-	return args.Int(0)
+	return m.zonesCount
 }

pkg/ingester/user_metrics_metadata_test.go

@@ -74,16 +74,14 @@ func TestUserMetricsMetadata(t *testing.T) {
 	for testName, testData := range tests {
 		t.Run(testName, func(t *testing.T) {
 			// Mock the ring
-			ring := &ringCountMock{}
-			ring.On("InstancesCount").Return(1)
-			ring.On("ZonesCount").Return(1)
+			ring := &ringCountMock{instancesCount: 1, zonesCount: 1}
 
 			limits, err := validation.NewOverrides(validation.Limits{
 				MaxGlobalMetricsWithMetadataPerUser: testData.maxMetadataPerUser,
 				MaxGlobalMetadataPerMetric:          testData.maxMetadataPerMetric,
 			}, nil)
 			require.NoError(t, err)
-			limiter := NewLimiter(limits, ring, 1, false)
+			limiter := NewLimiter(limits, ring, 1, false, "")
 
 			metrics := newIngesterMetrics(
 				prometheus.NewPedanticRegistry(),
@@ -132,13 +130,11 @@ func (t noopTestingT) Errorf(string, ...interface{}) {}
 
 func TestUserMetricsMetadataRequest(t *testing.T) {
 	// Mock the ring
-	ring := &ringCountMock{}
-	ring.On("InstancesCount").Return(1)
-	ring.On("ZonesCount").Return(1)
+	ring := &ringCountMock{instancesCount: 1, zonesCount: 1}
 
 	limits, err := validation.NewOverrides(validation.Limits{}, nil)
 	require.NoError(t, err)
-	limiter := NewLimiter(limits, ring, 1, false)
+	limiter := NewLimiter(limits, ring, 1, false, "")
 
 	metrics := newIngesterMetrics(
 		prometheus.NewPedanticRegistry(),

pkg/ingester/user_tsdb_test.go

@@ -217,7 +217,7 @@ func TestGetSeriesAndShardsForSeriesLimit(t *testing.T) {
 
 	db := userTSDB{
 		db:                   tsdbDB,
-		limiter:              NewLimiter(overrides, nil, 3, true),
+		limiter:              NewLimiter(overrides, nil, 3, true, "zone"),
 		ownedSeriesCount:     555,
 		ownedSeriesShardSize: 333,
 	}