leases: support resource management
Add three methods for lease service so that the client can use it to
manage the resource by lease, not just gc.root label. With the following
methods, it is easy for client to maintain their own cache system.
```
- AddResource(context.Context, Lease, Resource) error
- RemoveResource(context.Context, Lease, Resource) error
- ListResources(context.Context, Lease) ([]Resource, error)
```
And the resource is to be
```golang
type Resource {
ID string
Type string
}
```
For the snapshots, the Type field will be formatted by
snapshots/%{type}, like snapshots/overlayfs.
fix: #3295
Signed-off-by: Wei Fu <fuweid89@gmail.com>
This commit is contained in:
Wei Fu
@@ -18,6 +18,8 @@ package metadata
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/containerd/containerd/errdefs"
|
||||
@@ -167,6 +169,128 @@ func (lm *LeaseManager) List(ctx context.Context, fs ...string) ([]leases.Lease,
|
||||
return ll, nil
|
||||
}
|
||||
|
||||
// AddResource references the resource by the provided lease.
|
||||
func (lm *LeaseManager) AddResource(ctx context.Context, lease leases.Lease, r leases.Resource) error {
|
||||
namespace, err := namespaces.NamespaceRequired(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
topbkt := getBucket(lm.tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases, []byte(lease.ID))
|
||||
if topbkt == nil {
|
||||
return errors.Wrapf(errdefs.ErrNotFound, "lease %q", lease.ID)
|
||||
}
|
||||
|
||||
keys, ref, err := parseLeaseResource(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
bkt := topbkt
|
||||
for _, key := range keys {
|
||||
bkt, err = bkt.CreateBucketIfNotExists([]byte(key))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return bkt.Put([]byte(ref), nil)
|
||||
}
|
||||
|
||||
// DeleteResource dereferences the resource by the provided lease.
|
||||
func (lm *LeaseManager) DeleteResource(ctx context.Context, lease leases.Lease, r leases.Resource) error {
|
||||
namespace, err := namespaces.NamespaceRequired(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
topbkt := getBucket(lm.tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases, []byte(lease.ID))
|
||||
if topbkt == nil {
|
||||
return errors.Wrapf(errdefs.ErrNotFound, "lease %q", lease.ID)
|
||||
}
|
||||
|
||||
keys, ref, err := parseLeaseResource(r)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
bkt := topbkt
|
||||
for _, key := range keys {
|
||||
if bkt == nil {
|
||||
break
|
||||
}
|
||||
bkt = bkt.Bucket([]byte(key))
|
||||
}
|
||||
|
||||
if bkt == nil {
|
||||
return nil
|
||||
}
|
||||
return bkt.Delete([]byte(ref))
|
||||
}
|
||||
|
||||
// ListResources lists all the resources referenced by the lease.
|
||||
func (lm *LeaseManager) ListResources(ctx context.Context, lease leases.Lease) ([]leases.Resource, error) {
|
||||
namespace, err := namespaces.NamespaceRequired(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
topbkt := getBucket(lm.tx, bucketKeyVersion, []byte(namespace), bucketKeyObjectLeases, []byte(lease.ID))
|
||||
if topbkt == nil {
|
||||
return nil, errors.Wrapf(errdefs.ErrNotFound, "lease %q", lease.ID)
|
||||
}
|
||||
|
||||
rs := make([]leases.Resource, 0)
|
||||
|
||||
// content resources
|
||||
if cbkt := topbkt.Bucket(bucketKeyObjectContent); cbkt != nil {
|
||||
if err := cbkt.ForEach(func(k, _ []byte) error {
|
||||
rs = append(rs, leases.Resource{
|
||||
ID: string(k),
|
||||
Type: string(bucketKeyObjectContent),
|
||||
})
|
||||
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// ingest resources
|
||||
if lbkt := topbkt.Bucket(bucketKeyObjectIngests); lbkt != nil {
|
||||
if err := lbkt.ForEach(func(k, _ []byte) error {
|
||||
rs = append(rs, leases.Resource{
|
||||
ID: string(k),
|
||||
Type: string(bucketKeyObjectIngests),
|
||||
})
|
||||
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// snapshot resources
|
||||
if sbkt := topbkt.Bucket(bucketKeyObjectSnapshots); sbkt != nil {
|
||||
if err := sbkt.ForEach(func(sk, sv []byte) error {
|
||||
if sv != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
snbkt := sbkt.Bucket(sk)
|
||||
return snbkt.ForEach(func(k, _ []byte) error {
|
||||
rs = append(rs, leases.Resource{
|
||||
ID: string(k),
|
||||
Type: fmt.Sprintf("%s/%s", bucketKeyObjectSnapshots, sk),
|
||||
})
|
||||
return nil
|
||||
})
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return rs, nil
|
||||
}
|
||||
|
||||
func addSnapshotLease(ctx context.Context, tx *bolt.Tx, snapshotter, key string) error {
|
||||
lid, ok := leases.FromContext(ctx)
|
||||
if !ok {
|
||||
@@ -307,3 +431,36 @@ func removeIngestLease(ctx context.Context, tx *bolt.Tx, ref string) error {
|
||||
|
||||
return bkt.Delete([]byte(ref))
|
||||
}
|
||||
|
||||
func parseLeaseResource(r leases.Resource) ([]string, string, error) {
|
||||
var (
|
||||
ref = r.ID
|
||||
typ = r.Type
|
||||
keys = strings.Split(typ, "/")
|
||||
)
|
||||
|
||||
switch k := keys[0]; k {
|
||||
case string(bucketKeyObjectContent),
|
||||
string(bucketKeyObjectIngests):
|
||||
|
||||
if len(keys) != 1 {
|
||||
return nil, "", errors.Wrapf(errdefs.ErrInvalidArgument, "invalid resource type %s", typ)
|
||||
}
|
||||
|
||||
if k == string(bucketKeyObjectContent) {
|
||||
dgst, err := digest.Parse(ref)
|
||||
if err != nil {
|
||||
return nil, "", errors.Wrapf(errdefs.ErrInvalidArgument, "invalid content resource id %s: %v", ref, err)
|
||||
}
|
||||
ref = dgst.String()
|
||||
}
|
||||
case string(bucketKeyObjectSnapshots):
|
||||
if len(keys) != 2 {
|
||||
return nil, "", errors.Wrapf(errdefs.ErrInvalidArgument, "invalid snapshot resource type %s", typ)
|
||||
}
|
||||
default:
|
||||
return nil, "", errors.Wrapf(errdefs.ErrNotImplemented, "resource type %s not supported yet", typ)
|
||||
}
|
||||
|
||||
return keys, ref, nil
|
||||
}
|
||||
|
||||
@@ -264,3 +264,181 @@ func TestLeasesList(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestLeaseResource(t *testing.T) {
|
||||
ctx, db, cancel := testEnv(t)
|
||||
defer cancel()
|
||||
|
||||
var (
|
||||
leaseID = "l1"
|
||||
|
||||
lease = leases.Lease{
|
||||
ID: leaseID,
|
||||
}
|
||||
|
||||
snapshotterKey = "RstMI3X8vguKoPFkmIStZ5fQFI7F1L0o"
|
||||
)
|
||||
|
||||
// prepare lease
|
||||
if err := db.Update(func(tx *bolt.Tx) error {
|
||||
_, err0 := NewLeaseManager(tx).Create(ctx, leases.WithID(leaseID))
|
||||
return err0
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
testCases := []struct {
|
||||
lease leases.Lease
|
||||
resource leases.Resource
|
||||
err error
|
||||
}{
|
||||
{
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "sha256:29f5d56d12684887bdfa50dcd29fc31eea4aaf4ad3bec43daf19026a7ce69912",
|
||||
Type: "content",
|
||||
},
|
||||
},
|
||||
{
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "d2UdcINOwrBTQG9kS8rySAM3eMNBSojH",
|
||||
Type: "ingests",
|
||||
},
|
||||
},
|
||||
{
|
||||
// allow to add resource which exists
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "d2UdcINOwrBTQG9kS8rySAM3eMNBSojH",
|
||||
Type: "ingests",
|
||||
},
|
||||
},
|
||||
{
|
||||
// not allow to reference to lease
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "xCAV3F6PddlXitbtby0Vo23Qof6RTWpG",
|
||||
Type: "leases",
|
||||
},
|
||||
err: errdefs.ErrNotImplemented,
|
||||
},
|
||||
{
|
||||
// not allow to reference to container
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "05O9ljptPu5Qq9kZGOacEfymBwQFM8ZH",
|
||||
Type: "containers",
|
||||
},
|
||||
err: errdefs.ErrNotImplemented,
|
||||
},
|
||||
{
|
||||
// not allow to reference to image
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "qBUHpWBn03YaCt9cL3PPGKWoxBqTlLfu",
|
||||
Type: "image",
|
||||
},
|
||||
err: errdefs.ErrNotImplemented,
|
||||
},
|
||||
{
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "HMemOhlygombYhkhHhAZj5aRbDy2a3z2",
|
||||
Type: "snapshots",
|
||||
},
|
||||
err: errdefs.ErrInvalidArgument,
|
||||
},
|
||||
{
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: snapshotterKey,
|
||||
Type: "snapshots/overlayfs",
|
||||
},
|
||||
},
|
||||
{
|
||||
lease: lease,
|
||||
resource: leases.Resource{
|
||||
ID: "HMemOhlygombYhkhHhAZj5aRbDy2a3z2",
|
||||
Type: "snapshots/overlayfs/type1",
|
||||
},
|
||||
err: errdefs.ErrInvalidArgument,
|
||||
},
|
||||
{
|
||||
lease: leases.Lease{
|
||||
ID: "non-found",
|
||||
},
|
||||
resource: leases.Resource{
|
||||
ID: "HMemOhlygombYhkhHhAZj5aRbDy2a3z2",
|
||||
Type: "snapshots/overlayfs",
|
||||
},
|
||||
err: errdefs.ErrNotFound,
|
||||
},
|
||||
}
|
||||
|
||||
idxList := make(map[leases.Resource]bool)
|
||||
for i, tc := range testCases {
|
||||
if err := db.Update(func(tx *bolt.Tx) error {
|
||||
err0 := NewLeaseManager(tx).AddResource(ctx, tc.lease, tc.resource)
|
||||
if got := errors.Cause(err0); got != tc.err {
|
||||
return errors.Errorf("expect error (%v), but got (%v)", tc.err, err0)
|
||||
}
|
||||
|
||||
if err0 == nil {
|
||||
// not visited yet
|
||||
idxList[tc.resource] = false
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
t.Fatalf("failed to run case %d with resource: %v", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
// check list function
|
||||
var gotList []leases.Resource
|
||||
if err := db.View(func(tx *bolt.Tx) error {
|
||||
var err error
|
||||
gotList, err = NewLeaseManager(tx).ListResources(ctx, lease)
|
||||
return err
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if len(gotList) != len(idxList) {
|
||||
t.Fatalf("expected (%d) resources, but got (%d)", len(idxList), len(gotList))
|
||||
}
|
||||
|
||||
for _, r := range gotList {
|
||||
visited, ok := idxList[r]
|
||||
if !ok {
|
||||
t.Fatalf("unexpected resource(%v)", r)
|
||||
}
|
||||
if visited {
|
||||
t.Fatalf("duplicate resource(%v)", r)
|
||||
}
|
||||
idxList[r] = true
|
||||
}
|
||||
|
||||
// remove snapshots
|
||||
if err := db.Update(func(tx *bolt.Tx) error {
|
||||
return NewLeaseManager(tx).DeleteResource(ctx, lease, leases.Resource{
|
||||
ID: snapshotterKey,
|
||||
Type: "snapshots/overlayfs",
|
||||
})
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// check list number
|
||||
if err := db.View(func(tx *bolt.Tx) error {
|
||||
var err error
|
||||
gotList, err = NewLeaseManager(tx).ListResources(ctx, lease)
|
||||
return err
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if len(gotList)+1 != len(idxList) {
|
||||
t.Fatalf("expected (%d) resources, but got (%d)", len(idxList)-1, len(gotList))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user