dilim/buildkit
1
0
Fork 0
mirror of https://github.com/moby/buildkit.git synced 2026-08-08 00:30:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

exporter: add session export finalization

Browse Source 
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 <alberto.hierro@docker.com>
This commit is contained in:
Alberto Garcia Hierro
2026-07-22 23:11:31 +01:00
parent b8a2467139
commit 30a3c27393
9 changed files with 702 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
client/client_export_local_test.go
View File

@@ -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())

1
client/client_test.go
View File

@@ -92,6 +92,7 @@ var allTests = []func(t *testing.T, sb integration.Sandbox){
testExporterTargetExists,
testMultipleExporters,
testSessionExporter,
testSessionExporterFinalizeExport,
testTarExporterSymlink,
testTarExporterWithSocket,
testTarExporterWithSocketCopy,

129
session/exporter/exporter.pb.go
View File

@@ -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,
},

7
session/exporter/exporter.proto
View File

@@ -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<string, string> Attrs = 2;
}
message FinalizeExportRequest {
map<string, string> exporter_response = 1;
}
message FinalizeExportResponse {}

40
session/exporter/exporter_grpc.pb.go
View File

@@ -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",

425
session/exporter/exporter_vtproto.pb.go
View File

@@ -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
}

31
session/exporter/exporterprovider/provider.go
View File

@@ -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) {

23
solver/llbsolver/export.go
View File

@@ -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)

5
solver/llbsolver/solver.go
View File

@@ -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,