Separate out userStates options into config struct

Author: juliusv

ingester/ingester.go

@@ -104,21 +104,20 @@ type Config struct {
 	FlushCheckPeriod  time.Duration
 	MaxChunkIdle      time.Duration
 	MaxChunkAge       time.Duration
-	RateUpdatePeriod  time.Duration
 	ConcurrentFlushes int
-	MaxSeriesPerUser  int
 	ChunkEncoding     string
+	UserStatesConfig  UserStatesConfig
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.FlushCheckPeriod, "ingester.flush-period", 1*time.Minute, "Period with which to attempt to flush chunks.")
-	f.DurationVar(&cfg.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
 	f.DurationVar(&cfg.MaxChunkIdle, "ingester.max-chunk-idle", 1*time.Hour, "Maximum chunk idle time before flushing.")
 	f.DurationVar(&cfg.MaxChunkAge, "ingester.max-chunk-age", 12*time.Hour, "Maximum chunk age time before flushing.")
 	f.IntVar(&cfg.ConcurrentFlushes, "ingester.concurrent-flushes", DefaultConcurrentFlush, "Number of concurrent goroutines flushing to dynamodb.")
 	f.StringVar(&cfg.ChunkEncoding, "ingester.chunk-encoding", "1", "Encoding version to use for chunks.")
-	f.IntVar(&cfg.MaxSeriesPerUser, "ingester.max-series-per-user", DefaultMaxSeriesPerUser, "Maximum number of active series per user.")
+	f.DurationVar(&cfg.UserStatesConfig.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
+	f.IntVar(&cfg.UserStatesConfig.MaxSeriesPerUser, "ingester.max-series-per-user", DefaultMaxSeriesPerUser, "Maximum number of active series per user.")
 }
 
 type flushOp struct {
@@ -144,18 +143,18 @@ func New(cfg Config, chunkStore cortex_chunk.Store, ring *ring.Ring) (*Ingester,
 	if cfg.MaxChunkIdle == 0 {
 		cfg.MaxChunkIdle = 1 * time.Hour
 	}
-	if cfg.RateUpdatePeriod == 0 {
-		cfg.RateUpdatePeriod = 15 * time.Second
-	}
 	if cfg.ConcurrentFlushes <= 0 {
 		cfg.ConcurrentFlushes = DefaultConcurrentFlush
 	}
-	if cfg.MaxSeriesPerUser <= 0 {
-		cfg.MaxSeriesPerUser = DefaultMaxSeriesPerUser
-	}
 	if cfg.ChunkEncoding == "" {
 		cfg.ChunkEncoding = "1"
 	}
+	if cfg.UserStatesConfig.MaxSeriesPerUser <= 0 {
+		cfg.UserStatesConfig.MaxSeriesPerUser = DefaultMaxSeriesPerUser
+	}
+	if cfg.UserStatesConfig.RateUpdatePeriod == 0 {
+		cfg.UserStatesConfig.RateUpdatePeriod = 15 * time.Second
+	}
 
 	if err := chunk.DefaultEncoding.Set(cfg.ChunkEncoding); err != nil {
 		return nil, err
@@ -169,7 +168,7 @@ func New(cfg Config, chunkStore cortex_chunk.Store, ring *ring.Ring) (*Ingester,
 
 		startTime: time.Now(),
 
-		userStates:  newUserStates(cfg.RateUpdatePeriod, cfg.MaxSeriesPerUser),
+		userStates:  newUserStates(&cfg.UserStatesConfig),
 		flushQueues: make([]*util.PriorityQueue, cfg.ConcurrentFlushes, cfg.ConcurrentFlushes),
 
 		ingestedSamples: prometheus.NewCounter(prometheus.CounterOpts{
@@ -434,7 +433,7 @@ func (i *Ingester) loop() {
 	}()
 
 	flushTick := time.Tick(i.cfg.FlushCheckPeriod)
-	rateUpdateTick := time.Tick(i.cfg.RateUpdatePeriod)
+	rateUpdateTick := time.Tick(i.cfg.UserStatesConfig.RateUpdatePeriod)
 	for {
 		select {
 		case <-flushTick:

ingester/ingester_test.go

@@ -153,7 +153,9 @@ func TestIngesterSeriesLimitExceeded(t *testing.T) {
 	cfg := Config{
 		FlushCheckPeriod: 99999 * time.Hour,
 		MaxChunkIdle:     99999 * time.Hour,
-		MaxSeriesPerUser: 1,
+		UserStatesConfig: UserStatesConfig{
+			MaxSeriesPerUser: 1,
+		},
 	}
 	store := &testStore{
 		chunks: map[string][]chunk.Chunk{},

ingester/user_state.go

@@ -14,10 +14,9 @@ import (
 )
 
 type userStates struct {
-	mtx              sync.RWMutex
-	states           map[string]*userState
-	rateUpdatePeriod time.Duration
-	maxSeriesPerUser int
+	mtx    sync.RWMutex
+	states map[string]*userState
+	cfg    *UserStatesConfig
 }
 
 type userState struct {
@@ -29,11 +28,16 @@ type userState struct {
 	ingestedSamples *ewmaRate
 }
 
-func newUserStates(rateUpdatePeriod time.Duration, maxSeriesPerUser int) *userStates {
+// UserStatesConfig configures userStates properties.
+type UserStatesConfig struct {
+	RateUpdatePeriod time.Duration
+	MaxSeriesPerUser int
+}
+
+func newUserStates(cfg *UserStatesConfig) *userStates {
 	return &userStates{
-		states:           map[string]*userState{},
-		rateUpdatePeriod: rateUpdatePeriod,
-		maxSeriesPerUser: maxSeriesPerUser,
+		states: map[string]*userState{},
+		cfg:    cfg,
 	}
 }
 
@@ -126,7 +130,7 @@ func (us *userStates) getOrCreateSeries(ctx context.Context, metric model.Metric
 	us.mtx.RLock()
 	state, ok = us.states[userID]
 	if ok {
-		fp, series, err = state.unlockedGet(metric, us.maxSeriesPerUser)
+		fp, series, err = state.unlockedGet(metric, us.cfg.MaxSeriesPerUser)
 		if err != nil {
 			us.mtx.RUnlock()
 			return nil, fp, nil, err
@@ -140,7 +144,7 @@ func (us *userStates) getOrCreateSeries(ctx context.Context, metric model.Metric
 	us.mtx.Lock()
 	defer us.mtx.Unlock()
 	state = us.unlockedGetOrCreate(userID)
-	fp, series, err = state.unlockedGet(metric, us.maxSeriesPerUser)
+	fp, series, err = state.unlockedGet(metric, us.cfg.MaxSeriesPerUser)
 	return state, fp, series, err
 }
 
@@ -152,7 +156,7 @@ func (us *userStates) unlockedGetOrCreate(userID string) *userState {
 			fpToSeries:      newSeriesMap(),
 			fpLocker:        newFingerprintLocker(16),
 			index:           newInvertedIndex(),
-			ingestedSamples: newEWMARate(0.2, us.rateUpdatePeriod),
+			ingestedSamples: newEWMARate(0.2, us.cfg.RateUpdatePeriod),
 		}
 		state.mapper = newFPMapper(state.fpToSeries)
 		us.states[userID] = state