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

Update metadata plugin initialization

Browse Source 
Updates metadata plugin to require content and
snapshotter plugins be loaded and initializes with
those plugins, keeping the metadata database structure
static after initialization. Service plugins now only
require metadata plugin access snapshotter or content
stores through metadata, which was already required
behavior of the services.

Signed-off-by: Derek McGowan <derek@mcgstyle.net>
This commit is contained in:
Derek McGowan 2017-10-03 15:48:05 -07:00
parent 2ab70f21ac
commit 8d892a651b
No known key found for this signature in database
GPG Key ID: F58C5D0A4405ACDB
11 changed files with 84 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
differ/differ.go
View File

@ -25,19 +25,14 @@ func init() {
Type: plugin.DiffPlugin,
ID: "walking",
Requires: []plugin.Type{
plugin.ContentPlugin,
plugin.MetadataPlugin,
},
Init: func(ic *plugin.InitContext) (interface{}, error) {
c, err := ic.Get(plugin.ContentPlugin)
if err != nil {
return nil, err
}
md, err := ic.Get(plugin.MetadataPlugin)
if err != nil {
return nil, err
}
return NewWalkingDiff(metadata.NewContentStore(md.(*metadata.DB), c.(content.Store)))
return NewWalkingDiff(md.(*metadata.DB).ContentStore())
},
})
}

9
metadata/bolt.go
View File

@ -17,9 +17,14 @@ func WithTransactionContext(ctx context.Context, tx *bolt.Tx) context.Context {
return context.WithValue(ctx, transactionKey{}, tx)
}
type transactor interface {
View(fn func(*bolt.Tx) error) error
Update(fn func(*bolt.Tx) error) error
}
// view gets a bolt db transaction either from the context
// or starts a new one with the provided bolt database.
func view(ctx context.Context, db *DB, fn func(*bolt.Tx) error) error {
func view(ctx context.Context, db transactor, fn func(*bolt.Tx) error) error {
tx, ok := ctx.Value(transactionKey{}).(*bolt.Tx)
if !ok {
return db.View(fn)
@ -29,7 +34,7 @@ func view(ctx context.Context, db *DB, fn func(*bolt.Tx) error) error {
// update gets a writable bolt db transaction either from the context
// or starts a new one with the provided bolt database.
func update(ctx context.Context, db *DB, fn func(*bolt.Tx) error) error {
func update(ctx context.Context, db transactor, fn func(*bolt.Tx) error) error {
tx, ok := ctx.Value(transactionKey{}).(*bolt.Tx)
if !ok {
return db.Update(fn)

21
metadata/content.go
View File

@ -19,23 +19,16 @@ import (
type contentStore struct {
content.Store
db *DB
db transactor
}
// NewContentStore returns a namespaced content store using an existing
// newContentStore returns a namespaced content store using an existing
// content store interface.
func NewContentStore(db *DB, cs content.Store) content.Store {
db.storeL.Lock()
defer db.storeL.Unlock()
if db.cs == nil {
db.cs = &contentStore{
Store: cs,
db: db,
}
func newContentStore(db transactor, cs content.Store) content.Store {
return &contentStore{
Store: cs,
db: db,
}
return db.cs
}
func (cs *contentStore) Info(ctx context.Context, dgst digest.Digest) (content.Info, error) {
@ -360,7 +353,7 @@ type namespacedWriter struct {
content.Writer
ref string
namespace string
db *DB
db transactor
}
func (nw *namespacedWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...content.Opt) error {

2
metadata/content_test.go
View File

@ -23,7 +23,7 @@ func createContentStore(ctx context.Context, root string) (content.Store, func()
return nil, nil, err
}
return NewContentStore(NewDB(db), cs), func() error {
return NewDB(db, cs, nil).ContentStore(), func() error {
return db.Close()
}, nil
}

30
metadata/db.go
View File

@ -1,26 +1,40 @@
package metadata
import (
"sync"
"github.com/boltdb/bolt"
"github.com/containerd/containerd/content"
"github.com/containerd/containerd/snapshot"
)
type DB struct {
db *bolt.DB
storeL sync.Mutex
ss map[string]*snapshotter
cs *contentStore
ss map[string]snapshot.Snapshotter
cs content.Store
}
func NewDB(db *bolt.DB) *DB {
func NewDB(db *bolt.DB, cs content.Store, ss map[string]snapshot.Snapshotter) *DB {
return &DB{
db: db,
ss: map[string]*snapshotter{},
ss: ss,
cs: cs,
}
}
func (m *DB) ContentStore() content.Store {
if m.cs == nil {
return nil
}
return newContentStore(m, m.cs)
}
func (m *DB) Snapshotter(name string) snapshot.Snapshotter {
sn, ok := m.ss[name]
if !ok {
return nil
}
return newSnapshotter(m, name, sn)
}
func (m *DB) View(fn func(*bolt.Tx) error) error {
return m.db.View(fn)
}

31
metadata/snapshot.go
View File

@ -19,26 +19,17 @@ import (
type snapshotter struct {
snapshot.Snapshotter
name string
db *DB
db transactor
}
// NewSnapshotter returns a new Snapshotter which namespaces the given snapshot
// using the provided name and metadata store.
func NewSnapshotter(db *DB, name string, sn snapshot.Snapshotter) snapshot.Snapshotter {
db.storeL.Lock()
defer db.storeL.Unlock()
ss, ok := db.ss[name]
if !ok {
ss = &snapshotter{
Snapshotter: sn,
name: name,
db: db,
}
db.ss[name] = ss
// newSnapshotter returns a new Snapshotter which namespaces the given snapshot
// using the provided name and database.
func newSnapshotter(db transactor, name string, sn snapshot.Snapshotter) snapshot.Snapshotter {
return &snapshotter{
Snapshotter: sn,
name: name,
db: db,
}
return ss
}
func createKey(id uint64, namespace, key string) string {
@ -466,11 +457,7 @@ func (s *snapshotter) Remove(ctx context.Context, key string) error {
}
}
if err := bkt.DeleteBucket([]byte(key)); err != nil {
return err
}
return nil
return bkt.DeleteBucket([]byte(key))
})
}

6
metadata/snapshot_test.go
View File

@ -13,7 +13,7 @@ import (
"github.com/containerd/containerd/testutil"
)
func newSnapshotter(ctx context.Context, root string) (snapshot.Snapshotter, func() error, error) {
func newTestSnapshotter(ctx context.Context, root string) (snapshot.Snapshotter, func() error, error) {
naiveRoot := filepath.Join(root, "naive")
if err := os.Mkdir(naiveRoot, 0770); err != nil {
return nil, nil, err
@ -28,7 +28,7 @@ func newSnapshotter(ctx context.Context, root string) (snapshot.Snapshotter, fun
return nil, nil, err
}
sn := NewSnapshotter(NewDB(db), "naive", snapshotter)
sn := NewDB(db, nil, map[string]snapshot.Snapshotter{"naive": snapshotter}).Snapshotter("naive")
return sn, func() error {
return db.Close()
@ -38,5 +38,5 @@ func newSnapshotter(ctx context.Context, root string) (snapshot.Snapshotter, fun
func TestMetadata(t *testing.T) {
// Snapshot tests require mounting, still requires root
testutil.RequiresRoot(t)
testsuite.SnapshotterSuite(t, "Metadata", newSnapshotter)
testsuite.SnapshotterSuite(t, "Metadata", newTestSnapshotter)
}

31
server/server.go
View File

@ -12,19 +12,21 @@ import (
"github.com/boltdb/bolt"
containers "github.com/containerd/containerd/api/services/containers/v1"
content "github.com/containerd/containerd/api/services/content/v1"
contentapi "github.com/containerd/containerd/api/services/content/v1"
diff "github.com/containerd/containerd/api/services/diff/v1"
eventsapi "github.com/containerd/containerd/api/services/events/v1"
images "github.com/containerd/containerd/api/services/images/v1"
namespaces "github.com/containerd/containerd/api/services/namespaces/v1"
snapshot "github.com/containerd/containerd/api/services/snapshot/v1"
snapshotapi "github.com/containerd/containerd/api/services/snapshot/v1"
tasks "github.com/containerd/containerd/api/services/tasks/v1"
version "github.com/containerd/containerd/api/services/version/v1"
"github.com/containerd/containerd/content"
"github.com/containerd/containerd/content/local"
"github.com/containerd/containerd/events"
"github.com/containerd/containerd/log"
"github.com/containerd/containerd/metadata"
"github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/snapshot"
metrics "github.com/docker/go-metrics"
grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"golang.org/x/net/context"
@ -176,15 +178,34 @@ func loadPlugins(config *Config) ([]*plugin.Registration, error) {
plugin.Register(&plugin.Registration{
Type: plugin.MetadataPlugin,
ID: "bolt",
Requires: []plugin.Type{
plugin.ContentPlugin,
plugin.SnapshotPlugin,
},
Init: func(ic *plugin.InitContext) (interface{}, error) {
if err := os.MkdirAll(ic.Root, 0711); err != nil {
return nil, err
}
cs, err := ic.Get(plugin.ContentPlugin)
if err != nil {
return nil, err
}
rawSnapshotters, err := ic.GetAll(plugin.SnapshotPlugin)
if err != nil {
return nil, err
}
snapshotters := make(map[string]snapshot.Snapshotter)
for name, sn := range rawSnapshotters {
snapshotters[name] = sn.(snapshot.Snapshotter)
}
db, err := bolt.Open(filepath.Join(ic.Root, "meta.db"), 0644, nil)
if err != nil {
return nil, err
}
return metadata.NewDB(db), nil
return metadata.NewDB(db, cs.(content.Store), snapshotters), nil
},
})
@ -204,7 +225,7 @@ func interceptor(
ctx = log.WithModule(ctx, "tasks")
case containers.ContainersServer:
ctx = log.WithModule(ctx, "containers")
case content.ContentServer:
case contentapi.ContentServer:
ctx = log.WithModule(ctx, "content")
case images.ImagesServer:
ctx = log.WithModule(ctx, "images")
@ -212,7 +233,7 @@ func interceptor(
// No need to change the context
case version.VersionServer:
ctx = log.WithModule(ctx, "version")
case snapshot.SnapshotsServer:
case snapshotapi.SnapshotsServer:
ctx = log.WithModule(ctx, "snapshot")
case diff.DiffServer:
ctx = log.WithModule(ctx, "diff")

9
services/content/service.go
View File

@ -39,7 +39,6 @@ func init() {
Type: plugin.GRPCPlugin,
ID: "content",
Requires: []plugin.Type{
plugin.ContentPlugin,
plugin.MetadataPlugin,
},
Init: NewService,
@ -47,19 +46,13 @@ func init() {
}
func NewService(ic *plugin.InitContext) (interface{}, error) {
c, err := ic.Get(plugin.ContentPlugin)
if err != nil {
return nil, err
}
m, err := ic.Get(plugin.MetadataPlugin)
if err != nil {
return nil, err
}
cs := metadata.NewContentStore(m.(*metadata.DB), c.(content.Store))
return &Service{
store: cs,
store: m.(*metadata.DB).ContentStore(),
publisher: ic.Events,
}, nil
}

26
services/snapshot/service.go
View File

@ -14,7 +14,6 @@ import (
"github.com/containerd/containerd/plugin"
"github.com/containerd/containerd/snapshot"
protoempty "github.com/golang/protobuf/ptypes/empty"
"github.com/pkg/errors"
"golang.org/x/net/context"
"google.golang.org/grpc"
)
@ -24,7 +23,6 @@ func init() {
Type: plugin.GRPCPlugin,
ID: "snapshots",
Requires: []plugin.Type{
plugin.SnapshotPlugin,
plugin.MetadataPlugin,
},
Init: newService,
@ -34,31 +32,19 @@ func init() {
var empty = &protoempty.Empty{}
type service struct {
snapshotters map[string]snapshot.Snapshotter
publisher events.Publisher
db *metadata.DB
publisher events.Publisher
}
func newService(ic *plugin.InitContext) (interface{}, error) {
rawSnapshotters, err := ic.GetAll(plugin.SnapshotPlugin)
if err != nil {
return nil, err
}
md, err := ic.Get(plugin.MetadataPlugin)
if err != nil {
return nil, err
}
snapshotters := make(map[string]snapshot.Snapshotter)
for name, sn := range rawSnapshotters {
snapshotters[name] = metadata.NewSnapshotter(md.(*metadata.DB), name, sn.(snapshot.Snapshotter))
}
if len(snapshotters) == 0 {
return nil, errors.Errorf("failed to create snapshotter service: no snapshotters loaded")
}
return &service{
snapshotters: snapshotters,
publisher: ic.Events,
db: md.(*metadata.DB),
publisher: ic.Events,
}, nil
}
@ -67,8 +53,8 @@ func (s *service) getSnapshotter(name string) (snapshot.Snapshotter, error) {
return nil, errdefs.ToGRPCf(errdefs.ErrInvalidArgument, "snapshotter argument missing")
}
sn, ok := s.snapshotters[name]
if !ok {
sn := s.db.Snapshotter(name)
if sn == nil {
return nil, errdefs.ToGRPCf(errdefs.ErrInvalidArgument, "snapshotter not loaded: %s", name)
}
return sn, nil

7
services/tasks/service.go
View File

@ -44,7 +44,6 @@ func init() {
Requires: []plugin.Type{
plugin.RuntimePlugin,
plugin.MetadataPlugin,
plugin.ContentPlugin,
},
Init: New,
})
@ -59,11 +58,7 @@ func New(ic *plugin.InitContext) (interface{}, error) {
if err != nil {
return nil, err
}
ct, err := ic.Get(plugin.ContentPlugin)
if err != nil {
return nil, err
}
cs := metadata.NewContentStore(m.(*metadata.DB), ct.(content.Store))
cs := m.(*metadata.DB).ContentStore()
runtimes := make(map[string]runtime.Runtime)
for _, rr := range rt {
r := rr.(runtime.Runtime)