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)