read files bytewise for more control of whats happening - change transport protocol for more control over newlines

2 files changed, 33 insertions, 45 deletions

diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index 6757bd6..8a365a1 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -14,6 +14,7 @@ import (
 
 	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/protocol"
 	"github.com/mimecast/dtail/internal/regex"
 
 	"github.com/DataDog/zstd"
@@ -148,80 +149,64 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, t
 	if err != nil {
 		return err
 	}
-	rawLine := make([]byte, 0, 512)
 
 	lineLengthThreshold := 1024 * 1024 // 1mb
-	longLineWarning := false
+	warnedAboutLongLine := false
+	message := make([]byte, 0, 512)
 
 	for {
 		select {
 		case <-ctx.Done():
 			return nil
-		default:
-		}
-
-		select {
 		case <-truncate:
 			if isTruncated, err := f.truncated(fd); isTruncated {
 				return err
 			}
-			logger.Info(f.filePath, "Current offset", offset)
 		default:
 		}
 
-		// Read some bytes (max 4k at once as of go 1.12). isPrefix will
-		// be set if line does not fit into 4k buffer.
-		bytes, isPrefix, err := reader.ReadLine()
+		b, err := reader.ReadByte()
 
 		if err != nil {
-			// If EOF, sleep a couple of ms and return with nil error.
-			// If other error, return with non-nil error.
 			if err != io.EOF {
 				return err
 			}
 			if !f.seekEOF {
-				logger.Debug(f.FilePath(), "End of file reached")
+				logger.Info(f.FilePath(), "End of file reached")
 				return nil
 			}
 			time.Sleep(time.Millisecond * 100)
 			continue
 		}
+		offset++
 
-		rawLine = append(rawLine, bytes...)
-		offset += uint64(len(bytes))
-
-		if !isPrefix {
-			// last LineRead call returned contend until end of line.
-			rawLine = append(rawLine, '\n')
-			select {
-			case rawLines <- rawLine:
-			case <-ctx.Done():
-				return nil
+		switch b {
+		case '\n':
+			if len(message) == 0 {
+				time.Sleep(time.Millisecond * 100)
+				continue
 			}
-			rawLine = make([]byte, 0, 512)
-			if longLineWarning {
-				longLineWarning = false
-			}
-			continue
-		}
-
-		// Last LineRead call could not read content until end of line, buffer
-		// was too small. Determine whether we exceed the max line length we
-		// want dtail to send to the client at once. Possibly split up log line
-		// into multiple log lines.
-		if len(rawLine) >= lineLengthThreshold {
-			if !longLineWarning {
-				f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines")
-				// Only print out one warning per long log line.
-				longLineWarning = true
-			}
-			rawLine = append(rawLine, '\n')
 			select {
-			case rawLines <- rawLine:
+			case rawLines <- append(message, protocol.MessageDelimiter):
+				message = make([]byte, 0, 512)
+				warnedAboutLongLine = false
 			case <-ctx.Done():
 				return nil
 			}
-			rawLine = make([]byte, 0, 512)
+		default:
+			if len(message) >= lineLengthThreshold {
+				if !warnedAboutLongLine {
+					f.serverMessages <- logger.Warn(f.filePath, "Long log line, splitting into multiple lines")
+					warnedAboutLongLine = true
+				}
+				select {
+				case <-ctx.Done():
+					return nil
+				case rawLines <- append(message, protocol.MessageDelimiter):
+					message = make([]byte, 0, 512)
+				}
+			}
+			message = append(message, b)
 		}
 	}
 }
diff --git a/internal/io/logger/logger.go b/internal/io/logger/logger.go
index bb9dc02..3a3935d 100644
--- a/internal/io/logger/logger.go
+++ b/internal/io/logger/logger.go
@@ -205,7 +205,7 @@ func Trace(args ...interface{}) string {
 // Write log line to buffer and/or log file.
 func write(what, severity, message string) {
 	if Mode.logToStdout {
-		line := fmt.Sprintf("%s|%s|%s|%s\n", what, hostname, severity, message)
+		line := fmt.Sprintf("%s|%s|%s|%s", what, hostname, severity, message)
 
 		if config.Client.TermColorsEnable {
 			line = brush.Colorfy(line)
@@ -219,7 +219,7 @@ func write(what, severity, message string) {
 		timeStr := t.Format("20060102-150405")
 		fileLogBufCh <- buf{
 			time:    t,
-			message: fmt.Sprintf("%s|%s|%s|%s\n", severity, timeStr, what, message),
+			message: fmt.Sprintf("%s|%s|%s|%s", severity, timeStr, what, message),
 		}
 	}
 }
@@ -326,6 +326,7 @@ func Flush() {
 		select {
 		case message := <-stdoutBufCh:
 			stdoutWriter.WriteString(message)
+			stdoutWriter.WriteString("\n")
 		default:
 			stdoutWriter.Flush()
 			return
@@ -338,6 +339,7 @@ func writeToStdout(ctx context.Context) {
 		select {
 		case message := <-stdoutBufCh:
 			stdoutWriter.WriteString(message)
+			stdoutWriter.WriteString("\n")
 		case <-time.After(time.Millisecond * 100):
 			stdoutWriter.Flush()
 		case <-pauseCh:
@@ -365,6 +367,7 @@ func writeToFile(ctx context.Context) {
 			dateStr := buf.time.Format("20060102")
 			w := fileWriter(dateStr)
 			w.WriteString(buf.message)
+			w.WriteString("\n")
 		case <-pauseCh:
 		PAUSE:
 			for {