Merge lp:~wallyworld/juju-core/environ-resource-catalogs into lp:~go-bot/juju-core/trunk

[Ian Booth]

New environ ResourceCatalog

A ResourceCatalog is added, which knows how to persist
resource records. A resource record contains a path and
a hash and is used to catalog the data blobs stored in
the recently added ResourceStorage. A category attribute
is also provided and will be used later for defining
resources are tools vs charms etc.

When a resource entry with the same path is saved, a new
record is not created but instead the existing one has
its ref count incremented. The opposite occurs with
removal. Removing the last record deletes it.

The above will be used by a new ManagedResource implemention
which will know how to store data for environs and users and
do things like de-duping using checksums etc.

https://codereview.appspot.com/104720046/

[Ian Booth (wallyworld)]

Reviewers: mp+221484_code.launchpad.net,

Message:
Please take a look.

Description:
New environ ResourceCatalog

A ResourceCatalog is added, which knows how to persist
resource records. A resource record contains a path and
a hash and is used to catalog the data blobs stored in
the recently added ResourceStorage. A category attribute
is also provided and will be used later for defining
resources are tools vs charms etc.

When a resource entry with the same path is saved, a new
record is not created but instead the existing one has
its ref count incremented. The opposite occurs with
removal. Removing the last record deletes it.

The above will be used by a new ManagedResource implemention
which will know how to store data for environs and users and
do things like de-duping using checksums etc.

https://code.launchpad.net/~wallyworld/juju-core/environ-resource-catalogs/+merge/221484

(do not edit description out of merge proposal)

Please review this at https://codereview.appspot.com/104720046/

Affected files (+353, -0 lines):
   A [revision details]
   M state/interface.go
   M state/storage/interface.go
   A state/storage/resourcecatalog.go
   A state/storage/resourcecatalog_test.go

[William Reade (fwereade)]

Let's talk about this when you get back. I think it would be more
profitable to start by just straight-up implementing the Storage
interface, backed by gridfs, and leaving all considerations of
deduplication out.

https://codereview.appspot.com/104720046/diff/1/state/interface.go
File state/interface.go (right):

https://codereview.appspot.com/104720046/diff/1/state/interface.go#newcode197
state/interface.go:197: // MongoTransactionRunner instances execute
mongo transactions.
/me freaks out, and it's not pretty. I really don't think we should let
this implementation detail leak out of state.

https://codereview.appspot.com/104720046/diff/1/state/storage/interface.go
File state/storage/interface.go (right):

https://codereview.appspot.com/104720046/diff/1/state/storage/interface.go#newcode23
state/storage/interface.go:23: // Resources with the same Path value are
not duplicated; instead a reference count is incremented.
I think this is not quite at the correct level of abstraction. Path is a
distraction: it's the *content* that's important, and if the first cut
puts the emphasis on path I think it'll only make it harder to do the
deduplication right when we get to that phase.

https://codereview.appspot.com/104720046/diff/1/state/storage/resourcecatalog.go
File state/storage/resourcecatalog.go (right):

https://codereview.appspot.com/104720046/diff/1/state/storage/resourcecatalog.go#newcode52
state/storage/resourcecatalog.go:52: func NewResourceCatalog(collection
*mgo.Collection, txnRunner state.MongoTransactionRunner) ResourceCatalog
{
OK, I can see why you're doing this, but I still don't really like it.
It *could* be fixed by layering -- ie putting the runTransaction
functionality into "state/txn", and having both State and
ResourceCatalog use it -- but exposing it to other random clients of
state freaks me right out.

https://codereview.appspot.com/104720046/diff/1/state/storage/resourcecatalog.go#newcode93
state/storage/resourcecatalog.go:93: if ops, err :=
rc.resourceDecRefOps(id); err == mgo.ErrNotFound {
Hmm. Is this definitely a "you already did this", or is it perhaps
evidence that everything's totally screwed?

https://codereview.appspot.com/104720046/

Unmerged revisions

2811. By Ian Booth on 2014-05-30

New ResourceCatalog implementation

=== modified file 'state/interface.go'
--- state/interface.go	2014-03-26 00:24:30 +0000
+++ state/interface.go	2014-05-30 06:23:54 +0000
@@ -4,6 +4,8 @@
 package state
 
 import (
+	"labix.org/v2/mgo/txn"
+
 	"launchpad.net/juju-core/environs/config"
 	"launchpad.net/juju-core/instance"
 	"launchpad.net/juju-core/state/api/params"
@@ -191,3 +193,8 @@
 }
 
 var _ InstanceIdGetter = (*Machine)(nil)
+
+// MongoTransactionRunner instances execute mongo transactions.
+type MongoTransactionRunner interface {
+	RunTransaction(ops []txn.Op) error
+}

=== modified file 'state/storage/interface.go'
--- state/storage/interface.go	2014-05-29 05:35:37 +0000
+++ state/storage/interface.go	2014-05-30 06:23:54 +0000
@@ -18,3 +18,19 @@
 	// Remove deletes the data at the specified path.
 	Remove(path string) error
 }
+
+// ResourceCatalog instances persist Resources.
+// Resources with the same Path value are not duplicated; instead a reference count is incremented.
+// Similarly, when a Resources is removed, the reference count is decremented. When the reference
+// count reaches zero, the Resource is deleted.
+type ResourceCatalog interface {
+	// Get fetches a Resource with the given id.
+	Get(id string) (*Resource, error)
+
+	// Put stores a Resource and returns its id.
+	Put(r *Resource) (id string, err error)
+
+	// Remove decrements the reference count for a Resource with the given id, deleting it
+	// if the reference count reaches zero.
+	Remove(id string) error
+}

=== added file 'state/storage/resourcecatalog.go'
--- state/storage/resourcecatalog.go	1970-01-01 00:00:00 +0000
+++ state/storage/resourcecatalog.go	2014-05-30 06:23:54 +0000
@@ -0,0 +1,158 @@
+// Copyright 2014 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package storage
+
+import (
+	"github.com/juju/errors"
+	"labix.org/v2/mgo"
+	"labix.org/v2/mgo/bson"
+	"labix.org/v2/mgo/txn"
+
+	"launchpad.net/juju-core/state"
+)
+
+// Resource is a catalog entry for stored data.
+type Resource struct {
+	Category string
+	Path     string
+	Hash     string
+}
+
+// resourceDoc is the persistent representation of a Resource.
+type resourceDoc struct {
+	Id       bson.ObjectId `bson:"_id"`
+	Category string
+	Path     string
+	Hash     string
+	RefCount int64
+}
+
+// resourceCatalog is a mongo backed ResourceCatalog instance.
+type resourceCatalog struct {
+	txnRunner  state.MongoTransactionRunner
+	collection *mgo.Collection
+}
+
+var _ ResourceCatalog = (*resourceCatalog)(nil)
+
+// newResourceDoc constructs a resourceDoc from a Resource.
+func newResourceDoc(r *Resource) resourceDoc {
+	return resourceDoc{
+		Id:       bson.NewObjectId(),
+		Category: r.Category,
+		Path:     r.Path,
+		Hash:     r.Hash,
+		RefCount: 1,
+	}
+}
+
+// NewResourceCatalog creates a new ResourceCatalog using the transaction runner and
+// storing resource entries in the mongo collection.
+func NewResourceCatalog(collection *mgo.Collection, txnRunner state.MongoTransactionRunner) ResourceCatalog {
+	return &resourceCatalog{
+		txnRunner:  txnRunner,
+		collection: collection,
+	}
+}
+
+// Get is defined on the ResourceCatalog interface.
+func (rc *resourceCatalog) Get(id string) (*Resource, error) {
+	var doc resourceDoc
+	objId := bson.ObjectIdHex(id)
+	if err := rc.collection.FindId(objId).One(&doc); err == mgo.ErrNotFound {
+		return nil, errors.NotFoundf("resource with id %q", id)
+	} else if err != nil {
+		return nil, err
+	}
+	return &Resource{
+		Category: doc.Category,
+		Path:     doc.Path,
+		Hash:     doc.Hash,
+	}, nil
+}
+
+// Put is defined on the ResourceCatalog interface.
+func (rc *resourceCatalog) Put(r *Resource) (string, error) {
+	if id, ops, err := rc.resourceIncRefOps(r); err != nil {
+		return "", err
+	} else {
+		for i := 0; i < 3; i++ {
+			if err := rc.txnRunner.RunTransaction(ops); err == nil {
+				return id, nil
+			} else if err != txn.ErrAborted {
+				return "", err
+			}
+		}
+	}
+	return "", state.ErrExcessiveContention
+}
+
+// Remove is defined on the ResourceCatalog interface.
+func (rc *resourceCatalog) Remove(id string) error {
+	if ops, err := rc.resourceDecRefOps(id); err == mgo.ErrNotFound {
+		return nil
+	} else if err != nil {
+		return err
+	} else {
+		for i := 0; i < 3; i++ {
+			if err := rc.txnRunner.RunTransaction(ops); err == nil {
+				return nil
+			} else if err != txn.ErrAborted {
+				return err
+			}
+		}
+	}
+	return state.ErrExcessiveContention
+}
+
+func (rc *resourceCatalog) resourceIncRefOps(r *Resource) (string, []txn.Op, error) {
+	var doc resourceDoc
+	exists := false
+	err := rc.collection.Find(bson.M{"path": r.Path}).One(&doc)
+	if err != nil && err != mgo.ErrNotFound {
+		return "", nil, err
+	} else if err == nil {
+		exists = true
+	}
+	if !exists {
+		doc := newResourceDoc(r)
+		return doc.Id.Hex(), []txn.Op{{
+			C:      rc.collection.Name,
+			Id:     doc.Id,
+			Assert: txn.DocMissing,
+			Insert: doc,
+		}}, nil
+	}
+	// If we are incrementing the ref count, then the checksums must match.
+	if r.Hash != doc.Hash {
+		return "", nil, errors.Errorf("checksums do not match for resource %q: %s != %s", r.Path, r.Hash, doc.Hash)
+	}
+	return doc.Id.Hex(), []txn.Op{{
+		C:      rc.collection.Name,
+		Id:     doc.Id,
+		Assert: txn.DocExists,
+		Update: bson.D{{"$inc", bson.D{{"refcount", 1}}}},
+	}}, nil
+}
+
+func (rc *resourceCatalog) resourceDecRefOps(id string) ([]txn.Op, error) {
+	var doc resourceDoc
+	if err := rc.collection.FindId(bson.ObjectIdHex(id)).One(&doc); err != nil {
+		return nil, err
+	}
+	if doc.RefCount == 1 {
+		return []txn.Op{{
+			C:      rc.collection.Name,
+			Id:     doc.Id,
+			Assert: bson.D{{"refcount", 1}},
+			Remove: true,
+		}}, nil
+	}
+	return []txn.Op{{
+		C:      rc.collection.Name,
+		Id:     doc.Id,
+		Assert: bson.D{{"refcount", bson.D{{"$gt", 1}}}},
+		Update: bson.D{{"$inc", bson.D{{"refcount", -1}}}},
+	}}, nil
+}

=== added file 'state/storage/resourcecatalog_test.go'
--- state/storage/resourcecatalog_test.go	1970-01-01 00:00:00 +0000
+++ state/storage/resourcecatalog_test.go	2014-05-30 06:23:54 +0000
@@ -0,0 +1,170 @@
+// Copyright 2014 Canonical Ltd.
+// Licensed under the AGPLv3, see LICENCE file for details.
+
+package storage_test
+
+import (
+	"labix.org/v2/mgo"
+	"labix.org/v2/mgo/bson"
+	"labix.org/v2/mgo/txn"
+	gc "launchpad.net/gocheck"
+
+	"launchpad.net/juju-core/state"
+	"launchpad.net/juju-core/state/storage"
+	"launchpad.net/juju-core/testing"
+)
+
+var _ = gc.Suite(&resourceCatalogSuite{})
+
+type resourceCatalogSuite struct {
+	testing.BaseSuite
+	testing.MgoSuite
+	rCatalog   storage.ResourceCatalog
+	collection *mgo.Collection
+}
+
+func (s *resourceCatalogSuite) SetUpSuite(c *gc.C) {
+	s.BaseSuite.SetUpSuite(c)
+	s.MgoSuite.SetUpSuite(c)
+}
+
+func (s *resourceCatalogSuite) TearDownSuite(c *gc.C) {
+	s.MgoSuite.TearDownSuite(c)
+	s.BaseSuite.TearDownSuite(c)
+}
+
+type simpleTxnRunner struct {
+	runner *txn.Runner
+}
+
+var _ state.MongoTransactionRunner = (*simpleTxnRunner)(nil)
+
+func (tr *simpleTxnRunner) RunTransaction(ops []txn.Op) error {
+	return tr.runner.Run(ops, "", nil)
+}
+
+func (s *resourceCatalogSuite) SetUpTest(c *gc.C) {
+	s.BaseSuite.SetUpTest(c)
+	s.MgoSuite.SetUpTest(c)
+	db := s.Session.DB("juju")
+	s.collection = db.C("resourceCatalog")
+	txnRunner := &simpleTxnRunner{txn.NewRunner(db.C("txns"))}
+	s.rCatalog = storage.NewResourceCatalog(s.collection, txnRunner)
+}
+
+func (s *resourceCatalogSuite) TearDownTest(c *gc.C) {
+	s.MgoSuite.TearDownTest(c)
+	s.BaseSuite.TearDownTest(c)
+}
+
+func (s *resourceCatalogSuite) assertPut(c *gc.C, category, path, hash string) string {
+	r := &storage.Resource{
+		Category: category,
+		Path:     path,
+		Hash:     hash,
+	}
+	id, err := s.rCatalog.Put(r)
+	c.Assert(err, gc.IsNil)
+	c.Assert(id, gc.Not(gc.Equals), "")
+	s.assertGet(c, r, id)
+	return id
+}
+
+func (s *resourceCatalogSuite) assertGet(c *gc.C, expected *storage.Resource, id string) {
+	r, err := s.rCatalog.Get(id)
+	c.Assert(err, gc.IsNil)
+	c.Assert(*r, gc.DeepEquals, *expected)
+}
+
+type resourceDoc struct {
+	Id       bson.ObjectId `bson:"_id"`
+	RefCount int64
+}
+
+func (s *resourceCatalogSuite) assertRefCount(c *gc.C, id string, expected int64) {
+	var doc resourceDoc
+	objId := bson.ObjectIdHex(id)
+	err := s.collection.FindId(objId).One(&doc)
+	c.Assert(err, gc.IsNil)
+	c.Assert(doc.RefCount, gc.Equals, expected)
+}
+
+func (s *resourceCatalogSuite) TestPut(c *gc.C) {
+	id := s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	s.assertRefCount(c, id, 1)
+}
+
+func (s *resourceCatalogSuite) TestPutSamePathIncRefCount(c *gc.C) {
+	id := s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	s.assertRefCount(c, id, 2)
+}
+
+func (s *resourceCatalogSuite) TestPutSamePathMismatchedChecksums(c *gc.C) {
+	s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	r := &storage.Resource{
+		Category: "cat",
+		Path:     "/path/to/file",
+		Hash:     "abcdef",
+	}
+	_, err := s.rCatalog.Put(r)
+	c.Assert(err, gc.ErrorMatches, `checksums do not match .*`)
+}
+
+func (s *resourceCatalogSuite) TestGetNonExistent(c *gc.C) {
+	_, err := s.rCatalog.Get(bson.NewObjectId().Hex())
+	c.Assert(err, gc.ErrorMatches, `resource with id ".*" not found`)
+}
+
+func (s *resourceCatalogSuite) TestGet(c *gc.C) {
+	r := &storage.Resource{
+		Category: "cat",
+		Path:     "/path/to/file",
+		Hash:     "12345",
+	}
+	id, err := s.rCatalog.Put(r)
+	c.Assert(err, gc.IsNil)
+	s.assertGet(c, r, id)
+}
+
+func (s *resourceCatalogSuite) TestRemoveOnlyRecord(c *gc.C) {
+	id := s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	err := s.rCatalog.Remove(id)
+	c.Assert(err, gc.IsNil)
+	_, err = s.rCatalog.Get(id)
+	c.Assert(err, gc.ErrorMatches, `resource with id ".*" not found`)
+}
+
+func (s *resourceCatalogSuite) TestRemoveDecRefCount(c *gc.C) {
+	id := s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	s.assertRefCount(c, id, 2)
+	err := s.rCatalog.Remove(id)
+	c.Assert(err, gc.IsNil)
+	s.assertRefCount(c, id, 1)
+	// The record still exists.
+	r := &storage.Resource{
+		Category: "cat",
+		Path:     "/path/to/file",
+		Hash:     "1234567890",
+	}
+	s.assertGet(c, r, id)
+}
+
+func (s *resourceCatalogSuite) TestRemoveLastCopy(c *gc.C) {
+	id := s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	s.assertPut(c, "cat", "/path/to/file", "1234567890")
+	s.assertRefCount(c, id, 2)
+	err := s.rCatalog.Remove(id)
+	c.Assert(err, gc.IsNil)
+	s.assertRefCount(c, id, 1)
+	err = s.rCatalog.Remove(id)
+	c.Assert(err, gc.IsNil)
+	_, err = s.rCatalog.Get(id)
+	c.Assert(err, gc.ErrorMatches, `resource with id ".*" not found`)
+}
+
+func (s *resourceCatalogSuite) TestRemoveNonExistent(c *gc.C) {
+	err := s.rCatalog.Remove(bson.NewObjectId().Hex())
+	c.Assert(err, gc.IsNil)
+}