docker: use native gRPC for supported local endpoints

driver/docker/driver.go

@@ -5,12 +5,18 @@ import (
 	"net"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/docker/buildx/driver"
 	"github.com/docker/buildx/util/progress"
 	"github.com/moby/buildkit/client"
+	"github.com/moby/buildkit/util/flightcontrol"
+	"github.com/moby/buildkit/util/grpcerrors"
 	dockerclient "github.com/moby/moby/client"
+	"github.com/moby/moby/client/pkg/versions"
 	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc/codes"
 )
 
 type Driver struct {
@@ -21,6 +27,7 @@ type Driver struct {
 	// https://github.com/docker/docs/blob/main/content/build/drivers/docker.md
 	features    features
 	hostGateway hostGateway
+	nativeGRPC  flightcontrol.CachedGroup[bool]
 }
 
 func (d *Driver) Bootstrap(ctx context.Context, l progress.Logger) error {
@@ -64,16 +71,81 @@ func (d *Driver) Dial(ctx context.Context) (net.Conn, error) {
 }
 
 func (d *Driver) Client(ctx context.Context, opts ...client.ClientOpt) (*client.Client, error) {
+	// TODO: Support native gRPC with Desktop Resource Saver metadata. Keep /grpc
+	//  so Desktop's proxy can identify background connections and let the VM sleep.
+	//  See https://github.com/docker/desktop-build/pull/312.
+	if len(d.DialMeta) == 0 {
+		if err := context.Cause(ctx); err != nil {
+			return nil, err
+		}
+		native, err := d.nativeGRPC.Do(ctx, "", d.probeNativeGRPC)
+		if cause := context.Cause(ctx); cause != nil {
+			return nil, cause
+		}
+		if err == nil && native {
+			return client.New(ctx, d.DockerAPI.DaemonHost(), append(d.nativeClientOpts(), opts...)...)
+		}
+		if err != nil {
+			switch grpcerrors.Code(err) {
+			case codes.PermissionDenied, codes.Unauthenticated:
+				return nil, err
+			}
+			logrus.Debugf("docker driver: native gRPC unavailable, using /grpc: %v", err)
+		}
+	}
+
+	// Legacy transport for daemons and proxies without native gRPC:
+	// https://github.com/moby/moby/pull/50744
 	opts = append([]client.ClientOpt{
-		client.WithContextDialer(func(context.Context, string) (net.Conn, error) {
+		client.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
 			return d.Dial(ctx)
-		}), client.WithSessionDialer(func(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error) {
+		}),
+		client.WithSessionDialer(func(ctx context.Context, proto string, meta map[string][]string) (net.Conn, error) {
 			return d.DockerAPI.DialHijack(ctx, "/session", proto, meta)
 		}),
 	}, opts...)
 	return client.New(ctx, "", opts...)
 }
 
+func (d *Driver) probeNativeGRPC(ctx context.Context) (bool, error) {
+	// Remote endpoints keep Docker's configured transport, including TLS and SSH.
+	scheme, _, _ := strings.Cut(d.DockerAPI.DaemonHost(), "://")
+	if scheme != "unix" && scheme != "npipe" {
+		return false, nil
+	}
+
+	ctx, cancel := context.WithTimeoutCause(ctx, 10*time.Second, errors.New("native gRPC probe timed out"))
+	defer cancel()
+
+	ping, err := d.DockerAPI.Ping(ctx, dockerclient.PingOptions{})
+	if err != nil {
+		return false, err
+	}
+	// Engine 29.2 (API 1.53) introduced native gRPC; proxies may still reject it.
+	if ping.APIVersion == "" || versions.LessThan(ping.APIVersion, "1.53") {
+		return false, nil
+	}
+	c, err := client.New(ctx, d.DockerAPI.DaemonHost(), d.nativeClientOpts()...)
+	if err != nil {
+		return false, err
+	}
+	defer c.Close()
+	if _, err := c.ListWorkers(ctx); err != nil {
+		return false, err
+	}
+	logrus.Debug("docker driver: using native gRPC")
+	return true, nil
+}
+
+func (d *Driver) nativeClientOpts() []client.ClientOpt {
+	dial := d.DockerAPI.Dialer()
+	return []client.ClientOpt{
+		client.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
+			return dial(ctx)
+		}),
+	}
+}
+
 type features struct {
 	once sync.Once
 	list map[driver.Feature]bool

tests/docker.go

@@ -0,0 +1,89 @@
+package tests
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/containerd/continuity/fs/fstest"
+	"github.com/docker/buildx/driver"
+	dockerdriver "github.com/docker/buildx/driver/docker"
+	"github.com/docker/buildx/util/dockerutil"
+	"github.com/docker/cli/cli/command"
+	cliflags "github.com/docker/cli/cli/flags"
+	"github.com/moby/buildkit/util/testutil/integration"
+	"github.com/moby/moby/client/pkg/versions"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+)
+
+var dockerTests = []func(t *testing.T, sb integration.Sandbox){
+	testDockerNativeGRPC,
+	testDockerNativeGRPCClients,
+}
+
+func testDockerNativeGRPC(t *testing.T, sb integration.Sandbox) {
+	if !isDockerWorker(sb) {
+		t.Skip("only testing with docker worker")
+	}
+	out, err := dockerCmd(sb, withArgs("version", "--format", "{{.Server.APIVersion}}")).Output()
+	require.NoError(t, err)
+	native := !versions.LessThan(strings.TrimSpace(string(out)), "1.53")
+	dir := tmpdir(t,
+		fstest.CreateFile("Dockerfile", []byte("FROM scratch\nCOPY data /data\n"), 0600),
+		fstest.CreateFile("data", []byte(t.Name()), 0600),
+	)
+	dest := t.TempDir()
+	buildOut, err := buildCmd(sb, withArgs("--debug", "--output=type=local,dest="+dest, dir))
+	require.NoError(t, err, buildOut)
+	data, err := os.ReadFile(filepath.Join(dest, "data"))
+	require.NoError(t, err)
+	require.Equal(t, t.Name(), string(data))
+	if native {
+		require.Contains(t, buildOut, "docker driver: using native gRPC")
+	} else {
+		require.NotContains(t, buildOut, "docker driver: using native gRPC")
+	}
+}
+
+func testDockerNativeGRPCClients(t *testing.T, sb integration.Sandbox) {
+	if !isDockerWorker(sb) {
+		t.Skip("only testing with docker worker")
+	}
+	cli, err := command.NewDockerCli()
+	require.NoError(t, err)
+	opts := cliflags.NewClientOptions()
+	opts.Context = sb.DockerAddress()
+	require.NoError(t, cli.Initialize(opts))
+	api, err := dockerutil.NewClientAPI(cli, sb.DockerAddress())
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, api.Close()) })
+	d := &dockerdriver.Driver{InitConfig: driver.InitConfig{DockerAPI: api}}
+	ctx, cancel := context.WithCancelCause(t.Context())
+	cancel(context.Canceled)
+	_, err = d.Client(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+
+	ctx, stop := context.WithTimeoutCause(t.Context(), 30*time.Second, context.DeadlineExceeded)
+	defer stop()
+	var eg errgroup.Group
+	for range 16 {
+		eg.Go(func() error {
+			c, err := d.Client(ctx)
+			if err != nil {
+				return err
+			}
+			defer c.Close()
+			_, err = c.ListWorkers(ctx)
+			return err
+		})
+	}
+	require.NoError(t, eg.Wait())
+	require.True(t, d.Features(ctx)[driver.DefaultLoad])
+	ip, err := d.HostGatewayIP(ctx)
+	require.NoError(t, err)
+	require.NotNil(t, ip)
+}

tests/integration_test.go

@@ -24,6 +24,7 @@ func TestIntegration(t *testing.T) {
 	var tests []func(t *testing.T, sb integration.Sandbox)
 	tests = append(tests, commonTests...)
 	tests = append(tests, buildTests...)
+	tests = append(tests, dockerTests...)
 	tests = append(tests, debugTests...)
 	tests = append(tests, policyBuildTests...)
 	tests = append(tests, policyEvalTests...)