gateway: compress frontend input payloads

client/client_test.go

@@ -92,6 +92,7 @@ import (
 	"golang.org/x/mod/semver"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
+	grpcgzip "google.golang.org/grpc/encoding/gzip"
 )
 
 func init() {
@@ -237,6 +238,7 @@ var allTests = []func(t *testing.T, sb integration.Sandbox){
 	testImageManifestRegistryCacheImportExport,
 	testLLBMountPerformance,
 	testClientCustomGRPCOpts,
+	testClientGatewayFrontendInputsCompressed,
 	testMultipleRecordsWithSameLayersCacheImportExport,
 	testRegistryEmptyCacheExport,
 	testSnapshotWithMultipleBlobs,
@@ -13414,6 +13416,53 @@ func testClientCustomGRPCOpts(t *testing.T, sb integration.Sandbox) {
 	require.Contains(t, interceptedMethods, "/moby.buildkit.v1.Control/Solve")
 }
 
+func testClientGatewayFrontendInputsCompressed(t *testing.T, sb integration.Sandbox) {
+	var solveCalls atomic.Int64
+	var solveCompressed atomic.Bool
+	intercept := func(
+		ctx context.Context,
+		method string,
+		req,
+		reply any,
+		cc *grpc.ClientConn,
+		invoker grpc.UnaryInvoker,
+		opts ...grpc.CallOption,
+	) error {
+		if method == gatewaypb.LLBBridge_Solve_FullMethodName {
+			solveCalls.Add(1)
+			for _, opt := range opts {
+				compressor, ok := opt.(grpc.CompressorCallOption)
+				if ok && compressor.CompressorType == grpcgzip.Name {
+					solveCompressed.Store(true)
+				}
+			}
+		}
+		return invoker(ctx, method, req, reply, cc, opts...)
+	}
+	c, err := New(sb.Context(), sb.Address(), WithGRPCDialOption(grpc.WithChainUnaryInterceptor(intercept)))
+	require.NoError(t, err)
+	defer c.Close()
+
+	frontend := func(ctx context.Context, c gateway.Client) (*gateway.Result, error) {
+		st := llb.Scratch().File(llb.Mkfile("foo", 0600, []byte("data")))
+		def, err := st.Marshal(ctx)
+		if err != nil {
+			return nil, err
+		}
+		return c.Solve(ctx, gateway.SolveRequest{
+			Definition: def.ToPB(),
+			FrontendInputs: map[string]*pb.Definition{
+				"context": def.ToPB(),
+			},
+		})
+	}
+
+	_, err = c.Build(sb.Context(), SolveOpt{}, "", frontend, nil)
+	require.NoError(t, err)
+	require.NotZero(t, solveCalls.Load(), "expected a gateway Solve request")
+	require.True(t, solveCompressed.Load(), "expected gateway Solve request with frontend inputs to use gzip compression")
+}
+
 func testRunValidExitCodes(t *testing.T, sb integration.Sandbox) {
 	c, err := New(sb.Context(), sb.Address())
 	require.NoError(t, err)

frontend/dockerfile/dockerfile_test.go

@@ -19,12 +19,15 @@ import (
 	"runtime"
 	"slices"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	cacheimporttypes "github.com/moby/buildkit/cache/remotecache/v1/types"
 	"github.com/tonistiigi/fsutil"
 	"golang.org/x/sync/errgroup"
+	"google.golang.org/grpc"
+	grpcgzip "google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
@@ -44,6 +47,7 @@ import (
 	"github.com/moby/buildkit/frontend/dockerfile/builder"
 	"github.com/moby/buildkit/frontend/dockerui"
 	gateway "github.com/moby/buildkit/frontend/gateway/client"
+	gatewayapi "github.com/moby/buildkit/frontend/gateway/pb"
 	"github.com/moby/buildkit/frontend/subrequests"
 	"github.com/moby/buildkit/identity"
 	"github.com/moby/buildkit/session"
@@ -141,6 +145,7 @@ var allTests = integration.TestFuncs(
 	testFrontendUseForwardedSolveResults,
 	testFrontendEvaluate,
 	testFrontendInputs,
+	testFrontendInputsCompressed,
 	testErrorsSourceMap,
 	testMultiArgs,
 	testFrontendSubrequests,
@@ -7401,6 +7406,72 @@ COPY foo foo2
 	require.Equal(t, expected, actual)
 }
 
+type inputsCompressionInterceptor struct {
+	calls      atomic.Int64
+	compressed atomic.Bool
+}
+
+func (i *inputsCompressionInterceptor) unary(ctx context.Context, method string, req any, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+	if method == gatewayapi.LLBBridge_Inputs_FullMethodName {
+		i.calls.Add(1)
+		for _, opt := range opts {
+			if compressor, ok := opt.(grpc.CompressorCallOption); ok && compressor.CompressorType == grpcgzip.Name {
+				i.compressed.Store(true)
+			}
+		}
+	}
+	return invoker(ctx, method, req, reply, cc, opts...)
+}
+
+func testFrontendInputsCompressed(t *testing.T, sb integration.Sandbox) {
+	integration.SkipOnPlatform(t, "windows")
+	f := getFrontend(t, sb)
+	if _, ok := f.(*clientFrontend); !ok {
+		t.Skip("only test with client frontend")
+	}
+
+	inputsInterceptor := &inputsCompressionInterceptor{}
+	c, err := client.New(sb.Context(), sb.Address(), client.WithGRPCDialOption(grpc.WithChainUnaryInterceptor(inputsInterceptor.unary)))
+	require.NoError(t, err)
+	defer c.Close()
+
+	payload := bytes.Repeat([]byte("a"), 64<<10)
+	input := llb.Scratch().File(llb.Mkfile("/payload", 0600, payload))
+
+	destDir := t.TempDir()
+	dockerfile := []byte(`
+FROM scratch
+COPY payload payload
+`)
+	dir := integration.Tmpdir(
+		t,
+		fstest.CreateFile("Dockerfile", dockerfile, 0600),
+	)
+
+	_, err = f.Solve(sb.Context(), c, client.SolveOpt{
+		Exports: []client.ExportEntry{
+			{
+				Type:      client.ExporterLocal,
+				OutputDir: destDir,
+			},
+		},
+		LocalMounts: map[string]fsutil.FS{
+			dockerui.DefaultLocalNameDockerfile: dir,
+		},
+		FrontendInputs: map[string]llb.State{
+			dockerui.DefaultLocalNameContext: input,
+		},
+	}, nil)
+	require.NoError(t, err)
+
+	actual, err := os.ReadFile(filepath.Join(destDir, "payload"))
+	require.NoError(t, err)
+	require.Equal(t, payload, actual)
+
+	require.NotZero(t, inputsInterceptor.calls.Load(), "expected an Inputs request")
+	require.True(t, inputsInterceptor.compressed.Load(), "expected Inputs request to use gzip compression")
+}
+
 func testFrontendSubrequests(t *testing.T, sb integration.Sandbox) {
 	f := getFrontend(t, sb)
 	if _, ok := f.(*clientFrontend); !ok {

frontend/gateway/gateway.go

@@ -56,6 +56,7 @@ import (
 	spb "google.golang.org/genproto/googleapis/rpc/status"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
+	_ "google.golang.org/grpc/encoding/gzip" // register gzip compressor for gateway request decompression
 	"google.golang.org/grpc/health"
 	"google.golang.org/grpc/health/grpc_health_v1"
 	"google.golang.org/grpc/status"

frontend/gateway/grpcclient/client.go

@@ -35,6 +35,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/status"
 )
 
@@ -412,7 +413,12 @@ func (c *grpcClient) Solve(ctx context.Context, creq client.SolveRequest) (res *
 		}
 	}
 
-	resp, err := c.client.Solve(ctx, req)
+	var opts []grpc.CallOption
+	if len(req.FrontendInputs) > 0 && c.caps.Supports(pb.CapGatewayGzip) == nil {
+		opts = append(opts, grpc.UseCompressor(gzip.Name))
+	}
+
+	resp, err := c.client.Solve(ctx, req, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -800,7 +806,12 @@ func (c *grpcClient) Inputs(ctx context.Context) (map[string]llb.State, error) {
 		return nil, err
 	}
 
-	resp, err := c.client.Inputs(ctx, &pb.InputsRequest{})
+	var opts []grpc.CallOption
+	if c.caps.Supports(pb.CapGatewayGzip) == nil {
+		opts = append(opts, grpc.UseCompressor(gzip.Name))
+	}
+
+	resp, err := c.client.Inputs(ctx, &pb.InputsRequest{}, opts...)
 	if err != nil {
 		return nil, err
 	}

frontend/gateway/pb/caps.go

@@ -69,6 +69,9 @@ const (
 	// CapGatewayWarnings is the capability to log warnings from frontend
 	CapGatewayWarnings apicaps.CapID = "gateway.warnings"
 
+	// CapGatewayGzip is the capability to receive gzip-compressed gateway requests.
+	CapGatewayGzip apicaps.CapID = "gateway.grpc.gzip"
+
 	// CapAttestations is the capability to indicate that attestation
 	// references will be attached to results
 	CapAttestations apicaps.CapID = "reference.attestations"
@@ -246,6 +249,13 @@ func init() {
 		Status:  apicaps.CapStatusExperimental,
 	})
 
+	Caps.Init(apicaps.Cap{
+		ID:      CapGatewayGzip,
+		Name:    "gzip compressed gateway requests",
+		Enabled: true,
+		Status:  apicaps.CapStatusExperimental,
+	})
+
 	Caps.Init(apicaps.Cap{
 		ID:      CapAttestations,
 		Name:    "reference attestations",