Merge pull request #206 from pdxjohnny/etcd_lib_fixes

kontrol: etcd: Debug logging and TTL fix

Author: rjeczalik

kontrol/etcd.go

@@ -48,10 +48,31 @@ func NewEtcd(machines []string, log kite.Logger) *Etcd {
 		panic("cannot connect to etcd cluster: " + strings.Join(machines, ","))
 	}
 
-	return &Etcd{
-		client: etcd.NewKeysAPI(client),
+	e := &Etcd{
+		client: NewKeysAPILogger(etcd.NewKeysAPI(client), log),
 		log:    log,
 	}
+
+	// Create our prefix as a directory if it does not exist
+	_, err = e.client.Get(context.Background(), KitesPrefix, nil)
+	if err == nil {
+		return e
+	}
+
+	_, err = e.client.Set(
+		context.Background(),
+		KitesPrefix,
+		"",
+		&etcd.SetOptions{
+			Dir:       true,
+			PrevExist: etcd.PrevIgnore,
+		},
+	)
+	if err != nil {
+		log.Fatal("Could not create KitesPrefix %q: %v", KitesPrefix, err)
+	}
+
+	return e
 }
 
 func (e *Etcd) Delete(k *protocol.Kite) error {
@@ -95,7 +116,7 @@ func (e *Etcd) Add(k *protocol.Kite, value *kontrolprotocol.RegisterValue) error
 		etcdKey,
 		valueString,
 		&etcd.SetOptions{
-			TTL:       KeyTTL / time.Second,
+			TTL:       KeyTTL,
 			PrevExist: etcd.PrevExist,
 		},
 	)
@@ -108,7 +129,7 @@ func (e *Etcd) Add(k *protocol.Kite, value *kontrolprotocol.RegisterValue) error
 		etcdIDKey,
 		valueString,
 		&etcd.SetOptions{
-			TTL:       KeyTTL / time.Second,
+			TTL:       KeyTTL,
 			PrevExist: etcd.PrevExist,
 		},
 	)
@@ -136,7 +157,7 @@ func (e *Etcd) Update(k *protocol.Kite, value *kontrolprotocol.RegisterValue) er
 		etcdKey,
 		valueString,
 		&etcd.SetOptions{
-			TTL:       KeyTTL / time.Second,
+			TTL:       KeyTTL,
 			PrevExist: etcd.PrevExist,
 		},
 	)
@@ -153,7 +174,7 @@ func (e *Etcd) Update(k *protocol.Kite, value *kontrolprotocol.RegisterValue) er
 		etcdIDKey,
 		valueString,
 		&etcd.SetOptions{
-			TTL:       KeyTTL / time.Second,
+			TTL:       KeyTTL,
 			PrevExist: etcd.PrevExist,
 		},
 	)
@@ -166,7 +187,7 @@ func (e *Etcd) Update(k *protocol.Kite, value *kontrolprotocol.RegisterValue) er
 		KitesPrefix+"/"+k.Username,
 		"",
 		&etcd.SetOptions{
-			TTL:       KeyTTL / time.Second,
+			TTL:       KeyTTL,
 			PrevExist: etcd.PrevExist,
 		},
 	)

kontrol/etcd_keysapi_logger.go

@@ -0,0 +1,55 @@
+package kontrol
+
+import (
+	"context"
+
+	etcd "github.com/coreos/etcd/client"
+	"github.com/koding/kite"
+)
+
+type KeysAPILogger struct {
+	kapi etcd.KeysAPI
+	log  kite.Logger
+}
+
+func NewKeysAPILogger(kapi etcd.KeysAPI, log kite.Logger) KeysAPILogger {
+	return KeysAPILogger{
+		kapi: kapi,
+		log:  log,
+	}
+}
+
+func (k KeysAPILogger) Get(ctx context.Context, key string, opts *etcd.GetOptions) (*etcd.Response, error) {
+	k.log.Debug("Get: key: %v opts: %v", key, opts)
+	return k.kapi.Get(ctx, key, opts)
+}
+
+func (k KeysAPILogger) Set(ctx context.Context, key, value string, opts *etcd.SetOptions) (*etcd.Response, error) {
+	k.log.Debug("Set: key: %v value: %v opts: %v", key, value, opts)
+	return k.kapi.Set(ctx, key, value, opts)
+}
+
+func (k KeysAPILogger) Delete(ctx context.Context, key string, opts *etcd.DeleteOptions) (*etcd.Response, error) {
+	k.log.Debug("Delete: key: %v opts: %v", key, opts)
+	return k.kapi.Delete(ctx, key, opts)
+}
+
+func (k KeysAPILogger) Create(ctx context.Context, key, value string) (*etcd.Response, error) {
+	k.log.Debug("Create: key: %v value: %v", key, value)
+	return k.kapi.Create(ctx, key, value)
+}
+
+func (k KeysAPILogger) CreateInOrder(ctx context.Context, dir, value string, opts *etcd.CreateInOrderOptions) (*etcd.Response, error) {
+	k.log.Debug("CreateInOrder: dir: %v value: %v opts: %v", dir, value, opts)
+	return k.kapi.CreateInOrder(ctx, dir, value, opts)
+}
+
+func (k KeysAPILogger) Update(ctx context.Context, key, value string) (*etcd.Response, error) {
+	k.log.Debug("Update: key: %v value: %v", key, value)
+	return k.kapi.Update(ctx, key, value)
+}
+
+func (k KeysAPILogger) Watcher(key string, opts *etcd.WatcherOptions) etcd.Watcher {
+	k.log.Debug("Watcher: key: %v opts: %v", key, opts)
+	return k.kapi.Watcher(key, opts)
+}