Fix race on measurementFields

Both Shard and Engine had the same reference to the measurementField map,
but they each protected it with their own locks.  This causes a race when
write and queries are occurring because writes can add new fields to the
map while queries are reading from it.

The fix moves the ownership to the Engine and provides protected accessors
to that Shard now users.  For the most parts, the access on shard were old
dead code.

Fixing the measurementFields map race created a new race on the internal
fields map.  This is now unexported and protected via MeasurementFields
exported funcs.

Fixes #6188

tsdb/engine.go

@@ -29,14 +29,15 @@ type Engine interface {
 	Close() error
 
 	SetLogOutput(io.Writer)
-	LoadMetadataIndex(shard *Shard, index *DatabaseIndex, measurementFields map[string]*MeasurementFields) error
+	LoadMetadataIndex(shard *Shard, index *DatabaseIndex) error
 
 	CreateIterator(opt influxql.IteratorOptions) (influxql.Iterator, error)
 	SeriesKeys(opt influxql.IteratorOptions) (influxql.SeriesList, error)
-	WritePoints(points []models.Point, measurementFieldsToSave map[string]*MeasurementFields, seriesToCreate []*SeriesCreate) error
+	WritePoints(points []models.Point) error
 	DeleteSeries(keys []string) error
 	DeleteMeasurement(name string, seriesKeys []string) error
 	SeriesCount() (n int, err error)
+	MeasurementFields(measurement string) *MeasurementFields
 
 	// Format will return the format for the engine
 	Format() EngineFormat

tsdb/engine/tsm1/engine.go

@@ -78,8 +78,9 @@ func NewEngine(path string, walPath string, opt tsdb.EngineOptions) tsdb.Engine
 	}
 
 	e := &Engine{
-		path:   path,
-		logger: log.New(os.Stderr, "[tsm1] ", log.LstdFlags),
+		path:              path,
+		logger:            log.New(os.Stderr, "[tsm1] ", log.LstdFlags),
+		measurementFields: make(map[string]*tsdb.MeasurementFields),
 
 		WAL:   w,
 		Cache: cache,
@@ -110,13 +111,23 @@ func (e *Engine) Index() *tsdb.DatabaseIndex {
 }
 
 // MeasurementFields returns the measurement fields for a measurement.
-func (e *Engine) MeasurementFields(name string) *tsdb.MeasurementFields {
+func (e *Engine) MeasurementFields(measurement string) *tsdb.MeasurementFields {
+	e.mu.RLock()
+	m := e.measurementFields[measurement]
+	e.mu.RUnlock()
+
+	if m != nil {
+		return m
+	}
+
 	e.mu.Lock()
-	defer e.mu.Unlock()
-	if e.measurementFields[name] == nil {
-		e.measurementFields[name] = &tsdb.MeasurementFields{Fields: make(map[string]*tsdb.Field)}
+	m = e.measurementFields[measurement]
+	if m == nil {
+		m = tsdb.NewMeasurementFields()
+		e.measurementFields[measurement] = m
 	}
-	return e.measurementFields[name]
+	e.mu.Unlock()
+	return m
 }
 
 // Format returns the format type of this engine
@@ -187,10 +198,9 @@ func (e *Engine) Close() error {
 func (e *Engine) SetLogOutput(w io.Writer) {}
 
 // LoadMetadataIndex loads the shard metadata into memory.
-func (e *Engine) LoadMetadataIndex(sh *tsdb.Shard, index *tsdb.DatabaseIndex, measurementFields map[string]*tsdb.MeasurementFields) error {
+func (e *Engine) LoadMetadataIndex(sh *tsdb.Shard, index *tsdb.DatabaseIndex) error {
 	// Save reference to index for iterator creation.
 	e.index = index
-	e.measurementFields = measurementFields
 
 	start := time.Now()
 
@@ -200,7 +210,7 @@ func (e *Engine) LoadMetadataIndex(sh *tsdb.Shard, index *tsdb.DatabaseIndex, me
 			return err
 		}
 
-		if err := e.addToIndexFromKey(key, fieldType, index, measurementFields); err != nil {
+		if err := e.addToIndexFromKey(key, fieldType, index); err != nil {
 			return err
 		}
 		return nil
@@ -220,7 +230,7 @@ func (e *Engine) LoadMetadataIndex(sh *tsdb.Shard, index *tsdb.DatabaseIndex, me
 			continue
 		}
 
-		if err := e.addToIndexFromKey(key, fieldType, index, measurementFields); err != nil {
+		if err := e.addToIndexFromKey(key, fieldType, index); err != nil {
 			return err
 		}
 	}
@@ -297,19 +307,17 @@ func (e *Engine) writeFileToBackup(f FileStat, shardRelativePath string, tw *tar
 
 // addToIndexFromKey will pull the measurement name, series key, and field name from a composite key and add it to the
 // database index and measurement fields
-func (e *Engine) addToIndexFromKey(key string, fieldType influxql.DataType, index *tsdb.DatabaseIndex, measurementFields map[string]*tsdb.MeasurementFields) error {
+func (e *Engine) addToIndexFromKey(key string, fieldType influxql.DataType, index *tsdb.DatabaseIndex) error {
 	seriesKey, field := seriesAndFieldFromCompositeKey(key)
 	measurement := tsdb.MeasurementFromSeriesKey(seriesKey)
 
 	m := index.CreateMeasurementIndexIfNotExists(measurement)
 	m.SetFieldName(field)
 
-	mf := measurementFields[measurement]
+	mf := e.measurementFields[measurement]
 	if mf == nil {
-		mf = &tsdb.MeasurementFields{
-			Fields: map[string]*tsdb.Field{},
-		}
-		measurementFields[measurement] = mf
+		mf = tsdb.NewMeasurementFields()
+		e.measurementFields[measurement] = mf
 	}
 
 	if err := mf.CreateFieldIfNotExists(field, fieldType, false); err != nil {
@@ -330,7 +338,7 @@ func (e *Engine) addToIndexFromKey(key string, fieldType influxql.DataType, inde
 
 // WritePoints writes metadata and point data into the engine.
 // Returns an error if new points are added to an existing key.
-func (e *Engine) WritePoints(points []models.Point, measurementFieldsToSave map[string]*tsdb.MeasurementFields, seriesToCreate []*tsdb.SeriesCreate) error {
+func (e *Engine) WritePoints(points []models.Point) error {
 	values := map[string][]Value{}
 	for _, p := range points {
 		for k, v := range p.Fields() {
@@ -397,6 +405,10 @@ func (e *Engine) DeleteSeries(seriesKeys []string) error {
 
 // DeleteMeasurement deletes a measurement and all related series.
 func (e *Engine) DeleteMeasurement(name string, seriesKeys []string) error {
+	e.mu.Lock()
+	delete(e.measurementFields, name)
+	e.mu.Unlock()
+
 	return e.DeleteSeries(seriesKeys)
 }
 
@@ -740,7 +752,7 @@ func (e *Engine) SeriesKeys(opt influxql.IteratorOptions) (influxql.SeriesList,
 							return influxql.Unknown
 						}
 
-						f := mf.Fields[field]
+						f := mf.Field(field)
 						if f == nil {
 							return influxql.Unknown
 						}
@@ -904,7 +916,7 @@ func (e *Engine) buildCursor(measurement, seriesKey, field string, opt influxql.
 	}
 
 	// Find individual field.
-	f := mf.Fields[field]
+	f := mf.Field(field)
 	if f == nil {
 		return nil
 	}

tsdb/engine/tsm1/engine_test.go

@@ -34,7 +34,7 @@ func TestEngine_LoadMetadataIndex(t *testing.T) {
 
 	// Load metadata index.
 	index := tsdb.NewDatabaseIndex("db")
-	if err := e.LoadMetadataIndex(nil, index, make(map[string]*tsdb.MeasurementFields)); err != nil {
+	if err := e.LoadMetadataIndex(nil, index); err != nil {
 		t.Fatal(err)
 	}
 
@@ -57,7 +57,7 @@ func TestEngine_LoadMetadataIndex(t *testing.T) {
 
 	// Load metadata index.
 	index = tsdb.NewDatabaseIndex("db")
-	if err := e.LoadMetadataIndex(nil, index, make(map[string]*tsdb.MeasurementFields)); err != nil {
+	if err := e.LoadMetadataIndex(nil, index); err != nil {
 		t.Fatal(err)
 	}
 
@@ -71,7 +71,7 @@ func TestEngine_LoadMetadataIndex(t *testing.T) {
 	// Write a new point and ensure we can close and load index from TSM and WAL
 	if err := e.WritePoints([]models.Point{
 		MustParsePointString("cpu,host=B value=1.2 2000000000"),
-	}, nil, nil); err != nil {
+	}); err != nil {
 		t.Fatalf("failed to write points: %s", err.Error())
 	}
 
@@ -82,7 +82,7 @@ func TestEngine_LoadMetadataIndex(t *testing.T) {
 
 	// Load metadata index.
 	index = tsdb.NewDatabaseIndex("db")
-	if err := e.LoadMetadataIndex(nil, index, make(map[string]*tsdb.MeasurementFields)); err != nil {
+	if err := e.LoadMetadataIndex(nil, index); err != nil {
 		t.Fatal(err)
 	}
 
@@ -152,14 +152,14 @@ func TestEngine_Backup(t *testing.T) {
 		t.Fatalf("failed to open tsm1 engine: %s", err.Error())
 	}
 
-	if err := e.WritePoints([]models.Point{p1}, nil, nil); err != nil {
+	if err := e.WritePoints([]models.Point{p1}); err != nil {
 		t.Fatalf("failed to write points: %s", err.Error())
 	}
 	if err := e.WriteSnapshot(); err != nil {
 		t.Fatalf("failed to snapshot: %s", err.Error())
 	}
 
-	if err := e.WritePoints([]models.Point{p2}, nil, nil); err != nil {
+	if err := e.WritePoints([]models.Point{p2}); err != nil {
 		t.Fatalf("failed to write points: %s", err.Error())
 	}
 
@@ -189,7 +189,7 @@ func TestEngine_Backup(t *testing.T) {
 	// so this test won't work properly unless the file is at least a second past the last one
 	time.Sleep(time.Second)
 
-	if err := e.WritePoints([]models.Point{p3}, nil, nil); err != nil {
+	if err := e.WritePoints([]models.Point{p3}); err != nil {
 		t.Fatalf("failed to write points: %s", err.Error())
 	}
 
@@ -521,7 +521,7 @@ func MustOpenEngine() *Engine {
 	if err := e.Open(); err != nil {
 		panic(err)
 	}
-	if err := e.LoadMetadataIndex(nil, tsdb.NewDatabaseIndex("db"), make(map[string]*tsdb.MeasurementFields)); err != nil {
+	if err := e.LoadMetadataIndex(nil, tsdb.NewDatabaseIndex("db")); err != nil {
 		panic(err)
 	}
 	return e
@@ -559,7 +559,7 @@ func (e *Engine) MustWriteSnapshot() {
 
 // WritePointsString parses a string buffer and writes the points.
 func (e *Engine) WritePointsString(buf ...string) error {
-	return e.WritePoints(MustParsePointsString(strings.Join(buf, "\n")), nil, nil)
+	return e.WritePoints(MustParsePointsString(strings.Join(buf, "\n")))
 }
 
 // MustParsePointsString parses points from a string. Panic on error.

tsdb/meta.go

@@ -53,8 +53,9 @@ func NewDatabaseIndex(name string) *DatabaseIndex {
 // Series returns a series by key.
 func (d *DatabaseIndex) Series(key string) *Series {
 	d.mu.RLock()
-	defer d.mu.RUnlock()
-	return d.series[key]
+	s := d.series[key]
+	d.mu.RUnlock()
+	return s
 }
 
 // SeriesN returns the number of series.
@@ -146,7 +147,7 @@ func (d *DatabaseIndex) CreateMeasurementIndexIfNotExists(name string) *Measurem
 	// and acquire the write lock
 	m = d.measurements[name]
 	if m == nil {
-		m = NewMeasurement(name, d)
+		m = NewMeasurement(name)
 		d.measurements[name] = m
 		d.statMap.Add(statDatabaseMeasurements, 1)
 	}
@@ -293,7 +294,9 @@ func (d *DatabaseIndex) measurementsByTagFilters(filters []*TagFilter) Measureme
 	for _, m := range d.measurements {
 		// Iterate filters seeing if the measurement has a matching tag.
 		for _, f := range filters {
+			m.mu.RLock()
 			tagVals, ok := m.seriesByTagKeyValue[f.Key]
+			m.mu.RUnlock()
 			if !ok {
 				continue
 			}
@@ -352,10 +355,12 @@ func (d *DatabaseIndex) MeasurementsByRegex(re *regexp.Regexp) Measurements {
 
 // Measurements returns a list of all measurements.
 func (d *DatabaseIndex) Measurements() Measurements {
+	d.mu.RLock()
 	measurements := make(Measurements, 0, len(d.measurements))
 	for _, m := range d.measurements {
 		measurements = append(measurements, m)
 	}
+	d.mu.RUnlock()
 	return measurements
 }
 
@@ -408,7 +413,6 @@ type Measurement struct {
 	mu         sync.RWMutex
 	Name       string `json:"name,omitempty"`
 	fieldNames map[string]struct{}
-	index      *DatabaseIndex
 
 	// in-memory index fields
 	seriesByID          map[uint64]*Series // lookup table for series by their id
@@ -418,11 +422,10 @@ type Measurement struct {
 }
 
 // NewMeasurement allocates and initializes a new Measurement.
-func NewMeasurement(name string, idx *DatabaseIndex) *Measurement {
+func NewMeasurement(name string) *Measurement {
 	return &Measurement{
 		Name:       name,
 		fieldNames: make(map[string]struct{}),
-		index:      idx,
 
 		seriesByID:          make(map[uint64]*Series),
 		seriesByTagKeyValue: make(map[string]map[string]SeriesIDs),
@@ -433,7 +436,12 @@ func NewMeasurement(name string, idx *DatabaseIndex) *Measurement {
 // HasField returns true if the measurement has a field by the given name
 func (m *Measurement) HasField(name string) bool {
 	m.mu.RLock()
-	defer m.mu.RUnlock()
+	hasField := m.hasField(name)
+	m.mu.RUnlock()
+	return hasField
+}
+
+func (m *Measurement) hasField(name string) bool {
 	_, hasField := m.fieldNames[name]
 	return hasField
 }
@@ -606,8 +614,6 @@ func (m *Measurement) filters(condition influxql.Expr) (map[uint64]influxql.Expr
 // influx filter expression that goes with the series
 // TODO: this shouldn't be exported. However, until tx.go and the engine get refactored into tsdb, we need it.
 func (m *Measurement) TagSets(dimensions []string, condition influxql.Expr) ([]*influxql.TagSet, error) {
-	m.index.mu.RLock()
-	defer m.index.mu.RUnlock()
 	m.mu.RLock()
 	defer m.mu.RUnlock()
 
@@ -753,7 +759,7 @@ func (m *Measurement) idsForExpr(n *influxql.BinaryExpr) (SeriesIDs, influxql.Ex
 
 	// For fields, return all series IDs from this measurement and return
 	// the expression passed in, as the filter.
-	if name.Val != "_name" && m.HasField(name.Val) {
+	if name.Val != "_name" && m.hasField(name.Val) {
 		return m.seriesIDs, n, nil
 	}
 
@@ -1314,6 +1320,13 @@ func (s *Series) AssignShard(shardID uint64) {
 	s.mu.Unlock()
 }
 
+func (s *Series) Assigned(shardID uint64) bool {
+	s.mu.RLock()
+	b := s.shardIDs[shardID]
+	s.mu.RUnlock()
+	return b
+}
+
 // MarshalBinary encodes the object to a binary format.
 func (s *Series) MarshalBinary() ([]byte, error) {
 	s.mu.RLock()