upgrade to etcd/client/v3

go.mod

@@ -3,23 +3,16 @@ module github.com/flatcar-linux/locksmith
 go 1.14
 
 require (
-	github.com/coreos/go-semver v0.3.0 // indirect
 	github.com/coreos/go-systemd v0.0.0-20191104093116-d3cd4ed1dbcf
 	github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f
 	github.com/godbus/dbus v4.1.0+incompatible // indirect
-	github.com/godbus/dbus/v5 v5.0.3
-	github.com/gogo/protobuf v1.3.1 // indirect
+	github.com/godbus/dbus/v5 v5.0.4
 	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/rkt/rkt v1.30.0
-	go.etcd.io/etcd v0.0.0-00010101000000-000000000000
-	go.uber.org/zap v1.16.0 // indirect
-	golang.org/x/net v0.0.0-20201031054903-ff519b6c9102
-	google.golang.org/grpc v1.33.2 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.0
+	go.etcd.io/etcd/client/v3 v3.5.0
+	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
 )
 
-replace (
-	// Force updating etcd to most recent version.
-	go.etcd.io/etcd => go.etcd.io/etcd v0.5.0-alpha.5.0.20200824191128-ae9734ed278b
-	// Most recent etcd version is not compatible with grpc v1.31.x.
-	google.golang.org/grpc => google.golang.org/grpc v1.29.1
-)
+// Most recent etcd version is not compatible with grpc v1.31.x.
+replace google.golang.org/grpc => google.golang.org/grpc v1.29.1

lock/etcd.go

@@ -17,14 +17,19 @@ package lock
 import (
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/url"
 	"path"
 
-	"go.etcd.io/etcd/client"
+	client "go.etcd.io/etcd/client/v3"
 
 	"golang.org/x/net/context"
 )
 
+// ErrNotFound is used when a key is not found - which means
+// it returns 0 value.
+var ErrNotFound = errors.New("key not found")
+
 const (
 	keyPrefix       = "coreos.com/updateengine/rebootlock"
 	groupBranch     = "groups"
@@ -33,12 +38,11 @@ const (
 	SemaphorePrefix = keyPrefix + "/" + semaphoreBranch
 )
 
-// KeysAPI is the minimum etcd client.KeysAPI interface EtcdLockClient needs
+// KeysAPI is the minimum etcd client.KV interface EtcdLockClient needs
 // to do its job.
 type KeysAPI interface {
-	Get(ctx context.Context, key string, opts *client.GetOptions) (*client.Response, error)
-	Set(ctx context.Context, key, value string, opts *client.SetOptions) (*client.Response, error)
-	Create(ctx context.Context, key, value string) (*client.Response, error)
+	Get(ctx context.Context, key string, opts ...client.OpOption) (*client.GetResponse, error)
+	Txn(ctx context.Context) client.Txn
 }
 
 // EtcdLockClient is a wrapper around the etcd client that provides
@@ -60,46 +64,59 @@ func NewEtcdLockClient(keyapi KeysAPI, group string) (*EtcdLockClient, error) {
 
 	elc := &EtcdLockClient{keyapi, key}
 	if err := elc.Init(); err != nil {
-		return nil, err
+		return nil, fmt.Errorf("unable to init etcd lock client: %w", err)
 	}
 
 	return elc, nil
 }
 
 // Init sets an initial copy of the semaphore if it doesn't exist yet.
+// So we first try to get the value, if the value is not found we create the key
+// with a default semaphore value.
 func (c *EtcdLockClient) Init() error {
 	sem := newSemaphore()
-	b, err := json.Marshal(sem)
+	payload, err := json.Marshal(sem)
 	if err != nil {
-		return err
+		return fmt.Errorf("unable to marshal initial semaphore: %w", err)
 	}
 
-	if _, err := c.keyapi.Create(context.Background(), c.keypath, string(b)); err != nil {
-		eerr, ok := err.(client.Error)
-		if ok && eerr.Code == client.ErrorCodeNodeExist {
-			return nil
-		}
-
-		return err
+	if _, err := c.keyapi.Txn(context.TODO()).
+		If(
+			client.Compare(client.Version(c.keypath), "=", 0),
+		).
+		Then(
+			client.OpPut(c.keypath, string(payload)),
+		).
+		Commit(); err != nil {
+		return fmt.Errorf("unable to commit initial transaction: %w", err)
 	}
 
 	return nil
 }
 
 // Get fetches the Semaphore from etcd.
 func (c *EtcdLockClient) Get() (*Semaphore, error) {
-	resp, err := c.keyapi.Get(context.Background(), c.keypath, nil)
+	resp, err := c.keyapi.Get(context.Background(), c.keypath, client.WithLastCreate()...)
 	if err != nil {
 		return nil, err
 	}
 
+	// There is no proper way to handle non-existing value for a
+	// given key.
+	// See https://github.com/etcd-io/etcd/issues/6089 for more details.
+	if resp.Count == 0 {
+		return nil, ErrNotFound
+	}
+
+	kv := resp.Kvs[0]
+
 	sem := &Semaphore{}
-	err = json.Unmarshal([]byte(resp.Node.Value), sem)
+	err = json.Unmarshal(kv.Value, sem)
 	if err != nil {
 		return nil, err
 	}
 
-	sem.Index = resp.Node.ModifiedIndex
+	sem.Index = uint64(kv.Version)
 
 	return sem, nil
 }
@@ -114,10 +131,21 @@ func (c *EtcdLockClient) Set(sem *Semaphore) error {
 		return err
 	}
 
-	setopts := &client.SetOptions{
-		PrevIndex: sem.Index,
+	response, err := c.keyapi.Txn(context.Background()).
+		If(
+			client.Compare(client.Version(c.keypath), "=", int64(sem.Index)),
+		).
+		Then(
+			client.OpPut(c.keypath, string(b)),
+		).
+		Commit()
+	if err != nil {
+		return fmt.Errorf("making transaction: %w", err)
+	}
+
+	if !response.Succeeded {
+		return fmt.Errorf("failed to set the semaphore - it got updated in the meantime")
 	}
 
-	_, err = c.keyapi.Set(context.Background(), c.keypath, string(b), setopts)
-	return err
+	return nil
 }