Fix race condition leading to duplicate tailing (#181)

* Fix race condition leading to duplicate tailing

* Add threadsafe testWriter

* Make sure the test runs in a clean folder

* Remove line check, panic must have been caused by something else

* Switch to a test registry

* Fix race condition in logger

Author: Bowbaq

remote_syslog.go

@@ -26,7 +26,7 @@ var (
 type Server struct {
 	config   *Config
 	logger   *syslog.Logger
-	registry *WorkerRegistry
+	registry WorkerRegistry
 	stopChan chan struct{}
 	stopped  bool
 	mu       sync.RWMutex
@@ -35,7 +35,7 @@ type Server struct {
 func NewServer(config *Config) *Server {
 	return &Server{
 		config:   config,
-		registry: NewWorkerRegistry(),
+		registry: NewInMemoryRegistry(),
 		stopChan: make(chan struct{}),
 	}
 }
@@ -99,7 +99,6 @@ func (s *Server) closing() bool {
 // Tails a single file
 func (s *Server) tailOne(file, tag string, whence int) {
 	defer s.registry.Remove(file)
-	s.registry.Add(file)
 
 	t, err := tail.TailFile(file, tail.Config{
 		ReOpen:    true,
@@ -196,6 +195,7 @@ func (s *Server) globFiles(firstPass bool) {
 					whence = io.SeekEnd
 				}
 
+				s.registry.Add(file)
 				go s.tailOne(file, tag, whence)
 			}
 		}

remote_syslog_test.go

@@ -6,6 +6,8 @@ import (
 	"net"
 	"os"
 	"regexp"
+	"strconv"
+	"sync"
 	"testing"
 	"time"
 
@@ -83,6 +85,51 @@ func TestNewFileSeek(t *testing.T) {
 	}
 }
 
+func TestGlobCollisions(t *testing.T) {
+	assert := assert.New(t)
+
+	// Make sure we're running on a clean directory
+	os.RemoveAll(tmpdir)
+	os.Mkdir(tmpdir, 0755)
+
+	// Add colliding globs
+	config := testConfig()
+	config.Files = append(config.Files, LogFile{
+		Path: "tmp/*.log",
+	})
+
+	// Use an observable registry
+	testRegistry := &testRegistry{workers: make(map[string]int)}
+
+	s := NewServer(config)
+	s.registry = testRegistry
+	go s.Start()
+	defer s.Close()
+
+	// just a quick rest to get the server started
+	time.Sleep(1 * time.Second)
+
+	var files []*os.File
+	for i := 0; i < 50; i++ {
+		file := tmpLogFile()
+		files = append(files, file)
+		writeLog(file, "the most important message"+strconv.Itoa(i))
+	}
+
+	// NewFileCheckInterval = 1 second, so wait 1100ms for messages
+	time.Sleep(3000 * time.Millisecond)
+
+	testRegistry.mu.RLock()
+	for file, forwardCount := range testRegistry.workers {
+		assert.Equal(1, forwardCount, "Expected %s to be added once, got %d", file, forwardCount)
+	}
+	testRegistry.mu.RUnlock()
+
+	for _, file := range files {
+		file.Close()
+	}
+}
+
 // write to test log file
 func writeLog(file *os.File, msg string) {
 	w := bufio.NewWriterSize(file, 1024*32)
@@ -148,3 +195,30 @@ func testConfig() *Config {
 		},
 	}
 }
+
+// testRegistry is a WorkerRegistry implementation that keeps track of how many times a file was added
+type testRegistry struct {
+	mu      sync.RWMutex
+	workers map[string]int
+}
+
+func (tr *testRegistry) Exists(worker string) bool {
+	tr.mu.RLock()
+	defer tr.mu.RUnlock()
+	_, ok := tr.workers[worker]
+	return ok
+}
+
+func (tr *testRegistry) Add(worker string) {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	log.Tracef("Adding %s to worker registry", worker)
+	tr.workers[worker] += 1
+}
+
+func (tr *testRegistry) Remove(worker string) {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	log.Tracef("Removing %s from worker registry", worker)
+	delete(tr.workers, worker)
+}

syslog/syslog.go

@@ -98,7 +98,7 @@ type Logger struct {
 	connectTimeout   time.Duration
 	writeTimeout     time.Duration
 	tcpMaxLineLength int
-	mu               sync.Mutex
+	mu               sync.RWMutex
 	stopChan         chan struct{}
 	stopped          bool
 }
@@ -127,6 +127,9 @@ func Dial(clientHostname, network, raddr string, rootCAs *x509.CertPool, connect
 }
 
 func (l *Logger) Write(packet Packet) {
+	l.mu.RLock()
+	defer l.mu.RUnlock()
+
 	if l.stopped {
 		return
 	}

worker_registry.go

@@ -4,33 +4,46 @@ import (
 	"sync"
 )
 
-func NewWorkerRegistry() *WorkerRegistry {
-	return &WorkerRegistry{workers: make(map[string]bool)}
+// WorkerRegistry keeps track of which files are being actively tailed in order to avoid sending the same logs
+// multiple times. Implementations must be thread-safe.
+type WorkerRegistry interface {
+	// Exists returns true if a log file is currently being tailed
+	Exists(worker string) bool
+
+	// Add marks a log file as being currently tailed
+	Add(worker string)
+
+	// Remove clears a log file from the registry
+	Remove(worker string)
 }
 
-type WorkerRegistry struct {
-	workers map[string]bool
+// InMemoryRegistry is a simple WorkerRegistry implementation that uses a map protected by a sync.RWMutex.
+type InMemoryRegistry struct {
 	mu      sync.RWMutex
+	workers map[string]bool
+}
+
+func NewInMemoryRegistry() WorkerRegistry {
+	return &InMemoryRegistry{workers: make(map[string]bool)}
 }
 
-func (w *WorkerRegistry) Exists(worker string) bool {
-	w.mu.RLock()
-	defer w.mu.RUnlock()
-	_, ok := w.workers[worker]
+func (imr *InMemoryRegistry) Exists(worker string) bool {
+	imr.mu.RLock()
+	defer imr.mu.RUnlock()
+	_, ok := imr.workers[worker]
 	return ok
 }
 
-func (w *WorkerRegistry) Add(worker string) {
-	w.mu.Lock()
-	defer w.mu.Unlock()
+func (imr *InMemoryRegistry) Add(worker string) {
+	imr.mu.Lock()
+	defer imr.mu.Unlock()
 	log.Tracef("Adding %s to worker registry", worker)
-	w.workers[worker] = true
+	imr.workers[worker] = true
 }
 
-func (w *WorkerRegistry) Remove(worker string) error {
-	w.mu.Lock()
-	defer w.mu.Unlock()
+func (imr *InMemoryRegistry) Remove(worker string) {
+	imr.mu.Lock()
+	defer imr.mu.Unlock()
 	log.Tracef("Removing %s from worker registry", worker)
-	delete(w.workers, worker)
-	return nil
+	delete(imr.workers, worker)
 }