Add conmon-rs streaming server support

Signed-off-by: Sascha Grunert <sgrunert@redhat.com>

internal/oci/oci.go

@@ -78,6 +78,8 @@ type RuntimeImpl interface {
 	ReopenContainerLog(context.Context, *Container) error
 	CheckpointContainer(context.Context, *Container, *rspec.Spec, bool) error
 	RestoreContainer(context.Context, *Container, string, string) error
+	ServeExecContainer(context.Context, *Container, []string, bool, bool, bool, bool) (string, error)
+	ServeAttachContainer(context.Context, *Container, bool, bool, bool) (string, error)
 }
 
 // New creates a new Runtime with options provided
@@ -466,3 +468,21 @@ func (r *Runtime) RestoreContainer(ctx context.Context, c *Container, cgroupPare
 
 	return impl.RestoreContainer(ctx, c, cgroupParent, mountLabel)
 }
+
+func (r *Runtime) ServeExecContainer(ctx context.Context, c *Container, cmd []string, tty, stdin, stdout, stderr bool) (string, error) {
+	impl, err := r.RuntimeImpl(c)
+	if err != nil {
+		return "", err
+	}
+
+	return impl.ServeExecContainer(ctx, c, cmd, tty, stdin, stdout, stderr)
+}
+
+func (r *Runtime) ServeAttachContainer(ctx context.Context, c *Container, stdin, stdout, stderr bool) (string, error) {
+	impl, err := r.RuntimeImpl(c)
+	if err != nil {
+		return "", err
+	}
+
+	return impl.ServeAttachContainer(ctx, c, stdin, stdout, stderr)
+}

internal/oci/runtime_pod.go

@@ -171,6 +171,34 @@ func (r *runtimePod) RestoreContainer(
 	return r.oci.RestoreContainer(ctx, c, cgroupParent, mountLabel)
 }
 
+func (r *runtimePod) ServeExecContainer(ctx context.Context, c *Container, cmd []string, tty, stdin, stdout, stderr bool) (string, error) {
+	res, err := r.client.ServeExecContainer(ctx, &conmonClient.ServeExecContainerConfig{
+		ID:     c.ID(),
+		Cmd:    cmd,
+		Tty:    tty,
+		Stdin:  stdin,
+		Stdout: stdout,
+		Stderr: stderr,
+	})
+	if err != nil {
+		return "", fmt.Errorf("call ServeExecContainer RPC: %w", err)
+	}
+	return res.URL, nil
+}
+
+func (r *runtimePod) ServeAttachContainer(ctx context.Context, c *Container, stdin, stdout, stderr bool) (string, error) {
+	res, err := r.client.ServeAttachContainer(ctx, &conmonClient.ServeAttachContainerConfig{
+		ID:     c.ID(),
+		Stdin:  stdin,
+		Stdout: stdout,
+		Stderr: stderr,
+	})
+	if err != nil {
+		return "", fmt.Errorf("call ServeAttachContainer RPC: %w", err)
+	}
+	return res.URL, nil
+}
+
 func (r *runtimePod) ExecContainer(ctx context.Context, c *Container, cmd []string, stdin io.Reader, stdout, stderr io.WriteCloser, tty bool, resizeChan <-chan remotecommand.TerminalSize) error {
 	return r.oci.ExecContainer(ctx, c, cmd, stdin, stdout, stderr, tty, resizeChan)
 }

internal/oci/runtime_vm.go

@@ -1248,3 +1248,11 @@ func EncodeKataVirtualVolumeToBase64(ctx context.Context, volume *katavolume.Kat
 	option := base64.StdEncoding.EncodeToString(validKataVirtualVolumeJSON)
 	return option, nil
 }
+
+func (r *runtimeVM) ServeExecContainer(context.Context, *Container, []string, bool, bool, bool, bool) (string, error) {
+	return "", nil
+}
+
+func (r *runtimeVM) ServeAttachContainer(context.Context, *Container, bool, bool, bool) (string, error) {
+	return "", nil
+}

server/container_attach.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"errors"
+	"fmt"
 	"io"
 
 	"github.com/cri-o/cri-o/internal/log"
@@ -15,6 +16,20 @@ import (
 
 // Attach prepares a streaming endpoint to attach to a running container.
 func (s *Server) Attach(ctx context.Context, req *types.AttachRequest) (*types.AttachResponse, error) {
+	c, err := s.GetContainerFromShortID(ctx, req.ContainerId)
+	if err != nil {
+		return nil, fmt.Errorf("could not find container %q: %w", req.ContainerId, err)
+	}
+
+	url, err := s.Runtime().ServeAttachContainer(ctx, c, req.Stdin, req.Stdout, req.Stderr)
+	if err != nil {
+		return nil, fmt.Errorf("could not serve attach for container %q: %w", req.ContainerId, err)
+	}
+	if url != "" {
+		log.Infof(ctx, "Using attach URL from runtime: %v", url)
+		return &types.AttachResponse{Url: url}, nil
+	}
+
 	resp, err := s.getAttach(req)
 	if err != nil {
 		return nil, errors.New("unable to prepare attach endpoint")

server/container_exec.go

@@ -16,6 +16,20 @@ import (
 
 // Exec prepares a streaming endpoint to execute a command in the container.
 func (s *Server) Exec(ctx context.Context, req *types.ExecRequest) (*types.ExecResponse, error) {
+	c, err := s.GetContainerFromShortID(ctx, req.ContainerId)
+	if err != nil {
+		return nil, fmt.Errorf("could not find container %q: %w", req.ContainerId, err)
+	}
+
+	url, err := s.Runtime().ServeExecContainer(ctx, c, req.Cmd, req.Tty, req.Stdin, req.Stdout, req.Stderr)
+	if err != nil {
+		return nil, fmt.Errorf("could not serve exec for container %q: %w", req.ContainerId, err)
+	}
+	if url != "" {
+		log.Infof(ctx, "Using exec URL from runtime: %v", url)
+		return &types.ExecResponse{Url: url}, nil
+	}
+
 	resp, err := s.getExec(req)
 	if err != nil {
 		return nil, fmt.Errorf("unable to prepare exec endpoint: %w", err)

test/testdata/container_redis.json

@@ -1,71 +1,66 @@
 {
-	"metadata": {
-		"name": "podsandbox1-redis"
-	},
-	"image": {
-		"image": "quay.io/crio/fedora-crio-ci:latest"
-	},
-	"args": [
-		"docker-entrypoint.sh",
-		"redis-server"
-	],
-	"working_dir": "/data",
-	"envs": [
-		{
-			"key": "PATH",
-			"value": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
-		},
-		{
-			"key": "TERM",
-			"value": "xterm"
-		},
-		{
-			"key": "GLIBC_TUNABLES",
-			"value": "glibc.pthread.rseq=0"
-		},
-		{
-			"key": "REDIS_VERSION",
-			"value": "6.0.18"
-		},
-		{
-			"key": "REDIS_DOWNLOAD_URL",
-			"value": "http://download.redis.io/releases/redis-6.0.18.tar.gz"
-		},
-		{
-			"key": "REDIS_DOWNLOAD_SHA1",
-			"value": "d7b4f2a97fcab96727284092b0a4aa854af47d570803fa0e7a0345359743836e"
-		}
-	],
-	"labels": {
-		"tier": "backend"
-	},
-	"annotations": {
-		"pod": "podsandbox1"
-	},
-	"log_path": "",
-	"stdin": false,
-	"stdin_once": false,
-	"tty": false,
-	"linux": {
-		"resources": {
-			"memory_limit_in_bytes": 209715200,
-			"cpu_period": 10000,
-			"cpu_quota": 20000,
-			"cpu_shares": 512,
-			"oom_score_adj": 30,
-			"cpuset_cpus": "0",
-			"cpuset_mems": "0"
-		},
-		"security_context": {
-			"namespace_options": {
-				"pid": 1
-			},
-			"readonly_rootfs": false,
-			"capabilities": {
-				"add_capabilities": [
-					"sys_admin"
-				]
-			}
-		}
-	}
+  "metadata": {
+    "name": "podsandbox1-redis"
+  },
+  "image": {
+    "image": "quay.io/crio/fedora-crio-ci:latest"
+  },
+  "args": ["docker-entrypoint.sh", "redis-server"],
+  "working_dir": "/data",
+  "envs": [
+    {
+      "key": "PATH",
+      "value": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
+    },
+    {
+      "key": "TERM",
+      "value": "xterm"
+    },
+    {
+      "key": "GLIBC_TUNABLES",
+      "value": "glibc.pthread.rseq=0"
+    },
+    {
+      "key": "REDIS_VERSION",
+      "value": "6.0.18"
+    },
+    {
+      "key": "REDIS_DOWNLOAD_URL",
+      "value": "http://download.redis.io/releases/redis-6.0.18.tar.gz"
+    },
+    {
+      "key": "REDIS_DOWNLOAD_SHA1",
+      "value": "d7b4f2a97fcab96727284092b0a4aa854af47d570803fa0e7a0345359743836e"
+    }
+  ],
+  "labels": {
+    "tier": "backend"
+  },
+  "annotations": {
+    "pod": "podsandbox1"
+  },
+  "log_path": "",
+  "stdin": true,
+  "stdin_once": false,
+  "tty": true,
+  "linux": {
+    "resources": {
+      "memory_limit_in_bytes": 209715200,
+      "cpu_period": 10000,
+      "cpu_quota": 20000,
+      "cpu_shares": 512,
+      "oom_score_adj": 30,
+      "cpuset_cpus": "0",
+      "cpuset_mems": "0"
+    },
+    "security_context": {
+      "namespace_options": {
+        "pid": 1
+      },
+      "readonly_rootfs": false,
+      "capabilities": {
+        "add_capabilities": ["sys_admin"]
+      }
+    }
+  }
 }