DS farm: buffer per-DS update channel

Occasionally the daemon set farm locks up with the farm goroutine
blocking forever attempting to send an update to a daemon set worker
goroutine. This can happen due to a race where the worker thread might
exit for a number of reasons after the farm goroutine checks the child
map to determine a worker already exists but before sending an update.

This commit sidesteps the problem by buffering the per-daemon set update
channel so that the farm goroutine will never block sending to a worker.

If a worker dies, an existing routine grabs a mutex protecting the child
map and clears out the child entry and drains the buffered channel. The
next time an update is seen for the daemon set, the farm loop should
know that it needs to spawn another worker.

Author: mpuncel

pkg/ds/farm.go

@@ -87,7 +87,7 @@ type Farm struct {
 type childDS struct {
 	ds        DaemonSet
 	cancel    context.CancelFunc
-	updatedCh chan<- ds_fields.DaemonSet
+	updatedCh chan ds_fields.DaemonSet
 	deletedCh chan<- ds_fields.DaemonSet
 	errCh     <-chan error
 	unlocker  consul.TxnUnlocker
@@ -310,6 +310,10 @@ func (dsf *Farm) closeChild(dsID fields.ID) {
 	if child, ok := dsf.children[dsID]; ok {
 		dsf.logger.WithField("ds", dsID).Infoln("Releasing daemon set")
 		child.cancel()
+
+		// drain the updatedCh (it's buffered)
+		for range child.updatedCh {
+		}
 		close(child.updatedCh)
 		close(child.deletedCh)
 
@@ -560,7 +564,9 @@ func (dsf *Farm) spawnDaemonSet(
 		dsf.statusWritingInterval,
 	)
 
-	updatedCh := make(chan ds_fields.DaemonSet)
+	// updatedCh is buffered by 1 to protect the control loop by slow (or
+	// dead) readers
+	updatedCh := make(chan ds_fields.DaemonSet, 1)
 	deletedCh := make(chan ds_fields.DaemonSet)
 	ctx, cancel := context.WithCancel(ctx)
 
@@ -733,6 +739,14 @@ func (dsf *Farm) lockAndSpawn(ctx context.Context, dsFields ds_fields.DaemonSet,
 
 	// If we already are running the daemon set, just pass the update. Otherwise spawn one
 	if ok {
+		// try to drain the buffered value off the updatedCh if there is one (which
+		// indicates the worker goroutine was slow to read it or is dead)
+		select {
+		case <-child.updatedCh:
+			dsf.logger.Warnln("daemon set worker missed an update, sending a newer one")
+		default:
+		}
+
 		child.updatedCh <- dsFields
 	} else {
 		dsf.children[dsFields.ID] = dsf.spawnDaemonSet(