engine: refactor session+client management

This commit focuses on making sessions + associated state management
simpler:

* More comprehensible+centralized state management
  * Rather than spread all over the place and tied together in random
    places, all of the state associated with a given session is in a
    daggerSession object and all of the state associated with a given client
    in a session is a daggerClient object
  * The code is also a lot more structured and "boring" in terms of
    locking/mutating state/etc. Not a rube goldberg machine anymore
  * The whole "pre-register a nested client's state before it calls", which
    was a fountain of confusion and bugs, is gone.
* No more insane gRPC tunneling, the engine API is just an HTTP server now
  * graphQL http requests are just that, don't have to tunnel them through
    gRPC streams
  * session attachables are still gRPC based, but over a hijacked http conn
    (as opposed to a gRPC stream embedded in another gRPC stream)
  * This allowed us to move off the Session method from buildkit's upstream
    controller interface
  * That in turn let us delete huge chunks of complicated code around
    handing conns (i.e. engine/server/conn.go) and no longer need to be
    paranoid about gRPC max message limits in as many places

There are more details in the PR description (7315).

Signed-off-by: Erik Sipsma <[email protected]>

Author: sipsma

analytics/analytics.go

@@ -29,8 +29,8 @@ type Event struct {
 	Type       string            `json:"type,omitempty"`
 	Properties map[string]string `json:"properties,omitempty"`
 
-	DeviceID string `json:"device_id,omitempty"`
-	ServerID string `json:"server_id,omitempty"`
+	DeviceID  string `json:"device_id,omitempty"`
+	SessionID string `json:"session_id,omitempty"`
 
 	ClientVersion string `json:"client_version,omitempty"`
 	ClientOS      string `json:"client_os,omitempty"`
@@ -155,7 +155,7 @@ func (t *CloudTracker) Capture(ctx context.Context, event string, properties map
 		ev.GitAuthorHashed = fmt.Sprintf("%x", sha256.Sum256([]byte(author)))
 	}
 	if clientMetadata, err := engine.ClientMetadataFromContext(ctx); err == nil {
-		ev.ServerID = clientMetadata.ServerID
+		ev.SessionID = clientMetadata.SessionID
 	}
 
 	t.queue = append(t.queue, &queuedEvent{ctx: ctx, event: ev})

cmd/dagger/listen.go

@@ -11,6 +11,8 @@ import (
 	"github.com/rs/cors"
 	"github.com/spf13/cobra"
 	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
 
 	"dagger.io/dagger"
 	"github.com/dagger/dagger/engine/client"
@@ -46,20 +48,29 @@ func Listen(ctx context.Context, engineClient *client.Client, _ *dagger.Module,
 	defer sessionL.Close()
 
 	var handler http.Handler = engineClient
+
 	if allowCORS {
 		handler = cors.AllowAll().Handler(handler)
 	}
 
+	handler = otelhttp.NewHandler(handler, "listen", otelhttp.WithSpanNameFormatter(func(o string, r *http.Request) string {
+		return fmt.Sprintf("%s: HTTP %s %s", o, r.Method, r.URL.Path)
+	}))
+
+	http2Srv := &http2.Server{}
+	handler = h2c.NewHandler(handler, http2Srv)
+
 	srv := &http.Server{
-		Handler: otelhttp.NewHandler(handler, "listen", otelhttp.WithSpanNameFormatter(func(o string, r *http.Request) string {
-			return fmt.Sprintf("%s: HTTP %s %s", o, r.Method, r.URL.Path)
-		})),
+		Handler: handler,
 		// Gosec G112: prevent slowloris attacks
 		ReadHeaderTimeout: 10 * time.Second,
 		BaseContext: func(_ net.Listener) context.Context {
 			return ctx
 		},
 	}
+	if err := http2.ConfigureServer(srv, http2Srv); err != nil {
+		return fmt.Errorf("http2 server configuration: %w", err)
+	}
 
 	go func() {
 		<-ctx.Done()

cmd/dagger/shell.go

@@ -58,7 +58,7 @@ func (ts *terminalSession) Run() error {
 		NetDialContext: ts.Client.DialContext,
 	}
 
-	reqHeader := http.Header{}
+	reqHeader := ts.Client.AppendHTTPRequestHeaders(http.Header{})
 	if ts.Client.SecretToken != "" {
 		reqHeader["Authorization"] = []string{"Basic " + base64.StdEncoding.EncodeToString([]byte(ts.Client.SecretToken+":"))}
 	}

cmd/engine/debug.go

@@ -83,7 +83,7 @@ func logTraceMetrics(ctx context.Context) {
 	}
 }
 
-func logMetrics(ctx context.Context, engineStateRootDir string, eng *server.Engine) {
+func logMetrics(ctx context.Context, engineStateRootDir string, eng *server.Server) {
 	for range time.Tick(60 * time.Second) {
 		l := bklog.G(ctx)
 

cmd/engine/main.go

@@ -4,10 +4,11 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
-	goerrors "errors"
+	"errors"
 	"fmt"
 	"io"
 	"net"
+	"net/http"
 	"os"
 	"os/exec"
 	"os/user"
@@ -30,11 +31,12 @@ import (
 	"github.com/moby/buildkit/util/profiler"
 	"github.com/moby/buildkit/util/stack"
 	"github.com/moby/buildkit/version"
-	"github.com/pkg/errors"
 	sloglogrus "github.com/samber/slog-logrus/v2"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"golang.org/x/net/http2"
+	"golang.org/x/net/http2/h2c"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 
@@ -339,18 +341,18 @@ func main() { //nolint:gocyclo
 		cfg.Root = root
 
 		if err := os.MkdirAll(root, 0o700); err != nil {
-			return errors.Wrapf(err, "failed to create %s", root)
+			return fmt.Errorf("failed to create %s: %w", root, err)
 		}
 
 		bklog.G(ctx).Debug("creating engine lockfile")
 		lockPath := filepath.Join(root, "buildkitd.lock")
 		lock := flock.New(lockPath)
 		locked, err := lock.TryLock()
 		if err != nil {
-			return errors.Wrapf(err, "could not lock %s", lockPath)
+			return fmt.Errorf("could not lock %s: %w", lockPath, err)
 		}
 		if !locked {
-			return errors.Errorf("could not lock %s, another instance running?", lockPath)
+			return fmt.Errorf("could not lock %s, another instance running?", lockPath)
 		}
 		defer func() {
 			lock.Unlock()
@@ -367,7 +369,7 @@ func main() { //nolint:gocyclo
 				case "network.host":
 					cfg.Entitlements = append(cfg.Entitlements, e)
 				default:
-					return errors.Errorf("invalid entitlement : %s", e)
+					return fmt.Errorf("invalid entitlement : %s", e)
 				}
 			}
 		}
@@ -381,9 +383,7 @@ func main() { //nolint:gocyclo
 		if err != nil {
 			return fmt.Errorf("failed to create engine: %w", err)
 		}
-		defer eng.Close()
-
-		eng.Register(grpcServer)
+		defer srv.Close()
 
 		go logMetrics(context.Background(), cfg.Root, srv)
 		if cfg.Trace {
@@ -398,9 +398,26 @@ func main() { //nolint:gocyclo
 		}
 
 		// start serving on the listeners for actual clients
-		bklog.G(ctx).Debug("starting main engine grpc listeners")
+		bklog.G(ctx).Debug("starting main engine api listeners")
+		srv.Register(grpcServer)
+		http2Server := &http2.Server{}
+		httpServer := &http.Server{
+			ReadHeaderTimeout: 30 * time.Second,
+			Handler: h2c.NewHandler(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				if r.ProtoMajor == 2 && strings.HasPrefix(r.Header.Get("content-type"), "application/grpc") {
+					// The docs on grpcServer.ServeHTTP warn that some features are missing vs. serving fully "native" gRPC,
+					// but in practice it seems to work fine for us and only be relevant for some advanced features we don't use.
+					grpcServer.ServeHTTP(w, r)
+					return
+				}
+				srv.ServeHTTP(w, r)
+			}), http2Server),
+		}
+		if err := http2.ConfigureServer(httpServer, http2Server); err != nil {
+			return fmt.Errorf("failed to configure http2 server: %w", err)
+		}
 		errCh := make(chan error, 1)
-		if err := serveGRPC(cfg.GRPC, grpcServer, errCh); err != nil {
+		if err := serveAPI(cfg.GRPC, httpServer, errCh); err != nil {
 			return err
 		}
 
@@ -424,7 +441,7 @@ func main() { //nolint:gocyclo
 		if stopCacheErr != nil {
 			bklog.G(ctx).WithError(stopCacheErr).Error("failed to stop cache")
 		}
-		err = goerrors.Join(err, stopCacheErr)
+		err = errors.Join(err, stopCacheErr)
 		cancelNetworking()
 
 		bklog.G(ctx).Infof("stopping server")
@@ -449,7 +466,11 @@ func main() { //nolint:gocyclo
 	}
 }
 
-func serveGRPC(cfg config.GRPCConfig, server *grpc.Server, errCh chan error) error {
+func serveAPI(
+	cfg config.GRPCConfig,
+	httpServer *http.Server,
+	errCh chan error,
+) error {
 	addrs := cfg.Address
 	if len(addrs) == 0 {
 		return errors.New("--addr cannot be empty")
@@ -480,7 +501,12 @@ func serveGRPC(cfg config.GRPCConfig, server *grpc.Server, errCh chan error) err
 			eg.Go(func() error {
 				defer l.Close()
 				logrus.Infof("running server on %s", l.Addr())
-				return server.Serve(l)
+
+				err := httpServer.Serve(l)
+				if err != nil {
+					return fmt.Errorf("serve: %w", err)
+				}
+				return nil
 			})
 		}(l)
 	}
@@ -616,7 +642,7 @@ func applyMainFlags(c *cli.Context, cfg *config.Config) error {
 		} else {
 			maxParallelism, err = strconv.Atoi(maxParallelismStr)
 			if err != nil {
-				return errors.Wrap(err, "failed to parse oci-max-parallelism, should be positive integer, 0 for unlimited, or 'num-cpu' for setting to the number of CPUs")
+				return fmt.Errorf("failed to parse oci-max-parallelism, should be positive integer, 0 for unlimited, or 'num-cpu' for setting to the number of CPUs: %w", err)
 			}
 		}
 		cfg.Workers.OCI.MaxParallelism = maxParallelism
@@ -661,7 +687,7 @@ func grouptoGID(group string) (int, error) {
 func getListener(addr string, uid, gid int, tlsConfig *tls.Config) (net.Listener, error) {
 	addrSlice := strings.SplitN(addr, "://", 2)
 	if len(addrSlice) < 2 {
-		return nil, errors.Errorf("address %s does not contain proto, you meant unix://%s ?",
+		return nil, fmt.Errorf("address %s does not contain proto, you meant unix://%s ?",
 			addr, addr)
 	}
 	proto := addrSlice[0]
@@ -686,7 +712,7 @@ func getListener(addr string, uid, gid int, tlsConfig *tls.Config) (net.Listener
 		}
 		return tls.NewListener(l, tlsConfig), nil
 	default:
-		return nil, errors.Errorf("addr %s not supported", addr)
+		return nil, fmt.Errorf("addr %s not supported", addr)
 	}
 }
 
@@ -706,7 +732,7 @@ func serverCredentials(cfg config.TLSConfig) (*tls.Config, error) {
 	}
 	certificate, err := tls.LoadX509KeyPair(certFile, keyFile)
 	if err != nil {
-		return nil, errors.Wrap(err, "could not load server key pair")
+		return nil, fmt.Errorf("could not load server key pair: %w", err)
 	}
 	tlsConf := &tls.Config{
 		Certificates: []tls.Certificate{certificate},
@@ -716,7 +742,7 @@ func serverCredentials(cfg config.TLSConfig) (*tls.Config, error) {
 		certPool := x509.NewCertPool()
 		ca, err := os.ReadFile(caFile)
 		if err != nil {
-			return nil, errors.Wrap(err, "could not read ca certificate")
+			return nil, fmt.Errorf("could not read ca certificate: %w", err)
 		}
 		// Append the client certificates from the CA
 		if ok := certPool.AppendCertsFromPEM(ca); !ok {
@@ -733,7 +759,7 @@ func attrMap(sl []string) (map[string]string, error) {
 	for _, v := range sl {
 		parts := strings.SplitN(v, "=", 2)
 		if len(parts) != 2 {
-			return nil, errors.Errorf("invalid value %s", v)
+			return nil, fmt.Errorf("invalid value %s", v)
 		}
 		m[parts[0]] = parts[1]
 	}

core/container.go

@@ -301,7 +301,7 @@ func (container *Container) From(ctx context.Context, addr string) (*Container,
 		},
 	})
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to resolve image %s: %w", ref, err)
 	}
 
 	digested, err := reference.WithDigest(refName, digest)

core/container_exec.go

@@ -12,10 +12,8 @@ import (
 	"github.com/dagger/dagger/engine"
 	"github.com/dagger/dagger/engine/buildkit"
 	"github.com/moby/buildkit/client/llb"
+	"github.com/moby/buildkit/identity"
 	"github.com/pkg/errors"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/propagation"
-	"go.opentelemetry.io/otel/trace"
 )
 
 type ContainerExecOpts struct {
@@ -41,9 +39,8 @@ type ContainerExecOpts struct {
 	// Grant the process all root capabilities
 	InsecureRootCapabilities bool `default:"false"`
 
-	// (Internal-only) If this is a nested exec for a Function call, this should be set
-	// with the metadata for that call
-	NestedExecFunctionCall *FunctionCall `name:"-"`
+	// (Internal-only) If this is a nested exec, exec metadata to use for it
+	NestedExecMetadata *buildkit.ExecutionMetadata `name:"-"`
 }
 
 func (container *Container) WithExec(ctx context.Context, opts ContainerExecOpts) (*Container, error) { //nolint:gocyclo
@@ -72,18 +69,18 @@ func (container *Container) WithExec(ctx context.Context, opts ContainerExecOpts
 		return nil, err
 	}
 
-	execMD := buildkit.ExecutionMetadata{
-		ServerID: clientMetadata.ServerID,
-
-		HostAliases: make(map[string][]string),
-
-		RedirectStdoutPath: opts.RedirectStdout,
-		RedirectStderrPath: opts.RedirectStderr,
-
-		SystemEnvNames: container.SystemEnvNames,
-
-		EnabledGPUs: container.EnabledGPUs,
+	execMD := buildkit.ExecutionMetadata{}
+	if opts.NestedExecMetadata != nil {
+		execMD = *opts.NestedExecMetadata
+	}
+	execMD.SessionID = clientMetadata.SessionID
+	if execMD.HostAliases == nil {
+		execMD.HostAliases = make(map[string][]string)
 	}
+	execMD.RedirectStdoutPath = opts.RedirectStdout
+	execMD.RedirectStderrPath = opts.RedirectStderr
+	execMD.SystemEnvNames = container.SystemEnvNames
+	execMD.EnabledGPUs = container.EnabledGPUs
 
 	// if GPU parameters are set for this container pass them over:
 	if len(execMD.EnabledGPUs) > 0 {
@@ -94,44 +91,27 @@ func (container *Container) WithExec(ctx context.Context, opts ContainerExecOpts
 
 	// this allows executed containers to communicate back to this API
 	if opts.ExperimentalPrivilegedNesting {
-		callerOpts := opts.NestedExecFunctionCall
-		if callerOpts == nil {
-			// default to caching the nested exec
-			callerOpts = &FunctionCall{
-				Cache: true,
-			}
-		}
-
 		// directing telemetry to another span (i.e. a function call).
-		if callerOpts.SpanContext.IsValid() {
-			execMD.SpanContext = propagation.MapCarrier{}
-			otel.GetTextMapPropagator().Inject(
-				trace.ContextWithSpanContext(ctx, callerOpts.SpanContext),
-				execMD.SpanContext,
-			)
-
+		if execMD.SpanContext != nil {
 			// hide the exec span
 			spanName = buildkit.InternalPrefix + spanName
 		}
 
-		execMD.ClientID, err = container.Query.RegisterCaller(ctx, callerOpts)
-		if err != nil {
-			return nil, fmt.Errorf("register caller: %w", err)
-		}
+		execMD.ClientID = identity.NewID()
 
 		// include the engine version so that these execs get invalidated if the engine/API change
 		runOpts = append(runOpts, llb.AddEnv(buildkit.DaggerEngineVersionEnv, engine.Version))
 
 		// include a digest of the current call so that we scope of the cache of the ExecOp to this call
 		runOpts = append(runOpts, llb.AddEnv(buildkit.DaggerCallDigestEnv, string(dagql.CurrentID(ctx).Digest())))
 
-		if !callerOpts.Cache {
-			// include the ServerID here so that we bust cache once-per-session
+		if execMD.CachePerSession {
+			// include the SessionID here so that we bust cache once-per-session
 			clientMetadata, err := engine.ClientMetadataFromContext(ctx)
 			if err != nil {
 				return nil, err
 			}
-			runOpts = append(runOpts, llb.AddEnv(buildkit.DaggerServerIDEnv, clientMetadata.ServerID))
+			runOpts = append(runOpts, llb.AddEnv(buildkit.DaggerSessionIDEnv, clientMetadata.SessionID))
 		}
 	}
 

core/git.go

@@ -105,5 +105,5 @@ func (ref *GitRef) getState(ctx context.Context) (llb.State, error) {
 		return llb.State{}, err
 	}
 
-	return gitdns.Git(ref.Repo.URL, ref.Ref, clientMetadata.ServerID, opts...), nil
+	return gitdns.Git(ref.Repo.URL, ref.Ref, clientMetadata.SessionID, opts...), nil
 }