fix: panic index out of range for invalid series keys

cmd/influx_inspect/dumptsi/dumptsi.go

@@ -244,6 +244,9 @@ func (cmd *Command) printSeries(sfile *tsdb.SeriesFile) error {
 			break
 		}
 		name, tags := tsdb.ParseSeriesKey(sfile.SeriesKey(e.SeriesID))
+		if name == nil {
+			continue
+		}
 
 		if !cmd.matchSeries(name, tags) {
 			continue
@@ -368,6 +371,9 @@ func (cmd *Command) printTagValueSeries(sfile *tsdb.SeriesFile, fs *tsi1.FileSet
 		}
 
 		name, tags := tsdb.ParseSeriesKey(sfile.SeriesKey(e.SeriesID))
+		if name == nil {
+			continue
+		}
 
 		if !cmd.matchSeries(name, tags) {
 			continue

tsdb/index.go

@@ -137,6 +137,9 @@ func (itr *seriesIteratorAdapter) Next() (SeriesElem, error) {
 		}
 
 		name, tags := ParseSeriesKey(key)
+		if name == nil {
+			continue
+		}
 		deleted := itr.sfile.IsDeleted(elem.SeriesID)
 		return &seriesElemAdapter{
 			name:    name,
@@ -381,6 +384,9 @@ func (itr *seriesQueryAdapterIterator) Next() (*query.FloatPoint, error) {
 
 		// Convert to a key.
 		name, tags := ParseSeriesKey(seriesKey)
+		if name == nil {
+			continue
+		}
 		key := string(models.MakeKey(name, tags))
 
 		// Write auxiliary fields.
@@ -867,6 +873,9 @@ func (itr *seriesPointIterator) Next() (*query.FloatPoint, error) {
 		}
 
 		name, tags := ParseSeriesKey(itr.keys[0])
+		if name == nil {
+			continue
+		}
 		itr.keys = itr.keys[1:]
 
 		// TODO(edd): It seems to me like this authorisation check should be
@@ -2340,6 +2349,9 @@ func (itr *measurementSeriesKeyByExprIterator) Next() ([]byte, error) {
 		}
 
 		name, tags := ParseSeriesKey(seriesKey)
+		if name == nil {
+			continue
+		}
 
 		// Check leftover filters. All fields that might be filtered default to zero values
 		if e.Expr != nil {
@@ -2442,6 +2454,11 @@ func (is IndexSet) MeasurementSeriesKeysByExpr(name []byte, expr influxql.Expr)
 		}
 
 		name, tags := ParseSeriesKey(seriesKey)
+		// An invalid series key of 0 length should have been caught by the
+		// above check, but for extra safety we can pass over it here too.
+		if name == nil {
+			continue
+		}
 		keys = append(keys, models.MakeKey(name, tags))
 	}
 
@@ -2918,12 +2935,18 @@ func (is IndexSet) tagValuesByKeyAndExpr(auth query.FineAuthorizer, name []byte,
 
 		if auth != nil {
 			name, tags := ParseSeriesKey(buf)
+			if name == nil {
+				continue
+			}
 			if !auth.AuthorizeSeriesRead(database, name, tags) {
 				continue
 			}
 		}
 
 		_, buf = ReadSeriesKeyLen(buf)
+		if len(buf) == 0 {
+			continue
+		}
 		_, buf = ReadSeriesKeyMeasurement(buf)
 		tagN, buf := ReadSeriesKeyTagN(buf)
 		for i := 0; i < tagN; i++ {

tsdb/index/tsi1/log_file.go

@@ -713,6 +713,9 @@ func (f *LogFile) execSeriesEntry(e *LogEntry) {
 
 	// Read key size.
 	_, remainder := tsdb.ReadSeriesKeyLen(seriesKey)
+	if len(remainder) == 0 {
+		return
+	}
 
 	// Read measurement name.
 	name, remainder := tsdb.ReadSeriesKeyMeasurement(remainder)

tsdb/series_cursor.go

@@ -3,6 +3,7 @@ package tsdb
 import (
 	"bytes"
 	"errors"
+	"fmt"
 	"sort"
 	"sync"
 
@@ -112,6 +113,9 @@ func (cur *seriesCursor) Next() (*SeriesCursorRow, error) {
 		}
 
 		cur.row.Name, cur.row.Tags = ParseSeriesKey(cur.keys[cur.ofs])
+		if cur.row.Name == nil {
+			return nil, fmt.Errorf("series key was not valid: %+v", cur.keys[cur.ofs])
+		}
 		cur.ofs++
 
 		//if itr.opt.Authorizer != nil && !itr.opt.Authorizer.AuthorizeSeriesRead(itr.indexSet.Database(), name, tags) {

tsdb/series_file.go

@@ -414,6 +414,9 @@ func ParseSeriesKeyInto(data []byte, dstTags models.Tags) ([]byte, models.Tags)
 func parseSeriesKey(data []byte, dst models.Tags) ([]byte, models.Tags) {
 	var name []byte
 	_, data = ReadSeriesKeyLen(data)
+	if len(data) == 0 {
+		return nil, nil
+	}
 	name, data = ReadSeriesKeyMeasurement(data)
 	tagN, data := ReadSeriesKeyTagN(data)
 
@@ -445,8 +448,11 @@ func CompareSeriesKeys(a, b []byte) int {
 	}
 
 	// Read total size.
-	_, a = ReadSeriesKeyLen(a)
-	_, b = ReadSeriesKeyLen(b)
+	szA, a := ReadSeriesKeyLen(a)
+	szB, b := ReadSeriesKeyLen(b)
+	if szA == 0 && szB == 0 {
+		return 0
+	}
 
 	// Read names.
 	name0, a := ReadSeriesKeyMeasurement(a)

tsdb/series_file_test.go

@@ -50,6 +50,37 @@ func TestParseSeriesKeyInto(t *testing.T) {
 	}
 }
 
+func TestParseSeriesKey(t *testing.T) {
+	tests := []struct {
+		name        string
+		seriesKey   []byte
+		expectedKey []byte
+	}{
+		{
+			name:        "invalid zero length series key",
+			seriesKey:   tsdb.AppendSeriesKey(nil, []byte{}, nil),
+			expectedKey: nil,
+		},
+		{
+			name:        "valid series key with tags",
+			seriesKey:   tsdb.AppendSeriesKey(nil, []byte("foo"), models.NewTags(map[string]string{"tag1": "foo"})),
+			expectedKey: []byte("foo"),
+		},
+		{
+			name:        "valid series key with no tags",
+			seriesKey:   tsdb.AppendSeriesKey(nil, []byte("foo"), nil),
+			expectedKey: []byte("foo"),
+		},
+	}
+
+	for _, tt := range tests {
+		keyName, _ := tsdb.ParseSeriesKey(tt.seriesKey)
+		if res := bytes.Compare(keyName, tt.expectedKey); res != 0 {
+			t.Fatalf("invalid series key parsed for an %s: got %q, expected %q", tt.name, keyName, tt.expectedKey)
+		}
+	}
+}
+
 // Ensure that broken series files are closed
 func TestSeriesFile_Open_WhenFileCorrupt_ShouldReturnErr(t *testing.T) {
 	f := NewBrokenSeriesFile([]byte{0, 0, 0, 0, 0})

tsdb/series_partition.go

@@ -264,7 +264,13 @@ func (p *SeriesPartition) CreateSeriesListIfNotExists(keys [][]byte, keyPartitio
 		newKeyRanges = append(newKeyRanges, keyRange{id, offset})
 		if tracker.AddedMeasurementSeries != nil {
 			_, remainder := ReadSeriesKeyLen(key)
+			if len(remainder) == 0 {
+				continue
+			}
 			measurement, _ := ReadSeriesKeyMeasurement(remainder)
+			if len(measurement) == 0 {
+				continue
+			}
 			tracker.AddedMeasurementSeries(measurement, 1)
 		}
 	}