rethink server implementation

Signed-off-by: David Lawrence <david.lawrence@docker.com> (github: endophage)
This commit is contained in:
David Lawrence 2016-03-21 13:55:45 -07:00 committed by Riyaz Faizullabhoy
parent 72135193b0
commit 045721250f
4 changed files with 206 additions and 9 deletions

View File

@ -14,6 +14,7 @@ import (
"github.com/docker/notary/server"
"github.com/docker/notary/server/storage"
"github.com/docker/notary/signer/client"
"github.com/docker/notary/storage/rethinkdb"
"github.com/docker/notary/tuf/data"
"github.com/docker/notary/tuf/signed"
"github.com/docker/notary/utils"
@ -63,18 +64,27 @@ func grpcTLS(configuration *viper.Viper) (*tls.Config, error) {
// parses the configuration and returns a backing store for the TUF files
func getStore(configuration *viper.Viper, allowedBackends []string, hRegister healthRegister) (
storage.MetaStore, error) {
var (
store storage.MetaStore
err error
)
storeConfig, err := utils.ParseStorage(configuration, allowedBackends)
if err != nil {
return nil, err
}
logrus.Infof("Using %s backend", storeConfig.Backend)
if storeConfig.Backend == utils.MemoryBackend {
switch storeConfig.Backend {
case notary.MemoryBackend:
return storage.NewMemStorage(), nil
case notary.MySQLBackend:
store, err = storage.NewSQLStorage(storeConfig.Backend, storeConfig.Source)
case notary.RethinkDBBackend:
backend := rethinkdb.Connection(storeConfig.CA, storeConfig.Source, storeConfig.Password)
store, err = storage.NewRethinkDBStorage(backend)
default:
err = fmt.Errorf("%s not a supported storage backend", storeConfig.Backend)
}
store, err := storage.NewSQLStorage(storeConfig.Backend, storeConfig.Source)
if err != nil {
return nil, fmt.Errorf("Error starting DB driver: %s", err.Error())
}

182
server/storage/rethinkdb.go Normal file
View File

@ -0,0 +1,182 @@
package storage
import (
"time"
"github.com/dancannon/gorethink"
"github.com/docker/notary/storage/rethinkdb"
)
type RDBTUFFile struct {
rethinkdb.Timing
Gun string `gorethink:"gun"`
Role string `gorethink:"role"`
Version int `gorethink:"version"`
Sha256 string `gorethink:"sha256"`
Data []byte `gorethink:"data"`
}
func (_ RDBTufFile) TableName() string {
return "tuf_files"
}
func (_ RDBTufFile) DatabaseName() string {
return "notaryserver"
}
type RDBKey struct {
rethinkdb.Timing
Gun string `gorethink:"gun"`
Role string `gorethink:"role"`
Cipher string `gorethink:"cipher"`
Public []byte `gorethink:"public"`
}
func (_ RDBKey) TableName() string {
return "tuf_keys"
}
func (_ RDBKey) DatabaseName() string {
return "notaryserver"
}
// RethinkDB implements a MetaStore against the Rethink Database
type RethinkDB struct {
dbName string
rdb *gorethink.Session
}
// NewRethinkDBStorage initializes a RethinkDB object
func NewRethinkDBStorage(dbName string, sess *gorethink.Session) MetaStore {
return RethinkDB{
dbName: dbName,
rdb: sess,
}
}
// GetKey returns the cipher and public key for the given GUN and role.
// If the GUN+role don't exist, returns an error.
func (rdb RethinkDB) GetKey(gun, role string) (cipher string, public []byte, err error) {
var key RDBKey
res, err := gorethink.DB(rdb.dbName).Table(key.TableName()).Get(
RDBKey{
Gun: gun,
Role: role,
}).Run(rdb.sess)
if err != nil {
return "", nil, err
}
defer res.Close()
err = res.One(&key)
return key.Cipher, key.Public, err
}
// SetKey sets the cipher and public key for the given GUN and role if
// it doesn't already exist. Otherwise an error is returned.
func (rdb RethinkDB) SetKey(gun, role, cipher string, public []byte) error {
now := time.Now()
key := RDBKey{
Timing: rethinkdb.Timing{
CreatedAt: now,
UpdatedAt: now,
},
Gun: gun,
Role: role,
Cipher: cipher,
Public: public,
}
_, err := gorethink.DB(rdb.dbName).Table(key.TableName()).Insert(key).RunWrite(rdb.sess)
return err
}
// UpdateCurrent adds new metadata version for the given GUN if and only
// if it's a new role, or the version is greater than the current version
// for the role. Otherwise an error is returned.
func (rdb RethinkDB) UpdateCurrent(gun string, update MetaUpdate) error {
now := time.Now()
checksum := sha256.Sum256(update.Data)
file := RDBTUFFile{
Timing: rethinkdb.Timing{
CreatedAt: now,
UpdatedAt: now,
},
Gun: gun,
Role: update.Role,
Version: update.Version,
Sha256: hex.EncodeToString(checksum[:]),
Data: update.Data,
}
_, err := gorethink.DB(rdb.dbName).Table(file.TableName()).Insert(
file,
gorethink.InsertOpts{
Conflict: "error", // default but explicit for clarity of intent
},
).RunWrite(rdb.sess)
if gorethink.IsConflictErr(err) {
return &ErrOldVersion{}
}
return err
}
// UpdateMany adds multiple new metadata for the given GUN. RethinkDB does
// not support transactions, therefore we will attempt to insert the timestamp
// first as this represents a published version of the repo. If this is successful,
// we will insert the remaining roles (in any order). If any of those roles
// errors on insert, we will do a best effort rollback, at a minimum attempting
// to delete the timestamp so nobody pulls a broken repo.
func (rdb RethinkDB) UpdateMany(gun string, updates []MetaUpdate) error {
for _, up := range updates {
if err := rdb.UpdateCurrent(gun, up); err != nil {
return err
}
}
return nil
}
// GetCurrent returns the modification date and data part of the metadata for
// the latest version of the given GUN and role. If there is no data for
// the given GUN and role, an error is returned.
func (rdb RethinkDB) GetCurrent(gun, role string) (created *time.Time, data []byte, err error) {
var file RDBTUFFile
res, err := gorethink.DB(rdb.dbName).Table(file.TableName()).Get(
RDBTUFFile{
Gun: gun,
Role: role,
}).Run(rdb.sess)
if err != nil {
return nil, nil, err
}
defer res.Close()
err = res.One(&key)
return &file.CreatedAt, file.Data, err
}
// GetChecksum returns the given TUF role file and creation date for the
// GUN with the provided checksum. If the given (gun, role, checksum) are
// not found, it returns storage.ErrNotFound
func (rdb RethinkDB) GetChecksum(gun, tufRole, checksum string) (created *time.Time, data []byte, err error) {
var file RDBTUFFile
res, err := gorethink.DB(rdb.dbName).Table(file.TableName()).Get(
RDBTUFFile{
Gun: gun,
Role: role,
Sha256: checksum,
}).Run(rdb.sess)
if err != nil {
return nil, nil, err
}
defer res.Close()
err = res.One(&key)
return &file.CreatedAt, file.Data, err
}
// Delete removes all metadata for a given GUN. It does not return an
// error if no metadata exists for the given GUN.
func (rdb RethinkDB) Delete(gun string) error {
files := RDBTUFFile{Gun: gun}
_, err := gorethink.DB(rdb.dbName).Table(files.TableName()).Get(files).Delete().RunWrite(rdb.sess)
if err != nil {
return fmt.Errorf("unable to delete %s from database: %s", gun, err.Error())
}
return nil
}

View File

@ -57,7 +57,7 @@ func NewKeyRethinkDBStore(passphraseRetriever passphrase.Retriever, defaultPassA
// Name returns a user friendly name for the storage location
func (s *KeyRethinkDBStore) Name() string {
return "database"
return "RethinkDB"
}
// AddKey stores the contents of a private key. Both role and gun are ignored,
@ -74,7 +74,12 @@ func (s *KeyRethinkDBStore) AddKey(keyInfo trustmanager.KeyInfo, privKey data.Pr
return err
}
now := time.Now()
rethinkPrivKey := RethinkPrivateKey{
rethinkdb.Timing{
CreatedAt: now,
UpdatedAt: now,
},
KeyID: privKey.ID(),
EncryptionAlg: EncryptionAlg,
KeywrapAlg: KeywrapAlg,
@ -165,7 +170,7 @@ func (s *KeyRethinkDBStore) RemoveKey(keyID string) error {
dbPrivateKey := RethinkPrivateKey{KeyID: keyID}
_, err := gorethink.DB(dbPrivateKey.DatabaseName()).Table(dbPrivateKey.TableName()).Get(dbPrivateKey).Delete().RunWrite(s.session)
if err != nil {
return fmt.Errorf("Unable to delete private key from database")
return fmt.Errorf("unable to delete private key from database: %s", err.Error())
}
return nil

View File

@ -12,9 +12,9 @@ var session *gorethink.Session
// Timing can be embedded into other gorethink models to
// add time tracking fields
type Timing struct {
CreatedAt *time.Time `gorethink:"created_at"`
UpdatedAt *time.Time `gorethink:"updated_at"`
DeletedAt *time.Time `gorethink:"deleted_at"`
CreatedAt time.Time `gorethink:"created_at"`
UpdatedAt time.Time `gorethink:"updated_at"`
DeletedAt time.Time `gorethink:"deleted_at"`
}
// Connection sets up a RethinkDB connection to the host (`host:port` format)