Added a cancellable context to the distributed locking mechanism.

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/google/go-cmp v0.5.1 // indirect
 	github.com/google/uuid v1.2.0
 	github.com/gorilla/mux v1.8.0
+	github.com/gorilla/websocket v1.4.1 // indirect
 	github.com/hashicorp/go-hclog v0.14.1 // indirect
 	github.com/hashicorp/go-multierror v1.1.0 // indirect
 	github.com/namsral/flag v1.7.4-pre
@@ -26,4 +27,5 @@ require (
 	golang.org/x/text v0.3.3 // indirect
 	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
+	gopkg.in/yaml.v2 v2.2.8 // indirect
 )

go.sum

@@ -92,8 +92,9 @@ github.com/google/uuid v1.2.0 h1:qJYtXnJRWmpe7m/3XlyhrsLrEURqHRM2kxzoxXqyUDs=
 github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
-github.com/gorilla/websocket v1.4.0 h1:WDFjx/TMzVgy9VdMMQi2K2Emtwi2QcUQsztZ/zLaH/Q=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
+github.com/gorilla/websocket v1.4.1 h1:q7AeDBpnBk8AogcD4DSag/Ukw/KV+YhzLj2bP5HvKCM=
+github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 h1:Iju5GlWwrvL6UBg4zJJt3btmonfrMlCDdsejg4CZE7c=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
@@ -319,8 +320,9 @@ gopkg.in/resty.v1 v1.12.0/go.mod h1:mDo4pnntr5jdWRML875a/NmxYqAlA73dVijT2AXvQQo=
 gopkg.in/square/go-jose.v2 v2.3.1 h1:SK5KegNXmKmqE342YYN2qPHEnUYeoMiXXl1poUlI+o4=
 gopkg.in/square/go-jose.v2 v2.3.1/go.mod h1:M9dMgbHiYLoDGQrXy7OpJDJWiKiU//h+vD76mk0e1AI=
 gopkg.in/yaml.v2 v2.0.0-20170812160011-eb3733d160e7/go.mod h1:JAlM8MvJe8wmxCU4Bli9HhUf9+ttbYbLASfIpnQbh74=
-gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
+gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=

internal/storage/distlock_etcd_impl.go

@@ -23,6 +23,7 @@
 package storage
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -38,6 +39,8 @@ type ETCDLockProvider struct {
 	Logger *logrus.Logger
 	Duration time.Duration
 	mutex *sync.Mutex
+	ctx context.Context
+	ctxCancelFunc context.CancelFunc
 	kvHandle hmetcd.Kvi
 }
 
@@ -74,27 +77,31 @@ func (d *ETCDLockProvider) Ping() error {
 	return e.Ping()
 }
 
-func (d *ETCDLockProvider) DistributedTimedLock(maxLockTime time.Duration) error {
+func (d *ETCDLockProvider) DistributedTimedLock(maxLockTime time.Duration) (context.Context,error) {
 	if (maxLockTime < 1) {
-		return fmt.Errorf("Error: lock duration request invalid (%s seconds) -- must be >= 1.",
+		return nil,fmt.Errorf("Error: lock duration request invalid (%s seconds) -- must be >= 1.",
 					maxLockTime.String())
 	}
 	d.mutex.Lock()
 	defer d.mutex.Unlock()
+	d.ctx,d.ctxCancelFunc = context.WithTimeout(context.Background(),maxLockTime)
 	err := d.kvHandle.DistTimedLock(int(maxLockTime.Seconds()))
 	if (err != nil) {
-		return err
+		d.ctxCancelFunc()
+		return nil,err
 	}
 	d.Duration = maxLockTime
 
-	return nil
+	return d.ctx,nil
 }
 
 func (d *ETCDLockProvider) Unlock() error {
 	d.mutex.Lock()
 	defer d.mutex.Unlock()
 	d.Duration = 0
-	return d.kvHandle.DistUnlock()
+	d.kvHandle.DistUnlock()
+	d.ctxCancelFunc()
+	return nil
 }
 
 func (d *ETCDLockProvider) GetDuration() time.Duration {

internal/storage/distlock_mem_impl.go

@@ -23,6 +23,7 @@
 package storage
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"time"
@@ -41,20 +42,20 @@ type MEMLockProvider struct {
 	Logger *logrus.Logger
 	Duration time.Duration
 	mutex *sync.Mutex
+	ctx context.Context
+	ctxCancelFunc context.CancelFunc
 	kvHandle hmetcd.Kvi
 }
 
 func toStorageMEM(m *MEMLockProvider) *MEMStorage {
 	return &MEMStorage{Logger: m.Logger, mutex: m.mutex, kvHandle: m.kvHandle}
 }
 
-func fromStorageMEM(m *MEMStorage) *MEMLockProvider {
-	return &MEMLockProvider{Logger: m.Logger, mutex: m.mutex, kvHandle: m.kvHandle}
-}
-
 func toDistLockETCD(m *MEMLockProvider) *ETCDLockProvider {
 	return &ETCDLockProvider{Logger: m.Logger, Duration: m.Duration,
-	                         mutex: m.mutex, kvHandle: m.kvHandle}
+	                         mutex: m.mutex, ctx: m.ctx,
+	                         ctxCancelFunc: m.ctxCancelFunc,
+	                         kvHandle: m.kvHandle}
 }
 
 
@@ -83,14 +84,17 @@ func (d *MEMLockProvider) Ping() error {
 	return e.Ping()
 }
 
-func (d *MEMLockProvider) DistributedTimedLock(maxLockTime time.Duration) error {
+func (d *MEMLockProvider) DistributedTimedLock(maxLockTime time.Duration) (context.Context,error) {
 	if (maxLockTime < 1) {
-		return fmt.Errorf("Error: lock duration request invalid (%s seconds) -- must be >= 1.",
+		return nil,fmt.Errorf("Error: lock duration request invalid (%s seconds) -- must be >= 1.",
 					maxLockTime.String())
 	}
 	d.Duration = maxLockTime
 	e := toDistLockETCD(d)
-	return e.DistributedTimedLock(maxLockTime)
+	c,ce := e.DistributedTimedLock(maxLockTime)
+	d.ctx = e.ctx
+	d.ctxCancelFunc = e.ctxCancelFunc
+	return c,ce
 }
 
 func (d *MEMLockProvider) Unlock() error {

internal/storage/storage_interface.go

@@ -23,6 +23,7 @@
 package storage
 
 import (
+	"context"
 	"github.com/Cray-HPE/hms-power-control/internal/model"
 	"github.com/sirupsen/logrus"
 	"time"
@@ -43,6 +44,6 @@ type DistributedLockProvider interface {
 	InitFromStorage(si interface{}, Logger *logrus.Logger)
 	Ping() error
 	GetDuration() time.Duration
-	DistributedTimedLock(maxLockTime time.Duration) error
+	DistributedTimedLock(maxLockTime time.Duration) (context.Context,error)
 	Unlock() error
 }

internal/storage/storage_test.go

@@ -183,9 +183,9 @@ func TestPowerStatusStorage(t *testing.T) {
 	//for locking, so all we can exercise is the function calls.
 
 	lockDur := 10 * time.Second
-	err = ds.DistributedTimedLock(lockDur)
-	if err != nil {
-		t.Errorf("DistributedTimedLock() failed: %v", err)
+	ctx,dlerr := ds.DistributedTimedLock(lockDur)
+	if dlerr != nil {
+		t.Errorf("DistributedTimedLock() failed: %v", dlerr)
 	}
 	time.Sleep(1 * time.Second)
 	if ds.GetDuration() != lockDur {
@@ -200,4 +200,46 @@ func TestPowerStatusStorage(t *testing.T) {
 		t.Errorf("Lock duration readout failed, expecting 0s, got %s",
 			ds.GetDuration().String())
 	}
+
+	ctx,dlerr = ds.DistributedTimedLock(lockDur)
+	if dlerr != nil {
+		t.Errorf("DistributedTimedLock(2) failed: %v", dlerr)
+	}
+	gotCancelled := false
+	Outer:
+	for ix := 0; ix < 15; ix ++ {
+		select {
+			case <-ctx.Done():
+				t.Logf("Context cancelled!")
+				gotCancelled = true
+				break Outer
+			default:
+				time.Sleep(time.Second)
+		}
+	}
+
+	if (!gotCancelled) {
+		t.Errorf("Never saw context cancel.")
+	}
+	err = ds.Unlock()
+	if err != nil {
+		t.Errorf("Error releasing timed lock (outer): %v", err)
+	}
+
+
+	ctx,dlerr = ds.DistributedTimedLock(lockDur)
+	if dlerr != nil {
+		t.Errorf("DistributedTimedLock(2) failed: %v", dlerr)
+	}
+
+	envstr := os.Getenv("ETCD_HOST")
+	if (envstr != "") {
+		t.Logf("Running ETCD detected, checking dist lock contention.")
+		time.Sleep(3 * time.Second)
+		_,dlerr2 := ds.DistributedTimedLock(2 * time.Second)
+		if dlerr2 == nil {
+			t.Errorf("DistributedTimedLock(3) should have failed, did not.")
+		}
+		ds.Unlock()
+	}
 }

vendor/modules.txt

@@ -60,6 +60,8 @@ github.com/google/uuid
 # github.com/gorilla/mux v1.8.0
 ## explicit
 github.com/gorilla/mux
+# github.com/gorilla/websocket v1.4.1
+## explicit
 # github.com/hashicorp/errwrap v1.0.0
 github.com/hashicorp/errwrap
 # github.com/hashicorp/go-cleanhttp v0.5.1
@@ -251,5 +253,7 @@ gopkg.in/square/go-jose.v2
 gopkg.in/square/go-jose.v2/cipher
 gopkg.in/square/go-jose.v2/json
 gopkg.in/square/go-jose.v2/jwt
+# gopkg.in/yaml.v2 v2.2.8
+## explicit
 # gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c
 gopkg.in/yaml.v3