fix(tsm): use configured value instead of default when checking WAL s…

…egment size (#20839)



Co-authored-by: Tristan Su <[email protected]>

CHANGELOG.md

@@ -22,6 +22,7 @@ the endpoint has been removed. Use the `/metrics` endpoint to collect system sta
 1. [20886](https://github.com/influxdata/influxdb/pull/20886): Prevent "do not have an execution context" error when parsing Flux options in tasks.
 1. [20872](https://github.com/influxdata/influxdb/pull/20872): Respect 24 hour clock formats in the UI and allow more choices
 1. [20860](https://github.com/influxdata/influxdb/pull/20860): Remove unauthenticated, unsupported `/debug/vars` HTTP endpoint.
+1. [20839](https://github.com/influxdata/influxdb/pull/20839): Fix TSM WAL segment size check. Thanks @foobar!
 
 ## v2.0.4 [2021-02-08]
 ----------------------

tsdb/engine/tsm1/wal.go

@@ -464,7 +464,7 @@ func (l *WAL) writeToLog(entry WALEntry) (int, error) {
 // rollSegment checks if the current segment is due to roll over to a new segment;
 // and if so, opens a new segment file for future writes.
 func (l *WAL) rollSegment() error {
-	if l.currentSegmentWriter == nil || l.currentSegmentWriter.size > DefaultSegmentSize {
+	if l.currentSegmentWriter == nil || l.currentSegmentWriter.size > l.SegmentSize {
 		if err := l.newSegmentFile(); err != nil {
 			// A drop database or RP call could trigger this error if writes were in-flight
 			// when the drop statement executes.

tsdb/engine/tsm1/wal_test.go

@@ -3,6 +3,7 @@ package tsm1_test
 import (
 	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
 	"reflect"
 	"testing"
@@ -651,6 +652,55 @@ func TestWALSegmentReader_Corrupt(t *testing.T) {
 	}
 }
 
+func TestWALRollSegment(t *testing.T) {
+	dir := MustTempDir()
+	defer os.RemoveAll(dir)
+
+	w := tsm1.NewWAL(dir)
+	if err := w.Open(); err != nil {
+		t.Fatalf("error opening WAL: %v", err)
+	}
+	const segSize = 1024
+	w.SegmentSize = segSize
+
+	values := map[string][]tsm1.Value{
+		"cpu,host=A#!~#value": []tsm1.Value{tsm1.NewValue(1, 1.0)},
+		"cpu,host=B#!~#value": []tsm1.Value{tsm1.NewValue(1, 1.0)},
+		"cpu,host=C#!~#value": []tsm1.Value{tsm1.NewValue(1, 1.0)},
+	}
+	if _, err := w.WriteMulti(values); err != nil {
+		fatal(t, "write points", err)
+	}
+
+	files, err := ioutil.ReadDir(w.Path())
+	if err != nil {
+		fatal(t, "ReadDir", err)
+	}
+	if len(files) != 1 {
+		t.Fatalf("unexpected segments size %d", len(files))
+	}
+
+	encodeSize := files[0].Size()
+
+	for i := 0; i < 100; i++ {
+		if _, err := w.WriteMulti(values); err != nil {
+			fatal(t, "write points", err)
+		}
+	}
+	files, err = ioutil.ReadDir(w.Path())
+	if err != nil {
+		fatal(t, "ReadDir", err)
+	}
+	for _, f := range files {
+		if f.Size() > int64(segSize)+encodeSize {
+			t.Fatalf("unexpected segment size %d", f.Size())
+		}
+	}
+	if err := w.Close(); err != nil {
+		t.Fatalf("error closing wal: %v", err)
+	}
+}
+
 func TestWriteWALSegment_UnmarshalBinary_WriteWALCorrupt(t *testing.T) {
 	p1 := tsm1.NewValue(1, 1.1)
 	p2 := tsm1.NewValue(1, int64(1))