diff --git a/internal/machine/api/pb/docker.pb.go b/internal/machine/api/pb/docker.pb.go index f996cec5..e4590dcc 100644 --- a/internal/machine/api/pb/docker.pb.go +++ b/internal/machine/api/pb/docker.pb.go @@ -208,6 +208,110 @@ func (x *StartContainerRequest) GetOptions() []byte { return nil } +type PullImageRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Image string `protobuf:"bytes,1,opt,name=image,proto3" json:"image,omitempty"` + // JSON serialized image.PullOptions. + Options []byte `protobuf:"bytes,2,opt,name=options,proto3" json:"options,omitempty"` +} + +func (x *PullImageRequest) Reset() { + *x = PullImageRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_internal_machine_api_pb_docker_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *PullImageRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*PullImageRequest) ProtoMessage() {} + +func (x *PullImageRequest) ProtoReflect() protoreflect.Message { + mi := &file_internal_machine_api_pb_docker_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use PullImageRequest.ProtoReflect.Descriptor instead. +func (*PullImageRequest) Descriptor() ([]byte, []int) { + return file_internal_machine_api_pb_docker_proto_rawDescGZIP(), []int{3} +} + +func (x *PullImageRequest) GetImage() string { + if x != nil { + return x.Image + } + return "" +} + +func (x *PullImageRequest) GetOptions() []byte { + if x != nil { + return x.Options + } + return nil +} + +type JSONMessage struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + // JSON serialized jsonmessage.JSONMessage. + Message []byte `protobuf:"bytes,1,opt,name=message,proto3" json:"message,omitempty"` +} + +func (x *JSONMessage) Reset() { + *x = JSONMessage{} + if protoimpl.UnsafeEnabled { + mi := &file_internal_machine_api_pb_docker_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *JSONMessage) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*JSONMessage) ProtoMessage() {} + +func (x *JSONMessage) ProtoReflect() protoreflect.Message { + mi := &file_internal_machine_api_pb_docker_proto_msgTypes[4] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use JSONMessage.ProtoReflect.Descriptor instead. +func (*JSONMessage) Descriptor() ([]byte, []int) { + return file_internal_machine_api_pb_docker_proto_rawDescGZIP(), []int{4} +} + +func (x *JSONMessage) GetMessage() []byte { + if x != nil { + return x.Message + } + return nil +} + var File_internal_machine_api_pb_docker_proto protoreflect.FileDescriptor var file_internal_machine_api_pb_docker_proto_rawDesc = []byte{ @@ -233,21 +337,32 @@ var file_internal_machine_api_pb_docker_proto_rawDesc = []byte{ 0x61, 0x72, 0x74, 0x43, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x18, 0x0a, 0x07, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x18, 0x02, - 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x32, 0x9c, 0x01, - 0x0a, 0x06, 0x44, 0x6f, 0x63, 0x6b, 0x65, 0x72, 0x12, 0x4c, 0x0a, 0x0f, 0x43, 0x72, 0x65, 0x61, - 0x74, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x12, 0x1b, 0x2e, 0x61, 0x70, - 0x69, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, - 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x43, + 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x73, 0x22, 0x42, 0x0a, + 0x10, 0x50, 0x75, 0x6c, 0x6c, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x12, 0x14, 0x0a, 0x05, 0x69, 0x6d, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x05, 0x69, 0x6d, 0x61, 0x67, 0x65, 0x12, 0x18, 0x0a, 0x07, 0x6f, 0x70, 0x74, 0x69, 0x6f, + 0x6e, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x6f, 0x70, 0x74, 0x69, 0x6f, 0x6e, + 0x73, 0x22, 0x27, 0x0a, 0x0b, 0x4a, 0x53, 0x4f, 0x4e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, + 0x12, 0x18, 0x0a, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, + 0x0c, 0x52, 0x07, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x32, 0xd4, 0x01, 0x0a, 0x06, 0x44, + 0x6f, 0x63, 0x6b, 0x65, 0x72, 0x12, 0x4c, 0x0a, 0x0f, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, + 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x12, 0x1b, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x43, 0x72, 0x65, 0x61, 0x74, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x52, 0x65, - 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x44, 0x0a, 0x0e, 0x53, 0x74, 0x61, 0x72, 0x74, 0x43, - 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x12, 0x1a, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x53, - 0x74, 0x61, 0x72, 0x74, 0x43, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x71, - 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, - 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x42, 0x37, 0x5a, 0x35, - 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x70, 0x73, 0x76, 0x69, 0x64, - 0x65, 0x72, 0x73, 0x6b, 0x69, 0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, 0x75, 0x64, 0x2f, 0x69, 0x6e, - 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, 0x69, 0x6e, 0x65, 0x2f, 0x61, - 0x70, 0x69, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1c, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x43, 0x72, 0x65, 0x61, + 0x74, 0x65, 0x43, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x73, 0x70, 0x6f, + 0x6e, 0x73, 0x65, 0x12, 0x44, 0x0a, 0x0e, 0x53, 0x74, 0x61, 0x72, 0x74, 0x43, 0x6f, 0x6e, 0x74, + 0x61, 0x69, 0x6e, 0x65, 0x72, 0x12, 0x1a, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x53, 0x74, 0x61, 0x72, + 0x74, 0x43, 0x6f, 0x6e, 0x74, 0x61, 0x69, 0x6e, 0x65, 0x72, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, + 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x12, 0x36, 0x0a, 0x09, 0x50, 0x75, 0x6c, + 0x6c, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x12, 0x15, 0x2e, 0x61, 0x70, 0x69, 0x2e, 0x50, 0x75, 0x6c, + 0x6c, 0x49, 0x6d, 0x61, 0x67, 0x65, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x10, 0x2e, + 0x61, 0x70, 0x69, 0x2e, 0x4a, 0x53, 0x4f, 0x4e, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x30, + 0x01, 0x42, 0x37, 0x5a, 0x35, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, + 0x70, 0x73, 0x76, 0x69, 0x64, 0x65, 0x72, 0x73, 0x6b, 0x69, 0x2f, 0x75, 0x6e, 0x63, 0x6c, 0x6f, + 0x75, 0x64, 0x2f, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x6d, 0x61, 0x63, 0x68, + 0x69, 0x6e, 0x65, 0x2f, 0x61, 0x70, 0x69, 0x2f, 0x70, 0x62, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, + 0x6f, 0x33, } var ( @@ -262,20 +377,24 @@ func file_internal_machine_api_pb_docker_proto_rawDescGZIP() []byte { return file_internal_machine_api_pb_docker_proto_rawDescData } -var file_internal_machine_api_pb_docker_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_internal_machine_api_pb_docker_proto_msgTypes = make([]protoimpl.MessageInfo, 5) var file_internal_machine_api_pb_docker_proto_goTypes = []any{ (*CreateContainerRequest)(nil), // 0: api.CreateContainerRequest (*CreateContainerResponse)(nil), // 1: api.CreateContainerResponse (*StartContainerRequest)(nil), // 2: api.StartContainerRequest - (*emptypb.Empty)(nil), // 3: google.protobuf.Empty + (*PullImageRequest)(nil), // 3: api.PullImageRequest + (*JSONMessage)(nil), // 4: api.JSONMessage + (*emptypb.Empty)(nil), // 5: google.protobuf.Empty } var file_internal_machine_api_pb_docker_proto_depIdxs = []int32{ 0, // 0: api.Docker.CreateContainer:input_type -> api.CreateContainerRequest 2, // 1: api.Docker.StartContainer:input_type -> api.StartContainerRequest - 1, // 2: api.Docker.CreateContainer:output_type -> api.CreateContainerResponse - 3, // 3: api.Docker.StartContainer:output_type -> google.protobuf.Empty - 2, // [2:4] is the sub-list for method output_type - 0, // [0:2] is the sub-list for method input_type + 3, // 2: api.Docker.PullImage:input_type -> api.PullImageRequest + 1, // 3: api.Docker.CreateContainer:output_type -> api.CreateContainerResponse + 5, // 4: api.Docker.StartContainer:output_type -> google.protobuf.Empty + 4, // 5: api.Docker.PullImage:output_type -> api.JSONMessage + 3, // [3:6] is the sub-list for method output_type + 0, // [0:3] is the sub-list for method input_type 0, // [0:0] is the sub-list for extension type_name 0, // [0:0] is the sub-list for extension extendee 0, // [0:0] is the sub-list for field type_name @@ -323,6 +442,30 @@ func file_internal_machine_api_pb_docker_proto_init() { return nil } } + file_internal_machine_api_pb_docker_proto_msgTypes[3].Exporter = func(v any, i int) any { + switch v := v.(*PullImageRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_internal_machine_api_pb_docker_proto_msgTypes[4].Exporter = func(v any, i int) any { + switch v := v.(*JSONMessage); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } } type x struct{} out := protoimpl.TypeBuilder{ @@ -330,7 +473,7 @@ func file_internal_machine_api_pb_docker_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_internal_machine_api_pb_docker_proto_rawDesc, NumEnums: 0, - NumMessages: 3, + NumMessages: 5, NumExtensions: 0, NumServices: 1, }, diff --git a/internal/machine/api/pb/docker.proto b/internal/machine/api/pb/docker.proto index ad185d82..d64eede0 100644 --- a/internal/machine/api/pb/docker.proto +++ b/internal/machine/api/pb/docker.proto @@ -9,6 +9,7 @@ import "google/protobuf/empty.proto"; service Docker { rpc CreateContainer(CreateContainerRequest) returns (CreateContainerResponse); rpc StartContainer(StartContainerRequest) returns (google.protobuf.Empty); + rpc PullImage(PullImageRequest) returns (stream JSONMessage); } message CreateContainerRequest { @@ -33,3 +34,14 @@ message StartContainerRequest { // JSON serialized container.StartOptions. bytes options = 2; } + +message PullImageRequest { + string image = 1; + // JSON serialized image.PullOptions. + bytes options = 2; +} + +message JSONMessage { + // JSON serialized jsonmessage.JSONMessage. + bytes message = 1; +} diff --git a/internal/machine/api/pb/docker_grpc.pb.go b/internal/machine/api/pb/docker_grpc.pb.go index 181ee500..0ab9557d 100644 --- a/internal/machine/api/pb/docker_grpc.pb.go +++ b/internal/machine/api/pb/docker_grpc.pb.go @@ -22,6 +22,7 @@ const _ = grpc.SupportPackageIsVersion9 const ( Docker_CreateContainer_FullMethodName = "/api.Docker/CreateContainer" Docker_StartContainer_FullMethodName = "/api.Docker/StartContainer" + Docker_PullImage_FullMethodName = "/api.Docker/PullImage" ) // DockerClient is the client API for Docker service. @@ -30,6 +31,7 @@ const ( type DockerClient interface { CreateContainer(ctx context.Context, in *CreateContainerRequest, opts ...grpc.CallOption) (*CreateContainerResponse, error) StartContainer(ctx context.Context, in *StartContainerRequest, opts ...grpc.CallOption) (*emptypb.Empty, error) + PullImage(ctx context.Context, in *PullImageRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JSONMessage], error) } type dockerClient struct { @@ -60,12 +62,32 @@ func (c *dockerClient) StartContainer(ctx context.Context, in *StartContainerReq return out, nil } +func (c *dockerClient) PullImage(ctx context.Context, in *PullImageRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[JSONMessage], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &Docker_ServiceDesc.Streams[0], Docker_PullImage_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[PullImageRequest, JSONMessage]{ClientStream: stream} + if err := x.ClientStream.SendMsg(in); err != nil { + return nil, err + } + if err := x.ClientStream.CloseSend(); err != nil { + return nil, err + } + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Docker_PullImageClient = grpc.ServerStreamingClient[JSONMessage] + // DockerServer is the server API for Docker service. // All implementations must embed UnimplementedDockerServer // for forward compatibility. type DockerServer interface { CreateContainer(context.Context, *CreateContainerRequest) (*CreateContainerResponse, error) StartContainer(context.Context, *StartContainerRequest) (*emptypb.Empty, error) + PullImage(*PullImageRequest, grpc.ServerStreamingServer[JSONMessage]) error mustEmbedUnimplementedDockerServer() } @@ -82,6 +104,9 @@ func (UnimplementedDockerServer) CreateContainer(context.Context, *CreateContain func (UnimplementedDockerServer) StartContainer(context.Context, *StartContainerRequest) (*emptypb.Empty, error) { return nil, status.Errorf(codes.Unimplemented, "method StartContainer not implemented") } +func (UnimplementedDockerServer) PullImage(*PullImageRequest, grpc.ServerStreamingServer[JSONMessage]) error { + return status.Errorf(codes.Unimplemented, "method PullImage not implemented") +} func (UnimplementedDockerServer) mustEmbedUnimplementedDockerServer() {} func (UnimplementedDockerServer) testEmbeddedByValue() {} @@ -139,6 +164,17 @@ func _Docker_StartContainer_Handler(srv interface{}, ctx context.Context, dec fu return interceptor(ctx, in, info, handler) } +func _Docker_PullImage_Handler(srv interface{}, stream grpc.ServerStream) error { + m := new(PullImageRequest) + if err := stream.RecvMsg(m); err != nil { + return err + } + return srv.(DockerServer).PullImage(m, &grpc.GenericServerStream[PullImageRequest, JSONMessage]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type Docker_PullImageServer = grpc.ServerStreamingServer[JSONMessage] + // Docker_ServiceDesc is the grpc.ServiceDesc for Docker service. // It's only intended for direct use with grpc.RegisterService, // and not to be introspected or modified (even as a copy) @@ -155,6 +191,12 @@ var Docker_ServiceDesc = grpc.ServiceDesc{ Handler: _Docker_StartContainer_Handler, }, }, - Streams: []grpc.StreamDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "PullImage", + Handler: _Docker_PullImage_Handler, + ServerStreams: true, + }, + }, Metadata: "internal/machine/api/pb/docker.proto", } diff --git a/internal/machine/docker/client.go b/internal/machine/docker/client.go index 090b2c5d..344df3ff 100644 --- a/internal/machine/docker/client.go +++ b/internal/machine/docker/client.go @@ -5,9 +5,15 @@ import ( "encoding/json" "fmt" "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/image" "github.com/docker/docker/api/types/network" + "github.com/docker/docker/errdefs" + "github.com/docker/docker/pkg/jsonmessage" ocispec "github.com/opencontainers/image-spec/specs-go/v1" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "io" "uncloud/internal/machine/api/pb" ) @@ -66,6 +72,11 @@ func (c *Client) CreateContainer( Name: name, }) if err != nil { + if s, ok := status.FromError(err); ok { + if s.Code() == codes.NotFound { + return resp, errdefs.NotFound(err) + } + } return resp, err } @@ -76,15 +87,63 @@ func (c *Client) CreateContainer( } // StartContainer starts a container with the given ID and options. -func (c *Client) StartContainer(ctx context.Context, id string, options container.StartOptions) error { - optionsBytes, err := json.Marshal(options) +func (c *Client) StartContainer(ctx context.Context, id string, opts container.StartOptions) error { + optsBytes, err := json.Marshal(opts) if err != nil { return fmt.Errorf("marshal start options: %w", err) } _, err = c.grpcClient.StartContainer(ctx, &pb.StartContainerRequest{ Id: id, - Options: optionsBytes, + Options: optsBytes, }) return err } + +type PullImageMessage struct { + Message jsonmessage.JSONMessage + Err error +} + +func (c *Client) PullImage( + ctx context.Context, image string, opts image.PullOptions, +) (<-chan PullImageMessage, error) { + optsBytes, err := json.Marshal(opts) + if err != nil { + return nil, fmt.Errorf("marshal pull options: %w", err) + } + + stream, err := c.grpcClient.PullImage(ctx, &pb.PullImageRequest{ + Image: image, + Options: optsBytes, + }) + if err != nil { + return nil, err + } + + ch := make(chan PullImageMessage) + + go func() { + defer close(ch) + + for { + msg, err := stream.Recv() + if err == io.EOF { + return + } + if err != nil { + ch <- PullImageMessage{Err: err} + return + } + + var jm jsonmessage.JSONMessage + if err = json.Unmarshal(msg.Message, &jm); err != nil { + ch <- PullImageMessage{Err: fmt.Errorf("unmarshal JSON message: %w", err)} + return + } + ch <- PullImageMessage{Message: jm} + } + }() + + return ch, nil +} diff --git a/internal/machine/docker/server.go b/internal/machine/docker/server.go index 8dc9b517..1133ebee 100644 --- a/internal/machine/docker/server.go +++ b/internal/machine/docker/server.go @@ -3,13 +3,17 @@ package docker import ( "context" "encoding/json" + "errors" "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/image" "github.com/docker/docker/api/types/network" "github.com/docker/docker/client" ocispec "github.com/opencontainers/image-spec/specs-go/v1" + "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/emptypb" + "io" "uncloud/internal/machine/api/pb" ) @@ -47,6 +51,9 @@ func (s *Server) CreateContainer(ctx context.Context, req *pb.CreateContainerReq resp, err := s.client.ContainerCreate(ctx, &config, &hostConfig, &networkConfig, &platform, req.Name) if err != nil { + if client.IsErrNotFound(err) { + return nil, status.Errorf(codes.NotFound, "create container: %v", err) + } return nil, status.Errorf(codes.Internal, "create container: %v", err) } @@ -60,16 +67,65 @@ func (s *Server) CreateContainer(ctx context.Context, req *pb.CreateContainerReq // StartContainer starts a container with the given ID and options. func (s *Server) StartContainer(ctx context.Context, req *pb.StartContainerRequest) (*emptypb.Empty, error) { - var options container.StartOptions + var opts container.StartOptions if len(req.Options) > 0 { - if err := json.Unmarshal(req.Options, &options); err != nil { + if err := json.Unmarshal(req.Options, &opts); err != nil { return nil, status.Errorf(codes.InvalidArgument, "unmarshal start options: %v", err) } } - if err := s.client.ContainerStart(ctx, req.Id, options); err != nil { + if err := s.client.ContainerStart(ctx, req.Id, opts); err != nil { return nil, status.Errorf(codes.Internal, "start container: %v", err) } return &emptypb.Empty{}, nil } + +func (s *Server) PullImage( + ctx context.Context, req *pb.PullImageRequest, stream grpc.ServerStreamingServer[pb.JSONMessage], +) error { + var opts image.PullOptions + if len(req.Options) > 0 { + if err := json.Unmarshal(req.Options, &opts); err != nil { + return status.Errorf(codes.InvalidArgument, "unmarshal pull options: %v", err) + } + } + + respBody, err := s.client.ImagePull(ctx, req.Image, opts) + if err != nil { + return status.Errorf(codes.Internal, "pull image: %v", err) + } + defer respBody.Close() + + decoder := json.NewDecoder(respBody) + errCh := make(chan error, 1) + + go func() { + var raw json.RawMessage + for { + if err = decoder.Decode(&raw); err != nil { + if errors.Is(err, io.EOF) { + errCh <- nil + return + } + errCh <- status.Errorf(codes.Internal, "decode image pull message: %v", err) + return + } + + if err = stream.Send(&pb.JSONMessage{Message: raw}); err != nil { + errCh <- status.Errorf(codes.Internal, "send image pull message to stream: %v", err) + return + } + } + }() + + for { + select { + case err = <-errCh: + return err + case <-ctx.Done(): + return status.Errorf(codes.Canceled, "pull image: %v", ctx.Err()) + } + } + +}