Add ACME store

Signed-off-by: Emile Vauge <emile@vauge.com>

cluster/datastore.go

@@ -1,44 +1,62 @@
 package cluster
 
 import (
+	"encoding/json"
 	"fmt"
-	log "github.com/Sirupsen/logrus"
-	"github.com/cenkalti/backoff"
 	"github.com/containous/staert"
+	"github.com/containous/traefik/log"
 	"github.com/docker/libkv/store"
+	"github.com/emilevauge/backoff"
 	"github.com/satori/go.uuid"
 	"golang.org/x/net/context"
 	"sync"
 	"time"
 )
 
-// Object is the struct to store
-type Object interface{}
-
 // Metadata stores Object plus metadata
 type Metadata struct {
-	Lock string
+	object Object
+	Object []byte
+	Lock   string
 }
 
+func (m *Metadata) marshall() error {
+	var err error
+	m.Object, err = json.Marshal(m.object)
+	return err
+}
+
+func (m *Metadata) unmarshall() error {
+	if len(m.Object) == 0 {
+		return nil
+	}
+	return json.Unmarshal(m.Object, m.object)
+}
+
+// Listener is called when Object has been changed in KV store
+type Listener func(Object) error
+
+var _ Store = (*Datastore)(nil)
+
 // Datastore holds a struct synced in a KV store
 type Datastore struct {
-	kv        *staert.KvSource
+	kv        staert.KvSource
 	ctx       context.Context
 	localLock *sync.RWMutex
-	object    Object
 	meta      *Metadata
 	lockKey   string
+	listener  Listener
 }
 
 // NewDataStore creates a Datastore
-func NewDataStore(kvSource *staert.KvSource, ctx context.Context, object Object) (*Datastore, error) {
+func NewDataStore(kvSource staert.KvSource, ctx context.Context, object Object, listener Listener) (*Datastore, error) {
 	datastore := Datastore{
 		kv:        kvSource,
 		ctx:       ctx,
-		meta:      &Metadata{},
-		object:    object,
+		meta:      &Metadata{object: object},
 		lockKey:   kvSource.Prefix + "/lock",
 		localLock: &sync.RWMutex{},
+		listener:  listener,
 	}
 	err := datastore.watchChanges()
 	if err != nil {
@@ -67,17 +85,24 @@ func (d *Datastore) watchChanges() error {
 						return err
 					}
 					d.localLock.Lock()
-					err := d.kv.LoadConfig(d.object)
+					err := d.kv.LoadConfig(d.meta)
 					if err != nil {
 						d.localLock.Unlock()
 						return err
 					}
-					err = d.kv.LoadConfig(d.meta)
+					err = d.meta.unmarshall()
 					if err != nil {
 						d.localLock.Unlock()
 						return err
 					}
 					d.localLock.Unlock()
+					// log.Debugf("Datastore object change received: %+v", d.object)
+					if d.listener != nil {
+						err := d.listener(d.meta.object)
+						if err != nil {
+							log.Errorf("Error calling datastore listener: %s", err)
+						}
+					}
 				}
 			}
 		}
@@ -93,11 +118,12 @@ func (d *Datastore) watchChanges() error {
 }
 
 // Begin creates a transaction with the KV store.
-func (d *Datastore) Begin() (*Transaction, error) {
+func (d *Datastore) Begin() (Transaction, Object, error) {
 	id := uuid.NewV4().String()
+	log.Debugf("Transaction %s begins", id)
 	remoteLock, err := d.kv.NewLock(d.lockKey, &store.LockOptions{TTL: 20 * time.Second, Value: []byte(id)})
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	stopCh := make(chan struct{})
 	ctx, cancel := context.WithCancel(d.ctx)
@@ -109,11 +135,11 @@ func (d *Datastore) Begin() (*Transaction, error) {
 	select {
 	case <-ctx.Done():
 		if errLock != nil {
-			return nil, errLock
+			return nil, nil, errLock
 		}
 	case <-d.ctx.Done():
 		stopCh <- struct{}{}
-		return nil, d.ctx.Err()
+		return nil, nil, d.ctx.Err()
 	}
 
 	// we got the lock! Now make sure we are synced with KV store
@@ -131,14 +157,15 @@ func (d *Datastore) Begin() (*Transaction, error) {
 	ebo.MaxElapsedTime = 60 * time.Second
 	err = backoff.RetryNotify(operation, ebo, notify)
 	if err != nil {
-		return nil, fmt.Errorf("Datastore cannot sync: %v", err)
+		return nil, nil, fmt.Errorf("Datastore cannot sync: %v", err)
 	}
 
 	// we synced with KV store, we can now return Setter
-	return &Transaction{
+	return &datastoreTransaction{
 		Datastore:  d,
 		remoteLock: remoteLock,
-	}, nil
+		id:         id,
+	}, d.meta.object, nil
 }
 
 func (d *Datastore) get() *Metadata {
@@ -147,28 +174,50 @@ func (d *Datastore) get() *Metadata {
 	return d.meta
 }
 
+// Load load atomically a struct from the KV store
+func (d *Datastore) Load() (Object, error) {
+	d.localLock.Lock()
+	defer d.localLock.Unlock()
+	err := d.kv.LoadConfig(d.meta)
+	if err != nil {
+		return nil, err
+	}
+	err = d.meta.unmarshall()
+	if err != nil {
+		return nil, err
+	}
+	return d.meta.object, nil
+}
+
 // Get atomically a struct from the KV store
 func (d *Datastore) Get() Object {
 	d.localLock.RLock()
 	defer d.localLock.RUnlock()
-	return d.object
+	return d.meta.object
 }
 
-// Transaction allows to set a struct in the KV store
-type Transaction struct {
+var _ Transaction = (*datastoreTransaction)(nil)
+
+type datastoreTransaction struct {
 	*Datastore
 	remoteLock store.Locker
 	dirty      bool
+	id         string
 }
 
 // Commit allows to set an object in the KV store
-func (s *Transaction) Commit(object Object) error {
+func (s *datastoreTransaction) Commit(object Object) error {
 	s.localLock.Lock()
 	defer s.localLock.Unlock()
 	if s.dirty {
 		return fmt.Errorf("Transaction already used. Please begin a new one.")
 	}
-	err := s.kv.StoreConfig(object)
+	s.Datastore.meta.object = object
+	err := s.Datastore.meta.marshall()
+	if err != nil {
+		return err
+	}
+	err = s.kv.StoreConfig(s.Datastore.meta)
 	if err != nil {
 		return err
 	}
@@ -178,7 +227,8 @@ func (s *Transaction) Commit(object Object) error {
 		return err
 	}
 
-	s.Datastore.object = object
 	s.dirty = true
+	// log.Debugf("Datastore object saved: %+v", s.object)
+	log.Debugf("Transaction commited %s", s.id)
 	return nil
 }