merge develop

7 files changed, 394 insertions, 0 deletions

diff --git a/internal/io/dlog/loggers/factory.go b/internal/io/dlog/loggers/factory.go
new file mode 100644
index 0000000..a5cc7cf
--- /dev/null
+++ b/internal/io/dlog/loggers/factory.go
@@ -0,0 +1,54 @@
+package loggers
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+)
+
+var factoryMap map[string]Logger
+var factoryMutex sync.Mutex
+
+// Factory is there to retrieve a logger based on various settings.
+func Factory(sourceName, loggerName string, logRotation Strategy) Logger {
+	factoryMutex.Lock()
+	defer factoryMutex.Unlock()
+
+	id := fmt.Sprintf("sourceName:%s,fileBase:%s,loggerName:%s", sourceName,
+		logRotation.FileBase, loggerName)
+	if factoryMap == nil {
+		factoryMap = make(map[string]Logger)
+	}
+
+	singleton, ok := factoryMap[id]
+	if !ok {
+		switch strings.ToLower(loggerName) {
+		case "none":
+			singleton = none{}
+		case "stdout":
+			singleton = newStdout()
+			factoryMap[id] = singleton
+		case "file":
+			singleton = newFile(logRotation)
+			factoryMap[id] = singleton
+		case "fout":
+			singleton = newFout(logRotation)
+			factoryMap[id] = singleton
+		default:
+			panic(fmt.Sprintf("Unsupported logger type '%s'", loggerName))
+		}
+	}
+	return singleton
+}
+
+// FactoryRotate invokes a log rotation of all loggers.
+func FactoryRotate() {
+	factoryMutex.Lock()
+	defer factoryMutex.Unlock()
+	if factoryMap == nil {
+		return
+	}
+	for _, logger := range factoryMap {
+		logger.Rotate()
+	}
+}
diff --git a/internal/io/dlog/loggers/file.go b/internal/io/dlog/loggers/file.go
new file mode 100644
index 0000000..94824fe
--- /dev/null
+++ b/internal/io/dlog/loggers/file.go
@@ -0,0 +1,165 @@
+package loggers
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"os"
+	"runtime"
+	"sync"
+	"time"
+
+	"github.com/mimecast/dtail/internal/config"
+)
+
+type fileWriter struct{}
+
+type fileMessageBuf struct {
+	now     time.Time
+	message string
+}
+
+type file struct {
+	bufferCh     chan *fileMessageBuf
+	pauseCh      chan struct{}
+	resumeCh     chan struct{}
+	rotateCh     chan struct{}
+	flushCh      chan struct{}
+	fd           *os.File
+	writer       *bufio.Writer
+	mutex        sync.Mutex
+	started      bool
+	lastFileName string
+	strategy     Strategy
+}
+
+func newFile(strategy Strategy) *file {
+	return &file{
+		bufferCh: make(chan *fileMessageBuf, runtime.NumCPU()*100),
+		pauseCh:  make(chan struct{}),
+		resumeCh: make(chan struct{}),
+		rotateCh: make(chan struct{}),
+		flushCh:  make(chan struct{}),
+		strategy: strategy,
+	}
+}
+
+func (f *file) Start(ctx context.Context, wg *sync.WaitGroup) {
+	f.mutex.Lock()
+	defer func() {
+		f.started = true
+		f.mutex.Unlock()
+	}()
+
+	if f.started {
+		// Logger already started from another Goroutine.
+		wg.Done()
+		return
+	}
+
+	pause := func(ctx context.Context) {
+		select {
+		case <-f.resumeCh:
+			return
+		case <-ctx.Done():
+			return
+		}
+	}
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case m := <-f.bufferCh:
+				f.write(m)
+			case <-f.pauseCh:
+				pause(ctx)
+			case <-f.flushCh:
+				f.flush()
+			case <-ctx.Done():
+				f.flush()
+				f.fd.Close()
+				return
+			}
+		}
+	}()
+}
+
+func (f *file) Log(now time.Time, message string) {
+	f.bufferCh <- &fileMessageBuf{now, message}
+}
+
+func (f *file) LogWithColors(now time.Time, message, coloredMessage string) {
+	panic("Colors not supported in file logger")
+}
+
+func (f *file) Pause()  { f.pauseCh <- struct{}{} }
+func (f *file) Resume() { f.resumeCh <- struct{}{} }
+func (f *file) Flush()  { f.flushCh <- struct{}{} }
+
+func (f *file) Rotate()            { f.rotateCh <- struct{}{} }
+func (*file) SupportsColors() bool { return false }
+
+func (f *file) write(m *fileMessageBuf) {
+	select {
+	case <-f.rotateCh:
+		// Force re-opening the outfile next time in getWriter.
+		f.lastFileName = ""
+	default:
+	}
+
+	var writer *bufio.Writer
+	if f.strategy.Rotation == DailyRotation {
+		writer = f.getWriter(m.now.Format("20060102"))
+	} else {
+		writer = f.getWriter(f.strategy.FileBase)
+	}
+
+	writer.WriteString(m.message)
+	writer.WriteByte('\n')
+}
+
+func (f *file) getWriter(name string) *bufio.Writer {
+	if f.lastFileName == name {
+		return f.writer
+	}
+	if _, err := os.Stat(config.Common.LogDir); os.IsNotExist(err) {
+		if err = os.MkdirAll(config.Common.LogDir, 0755); err != nil {
+			panic(err)
+		}
+	}
+
+	logFile := fmt.Sprintf("%s/%s.log", config.Common.LogDir, name)
+	newFd, err := os.OpenFile(logFile, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644)
+	if err != nil {
+		panic(err)
+	}
+
+	// Close old writer.
+	if f.fd != nil {
+		f.writer.Flush()
+		f.fd.Close()
+	}
+	// Set new writer.
+	f.fd = newFd
+	f.writer = bufio.NewWriterSize(f.fd, 1)
+	f.lastFileName = name
+
+	return f.writer
+}
+
+func (f *file) flush() {
+	defer func() {
+		if f.writer != nil {
+			f.writer.Flush()
+		}
+	}()
+	for {
+		select {
+		case m := <-f.bufferCh:
+			f.write(m)
+		default:
+			return
+		}
+	}
+}
diff --git a/internal/io/dlog/loggers/fout.go b/internal/io/dlog/loggers/fout.go
new file mode 100644
index 0000000..60c318d
--- /dev/null
+++ b/internal/io/dlog/loggers/fout.go
@@ -0,0 +1,46 @@
+package loggers
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+type fout struct {
+	file   *file
+	stdout *stdout
+}
+
+// Logs to both, a file and stdout
+func newFout(strategy Strategy) *fout {
+	return &fout{file: newFile(strategy), stdout: newStdout()}
+}
+
+func (f *fout) Start(ctx context.Context, wg *sync.WaitGroup) {
+	go func() {
+		defer wg.Done()
+
+		var wg2 sync.WaitGroup
+		wg2.Add(2)
+		f.file.Start(ctx, &wg2)
+		f.stdout.Start(ctx, &wg2)
+		wg2.Wait()
+	}()
+}
+
+func (f *fout) Log(now time.Time, message string) {
+	f.stdout.Log(now, message)
+	f.file.Log(now, message)
+}
+
+func (f *fout) LogWithColors(now time.Time, message, coloredMessage string) {
+	f.stdout.LogWithColors(now, "", coloredMessage)
+	f.file.Log(now, message)
+}
+
+func (f *fout) Flush()  { f.stdout.Flush(); f.file.Flush() }
+func (f *fout) Pause()  { f.stdout.Pause(); f.file.Pause() }
+func (f *fout) Resume() { f.stdout.Resume(); f.file.Resume() }
+func (f *fout) Rotate() { f.file.Rotate() }
+
+func (fout) SupportsColors() bool { return true }
diff --git a/internal/io/dlog/loggers/logger.go b/internal/io/dlog/loggers/logger.go
new file mode 100644
index 0000000..d4e85de
--- /dev/null
+++ b/internal/io/dlog/loggers/logger.go
@@ -0,0 +1,19 @@
+package loggers
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// Logger is there to plug in your own log implementation.
+type Logger interface {
+	Log(now time.Time, message string)
+	LogWithColors(now time.Time, message, messageWithColors string)
+	Start(ctx context.Context, wg *sync.WaitGroup)
+	Flush()
+	Pause()
+	Resume()
+	Rotate()
+	SupportsColors() bool
+}
diff --git a/internal/io/dlog/loggers/none.go b/internal/io/dlog/loggers/none.go
new file mode 100644
index 0000000..270027f
--- /dev/null
+++ b/internal/io/dlog/loggers/none.go
@@ -0,0 +1,21 @@
+package loggers
+
+import (
+	"context"
+	"sync"
+	"time"
+)
+
+// don't log anything
+type none struct{}
+
+func (none) Start(ctx context.Context, wg *sync.WaitGroup) { wg.Done() }
+func (none) Log(now time.Time, message string)             {}
+
+func (none) LogWithColors(now time.Time, message, coloredMessage string) {}
+
+func (none) Flush()               {}
+func (none) Pause()               {}
+func (none) Resume()              {}
+func (none) Rotate()              {}
+func (none) SupportsColors() bool { return false }
diff --git a/internal/io/dlog/loggers/stdout.go b/internal/io/dlog/loggers/stdout.go
new file mode 100644
index 0000000..05485c6
--- /dev/null
+++ b/internal/io/dlog/loggers/stdout.go
@@ -0,0 +1,54 @@
+package loggers
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+)
+
+type stdout struct {
+	pauseCh  chan struct{}
+	resumeCh chan struct{}
+	mutex    sync.Mutex
+}
+
+func newStdout() *stdout {
+	return &stdout{
+		pauseCh:  make(chan struct{}),
+		resumeCh: make(chan struct{}),
+	}
+}
+
+func (s *stdout) Start(ctx context.Context, wg *sync.WaitGroup) {
+	wg.Done()
+}
+
+func (s *stdout) Log(now time.Time, message string) {
+	s.log(message)
+}
+
+func (s *stdout) LogWithColors(now time.Time, message, coloredMessage string) {
+	s.log(coloredMessage)
+}
+
+func (s *stdout) log(message string) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	select {
+	case <-s.pauseCh:
+		// Pause until resumed.
+		<-s.resumeCh
+	default:
+	}
+
+	fmt.Println(message)
+}
+
+func (s *stdout) Pause()  { s.pauseCh <- struct{}{} }
+func (s *stdout) Resume() { s.resumeCh <- struct{}{} }
+func (s *stdout) Flush()  {}
+func (s *stdout) Rotate() {}
+
+func (stdout) SupportsColors() bool { return true }
diff --git a/internal/io/dlog/loggers/strategy.go b/internal/io/dlog/loggers/strategy.go
new file mode 100644
index 0000000..48e7d44
--- /dev/null
+++ b/internal/io/dlog/loggers/strategy.go
@@ -0,0 +1,35 @@
+package loggers
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+// Rotation is the actual strategy used for log rotation..
+type Rotation int
+
+const (
+	// DailyRotation tells DTail to rotate its logs on a daily basis or on SIGHUP.
+	DailyRotation Rotation = iota
+	// SignalRotation tells DTail to rotate its logs only on SIGHUP.
+	SignalRotation Rotation = iota
+)
+
+// Strategy is a pair of the rotation and the file base.
+type Strategy struct {
+	// Rotation is the actual rotation strategy used.
+	Rotation Rotation
+	// FileBase can be a name (e.g. "dserver", "dmap") when signal rotation is used.
+	FileBase string
+}
+
+// NewStrategy returns the stratey based on its name.
+func NewStrategy(name string) Strategy {
+	switch strings.ToLower(name) {
+	case "daily":
+		return Strategy{DailyRotation, ""}
+	default:
+		return Strategy{SignalRotation, filepath.Base(os.Args[0])}
+	}
+}