Fix concurrent SetTags race on config.Tags

serf/delegate.go

@@ -20,9 +20,10 @@ type delegate struct {
 var _ memberlist.Delegate = &delegate{}
 
 func (d *delegate) NodeMeta(limit int) []byte {
-	roleBytes := d.serf.encodeTags(d.serf.config.Tags)
+	tags := d.serf.getTags()
+	roleBytes := d.serf.encodeTags(tags)
 	if len(roleBytes) > limit {
-		panic(fmt.Errorf("Node tags '%v' exceeds length limit of %d bytes", d.serf.config.Tags, limit))
+		panic(fmt.Errorf("Node tags '%v' exceeds length limit of %d bytes", tags, limit))
 	}
 
 	return roleBytes

serf/query.go

@@ -242,8 +242,7 @@ func (s *Serf) shouldProcessQuery(filters [][]byte) bool {
 			}
 
 			// Check if we match this regex
-			tags := s.config.Tags
-			matched, err := regexp.MatchString(filt.Expr, tags[filt.Tag])
+			matched, err := regexp.MatchString(filt.Expr, s.getTags()[filt.Tag])
 			if err != nil {
 				s.logger.Printf("[WARN] serf: failed to compile filter regex (%s): %v", filt.Expr, err)
 				return false

serf/serf.go

@@ -67,6 +67,8 @@ type Serf struct {
 
 	broadcasts    *memberlist.TransmitLimitedQueue
 	config        *Config
+	tagsLock      sync.RWMutex // protects config.Tags
+	setTagsLock   sync.Mutex   // serializes memberlist.UpdateNode from SetTags
 	failedMembers []*memberState
 	leftMembers   []*memberState
 	memberlist    *memberlist.Memberlist
@@ -621,13 +623,23 @@ func (s *Serf) SetTags(tags map[string]string) error {
 			memberlist.MetaMaxSize)
 	}
 
-	// Update the config
+	s.tagsLock.Lock()
 	s.config.Tags = tags
+	s.tagsLock.Unlock()
 
-	// Trigger a memberlist update
+	// Serialize UpdateNode; memberlist is not safe for concurrent calls.
+	s.setTagsLock.Lock()
+	defer s.setTagsLock.Unlock()
 	return s.memberlist.UpdateNode(s.config.BroadcastTimeout)
 }
 
+func (s *Serf) getTags() map[string]string {
+	s.tagsLock.RLock()
+	tags := s.config.Tags
+	s.tagsLock.RUnlock()
+	return tags
+}
+
 // Join joins an existing Serf cluster. Returns the number of nodes
 // successfully contacted. The returned error will be non-nil only in the
 // case that no nodes could be contacted. If ignoreOld is true, then any

serf/serf_test.go

@@ -2005,6 +2005,41 @@ func TestSerf_SetTags(t *testing.T) {
 		[]EventType{EventMemberJoin, EventMemberUpdate})
 }
 
+func TestSerf_SetTags_Concurrent(t *testing.T) {
+	ip1, returnFn1 := testutil.TakeIP()
+	defer returnFn1()
+
+	s1Config := testConfig(t, ip1)
+	s1Config.BroadcastTimeout = time.Millisecond
+	s1, err := Create(s1Config)
+	if err != nil {
+		t.Fatalf("err: %v", err)
+	}
+	defer s1.Shutdown()
+
+	const goroutines = 16
+	const iterations = 32
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	for i := 0; i < goroutines; i++ {
+		wg.Go(func() {
+			<-start
+			for j := 0; j < iterations; j++ {
+				tags := map[string]string{
+					"id": strconv.Itoa(i),
+					"n":  strconv.Itoa(j),
+				}
+				if err := s1.SetTags(tags); err != nil {
+					t.Errorf("SetTags: %v", err)
+					return
+				}
+			}
+		})
+	}
+	close(start)
+	wg.Wait()
+}
+
 func TestSerf_Query(t *testing.T) {
 	ip1, returnFn1 := testutil.TakeIP()
 	defer returnFn1()