Make MinChunkLength settable per-user

pkg/ingester/flush.go

@@ -266,25 +266,27 @@ func (i *Ingester) flushUserSeries(flushQueueIndex int, userID string, fp model.
 		chunks = chunks[:len(chunks)-1]
 	}
 
-	if reason == reasonIdle && series.headChunkClosed && i.cfg.MinChunkLength > 0 {
-		chunkLength := 0
-		for _, c := range chunks {
-			chunkLength += c.C.Len()
-		}
-		if chunkLength < i.cfg.MinChunkLength {
-			userState.removeSeries(fp, series.metric)
-			memoryChunks.Sub(float64(len(chunks)))
-			droppedChunks.Add(float64(len(chunks)))
-			util.Event().Log(
-				"msg", "dropped chunks",
-				"userID", userID,
-				"numChunks", len(chunks),
-				"chunkLength", chunkLength,
-				"fp", fp,
-				"series", series.metric,
-				"queue", flushQueueIndex,
-			)
-			chunks = nil
+	if reason == reasonIdle && series.headChunkClosed {
+		if minChunkLength := i.limits.MinChunkLength(userID); minChunkLength > 0 {
+			chunkLength := 0
+			for _, c := range chunks {
+				chunkLength += c.C.Len()
+			}
+			if chunkLength < minChunkLength {
+				userState.removeSeries(fp, series.metric)
+				memoryChunks.Sub(float64(len(chunks)))
+				droppedChunks.Add(float64(len(chunks)))
+				util.Event().Log(
+					"msg", "dropped chunks",
+					"userID", userID,
+					"numChunks", len(chunks),
+					"chunkLength", chunkLength,
+					"fp", fp,
+					"series", series.metric,
+					"queue", flushQueueIndex,
+				)
+				chunks = nil
+			}
 		}
 	}
 

pkg/ingester/ingester.go

@@ -117,7 +117,6 @@ type Config struct {
 	ChunkAgeJitter    time.Duration
 	ConcurrentFlushes int
 	SpreadFlushes     bool
-	MinChunkLength    int
 
 	RateUpdatePeriod time.Duration
 
@@ -137,7 +136,6 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.MaxChunkAge, "ingester.max-chunk-age", 12*time.Hour, "Maximum chunk age before flushing.")
 	f.DurationVar(&cfg.ChunkAgeJitter, "ingester.chunk-age-jitter", 20*time.Minute, "Range of time to subtract from MaxChunkAge to spread out flushes")
 	f.BoolVar(&cfg.SpreadFlushes, "ingester.spread-flushes", false, "If true, spread series flushes across the whole period of MaxChunkAge")
-	f.IntVar(&cfg.MinChunkLength, "ingester.min-chunk-length", 0, "Minimum number of samples in an idle chunk to flush it to the store. Use with care, if chunks are less than this size they will be discarded.")
 	f.IntVar(&cfg.ConcurrentFlushes, "ingester.concurrent-flushes", 50, "Number of concurrent goroutines flushing to dynamodb.")
 	f.DurationVar(&cfg.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
 }

pkg/util/validation/limits.go

@@ -30,6 +30,7 @@ type Limits struct {
 	MaxSamplesPerQuery int `yaml:"max_samples_per_query"`
 	MaxSeriesPerUser   int `yaml:"max_series_per_user"`
 	MaxSeriesPerMetric int `yaml:"max_series_per_metric"`
+	MinChunkLength     int `yaml:"min_chunk_length"`
 
 	// Querier enforced limits.
 	MaxChunksPerQuery   int           `yaml:"max_chunks_per_query"`
@@ -61,6 +62,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&l.MaxSamplesPerQuery, "ingester.max-samples-per-query", 1000000, "The maximum number of samples that a query can return.")
 	f.IntVar(&l.MaxSeriesPerUser, "ingester.max-series-per-user", 5000000, "Maximum number of active series per user.")
 	f.IntVar(&l.MaxSeriesPerMetric, "ingester.max-series-per-metric", 50000, "Maximum number of active series per metric name.")
+	f.IntVar(&l.MinChunkLength, "ingester.min-chunk-length", 0, "Minimum number of samples in an idle chunk to flush it to the store. Use with care, if chunks are less than this size they will be discarded.")
 
 	f.IntVar(&l.MaxChunksPerQuery, "store.query-chunk-limit", 2e6, "Maximum number of chunks that can be fetched in a single query.")
 	f.DurationVar(&l.MaxQueryLength, "store.max-query-length", 0, "Limit to length of chunk store queries, 0 to disable.")
@@ -229,6 +231,11 @@ func (o *Overrides) CardinalityLimit(userID string) int {
 	return o.overridesManager.GetLimits(userID).(*Limits).CardinalityLimit
 }
 
+// MinChunkLength returns the minimum size of chunk that will be saved by ingesters
+func (o *Overrides) MinChunkLength(userID string) int {
+	return o.overridesManager.GetLimits(userID).(*Limits).MinChunkLength
+}
+
 // Loads overrides and returns the limits as an interface to store them in OverridesManager.
 // We need to implement it here since OverridesManager must store type Limits in an interface but
 // it doesn't know its definition to initialize it.