[shimV2] implements task, shimdiag service and wires up pod/container controllers

cmd/containerd-shim-lcow-v2/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/Microsoft/hcsshim/internal/log"
 	"github.com/Microsoft/hcsshim/internal/oc"
 	"github.com/Microsoft/hcsshim/internal/shim"
+	"github.com/Microsoft/hcsshim/osversion"
 
 	"github.com/containerd/errdefs"
 	"github.com/sirupsen/logrus"
@@ -42,6 +43,13 @@ func main() {
 		os.Exit(1)
 	}
 
+	// This shim is supported on Windows Build 26100 and later.
+	if osversion.Build() < osversion.V25H1Server {
+		_, _ = fmt.Fprintf(os.Stderr,
+			"%s: Windows version [%v] is not supported", service.ShimName, osversion.Build())
+		os.Exit(1)
+	}
+
 	// Start the shim manager event loop. The manager is responsible for
 	// handling containerd start/stop lifecycle calls for the shim process.
 	shim.Run(context.Background(), newShimManager(service.ShimName), func(c *shim.Config) {

cmd/containerd-shim-lcow-v2/manager.go

@@ -217,19 +217,19 @@ func (m *shimManager) Stop(_ context.Context, id string) (resp shim.StopStatus,
 	if sys, _ := hcs.OpenComputeSystem(ctx, id); sys != nil {
 		defer sys.Close()
 		if err := sys.Terminate(ctx); err != nil {
-			fmt.Fprintf(os.Stderr, "failed to terminate '%s': %v", id, err)
+			fmt.Fprintf(os.Stderr, "failed to terminate %q: %v", id, err)
 		} else {
 			ch := make(chan error, 1)
 			go func() { ch <- sys.Wait() }()
 			t := time.NewTimer(time.Second * 30)
 			select {
 			case <-t.C:
 				sys.Close()
-				return resp, fmt.Errorf("timed out waiting for '%s' to terminate", id)
+				return resp, fmt.Errorf("timed out waiting for %q to terminate", id)
 			case err := <-ch:
 				t.Stop()
 				if err != nil {
-					fmt.Fprintf(os.Stderr, "failed to wait for '%s' to terminate: %v", id, err)
+					fmt.Fprintf(os.Stderr, "failed to wait for %q to terminate: %v", id, err)
 				}
 			}
 		}
@@ -249,20 +249,20 @@ func (m *shimManager) Stop(_ context.Context, id string) (resp shim.StopStatus,
 func limitedRead(filePath string, readLimitBytes int64) ([]byte, error) {
 	f, err := os.Open(filePath)
 	if err != nil {
-		return nil, fmt.Errorf("limited read failed to open file: %s: %w", filePath, err)
+		return nil, fmt.Errorf("open file %s: %w", filePath, err)
 	}
 	defer f.Close()
 	fi, err := f.Stat()
 	if err != nil {
-		return []byte{}, fmt.Errorf("limited read failed during file stat: %s: %w", filePath, err)
+		return []byte{}, fmt.Errorf("stat file %s: %w", filePath, err)
 	}
 	if fi.Size() < readLimitBytes {
 		readLimitBytes = fi.Size()
 	}
 	buf := make([]byte, readLimitBytes)
 	_, err = f.Read(buf)
 	if err != nil {
-		return []byte{}, fmt.Errorf("limited read failed during file read: %s: %w", filePath, err)
+		return []byte{}, fmt.Errorf("read file %s: %w", filePath, err)
 	}
 	return buf, nil
 }

cmd/containerd-shim-lcow-v2/service/plugin/plugin.go

@@ -7,6 +7,7 @@ import (
 	"os"
 
 	"github.com/Microsoft/hcsshim/cmd/containerd-shim-lcow-v2/service"
+	"github.com/Microsoft/hcsshim/internal/logfields"
 	"github.com/Microsoft/hcsshim/internal/shim"
 	"github.com/Microsoft/hcsshim/internal/shimdiag"
 	hcsversion "github.com/Microsoft/hcsshim/internal/version"
@@ -30,7 +31,7 @@ const (
 var svc *service.Service
 
 func init() {
-	// Provider ID: 64F6FC7F-8326-5EE8-B890-3734AE584136
+	// Provider ID: A6BD4B70-8A0B-5913-5C8E-E2780DC7F06F
 	// Provider and hook aren't closed explicitly, as they will exist until process exit.
 	provider, err := etw.NewProvider(etwProviderName, etwCallback)
 	if err != nil {
@@ -106,7 +107,7 @@ func etwCallback(sourceID guid.GUID, state etw.ProviderState, level etw.Level, m
 		if err != nil {
 			return
 		}
-		log := logrus.WithField("sandboxID", svc.SandboxID())
+		log := logrus.WithField(logfields.SandboxID, svc.SandboxID())
 		log.WithField("stack", resp.Stacks).Info("goroutine stack dump")
 		if resp.GuestStacks != "" {
 			log.WithField("stack", resp.GuestStacks).Info("guest stack dump")

cmd/containerd-shim-lcow-v2/service/service.go

@@ -4,19 +4,22 @@ package service
 
 import (
 	"context"
+	"fmt"
 	"sync"
 
 	"github.com/Microsoft/hcsshim/internal/builder/vm/lcow"
+	"github.com/Microsoft/hcsshim/internal/controller/pod"
 	"github.com/Microsoft/hcsshim/internal/controller/vm"
 	"github.com/Microsoft/hcsshim/internal/log"
 	"github.com/Microsoft/hcsshim/internal/shim"
 	"github.com/Microsoft/hcsshim/internal/shimdiag"
 
 	sandboxsvc "github.com/containerd/containerd/api/runtime/sandbox/v1"
-	tasksvc "github.com/containerd/containerd/api/runtime/task/v3"
+	tasksvc "github.com/containerd/containerd/api/runtime/task/v2"
 	"github.com/containerd/containerd/v2/core/runtime"
 	"github.com/containerd/containerd/v2/pkg/namespaces"
 	"github.com/containerd/containerd/v2/pkg/shutdown"
+	"github.com/containerd/errdefs"
 	"github.com/containerd/ttrpc"
 )
 
@@ -29,7 +32,7 @@ const (
 // All Service methods (sandbox, task, and shimdiag) operate on this shared struct.
 type Service struct {
 	// mu is used to synchronize access to shared state within the Service.
-	mu sync.Mutex
+	mu sync.RWMutex
 
 	// publisher is used to publish events from the shim to containerd.
 	publisher shim.Publisher
@@ -47,6 +50,14 @@ type Service struct {
 	// vmController is responsible for managing the lifecycle of the underlying utility VM and its associated resources.
 	vmController *vm.Controller
 
+	// podControllers maps podID -> PodController for each active pod.
+	podControllers map[string]*pod.Controller
+
+	// containerPodMapping maps containerID -> podID, allowing callers to look up
+	// which pod a container belongs to and then retrieve the corresponding controller
+	// from podControllers.
+	containerPodMapping map[string]string
+
 	// shutdown manages graceful shutdown operations and allows registration of cleanup callbacks.
 	shutdown shutdown.Service
 }
@@ -56,10 +67,12 @@ var _ shim.TTRPCService = (*Service)(nil)
 // NewService creates a new instance of the Service with the shared state.
 func NewService(ctx context.Context, eventsPublisher shim.Publisher, sd shutdown.Service) *Service {
 	svc := &Service{
-		publisher:    eventsPublisher,
-		events:       make(chan interface{}, 128), // Buffered channel for events
-		vmController: vm.New(),
-		shutdown:     sd,
+		publisher:           eventsPublisher,
+		events:              make(chan interface{}, 128), // Buffered channel for events
+		vmController:        vm.New(),
+		podControllers:      make(map[string]*pod.Controller),
+		containerPodMapping: make(map[string]string),
+		shutdown:            sd,
 	}
 
 	go svc.forward(ctx, eventsPublisher)
@@ -83,23 +96,27 @@ func NewService(ctx context.Context, eventsPublisher shim.Publisher, sd shutdown
 // RegisterTTRPC registers the Task, Sandbox, and ShimDiag TTRPC services on
 // the provided server so that containerd can call into the shim over TTRPC.
 func (s *Service) RegisterTTRPC(server *ttrpc.Server) error {
-	tasksvc.RegisterTTRPCTaskService(server, s)
+	tasksvc.RegisterTaskService(server, s)
 	sandboxsvc.RegisterTTRPCSandboxService(server, s)
 	shimdiag.RegisterShimDiagService(server, s)
 	return nil
 }
 
+// ensureVMRunning returns an error if the VM is not in the running state.
+func (s *Service) ensureVMRunning() error {
+	if state := s.vmController.State(); state != vm.StateRunning {
+		return fmt.Errorf("vm is not running (state: %s): %w", state, errdefs.ErrFailedPrecondition)
+	}
+	return nil
+}
+
 // SandboxID returns the unique identifier for the sandbox managed by this Service.
 func (s *Service) SandboxID() string {
 	return s.sandboxID
 }
 
 // send enqueues an event onto the internal events channel so that it can be
 // forwarded to containerd asynchronously by the forward goroutine.
-//
-// TODO: wire up send() for task events once task lifecycle methods are implemented.
-//
-//nolint:unused
 func (s *Service) send(evt interface{}) {
 	s.events <- evt
 }

cmd/containerd-shim-lcow-v2/service/service_sandbox_internal.go

@@ -124,7 +124,7 @@ func (s *Service) startSandboxInternal(ctx context.Context, request *sandbox.Sta
 		}
 	}
 
-	// VM controller ensures that only once of the Start call goes through.
+	// VM controller ensures that only one of the Start calls goes through.
 	err := s.vmController.StartVM(ctx, &vm.StartOptions{
 		GCSServiceID:        winio.VsockServiceID(prot.LinuxGcsVsockPort),
 		ConfidentialOptions: confidentialOpts,
@@ -264,7 +264,7 @@ func (s *Service) pingSandboxInternal(_ context.Context, _ *sandbox.PingRequest)
 // The sandbox must already be in the stopped state before shutdown is accepted.
 func (s *Service) shutdownSandboxInternal(ctx context.Context, request *sandbox.ShutdownSandboxRequest) (*sandbox.ShutdownSandboxResponse, error) {
 	if s.sandboxID != request.SandboxID {
-		return &sandbox.ShutdownSandboxResponse{}, fmt.Errorf("sandbox ID mismatch, expected %s, got %s", s.sandboxID, request.SandboxID)
+		return nil, fmt.Errorf("sandbox ID mismatch, expected %s, got %s", s.sandboxID, request.SandboxID)
 	}
 
 	// Ensure the VM is terminated. If the VM is already terminated,
@@ -298,17 +298,17 @@ func (s *Service) shutdownSandboxInternal(ctx context.Context, request *sandbox.
 // It collects and returns runtime statistics from the vmController.
 func (s *Service) sandboxMetricsInternal(ctx context.Context, request *sandbox.SandboxMetricsRequest) (*sandbox.SandboxMetricsResponse, error) {
 	if s.sandboxID != request.SandboxID {
-		return &sandbox.SandboxMetricsResponse{}, fmt.Errorf("sandbox ID mismatch, expected %s, got %s", s.sandboxID, request.SandboxID)
+		return nil, fmt.Errorf("sandbox ID mismatch, expected %s, got %s", s.sandboxID, request.SandboxID)
 	}
 
 	stats, err := s.vmController.Stats(ctx)
 	if err != nil {
-		return &sandbox.SandboxMetricsResponse{}, fmt.Errorf("failed to get sandbox metrics: %w", err)
+		return nil, fmt.Errorf("failed to get sandbox metrics: %w", err)
 	}
 
 	anyStat, err := typeurl.MarshalAny(stats)
 	if err != nil {
-		return &sandbox.SandboxMetricsResponse{}, fmt.Errorf("failed to marshal sandbox metrics: %w", err)
+		return nil, fmt.Errorf("failed to marshal sandbox metrics: %w", err)
 	}
 
 	return &sandbox.SandboxMetricsResponse{

cmd/containerd-shim-lcow-v2/service/service_shimdiag.go

@@ -7,11 +7,13 @@ import (
 	"os"
 	"strings"
 
+	"github.com/Microsoft/hcsshim/internal/log"
 	"github.com/Microsoft/hcsshim/internal/logfields"
 	"github.com/Microsoft/hcsshim/internal/oc"
 	"github.com/Microsoft/hcsshim/internal/shimdiag"
 
 	"github.com/containerd/errdefs/pkg/errgrpc"
+	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
 )
 
@@ -34,6 +36,9 @@ func (s *Service) DiagExecInHost(ctx context.Context, request *shimdiag.ExecProc
 		trace.StringAttribute(logfields.Stdout, request.Stdout),
 		trace.StringAttribute(logfields.Stderr, request.Stderr))
 
+	// Set the sandbox ID in the logger context for all subsequent logs in this request.
+	ctx, _ = log.WithContext(ctx, logrus.WithField(logfields.SandboxID, s.sandboxID))
+
 	r, e := s.diagExecInHostInternal(ctx, request)
 	return r, errgrpc.ToGRPC(e)
 }
@@ -49,6 +54,9 @@ func (s *Service) DiagTasks(ctx context.Context, request *shimdiag.TasksRequest)
 		trace.StringAttribute(logfields.SandboxID, s.sandboxID),
 		trace.BoolAttribute(logfields.Execs, request.Execs))
 
+	// Set the sandbox ID in the logger context for all subsequent logs in this request.
+	ctx, _ = log.WithContext(ctx, logrus.WithField(logfields.SandboxID, s.sandboxID))
+
 	r, e := s.diagTasksInternal(ctx, request)
 	return r, errgrpc.ToGRPC(e)
 }
@@ -66,20 +74,26 @@ func (s *Service) DiagShare(ctx context.Context, request *shimdiag.ShareRequest)
 		trace.StringAttribute(logfields.UVMPath, request.UvmPath),
 		trace.BoolAttribute(logfields.ReadOnly, request.ReadOnly))
 
+	// Set the sandbox ID in the logger context for all subsequent logs in this request.
+	ctx, _ = log.WithContext(ctx, logrus.WithField(logfields.SandboxID, s.sandboxID))
+
 	r, e := s.diagShareInternal(ctx, request)
 	return r, errgrpc.ToGRPC(e)
 }
 
 // DiagStacks returns the stack traces of all goroutines in the shim.
 // This method is part of the instrumentation layer and business logic is included in diagStacksInternal.
-func (s *Service) DiagStacks(ctx context.Context, request *shimdiag.StacksRequest) (resp *shimdiag.StacksResponse, err error) {
+func (s *Service) DiagStacks(ctx context.Context, _ *shimdiag.StacksRequest) (resp *shimdiag.StacksResponse, err error) {
 	ctx, span := oc.StartSpan(ctx, "DiagStacks")
 	defer span.End()
 	defer func() { oc.SetSpanStatus(span, err) }()
 
 	span.AddAttributes(trace.StringAttribute(logfields.SandboxID, s.sandboxID))
 
-	r, e := s.diagStacksInternal(ctx, request)
+	// Set the sandbox ID in the logger context for all subsequent logs in this request.
+	ctx, _ = log.WithContext(ctx, logrus.WithField(logfields.SandboxID, s.sandboxID))
+
+	r, e := s.diagStacksInternal(ctx)
 	return r, errgrpc.ToGRPC(e)
 }