oci: make ExecSync handle split std{out,err}

Now that conmon splits std{out,err} for !terminal containers, ExecSync
can parse that output to return the correct std{out,err} split to the
kubelet. Invalid log lines are ignored but complained about.

Signed-off-by: Aleksa Sarai <asarai@suse.de>

conmon/conmon.c

@@ -533,24 +533,9 @@ int main(int argc, char *argv[])
 				if (num_read <= 0)
 					goto out;
 
-				if (exec) {
+				if (write_k8s_log(logfd, pipe, buf, num_read) < 0) {
-					/*
+					nwarn("write_k8s_log failed");
-					 * If we're in ExecSync we don't output the k8s log
+					goto out;
-					 * format. TODO(cyphar): This code really should be
-					 * rewritten so that we have a single conmon per
-					 * container and the conmon is logging the main
-					 * container process as a separate piece of logic to
-					 * the streaming to Exec[Sync] clients.
-					 */
-					if (write(logfd, buf, num_read) < 0) {
-						nwarn("write failed");
-						goto out;
-					}
-				} else {
-					if (write_k8s_log(logfd, pipe, buf, num_read) < 0) {
-						nwarn("write_k8s_log failed");
-						goto out;
-					}
 				}
 			} else if (evlist[i].events & (EPOLLHUP | EPOLLERR)) {
 				printf("closing fd %d\n", evlist[i].data.fd);

oci/oci.go

@@ -236,6 +236,42 @@ func prepareExec() (pidFile, parentPipe, childPipe *os.File, err error) {
 	return
 }
 
+func parseLog(log []byte) (stdout, stderr []byte) {
+	// Split the log on newlines, which is what separates entries.
+	lines := bytes.SplitAfter(log, []byte{'\n'})
+	for _, line := range lines {
+		// Ignore empty lines.
+		if len(line) == 0 {
+			continue
+		}
+
+		// The format of log lines is "DATE pipe REST".
+		parts := bytes.SplitN(line, []byte{' '}, 3)
+		if len(parts) < 3 {
+			// 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))
+			continue
+		}
+
+		pipe := string(parts[1])
+		content := parts[2]
+
+		switch pipe {
+		case "stdout":
+			stdout = append(stdout, content...)
+		case "stderr":
+			stderr = append(stderr, content...)
+		default:
+			// Complain about unknown pipes.
+			logrus.Warnf("hit invalid log format [unknown pipe %s]: %q", pipe, string(line))
+			continue
+		}
+	}
+
+	return stdout, stderr
+}
+
 // ExecSync execs a command in a container and returns it's stdout, stderr and return code.
 func (r *Runtime) ExecSync(c *Container, command []string, timeout int64) (resp *ExecSyncResponse, err error) {
 	pidFile, parentPipe, childPipe, err := prepareExec()
@@ -386,7 +422,7 @@ func (r *Runtime) ExecSync(c *Container, command []string, timeout int64) (resp
 	// XXX: Currently runC dups the same console over both stdout and stderr,
 	//      so we can't differentiate between the two.
 
-	outputBytes, err := ioutil.ReadFile(logPath)
+	logBytes, err := ioutil.ReadFile(logPath)
 	if err != nil {
 		return nil, ExecSyncError{
 			Stdout:   stdoutBuf,
@@ -396,9 +432,11 @@ func (r *Runtime) ExecSync(c *Container, command []string, timeout int64) (resp
 		}
 	}
 
+	// We have to parse the log output into {stdout, stderr} buffers.
+	stdoutBytes, stderrBytes := parseLog(logBytes)
 	return &ExecSyncResponse{
-		Stdout:   outputBytes,
+		Stdout:   stdoutBytes,
-		Stderr:   nil,
+		Stderr:   stderrBytes,
 		ExitCode: ec.ExitCode,
 	}, nil
 }