github/containerd
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Merge pull request #1777 from jessvalarezo/migrate-client-code

Browse Source 
client: migrate client code to containerd
This commit is contained in:
Michael Crosby
2017-11-17 09:32:34 -05:00
committed by GitHub
parent 77cb722e1c 8c5dede74e
commit a4968710fd
12 changed files with 245 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
services/content/reader.go
View File

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

20
services/content/service.go
View File

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

208
services/content/store.go
View File

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

123
services/content/writer.go
View File

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

83
services/diff/client.go
View File

@@ -1,83 +0,0 @@
package diff
import (
diffapi "github.com/containerd/containerd/api/services/diff/v1"
"github.com/containerd/containerd/api/types"
"github.com/containerd/containerd/diff"
"github.com/containerd/containerd/mount"
ocispec "github.com/opencontainers/image-spec/specs-go/v1"
"golang.org/x/net/context"
)
// NewDiffServiceFromClient returns a new diff service which communicates
// over a GRPC connection.
func NewDiffServiceFromClient(client diffapi.DiffClient) diff.Differ {
return &remote{
client: client,
}
}
type remote struct {
client diffapi.DiffClient
}
func (r *remote) Apply(ctx context.Context, diff ocispec.Descriptor, mounts []mount.Mount) (ocispec.Descriptor, error) {
req := &diffapi.ApplyRequest{
Diff: fromDescriptor(diff),
Mounts: fromMounts(mounts),
}
resp, err := r.client.Apply(ctx, req)
if err != nil {
return ocispec.Descriptor{}, err
}
return toDescriptor(resp.Applied), nil
}
func (r *remote) DiffMounts(ctx context.Context, a, b []mount.Mount, opts ...diff.Opt) (ocispec.Descriptor, error) {
var config diff.Config
for _, opt := range opts {
if err := opt(&config); err != nil {
return ocispec.Descriptor{}, err
}
}
req := &diffapi.DiffRequest{
Left: fromMounts(a),
Right: fromMounts(b),
MediaType: config.MediaType,
Ref: config.Reference,
Labels: config.Labels,
}
resp, err := r.client.Diff(ctx, req)
if err != nil {
return ocispec.Descriptor{}, err
}
return toDescriptor(resp.Diff), nil
}
func toDescriptor(d *types.Descriptor) ocispec.Descriptor {
return ocispec.Descriptor{
MediaType: d.MediaType,
Digest: d.Digest,
Size: d.Size_,
}
}
func fromDescriptor(d ocispec.Descriptor) *types.Descriptor {
return &types.Descriptor{
MediaType: d.MediaType,
Digest: d.Digest,
Size_: d.Size,
}
}
func fromMounts(mounts []mount.Mount) []*types.Mount {
apiMounts := make([]*types.Mount, len(mounts))
for i, m := range mounts {
apiMounts[i] = &types.Mount{
Type: m.Type,
Source: m.Source,
Options: m.Options,
}
}
return apiMounts
}

16
services/diff/service.go
View File

@@ -140,3 +140,19 @@ func toMounts(apim []*types.Mount) []mount.Mount {
}
return mounts
}
func toDescriptor(d *types.Descriptor) ocispec.Descriptor {
return ocispec.Descriptor{
MediaType: d.MediaType,
Digest: d.Digest,
Size: d.Size_,
}
}
func fromDescriptor(d ocispec.Descriptor) *types.Descriptor {
return &types.Descriptor{
MediaType: d.MediaType,
Digest: d.Digest,
Size_: d.Size,
}
}

81
services/images/client.go
View File

@@ -1,81 +0,0 @@
package images
import (
"context"
imagesapi "github.com/containerd/containerd/api/services/images/v1"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/images"
ptypes "github.com/gogo/protobuf/types"
)
type remoteStore struct {
client imagesapi.ImagesClient
}
// NewStoreFromClient returns a new image store client
func NewStoreFromClient(client imagesapi.ImagesClient) images.Store {
return &remoteStore{
client: client,
}
}
func (s *remoteStore) Get(ctx context.Context, name string) (images.Image, error) {
resp, err := s.client.Get(ctx, &imagesapi.GetImageRequest{
Name: name,
})
if err != nil {
return images.Image{}, errdefs.FromGRPC(err)
}
return imageFromProto(resp.Image), nil
}
func (s *remoteStore) List(ctx context.Context, filters ...string) ([]images.Image, error) {
resp, err := s.client.List(ctx, &imagesapi.ListImagesRequest{
Filters: filters,
})
if err != nil {
return nil, errdefs.FromGRPC(err)
}
return imagesFromProto(resp.Images), nil
}
func (s *remoteStore) Create(ctx context.Context, image images.Image) (images.Image, error) {
created, err := s.client.Create(ctx, &imagesapi.CreateImageRequest{
Image: imageToProto(&image),
})
if err != nil {
return images.Image{}, errdefs.FromGRPC(err)
}
return imageFromProto(&created.Image), nil
}
func (s *remoteStore) Update(ctx context.Context, image images.Image, fieldpaths ...string) (images.Image, error) {
var updateMask *ptypes.FieldMask
if len(fieldpaths) > 0 {
updateMask = &ptypes.FieldMask{
Paths: fieldpaths,
}
}
updated, err := s.client.Update(ctx, &imagesapi.UpdateImageRequest{
Image: imageToProto(&image),
UpdateMask: updateMask,
})
if err != nil {
return images.Image{}, errdefs.FromGRPC(err)
}
return imageFromProto(&updated.Image), nil
}
func (s *remoteStore) Delete(ctx context.Context, name string) error {
_, err := s.client.Delete(ctx, &imagesapi.DeleteImageRequest{
Name: name,
})
return errdefs.FromGRPC(err)
}

97
services/namespaces/client.go
View File

@@ -1,97 +0,0 @@
package namespaces
import (
"context"
"strings"
api "github.com/containerd/containerd/api/services/namespaces/v1"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/namespaces"
"github.com/gogo/protobuf/types"
)
// NewStoreFromClient returns a new namespace store
func NewStoreFromClient(client api.NamespacesClient) namespaces.Store {
return &remote{client: client}
}
type remote struct {
client api.NamespacesClient
}
func (r *remote) Create(ctx context.Context, namespace string, labels map[string]string) error {
var req api.CreateNamespaceRequest
req.Namespace = api.Namespace{
Name: namespace,
Labels: labels,
}
_, err := r.client.Create(ctx, &req)
if err != nil {
return errdefs.FromGRPC(err)
}
return nil
}
func (r *remote) Labels(ctx context.Context, namespace string) (map[string]string, error) {
var req api.GetNamespaceRequest
req.Name = namespace
resp, err := r.client.Get(ctx, &req)
if err != nil {
return nil, errdefs.FromGRPC(err)
}
return resp.Namespace.Labels, nil
}
func (r *remote) SetLabel(ctx context.Context, namespace, key, value string) error {
var req api.UpdateNamespaceRequest
req.Namespace = api.Namespace{
Name: namespace,
Labels: map[string]string{key: value},
}
req.UpdateMask = &types.FieldMask{
Paths: []string{strings.Join([]string{"labels", key}, ".")},
}
_, err := r.client.Update(ctx, &req)
if err != nil {
return errdefs.FromGRPC(err)
}
return nil
}
func (r *remote) List(ctx context.Context) ([]string, error) {
var req api.ListNamespacesRequest
resp, err := r.client.List(ctx, &req)
if err != nil {
return nil, errdefs.FromGRPC(err)
}
var namespaces []string
for _, ns := range resp.Namespaces {
namespaces = append(namespaces, ns.Name)
}
return namespaces, nil
}
func (r *remote) Delete(ctx context.Context, namespace string) error {
var req api.DeleteNamespaceRequest
req.Name = namespace
_, err := r.client.Delete(ctx, &req)
if err != nil {
return errdefs.FromGRPC(err)
}
return nil
}

208
services/snapshot/client.go
View File

@@ -1,208 +0,0 @@
package snapshot
import (
"context"
"io"
snapshotapi "github.com/containerd/containerd/api/services/snapshot/v1"
"github.com/containerd/containerd/api/types"
"github.com/containerd/containerd/errdefs"
"github.com/containerd/containerd/mount"
"github.com/containerd/containerd/snapshot"
protobuftypes "github.com/gogo/protobuf/types"
)
// NewSnapshotterFromClient returns a new Snapshotter which communicates
// over a GRPC connection.
func NewSnapshotterFromClient(client snapshotapi.SnapshotsClient, snapshotterName string) snapshot.Snapshotter {
return &remoteSnapshotter{
client: client,
snapshotterName: snapshotterName,
}
}
type remoteSnapshotter struct {
client snapshotapi.SnapshotsClient
snapshotterName string
}
func (r *remoteSnapshotter) Stat(ctx context.Context, key string) (snapshot.Info, error) {
resp, err := r.client.Stat(ctx,
&snapshotapi.StatSnapshotRequest{
Snapshotter: r.snapshotterName,
Key: key,
})
if err != nil {
return snapshot.Info{}, errdefs.FromGRPC(err)
}
return toInfo(resp.Info), nil
}
func (r *remoteSnapshotter) Update(ctx context.Context, info snapshot.Info, fieldpaths ...string) (snapshot.Info, error) {
resp, err := r.client.Update(ctx,
&snapshotapi.UpdateSnapshotRequest{
Snapshotter: r.snapshotterName,
Info: fromInfo(info),
UpdateMask: &protobuftypes.FieldMask{
Paths: fieldpaths,
},
})
if err != nil {
return snapshot.Info{}, errdefs.FromGRPC(err)
}
return toInfo(resp.Info), nil
}
func (r *remoteSnapshotter) Usage(ctx context.Context, key string) (snapshot.Usage, error) {
resp, err := r.client.Usage(ctx, &snapshotapi.UsageRequest{
Snapshotter: r.snapshotterName,
Key: key,
})
if err != nil {
return snapshot.Usage{}, errdefs.FromGRPC(err)
}
return toUsage(resp), nil
}
func (r *remoteSnapshotter) Mounts(ctx context.Context, key string) ([]mount.Mount, error) {
resp, err := r.client.Mounts(ctx, &snapshotapi.MountsRequest{
Snapshotter: r.snapshotterName,
Key: key,
})
if err != nil {
return nil, errdefs.FromGRPC(err)
}
return toMounts(resp.Mounts), nil
}
func (r *remoteSnapshotter) Prepare(ctx context.Context, key, parent string, opts ...snapshot.Opt) ([]mount.Mount, error) {
var local snapshot.Info
for _, opt := range opts {
if err := opt(&local); err != nil {
return nil, err
}
}
resp, err := r.client.Prepare(ctx, &snapshotapi.PrepareSnapshotRequest{
Snapshotter: r.snapshotterName,
Key: key,
Parent: parent,
Labels: local.Labels,
})
if err != nil {
return nil, errdefs.FromGRPC(err)
}
return toMounts(resp.Mounts), nil
}
func (r *remoteSnapshotter) View(ctx context.Context, key, parent string, opts ...snapshot.Opt) ([]mount.Mount, error) {
var local snapshot.Info
for _, opt := range opts {
if err := opt(&local); err != nil {
return nil, err
}
}
resp, err := r.client.View(ctx, &snapshotapi.ViewSnapshotRequest{
Snapshotter: r.snapshotterName,
Key: key,
Parent: parent,
Labels: local.Labels,
})
if err != nil {
return nil, errdefs.FromGRPC(err)
}
return toMounts(resp.Mounts), nil
}
func (r *remoteSnapshotter) Commit(ctx context.Context, name, key string, opts ...snapshot.Opt) error {
var local snapshot.Info
for _, opt := range opts {
if err := opt(&local); err != nil {
return err
}
}
_, err := r.client.Commit(ctx, &snapshotapi.CommitSnapshotRequest{
Snapshotter: r.snapshotterName,
Name: name,
Key: key,
Labels: local.Labels,
})
return errdefs.FromGRPC(err)
}
func (r *remoteSnapshotter) Remove(ctx context.Context, key string) error {
_, err := r.client.Remove(ctx, &snapshotapi.RemoveSnapshotRequest{
Snapshotter: r.snapshotterName,
Key: key,
})
return errdefs.FromGRPC(err)
}
func (r *remoteSnapshotter) Walk(ctx context.Context, fn func(context.Context, snapshot.Info) error) error {
sc, err := r.client.List(ctx, &snapshotapi.ListSnapshotsRequest{
Snapshotter: r.snapshotterName,
})
if err != nil {
return errdefs.FromGRPC(err)
}
for {
resp, err := sc.Recv()
if err != nil {
if err == io.EOF {
return nil
}
return errdefs.FromGRPC(err)
}
if resp == nil {
return nil
}
for _, info := range resp.Info {
if err := fn(ctx, toInfo(info)); err != nil {
return err
}
}
}
}
func (r *remoteSnapshotter) Close() error {
return nil
}
func toKind(kind snapshotapi.Kind) snapshot.Kind {
if kind == snapshotapi.KindActive {
return snapshot.KindActive
}
if kind == snapshotapi.KindView {
return snapshot.KindView
}
return snapshot.KindCommitted
}
func toInfo(info snapshotapi.Info) snapshot.Info {
return snapshot.Info{
Name: info.Name,
Parent: info.Parent,
Kind: toKind(info.Kind),
Created: info.CreatedAt,
Updated: info.UpdatedAt,
Labels: info.Labels,
}
}
func toUsage(resp *snapshotapi.UsageResponse) snapshot.Usage {
return snapshot.Usage{
Inodes: resp.Inodes,
Size: resp.Size_,
}
}
func toMounts(mm []*types.Mount) []mount.Mount {
mounts := make([]mount.Mount, len(mm))
for i, m := range mm {
mounts[i] = mount.Mount{
Type: m.Type,
Source: m.Source,
Options: m.Options,
}
}
return mounts
}

21
services/snapshot/service.go
View File

@@ -293,3 +293,24 @@ func fromMounts(mounts []mount.Mount) []*types.Mount {
}
return out
}
func toInfo(info snapshotapi.Info) snapshot.Info {
return snapshot.Info{
Name: info.Name,
Parent: info.Parent,
Kind: toKind(info.Kind),
Created: info.CreatedAt,
Updated: info.UpdatedAt,
Labels: info.Labels,
}
}
func toKind(kind snapshotapi.Kind) snapshot.Kind {
if kind == snapshotapi.KindActive {
return snapshot.KindActive
}
if kind == snapshotapi.KindView {
return snapshot.KindView
}
return snapshot.KindCommitted
}