Move cio to pkg/cio

Signed-off-by: Derek McGowan <derek@mcg.dev>

pkg/cio/io.go

@@ -0,0 +1,388 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package cio
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+
+	"github.com/containerd/containerd/v2/defaults"
+)
+
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 32<<10)
+		return &buffer
+	},
+}
+
+// Config holds the IO configurations.
+type Config struct {
+	// Terminal is true if one has been allocated
+	Terminal bool
+	// Stdin path
+	Stdin string
+	// Stdout path
+	Stdout string
+	// Stderr path
+	Stderr string
+}
+
+// IO holds the io information for a task or process
+type IO interface {
+	// Config returns the IO configuration.
+	Config() Config
+	// Cancel aborts all current io operations.
+	Cancel()
+	// Wait blocks until all io copy operations have completed.
+	Wait()
+	// Close cleans up all open io resources. Cancel() is always called before
+	// Close()
+	Close() error
+}
+
+// Creator creates new IO sets for a task
+type Creator func(id string) (IO, error)
+
+// Attach allows callers to reattach to running tasks
+//
+// There should only be one reader for a task's IO set
+// because fifo's can only be read from one reader or the output
+// will be sent only to the first reads
+type Attach func(*FIFOSet) (IO, error)
+
+// FIFOSet is a set of file paths to FIFOs for a task's standard IO streams
+type FIFOSet struct {
+	Config
+	close func() error
+}
+
+// Close the FIFOSet
+func (f *FIFOSet) Close() error {
+	if f != nil && f.close != nil {
+		return f.close()
+	}
+	return nil
+}
+
+// NewFIFOSet returns a new FIFOSet from a Config and a close function
+func NewFIFOSet(config Config, close func() error) *FIFOSet {
+	return &FIFOSet{Config: config, close: close}
+}
+
+// Streams used to configure a Creator or Attach
+type Streams struct {
+	Stdin    io.Reader
+	Stdout   io.Writer
+	Stderr   io.Writer
+	Terminal bool
+	FIFODir  string
+}
+
+// Opt customize options for creating a Creator or Attach
+type Opt func(*Streams)
+
+// WithStdio sets stream options to the standard input/output streams
+func WithStdio(opt *Streams) {
+	WithStreams(os.Stdin, os.Stdout, os.Stderr)(opt)
+}
+
+// WithTerminal sets the terminal option
+func WithTerminal(opt *Streams) {
+	opt.Terminal = true
+}
+
+// WithStreams sets the stream options to the specified Reader and Writers
+func WithStreams(stdin io.Reader, stdout, stderr io.Writer) Opt {
+	return func(opt *Streams) {
+		opt.Stdin = stdin
+		opt.Stdout = stdout
+		opt.Stderr = stderr
+	}
+}
+
+// WithFIFODir sets the fifo directory.
+// e.g. "/run/containerd/fifo", "/run/users/1001/containerd/fifo"
+func WithFIFODir(dir string) Opt {
+	return func(opt *Streams) {
+		opt.FIFODir = dir
+	}
+}
+
+// NewCreator returns an IO creator from the options
+func NewCreator(opts ...Opt) Creator {
+	streams := &Streams{}
+	for _, opt := range opts {
+		opt(streams)
+	}
+	if streams.FIFODir == "" {
+		streams.FIFODir = defaults.DefaultFIFODir
+	}
+	return func(id string) (IO, error) {
+		fifos, err := NewFIFOSetInDir(streams.FIFODir, id, streams.Terminal)
+		if err != nil {
+			return nil, err
+		}
+		if streams.Stdin == nil {
+			fifos.Stdin = ""
+		}
+		if streams.Stdout == nil {
+			fifos.Stdout = ""
+		}
+		if streams.Stderr == nil {
+			fifos.Stderr = ""
+		}
+		return copyIO(fifos, streams)
+	}
+}
+
+// NewAttach attaches the existing io for a task to the provided io.Reader/Writers
+func NewAttach(opts ...Opt) Attach {
+	streams := &Streams{}
+	for _, opt := range opts {
+		opt(streams)
+	}
+	return func(fifos *FIFOSet) (IO, error) {
+		if fifos == nil {
+			return nil, fmt.Errorf("cannot attach, missing fifos")
+		}
+		if streams.Stdin == nil {
+			fifos.Stdin = ""
+		}
+		if streams.Stdout == nil {
+			fifos.Stdout = ""
+		}
+		if streams.Stderr == nil {
+			fifos.Stderr = ""
+		}
+		return copyIO(fifos, streams)
+	}
+}
+
+// NullIO redirects the container's IO into /dev/null
+func NullIO(_ string) (IO, error) {
+	return &cio{}, nil
+}
+
+// cio is a basic container IO implementation.
+type cio struct {
+	config  Config
+	wg      *sync.WaitGroup
+	closers []io.Closer
+	cancel  context.CancelFunc
+}
+
+func (c *cio) Config() Config {
+	return c.config
+}
+
+func (c *cio) Wait() {
+	if c.wg != nil {
+		c.wg.Wait()
+	}
+}
+
+func (c *cio) Close() error {
+	var lastErr error
+	for _, closer := range c.closers {
+		if closer == nil {
+			continue
+		}
+		if err := closer.Close(); err != nil {
+			lastErr = err
+		}
+	}
+	return lastErr
+}
+
+func (c *cio) Cancel() {
+	if c.cancel != nil {
+		c.cancel()
+	}
+}
+
+type pipes struct {
+	Stdin  io.WriteCloser
+	Stdout io.ReadCloser
+	Stderr io.ReadCloser
+}
+
+// DirectIO allows task IO to be handled externally by the caller
+type DirectIO struct {
+	pipes
+	cio
+}
+
+var (
+	_ IO = &DirectIO{}
+	_ IO = &logURI{}
+)
+
+// LogURI provides the raw logging URI
+func LogURI(uri *url.URL) Creator {
+	return func(_ string) (IO, error) {
+		return &logURI{
+			config: Config{
+				Stdout: uri.String(),
+				Stderr: uri.String(),
+			},
+		}, nil
+	}
+}
+
+// TerminalLogURI provides the raw logging URI
+// as well as sets the terminal option to true.
+func TerminalLogURI(uri *url.URL) Creator {
+	return func(_ string) (IO, error) {
+		return &logURI{
+			config: Config{
+				Stdout:   uri.String(),
+				Stderr:   uri.String(),
+				Terminal: true,
+			},
+		}, nil
+	}
+}
+
+// BinaryIO forwards container STDOUT|STDERR directly to a logging binary
+func BinaryIO(binary string, args map[string]string) Creator {
+	return func(_ string) (IO, error) {
+		uri, err := LogURIGenerator("binary", binary, args)
+		if err != nil {
+			return nil, err
+		}
+
+		res := uri.String()
+		return &logURI{
+			config: Config{
+				Stdout: res,
+				Stderr: res,
+			},
+		}, nil
+	}
+}
+
+// TerminalBinaryIO forwards container STDOUT|STDERR directly to a logging binary
+// It also sets the terminal option to true
+func TerminalBinaryIO(binary string, args map[string]string) Creator {
+	return func(_ string) (IO, error) {
+		uri, err := LogURIGenerator("binary", binary, args)
+		if err != nil {
+			return nil, err
+		}
+
+		res := uri.String()
+		return &logURI{
+			config: Config{
+				Stdout:   res,
+				Stderr:   res,
+				Terminal: true,
+			},
+		}, nil
+	}
+}
+
+// LogFile creates a file on disk that logs the task's STDOUT,STDERR.
+// If the log file already exists, the logs will be appended to the file.
+func LogFile(path string) Creator {
+	return func(_ string) (IO, error) {
+		uri, err := LogURIGenerator("file", path, nil)
+		if err != nil {
+			return nil, err
+		}
+
+		res := uri.String()
+		return &logURI{
+			config: Config{
+				Stdout: res,
+				Stderr: res,
+			},
+		}, nil
+	}
+}
+
+// LogURIGenerator is the helper to generate log uri with specific scheme.
+func LogURIGenerator(scheme string, path string, args map[string]string) (*url.URL, error) {
+	path = filepath.Clean(path)
+	if !filepath.IsAbs(path) {
+		return nil, fmt.Errorf("%q must be absolute", path)
+	}
+
+	// Without adding / here, C:\foo\bar.txt will become file://C:/foo/bar.txt
+	// which is invalid. The path must have three slashes.
+	//
+	// https://learn.microsoft.com/en-us/archive/blogs/ie/file-uris-in-windows
+	// > In the case of a local Windows file path, there is no hostname,
+	// > and thus another slash and the path immediately follow.
+	p := filepath.ToSlash(path)
+	if !strings.HasPrefix(path, "/") {
+		p = "/" + p
+	}
+	uri := &url.URL{Scheme: scheme, Path: p}
+
+	if len(args) == 0 {
+		return uri, nil
+	}
+
+	q := uri.Query()
+	for k, v := range args {
+		q.Set(k, v)
+	}
+	uri.RawQuery = q.Encode()
+	return uri, nil
+}
+
+type logURI struct {
+	config Config
+}
+
+func (l *logURI) Config() Config {
+	return l.config
+}
+
+func (l *logURI) Cancel() {
+
+}
+
+func (l *logURI) Wait() {
+
+}
+
+func (l *logURI) Close() error {
+	return nil
+}
+
+// Load the io for a container but do not attach
+//
+// Allows io to be loaded on the task for deletion without
+// starting copy routines
+func Load(set *FIFOSet) (IO, error) {
+	return &cio{
+		config:  set.Config,
+		closers: []io.Closer{set},
+	}, nil
+}
+
+func (p *pipes) closers() []io.Closer {
+	return []io.Closer{p.Stdin, p.Stdout, p.Stderr}
+}

pkg/cio/io_test.go

@@ -0,0 +1,107 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package cio
+
+import (
+	"fmt"
+	"net/url"
+	"runtime"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	prefix    string
+	urlPrefix string
+)
+
+func init() {
+	if runtime.GOOS == "windows" {
+		prefix = "C:"
+		urlPrefix = "/C:"
+	}
+}
+
+func TestBinaryIOArgs(t *testing.T) {
+	res, err := BinaryIO(prefix+"/file.bin", map[string]string{"id": "1"})("")
+	require.NoError(t, err)
+	expected := fmt.Sprintf("binary://%s/file.bin?id=1", urlPrefix)
+	assert.Equal(t, expected, res.Config().Stdout)
+	assert.Equal(t, expected, res.Config().Stderr)
+}
+
+func TestBinaryIOAbsolutePath(t *testing.T) {
+	res, err := BinaryIO(prefix+"/full/path/bin", nil)("!")
+	require.NoError(t, err)
+
+	// Test parse back
+	parsed, err := url.Parse(res.Config().Stdout)
+	require.NoError(t, err)
+	assert.Equal(t, "binary", parsed.Scheme)
+	assert.Equal(t, urlPrefix+"/full/path/bin", parsed.Path)
+}
+
+func TestBinaryIOFailOnRelativePath(t *testing.T) {
+	_, err := BinaryIO("./bin", nil)("!")
+	assert.Error(t, err, "absolute path needed")
+}
+
+func TestLogFileAbsolutePath(t *testing.T) {
+	res, err := LogFile(prefix + "/full/path/file.txt")("!")
+	require.NoError(t, err)
+	expected := fmt.Sprintf("file://%s/full/path/file.txt", urlPrefix)
+	assert.Equal(t, expected, res.Config().Stdout)
+	assert.Equal(t, expected, res.Config().Stderr)
+
+	// Test parse back
+	parsed, err := url.Parse(res.Config().Stdout)
+	assert.NoError(t, err)
+	assert.Equal(t, "file", parsed.Scheme)
+	assert.Equal(t, urlPrefix+"/full/path/file.txt", parsed.Path)
+}
+
+func TestLogFileFailOnRelativePath(t *testing.T) {
+	_, err := LogFile("./file.txt")("!")
+	assert.Error(t, err, "absolute path needed")
+}
+
+type LogURIGeneratorTestCase struct {
+	// Arbitrary scheme string (e.g. "binary")
+	scheme string
+	// Path to executable/file: (e.g. "/some/path/to/bin.exe")
+	path string
+	// Extra query args expected to be in the URL (e.g. "id=123")
+	args map[string]string
+	// What the test case is expecting as an output (e.g. "binary:///some/path/to/bin.exe?id=123")
+	expected string
+	// Error string to be expected:
+	err string
+}
+
+func baseTestLogURIGenerator(t *testing.T, testCases []LogURIGeneratorTestCase) {
+	for _, tc := range testCases {
+		uri, err := LogURIGenerator(tc.scheme, tc.path, tc.args)
+		if tc.err != "" {
+			assert.ErrorContains(t, err, tc.err)
+			continue
+		}
+		assert.NoError(t, err)
+		assert.Equal(t, tc.expected, uri.String())
+	}
+}

pkg/cio/io_unix.go

@@ -0,0 +1,160 @@
+//go:build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package cio
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+	"syscall"
+
+	"github.com/containerd/fifo"
+)
+
+// NewFIFOSetInDir returns a new FIFOSet with paths in a temporary directory under root
+func NewFIFOSetInDir(root, id string, terminal bool) (*FIFOSet, error) {
+	if root != "" {
+		if err := os.MkdirAll(root, 0700); err != nil {
+			return nil, err
+		}
+	}
+	dir, err := os.MkdirTemp(root, "")
+	if err != nil {
+		return nil, err
+	}
+	closer := func() error {
+		return os.RemoveAll(dir)
+	}
+	return NewFIFOSet(Config{
+		Stdin:    filepath.Join(dir, id+"-stdin"),
+		Stdout:   filepath.Join(dir, id+"-stdout"),
+		Stderr:   filepath.Join(dir, id+"-stderr"),
+		Terminal: terminal,
+	}, closer), nil
+}
+
+func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
+	var ctx, cancel = context.WithCancel(context.Background())
+	pipes, err := openFifos(ctx, fifos)
+	if err != nil {
+		cancel()
+		return nil, err
+	}
+
+	if fifos.Stdin != "" {
+		go func() {
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(pipes.Stdin, ioset.Stdin, *p)
+			pipes.Stdin.Close()
+		}()
+	}
+
+	var wg = &sync.WaitGroup{}
+	if fifos.Stdout != "" {
+		wg.Add(1)
+		go func() {
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(ioset.Stdout, pipes.Stdout, *p)
+			pipes.Stdout.Close()
+			wg.Done()
+		}()
+	}
+
+	if !fifos.Terminal && fifos.Stderr != "" {
+		wg.Add(1)
+		go func() {
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(ioset.Stderr, pipes.Stderr, *p)
+			pipes.Stderr.Close()
+			wg.Done()
+		}()
+	}
+	return &cio{
+		config:  fifos.Config,
+		wg:      wg,
+		closers: append(pipes.closers(), fifos),
+		cancel: func() {
+			cancel()
+			for _, c := range pipes.closers() {
+				if c != nil {
+					c.Close()
+				}
+			}
+		},
+	}, nil
+}
+
+func openFifos(ctx context.Context, fifos *FIFOSet) (f pipes, retErr error) {
+	defer func() {
+		if retErr != nil {
+			fifos.Close()
+		}
+	}()
+
+	if fifos.Stdin != "" {
+		if f.Stdin, retErr = fifo.OpenFifo(ctx, fifos.Stdin, syscall.O_WRONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); retErr != nil {
+			return f, fmt.Errorf("failed to open stdin fifo: %w", retErr)
+		}
+		defer func() {
+			if retErr != nil && f.Stdin != nil {
+				f.Stdin.Close()
+			}
+		}()
+	}
+	if fifos.Stdout != "" {
+		if f.Stdout, retErr = fifo.OpenFifo(ctx, fifos.Stdout, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); retErr != nil {
+			return f, fmt.Errorf("failed to open stdout fifo: %w", retErr)
+		}
+		defer func() {
+			if retErr != nil && f.Stdout != nil {
+				f.Stdout.Close()
+			}
+		}()
+	}
+	if !fifos.Terminal && fifos.Stderr != "" {
+		if f.Stderr, retErr = fifo.OpenFifo(ctx, fifos.Stderr, syscall.O_RDONLY|syscall.O_CREAT|syscall.O_NONBLOCK, 0700); retErr != nil {
+			return f, fmt.Errorf("failed to open stderr fifo: %w", retErr)
+		}
+	}
+	return f, nil
+}
+
+// NewDirectIO returns an IO implementation that exposes the IO streams as io.ReadCloser
+// and io.WriteCloser.
+func NewDirectIO(ctx context.Context, fifos *FIFOSet) (*DirectIO, error) {
+	ctx, cancel := context.WithCancel(ctx)
+	pipes, err := openFifos(ctx, fifos)
+	return &DirectIO{
+		pipes: pipes,
+		cio: cio{
+			config:  fifos.Config,
+			closers: append(pipes.closers(), fifos),
+			cancel:  cancel,
+		},
+	}, err
+}

pkg/cio/io_unix_test.go

@@ -0,0 +1,305 @@
+//go:build !windows
+
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package cio
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"testing"
+
+	"github.com/containerd/fifo"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOpenFifos(t *testing.T) {
+	scenarios := []*FIFOSet{
+		{
+			Config: Config{
+				Stdin:  "",
+				Stdout: filepath.Join("This/does/not/exist", "test-stdout"),
+				Stderr: filepath.Join("This/does/not/exist", "test-stderr"),
+			},
+		},
+		{
+			Config: Config{
+				Stdin:  filepath.Join("This/does/not/exist", "test-stdin"),
+				Stdout: "",
+				Stderr: filepath.Join("This/does/not/exist", "test-stderr"),
+			},
+		},
+		{
+			Config: Config{
+				Stdin:  "",
+				Stdout: "",
+				Stderr: filepath.Join("This/does/not/exist", "test-stderr"),
+			},
+		},
+	}
+	for _, scenario := range scenarios {
+		_, err := openFifos(context.Background(), scenario)
+		assert.Error(t, err, scenario)
+	}
+}
+
+// TestOpenFifosWithTerminal tests openFifos should not open stderr if terminal
+// is set.
+func TestOpenFifosWithTerminal(t *testing.T) {
+	var ctx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+
+	ioFifoDir := t.TempDir()
+
+	cfg := Config{
+		Stdout: filepath.Join(ioFifoDir, "test-stdout"),
+		Stderr: filepath.Join(ioFifoDir, "test-stderr"),
+	}
+
+	// Without terminal, pipes.Stderr should not be nil
+	{
+		p, err := openFifos(ctx, NewFIFOSet(cfg, nil))
+		if err != nil {
+			t.Fatalf("unexpected error during openFifos: %v", err)
+		}
+
+		if p.Stderr == nil {
+			t.Fatalf("unexpected empty stderr pipe")
+		}
+	}
+
+	// With terminal, pipes.Stderr should be nil
+	{
+		cfg.Terminal = true
+		p, err := openFifos(ctx, NewFIFOSet(cfg, nil))
+		if err != nil {
+			t.Fatalf("unexpected error during openFifos: %v", err)
+		}
+
+		if p.Stderr != nil {
+			t.Fatalf("unexpected stderr pipe")
+		}
+	}
+}
+
+func assertHasPrefix(t *testing.T, s, prefix string) {
+	t.Helper()
+	if !strings.HasPrefix(s, prefix) {
+		t.Fatalf("expected %s to start with %s", s, prefix)
+	}
+}
+
+func TestNewFIFOSetInDir(t *testing.T) {
+	root := t.TempDir()
+
+	fifos, err := NewFIFOSetInDir(root, "theid", true)
+	assert.NoError(t, err)
+
+	dir := filepath.Dir(fifos.Stdin)
+	assertHasPrefix(t, dir, root)
+	expected := &FIFOSet{
+		Config: Config{
+			Stdin:    filepath.Join(dir, "theid-stdin"),
+			Stdout:   filepath.Join(dir, "theid-stdout"),
+			Stderr:   filepath.Join(dir, "theid-stderr"),
+			Terminal: true,
+		},
+	}
+
+	assert.Equal(t, fifos.Config, expected.Config)
+
+	files, err := os.ReadDir(root)
+	assert.NoError(t, err)
+	assert.Len(t, files, 1)
+
+	assert.Nil(t, fifos.Close())
+	files, err = os.ReadDir(root)
+	assert.NoError(t, err)
+	assert.Len(t, files, 0)
+}
+
+func TestNewAttach(t *testing.T) {
+	testCases := []struct {
+		name                                          string
+		expectedStdin, expectedStdout, expectedStderr string
+	}{
+		{
+			name:           "attach to all streams (stdin, stdout, and stderr)",
+			expectedStdin:  "this is the stdin",
+			expectedStdout: "this is the stdout",
+			expectedStderr: "this is the stderr",
+		},
+		{
+			name:           "don't attach to stdin",
+			expectedStdout: "this is the stdout",
+			expectedStderr: "this is the stderr",
+		},
+		{
+			name:           "don't attach to stdout",
+			expectedStdin:  "this is the stdin",
+			expectedStderr: "this is the stderr",
+		},
+		{
+			name:           "don't attach to stderr",
+			expectedStdin:  "this is the stdin",
+			expectedStdout: "this is the stdout",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var (
+				stdin  = bytes.NewBufferString(tc.expectedStdin)
+				stdout = new(bytes.Buffer)
+				stderr = new(bytes.Buffer)
+
+				// The variables below have to be of the interface type (i.e., io.Reader/io.Writer)
+				// instead of the concrete type (i.e., *bytes.Buffer) *before* being passed to NewAttach.
+				// Otherwise, in NewAttach, the interface value won't be nil
+				// (it's just that the concrete value inside the interface itself is nil. [1]),
+				// which means that the corresponding FIFO path won't be set to be an empty string,
+				// and that's not what we want.
+				//
+				// [1] https://go.dev/tour/methods/12
+				stdinArg             io.Reader
+				stdoutArg, stderrArg io.Writer
+			)
+			if tc.expectedStdin != "" {
+				stdinArg = stdin
+			}
+			if tc.expectedStdout != "" {
+				stdoutArg = stdout
+			}
+			if tc.expectedStderr != "" {
+				stderrArg = stderr
+			}
+
+			attacher := NewAttach(WithStreams(stdinArg, stdoutArg, stderrArg))
+
+			fifos, err := NewFIFOSetInDir("", "theid", false)
+			assert.NoError(t, err)
+
+			attachedFifos, err := attacher(fifos)
+			assert.NoError(t, err)
+			defer attachedFifos.Close()
+
+			producers := setupFIFOProducers(t, attachedFifos.Config())
+			initProducers(t, producers, tc.expectedStdout, tc.expectedStderr)
+
+			var actualStdin []byte
+			if producers.Stdin != nil {
+				actualStdin, err = io.ReadAll(producers.Stdin)
+				assert.NoError(t, err)
+			}
+
+			attachedFifos.Wait()
+			attachedFifos.Cancel()
+			assert.Nil(t, attachedFifos.Close())
+
+			assert.Equal(t, tc.expectedStdout, stdout.String())
+			assert.Equal(t, tc.expectedStderr, stderr.String())
+			assert.Equal(t, tc.expectedStdin, string(actualStdin))
+		})
+	}
+}
+
+type producers struct {
+	Stdin  io.ReadCloser
+	Stdout io.WriteCloser
+	Stderr io.WriteCloser
+}
+
+func setupFIFOProducers(t *testing.T, fifos Config) producers {
+	var (
+		err   error
+		pipes producers
+		ctx   = context.Background()
+	)
+
+	if fifos.Stdin != "" {
+		pipes.Stdin, err = fifo.OpenFifo(ctx, fifos.Stdin, syscall.O_RDONLY, 0)
+		assert.NoError(t, err)
+	}
+
+	if fifos.Stdout != "" {
+		pipes.Stdout, err = fifo.OpenFifo(ctx, fifos.Stdout, syscall.O_WRONLY, 0)
+		assert.NoError(t, err)
+	}
+
+	if fifos.Stderr != "" {
+		pipes.Stderr, err = fifo.OpenFifo(ctx, fifos.Stderr, syscall.O_WRONLY, 0)
+		assert.NoError(t, err)
+	}
+
+	return pipes
+}
+
+func initProducers(t *testing.T, producers producers, stdout, stderr string) {
+	if producers.Stdout != nil {
+		_, err := producers.Stdout.Write([]byte(stdout))
+		assert.NoError(t, err)
+		assert.Nil(t, producers.Stdout.Close())
+	}
+
+	if producers.Stderr != nil {
+		_, err := producers.Stderr.Write([]byte(stderr))
+		assert.NoError(t, err)
+		assert.Nil(t, producers.Stderr.Close())
+	}
+}
+
+func TestLogURIGenerator(t *testing.T) {
+	baseTestLogURIGenerator(t, []LogURIGeneratorTestCase{
+		{
+			scheme:   "fifo",
+			path:     "/full/path/pipe.fifo",
+			expected: "fifo:///full/path/pipe.fifo",
+		},
+		{
+			scheme: "file",
+			path:   "/full/path/file.txt",
+			args: map[string]string{
+				"maxSize": "100MB",
+			},
+			expected: "file:///full/path/file.txt?maxSize=100MB",
+		},
+		{
+			scheme: "binary",
+			path:   "/full/path/bin",
+			args: map[string]string{
+				"id": "testing",
+			},
+			expected: "binary:///full/path/bin?id=testing",
+		},
+		{
+			scheme: "unknown",
+			path:   "nowhere",
+			err:    "must be absolute",
+		},
+		{
+			scheme: "binary",
+			path:   "C:\\path\\to\\binary",
+			// NOTE: Windows paths should not be parse-able outside of Windows:
+			err: "must be absolute",
+		},
+	})
+}

pkg/cio/io_windows.go

@@ -0,0 +1,157 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package cio
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	winio "github.com/Microsoft/go-winio"
+	"github.com/containerd/log"
+)
+
+const pipeRoot = `\\.\pipe`
+
+// NewFIFOSetInDir returns a new set of fifos for the task
+func NewFIFOSetInDir(_, id string, terminal bool) (*FIFOSet, error) {
+	stderrPipe := ""
+	if !terminal {
+		stderrPipe = fmt.Sprintf(`%s\ctr-%s-stderr`, pipeRoot, id)
+	}
+	return NewFIFOSet(Config{
+		Terminal: terminal,
+		Stdin:    fmt.Sprintf(`%s\ctr-%s-stdin`, pipeRoot, id),
+		Stdout:   fmt.Sprintf(`%s\ctr-%s-stdout`, pipeRoot, id),
+		Stderr:   stderrPipe,
+	}, nil), nil
+}
+
+func copyIO(fifos *FIFOSet, ioset *Streams) (_ *cio, retErr error) {
+	cios := &cio{config: fifos.Config}
+
+	defer func() {
+		if retErr != nil {
+			_ = cios.Close()
+		}
+	}()
+
+	if fifos.Stdin != "" {
+		l, err := winio.ListenPipe(fifos.Stdin, nil)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create stdin pipe %s: %w", fifos.Stdin, err)
+		}
+		cios.closers = append(cios.closers, l)
+
+		go func() {
+			c, err := l.Accept()
+			if err != nil {
+				log.L.WithError(err).Errorf("failed to accept stdin connection on %s", fifos.Stdin)
+				return
+			}
+
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(c, ioset.Stdin, *p)
+			c.Close()
+			l.Close()
+		}()
+	}
+
+	if fifos.Stdout != "" {
+		l, err := winio.ListenPipe(fifos.Stdout, nil)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create stdout pipe %s: %w", fifos.Stdout, err)
+		}
+		cios.closers = append(cios.closers, l)
+
+		go func() {
+			c, err := l.Accept()
+			if err != nil {
+				log.L.WithError(err).Errorf("failed to accept stdout connection on %s", fifos.Stdout)
+				return
+			}
+
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(ioset.Stdout, c, *p)
+			c.Close()
+			l.Close()
+		}()
+	}
+
+	if fifos.Stderr != "" {
+		l, err := winio.ListenPipe(fifos.Stderr, nil)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create stderr pipe %s: %w", fifos.Stderr, err)
+		}
+		cios.closers = append(cios.closers, l)
+
+		go func() {
+			c, err := l.Accept()
+			if err != nil {
+				log.L.WithError(err).Errorf("failed to accept stderr connection on %s", fifos.Stderr)
+				return
+			}
+
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(ioset.Stderr, c, *p)
+			c.Close()
+			l.Close()
+		}()
+	}
+
+	return cios, nil
+}
+
+// NewDirectIO returns an IO implementation that exposes the IO streams as io.ReadCloser
+// and io.WriteCloser.
+func NewDirectIO(stdin io.WriteCloser, stdout, stderr io.ReadCloser, terminal bool) *DirectIO {
+	return &DirectIO{
+		pipes: pipes{
+			Stdin:  stdin,
+			Stdout: stdout,
+			Stderr: stderr,
+		},
+		cio: cio{
+			config: Config{Terminal: terminal},
+		},
+	}
+}
+
+// NewDirectIOFromFIFOSet returns an IO implementation that exposes the IO streams as io.ReadCloser
+// and io.WriteCloser.
+func NewDirectIOFromFIFOSet(ctx context.Context, stdin io.WriteCloser, stdout, stderr io.ReadCloser, fifos *FIFOSet) *DirectIO {
+	_, cancel := context.WithCancel(ctx)
+	pipes := pipes{
+		Stdin:  stdin,
+		Stdout: stdout,
+		Stderr: stderr,
+	}
+	return &DirectIO{
+		pipes: pipes,
+		cio: cio{
+			config:  fifos.Config,
+			closers: append(pipes.closers(), fifos),
+			cancel:  cancel,
+		},
+	}
+}

pkg/cio/io_windows_test.go

@@ -0,0 +1,111 @@
+/*
+   Copyright The containerd Authors.
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package cio
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewFifoSetInDir_NoTerminal(t *testing.T) {
+	set, err := NewFIFOSetInDir("", t.Name(), false)
+	if err != nil {
+		t.Fatalf("NewFifoSetInDir failed with: %v", err)
+	}
+
+	assert.True(t, !set.Terminal, "FIFOSet.Terminal should be false")
+	assert.NotEmpty(t, set.Stdin, "FIFOSet.Stdin should be set")
+	assert.NotEmpty(t, set.Stdout, "FIFOSet.Stdout should be set")
+	assert.NotEmpty(t, set.Stderr, "FIFOSet.Stderr should be set")
+}
+
+func TestNewFifoSetInDir_Terminal(t *testing.T) {
+	set, err := NewFIFOSetInDir("", t.Name(), true)
+	if err != nil {
+		t.Fatalf("NewFifoSetInDir failed with: %v", err)
+	}
+
+	assert.True(t, set.Terminal, "FIFOSet.Terminal should be true")
+	assert.NotEmpty(t, set.Stdin, "FIFOSet.Stdin should be set")
+	assert.NotEmpty(t, set.Stdout, "FIFOSet.Stdout should be set")
+	assert.Empty(t, set.Stderr, "FIFOSet.Stderr should not be set")
+}
+
+func TestLogFileBackslash(t *testing.T) {
+	testcases := []struct {
+		path string
+	}{
+		{`C:/foo/bar.log`},
+		{`C:\foo\bar.log`},
+	}
+	for _, tc := range testcases {
+		f := LogFile(tc.path)
+		res, err := f("unused")
+		require.NoError(t, err)
+		assert.Equal(t, res.Config().Stdout, res.Config().Stderr)
+		assert.Equal(t, "file:///C:/foo/bar.log", res.Config().Stdout)
+	}
+}
+
+func TestLogURIGenerator(t *testing.T) {
+	baseTestLogURIGenerator(t, []LogURIGeneratorTestCase{
+		{
+			scheme:   "slashes",
+			path:     "C:/full/path/pipe.fifo",
+			expected: "slashes:///C:/full/path/pipe.fifo",
+		},
+		{
+			scheme:   "backslashes",
+			path:     "C:\\full\\path\\pipe.fifo",
+			expected: "backslashes:///C:/full/path/pipe.fifo",
+		},
+		{
+			scheme:   "mixedslashes",
+			path:     "C:\\full/path/pipe.fifo",
+			expected: "mixedslashes:///C:/full/path/pipe.fifo",
+		},
+		{
+			scheme: "file",
+			path:   "C:/full/path/file.txt",
+			args: map[string]string{
+				"maxSize": "100MB",
+			},
+			expected: "file:///C:/full/path/file.txt?maxSize=100MB",
+		},
+		{
+			scheme: "binary",
+			path:   "C:/full/path/bin",
+			args: map[string]string{
+				"id": "testing",
+			},
+			expected: "binary:///C:/full/path/bin?id=testing",
+		},
+		{
+			scheme: "unknown",
+			path:   "nowhere",
+			err:    "must be absolute",
+		},
+		{
+			scheme: "unixpath",
+			path:   "/some/unix/path",
+			// NOTE: Unix paths should not be usable on Windows:
+			err: "must be absolute",
+		},
+	})
+}

pkg/cri/io/container_io.go

@@ -22,7 +22,7 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/containerd/containerd/v2/cio"
+	"github.com/containerd/containerd/v2/pkg/cio"
 	"github.com/containerd/log"
 
 	"github.com/containerd/containerd/v2/pkg/cri/util"

pkg/cri/io/exec_io.go

@@ -20,7 +20,7 @@ import (
 	"io"
 	"sync"
 
-	"github.com/containerd/containerd/v2/cio"
+	"github.com/containerd/containerd/v2/pkg/cio"
 	cioutil "github.com/containerd/containerd/v2/pkg/ioutil"
 	"github.com/containerd/log"
 )

pkg/cri/io/helpers.go

@@ -24,7 +24,7 @@ import (
 	"sync"
 	"syscall"
 
-	"github.com/containerd/containerd/v2/cio"
+	"github.com/containerd/containerd/v2/pkg/cio"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 )
 

pkg/cri/server/container_execsync.go

@@ -24,10 +24,10 @@ import (
 	"syscall"
 	"time"
 
-	containerdio "github.com/containerd/containerd/v2/cio"
 	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/errdefs"
 	"github.com/containerd/containerd/v2/oci"
+	containerdio "github.com/containerd/containerd/v2/pkg/cio"
 	"github.com/containerd/log"
 	"k8s.io/client-go/tools/remotecommand"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"

pkg/cri/server/container_execsync_test.go

@@ -24,8 +24,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/containerd/containerd/v2/cio"
 	containerd "github.com/containerd/containerd/v2/client"
+	"github.com/containerd/containerd/v2/pkg/cio"
 	cioutil "github.com/containerd/containerd/v2/pkg/ioutil"
 	"github.com/stretchr/testify/assert"
 )

pkg/cri/server/container_start.go

@@ -23,9 +23,9 @@ import (
 	"io"
 	"time"
 
-	containerdio "github.com/containerd/containerd/v2/cio"
 	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/errdefs"
+	containerdio "github.com/containerd/containerd/v2/pkg/cio"
 	"github.com/containerd/log"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 

pkg/cri/server/events.go

@@ -30,10 +30,10 @@ import (
 
 	eventtypes "github.com/containerd/containerd/v2/api/events"
 	apitasks "github.com/containerd/containerd/v2/api/services/tasks/v1"
-	containerdio "github.com/containerd/containerd/v2/cio"
 	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/errdefs"
 	"github.com/containerd/containerd/v2/events"
+	containerdio "github.com/containerd/containerd/v2/pkg/cio"
 	"github.com/containerd/containerd/v2/pkg/cri/constants"
 	containerstore "github.com/containerd/containerd/v2/pkg/cri/store/container"
 	sandboxstore "github.com/containerd/containerd/v2/pkg/cri/store/sandbox"

pkg/cri/server/podsandbox/sandbox_run.go

@@ -29,11 +29,11 @@ import (
 	"github.com/opencontainers/selinux/go-selinux"
 	runtime "k8s.io/cri-api/pkg/apis/runtime/v1"
 
-	containerdio "github.com/containerd/containerd/v2/cio"
 	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/sandbox"
 	"github.com/containerd/containerd/v2/core/snapshots"
 	"github.com/containerd/containerd/v2/errdefs"
+	containerdio "github.com/containerd/containerd/v2/pkg/cio"
 	criconfig "github.com/containerd/containerd/v2/pkg/cri/config"
 	crilabels "github.com/containerd/containerd/v2/pkg/cri/labels"
 	customopts "github.com/containerd/containerd/v2/pkg/cri/opts"

pkg/cri/server/restart.go

@@ -23,9 +23,9 @@ import (
 	"path/filepath"
 	"time"
 
-	containerdio "github.com/containerd/containerd/v2/cio"
 	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/errdefs"
+	containerdio "github.com/containerd/containerd/v2/pkg/cio"
 	criconfig "github.com/containerd/containerd/v2/pkg/cri/config"
 	crilabels "github.com/containerd/containerd/v2/pkg/cri/labels"
 	"github.com/containerd/containerd/v2/pkg/cri/server/podsandbox"