From 30a3c273938bb534f7ad2a350bbd014db970796c Mon Sep 17 00:00:00 2001 From: Alberto Garcia Hierro Date: Wed, 22 Jul 2026 23:11:31 +0100 Subject: [PATCH] exporter: add session export finalization Add a FinalizeExport RPC to pass the combined exporter response back to the session provider after all exports complete. This lets clients run completion work while the build's gateway references remain available. Keep the callback optional for compatibility with existing session exporters. Ignore unsupported finalization, propagate callback failures, and support providers that only register a finalization callback. Signed-off-by: Alberto Garcia Hierro --- client/client_export_local_test.go | 64 +++ client/client_test.go | 1 + session/exporter/exporter.pb.go | 129 +++++- session/exporter/exporter.proto | 7 + session/exporter/exporter_grpc.pb.go | 40 +- session/exporter/exporter_vtproto.pb.go | 425 ++++++++++++++++++ session/exporter/exporterprovider/provider.go | 31 +- solver/llbsolver/export.go | 23 + solver/llbsolver/solver.go | 5 + 9 files changed, 702 insertions(+), 23 deletions(-) diff --git a/client/client_export_local_test.go b/client/client_export_local_test.go index 2e02b4b7d..022d9eba3 100644 --- a/client/client_export_local_test.go +++ b/client/client_export_local_test.go @@ -32,7 +32,10 @@ import ( "github.com/moby/buildkit/util/testutil/workers" ocispecs "github.com/opencontainers/image-spec/specs-go/v1" "github.com/pkg/errors" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" ) func testExportLocalForcePlatformSplit(t *testing.T, sb integration.Sandbox) { @@ -664,6 +667,8 @@ func testSessionExporter(t *testing.T, sb integration.Sandbox) { outW := bytes.NewBuffer(nil) exporterCalled := false + // The provider has no finalization callback; FinalizeExport returning + // Unimplemented must not fail the solve. exporter := exporterprovider.New(func(ctx context.Context, md map[string][]byte, refs []string) ([]*exporter.ExporterRequest, error) { require.Len(t, refs, 1) g := c.GatewayClientForBuild(buildID) @@ -731,6 +736,65 @@ func testSessionExporter(t *testing.T, sb integration.Sandbox) { } } +func testSessionExporterFinalizeExport(t *testing.T, sb integration.Sandbox) { + c, err := New(sb.Context(), sb.Address()) + require.NoError(t, err) + defer c.Close() + + st := llb.Image(integration.UnixOrWindows("busybox:latest", "nanoserver:latest")) + def, err := st.Marshal(sb.Context()) + require.NoError(t, err) + + build := func(name string, p session.Attachable) error { + _, err := c.Build(sb.Context(), SolveOpt{ + EnableSessionExporter: true, + Exports: []ExportEntry{ + { + Type: ExporterImage, + Attrs: map[string]string{ + "name": "session-exporter-finalize-" + name, + }, + }, + }, + Session: []session.Attachable{p}, + Ref: identity.NewID(), + }, "", func(ctx context.Context, c gateway.Client) (*gateway.Result, error) { + return c.Solve(ctx, gateway.SolveRequest{ + Definition: def.ToPB(), + }) + }, nil) + return err + } + + callbackErr := errors.New("finalize callback failed") + for _, tc := range []struct { + name string + callbackErr error + wantErr bool + }{ + {name: "success"}, + {name: "unavailable", callbackErr: status.Error(codes.Unavailable, "finalize callback unavailable")}, + {name: "error", callbackErr: callbackErr, wantErr: true}, + } { + t.Run(tc.name, func(t *testing.T) { + called := false + p := exporterprovider.New(nil, exporterprovider.WithFinalizeCallback(func(_ context.Context, exporterResponse map[string]string) error { + called = true + assert.NotEmpty(t, exporterResponse[exptypes.ExporterImageDescriptorKey]) + return tc.callbackErr + })) + + err := build(tc.name, p) + if tc.wantErr { + require.ErrorContains(t, err, tc.callbackErr.Error()) + } else { + require.NoError(t, err) + } + assert.True(t, called) + }) + } +} + // moby/buildkit#1418 func testTarExporterSymlink(t *testing.T, sb integration.Sandbox) { c, err := New(sb.Context(), sb.Address()) diff --git a/client/client_test.go b/client/client_test.go index 2f3fda52e..9009e5594 100644 --- a/client/client_test.go +++ b/client/client_test.go @@ -92,6 +92,7 @@ var allTests = []func(t *testing.T, sb integration.Sandbox){ testExporterTargetExists, testMultipleExporters, testSessionExporter, + testSessionExporterFinalizeExport, testTarExporterSymlink, testTarExporterWithSocket, testTarExporterWithSocketCopy, diff --git a/session/exporter/exporter.pb.go b/session/exporter/exporter.pb.go index 413efd72b..6b25e780e 100644 --- a/session/exporter/exporter.pb.go +++ b/session/exporter/exporter.pb.go @@ -169,6 +169,86 @@ func (x *ExporterRequest) GetAttrs() map[string]string { return nil } +type FinalizeExportRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + ExporterResponse map[string]string `protobuf:"bytes,1,rep,name=exporter_response,json=exporterResponse,proto3" json:"exporter_response,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *FinalizeExportRequest) Reset() { + *x = FinalizeExportRequest{} + mi := &file_github_com_moby_buildkit_session_exporter_exporter_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *FinalizeExportRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*FinalizeExportRequest) ProtoMessage() {} + +func (x *FinalizeExportRequest) ProtoReflect() protoreflect.Message { + mi := &file_github_com_moby_buildkit_session_exporter_exporter_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use FinalizeExportRequest.ProtoReflect.Descriptor instead. +func (*FinalizeExportRequest) Descriptor() ([]byte, []int) { + return file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDescGZIP(), []int{3} +} + +func (x *FinalizeExportRequest) GetExporterResponse() map[string]string { + if x != nil { + return x.ExporterResponse + } + return nil +} + +type FinalizeExportResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *FinalizeExportResponse) Reset() { + *x = FinalizeExportResponse{} + mi := &file_github_com_moby_buildkit_session_exporter_exporter_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *FinalizeExportResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*FinalizeExportResponse) ProtoMessage() {} + +func (x *FinalizeExportResponse) ProtoReflect() protoreflect.Message { + mi := &file_github_com_moby_buildkit_session_exporter_exporter_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use FinalizeExportResponse.ProtoReflect.Descriptor instead. +func (*FinalizeExportResponse) Descriptor() ([]byte, []int) { + return file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDescGZIP(), []int{4} +} + var File_github_com_moby_buildkit_session_exporter_exporter_proto protoreflect.FileDescriptor const file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDesc = "" + @@ -188,9 +268,16 @@ const file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDesc = "" "\n" + "AttrsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x012l\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xc8\x01\n" + + "\x15FinalizeExportRequest\x12j\n" + + "\x11exporter_response\x18\x01 \x03(\v2=.moby.exporter.v1.FinalizeExportRequest.ExporterResponseEntryR\x10exporterResponse\x1aC\n" + + "\x15ExporterResponseEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\x18\n" + + "\x16FinalizeExportResponse2\xd1\x01\n" + "\bExporter\x12`\n" + - "\rFindExporters\x12&.moby.exporter.v1.FindExportersRequest\x1a'.moby.exporter.v1.FindExportersResponseB+Z)github.com/moby/buildkit/session/exporterb\x06proto3" + "\rFindExporters\x12&.moby.exporter.v1.FindExportersRequest\x1a'.moby.exporter.v1.FindExportersResponse\x12c\n" + + "\x0eFinalizeExport\x12'.moby.exporter.v1.FinalizeExportRequest\x1a(.moby.exporter.v1.FinalizeExportResponseB+Z)github.com/moby/buildkit/session/exporterb\x06proto3" var ( file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDescOnce sync.Once @@ -204,25 +291,31 @@ func file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDescGZIP() return file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDescData } -var file_github_com_moby_buildkit_session_exporter_exporter_proto_msgTypes = make([]protoimpl.MessageInfo, 5) +var file_github_com_moby_buildkit_session_exporter_exporter_proto_msgTypes = make([]protoimpl.MessageInfo, 8) var file_github_com_moby_buildkit_session_exporter_exporter_proto_goTypes = []any{ - (*FindExportersRequest)(nil), // 0: moby.exporter.v1.FindExportersRequest - (*FindExportersResponse)(nil), // 1: moby.exporter.v1.FindExportersResponse - (*ExporterRequest)(nil), // 2: moby.exporter.v1.ExporterRequest - nil, // 3: moby.exporter.v1.FindExportersRequest.MetadataEntry - nil, // 4: moby.exporter.v1.ExporterRequest.AttrsEntry + (*FindExportersRequest)(nil), // 0: moby.exporter.v1.FindExportersRequest + (*FindExportersResponse)(nil), // 1: moby.exporter.v1.FindExportersResponse + (*ExporterRequest)(nil), // 2: moby.exporter.v1.ExporterRequest + (*FinalizeExportRequest)(nil), // 3: moby.exporter.v1.FinalizeExportRequest + (*FinalizeExportResponse)(nil), // 4: moby.exporter.v1.FinalizeExportResponse + nil, // 5: moby.exporter.v1.FindExportersRequest.MetadataEntry + nil, // 6: moby.exporter.v1.ExporterRequest.AttrsEntry + nil, // 7: moby.exporter.v1.FinalizeExportRequest.ExporterResponseEntry } var file_github_com_moby_buildkit_session_exporter_exporter_proto_depIdxs = []int32{ - 3, // 0: moby.exporter.v1.FindExportersRequest.metadata:type_name -> moby.exporter.v1.FindExportersRequest.MetadataEntry + 5, // 0: moby.exporter.v1.FindExportersRequest.metadata:type_name -> moby.exporter.v1.FindExportersRequest.MetadataEntry 2, // 1: moby.exporter.v1.FindExportersResponse.exporters:type_name -> moby.exporter.v1.ExporterRequest - 4, // 2: moby.exporter.v1.ExporterRequest.Attrs:type_name -> moby.exporter.v1.ExporterRequest.AttrsEntry - 0, // 3: moby.exporter.v1.Exporter.FindExporters:input_type -> moby.exporter.v1.FindExportersRequest - 1, // 4: moby.exporter.v1.Exporter.FindExporters:output_type -> moby.exporter.v1.FindExportersResponse - 4, // [4:5] is the sub-list for method output_type - 3, // [3:4] is the sub-list for method input_type - 3, // [3:3] is the sub-list for extension type_name - 3, // [3:3] is the sub-list for extension extendee - 0, // [0:3] is the sub-list for field type_name + 6, // 2: moby.exporter.v1.ExporterRequest.Attrs:type_name -> moby.exporter.v1.ExporterRequest.AttrsEntry + 7, // 3: moby.exporter.v1.FinalizeExportRequest.exporter_response:type_name -> moby.exporter.v1.FinalizeExportRequest.ExporterResponseEntry + 0, // 4: moby.exporter.v1.Exporter.FindExporters:input_type -> moby.exporter.v1.FindExportersRequest + 3, // 5: moby.exporter.v1.Exporter.FinalizeExport:input_type -> moby.exporter.v1.FinalizeExportRequest + 1, // 6: moby.exporter.v1.Exporter.FindExporters:output_type -> moby.exporter.v1.FindExportersResponse + 4, // 7: moby.exporter.v1.Exporter.FinalizeExport:output_type -> moby.exporter.v1.FinalizeExportResponse + 6, // [6:8] is the sub-list for method output_type + 4, // [4:6] is the sub-list for method input_type + 4, // [4:4] is the sub-list for extension type_name + 4, // [4:4] is the sub-list for extension extendee + 0, // [0:4] is the sub-list for field type_name } func init() { file_github_com_moby_buildkit_session_exporter_exporter_proto_init() } @@ -236,7 +329,7 @@ func file_github_com_moby_buildkit_session_exporter_exporter_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDesc), len(file_github_com_moby_buildkit_session_exporter_exporter_proto_rawDesc)), NumEnums: 0, - NumMessages: 5, + NumMessages: 8, NumExtensions: 0, NumServices: 1, }, diff --git a/session/exporter/exporter.proto b/session/exporter/exporter.proto index 0a0423372..76f4e1c9d 100644 --- a/session/exporter/exporter.proto +++ b/session/exporter/exporter.proto @@ -6,6 +6,7 @@ option go_package = "github.com/moby/buildkit/session/exporter"; service Exporter { rpc FindExporters(FindExportersRequest) returns (FindExportersResponse); + rpc FinalizeExport(FinalizeExportRequest) returns (FinalizeExportResponse); } message FindExportersRequest{ @@ -21,3 +22,9 @@ message ExporterRequest { string Type = 1; map Attrs = 2; } + +message FinalizeExportRequest { + map exporter_response = 1; +} + +message FinalizeExportResponse {} diff --git a/session/exporter/exporter_grpc.pb.go b/session/exporter/exporter_grpc.pb.go index 68c6b28da..af1559be5 100644 --- a/session/exporter/exporter_grpc.pb.go +++ b/session/exporter/exporter_grpc.pb.go @@ -19,7 +19,8 @@ import ( const _ = grpc.SupportPackageIsVersion9 const ( - Exporter_FindExporters_FullMethodName = "/moby.exporter.v1.Exporter/FindExporters" + Exporter_FindExporters_FullMethodName = "/moby.exporter.v1.Exporter/FindExporters" + Exporter_FinalizeExport_FullMethodName = "/moby.exporter.v1.Exporter/FinalizeExport" ) // ExporterClient is the client API for Exporter service. @@ -27,6 +28,7 @@ const ( // For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. type ExporterClient interface { FindExporters(ctx context.Context, in *FindExportersRequest, opts ...grpc.CallOption) (*FindExportersResponse, error) + FinalizeExport(ctx context.Context, in *FinalizeExportRequest, opts ...grpc.CallOption) (*FinalizeExportResponse, error) } type exporterClient struct { @@ -47,11 +49,22 @@ func (c *exporterClient) FindExporters(ctx context.Context, in *FindExportersReq return out, nil } +func (c *exporterClient) FinalizeExport(ctx context.Context, in *FinalizeExportRequest, opts ...grpc.CallOption) (*FinalizeExportResponse, error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + out := new(FinalizeExportResponse) + err := c.cc.Invoke(ctx, Exporter_FinalizeExport_FullMethodName, in, out, cOpts...) + if err != nil { + return nil, err + } + return out, nil +} + // ExporterServer is the server API for Exporter service. // All implementations should embed UnimplementedExporterServer // for forward compatibility. type ExporterServer interface { FindExporters(context.Context, *FindExportersRequest) (*FindExportersResponse, error) + FinalizeExport(context.Context, *FinalizeExportRequest) (*FinalizeExportResponse, error) } // UnimplementedExporterServer should be embedded to have @@ -64,6 +77,9 @@ type UnimplementedExporterServer struct{} func (UnimplementedExporterServer) FindExporters(context.Context, *FindExportersRequest) (*FindExportersResponse, error) { return nil, status.Error(codes.Unimplemented, "method FindExporters not implemented") } +func (UnimplementedExporterServer) FinalizeExport(context.Context, *FinalizeExportRequest) (*FinalizeExportResponse, error) { + return nil, status.Error(codes.Unimplemented, "method FinalizeExport not implemented") +} func (UnimplementedExporterServer) testEmbeddedByValue() {} // UnsafeExporterServer may be embedded to opt out of forward compatibility for this service. @@ -102,6 +118,24 @@ func _Exporter_FindExporters_Handler(srv interface{}, ctx context.Context, dec f return interceptor(ctx, in, info, handler) } +func _Exporter_FinalizeExport_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { + in := new(FinalizeExportRequest) + if err := dec(in); err != nil { + return nil, err + } + if interceptor == nil { + return srv.(ExporterServer).FinalizeExport(ctx, in) + } + info := &grpc.UnaryServerInfo{ + Server: srv, + FullMethod: Exporter_FinalizeExport_FullMethodName, + } + handler := func(ctx context.Context, req interface{}) (interface{}, error) { + return srv.(ExporterServer).FinalizeExport(ctx, req.(*FinalizeExportRequest)) + } + return interceptor(ctx, in, info, handler) +} + // Exporter_ServiceDesc is the grpc.ServiceDesc for Exporter service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -113,6 +147,10 @@ var Exporter_ServiceDesc = grpc.ServiceDesc{ MethodName: "FindExporters", Handler: _Exporter_FindExporters_Handler, }, + { + MethodName: "FinalizeExport", + Handler: _Exporter_FinalizeExport_Handler, + }, }, Streams: []grpc.StreamDesc{}, Metadata: "github.com/moby/buildkit/session/exporter/exporter.proto", diff --git a/session/exporter/exporter_vtproto.pb.go b/session/exporter/exporter_vtproto.pb.go index 469754619..466c1f436 100644 --- a/session/exporter/exporter_vtproto.pb.go +++ b/session/exporter/exporter_vtproto.pb.go @@ -96,6 +96,45 @@ func (m *ExporterRequest) CloneMessageVT() proto.Message { return m.CloneVT() } +func (m *FinalizeExportRequest) CloneVT() *FinalizeExportRequest { + if m == nil { + return (*FinalizeExportRequest)(nil) + } + r := new(FinalizeExportRequest) + if rhs := m.ExporterResponse; rhs != nil { + tmpContainer := make(map[string]string, len(rhs)) + for k, v := range rhs { + tmpContainer[k] = v + } + r.ExporterResponse = tmpContainer + } + if len(m.unknownFields) > 0 { + r.unknownFields = make([]byte, len(m.unknownFields)) + copy(r.unknownFields, m.unknownFields) + } + return r +} + +func (m *FinalizeExportRequest) CloneMessageVT() proto.Message { + return m.CloneVT() +} + +func (m *FinalizeExportResponse) CloneVT() *FinalizeExportResponse { + if m == nil { + return (*FinalizeExportResponse)(nil) + } + r := new(FinalizeExportResponse) + if len(m.unknownFields) > 0 { + r.unknownFields = make([]byte, len(m.unknownFields)) + copy(r.unknownFields, m.unknownFields) + } + return r +} + +func (m *FinalizeExportResponse) CloneMessageVT() proto.Message { + return m.CloneVT() +} + func (this *FindExportersRequest) EqualVT(that *FindExportersRequest) bool { if this == that { return true @@ -197,6 +236,50 @@ func (this *ExporterRequest) EqualMessageVT(thatMsg proto.Message) bool { } return this.EqualVT(that) } +func (this *FinalizeExportRequest) EqualVT(that *FinalizeExportRequest) bool { + if this == that { + return true + } else if this == nil || that == nil { + return false + } + if len(this.ExporterResponse) != len(that.ExporterResponse) { + return false + } + for i, vx := range this.ExporterResponse { + vy, ok := that.ExporterResponse[i] + if !ok { + return false + } + if vx != vy { + return false + } + } + return string(this.unknownFields) == string(that.unknownFields) +} + +func (this *FinalizeExportRequest) EqualMessageVT(thatMsg proto.Message) bool { + that, ok := thatMsg.(*FinalizeExportRequest) + if !ok { + return false + } + return this.EqualVT(that) +} +func (this *FinalizeExportResponse) EqualVT(that *FinalizeExportResponse) bool { + if this == that { + return true + } else if this == nil || that == nil { + return false + } + return string(this.unknownFields) == string(that.unknownFields) +} + +func (this *FinalizeExportResponse) EqualMessageVT(thatMsg proto.Message) bool { + that, ok := thatMsg.(*FinalizeExportResponse) + if !ok { + return false + } + return this.EqualVT(that) +} func (m *FindExportersRequest) MarshalVT() (dAtA []byte, err error) { if m == nil { return nil, nil @@ -362,6 +445,91 @@ func (m *ExporterRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { return len(dAtA) - i, nil } +func (m *FinalizeExportRequest) MarshalVT() (dAtA []byte, err error) { + if m == nil { + return nil, nil + } + size := m.SizeVT() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBufferVT(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *FinalizeExportRequest) MarshalToVT(dAtA []byte) (int, error) { + size := m.SizeVT() + return m.MarshalToSizedBufferVT(dAtA[:size]) +} + +func (m *FinalizeExportRequest) MarshalToSizedBufferVT(dAtA []byte) (int, error) { + if m == nil { + return 0, nil + } + i := len(dAtA) + _ = i + var l int + _ = l + if m.unknownFields != nil { + i -= len(m.unknownFields) + copy(dAtA[i:], m.unknownFields) + } + if len(m.ExporterResponse) > 0 { + for k := range m.ExporterResponse { + v := m.ExporterResponse[k] + baseI := i + i -= len(v) + copy(dAtA[i:], v) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(v))) + i-- + dAtA[i] = 0x12 + i -= len(k) + copy(dAtA[i:], k) + i = protohelpers.EncodeVarint(dAtA, i, uint64(len(k))) + i-- + dAtA[i] = 0xa + i = protohelpers.EncodeVarint(dAtA, i, uint64(baseI-i)) + i-- + dAtA[i] = 0xa + } + } + return len(dAtA) - i, nil +} + +func (m *FinalizeExportResponse) MarshalVT() (dAtA []byte, err error) { + if m == nil { + return nil, nil + } + size := m.SizeVT() + dAtA = make([]byte, size) + n, err := m.MarshalToSizedBufferVT(dAtA[:size]) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *FinalizeExportResponse) MarshalToVT(dAtA []byte) (int, error) { + size := m.SizeVT() + return m.MarshalToSizedBufferVT(dAtA[:size]) +} + +func (m *FinalizeExportResponse) MarshalToSizedBufferVT(dAtA []byte) (int, error) { + if m == nil { + return 0, nil + } + i := len(dAtA) + _ = i + var l int + _ = l + if m.unknownFields != nil { + i -= len(m.unknownFields) + copy(dAtA[i:], m.unknownFields) + } + return len(dAtA) - i, nil +} + func (m *FindExportersRequest) SizeVT() (n int) { if m == nil { return 0 @@ -425,6 +593,34 @@ func (m *ExporterRequest) SizeVT() (n int) { return n } +func (m *FinalizeExportRequest) SizeVT() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + if len(m.ExporterResponse) > 0 { + for k, v := range m.ExporterResponse { + _ = k + _ = v + mapEntrySize := 1 + len(k) + protohelpers.SizeOfVarint(uint64(len(k))) + 1 + len(v) + protohelpers.SizeOfVarint(uint64(len(v))) + n += mapEntrySize + 1 + protohelpers.SizeOfVarint(uint64(mapEntrySize)) + } + } + n += len(m.unknownFields) + return n +} + +func (m *FinalizeExportResponse) SizeVT() (n int) { + if m == nil { + return 0 + } + var l int + _ = l + n += len(m.unknownFields) + return n +} + func (m *FindExportersRequest) UnmarshalVT(dAtA []byte) error { l := len(dAtA) iNdEx := 0 @@ -931,3 +1127,232 @@ func (m *ExporterRequest) UnmarshalVT(dAtA []byte) error { } return nil } +func (m *FinalizeExportRequest) UnmarshalVT(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: FinalizeExportRequest: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: FinalizeExportRequest: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field ExporterResponse", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= int(b&0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return protohelpers.ErrInvalidLength + } + postIndex := iNdEx + msglen + if postIndex < 0 { + return protohelpers.ErrInvalidLength + } + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.ExporterResponse == nil { + m.ExporterResponse = make(map[string]string) + } + var mapkey string + var mapvalue string + for iNdEx < postIndex { + entryPreIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + if fieldNum == 1 { + var stringLenmapkey uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapkey |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapkey := int(stringLenmapkey) + if intStringLenmapkey < 0 { + return protohelpers.ErrInvalidLength + } + postStringIndexmapkey := iNdEx + intStringLenmapkey + if postStringIndexmapkey < 0 { + return protohelpers.ErrInvalidLength + } + if postStringIndexmapkey > l { + return io.ErrUnexpectedEOF + } + mapkey = string(dAtA[iNdEx:postStringIndexmapkey]) + iNdEx = postStringIndexmapkey + } else if fieldNum == 2 { + var stringLenmapvalue uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLenmapvalue |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + intStringLenmapvalue := int(stringLenmapvalue) + if intStringLenmapvalue < 0 { + return protohelpers.ErrInvalidLength + } + postStringIndexmapvalue := iNdEx + intStringLenmapvalue + if postStringIndexmapvalue < 0 { + return protohelpers.ErrInvalidLength + } + if postStringIndexmapvalue > l { + return io.ErrUnexpectedEOF + } + mapvalue = string(dAtA[iNdEx:postStringIndexmapvalue]) + iNdEx = postStringIndexmapvalue + } else { + iNdEx = entryPreIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > postIndex { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + m.ExporterResponse[mapkey] = mapvalue + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.unknownFields = append(m.unknownFields, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *FinalizeExportResponse) UnmarshalVT(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return protohelpers.ErrIntOverflow + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= uint64(b&0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: FinalizeExportResponse: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: FinalizeExportResponse: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + default: + iNdEx = preIndex + skippy, err := protohelpers.Skip(dAtA[iNdEx:]) + if err != nil { + return err + } + if (skippy < 0) || (iNdEx+skippy) < 0 { + return protohelpers.ErrInvalidLength + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + m.unknownFields = append(m.unknownFields, dAtA[iNdEx:iNdEx+skippy]...) + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} diff --git a/session/exporter/exporterprovider/provider.go b/session/exporter/exporterprovider/provider.go index abdc28894..ac40d84fa 100644 --- a/session/exporter/exporterprovider/provider.go +++ b/session/exporter/exporterprovider/provider.go @@ -11,14 +11,37 @@ import ( type Callback func(ctx context.Context, md map[string][]byte, refs []string) ([]*exporter.ExporterRequest, error) -func New(cb Callback) *Exporter { - return &Exporter{ - cb: cb, +type FinalizeCallback func(ctx context.Context, exporterResponse map[string]string) error + +type Option func(*Exporter) + +func WithFinalizeCallback(cb FinalizeCallback) Option { + return func(e *Exporter) { + e.finalize = cb } } +func New(cb Callback, opts ...Option) *Exporter { + e := &Exporter{cb: cb} + for _, opt := range opts { + opt(e) + } + return e +} + type Exporter struct { - cb Callback + cb Callback + finalize FinalizeCallback +} + +func (e *Exporter) FinalizeExport(ctx context.Context, in *exporter.FinalizeExportRequest) (*exporter.FinalizeExportResponse, error) { + if e.finalize == nil { + return nil, status.Errorf(codes.Unimplemented, "no exporter finalize callback registered") + } + if err := e.finalize(ctx, in.ExporterResponse); err != nil { + return nil, err + } + return &exporter.FinalizeExportResponse{}, nil } func (e *Exporter) Register(server *grpc.Server) { diff --git a/solver/llbsolver/export.go b/solver/llbsolver/export.go index 8e59a9b2a..414852eea 100644 --- a/solver/llbsolver/export.go +++ b/solver/llbsolver/export.go @@ -82,6 +82,29 @@ func (s *Solver) getSessionExporters(ctx context.Context, sessionID string, id i return out, nil } +func (s *Solver) finalizeSessionExport(ctx context.Context, sessionID string, exporterResponse map[string]string) error { + timeoutCtx, cancel := context.WithCancelCause(ctx) + timeoutCtx, _ = context.WithTimeoutCause(timeoutCtx, 5*time.Second, errors.WithStack(context.DeadlineExceeded)) //nolint:govet + defer func() { cancel(errors.WithStack(context.Canceled)) }() + + caller, err := s.sm.Get(timeoutCtx, sessionID, false) + if err != nil { + return err + } + + client := sessionexporter.NewExporterClient(caller.Conn()) + _, err = client.FinalizeExport(caller.Context(ctx), &sessionexporter.FinalizeExportRequest{ + ExporterResponse: exporterResponse, + }) + if err != nil { + switch grpcerrors.Code(err) { + case codes.Unavailable, codes.Unimplemented: + return nil + } + } + return err +} + func runCacheExporters(ctx context.Context, exporters []RemoteCacheExporter, j *solver.Job, cached *result.Result[solver.CachedResult], inp *result.Result[cache.ImmutableRef]) (map[string]string, error) { eg, ctx := errgroup.WithContext(ctx) g := session.NewGroup(j.SessionID) diff --git a/solver/llbsolver/solver.go b/solver/llbsolver/solver.go index 378df4603..d168bdd6f 100644 --- a/solver/llbsolver/solver.go +++ b/solver/llbsolver/solver.go @@ -443,6 +443,11 @@ func (s *Solver) Solve(ctx context.Context, id string, sessionID string, req fro exporterResponse[k] = v } } + if exp.EnableSessionExporter { + if err := s.finalizeSessionExport(ctx, j.SessionID, exporterResponse); err != nil { + return nil, err + } + } return &client.SolveResponse{ ExporterResponse: exporterResponse,