Merge pull request #6170 from influxdata/jw-tsm-panic

Fix TSM UnmarshalBinary panic

CHANGELOG.md

@@ -31,6 +31,7 @@
 - [#6094](https://github.com/influxdata/influxdb/issues/6094): Ensure CREATE RETENTION POLICY and CREATE CONTINUOUS QUERY are idempotent in the correct way.
 - [#6092](https://github.com/influxdata/influxdb/issues/6092): Upgrading directly from 0.9.6.1 to 0.11.0 fails
 - [#6061](https://github.com/influxdata/influxdb/issues/6061): [0.12 / master] POST to /write does not write points if request has header 'Content-Type: application/x-www-form-urlencoded'
+- [#6121](https://github.com/influxdata/influxdb/issues/6121): Fix panic: slice index out of bounds in TSM index
 
 ## v0.11.0 [2016-03-22]
 

models/points.go

@@ -27,8 +27,13 @@ var (
 		'=': []byte(`\=`),
 	}
 
-	ErrPointMustHaveAField = errors.New("point without fields is unsupported")
-	ErrInvalidNumber       = errors.New("invalid number")
+	ErrPointMustHaveAField  = errors.New("point without fields is unsupported")
+	ErrInvalidNumber        = errors.New("invalid number")
+	ErrMaxKeyLengthExceeded = errors.New("max key length exceeded")
+)
+
+const (
+	MaxKeyLength = 65535
 )
 
 // Point defines the values that will be written to the database
@@ -204,6 +209,10 @@ func parsePoint(buf []byte, defaultTime time.Time, precision string) (Point, err
 		return nil, fmt.Errorf("missing measurement")
 	}
 
+	if len(key) > MaxKeyLength {
+		return nil, fmt.Errorf("max key length exceeded: %v > %v", len(key), MaxKeyLength)
+	}
+
 	// scan the second block is which is field1=value1[,field2=value2,...]
 	pos, fields, err := scanFields(buf, pos)
 	if err != nil {
@@ -1102,8 +1111,13 @@ func NewPoint(name string, tags Tags, fields Fields, time time.Time) (Point, err
 		}
 	}
 
+	key := MakeKey([]byte(name), tags)
+	if len(key) > MaxKeyLength {
+		return nil, fmt.Errorf("max key length exceeded: %v > %v", len(key), MaxKeyLength)
+	}
+
 	return &point{
-		key:    MakeKey([]byte(name), tags),
+		key:    key,
 		time:   time,
 		fields: fields.MarshalBinary(),
 	}, nil

models/points_test.go

@@ -1791,3 +1791,18 @@ func TestNewPointsRejectsEmptyFieldNames(t *testing.T) {
 		t.Fatalf("new point with empty field name. got: nil, expected: error")
 	}
 }
+
+func TestNewPointsRejectsMaxKey(t *testing.T) {
+	var key string
+	for i := 0; i < 65536; i++ {
+		key += "a"
+	}
+
+	if _, err := models.NewPoint(key, nil, models.Fields{"value": 1}, time.Now()); err == nil {
+		t.Fatalf("new point with max key. got: nil, expected: error")
+	}
+
+	if _, err := models.ParsePointsString(fmt.Sprintf("%v value=1", key)); err == nil {
+		t.Fatalf("parse point with max key. got: nil, expected: error")
+	}
+}

tsdb/engine/tsm1/writer.go

@@ -94,11 +94,15 @@ const (
 
 	// Max number of blocks for a given key that can exist in a single file
 	maxIndexEntries = (1 << (indexCountSize * 8)) - 1
+
+	// max length of a key in an index entry (measurement + tags)
+	maxKeyLength = (1 << (2 * 8)) - 1
 )
 
 var (
-	ErrNoValues  = fmt.Errorf("no values written")
-	ErrTSMClosed = fmt.Errorf("tsm file closed")
+	ErrNoValues             = fmt.Errorf("no values written")
+	ErrTSMClosed            = fmt.Errorf("tsm file closed")
+	ErrMaxKeyLengthExceeded = fmt.Errorf("max key length exceeded")
 )
 
 // TSMWriter writes TSM formatted key and values.
@@ -542,6 +546,10 @@ func (t *tsmWriter) Write(key string, values Values) error {
 		return nil
 	}
 
+	if len(key) > maxKeyLength {
+		return ErrMaxKeyLengthExceeded
+	}
+
 	// Write header only after we have some data to write.
 	if t.n == 0 {
 		if err := t.writeHeader(); err != nil {

tsdb/engine/tsm1/writer_test.go

@@ -3,6 +3,7 @@ package tsm1_test
 import (
 	"bytes"
 	"encoding/binary"
+	"os"
 	"testing"
 
 	"github.com/influxdata/influxdb/tsdb/engine/tsm1"
@@ -510,3 +511,23 @@ func TestTSMWriter_WriteBlock_Multiple(t *testing.T) {
 		}
 	}
 }
+
+func TestTSMWriter_Write_MaxKey(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 created writer: %v", err)
+	}
+
+	var key string
+	for i := 0; i < 100000; i++ {
+		key += "a"
+	}
+	if err := w.Write(key, []tsm1.Value{tsm1.NewValue(0, 1.0)}); err != tsm1.ErrMaxKeyLengthExceeded {
+		t.Fatalf("expected max key length error writing key: %v", err)
+	}
+}