Tenant/unordered

pkg/ingester/instance.go

@@ -132,7 +132,7 @@ func (i *instance) consumeChunk(ctx context.Context, ls labels.Labels, chunk *lo
 	if !ok {
 
 		sortedLabels := i.index.Add(cortexpb.FromLabelsToLabelAdapters(ls), fp)
-		stream = newStream(i.cfg, i.instanceID, fp, sortedLabels, i.metrics)
+		stream = newStream(i.cfg, i.instanceID, fp, sortedLabels, i.limiter.limits.UnorderedWrites(i.instanceID), i.metrics)
 		i.streamsByFP[fp] = stream
 		i.streams[stream.labelsString] = stream
 		i.streamsCreatedTotal.Inc()
@@ -243,7 +243,7 @@ func (i *instance) getOrCreateStream(pushReqStream logproto.Stream, lock bool, r
 	fp := i.getHashForLabels(labels)
 
 	sortedLabels := i.index.Add(cortexpb.FromLabelsToLabelAdapters(labels), fp)
-	stream = newStream(i.cfg, i.instanceID, fp, sortedLabels, i.metrics)
+	stream = newStream(i.cfg, i.instanceID, fp, sortedLabels, i.limiter.limits.UnorderedWrites(i.instanceID), i.metrics)
 	i.streams[pushReqStream.Labels] = stream
 	i.streamsByFP[fp] = stream
 

pkg/ingester/instance_test.go

@@ -178,7 +178,7 @@ func Test_SeriesQuery(t *testing.T) {
 	for _, testStream := range testStreams {
 		stream, err := instance.getOrCreateStream(testStream, false, recordPool.GetRecord())
 		require.NoError(t, err)
-		chunk := newStream(cfg, "fake", 0, nil, NilMetrics).NewChunk()
+		chunk := newStream(cfg, "fake", 0, nil, true, NilMetrics).NewChunk()
 		for _, entry := range testStream.Entries {
 			err = chunk.Append(&entry)
 			require.NoError(t, err)
@@ -334,7 +334,7 @@ func Benchmark_instance_addNewTailer(b *testing.B) {
 	lbs := makeRandomLabels()
 	b.Run("addTailersToNewStream", func(b *testing.B) {
 		for n := 0; n < b.N; n++ {
-			inst.addTailersToNewStream(newStream(nil, "fake", 0, lbs, NilMetrics))
+			inst.addTailersToNewStream(newStream(nil, "fake", 0, lbs, true, NilMetrics))
 		}
 	})
 }

pkg/ingester/stream.go

@@ -95,6 +95,8 @@ type stream struct {
 	// errors were used to detect this, but this counter has been
 	// introduced to facilitate removing the ordering constraint.
 	entryCt int64
+
+	unorderedWrites bool
 }
 
 type chunkDesc struct {
@@ -111,15 +113,16 @@ type entryWithError struct {
 	e     error
 }
 
-func newStream(cfg *Config, tenant string, fp model.Fingerprint, labels labels.Labels, metrics *ingesterMetrics) *stream {
+func newStream(cfg *Config, tenant string, fp model.Fingerprint, labels labels.Labels, unorderedWrites bool, metrics *ingesterMetrics) *stream {
 	return &stream{
-		cfg:          cfg,
-		fp:           fp,
-		labels:       labels,
-		labelsString: labels.String(),
-		tailers:      map[uint32]*tailer{},
-		metrics:      metrics,
-		tenant:       tenant,
+		cfg:             cfg,
+		fp:              fp,
+		labels:          labels,
+		labelsString:    labels.String(),
+		tailers:         map[uint32]*tailer{},
+		metrics:         metrics,
+		tenant:          tenant,
+		unorderedWrites: unorderedWrites,
 	}
 }
 
@@ -159,7 +162,7 @@ func (s *stream) setChunks(chunks []Chunk) (bytesAdded, entriesAdded int, err er
 
 func (s *stream) NewChunk() *chunkenc.MemChunk {
 	hbType := chunkenc.OrderedHeadBlockFmt
-	if s.cfg.UnorderedWrites {
+	if s.unorderedWrites {
 		hbType = chunkenc.UnorderedHeadBlockFmt
 	}
 	return chunkenc.NewMemChunk(s.cfg.parsedEncoding, hbType, s.cfg.BlockSize, s.cfg.TargetChunkSize)
@@ -225,7 +228,7 @@ func (s *stream) Push(
 		}
 
 		// The validity window for unordered writes is the highest timestamp present minus 1/2 * max-chunk-age.
-		if s.cfg.UnorderedWrites && !s.highestTs.IsZero() && s.highestTs.Add(-s.cfg.MaxChunkAge/2).After(entries[i].Timestamp) {
+		if s.unorderedWrites && !s.highestTs.IsZero() && s.highestTs.Add(-s.cfg.MaxChunkAge/2).After(entries[i].Timestamp) {
 			failedEntriesWithError = append(failedEntriesWithError, entryWithError{&entries[i], chunkenc.ErrOutOfOrder})
 			outOfOrderSamples++
 			outOfOrderBytes += len(entries[i].Line)

pkg/ingester/stream_test.go

@@ -53,6 +53,7 @@ func TestMaxReturnedStreamsErrors(t *testing.T) {
 				labels.Labels{
 					{Name: "foo", Value: "bar"},
 				},
+				true,
 				NilMetrics,
 			)
 
@@ -92,6 +93,7 @@ func TestPushDeduplication(t *testing.T) {
 		labels.Labels{
 			{Name: "foo", Value: "bar"},
 		},
+		true,
 		NilMetrics,
 	)
 
@@ -115,6 +117,7 @@ func TestPushRejectOldCounter(t *testing.T) {
 		labels.Labels{
 			{Name: "foo", Value: "bar"},
 		},
+		true,
 		NilMetrics,
 	)
 
@@ -203,6 +206,7 @@ func TestUnorderedPush(t *testing.T) {
 		labels.Labels{
 			{Name: "foo", Value: "bar"},
 		},
+		true,
 		NilMetrics,
 	)
 
@@ -300,7 +304,7 @@ func Benchmark_PushStream(b *testing.B) {
 		labels.Label{Name: "job", Value: "loki-dev/ingester"},
 		labels.Label{Name: "container", Value: "ingester"},
 	}
-	s := newStream(&Config{}, "fake", model.Fingerprint(0), ls, NilMetrics)
+	s := newStream(&Config{}, "fake", model.Fingerprint(0), ls, true, NilMetrics)
 	t, err := newTailer("foo", `{namespace="loki-dev"}`, &fakeTailServer{})
 	require.NoError(b, err)
 

pkg/validation/limits.go

@@ -42,8 +42,9 @@ type Limits struct {
 	MaxLineSizeTruncate    bool             `yaml:"max_line_size_truncate" json:"max_line_size_truncate"`
 
 	// Ingester enforced limits.
-	MaxLocalStreamsPerUser  int `yaml:"max_streams_per_user" json:"max_streams_per_user"`
-	MaxGlobalStreamsPerUser int `yaml:"max_global_streams_per_user" json:"max_global_streams_per_user"`
+	MaxLocalStreamsPerUser  int  `yaml:"max_streams_per_user" json:"max_streams_per_user"`
+	MaxGlobalStreamsPerUser int  `yaml:"max_global_streams_per_user" json:"max_global_streams_per_user"`
+	UnorderedWrites         bool `yaml:"unordered_writes" json:"unordered_writes"`
 
 	// Querier enforced limits.
 	MaxChunksPerQuery          int            `yaml:"max_chunks_per_query" json:"max_chunks_per_query"`
@@ -105,6 +106,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) {
 
 	f.IntVar(&l.MaxLocalStreamsPerUser, "ingester.max-streams-per-user", 10e3, "Maximum number of active streams per user, per ingester. 0 to disable.")
 	f.IntVar(&l.MaxGlobalStreamsPerUser, "ingester.max-global-streams-per-user", 0, "Maximum number of active streams per user, across the cluster. 0 to disable.")
+	f.BoolVar(&l.UnorderedWrites, "ingester.unordered-writes", false, "(Experimental) Allow out of order writes.")
 
 	f.IntVar(&l.MaxChunksPerQuery, "store.query-chunk-limit", 2e6, "Maximum number of chunks that can be fetched in a single query.")
 
@@ -400,6 +402,10 @@ func (o *Overrides) StreamRetention(userID string) []StreamRetention {
 	return o.getOverridesForUser(userID).StreamRetention
 }
 
+func (o *Overrides) UnorderedWrites(userID string) bool {
+	return o.getOverridesForUser(userID).UnorderedWrites
+}
+
 func (o *Overrides) ForEachTenantLimit(callback ForEachTenantLimitCallback) {
 	o.tenantLimits.ForEachTenantLimit(callback)
 }