diff --git a/client/client_test.go b/client/client_test.go index 0763cc12d972..ab9a33543025 100644 --- a/client/client_test.go +++ b/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) diff --git a/frontend/dockerfile/dockerfile_test.go b/frontend/dockerfile/dockerfile_test.go index 2ea776dc75a7..01e17a26b909 100644 --- a/frontend/dockerfile/dockerfile_test.go +++ b/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 { diff --git a/frontend/gateway/gateway.go b/frontend/gateway/gateway.go index d79ee40957c5..38991de756d9 100644 --- a/frontend/gateway/gateway.go +++ b/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" diff --git a/frontend/gateway/grpcclient/client.go b/frontend/gateway/grpcclient/client.go index 205e260a677d..19bc5706afe1 100644 --- a/frontend/gateway/grpcclient/client.go +++ b/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 } diff --git a/frontend/gateway/pb/caps.go b/frontend/gateway/pb/caps.go index a8e45fca9f3c..4c278a5cb3a8 100644 --- a/frontend/gateway/pb/caps.go +++ b/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",