Fix cio.NewAttach

cio.NewAttach didn't work with with firecracker-containerd.
The operation is doing the following;

- Calls State() to find FIFO files that are connected to the task
- Passes the FIFO files to the client.
- The client can open the FIFO files to interact with the task's
  stdio streams.

However firecracker-containerd was proxying the request to in-VM agent
as is. As a result the FIFO files on the response are pointing
the files inside the VM, which the client couldn't access.

Signed-off-by: Kazuyoshi Kato <katokazu@amazon.com>

Author: kzys

runtime/service.go

@@ -32,6 +32,7 @@ import (
 	// secure randomness
 	"math/rand" // #nosec
 
+	"github.com/containerd/containerd/cio"
 	"github.com/containerd/containerd/errdefs"
 	"github.com/containerd/containerd/events/exchange"
 	"github.com/containerd/containerd/log"
@@ -87,6 +88,10 @@ const (
 
 	// StopEventName is the topic published to when a VM stops
 	StopEventName = "/firecracker-vm/stop"
+
+	// taskExecID is a special exec ID that is pointing its task itself.
+	// While the constant is defined here, the convention is coming from containerd.
+	taskExecID = ""
 )
 
 var (
@@ -145,6 +150,10 @@ type service struct {
 	machineConfig    *firecracker.Config
 	vsockIOPortCount uint32
 	vsockPortMu      sync.Mutex
+
+	// fifos have stdio FIFOs containerd passed to the shim. The key is [taskID][execID].
+	fifos   map[string]map[string]cio.Config
+	fifosMu sync.Mutex
 }
 
 func shimOpts(shimCtx context.Context) (*shim.Opts, error) {
@@ -209,6 +218,7 @@ func NewService(shimCtx context.Context, id string, remotePublisher shim.Publish
 
 		vmReady: make(chan struct{}),
 		jailer:  newNoopJailer(shimCtx, logger, shimDir),
+		fifos:   make(map[string]map[string]cio.Config),
 	}
 
 	s.startEventForwarders(remotePublisher)
@@ -884,6 +894,39 @@ func (s *service) newIOProxy(logger *logrus.Entry, stdin, stdout, stderr string,
 	return ioConnectorSet, nil
 }
 
+func (s *service) addFIFOs(taskID, execID string, config cio.Config) error {
+	s.fifosMu.Lock()
+	defer s.fifosMu.Unlock()
+
+	_, exists := s.fifos[taskID]
+	if !exists {
+		s.fifos[taskID] = make(map[string]cio.Config)
+	}
+
+	value, exists := s.fifos[taskID][execID]
+	if exists {
+		return fmt.Errorf("failed to add FIFO files for task %q (exec=%q). There was %+v already", taskID, execID, value)
+	}
+	s.fifos[taskID][execID] = config
+	return nil
+}
+
+func (s *service) deleteFIFOs(taskID, execID string) error {
+	s.fifosMu.Lock()
+	defer s.fifosMu.Unlock()
+
+	_, exists := s.fifos[taskID][execID]
+	if !exists {
+		return fmt.Errorf("task %q (exec=%q) doesn't have corresponding FIFOs to delete", taskID, execID)
+	}
+	delete(s.fifos[taskID], execID)
+
+	if execID == taskExecID {
+		delete(s.fifos, taskID)
+	}
+	return nil
+}
+
 func (s *service) Create(requestCtx context.Context, request *taskAPI.CreateTaskRequest) (*taskAPI.CreateTaskResponse, error) {
 	logger := s.logger.WithField("task_id", request.ID)
 	defer logPanicAndDie(logger)
@@ -975,6 +1018,15 @@ func (s *service) Create(requestCtx context.Context, request *taskAPI.CreateTask
 		return nil, err
 	}
 
+	err = s.addFIFOs(request.ID, taskExecID, cio.Config{
+		Stdin:  request.Stdin,
+		Stdout: request.Stdout,
+		Stderr: request.Stderr,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	return resp, nil
 }
 
@@ -1001,6 +1053,11 @@ func (s *service) Delete(requestCtx context.Context, req *taskAPI.DeleteRequest)
 		return nil, err
 	}
 
+	err = s.deleteFIFOs(req.ID, req.ExecID)
+	if err != nil {
+		return nil, err
+	}
+
 	// Only delete a process as like runc when there is ExecID
 	// https://github.com/containerd/containerd/blob/f3e148b1ccf268450c87427b5dbb6187db3d22f1/runtime/v2/runc/container.go#L320
 	if req.ExecID != "" {
@@ -1062,6 +1119,16 @@ func (s *service) Exec(requestCtx context.Context, req *taskAPI.ExecProcessReque
 		return nil, err
 	}
 
+	err = s.addFIFOs(req.ID, req.ExecID, cio.Config{
+		Terminal: req.Terminal,
+		Stdin:    req.Stdin,
+		Stdout:   req.Stdout,
+		Stderr:   req.Stderr,
+	})
+	if err != nil {
+		return nil, err
+	}
+
 	return resp, nil
 }
 
@@ -1088,6 +1155,14 @@ func (s *service) State(requestCtx context.Context, req *taskAPI.StateRequest) (
 		return nil, err
 	}
 
+	// These fields are pointing files inside the VM.
+	// Replace them with the corresponding files on the host, so clients can access.
+	s.fifosMu.Lock()
+	defer s.fifosMu.Unlock()
+	resp.Stdin = s.fifos[req.ID][req.ExecID].Stdin
+	resp.Stdout = s.fifos[req.ID][req.ExecID].Stdout
+	resp.Stderr = s.fifos[req.ID][req.ExecID].Stderr
+
 	return resp, nil
 }
 

runtime/service_integ_test.go

@@ -2025,64 +2025,85 @@ func TestAttach_Isolated(t *testing.T) {
 	image, err := alpineImage(ctx, client, defaultSnapshotterName)
 	require.NoError(t, err, "failed to get alpine image")
 
-	vmID := testNameToVMID(t.Name())
+	testcases := []struct {
+		name     string
+		newIO    func(context.Context, string) (cio.IO, error)
+		expected string
+	}{
+		{
+			name: "attach",
+			newIO: func(ctx context.Context, id string) (cio.IO, error) {
+				set, err := cio.NewFIFOSetInDir("", id, false)
+				if err != nil {
+					return nil, err
+				}
 
-	c, err := client.NewContainer(ctx,
-		"container-"+vmID,
-		containerd.WithSnapshotter(defaultSnapshotterName),
-		containerd.WithNewSnapshot("snapshot-"+vmID, image),
-		containerd.WithNewSpec(oci.WithProcessArgs(
-			"/bin/cat",
-		)),
-	)
-	require.NoError(t, err)
+				return cio.NewDirectIO(ctx, set)
+			},
+			expected: "hello\n",
+		},
+		{
+			name: "null io",
 
-	// cio.NewCreator creates FIFOs and *the readers* implicitly. Use cio.NewDirectIO() to
-	// only create FIFOs.
-	fifos, err := cio.NewFIFOSetInDir("", t.Name(), false)
-	require.NoError(t, err)
+			// firecracker-containerd doesn't create IO Proxy objects in this case.
+			newIO: func(ctx context.Context, id string) (cio.IO, error) {
+				return cio.NullIO(id)
+			},
 
-	io, err := cio.NewDirectIO(ctx, fifos)
-	require.NoError(t, err)
+			// So, attaching new IOs doesn't work.
+			// While it looks odd, containerd's v2 shim has the same behavior.
+			expected: "",
+		},
+	}
 
-	task1, err := c.NewTask(ctx, func(id string) (cio.IO, error) {
-		// Pass FIFO files, but don't create the readers.
-		return io, nil
-	})
-	require.NoError(t, err, "failed to create task for container %s", c.ID())
-	defer task1.Delete(ctx)
+	for _, tc := range testcases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			name := testNameToVMID(t.Name())
 
-	err = task1.Start(ctx)
-	require.NoError(t, err, "failed to start task for container %s", c.ID())
+			c, err := client.NewContainer(ctx,
+				"container-"+name,
+				containerd.WithSnapshotter(defaultSnapshotterName),
+				containerd.WithNewSnapshot("snapshot-"+name, image),
+				containerd.WithNewSpec(oci.WithProcessArgs("/bin/cat")),
+			)
+			require.NoError(t, err)
 
-	// Directly reading/writing bytes to make sure "cat" is working.
-	input := "line1\n"
-	io.Stdin.Write([]byte(input))
+			io, err := tc.newIO(ctx, name)
+			require.NoError(t, err)
 
-	output := make([]byte, len(input))
-	io.Stdout.Read(output)
-	assert.Equal(t, input, string(output))
+			t1, err := c.NewTask(ctx, func(id string) (cio.IO, error) {
+				return io, nil
+			})
+			require.NoError(t, err)
 
-	c, err = client.LoadContainer(ctx, "container-"+vmID)
-	require.NoError(t, err)
+			ch, err := t1.Wait(ctx)
+			require.NoError(t, err)
 
-	var stderr, stdout bytes.Buffer
-	task2, err := c.Task(
-		ctx,
-		cio.NewAttach(cio.WithStreams(bytes.NewBufferString("line2\n"), &stdout, &stderr)),
-	)
-	require.NoError(t, err, "failed to load the task")
+			err = t1.Start(ctx)
+			require.NoError(t, err)
 
-	assert.Equal(t, task1.ID(), task2.ID(), "task1 and task2 are pointing the same task")
+			stdin := bytes.NewBufferString("hello\n")
+			var stdout bytes.Buffer
+			t2, err := c.Task(
+				ctx,
+				cio.NewAttach(cio.WithStreams(stdin, &stdout, nil)),
+			)
+			require.NoError(t, err)
+			assert.Equal(t, t1.ID(), t2.ID())
 
-	ch, err := task2.Wait(ctx)
-	require.NoError(t, err)
+			err = io.Close()
+			assert.NoError(t, err)
 
-	err = task2.CloseIO(ctx, containerd.WithStdinCloser)
-	require.NoError(t, err)
+			err = t2.CloseIO(ctx, containerd.WithStdinCloser)
+			assert.NoError(t, err)
+
+			<-ch
 
-	<-ch
+			_, err = t2.Delete(ctx)
+			require.NoError(t, err)
 
-	assert.Equal(t, "", stderr.String(), "stderr")
-	assert.Equal(t, "line2\n", stdout.String(), "stdout")
+			assert.Equal(t, tc.expected, stdout.String())
+		})
+	}
 }