Fix spoolfile message formatting

producer/socket.go

@@ -181,7 +181,7 @@ func (prod *Socket) sendBatch() {
 	// Flush the buffer to the connection if it is active
 	if prod.tryConnect() {
 		prod.batch.Flush(prod.assembly.Write)
-	} else if prod.IsStopping() {
+	} else {
 		prod.batch.Flush(prod.assembly.Flush)
 	}
 }

producer/spoolfile.go

@@ -25,7 +25,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/rcrowley/go-metrics"
+	metrics "github.com/rcrowley/go-metrics"
 
 	"github.com/trivago/gollum/core"
 	"github.com/trivago/tgo/tio"
@@ -170,13 +170,19 @@ func (spool *spoolFile) decode(data []byte) {
 	// Base64 decode, than deserialize
 	decoded := make([]byte, base64.StdEncoding.DecodedLen(len(data)))
 
-	if size, err := base64.StdEncoding.Decode(decoded, data); err != nil {
-		spool.prod.Logger.Error("File read: ", err)
-	} else if msg, err := core.DeserializeMessage(decoded[:size]); err != nil {
-		spool.prod.Logger.Error("File read: ", err)
-	} else {
-		spool.prod.routeToOrigin(msg)
+	size, err := base64.StdEncoding.Decode(decoded, data)
+	if err != nil {
+		spool.prod.Logger.WithError(err).Error("failed deserialiying spooled message from base64")
+		return
+	}
+
+	msg, err := core.DeserializeMessage(decoded[:size])
+	if err != nil {
+		spool.prod.Logger.WithError(err).Error("failed deserializing spooled message from protobuf")
+		return
 	}
+
+	spool.prod.routeToOrigin(msg)
 }
 
 func (spool *spoolFile) waitForReader() {
@@ -185,13 +191,17 @@ func (spool *spoolFile) waitForReader() {
 
 func (spool *spoolFile) read() {
 	spool.prod.AddWorker()
-	spool.readWorker.Add(1)
 	defer spool.prod.WorkerDone()
+
+	spool.readWorker.Add(1)
 	defer spool.readWorker.Done()
 
+nextFile:
 	for !spool.prod.IsStopping() {
 		minSuffix, _ := spool.getFileNumbering()
 
+		// Wait for first spooling file to be rolled.
+
 		spoolFileName := fmt.Sprintf(spoolFileFormatString, spool.basePath, minSuffix)
 		if minSuffix == 0 || minSuffix > maxSpoolFileNumber || (spool.file != nil && spool.file.Name() == spoolFileName) {
 			if minSuffix > maxSpoolFileNumber {
@@ -201,6 +211,7 @@ func (spool *spoolFile) read() {
 			}
 
 			spool.prod.WorkerDone() // to avoid being stuck during shutdown
+			spool.readWorker.Done()
 
 			retry := time.After(spool.prod.maxFileAge / 2)
 			select {
@@ -209,47 +220,69 @@ func (spool *spoolFile) read() {
 			}
 
 			spool.prod.AddWorker() // worker done is always called at exit
-			continue               // ### continue, try again ###
+			spool.readWorker.Add(1)
+			continue // ### continue, try again ###
+		}
+
+		// Only spool back if target is not busy
+		for !spool.prod.IsStopping() && spool.source != nil && spool.source.IsBlocked() {
+			time.Sleep(time.Millisecond * 100)
+		}
+
+		if spool.prod.IsStopping() {
+			return // ### return, stop requested ###
 		}
 
 		file, err := os.OpenFile(spoolFileName, os.O_RDONLY, 0600)
 		if err != nil {
 			spool.prod.Logger.Error("Read open error ", err)
+			// TODO: how to mitigate a possible endless loop?
 			continue // ### continue, try again ###
 		}
 
 		spool.prod.Logger.Debug("Opened ", spoolFileName, " for reading")
 		spool.reader.Reset(0)
-		readFailed := false
 
-		for spool.prod.IsStopping() {
-			// Only spool back if target is not busy
-			if spool.source != nil && spool.source.IsBlocked() {
-				time.Sleep(time.Millisecond * 100)
-				continue // ### continue, busy source ###
+		var data []byte
+		more := true
+
+		for more {
+			// read line by line
+			data, more, err = spool.reader.ReadOne(file)
+
+			// Any error marks the file as failed but does not delete it, so messages can eventually be recovered
+			if err != io.EOF && err != nil {
+				spool.prod.Logger.WithError(err).Error("failed to read spooling file ", spoolFileName)
+				if err := file.Close(); err != nil {
+					spool.prod.Logger.WithError(err).Error("failed to close spooling file ", spoolFileName)
+				}
+
+				spool.prod.Logger.Debug("renaming ", spoolFileName)
+				os.Rename(spoolFileName, spoolFileName+".failed")
+				continue nextFile // ### continue, read error or EOF ###
 			}
 
-			// Any error cancels the loop
-			if err := spool.reader.ReadAll(file, spool.decode); err != nil {
-				if err != io.EOF {
-					readFailed = true
-					spool.prod.Logger.Error("Read error: ", err)
+			// len(data) == 0 may be an incomplete message, i.e. we need to read once more to get the rest
+			if len(data) > 0 {
+				spool.decode(data)
+			}
+
+			// Check if we're forced to stop reading
+			if spool.prod.IsStopping() {
+				spool.prod.Logger.Warning("file ", spoolFileName, " will be read again after restart")
+				if err := file.Close(); err != nil {
+					spool.prod.Logger.WithError(err).Error("failed to close spooling file ", spoolFileName)
 				}
-				break // ### break, read error or EOF ###
+				return // ### return, stop requested ###
 			}
 		}
 
-		// Close and remove file
-		spool.prod.Logger.Debug("Removing ", spoolFileName)
-		file.Close()
-		if readFailed {
-			// Rename file for future processing
-			spool.prod.Logger.Debug("Renaming ", spoolFileName)
-			os.Rename(spoolFileName, spoolFileName+".failed")
-		} else {
-			// Delete file
-			spool.prod.Logger.Debug("Removing ", spoolFileName)
-			os.Remove(spoolFileName)
+		// Close and remove file as it has been completely read
+		if err := file.Close(); err != nil {
+			spool.prod.Logger.WithError(err).Error("failed to close spooling file ", spoolFileName)
 		}
+
+		spool.prod.Logger.Debug("removing ", spoolFileName)
+		os.Remove(spoolFileName)
 	}
 }

producer/spooling.go

@@ -15,13 +15,14 @@
 package producer
 
 import (
+	"encoding/base64"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sync"
 	"time"
 
-	"github.com/rcrowley/go-metrics"
+	metrics "github.com/rcrowley/go-metrics"
 
 	"github.com/trivago/gollum/core"
 	"github.com/trivago/gollum/core/components"
@@ -107,7 +108,6 @@ type Spooling struct {
 	batchTimeout          time.Duration           `config:"Batch/TimeoutSec" default:"5" metric:"sec"`
 	readDelay             time.Duration
 	spoolCheck            *time.Timer
-	serialze              core.Formatter
 	metricsRegistry       metrics.Registry
 }
 
@@ -122,14 +122,6 @@ func (prod *Spooling) Configure(conf core.PluginConfigReader) {
 	prod.SetRollCallback(prod.onRoll)
 	prod.metricsRegistry = core.NewMetricsRegistryForPlugin(prod)
 
-	serializePlugin, err := core.NewPluginWithConfig(core.NewPluginConfig("", "format.Serialize"))
-	conf.Errors.Push(err)
-	if serializeFormatter, isFormatter := serializePlugin.(core.Formatter); isFormatter {
-		prod.serialze = serializeFormatter
-	} else {
-		conf.Errors.Pushf("Failed to instantiate format.Serialize")
-	}
-
 	prod.outfileGuard = new(sync.RWMutex)
 	prod.outfile = make(map[core.MessageStreamID]*spoolFile)
 
@@ -146,13 +138,6 @@ func (prod *Spooling) Configure(conf core.PluginConfigReader) {
 	prod.rotation.SizeByte = prod.maxFileSize
 }
 
-// Modulate enforces the serialize formatter at the end of the modulation chain
-func (prod *Spooling) Modulate(msg *core.Message) core.ModulateResult {
-	result := prod.BufferedProducer.Modulate(msg)
-	prod.serialze.ApplyFormatter(msg) // Ignore result
-	return result
-}
-
 // TryFallback reverts the message stream before dropping
 func (prod *Spooling) TryFallback(msg *core.Message) {
 	if prod.revertOnDrop {
@@ -161,6 +146,21 @@ func (prod *Spooling) TryFallback(msg *core.Message) {
 	prod.BufferedProducer.TryFallback(msg)
 }
 
+func (prod *Spooling) encode(msg *core.Message) ([]byte, error) {
+	serialized, err := msg.Serialize()
+	if err != nil {
+		return nil, err
+	}
+
+	encodedLen := base64.StdEncoding.EncodedLen(len(serialized)) + 1
+	encoded := make([]byte, encodedLen)
+
+	base64.StdEncoding.Encode(encoded, serialized)
+	encoded[encodedLen-1] = '\n'
+
+	return encoded, nil
+}
+
 func (prod *Spooling) writeToFile(msg *core.Message) {
 	// Get the correct file state for this stream
 	streamID := msg.GetPrevStreamID()
@@ -193,6 +193,16 @@ func (prod *Spooling) writeToFile(msg *core.Message) {
 		return // ### return, could not spool to disk ###
 	}
 
+	// Convert to expected payload format
+	encodedPayload, err := prod.encode(msg)
+	if err != nil {
+		prod.Logger.WithError(err).Error("Failed to encode message for spooling")
+		prod.TryFallback(msg)
+		return // ### return, could not spool to disk ###
+	}
+
+	msg.StorePayload(encodedPayload)
+
 	// Append to buffer
 	spool.batch.AppendOrFlush(msg, spool.flush, prod.IsActiveOrStopping, prod.TryFallback)
 	spool.countWrite()
@@ -248,11 +258,10 @@ func (prod *Spooling) routeToOrigin(msg *core.Message) {
 }
 
 func (prod *Spooling) waitForReader() {
-	prod.outfileGuard.Lock()
-	defer prod.outfileGuard.Unlock()
+	prod.outfileGuard.RLock()
+	defer prod.outfileGuard.RUnlock()
 
-	outfiles := prod.outfile
-	for _, spool := range outfiles {
+	for _, spool := range prod.outfile {
 		spool.waitForReader()
 	}
 }
@@ -266,8 +275,9 @@ func (prod *Spooling) close() {
 	}
 	prod.DefaultClose()
 
-	prod.outfileGuard.Lock()
-	defer prod.outfileGuard.Unlock()
+	prod.outfileGuard.RLock()
+	defer prod.outfileGuard.RUnlock()
+
 	for _, spool := range prod.outfile {
 		spool.close()
 	}