Allow multiple json messages per TCP stream

protolog/loglistener.go

@@ -1,6 +1,7 @@
 package protolog
 
 import (
+	"bufio"
 	"fmt"
 	"net"
 	"strconv"
@@ -75,20 +76,7 @@ func (ll *LogListener) startTCP(proto string, address string) {
 			logp.Err("Error accepting log event: %v", err.Error())
 			continue
 		}
-
-		buffer := make([]byte, ll.config.MaxMsgSize)
-
-		length, err := conn.Read(buffer)
-		if err != nil {
-			e, ok := err.(net.Error)
-			if ok && e.Timeout() {
-				logp.Err("Timeout reading from socket: %v", err)
-				ll.logEntriesError <- true
-				return
-			}
-		}
-		go ll.processMessage(strings.TrimSpace(string(buffer[:length])))
-
+		go ll.processConnection(conn)
 	}
 }
 
@@ -144,86 +132,128 @@ func (ll *LogListener) Shutdown() {
 	close(ll.logEntriesRecieved)
 }
 
+func (ll *LogListener) processConnection(conn net.Conn) {
+	if ll.config.JsonMode {
+		scanner := bufio.NewScanner(conn)
+		scanner.Buffer(make([]byte, bufio.MaxScanTokenSize), ll.config.MaxMsgSize)
+		for scanner.Scan() {
+			log := common.MapStr{}
+			if err := ffjson.Unmarshal(scanner.Bytes(), &log); err != nil {
+				logp.Err("Could not parse JSON: %v", err)
+				ll.processErr(scanner.Text(), err, []string{"_protologbeat_json_parse_failure"})
+				continue
+			}
+			ll.processJson(log, scanner.Text())
+		}
+	} else {
+		buffer := make([]byte, ll.config.MaxMsgSize)
+		length, err := conn.Read(buffer)
+		if err != nil {
+			e, ok := err.(net.Error)
+			if ok && e.Timeout() {
+				logp.Err("Timeout reading from socket: %v", err)
+				ll.logEntriesError <- true
+				return
+			}
+		}
+
+		ll.processMessage(strings.TrimSpace(string(buffer[:length])))
+	}
+}
+
 func (ll *LogListener) processMessage(logData string) {
 
 	if logData == "" {
 		logp.Err("Event is empty")
 		return
 	}
-	event := common.MapStr{}
 
 	if ll.config.EnableSyslogFormatOnly {
 		msg, facility, severity, err := GetSyslogMsgDetails(logData)
 		if err == nil {
+			event := common.MapStr{}
 			event["facility"] = facility
 			event["severity"] = severity
 			event["message"] = msg
+			ll.sendEvent(event)
 		}
 	} else if ll.config.JsonMode {
-		if ll.config.MergeFieldsToRoot {
-			if err := ffjson.Unmarshal([]byte(logData), &event); err != nil {
-				logp.Err("Could not parse JSON: %v", err)
-				event["message"] = logData
-				event["tags"] = []string{"_protologbeat_json_parse_failure"}
-				goto PreSend
-			}
-		} else {
-			event = common.MapStr{}
-			nestedData := common.MapStr{}
-			if err := ffjson.Unmarshal([]byte(logData), &nestedData); err != nil {
-				logp.Err("Could not parse JSON: %v", err)
-				event["message"] = logData
-				event["tags"] = []string{"_protologbeat_json_parse_failure"}
-				goto PreSend
-			} else {
-				event["log"] = nestedData
-			}
+		log := common.MapStr{}
+		if err := ffjson.Unmarshal([]byte(logData), &log); err != nil {
+			logp.Err("Could not parse JSON: %v", err)
+			ll.processErr(logData, err, []string{"_protologbeat_json_parse_failure"})
+			return
 		}
+		ll.processJson(log, logData)
+	} else {
+		event := common.MapStr{}
+		event["message"] = logData
+		ll.sendEvent(event)
+	}
+}
 
-		schemaSet := false
-		hasType := false
-		if _, ok := event["type"]; ok {
-			hasType = true
-		}
+func (ll *LogListener) processJson(log common.MapStr, logData string) {
+	if len(log) == 0 {
+		logp.Err("Event is empty")
+		return
+	}
+
+	var event common.MapStr
+	if ll.config.MergeFieldsToRoot {
+		event = log
+	} else {
+		event["log"] = log
+	}
+
+	schemaSet := false
+	hasType := false
+	if _, ok := event["type"]; ok {
+		hasType = true
+	}
+
+	if hasType {
+		_, schemaSet = ll.jsonSchema[event["type"].(string)]
+	}
 
-		if hasType {
-			_, schemaSet = ll.jsonSchema[event["type"].(string)]
+	if ll.config.ValidateAllJSONTypes && !schemaSet {
+		if ll.config.Debug && hasType {
+			logp.Err("Log entry of type '%s' has no JSON schema set.", event["type"].(string))
+		} else if ll.config.Debug {
+			logp.Err("Log entry has no type.")
 		}
+		return
+	}
 
-		if ll.config.ValidateAllJSONTypes && !schemaSet {
-			if ll.config.Debug && hasType {
-				logp.Err("Log entry of type '%s' has no JSON schema set.", event["type"].(string))
-			} else if ll.config.Debug {
-				logp.Err("Log entry has no type.")
+	if ll.config.EnableJsonValidation && schemaSet {
+		result, err := gojsonschema.Validate(ll.jsonSchema[event["type"].(string)], gojsonschema.NewStringLoader(logData))
+		if err != nil {
+			if ll.config.Debug {
+				logp.Err("Error with JSON object: %s", err.Error())
 			}
 			return
 		}
 
-		if ll.config.EnableJsonValidation && schemaSet {
-
-			result, err := gojsonschema.Validate(ll.jsonSchema[event["type"].(string)], gojsonschema.NewStringLoader(logData))
-			if err != nil {
-				if ll.config.Debug {
-					logp.Err("Error with JSON object: %s", err.Error())
-				}
-				return
-			}
-
-			if !result.Valid() {
-				if ll.config.Debug {
-					logp.Err("Log entry does not match specified schema for type '%s'. (Note: ensure you have 'type' field (string) at the root level in your schema)", event["type"].(string))
-				}
-				return
+		if !result.Valid() {
+			if ll.config.Debug {
+				logp.Err("Log entry does not match specified schema for type '%s'. (Note: ensure you have 'type' field (string) at the root level in your schema)", event["type"].(string))
 			}
+			return
 		}
-
-	} else {
-		event["message"] = logData
 	}
 
-PreSend:
-	event["@timestamp"] = common.Time(time.Now())
+	ll.sendEvent(event)
+}
 
+func (ll *LogListener) processErr(logData string, err error, tags []string) {
+	event := common.MapStr{}
+	event["message"] = logData
+	event["error"] = fmt.Sprintf("%v", err.Error())
+	event["tags"] = tags
+	ll.sendEvent(event)
+}
+
+func (ll *LogListener) sendEvent(event common.MapStr) {
+	event["@timestamp"] = common.Time(time.Now())
 	ll.logEntriesRecieved <- event
 }