Add support for using terminals with Containers and Executables

AGENTS.md

@@ -62,6 +62,9 @@ Place new code in the correct location according to the project's structure:
 - Run `make generate` after making changes to API definitions (files under `api/v1` folder).
 - Run `make generate-grpc` after making changes to protobuf definitions (files with `.proto` extension).
 
+## Test authoring
+- Unless a tight, specific timeout is ESSENTIAL for testing a specific funtion, avoid using hard-coded timeouts (e.g. obtained via `time.After()`). Instead, use test context (created via `testutil.GetTestContext()`) and test context `Done()` channel for timeouts in tests.
+
 
 # Working inside the repository
 
@@ -79,3 +82,4 @@ Place new code in the correct location according to the project's structure:
 - Tests for specific packages can be run with `go test -count 1 -parallel 32 <package>` (after building prerequisites), where `<package>` indicates the package that was changed and needs to be tested. This is significantly faster than running all tests, and should be used when working on a specific code change. For final verification, run `make test` to ensure all tests pass.
 - If working on a change involving a lot of goroutine synchronization, channel operations, and locking, run tests with `-race` flag to enable race detection. This works on MacOS and Linux ONLY, do not try this when the current OS is Windows.
 - Our test context (obtained via `GetTestContext` function from `pkg/testutil` package) can be adjusted via an environment variable to use non-standard test timeout. Set `TEST_CONTEXT_TIMEOUT` environment variable to number of seconds that will be used as timeout value. For example, `TEST_CONTEXT_TIMEOUT=30` will make the test context expire after 30 seconds. 30 seconds is a good value when running individual tests.
+- No single test (not even those classified as "advanced" e.g. tests using real container orchestrator) should take more than 3 minutes (180 seconds) to run. Use AT MOST 3 minutes as a timeout when running individual tests, or subset of tests, with "go test" command ("-timeout" flag).

Makefile

@@ -114,6 +114,7 @@ DELAY_TOOL ?= $(TOOL_BIN)/delay$(exe_suffix)
 LFWRITER_TOOL ?= $(TOOL_BIN)/lfwriter$(exe_suffix)
 PARROT_TOOL ?= $(TOOL_BIN)/parrot$(exe_suffix)
 PARROT_TOOL_CONTAINER_BINARY ?= $(TOOL_BIN)/parrot_c
+TERMCHILD_TOOL ?= $(TOOL_BIN)/termchild$(exe_suffix)
 GO_LICENSES ?= $(TOOL_BIN)/go-licenses$(exe_suffix)
 PROTOC ?= $(TOOL_BIN)/protoc/bin/protoc$(exe_suffix)
 
@@ -339,9 +340,9 @@ endif
 # mirrored there.
 
 ifeq (4.4,$(firstword $(sort $(MAKE_VERSION) 4.4)))
-TEST_PREREQS := generate-grpc .WAIT build-dcp build-dcptun-containerexe delay-tool lfwriter-tool parrot-tool parrot-tool-containerexe
+TEST_PREREQS := generate-grpc .WAIT build-dcp build-dcptun-containerexe delay-tool lfwriter-tool parrot-tool parrot-tool-containerexe termchild-tool
 else
-TEST_PREREQS := generate-grpc build-dcp build-dcptun-containerexe delay-tool lfwriter-tool parrot-tool parrot-tool-containerexe
+TEST_PREREQS := generate-grpc build-dcp build-dcptun-containerexe delay-tool lfwriter-tool parrot-tool parrot-tool-containerexe termchild-tool
 endif
 
 .PHONY: test-prereqs
@@ -407,6 +408,12 @@ delay-tool: $(DELAY_TOOL)
 $(DELAY_TOOL): $(wildcard ./test/delay/*.go) | $(TOOL_BIN)
 	$(GO_BIN) build -o $(DELAY_TOOL) github.com/microsoft/dcp/test/delay
 
+# termchild-tool is used for internal/termpty pseudo-terminal tests
+.PHONY: termchild-tool
+termchild-tool: $(TERMCHILD_TOOL)
+$(TERMCHILD_TOOL): $(wildcard ./test/termchild/*.go) | $(TOOL_BIN)
+	$(GO_BIN) build -o $(TERMCHILD_TOOL) github.com/microsoft/dcp/test/termchild
+
 # lfwriter tool is used for testing lockfile package
 .PHONY: lfwriter-tool
 lfwriter-tool: $(LFWRITER_TOOL)

api/v1/common_types.go

@@ -105,6 +105,12 @@ type StdIoStreamableResource interface {
 	GetResourceId() string
 	Done() bool
 
+	// HasTerminal reports whether the resource is configured to bridge its
+	// stdin/stdout/stderr to a pseudo-terminal. When true, the resource does
+	// not produce stdout/stderr log files and API requests for those log
+	// streams should fail. System logs are unaffected.
+	HasTerminal() bool
+
 	// This is set by Kubernetes with 1-second precision when the resource is deleted
 	// Hence we use metav1.Time here instead of metav1.MicroTime
 	GetDeletionTimestamp() *metav1.Time

api/v1/container_exec_types.go

@@ -114,6 +114,12 @@ func (ce *ContainerExec) HasStdErr() bool {
 	return true
 }
 
+// HasTerminal implements StdIoStreamableResource. ContainerExec does not
+// support pseudo-terminal allocation, so this always returns false.
+func (ce *ContainerExec) HasTerminal() bool {
+	return false
+}
+
 // StdOutFile implements StdOutStreamableResource.
 func (ce *ContainerExec) GetStdOutFile() string {
 	return ce.Status.StdOutFile

api/v1/container_types.go

@@ -695,6 +695,13 @@ type ContainerSpec struct {
 	// PEM formatted public certificates to be created in the container
 	// +optional
 	PemCertificates *ContainerPemCertificates `json:"pemCertificates,omitempty"`
+
+	// Optional terminal/PTY configuration. When set, the container's primary process
+	// is started with connection to a pseudo-terminal
+	// and its stdin/stdout/stderr are bridged to the configured UDS via HMP v1,
+	// instead of the container being run detached with separate log capture.
+	// +optional
+	Terminal *TerminalSpec `json:"terminal,omitempty"`
 }
 
 func (cs *ContainerSpec) Equal(other *ContainerSpec) bool {
@@ -812,6 +819,10 @@ func (cs *ContainerSpec) Equal(other *ContainerSpec) bool {
 		return false
 	}
 
+	if !cs.Terminal.Equal(other.Terminal) {
+		return false
+	}
+
 	return true
 }
 
@@ -1020,6 +1031,12 @@ func (cs *ContainerSpec) GetLifecycleKey() (string, bool, error) {
 		}
 	}
 
+	if cs.Terminal != nil {
+		// Columns and rows do not matter that much (the client can always resize the terminal as necessary),
+		// but once a Container is started with terminal support, the UDS path does not change.
+		hashErr = errors.Join(hashErr, encoder.Encode(cs.Terminal.UDSPath))
+	}
+
 	// Compute the hash for the lifecycle key
 	lifecycleKey := fmt.Sprintf("%x", fnvHash.Sum(nil))
 
@@ -1286,6 +1303,10 @@ func (c *Container) Validate(ctx context.Context) field.ErrorList {
 		errorList = append(errorList, field.Required(field.NewPath("spec", "containerName"), "containerName must be set to a value when persistent is true"))
 	}
 
+	if c.Spec.Persistent && c.Spec.Terminal != nil {
+		errorList = append(errorList, field.Forbidden(field.NewPath("spec", "terminal"), "Persistent Containers cannot use a terminal."))
+	}
+
 	monitorTimestampSet := !c.Spec.MonitorTimestamp.IsZero()
 	if c.Spec.MonitorPID != nil && *c.Spec.MonitorPID <= 0 {
 		errorList = append(errorList, field.Invalid(field.NewPath("spec", "monitorPid"), *c.Spec.MonitorPID, "monitorPid must be positive"))
@@ -1342,6 +1363,9 @@ func (c *Container) Validate(ctx context.Context) field.ErrorList {
 	// Validate PEM certificates configuration
 	errorList = append(errorList, c.Spec.PemCertificates.Validate(field.NewPath("spec", "pemCertificates"))...)
 
+	// Validate terminal configuration
+	errorList = append(errorList, c.Spec.Terminal.Validate(field.NewPath("spec", "terminal"))...)
+
 	// Validate that annotations don't exceed the Kubernetes size limit.
 	// This provides a clearer error message than the generic Kubernetes API server error,
 	// especially when long arguments or environment variables are stored in annotations.
@@ -1443,6 +1467,8 @@ func (c *Container) ValidateUpdate(ctx context.Context, obj runtime.Object) fiel
 		}
 	}
 
+	errorList = append(errorList, c.Spec.Terminal.ValidateUpdate(oldContainer.Spec.Terminal, field.NewPath("spec", "terminal"))...)
+
 	return errorList
 }
 

api/v1/container_types_test.go

@@ -146,6 +146,65 @@ func TestContainerValidateMonitorFields(t *testing.T) {
 	}
 }
 
+func TestContainerValidateTerminalPersistentCombination(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name        string
+		spec        ContainerSpec
+		expectError bool
+	}{
+		{
+			name: "non-persistent with terminal is allowed",
+			spec: ContainerSpec{
+				Image: "api:dev",
+				Terminal: &TerminalSpec{
+					UDSPath: "/tmp/api.sock",
+					Cols:    80,
+					Rows:    24,
+				},
+			},
+		},
+		{
+			name: "persistent without terminal is allowed",
+			spec: ContainerSpec{
+				Image:         "api:dev",
+				ContainerName: "api",
+				Persistent:    true,
+			},
+		},
+		{
+			name: "persistent with terminal is rejected",
+			spec: ContainerSpec{
+				Image:         "api:dev",
+				ContainerName: "api",
+				Persistent:    true,
+				Terminal: &TerminalSpec{
+					UDSPath: "/tmp/api.sock",
+					Cols:    80,
+					Rows:    24,
+				},
+			},
+			expectError: true,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			t.Parallel()
+
+			container := &Container{Spec: testCase.spec}
+			errors := container.Validate(nil)
+
+			if testCase.expectError {
+				require.NotEmpty(t, errors)
+			} else {
+				require.Empty(t, errors)
+			}
+		})
+	}
+}
+
 func TestImageLayerValidate(t *testing.T) {
 	t.Parallel()
 

api/v1/executable_types.go

@@ -291,6 +291,15 @@ type ExecutableSpec struct {
 	// PEM formatted certificates to be written for the Executable
 	// +optional
 	PemCertificates *ExecutablePemCertificates `json:"pemCertificates,omitempty"`
+
+	// Optional terminal/PTY configuration. When set, the Executable process
+	// is started with connection to a pseudo-terminal
+	// and its stdin/stdout/stderr are bridged to the configured UDS via HMP v1.
+	// When terminal configuration is present, there will be no "logs",
+	// i.e. ExecutableStatus.StdOutFile and ExecutableStatus.StdErrFile will be empty,
+	// and API requests to fetch logs will fail.
+	// +optional
+	Terminal *TerminalSpec `json:"terminal,omitempty"`
 }
 
 func (es ExecutableSpec) Equal(other ExecutableSpec) bool {
@@ -364,6 +373,10 @@ func (es ExecutableSpec) Equal(other ExecutableSpec) bool {
 		return false
 	}
 
+	if !es.Terminal.Equal(other.Terminal) {
+		return false
+	}
+
 	return true
 }
 
@@ -423,6 +436,12 @@ func (es *ExecutableSpec) GetLifecycleKey() (string, bool, error) {
 		hashErr = errors.Join(hashErr, encoder.Encode(es.PemCertificates.ContinueOnError))
 	}
 
+	if es.Terminal != nil {
+		// Columns and rows do not matter that much (the client can always resize the terminal as necessary),
+		// but once an Executable is started with terminal support, the UDS path does not change.
+		hashErr = errors.Join(hashErr, encoder.Encode(es.Terminal.UDSPath))
+	}
+
 	lifecycleKey := fmt.Sprintf("%x", fnvHash.Sum(nil))
 	return lifecycleKey, true, hashErr
 }
@@ -570,6 +589,9 @@ func (es ExecutableSpec) Validate(specPath *field.Path) field.ErrorList {
 	if es.Persistent && len(es.FallbackExecutionTypes) > 0 {
 		errorList = append(errorList, field.Invalid(specPath.Child("fallbackExecutionTypes"), es.FallbackExecutionTypes, "Persistent Executables cannot use fallback execution types."))
 	}
+	if es.Persistent && es.Terminal != nil {
+		errorList = append(errorList, field.Forbidden(specPath.Child("terminal"), "Persistent Executables cannot use a terminal."))
+	}
 
 	monitorTimestampSet := !es.MonitorTimestamp.IsZero()
 	if es.MonitorPID != nil && *es.MonitorPID <= 0 {
@@ -595,6 +617,8 @@ func (es ExecutableSpec) Validate(specPath *field.Path) field.ErrorList {
 
 	errorList = append(errorList, es.PemCertificates.Validate(specPath.Child("pemCertificates"))...)
 
+	errorList = append(errorList, es.Terminal.Validate(specPath.Child("terminal"))...)
+
 	return errorList
 }
 
@@ -678,6 +702,13 @@ func (ce *Executable) HasStdErr() bool {
 	return true
 }
 
+// HasTerminal implements StdIoStreamableResource. It reports whether the
+// Executable is configured to bridge stdin/stdout/stderr to a pseudo-terminal,
+// in which case no stdout/stderr log files are captured.
+func (e *Executable) HasTerminal() bool {
+	return e.Spec.Terminal != nil
+}
+
 // StdOutFile implements StdOutStreamableResource.
 func (e *Executable) GetStdOutFile() string {
 	return e.Status.StdOutFile
@@ -799,6 +830,8 @@ func (e *Executable) ValidateUpdate(ctx context.Context, obj runtime.Object) fie
 		errorList = append(errorList, field.Forbidden(field.NewPath("spec", "pemCertificates"), "pemCertificates cannot be changed once an Executable is created."))
 	}
 
+	errorList = append(errorList, e.Spec.Terminal.ValidateUpdate(oldExe.Spec.Terminal, field.NewPath("spec", "terminal"))...)
+
 	return errorList
 }
 

api/v1/executable_types_test.go

@@ -202,6 +202,63 @@ func TestExecutableValidateMonitorFields(t *testing.T) {
 	}
 }
 
+func TestExecutableValidateTerminalPersistentCombination(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		name        string
+		spec        ExecutableSpec
+		expectError bool
+	}{
+		{
+			name: "non-persistent with terminal is allowed",
+			spec: ExecutableSpec{
+				ExecutablePath: "/path/to/app",
+				Terminal: &TerminalSpec{
+					UDSPath: "/tmp/app.sock",
+					Cols:    80,
+					Rows:    24,
+				},
+			},
+		},
+		{
+			name: "persistent without terminal is allowed",
+			spec: ExecutableSpec{
+				ExecutablePath: "/path/to/app",
+				Persistent:     true,
+			},
+		},
+		{
+			name: "persistent with terminal is rejected",
+			spec: ExecutableSpec{
+				ExecutablePath: "/path/to/app",
+				Persistent:     true,
+				Terminal: &TerminalSpec{
+					UDSPath: "/tmp/app.sock",
+					Cols:    80,
+					Rows:    24,
+				},
+			},
+			expectError: true,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			t.Parallel()
+
+			exe := &Executable{Spec: testCase.spec}
+			errors := exe.Validate(context.Background())
+
+			if testCase.expectError {
+				require.NotEmpty(t, errors)
+			} else {
+				require.Empty(t, errors)
+			}
+		})
+	}
+}
+
 func TestExecutableGetLifecycleKeyIgnoresImplicitEffectiveEnv(t *testing.T) {
 	t.Parallel()
 

api/v1/log_storage.go

@@ -43,6 +43,15 @@ const (
 // ResourceLogStreamFactory is an entity that knows how to stream logs for a given resource.
 // +kubebuilder:object:generate=false
 type ResourceLogStreamer interface {
+	// Preflight is called synchronously by the LogStorage before any streaming goroutine is launched.
+	// It offers an opportunity for a streamer to surface a proper Kubernetes API error to the client,
+	// because once StreamLogs starts running its results are delivered through an io.Pipe
+	// whose (successful) HTTP response status has already been committed.
+	// Implementations should only perform cheap, side-effect-free checks here
+	// (e.g. validating that the requested log source is compatible with the resource's configuration).
+	// Returning nil allows streaming to proceed; returning an error short-circuits the request.
+	Preflight(obj apiserver_resource.Object, opts *LogOptions) error
+
 	// StreamLogs returns a boolean indicating if the logs are ready to be streamed, a channel that will be closed when the logs are done streaming, and an error if one occurred.
 	StreamLogs(ctx context.Context, dest io.Writer, obj apiserver_resource.Object, opts *LogOptions, log logr.Logger) (ResourceStreamStatus, <-chan struct{}, error)
 
@@ -447,6 +456,10 @@ func (ls *LogStorage) resourceStreamerFactory(resourceName string, options *LogO
 			return nil, false, "", apierrors.NewInternalError(fmt.Errorf("parent storage returned object of wrong type: %s", obj.GetObjectKind().GroupVersionKind().String()))
 		}
 
+		if preflightErr := ls.logStreamer.Preflight(apiObj, options); preflightErr != nil {
+			return nil, false, "", preflightErr
+		}
+
 		log := contextdata.GetContextLogger(ctx)
 		log = log.WithName("Logstorage").WithValues(
 			"Kind", apiObj.GetObjectKind().GroupVersionKind().String(),

api/v1/service_types.go

@@ -152,6 +152,12 @@ func (ce *Service) HasStdErr() bool {
 	return false
 }
 
+// HasTerminal implements StdIoStreamableResource. Service does not support
+// pseudo-terminal allocation, so this always returns false.
+func (ce *Service) HasTerminal() bool {
+	return false
+}
+
 // GetStdErrFile implements StdIoStreamableResource.
 func (svc *Service) GetStdErrFile() string {
 	return ""