conmon: Add support for partial/newline log tags

This is for ttps://github.com/kubernetes/kubernetes/pull/55922

Signed-off-by: Mrunal Patel <mrunalp@gmail.com>

conmon/conmon.c

@@ -296,7 +296,6 @@ const char *stdpipe_name(stdpipe_t pipe)
 static int write_k8s_log(int fd, stdpipe_t pipe, const char *buf, ssize_t buflen)
 {
 	char tsbuf[TSBUFLEN];
-	static stdpipe_t trailing_line = NO_PIPE;
 	writev_buffer_t bufv = {0};
 	static int64_t bytes_written = 0;
 	int64_t bytes_to_be_written = 0;
@@ -313,36 +312,23 @@ static int write_k8s_log(int fd, stdpipe_t pipe, const char *buf, ssize_t buflen
 	while (buflen > 0) {
 		const char *line_end = NULL;
 		ptrdiff_t line_len = 0;
-		bool insert_newline = FALSE;
-		bool insert_timestamp = FALSE;
+		bool partial = FALSE;
 
 		/* Find the end of the line, or alternatively the end of the buffer. */
 		line_end = memchr(buf, '\n', buflen);
-		if (line_end == NULL)
+		if (line_end == NULL) {
 			line_end = &buf[buflen-1];
+			partial = TRUE;
+		}
 		line_len = line_end - buf + 1;
 
-		bytes_to_be_written = line_len;
-		if (trailing_line != pipe) {
-			/*
-			 * Write the (timestamp, stream) tuple if there isn't any trailing
-			 * output from the previous line (or if there is trailing output but
-			 * the current buffer being printed is from a different pipe).
-			 */
-			insert_timestamp = TRUE;
-			bytes_to_be_written += (TSBUFLEN - 1);
-			/*
-			 * If there was a trailing line from a different pipe, prepend a
-			 * newline to split it properly. This technically breaks the flow
-			 * of the previous line (adding a newline in the log where there
-			 * wasn't one output) but without modifying the file in a
-			 * non-append-only way there's not much we can do.
-			 */
-			if (trailing_line != NO_PIPE) {
-				insert_newline = TRUE;
+		/* This is line_len bytes + TSBUFLEN - 1 + 2 (- 1 is for ignoring \0). */
+		bytes_to_be_written = line_len + TSBUFLEN + 1;
+
+		/* If partial, then we add a \n */
+		if (partial) {
 			bytes_to_be_written += 1;
 		}
-		}
 
 		/*
 		 * We re-open the log file if writing out the bytes will exceed the max
@@ -351,8 +337,6 @@ static int write_k8s_log(int fd, stdpipe_t pipe, const char *buf, ssize_t buflen
 		 */
 		if ((opt_log_size_max > 0) && (bytes_written + bytes_to_be_written) > opt_log_size_max) {
 			ninfo("Creating new log file");
-			insert_newline = FALSE;
-			insert_timestamp = TRUE;
 			bytes_written = 0;
 
 			/* Close the existing fd */
@@ -370,20 +354,23 @@ static int write_k8s_log(int fd, stdpipe_t pipe, const char *buf, ssize_t buflen
 			fd = log_fd;
 		}
 
-		/* Output a newline */
-		if (insert_newline) {
-			if (writev_buffer_append_segment(fd, &bufv, "\n", -1) < 0) {
-				nwarn("failed to write newline to log");
-				goto next;
-			}
-		}
-
-		/* Output a timestamp */
-		if (insert_timestamp) {
+		/* Output the timestamp */
 		if (writev_buffer_append_segment(fd, &bufv, tsbuf, -1) < 0) {
 			nwarn("failed to write (timestamp, stream) to log");
 			goto next;
 		}
+
+		/* Output log tag for partial or newline */
+		if (partial) {
+			if (writev_buffer_append_segment(fd, &bufv, "P ", -1) < 0) {
+				nwarn("failed to write partial log tag");
+				goto next;
+			}
+		} else {
+			if (writev_buffer_append_segment(fd, &bufv, "F ", -1) < 0) {
+				nwarn("failed to write end log tag");
+				goto next;
+			}
 		}
 
 		/* Output the actual contents. */
@@ -392,11 +379,15 @@ static int write_k8s_log(int fd, stdpipe_t pipe, const char *buf, ssize_t buflen
 			goto next;
 		}
 
+		/* Output a newline for partial */
+		if (partial) {
+			if (writev_buffer_append_segment(fd, &bufv, "\n", -1) < 0) {
+				nwarn("failed to write newline to log");
+				goto next;
+			}
+		}
+
 		bytes_written += bytes_to_be_written;
-
-		/* If we did not output a full line, then we are a trailing_line. */
-		trailing_line = (*line_end == '\n') ? NO_PIPE : pipe;
-
 next:
 		/* Update the head of the buffer remaining to output. */
 		buf += line_len;

contrib/test/integration/build/cri-tools.yml

@@ -4,7 +4,6 @@
   git:
     repo: "https://github.com/kubernetes-incubator/cri-tools.git"
     dest: "{{ ansible_env.GOPATH }}/src/github.com/kubernetes-incubator/cri-tools"
-    version: "b42fc3f364dd48f649d55926c34492beeb9b2e99"
     version: "{{ cri_tools_git_version }}"
     force: "{{ force_clone | default(False) | bool}}"
 

contrib/test/integration/build/kubernetes.yml

@@ -2,7 +2,7 @@
 
 - name: clone kubernetes source repo
   git:
-    repo: "https://github.com/runcom/kubernetes.git"
+    repo: "https://github.com/{{ k8s_github_fork }}/kubernetes.git"
     dest: "{{ ansible_env.GOPATH }}/src/k8s.io/kubernetes"
     # based on kube v1.9.0-alpha.2, update as needed
     version: "{{ k8s_git_version }}"

contrib/test/integration/main.yml

@@ -25,6 +25,7 @@
       include: "build/kubernetes.yml"
       vars:
         k8s_git_version: "cri-o-node-e2e-patched-logs"
+        k8s_github_fork: "runcom"
         crio_socket: "/var/run/crio.sock"
 
     - name: clone build and install runc
@@ -55,7 +56,7 @@
       include: "build/cri-tools.yml"
       vars:
         force_clone: True
-        cri_tools_git_version: "d9476a4d8632da6c28bf272018a0b9e059c3542f"
+        cri_tools_git_version: "a9e38a4a000bc1a4052fb33de1c967b8cfe9ad40"
     - name: run cri-o integration tests
       include: test.yml
 
@@ -74,14 +75,15 @@
       include: "build/cri-tools.yml"
       vars:
           force_clone: True
-          cri_tools_git_version: "d9476a4d8632da6c28bf272018a0b9e059c3542f"
+          cri_tools_git_version: "a9e38a4a000bc1a4052fb33de1c967b8cfe9ad40"
     - name: run critest validation and benchmarks
       include: critest.yml
     - name: clone build and install kubernetes
       include: "build/kubernetes.yml"
       vars:
           force_clone: True
-          k8s_git_version: "cri-o-patched-1.9"
+          k8s_git_version: "release-1.9"
+          k8s_github_fork: "kubernetes"
           crio_socket: "/var/run/crio/crio.sock"
     - name: run k8s e2e tests
       include: e2e.yml

oci/oci.go

@@ -359,9 +359,9 @@ func parseLog(log []byte) (stdout, stderr []byte) {
 			continue
 		}
 
-		// The format of log lines is "DATE pipe REST".
-		parts := bytes.SplitN(line, []byte{' '}, 3)
-		if len(parts) < 3 {
+		// The format of log lines is "DATE pipe LogTag REST".
+		parts := bytes.SplitN(line, []byte{' '}, 4)
+		if len(parts) < 4 {
 			// Ignore the line if it's formatted incorrectly, but complain
 			// about it so it can be debugged.
 			logrus.Warnf("hit invalid log format: %q", string(line))
@@ -369,7 +369,15 @@ func parseLog(log []byte) (stdout, stderr []byte) {
 		}
 
 		pipe := string(parts[1])
-		content := parts[2]
+		content := parts[3]
+
+		linetype := string(parts[2])
+		if linetype == "P" {
+			contentLen := len(content)
+			if content[contentLen-1] == '\n' {
+				content = content[:contentLen-1]
+			}
+		}
 
 		switch pipe {
 		case "stdout":

test/ctr.bats

@@ -191,8 +191,8 @@ function teardown() {
 	logpath="$DEFAULT_LOG_PATH/$pod_id/$ctr_id.log"
 	[ -f "$logpath" ]
 	echo "$logpath :: $(cat "$logpath")"
-	grep -E "^[^\n]+ stdout here is some output$" "$logpath"
-	grep -E "^[^\n]+ stderr and some from stderr$" "$logpath"
+	grep -E "^[^\n]+ stdout F here is some output$" "$logpath"
+	grep -E "^[^\n]+ stderr F and some from stderr$" "$logpath"
 
 	run crictl stops "$pod_id"
 	echo "$output"
@@ -238,7 +238,7 @@ function teardown() {
 	logpath="$DEFAULT_LOG_PATH/$pod_id/$ctr_id.log"
 	[ -f "$logpath" ]
 	echo "$logpath :: $(cat "$logpath")"
-	grep --binary -P "^[^\n]+ stdout here is some output\x0d$" "$logpath"
+	grep --binary -P "^[^\n]+ stdout F here is some output\x0d$" "$logpath"
 
 	run crictl stops "$pod_id"
 	echo "$output"