it compiles!

Protobuild.toml

@@ -3,11 +3,9 @@ generators = ["go", "go-grpc"]
 
 # Control protoc include paths.
 [includes]
-  before = ["./protobuf"]
-
   # defaults are "/usr/local/include" and "/usr/include", which don't exist on Windows.
   # override defaults to supress errors about non-existant directories.
-  after = []
+  # after = []
 
 # This section maps protobuf imports to Go packages.
 [packages]
@@ -21,5 +19,20 @@ prefixes = [
   "github.com/Microsoft/hcsshim/internal/computeagent",
   "github.com/Microsoft/hcsshim/internal/ncproxyttrpc",
   "github.com/Microsoft/hcsshim/internal/vmservice",
+  "github.com/Microsoft/hcsshim/internal/save",
+]
+generators = ["go", "go-ttrpc"]
+
+[[overrides]]
+prefixes = [
+  "github.com/Microsoft/hcsshim/internal/lm/proto",
 ]
 generators = ["go", "go-ttrpc"]
+# Override the go package path for the generated code. This allows us to not need a go_package
+# directive in the proto file itself, which is nice so we can share it with other projects easily.
+parameters.go."Mgithub.com/Microsoft/hcsshim/internal/lm/proto/lm.proto" = '"github.com/Microsoft/hcsshim/internal/lm/proto;lmproto"'
+parameters.go-ttrpc."Mgithub.com/Microsoft/hcsshim/internal/lm/proto/lm.proto" = '"github.com/Microsoft/hcsshim/internal/lm/proto;lmproto"'
+
+[[overrides]]
+prefixes = ["github.com/Microsoft/hcsshim/internal/state"]
+generators = ["go"]

cmd/containerd-shim-runhcs-v1/main.go

@@ -3,7 +3,6 @@
 package main
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -20,7 +19,6 @@ import (
 
 	"github.com/Microsoft/hcsshim/internal/log"
 	"github.com/Microsoft/hcsshim/internal/oc"
-	"github.com/Microsoft/hcsshim/internal/shimdiag"
 	hcsversion "github.com/Microsoft/hcsshim/internal/version"
 
 	// register common types spec with typeurl
@@ -55,17 +53,17 @@ var (
 )
 
 func etwCallback(sourceID guid.GUID, state etw.ProviderState, level etw.Level, matchAnyKeyword uint64, matchAllKeyword uint64, filterData uintptr) {
-	if state == etw.ProviderStateCaptureState {
-		resp, err := svc.DiagStacks(context.Background(), &shimdiag.StacksRequest{})
-		if err != nil {
-			return
-		}
-		log := logrus.WithField("tid", svc.tid)
-		log.WithField("stack", resp.Stacks).Info("goroutine stack dump")
-		if resp.GuestStacks != "" {
-			log.WithField("stack", resp.GuestStacks).Info("guest stack dump")
-		}
-	}
+	// if state == etw.ProviderStateCaptureState {
+	// 	resp, err := svc.DiagStacks(context.Background(), &shimdiag.StacksRequest{})
+	// 	if err != nil {
+	// 		return
+	// 	}
+	// 	log := logrus.WithField("tid", svc.tid)
+	// 	log.WithField("stack", resp.Stacks).Info("goroutine stack dump")
+	// 	if resp.GuestStacks != "" {
+	// 		log.WithField("stack", resp.GuestStacks).Info("guest stack dump")
+	// 	}
+	// }
 }
 
 func main() {

cmd/containerd-shim-runhcs-v1/options/runhcs.proto

@@ -107,6 +107,15 @@ message Options {
 
 	// scrub_logs enables removing environment variables and other potentially sensitive information from logs
 	bool scrub_logs = 20;
+
+	BundleType bundle_type = 22;
+}
+
+enum BundleType {
+	BUNDLE_OCI = 0;
+	BUNDLE_POD_RESTORE = 1;
+	BUNDLE_CONTAINER_RESTORE = 2;
+	BUNDLE_POD_LM = 3;
 }
 
 // ProcessDetails contains additional information about a process. This is the additional

cmd/containerd-shim-runhcs-v1/serve.go

@@ -24,13 +24,15 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	runhcsopts "github.com/Microsoft/hcsshim/cmd/containerd-shim-runhcs-v1/options"
-	"github.com/Microsoft/hcsshim/internal/extendedtask"
+	"github.com/Microsoft/hcsshim/internal/ctrdpub"
 	hcslog "github.com/Microsoft/hcsshim/internal/log"
 	"github.com/Microsoft/hcsshim/internal/shimdiag"
+	"github.com/Microsoft/hcsshim/internal/taskserver"
+	"github.com/Microsoft/hcsshim/internal/winapi"
 	"github.com/Microsoft/hcsshim/pkg/octtrpc"
 )
 
-var svc *service
+var svc task.TaskService
 
 var serveCommand = cli.Command{
 	Name:           "serve",
@@ -47,6 +49,17 @@ var serveCommand = cli.Command{
 		},
 	},
 	Action: func(ctx *cli.Context) error {
+
+		_, err := os.Stat(`c:\debugwait`)
+		if err == nil {
+			for {
+				if winapi.IsDebuggerPresent() {
+					break
+				}
+				time.Sleep(1 * time.Second)
+			}
+		}
+
 		// On Windows the serve command is internally used to actually create
 		// the process that hosts the containerd/ttrpc entrypoint to the Runtime
 		// V2 API's. The model requires this 2nd invocation of the shim process
@@ -175,20 +188,27 @@ var serveCommand = cli.Command{
 		}
 
 		ttrpcAddress := os.Getenv(ttrpcAddressEnv)
-		ttrpcEventPublisher, err := newEventPublisher(ttrpcAddress, namespaceFlag)
+		// ttrpcEventPublisher, err := newEventPublisher(ttrpcAddress, namespaceFlag)
+		// if err != nil {
+		// 	return err
+		// }
+		// defer func() {
+		// 	if err != nil {
+		// 		ttrpcEventPublisher.close()
+		// 	}
+		// }()
+		publisher, err := ctrdpub.NewPublisher(ttrpcAddress, namespaceFlag)
 		if err != nil {
 			return err
 		}
-		defer func() {
-			if err != nil {
-				ttrpcEventPublisher.close()
-			}
-		}()
 
 		// Setup the ttrpc server
-		svc, err = NewService(WithEventPublisher(ttrpcEventPublisher),
-			WithTID(idFlag),
-			WithIsSandbox(ctx.Bool("is-sandbox")))
+		closeCh := make(chan struct{})
+		coreSvc := taskserver.NewService(closeCh, publisher)
+		taskSvc := taskserver.NewInstrumentedService(coreSvc)
+		// svc, err = NewService(WithEventPublisher(ttrpcEventPublisher),
+		// 	WithTID(idFlag),
+		// 	WithIsSandbox(ctx.Bool("is-sandbox")))
 		if err != nil {
 			return fmt.Errorf("failed to create new service: %w", err)
 		}
@@ -198,9 +218,10 @@ var serveCommand = cli.Command{
 			return err
 		}
 		defer s.Close()
-		task.RegisterTaskService(s, svc)
-		shimdiag.RegisterShimDiagService(s, svc)
-		extendedtask.RegisterExtendedTaskService(s, svc)
+		task.RegisterTaskService(s, taskSvc)
+		shimdiag.RegisterShimDiagService(s, coreSvc)
+		// extendedtask.RegisterExtendedTaskService(s, svc)
+		// save.RegisterSaveService(s, svc)
 
 		sl, err := winio.ListenPipe(socket, nil)
 		if err != nil {
@@ -246,12 +267,12 @@ var serveCommand = cli.Command{
 		select {
 		case err = <-serrs:
 			// the ttrpc server shutdown without processing a shutdown request
-		case <-svc.Done():
-			if !svc.gracefulShutdown {
-				// Return immediately, but still close ttrpc server, pipes, and spans
-				// Shouldn't need to os.Exit without clean up (ie, deferred `.Close()`s)
-				return nil
-			}
+		case <-closeCh:
+			// if !svc.gracefulShutdown {
+			// 	// Return immediately, but still close ttrpc server, pipes, and spans
+			// 	// Shouldn't need to os.Exit without clean up (ie, deferred `.Close()`s)
+			// 	return nil
+			// }
 			// currently the ttrpc shutdown is the only clean up to wait on
 			sctx, cancel := context.WithTimeout(context.Background(), gracefulShutdownTimeout)
 			defer cancel()

cmd/gcs/main.go

@@ -4,25 +4,29 @@
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"runtime"
 	"syscall"
 	"time"
 
 	"github.com/cenkalti/backoff/v4"
 	cgroups "github.com/containerd/cgroups/v3/cgroup1"
 	cgroupstats "github.com/containerd/cgroups/v3/cgroup1/stats"
+	"github.com/linuxkit/virtsock/pkg/vsock"
 	oci "github.com/opencontainers/runtime-spec/specs-go"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"go.opencensus.io/trace"
 
 	"github.com/Microsoft/hcsshim/internal/guest/bridge"
 	"github.com/Microsoft/hcsshim/internal/guest/kmsg"
+	"github.com/Microsoft/hcsshim/internal/guest/reconn"
 	"github.com/Microsoft/hcsshim/internal/guest/runtime/hcsv2"
 	"github.com/Microsoft/hcsshim/internal/guest/runtime/runc"
 	"github.com/Microsoft/hcsshim/internal/guest/transport"
@@ -182,9 +186,9 @@ func main() {
 		"",
 		"Logging Target: An optional file name/path. Omit for console output.")
 	logFormat := flag.String("log-format", "text", "Logging Format: text or json")
-	useInOutErr := flag.Bool("use-inouterr",
-		false,
-		"If true use stdin/stdout for bridge communication and stderr for logging")
+	// useInOutErr := flag.Bool("use-inouterr",
+	// 	false,
+	// 	"If true use stdin/stdout for bridge communication and stderr for logging")
 	v4 := flag.Bool("v4", false, "enable the v4 protocol support and v2 schema")
 	rootMemReserveBytes := flag.Uint64("root-mem-reserve-bytes",
 		75*1024*1024, // 75Mib
@@ -218,7 +222,8 @@ func main() {
 
 	logrus.AddHook(log.NewHook())
 
-	var logWriter *os.File
+	tport := &transport.VsockTransport{}
+	var logWriter io.Writer = os.Stderr
 	if *logFile != "" {
 		logFileHandle, err := os.OpenFile(*logFile, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
 		if err != nil {
@@ -229,8 +234,35 @@ func main() {
 		}
 		logWriter = logFileHandle
 	} else {
+		l := logrus.New()
+		l.SetOutput(os.Stderr)
 		// logrus uses os.Stderr. see logrus.New()
-		logWriter = os.Stderr
+		d := func(ctx context.Context) (reconn.Conn, error) {
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
+			select {
+			case <-ctx.Done():
+				return nil, ctx.Err()
+			default:
+				l.Info("dial attempt starting")
+				c, err := vsock.Dial(vsock.CIDHost, 109)
+				l.WithError(err).Info("dial attempt complete")
+				return c, err
+			}
+		}
+		c, err := d(context.Background())
+		if err != nil {
+			l.WithError(err).Fatal("dial log pipe")
+		}
+		logWriter = reconn.NewPipe(
+			d,
+			c,
+			backoff.NewConstantBackOff(5*time.Second),
+			func(err error) bool {
+				l.WithError(err).Warn("log pipe disconnected")
+				return true
+			},
+		)
 	}
 
 	// set up our initial stance policy enforcer
@@ -294,36 +326,12 @@ func main() {
 	// Continuously log /dev/kmsg
 	go kmsg.ReadForever(kmsg.LogLevel(*kmsgLogLevel))
 
-	tport := &transport.VsockTransport{}
 	rtime, err := runc.NewRuntime(baseLogPath)
 	if err != nil {
 		logrus.WithError(err).Fatal("failed to initialize new runc runtime")
 	}
 	mux := bridge.NewBridgeMux()
-	b := bridge.Bridge{
-		Handler:  mux,
-		EnableV4: *v4,
-	}
 	h := hcsv2.NewHost(rtime, tport, initialEnforcer, logWriter)
-	b.AssignHandlers(mux, h)
-
-	var bridgeIn io.ReadCloser
-	var bridgeOut io.WriteCloser
-	if *useInOutErr {
-		bridgeIn = os.Stdin
-		bridgeOut = os.Stdout
-	} else {
-		const commandPort uint32 = 0x40000000
-		bridgeCon, err := tport.Dial(commandPort)
-		if err != nil {
-			logrus.WithFields(logrus.Fields{
-				"port":          commandPort,
-				logrus.ErrorKey: err,
-			}).Fatal("failed to dial host vsock connection")
-		}
-		bridgeIn = bridgeCon
-		bridgeOut = bridgeCon
-	}
 
 	// Setup the UVM cgroups to protect against a workload taking all available
 	// memory and causing the GCS to malfunction we create two cgroups: gcs,
@@ -390,10 +398,32 @@ func main() {
 
 	go readMemoryEvents(startTime, gefdFile, "/gcs", int64(*gcsMemLimitBytes), gcsControl)
 	go readMemoryEvents(startTime, oomFile, "/containers", containersLimit, containersControl)
-	err = b.ListenAndServe(bridgeIn, bridgeOut)
-	if err != nil {
-		logrus.WithFields(logrus.Fields{
-			logrus.ErrorKey: err,
-		}).Fatal("failed to serve gcs service")
+
+	var p bridge.Publisher
+	for {
+		const commandPort uint32 = 0x40000000
+		bridgeCon, err := tport.Dial(commandPort)
+		if err != nil {
+			logrus.WithFields(logrus.Fields{
+				"port":          commandPort,
+				logrus.ErrorKey: err,
+			}).Error("failed to dial host vsock connection")
+		}
+		b := bridge.Bridge{
+			Handler:   mux,
+			EnableV4:  *v4,
+			Publisher: &p,
+		}
+		b.AssignHandlers(mux, h)
+		p.SetBridge(&b)
+		err = b.ListenAndServe(bridgeCon, bridgeCon)
+		if err != nil {
+			logrus.WithFields(logrus.Fields{
+				logrus.ErrorKey: err,
+			}).Error("failed to serve gcs service")
+		}
+		time.Sleep(3 * time.Second)
 	}
+
+	runtime.KeepAlive(logWriter)
 }