-
Notifications
You must be signed in to change notification settings - Fork 3.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Introduces per stream chunks mutex (#3000)
* marshalable chunks * wal record types custom serialization * proto types for wal checkpoints * byteswith output unaffected by buffer * wal & record pool ifcs * wal record can hold entries from multiple series * entry pool * ingester uses noopWal * removes duplicate argument passing in ingester code. adds ingester config validation & derives chunk encoding. * segment writing * [WIP] wal recovery from segments * replay uses sync.Maps & preserves WAL fingerprints * in memory wal recovery * wal segment recovery * ingester metrics struct * wal replay locks streamsMtx in instances, adds checkpoint codec * ingester metrics * checkpointer * WAL checkpoint writer * checkpointwriter can write multiple checkpoints * reorgs checkpointing * wires up checkpointwriter to wal * ingester SeriesIter impl * wires up ingesterRecoverer to consume checkpoints * generic recovery fn * generic recovery fn * recover from both wal types * cleans up old tmp checkpoints & allows aborting in flight checkpoints * wires up wal checkpointing * more granular wal logging * fixes off by 1 wal truncation & removes double logging * adds userID to wal records correctly * wire chunk encoding tests * more granular wal metrics * checkpoint encoding test * ignores debug bins * segment replay ignores out of orders * fixes bug between WAL reading []byte validity and proto unmarshalling refs * conf validations, removes comments * flush on shutdown config * POST /ingester/shutdown * renames flush on shutdown * wal & checkpoint use same segment size * writes entries to wal regardless of tailers * makes wal checkpoing duration default to 5m * recovery metrics * encodes headchunks separately for wal purposes * merge upstream * linting * adds chunkMtx to ingester streams * all flush locking locks streams -> chunks in order to prevent deadlocks * wal integration tests * addresses pr feedback uses entry pool in stream push/tailer removes unnecessary pool interaction checkpointbytes comment fillchunk helper, record resetting in tests via pool redundant comment defers wg done in recovery s/num/count/ checkpoint wal uses a logger encodeWithTypeHeader now creates its own []byte removes pool from decodeEntries wal stop can error * cleans up merge * lint * inline functions to simplify unlocking
- Loading branch information
Showing
9 changed files
with
298 additions
and
86 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,143 @@ | ||
| package ingester | ||
|
|
||
| import ( | ||
| "context" | ||
| fmt "fmt" | ||
| "io/ioutil" | ||
| "os" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/cortexproject/cortex/pkg/chunk" | ||
| "github.com/cortexproject/cortex/pkg/util/services" | ||
| "github.com/stretchr/testify/require" | ||
| "github.com/weaveworks/common/user" | ||
|
|
||
| "github.com/grafana/loki/pkg/ingester/client" | ||
| "github.com/grafana/loki/pkg/logproto" | ||
| "github.com/grafana/loki/pkg/util/validation" | ||
| ) | ||
|
|
||
| // small util for ensuring data exists as we expect | ||
| func ensureIngesterData(ctx context.Context, t *testing.T, start, end time.Time, i *Ingester) { | ||
| result := mockQuerierServer{ | ||
| ctx: ctx, | ||
| } | ||
| err := i.Query(&logproto.QueryRequest{ | ||
| Selector: `{foo="bar"}`, | ||
| Limit: 100, | ||
| Start: start, | ||
| End: end, | ||
| }, &result) | ||
|
|
||
| ln := int(end.Sub(start) / time.Second) | ||
| require.NoError(t, err) | ||
| require.Len(t, result.resps, 1) | ||
| require.Len(t, result.resps[0].Streams, 2) | ||
| require.Len(t, result.resps[0].Streams[0].Entries, ln) | ||
| require.Len(t, result.resps[0].Streams[1].Entries, ln) | ||
| } | ||
|
|
||
| func TestIngesterWAL(t *testing.T) { | ||
|
|
||
| walDir, err := ioutil.TempDir(os.TempDir(), "loki-wal") | ||
| require.Nil(t, err) | ||
| defer os.RemoveAll(walDir) | ||
|
|
||
| ingesterConfig := defaultIngesterTestConfig(t) | ||
| ingesterConfig.MaxTransferRetries = 0 | ||
| ingesterConfig.WAL = WALConfig{ | ||
| Enabled: true, | ||
| Dir: walDir, | ||
| Recover: true, | ||
| CheckpointDuration: time.Second, | ||
| } | ||
| limits, err := validation.NewOverrides(defaultLimitsTestConfig(), nil) | ||
| require.NoError(t, err) | ||
|
|
||
| newStore := func() *mockStore { | ||
| return &mockStore{ | ||
| chunks: map[string][]chunk.Chunk{}, | ||
| } | ||
| } | ||
|
|
||
| i, err := New(ingesterConfig, client.Config{}, newStore(), limits, nil) | ||
| require.NoError(t, err) | ||
| require.Nil(t, services.StartAndAwaitRunning(context.Background(), i)) | ||
| defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck | ||
|
|
||
| req := logproto.PushRequest{ | ||
| Streams: []logproto.Stream{ | ||
| { | ||
| Labels: `{foo="bar",bar="baz1"}`, | ||
| }, | ||
| { | ||
| Labels: `{foo="bar",bar="baz2"}`, | ||
| }, | ||
| }, | ||
| } | ||
|
|
||
| start := time.Now() | ||
| steps := 10 | ||
| end := start.Add(time.Second * time.Duration(steps)) | ||
|
|
||
| for i := 0; i < steps; i++ { | ||
| req.Streams[0].Entries = append(req.Streams[0].Entries, logproto.Entry{ | ||
| Timestamp: start.Add(time.Duration(i) * time.Second), | ||
| Line: fmt.Sprintf("line %d", i), | ||
| }) | ||
| req.Streams[1].Entries = append(req.Streams[1].Entries, logproto.Entry{ | ||
| Timestamp: start.Add(time.Duration(i) * time.Second), | ||
| Line: fmt.Sprintf("line %d", i), | ||
| }) | ||
| } | ||
|
|
||
| ctx := user.InjectOrgID(context.Background(), "test") | ||
| _, err = i.Push(ctx, &req) | ||
| require.NoError(t, err) | ||
|
|
||
| ensureIngesterData(ctx, t, start, end, i) | ||
|
|
||
| require.Nil(t, services.StopAndAwaitTerminated(context.Background(), i)) | ||
|
|
||
| // ensure we haven't checkpointed yet | ||
| expectCheckpoint(t, walDir, false) | ||
|
|
||
| // restart the ingester | ||
| i, err = New(ingesterConfig, client.Config{}, newStore(), limits, nil) | ||
| require.NoError(t, err) | ||
| defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck | ||
| require.Nil(t, services.StartAndAwaitRunning(context.Background(), i)) | ||
|
|
||
| // ensure we've recovered data from wal segments | ||
| ensureIngesterData(ctx, t, start, end, i) | ||
|
|
||
| time.Sleep(ingesterConfig.WAL.CheckpointDuration + time.Second) // give a bit of buffer | ||
| // ensure we have checkpointed now | ||
| expectCheckpoint(t, walDir, true) | ||
|
|
||
| require.Nil(t, services.StopAndAwaitTerminated(context.Background(), i)) | ||
|
|
||
| // restart the ingester | ||
| i, err = New(ingesterConfig, client.Config{}, newStore(), limits, nil) | ||
| require.NoError(t, err) | ||
| defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck | ||
| require.Nil(t, services.StartAndAwaitRunning(context.Background(), i)) | ||
|
|
||
| // ensure we've recovered data from checkpoint+wal segments | ||
| ensureIngesterData(ctx, t, start, end, i) | ||
|
|
||
| } | ||
|
|
||
| func expectCheckpoint(t *testing.T, walDir string, shouldExist bool) { | ||
| fs, err := ioutil.ReadDir(walDir) | ||
| require.Nil(t, err) | ||
| var found bool | ||
| for _, f := range fs { | ||
| if _, err := checkpointIndex(f.Name(), false); err == nil { | ||
| found = true | ||
| } | ||
| } | ||
|
|
||
| require.True(t, found == shouldExist) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.