Merge pull request #11 from mrunalp/conmon_integration

Conmon integration

conmon/conmon.c

@@ -76,6 +76,9 @@ int main(int argc, char *argv[])
 	struct termios t;
 	struct epoll_event ev;
 	struct epoll_event evlist[MAX_EVENTS];
+	int child_pipe = -1;
+	char *sync_pipe, *endptr;
+	int len;
 
 	while ((opt = getopt(argc, argv, "tc:")) != -1) {
 		switch(opt) {
@@ -101,6 +104,14 @@ int main(int argc, char *argv[])
 		nexit("Container ID not passed");
 	}
 
+	sync_pipe = getenv("_OCI_SYNCPIPE");
+	if (sync_pipe) {
+		errno = 0;
+		child_pipe = strtol(sync_pipe, &endptr, 10);
+		if (errno != 0 || *endptr != '\0')
+			pexit("unable to parse _OCI_SYNCPIPE");
+	}
+
 	/*
 	 * Set self as subreaper so we can wait for container process
 	 * and return its exit code.
@@ -155,6 +166,14 @@ int main(int argc, char *argv[])
 	cpid = atoi(contents);
 	printf("container PID: %d\n", cpid);
 
+	/* Send the container pid back to parent */
+	if (child_pipe > 0) {
+		len = snprintf(buf, BUF_SIZE, "{\"pid\": %d}\n", cpid);
+		if (len < 0 || write(child_pipe, buf, len) != len) {
+			pexit("unable to send container pid to parent");
+		}
+	}
+
 	if (terminal) {
 		/* Save exiting termios settings */
 		if (tcgetattr(STDIN_FILENO, &tty_orig) == -1)

oci/oci.go

@@ -9,8 +9,10 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"syscall"
 	"time"
 
+	"github.com/Sirupsen/logrus"
 	"github.com/kubernetes-incubator/ocid/utils"
 	"github.com/opencontainers/runtime-spec/specs-go"
 )
@@ -32,6 +34,11 @@ type Runtime struct {
 	containerDir string
 }
 
+// syncInfo is used to return data from monitor process to daemon
+type syncInfo struct {
+	Pid int `"json:pid"`
+}
+
 // Name returns the name of the OCI Runtime
 func (r *Runtime) Name() string {
 	return r.name
@@ -69,7 +76,46 @@ func getOCIVersion(name string, args ...string) (string, error) {
 
 // CreateContainer creates a container.
 func (r *Runtime) CreateContainer(c *Container) error {
-	return utils.ExecCmdWithStdStreams(os.Stdin, os.Stdout, os.Stderr, r.path, "--systemd-cgroup", "create", "--bundle", c.bundlePath, c.name)
+	parentPipe, childPipe, err := newPipe()
+	if err != nil {
+		return fmt.Errorf("error creating socket pair: %v", err)
+	}
+	defer parentPipe.Close()
+
+	args := []string{"-c", c.name}
+	if c.terminal {
+		args = append(args, "-t")
+	}
+
+	cmd := exec.Command("conmon", args...)
+	cmd.Dir = c.bundlePath
+	cmd.SysProcAttr = &syscall.SysProcAttr{
+		Setpgid: true,
+	}
+	cmd.Stdin = os.Stdin
+	cmd.Stdout = os.Stdout
+	cmd.Stderr = os.Stderr
+	cmd.ExtraFiles = append(cmd.ExtraFiles, childPipe)
+	// 0, 1 and 2 are stdin, stdout and stderr
+	cmd.Env = append(cmd.Env, fmt.Sprintf("_OCI_SYNCPIPE=%d", 3))
+
+	err = cmd.Start()
+	if err != nil {
+		childPipe.Close()
+		return err
+	}
+
+	// We don't need childPipe on the parent side
+	childPipe.Close()
+
+	// Wait to get container pid from conmon
+	// TODO(mrunalp): Add a timeout here
+	var si *syncInfo
+	if err := json.NewDecoder(parentPipe).Decode(&si); err != nil {
+		return fmt.Errorf("reading pid from init pipe: %v", err)
+	}
+	logrus.Infof("Received container pid: %v", si.Pid)
+	return nil
 }
 
 // StartContainer starts a container.
@@ -121,6 +167,7 @@ type Container struct {
 	logPath    string
 	labels     map[string]string
 	sandbox    string
+	terminal   bool
 	state      *ContainerState
 	stateLock  sync.Mutex
 }
@@ -133,13 +180,14 @@ type ContainerState struct {
 }
 
 // NewContainer creates a container object.
-func NewContainer(name string, bundlePath string, logPath string, labels map[string]string, sandbox string) (*Container, error) {
+func NewContainer(name string, bundlePath string, logPath string, labels map[string]string, sandbox string, terminal bool) (*Container, error) {
 	c := &Container{
 		name:       name,
 		bundlePath: bundlePath,
 		logPath:    logPath,
 		labels:     labels,
 		sandbox:    sandbox,
+		terminal:   terminal,
 	}
 	return c, nil
 }
@@ -176,3 +224,12 @@ func (c *Container) NetNsPath() (string, error) {
 	}
 	return fmt.Sprintf("/proc/%d/ns/net", c.state.Pid), nil
 }
+
+// newPipe creates a unix socket pair for communication
+func newPipe() (parent *os.File, child *os.File, err error) {
+	fds, err := syscall.Socketpair(syscall.AF_LOCAL, syscall.SOCK_STREAM|syscall.SOCK_CLOEXEC, 0)
+	if err != nil {
+		return nil, nil, err
+	}
+	return os.NewFile(uintptr(fds[1]), "parent"), os.NewFile(uintptr(fds[0]), "child"), nil
+}

server/runtime.go

@@ -139,7 +139,7 @@ func (s *Server) CreatePodSandbox(ctx context.Context, req *pb.CreatePodSandboxR
 	}
 
 	containerName := name + "-infra"
-	container, err := oci.NewContainer(containerName, podSandboxDir, podSandboxDir, labels, name)
+	container, err := oci.NewContainer(containerName, podSandboxDir, podSandboxDir, labels, name, false)
 	if err != nil {
 		return nil, err
 	}
@@ -540,7 +540,7 @@ func (s *Server) CreateContainer(ctx context.Context, req *pb.CreateContainerReq
 		return nil, err
 	}
 
-	container, err := oci.NewContainer(name, containerDir, logPath, labels, podSandboxId)
+	container, err := oci.NewContainer(name, containerDir, logPath, labels, podSandboxId, containerConfig.GetTty())
 	if err != nil {
 		return nil, err
 	}