Prevent concurrent access to a Series' shard IDs

Fixes #6235

CHANGELOG.md

@@ -1,5 +1,17 @@
+## v0.13.0 [unreleased]
+
+### Release Notes
+
+### Features
+
+### Bugfixes
+
+- [#6235](https://github.com/influxdata/influxdb/issues/6235): Fix panic caused by concurrent access to Series' shard IDs.
+
 ## v0.12.0 [2016-04-05]
+
 ### Release Notes
+
 Upgrading to this release requires a little more than just installing the new binary and starting it up. The upgrade process is very quick and should only require a minute of downtime or less. Details on [upgrading to 0.12 are here](https://docs.influxdata.com/influxdb/v0.12/administration/upgrading/).
 
 This release removes all of the old clustering code. It operates as a standalone server. For a free open source HA setup see the [InfluxDB Relay](https://github.com/influxdata/influxdb-relay).

tsdb/meta.go

@@ -1314,6 +1314,14 @@ func (s *Series) AssignShard(shardID uint64) {
 	s.mu.Unlock()
 }
 
+// InShard determins if the shard specified by shardID has defined
+// the Series.
+func (s *Series) InShard(shardID uint64) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.shardIDs[shardID]
+}
+
 // MarshalBinary encodes the object to a binary format.
 func (s *Series) MarshalBinary() ([]byte, error) {
 	s.mu.RLock()

tsdb/shard.go

@@ -371,7 +371,7 @@ func (s *Shard) validateSeriesAndFields(points []models.Point) ([]*SeriesCreate,
 			series := NewSeries(string(p.Key()), p.Tags())
 			seriesToCreate = append(seriesToCreate, &SeriesCreate{p.Name(), series})
 			seriesToAddShardTo = append(seriesToAddShardTo, series.Key)
-		} else if !ss.shardIDs[s.id] {
+		} else if !ss.InShard(s.id) {
 			// this is the first time this series is being written into this shard, persist it
 			seriesToCreate = append(seriesToCreate, &SeriesCreate{p.Name(), ss})
 			seriesToAddShardTo = append(seriesToAddShardTo, ss.Key)