Add support for log tags in newest CRI spec

CHANGELOG.asciidoc

@@ -60,6 +60,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 - Fix date format in Mongodb Ingest pipeline. {pull}7974[7974]
 - Mark the TCP and UDP input as GA. {pull}8125[8125]
 - Fixed a docker input error due to the offset update bug in partial log join.{pull}8177[8177]
+- Update CRI format to support partial/full tags. {pull}8265[8265]
 
 *Heartbeat*
 

filebeat/docs/inputs/input-docker.asciidoc

@@ -55,6 +55,12 @@ Enable partial messages joining. Docker `json-file` driver splits log lines larg
 end of line (`\n`) is present for common lines in the resulting file, while it's not the for the lines
 that have been split. `combine_partial` joins them back together when enabled. It is enabled by default.
 
+===== `cri.parse_flags`
+
+Enable CRI flags parsing from the log file. CRI uses flags to signal a partial line, enabling this will
+ensure partial lines are rejoined. It is disabled by default.
+
+
 The following input configures {beatname_uc} to read the `stdout` stream from
 all containers under the default Docker containers path:
 

filebeat/input/docker/config.go

@@ -31,12 +31,15 @@ type config struct {
 
 	// Partial configures the prospector to join partial lines
 	Partial bool `config:"combine_partials"`
+
+	// Enable CRI flags parsing (to be switched to default in 7.0)
+	CRIFlags bool `config:"cri.parse_flags"`
 }
 
 type containers struct {
 	IDs  []string `config:"ids"`
 	Path string   `config:"path"`
 
-	// Stream can be all,stdout or stderr
+	// Stream can be all, stdout or stderr
 	Stream string `config:"stream"`
 }

filebeat/input/docker/input.go

@@ -86,6 +86,10 @@ func NewInput(
 		return nil, errors.Wrap(err, "update input config")
 	}
 
+	if err := cfg.SetBool("docker-json.cri_flags", -1, config.Partial); err != nil {
+		return nil, errors.Wrap(err, "update input config")
+	}
+
 	// Add stream to meta to ensure different state per stream
 	if config.Containers.Stream != "all" {
 		if context.Meta == nil {

filebeat/input/log/config.go

@@ -104,10 +104,9 @@ type config struct {
 
 	// Hidden on purpose, used by the docker input:
 	DockerJSON *struct {
-		Stream string `config:"stream"`
-
-		// TODO move this to true by default
-		Partial bool `config:"partial"`
+		Stream   string `config:"stream"`
+		Partial  bool   `config:"partial"`
+		CRIFlags bool   `config:"cri_flags"`
 	} `config:"docker-json"`
 }
 

filebeat/input/log/harvester.go

@@ -556,7 +556,7 @@ func (h *Harvester) newLogFileReader() (reader.Reader, error) {
 
 	if h.config.DockerJSON != nil {
 		// Docker json-file format, add custom parsing to the pipeline
-		r = readjson.New(r, h.config.DockerJSON.Stream, h.config.DockerJSON.Partial)
+		r = readjson.New(r, h.config.DockerJSON.Stream, h.config.DockerJSON.Partial, h.config.DockerJSON.CRIFlags)
 	}
 
 	if h.config.JSON != nil {

libbeat/reader/readjson/docker_json.go

@@ -37,76 +37,118 @@ type DockerJSONReader struct {
 
 	// join partial lines
 	partial bool
-}
 
-type dockerLog struct {
-	Timestamp string `json:"time"`
-	Log       string `json:"log"`
-	Stream    string `json:"stream"`
+	// parse CRI flags
+	criflags bool
 }
 
-type crioLog struct {
-	Timestamp time.Time
-	Stream    string
-	Log       []byte
+type logLine struct {
+	Partial   bool      `json:"-"`
+	Timestamp time.Time `json:"-"`
+	Time      string    `json:"time"`
+	Stream    string    `json:"stream"`
+	Log       string    `json:"log"`
 }
 
 // New creates a new reader renaming a field
-func New(r reader.Reader, stream string, partial bool) *DockerJSONReader {
+func New(r reader.Reader, stream string, partial bool, CRIFlags bool) *DockerJSONReader {
 	return &DockerJSONReader{
-		stream:  stream,
-		partial: partial,
-		reader:  r,
+		stream:   stream,
+		partial:  partial,
+		reader:   r,
+		criflags: CRIFlags,
 	}
 }
 
 // parseCRILog parses logs in CRI log format.
 // CRI log format example :
 // 2017-09-12T22:32:21.212861448Z stdout 2017-09-12 22:32:21.212 [INFO][88] table.go 710: Invalidating dataplane cache
-func parseCRILog(message reader.Message, msg *crioLog) (reader.Message, error) {
-	log := strings.SplitN(string(message.Content), " ", 3)
-	if len(log) < 3 {
-		return message, errors.New("invalid CRI log")
+func (p *DockerJSONReader) parseCRILog(message *reader.Message, msg *logLine) error {
+	split := 3
+	// read line tags if split is enabled:
+	if p.criflags {
+		split = 4
 	}
-	ts, err := time.Parse(time.RFC3339, log[0])
+
+	// current field
+	i := 0
+
+	// timestamp
+	log := strings.SplitN(string(message.Content), " ", split)
+	if len(log) < split {
+		return errors.New("invalid CRI log format")
+	}
+	ts, err := time.Parse(time.RFC3339, log[i])
 	if err != nil {
-		return message, errors.Wrap(err, "parsing CRI timestamp")
+		return errors.Wrap(err, "parsing CRI timestamp")
+	}
+	message.Ts = ts
+	i++
+
+	// stream
+	msg.Stream = log[i]
+	i++
+
+	// tags
+	partial := false
+	if p.criflags {
+		// currently only P(artial) or F(ull) are available
+		tags := strings.Split(log[i], ":")
+		for _, tag := range tags {
+			if tag == "P" {
+				partial = true
+			}
+		}
+		i++
 	}
 
-	msg.Timestamp = ts
-	msg.Stream = log[1]
-	msg.Log = []byte(log[2])
+	msg.Partial = partial
 	message.AddFields(common.MapStr{
 		"stream": msg.Stream,
 	})
-	message.Content = msg.Log
-	message.Ts = ts
+	// Remove ending \n for partial messages
+	message.Content = []byte(log[i])
+	if partial {
+		message.Content = bytes.TrimRightFunc(message.Content, func(r rune) bool {
+			return r == '\n' || r == '\r'
+		})
+	}
 
-	return message, nil
+	return nil
 }
 
 // parseReaderLog parses logs in Docker JSON log format.
 // Docker JSON log format example:
 // {"log":"1:M 09 Nov 13:27:36.276 # User requested shutdown...\n","stream":"stdout"}
-func parseDockerJSONLog(message reader.Message, msg *dockerLog) (reader.Message, error) {
+func (p *DockerJSONReader) parseDockerJSONLog(message *reader.Message, msg *logLine) error {
 	dec := json.NewDecoder(bytes.NewReader(message.Content))
+
 	if err := dec.Decode(&msg); err != nil {
-		return message, errors.Wrap(err, "decoding docker JSON")
+		return errors.Wrap(err, "decoding docker JSON")
 	}
 
 	// Parse timestamp
-	ts, err := time.Parse(time.RFC3339, msg.Timestamp)
+	ts, err := time.Parse(time.RFC3339, msg.Time)
 	if err != nil {
-		return message, errors.Wrap(err, "parsing docker timestamp")
+		return errors.Wrap(err, "parsing docker timestamp")
 	}
 
 	message.AddFields(common.MapStr{
 		"stream": msg.Stream,
 	})
 	message.Content = []byte(msg.Log)
 	message.Ts = ts
+	msg.Partial = message.Content[len(message.Content)-1] != byte('\n')
+
+	return nil
+}
+
+func (p *DockerJSONReader) parseLine(message *reader.Message, msg *logLine) error {
+	if strings.HasPrefix(string(message.Content), "{") {
+		return p.parseDockerJSONLog(message, msg)
+	}
 
-	return message, nil
+	return p.parseCRILog(message, msg)
 }
 
 // Next returns the next line.
@@ -117,32 +159,27 @@ func (p *DockerJSONReader) Next() (reader.Message, error) {
 			return message, err
 		}
 
-		var dockerLine dockerLog
-		var crioLine crioLog
+		var logLine logLine
+		err = p.parseLine(&message, &logLine)
+		if err != nil {
+			return message, err
+		}
 
-		if strings.HasPrefix(string(message.Content), "{") {
-			message, err = parseDockerJSONLog(message, &dockerLine)
+		// Handle multiline messages, join partial lines
+		for p.partial && logLine.Partial {
+			next, err := p.reader.Next()
 			if err != nil {
 				return message, err
 			}
-			// Handle multiline messages, join lines that don't end with \n
-			for p.partial && message.Content[len(message.Content)-1] != byte('\n') {
-				next, err := p.reader.Next()
-				if err != nil {
-					return message, err
-				}
-				next, err = parseDockerJSONLog(next, &dockerLine)
-				if err != nil {
-					return message, err
-				}
-				message.Content = append(message.Content, next.Content...)
-				message.Bytes += next.Bytes
+			err = p.parseLine(&next, &logLine)
+			if err != nil {
+				return message, err
 			}
-		} else {
-			message, err = parseCRILog(message, &crioLine)
+			message.Content = append(message.Content, next.Content...)
+			message.Bytes += next.Bytes
 		}
 
-		if p.stream != "all" && p.stream != dockerLine.Stream && p.stream != crioLine.Stream {
+		if p.stream != "all" && p.stream != logLine.Stream {
 			continue
 		}