host/logmux: Add timeouts to sink writes

host/logmux/logmux.go

@@ -5,6 +5,7 @@ import (
 	"bytes"
 	"io"
 	"io/ioutil"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"regexp"
@@ -95,9 +96,18 @@ func (m *Mux) broadcast(app string, msg message) {
 	m.subscribersMtx.RLock()
 	defer m.subscribersMtx.RUnlock()
 
+	timeout := time.NewTimer(time.Second)
+	l := m.logger.New("fn", "broadcast", "sample", "0.1")
+	r := rand.New(rand.NewSource(time.Now().Unix()))
 	for ch := range m.subscribers[firehoseApp] {
-		ch <- msg
-		// TODO: if blocking write, drop+notify, eventually close
+		timeout.Reset(time.Second)
+		select {
+		case ch <- msg:
+		case <-timeout.C:
+			if r.Intn(9) == 0 {
+				l.Error("dropping log line due to sink write timeout")
+			}
+		}
 	}
 	for ch := range m.subscribers[app] {
 		ch <- msg

host/logmux/sink.go

@@ -442,6 +442,7 @@ func (s *LogAggregatorSink) GetCursor(hostID string) (*utils.HostCursor, error)
 }
 
 func (s *LogAggregatorSink) Write(m message) error {
+	s.conn.SetWriteDeadline(time.Now().Add(time.Second))
 	_, err := s.conn.Write(rfc6587.Bytes(m.Message))
 	return err
 }
@@ -650,6 +651,7 @@ func (s *SyslogSink) Write(m message) error {
 	case ct.SyslogFormatNewline:
 		data = append(msg.Bytes(), '\n')
 	}
+	s.conn.SetWriteDeadline(time.Now().Add(time.Second))
 	_, err := s.conn.Write(data)
 	if err != nil {
 		return err