From 29c2b1958e38072aa62f523782a1262701eb79e2 Mon Sep 17 00:00:00 2001 From: Jason Wilder Date: Mon, 8 May 2017 14:52:28 -0600 Subject: [PATCH] Fix deletes triggering unnecessary compactions Tombstone files would be written to all TSM files even if the deleted keys or timerange did not exist in the TSM file. This had the side effect of causing shards to get recompacted back to the same state. If any shards or large numbers of TSM files existed, disk usage and CPU utilization would spike causing issues. This prevents tombstones being written for TSM files that could not possiby contain the series keys being deleted or if the delted time range is outside the range of the file. --- CHANGELOG.md | 1 + tsdb/engine/tsm1/reader.go | 34 ++++++++- tsdb/engine/tsm1/reader_test.go | 129 ++++++++++++++++++++++++++++++++ 3 files changed, 163 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c12d9cd493a..0e9dade1268 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -59,6 +59,7 @@ The admin UI is removed and unusable in this release. The `[admin]` configuratio - [#7425](https://github.com/influxdata/influxdb/issues/7425): Fix compaction aborted log messages - [#8123](https://github.com/influxdata/influxdb/issues/8123): TSM compaction does not remove .tmp on error - [#8343](https://github.com/influxdata/influxdb/issues/8343): Set the CSV output to an empty string for null values. +- [#8368](https://github.com/influxdata/influxdb/issues/8368): Compaction exhausting disk resources in InfluxDB ## v1.2.3 [unreleased] diff --git a/tsdb/engine/tsm1/reader.go b/tsdb/engine/tsm1/reader.go index 7e5abf3b992..59b41511c34 100644 --- a/tsdb/engine/tsm1/reader.go +++ b/tsdb/engine/tsm1/reader.go @@ -74,6 +74,12 @@ type TSMIndex interface { // KeyCount returns the count of unique keys in the index. KeyCount() int + // OverlapsTimeRange returns true if the time range of the file intersect min and max. + OverlapsTimeRange(min, max int64) bool + + // OverlapsKeyRange returns true if the min and max keys of the file overlap the arguments min and max. + OverlapsKeyRange(min, max string) bool + // Size returns the size of the current index in bytes. Size() uint32 @@ -403,8 +409,24 @@ func (t *TSMReader) ContainsValue(key string, ts int64) bool { return t.index.ContainsValue(key, ts) } -// DeleteRange removes the given points for keys between minTime and maxTime. +// DeleteRange removes the given points for keys between minTime and maxTime. The series +// keys passed in must be sorted. func (t *TSMReader) DeleteRange(keys []string, minTime, maxTime int64) error { + if len(keys) == 0 { + return nil + } + + // If the keys can't exist in this TSM file, skip it. + minKey, maxKey := keys[0], keys[len(keys)-1] + if !t.index.OverlapsKeyRange(minKey, maxKey) { + return nil + } + + // If the timerange can't exist in this TSM file, skip it. + if !t.index.OverlapsTimeRange(minTime, maxTime) { + return nil + } + if err := t.tombstoner.AddRange(keys, minTime, maxTime); err != nil { return err } @@ -866,6 +888,16 @@ func (d *indirectIndex) Type(key string) (byte, error) { return 0, fmt.Errorf("key does not exist: %v", key) } +// OverlapsTimeRange returns true if the time range of the file intersect min and max. +func (d *indirectIndex) OverlapsTimeRange(min, max int64) bool { + return d.minTime <= max && d.maxTime >= min +} + +// OverlapsKeyRange returns true if the min and max keys of the file overlap the arguments min and max. +func (d *indirectIndex) OverlapsKeyRange(min, max string) bool { + return d.minKey <= max && d.maxKey >= max +} + // KeyRange returns the min and max keys in the index. func (d *indirectIndex) KeyRange() (string, string) { return d.minKey, d.maxKey diff --git a/tsdb/engine/tsm1/reader_test.go b/tsdb/engine/tsm1/reader_test.go index 7c2e0a6f81a..6d3b00397ec 100644 --- a/tsdb/engine/tsm1/reader_test.go +++ b/tsdb/engine/tsm1/reader_test.go @@ -404,6 +404,135 @@ func TestTSMReader_MMAP_TombstoneRange(t *testing.T) { } } +func TestTSMReader_MMAP_TombstoneOutsideTimeRange(t *testing.T) { + dir := MustTempDir() + defer os.RemoveAll(dir) + f := MustTempFile(dir) + defer f.Close() + + w, err := tsm1.NewTSMWriter(f) + if err != nil { + t.Fatalf("unexpected error creating writer: %v", err) + } + + expValues := []tsm1.Value{ + tsm1.NewValue(1, 1.0), + tsm1.NewValue(2, 2.0), + tsm1.NewValue(3, 3.0), + } + if err := w.Write("cpu", expValues); err != nil { + t.Fatalf("unexpected error writing: %v", err) + } + + if err := w.WriteIndex(); err != nil { + t.Fatalf("unexpected error writing index: %v", err) + } + + if err := w.Close(); err != nil { + t.Fatalf("unexpected error closing: %v", err) + } + + f, err = os.Open(f.Name()) + if err != nil { + t.Fatalf("unexpected error open file: %v", err) + } + + r, err := tsm1.NewTSMReader(f) + if err != nil { + t.Fatalf("unexpected error created reader: %v", err) + } + + if err := r.DeleteRange([]string{"cpu"}, 0, 0); err != nil { + t.Fatalf("unexpected error deleting: %v", err) + } + defer r.Close() + + if got, exp := r.ContainsValue("cpu", 1), true; got != exp { + t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp) + } + + if got, exp := r.ContainsValue("cpu", 2), true; got != exp { + t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp) + } + + if got, exp := r.ContainsValue("cpu", 3), true; got != exp { + t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp) + } + + if got, exp := r.HasTombstones(), false; got != exp { + t.Fatalf("HasTombstones mismatch: got %v, exp %v", got, exp) + } + + if got, exp := len(r.TombstoneFiles()), 0; got != exp { + t.Fatalf("TombstoneFiles len mismatch: got %v, exp %v", got, exp) + } +} + +func TestTSMReader_MMAP_TombstoneOutsideKeyRange(t *testing.T) { + dir := MustTempDir() + defer os.RemoveAll(dir) + f := MustTempFile(dir) + defer f.Close() + + w, err := tsm1.NewTSMWriter(f) + if err != nil { + t.Fatalf("unexpected error creating writer: %v", err) + } + + expValues := []tsm1.Value{ + tsm1.NewValue(1, 1.0), + tsm1.NewValue(2, 2.0), + tsm1.NewValue(3, 3.0), + } + if err := w.Write("cpu", expValues); err != nil { + t.Fatalf("unexpected error writing: %v", err) + } + + if err := w.WriteIndex(); err != nil { + t.Fatalf("unexpected error writing index: %v", err) + } + + if err := w.Close(); err != nil { + t.Fatalf("unexpected error closing: %v", err) + } + + f, err = os.Open(f.Name()) + if err != nil { + t.Fatalf("unexpected error open file: %v", err) + } + + r, err := tsm1.NewTSMReader(f) + if err != nil { + t.Fatalf("unexpected error created reader: %v", err) + } + + if err := r.DeleteRange([]string{"mem"}, 0, 3); err != nil { + t.Fatalf("unexpected error deleting: %v", err) + } + defer r.Close() + + if got, exp := r.ContainsValue("cpu", 1), true; got != exp { + t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp) + } + + if got, exp := r.ContainsValue("cpu", 2), true; got != exp { + t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp) + } + + if got, exp := r.ContainsValue("cpu", 3), true; got != exp { + t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp) + } + + if got, exp := r.HasTombstones(), false; got != exp { + t.Fatalf("HasTombstones mismatch: got %v, exp %v", got, exp) + } + + if got, exp := len(r.TombstoneFiles()), 0; got != exp { + t.Fatalf("TombstoneFiles len mismatch: got %v, exp %v", got, exp) + + } +} + func TestTSMReader_MMAP_TombstoneFullRange(t *testing.T) { dir := MustTempDir() defer os.RemoveAll(dir)