mirror of https://github.com/grafana/loki
Loki: Implement stream sharding (#6952)
* introduce 'StreamSharder' interface to decouple sharding from managing shards
* make the streamsharder take a whole stream instead of just and ID
* Implement stream sharding usage.
* move streamShard to distributor
* Implement the stream sharder and shard iter (#6941)
* Modify sharding to not duplicate all entries for all shards (#6942)
* Modify sharding to not duplicate all entries for all shards.
* Rename ShardIter and get rid of NextShardId method.
* Get rid of ShardIter and ShardStats.
* Implement a StreamSharder mock and use it in tests.
* Rewrite stubbing and remove flags.
* Add auto-splitting.
* Only shard stream when approppriate.
* Make sharding mode an enum.
* Bench shardStream function.
* Modify shard comparison to be <= 1.
* Add a guard clause for when lowerbound>upperbound.
* Apply suggestions from code review
Avoid using `fmt.Sprintf`.
Co-authored-by: Danny Kopping <danny.kopping@grafana.com>
* Update pkg/distributor/distributor.go
Rename metric name.
Co-authored-by: Danny Kopping <danny.kopping@grafana.com>
* Avoid double-negative by Danny's suggestion.
* Move TODO to its right place.
* Log scenario where upperbound < lowerbound.
* Reuse compiled regex.
* Add docstring explaining why we use max(shards*2, 2).
* make sharder use a RWMutex
* simplify sharding config
* replace doc string
* Revert "Add auto-splitting" for standalone PR.
This reverts commit e3886a6210.
* shard loop refactor
* move shard creation into it's own function
* fix after refactor
* make config more explicit and document it
* reduce allocations
* review feedback
Co-authored-by: Travis Patterson <travis.patterson@grafana.com>
Co-authored-by: Danny Kopping <danny.kopping@grafana.com>
pull/6970/head
Dylan Guedes
3 years ago
committed by
GitHub
parent
15f8f42295
commit
5d8c48bd10
@ -0,0 +1,49 @@ |
||||
package distributor |
||||
|
||||
import ( |
||||
"sync" |
||||
|
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
type streamSharder struct { |
||||
mu sync.RWMutex |
||||
streams map[string]int |
||||
} |
||||
|
||||
func NewStreamSharder() StreamSharder { |
||||
return &streamSharder{ |
||||
streams: make(map[string]int), |
||||
} |
||||
} |
||||
|
||||
func (s *streamSharder) ShardCountFor(stream logproto.Stream) (int, bool) { |
||||
s.mu.RLock() |
||||
defer s.mu.RUnlock() |
||||
|
||||
shards := s.streams[stream.Labels] |
||||
if shards > 0 { |
||||
return shards, true |
||||
} |
||||
|
||||
return 0, false |
||||
} |
||||
|
||||
// IncreaseShardsFor shards the given stream by doubling its number of shards.
|
||||
func (s *streamSharder) IncreaseShardsFor(stream logproto.Stream) { |
||||
s.mu.Lock() |
||||
defer s.mu.Unlock() |
||||
|
||||
shards := s.streams[stream.Labels] |
||||
|
||||
// Since the number of shards of a stream that is being sharded for the first time is 0,
|
||||
// we assign to it shards = max(shards*2, 2) such that its number of shards will be no less than 2.
|
||||
s.streams[stream.Labels] = max(shards*2, 2) |
||||
} |
||||
|
||||
func max(a, b int) int { |
||||
if a > b { |
||||
return a |
||||
} |
||||
return b |
||||
} |
||||
@ -0,0 +1,73 @@ |
||||
package distributor |
||||
|
||||
import ( |
||||
"testing" |
||||
|
||||
"github.com/grafana/loki/pkg/logproto" |
||||
|
||||
"github.com/stretchr/testify/require" |
||||
) |
||||
|
||||
func TestStreamSharder(t *testing.T) { |
||||
stream := logproto.Stream{Entries: make([]logproto.Entry, 11), Labels: "test-stream"} |
||||
stream2 := logproto.Stream{Entries: make([]logproto.Entry, 11), Labels: "test-stream-2"} |
||||
|
||||
t.Run("it returns not ok when a stream should not be sharded", func(t *testing.T) { |
||||
sharder := NewStreamSharder() |
||||
|
||||
shards, ok := sharder.ShardCountFor(stream) |
||||
require.Equal(t, shards, 0) |
||||
require.False(t, ok) |
||||
}) |
||||
|
||||
t.Run("it keeps track of multiple streams", func(t *testing.T) { |
||||
sharder := NewStreamSharder() |
||||
sharder.IncreaseShardsFor(stream) |
||||
sharder.IncreaseShardsFor(stream) |
||||
sharder.IncreaseShardsFor(stream2) |
||||
|
||||
shards, ok := sharder.ShardCountFor(stream) |
||||
require.True(t, ok) |
||||
|
||||
require.Equal(t, 4, shards) |
||||
|
||||
shards, ok = sharder.ShardCountFor(stream2) |
||||
require.True(t, ok) |
||||
|
||||
require.Equal(t, 2, shards) |
||||
}) |
||||
} |
||||
|
||||
type StreamSharderMock struct { |
||||
calls map[string]int |
||||
|
||||
wantShards int |
||||
} |
||||
|
||||
func NewStreamSharderMock(shards int) *StreamSharderMock { |
||||
return &StreamSharderMock{ |
||||
calls: make(map[string]int), |
||||
wantShards: shards, |
||||
} |
||||
} |
||||
|
||||
func (s *StreamSharderMock) IncreaseShardsFor(stream logproto.Stream) { |
||||
s.increaseCallsFor("IncreaseShardsFor") |
||||
} |
||||
|
||||
func (s *StreamSharderMock) ShardCountFor(stream logproto.Stream) (int, bool) { |
||||
s.increaseCallsFor("ShardCountFor") |
||||
if s.wantShards < 0 { |
||||
return 0, false |
||||
} |
||||
return s.wantShards, true |
||||
} |
||||
|
||||
func (s *StreamSharderMock) increaseCallsFor(funcName string) { |
||||
if _, ok := s.calls[funcName]; ok { |
||||
s.calls[funcName]++ |
||||
return |
||||
} |
||||
|
||||
s.calls[funcName] = 1 |
||||
} |
||||
Loading…
Reference in new issue