client: move content store client to containerd package
Signed-off-by: Jess Valarezo <valarezo.jessica@gmail.com>
This commit is contained in:
Jess Valarezo
@@ -1,49 +0,0 @@
|
||||
package content
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
contentapi "github.com/containerd/containerd/api/services/content/v1"
|
||||
digest "github.com/opencontainers/go-digest"
|
||||
)
|
||||
|
||||
type remoteReaderAt struct {
|
||||
ctx context.Context
|
||||
digest digest.Digest
|
||||
size int64
|
||||
client contentapi.ContentClient
|
||||
}
|
||||
|
||||
func (ra *remoteReaderAt) Size() int64 {
|
||||
return ra.size
|
||||
}
|
||||
|
||||
func (ra *remoteReaderAt) ReadAt(p []byte, off int64) (n int, err error) {
|
||||
rr := &contentapi.ReadContentRequest{
|
||||
Digest: ra.digest,
|
||||
Offset: off,
|
||||
Size_: int64(len(p)),
|
||||
}
|
||||
rc, err := ra.client.Read(ra.ctx, rr)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
for len(p) > 0 {
|
||||
var resp *contentapi.ReadContentResponse
|
||||
// fill our buffer up until we can fill p.
|
||||
resp, err = rc.Recv()
|
||||
if err != nil {
|
||||
return n, err
|
||||
}
|
||||
|
||||
copied := copy(p, resp.Data)
|
||||
n += copied
|
||||
p = p[copied:]
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (ra *remoteReaderAt) Close() error {
|
||||
return nil
|
||||
}
|
||||
@@ -452,3 +452,23 @@ func (s *service) Abort(ctx context.Context, req *api.AbortRequest) (*ptypes.Emp
|
||||
|
||||
return &ptypes.Empty{}, nil
|
||||
}
|
||||
|
||||
func infoToGRPC(info content.Info) api.Info {
|
||||
return api.Info{
|
||||
Digest: info.Digest,
|
||||
Size_: info.Size,
|
||||
CreatedAt: info.CreatedAt,
|
||||
UpdatedAt: info.UpdatedAt,
|
||||
Labels: info.Labels,
|
||||
}
|
||||
}
|
||||
|
||||
func infoFromGRPC(info api.Info) content.Info {
|
||||
return content.Info{
|
||||
Digest: info.Digest,
|
||||
Size: info.Size_,
|
||||
CreatedAt: info.CreatedAt,
|
||||
UpdatedAt: info.UpdatedAt,
|
||||
Labels: info.Labels,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,208 +0,0 @@
|
||||
package content
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
|
||||
contentapi "github.com/containerd/containerd/api/services/content/v1"
|
||||
"github.com/containerd/containerd/content"
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
protobuftypes "github.com/gogo/protobuf/types"
|
||||
digest "github.com/opencontainers/go-digest"
|
||||
)
|
||||
|
||||
type remoteStore struct {
|
||||
client contentapi.ContentClient
|
||||
}
|
||||
|
||||
// NewStoreFromClient returns a new content store
|
||||
func NewStoreFromClient(client contentapi.ContentClient) content.Store {
|
||||
return &remoteStore{
|
||||
client: client,
|
||||
}
|
||||
}
|
||||
|
||||
func (rs *remoteStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) {
|
||||
resp, err := rs.client.Info(ctx, &contentapi.InfoRequest{
|
||||
Digest: dgst,
|
||||
})
|
||||
if err != nil {
|
||||
return content.Info{}, errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
return infoFromGRPC(resp.Info), nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) Walk(ctx context.Context, fn content.WalkFunc, filters ...string) error {
|
||||
session, err := rs.client.List(ctx, &contentapi.ListContentRequest{
|
||||
Filters: filters,
|
||||
})
|
||||
if err != nil {
|
||||
return errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
for {
|
||||
msg, err := session.Recv()
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
return errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
break
|
||||
}
|
||||
|
||||
for _, info := range msg.Info {
|
||||
if err := fn(infoFromGRPC(info)); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) Delete(ctx context.Context, dgst digest.Digest) error {
|
||||
if _, err := rs.client.Delete(ctx, &contentapi.DeleteContentRequest{
|
||||
Digest: dgst,
|
||||
}); err != nil {
|
||||
return errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) ReaderAt(ctx context.Context, dgst digest.Digest) (content.ReaderAt, error) {
|
||||
i, err := rs.Info(ctx, dgst)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &remoteReaderAt{
|
||||
ctx: ctx,
|
||||
digest: dgst,
|
||||
size: i.Size,
|
||||
client: rs.client,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) Status(ctx context.Context, ref string) (content.Status, error) {
|
||||
resp, err := rs.client.Status(ctx, &contentapi.StatusRequest{
|
||||
Ref: ref,
|
||||
})
|
||||
if err != nil {
|
||||
return content.Status{}, errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
status := resp.Status
|
||||
return content.Status{
|
||||
Ref: status.Ref,
|
||||
StartedAt: status.StartedAt,
|
||||
UpdatedAt: status.UpdatedAt,
|
||||
Offset: status.Offset,
|
||||
Total: status.Total,
|
||||
Expected: status.Expected,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) Update(ctx context.Context, info content.Info, fieldpaths ...string) (content.Info, error) {
|
||||
resp, err := rs.client.Update(ctx, &contentapi.UpdateRequest{
|
||||
Info: infoToGRPC(info),
|
||||
UpdateMask: &protobuftypes.FieldMask{
|
||||
Paths: fieldpaths,
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
return content.Info{}, errdefs.FromGRPC(err)
|
||||
}
|
||||
return infoFromGRPC(resp.Info), nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) ListStatuses(ctx context.Context, filters ...string) ([]content.Status, error) {
|
||||
resp, err := rs.client.ListStatuses(ctx, &contentapi.ListStatusesRequest{
|
||||
Filters: filters,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
var statuses []content.Status
|
||||
for _, status := range resp.Statuses {
|
||||
statuses = append(statuses, content.Status{
|
||||
Ref: status.Ref,
|
||||
StartedAt: status.StartedAt,
|
||||
UpdatedAt: status.UpdatedAt,
|
||||
Offset: status.Offset,
|
||||
Total: status.Total,
|
||||
Expected: status.Expected,
|
||||
})
|
||||
}
|
||||
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) Writer(ctx context.Context, ref string, size int64, expected digest.Digest) (content.Writer, error) {
|
||||
wrclient, offset, err := rs.negotiate(ctx, ref, size, expected)
|
||||
if err != nil {
|
||||
return nil, errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
return &remoteWriter{
|
||||
ref: ref,
|
||||
client: wrclient,
|
||||
offset: offset,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// Abort implements asynchronous abort. It starts a new write session on the ref l
|
||||
func (rs *remoteStore) Abort(ctx context.Context, ref string) error {
|
||||
if _, err := rs.client.Abort(ctx, &contentapi.AbortRequest{
|
||||
Ref: ref,
|
||||
}); err != nil {
|
||||
return errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rs *remoteStore) negotiate(ctx context.Context, ref string, size int64, expected digest.Digest) (contentapi.Content_WriteClient, int64, error) {
|
||||
wrclient, err := rs.client.Write(ctx)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
if err := wrclient.Send(&contentapi.WriteContentRequest{
|
||||
Action: contentapi.WriteActionStat,
|
||||
Ref: ref,
|
||||
Total: size,
|
||||
Expected: expected,
|
||||
}); err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
resp, err := wrclient.Recv()
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
|
||||
return wrclient, resp.Offset, nil
|
||||
}
|
||||
|
||||
func infoToGRPC(info content.Info) contentapi.Info {
|
||||
return contentapi.Info{
|
||||
Digest: info.Digest,
|
||||
Size_: info.Size,
|
||||
CreatedAt: info.CreatedAt,
|
||||
UpdatedAt: info.UpdatedAt,
|
||||
Labels: info.Labels,
|
||||
}
|
||||
}
|
||||
|
||||
func infoFromGRPC(info contentapi.Info) content.Info {
|
||||
return content.Info{
|
||||
Digest: info.Digest,
|
||||
Size: info.Size_,
|
||||
CreatedAt: info.CreatedAt,
|
||||
UpdatedAt: info.UpdatedAt,
|
||||
Labels: info.Labels,
|
||||
}
|
||||
}
|
||||
@@ -1,123 +0,0 @@
|
||||
package content
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
|
||||
contentapi "github.com/containerd/containerd/api/services/content/v1"
|
||||
"github.com/containerd/containerd/content"
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
digest "github.com/opencontainers/go-digest"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
type remoteWriter struct {
|
||||
ref string
|
||||
client contentapi.Content_WriteClient
|
||||
offset int64
|
||||
digest digest.Digest
|
||||
}
|
||||
|
||||
// send performs a synchronous req-resp cycle on the client.
|
||||
func (rw *remoteWriter) send(req *contentapi.WriteContentRequest) (*contentapi.WriteContentResponse, error) {
|
||||
if err := rw.client.Send(req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err := rw.client.Recv()
|
||||
|
||||
if err == nil {
|
||||
// try to keep these in sync
|
||||
if resp.Digest != "" {
|
||||
rw.digest = resp.Digest
|
||||
}
|
||||
}
|
||||
|
||||
return resp, err
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Status() (content.Status, error) {
|
||||
resp, err := rw.send(&contentapi.WriteContentRequest{
|
||||
Action: contentapi.WriteActionStat,
|
||||
})
|
||||
if err != nil {
|
||||
return content.Status{}, errors.Wrap(err, "error getting writer status")
|
||||
}
|
||||
|
||||
return content.Status{
|
||||
Ref: rw.ref,
|
||||
Offset: resp.Offset,
|
||||
Total: resp.Total,
|
||||
StartedAt: resp.StartedAt,
|
||||
UpdatedAt: resp.UpdatedAt,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Digest() digest.Digest {
|
||||
return rw.digest
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Write(p []byte) (n int, err error) {
|
||||
offset := rw.offset
|
||||
|
||||
resp, err := rw.send(&contentapi.WriteContentRequest{
|
||||
Action: contentapi.WriteActionWrite,
|
||||
Offset: offset,
|
||||
Data: p,
|
||||
})
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
n = int(resp.Offset - offset)
|
||||
if n < len(p) {
|
||||
err = io.ErrShortWrite
|
||||
}
|
||||
|
||||
rw.offset += int64(n)
|
||||
if resp.Digest != "" {
|
||||
rw.digest = resp.Digest
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error {
|
||||
var base content.Info
|
||||
for _, opt := range opts {
|
||||
if err := opt(&base); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
resp, err := rw.send(&contentapi.WriteContentRequest{
|
||||
Action: contentapi.WriteActionCommit,
|
||||
Total: size,
|
||||
Offset: rw.offset,
|
||||
Expected: expected,
|
||||
Labels: base.Labels,
|
||||
})
|
||||
if err != nil {
|
||||
return errdefs.FromGRPC(err)
|
||||
}
|
||||
|
||||
if size != 0 && resp.Offset != size {
|
||||
return errors.Errorf("unexpected size: %v != %v", resp.Offset, size)
|
||||
}
|
||||
|
||||
if expected != "" && resp.Digest != expected {
|
||||
return errors.Errorf("unexpected digest: %v != %v", resp.Digest, expected)
|
||||
}
|
||||
|
||||
rw.digest = resp.Digest
|
||||
rw.offset = resp.Offset
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Truncate(size int64) error {
|
||||
// This truncation won't actually be validated until a write is issued.
|
||||
rw.offset = size
|
||||
return nil
|
||||
}
|
||||
|
||||
func (rw *remoteWriter) Close() error {
|
||||
return rw.client.CloseSend()
|
||||
}
|
||||
Reference in New Issue
Block a user