Do not drop unmarshallable service logs (#110)

Do not drop unmarshallable service logs. Wrap the unmarshallable value
with default values so it can be passed to appender.

Author: medzin

servicelog/scraper/json.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"encoding/json"
 	"io"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 
@@ -26,10 +27,9 @@ func (j *JSON) StartScraping(reader io.Reader) <-chan servicelog.Entry {
 		for scanner.Scan() {
 			logEntry := servicelog.Entry{}
 			if err := json.Unmarshal(scanner.Bytes(), &logEntry); err != nil {
-				log.WithError(err).Warn("Unable to unmarshal log entry - skipping line")
-				continue
-			}
-			if j.KeyFilter != nil {
+				log.WithError(err).Debug("Unable to unmarshal log entry - wrapping in default entry")
+				logEntry = j.wrapInDefault(scanner.Bytes())
+			} else if j.KeyFilter != nil {
 				for key := range logEntry {
 					if j.KeyFilter.Match([]byte(key)) {
 						delete(logEntry, key)
@@ -43,3 +43,12 @@ func (j *JSON) StartScraping(reader io.Reader) <-chan servicelog.Entry {
 
 	return logEntries
 }
+
+func (j *JSON) wrapInDefault(bytes []byte) servicelog.Entry {
+	return servicelog.Entry{
+		"time":   time.Now().Format(time.RFC3339Nano),
+		"level":  "INFO",
+		"logger": "invalid-format",
+		"msg":    string(bytes),
+	}
+}

servicelog/scraper/json_test.go

@@ -35,3 +35,17 @@ func TestIfFiltersKeysFromScrapedJSONs(t *testing.T) {
 	assert.Equal(t, "d", entry["c"])
 	assert.Len(t, entry, 1)
 }
+
+func TestIfWrapsInDefualtValuesInvalidLogEntries(t *testing.T) {
+	reader, writer := io.Pipe()
+	scraper := JSON{}
+
+	entries := scraper.StartScraping(reader)
+	go writer.Write([]byte("ERROR my invalid format\n"))
+
+	entry := <-entries
+
+	assert.Equal(t, "ERROR my invalid format", entry["msg"])
+	assert.Equal(t, "invalid-format", entry["logger"])
+	assert.Equal(t, "INFO", entry["level"])
+}