Merge branch 'master' into crawl-rule

Author: oliviagolden0

go.mod

@@ -8,8 +8,8 @@ require (
 	github.com/gorilla/mux v1.8.1
 	github.com/prometheus/client_golang v1.22.0
 	github.com/stretchr/testify v1.10.0
-	go.etcd.io/etcd/api/v3 v3.6.0
-	go.etcd.io/etcd/client/v3 v3.6.0
+	go.etcd.io/etcd/api/v3 v3.6.1
+	go.etcd.io/etcd/client/v3 v3.6.1
 	go.uber.org/zap v1.27.0
 	golang.org/x/net v0.41.0
 )
@@ -28,7 +28,7 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.62.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
-	go.etcd.io/etcd/client/pkg/v3 v3.6.0 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.6.1 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/sys v0.33.0 // indirect
 	golang.org/x/text v0.26.0 // indirect

go.sum

@@ -53,12 +53,12 @@ github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOf
 github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
-go.etcd.io/etcd/api/v3 v3.6.0 h1:vdbkcUBGLf1vfopoGE/uS3Nv0KPyIpUV/HM6w9yx2kM=
-go.etcd.io/etcd/api/v3 v3.6.0/go.mod h1:Wt5yZqEmxgTNJGHob7mTVBJDZNXiHPtXTcPab37iFOw=
-go.etcd.io/etcd/client/pkg/v3 v3.6.0 h1:nchnPqpuxvv3UuGGHaz0DQKYi5EIW5wOYsgUNRc365k=
-go.etcd.io/etcd/client/pkg/v3 v3.6.0/go.mod h1:Jv5SFWMnGvIBn8o3OaBq/PnT0jjsX8iNokAUessNjoA=
-go.etcd.io/etcd/client/v3 v3.6.0 h1:/yjKzD+HW5v/3DVj9tpwFxzNbu8hjcKID183ug9duWk=
-go.etcd.io/etcd/client/v3 v3.6.0/go.mod h1:Jzk/Knqe06pkOZPHXsQ0+vNDvMQrgIqJ0W8DwPdMJMg=
+go.etcd.io/etcd/api/v3 v3.6.1 h1:yJ9WlDih9HT457QPuHt/TH/XtsdN2tubyxyQHSHPsEo=
+go.etcd.io/etcd/api/v3 v3.6.1/go.mod h1:lnfuqoGsXMlZdTJlact3IB56o3bWp1DIlXPIGKRArto=
+go.etcd.io/etcd/client/pkg/v3 v3.6.1 h1:CxDVv8ggphmamrXM4Of8aCC8QHzDM4tGcVr9p2BSoGk=
+go.etcd.io/etcd/client/pkg/v3 v3.6.1/go.mod h1:aTkCp+6ixcVTZmrJGa7/Mc5nMNs59PEgBbq+HCmWyMc=
+go.etcd.io/etcd/client/v3 v3.6.1 h1:KelkcizJGsskUXlsxjVrSmINvMMga0VWwFF0tSPGEP0=
+go.etcd.io/etcd/client/v3 v3.6.1/go.mod h1:fCbPUdjWNLfx1A6ATo9syUmFVxqHH9bCnPLBZmnLmMY=
 go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
 go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
 go.opentelemetry.io/otel v1.34.0 h1:zRLXxLCgL1WyKsPVrgbSdMN4c0FMkDAskSTQP+0hdUY=

rules/engine.go

@@ -116,10 +116,10 @@ func newV3Engine(logger *zap.Logger, cl *v3.Client, options ...EngineOption) v3E
 	}
 	var baseEtcdLocker lock.RuleLocker
 	if opts.dontUseSharedLockSession {
-		baseEtcdLocker = lock.NewV3Locker(cl, opts.lockAcquisitionTimeout, opts.useTryLock)
+		baseEtcdLocker = lock.NewV3Locker(cl, opts.lockAcquisitionTimeout, opts.useTryLock, logger)
 	} else {
 		sessionManager := concurrency.NewSessionManager(cl, logger)
-		baseEtcdLocker = lock.NewSessionLocker(sessionManager.GetSession, opts.lockAcquisitionTimeout, false, opts.useTryLock)
+		baseEtcdLocker = lock.NewSessionLocker(sessionManager.GetSession, opts.lockAcquisitionTimeout, false, opts.useTryLock, logger)
 	}
 	metricsEtcdLocker := lock.WithMetrics(baseEtcdLocker, "etcd")
 	var baseLocker lock.RuleLocker

rules/engine_lock_test.go

@@ -39,9 +39,9 @@ func Test_ReentrantLocking(t *testing.T) {
 					var rlckr lock.RuleLocker
 					if useShared {
 						sessionManager := concurrency.NewSessionManager(cl, logger)
-						rlckr = lock.NewSessionLocker(sessionManager.GetSession, 5, closeSession, useTryLock)
+						rlckr = lock.NewSessionLocker(sessionManager.GetSession, 5, closeSession, useTryLock, logger)
 					} else {
-						baseEtcdLocker := lock.NewV3Locker(cl, 5, useTryLock)
+						baseEtcdLocker := lock.NewV3Locker(cl, 5, useTryLock, logger)
 						baseMapLocker := lock.NewMapLocker()
 						rlckr = lock.NewNestedLocker(baseMapLocker, baseEtcdLocker)
 					}
@@ -54,11 +54,7 @@ func Test_ReentrantLocking(t *testing.T) {
 					} else {
 						assert.Error(t, err2)
 					}
-					if closeSession {
-						assert.Error(t, rlck.Unlock())
-					} else {
-						assert.NoError(t, rlck.Unlock())
-					}
+					assert.NoError(t, rlck.Unlock())
 				})
 			}
 		}

rules/etcd.go

@@ -16,6 +16,8 @@ type etcdV3ReadAPI struct {
 	kV v3.KV
 }
 
+var ErrWatcherClosing = errors.New("Watcher closing")
+
 // This method is currently not used but is being kept around to limit
 // the blast radius of implementing batch gets for rule evaluations.
 // The arrangement of interfaces is not ideal and should be addressed
@@ -142,7 +144,7 @@ func (ev3kw *etcdV3KeyWatcher) next() (string, *string, error) {
 			ev3kw.reset()
 			err := ev3kw.w.Close()
 			if err == nil {
-				err = errors.New("Watcher closing")
+				err = ErrWatcherClosing
 			}
 			return "", nil, err
 		case wr, stillOpen := <-ev3kw.ch:

rules/lock/lock.go

@@ -2,8 +2,10 @@ package lock
 
 import (
 	"errors"
+	"strings"
 	"time"
 
+	"go.uber.org/zap"
 	"golang.org/x/net/context"
 
 	v3 "go.etcd.io/etcd/client/v3"
@@ -21,28 +23,29 @@ type RuleLock interface {
 type GetSession func(context.Context) (*v3c.Session, error)
 
 // NewV3Locker creates a locker backed by etcd V3.
-func NewV3Locker(cl *v3.Client, lockTimeout int, useTryLock bool) RuleLocker {
+func NewV3Locker(cl *v3.Client, lockTimeout int, useTryLock bool, logger *zap.Logger) RuleLocker {
 	// The TTL is for the lease associated with the session, in seconds. While the session is still open,
 	// the lease's TTL will keep getting renewed to keep it from expiring, so all this really does is
 	// set the amount of time it takes for the lease to expire if the lease stops being renewed due
 	// to the application shutting down before a session could be properly closed.
 	newSession := func(_ context.Context) (*v3c.Session, error) {
 		return v3c.NewSession(cl, v3c.WithTTL(30))
 	}
-	return NewSessionLocker(newSession, lockTimeout, true, useTryLock)
+	return NewSessionLocker(newSession, lockTimeout, true, useTryLock, logger)
 }
 
 // NewSessionLocker creates a new locker with the provided session constructor. Note that
 // if closeSession is false, it means that the session provided by getSession will not be
 // closed but instead be reused. In that case the locker must be protected by another locker
 // (for instance an in-memory locker) because locks within the same session are reentrant
 // which means that two goroutines can obtain the same lock.
-func NewSessionLocker(getSession GetSession, lockTimeout int, closeSession, useTryLock bool) RuleLocker {
+func NewSessionLocker(getSession GetSession, lockTimeout int, closeSession, useTryLock bool, logger *zap.Logger) RuleLocker {
 	return &v3Locker{
 		lockTimeout:  lockTimeout,
 		newSession:   getSession,
 		closeSession: closeSession,
 		useTryLock:   useTryLock,
+		logger:       logger,
 	}
 }
 
@@ -51,12 +54,13 @@ type v3Locker struct {
 	newSession   GetSession
 	closeSession bool
 	useTryLock   bool
+	logger       *zap.Logger
 }
 
 func (v3l *v3Locker) Lock(key string, options ...Option) (RuleLock, error) {
-	return v3l.lockWithTimeout(key, v3l.lockTimeout)
+	return v3l.lockWithTimeout(key, v3l.lockTimeout, v3l.logger)
 }
-func (v3l *v3Locker) lockWithTimeout(key string, timeout int) (RuleLock, error) {
+func (v3l *v3Locker) lockWithTimeout(key string, timeout int, logger *zap.Logger) (RuleLock, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
 	defer cancel()
 	s, err := v3l.newSession(ctx)
@@ -73,7 +77,8 @@ func (v3l *v3Locker) lockWithTimeout(key string, timeout int) (RuleLock, error)
 		return nil, err
 	}
 	lock := &v3Lock{
-		mutex: m,
+		mutex:  m,
+		logger: logger,
 	}
 	if v3l.closeSession {
 		lock.session = s
@@ -84,24 +89,45 @@ func (v3l *v3Locker) lockWithTimeout(key string, timeout int) (RuleLock, error)
 type v3Lock struct {
 	mutex   *v3c.Mutex
 	session *v3c.Session
+	logger  *zap.Logger
 }
 
 // ErrNilMutex indicates that the lock has a nil mutex
 var ErrNilMutex = errors.New("mutex is nil")
 
+// Max number of unlock retries
+var unlockMaxRetries = 5
+
+// This should be given every chance to complete, otherwise
+// a lock could prevent future interactions with a resource.
 func (v3l *v3Lock) Unlock(_ ...Option) error {
 	if v3l.mutex != nil {
-		// This should be given every chance to complete, otherwise
-		// a lock could prevent future interactions with a resource.
-		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
-		defer cancel()
-		err := v3l.mutex.Unlock(ctx)
+		var err error
+		for i := range unlockMaxRetries {
+			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
+			err = v3l.mutex.Unlock(ctx)
+			cancel()
+			if err != nil {
+				v3l.logger.Warn("Unlock error", zap.String("key", v3l.mutex.Key()), zap.Int("attempt", i+1), zap.Error(err))
+				time.Sleep(time.Minute)
+			} else {
+				break
+			}
+		}
+		if err != nil {
+			err = errors.Join(errors.New("Unlock:"), err)
+		}
+
 		// If the lock failed to be released, as least closing the session
 		// will allow the lease it is associated with to expire.
 		if v3l.session != nil {
 			serr := v3l.session.Close()
-			if err == nil {
-				err = serr
+			// The Unlock will close the session in some cases
+			if serr != nil && strings.Contains(serr.Error(), "lease not found") {
+				serr = nil
+			}
+			if err == nil && serr != nil {
+				err = errors.Join(errors.New("Close:"), serr)
 			}
 		}
 		return err

rules/lock/lock_test.go

@@ -7,12 +7,14 @@ import (
 	"github.com/stretchr/testify/require"
 	v3 "go.etcd.io/etcd/client/v3"
 	v3c "go.etcd.io/etcd/client/v3/concurrency"
+	"go.uber.org/zap"
 	"golang.org/x/net/context"
 
 	"github.com/IBM-Cloud/go-etcd-rules/rules/teststore"
 )
 
 func Test_V3Locker(t *testing.T) {
+	logger, _ := zap.NewDevelopment()
 	cfg, cl := teststore.InitV3Etcd(t)
 	c, err := v3.New(cfg)
 	require.NoError(t, err)
@@ -31,29 +33,31 @@ func Test_V3Locker(t *testing.T) {
 			rlckr := v3Locker{
 				newSession:  newSession,
 				lockTimeout: 5,
+				logger:      logger,
 			}
-			rlck, err1 := rlckr.Lock("test")
-			assert.NoError(t, err1)
-			_, err2 := rlckr.lockWithTimeout("test", 10)
-			assert.Error(t, err2)
-			assert.NoError(t, rlck.Unlock())
+			rlck, err := rlckr.Lock("test")
+			assert.NoError(t, err, err)
+			_, err = rlckr.lockWithTimeout("test", 10, logger)
+			assert.Error(t, err, err)
+			err = rlck.Unlock()
+			assert.NoError(t, err, err)
 
 			done1 := make(chan bool)
 			done2 := make(chan bool)
-
 			go func() {
-				lckr := NewV3Locker(c, 5, useTryLock)
-				lck, lErr := lckr.Lock("test1")
-				assert.NoError(t, lErr)
+				lckr := NewV3Locker(c, 5, useTryLock, logger)
+				lck, err := lckr.Lock("test1")
+				assert.NoError(t, err, err)
 				done1 <- true
 				<-done2
 				if lck != nil {
-					assert.NoError(t, lck.Unlock())
+					err = lck.Unlock()
+					assert.NoError(t, err, err)
 				}
 			}()
 			<-done1
 			_, err = rlckr.Lock("test1")
-			assert.Error(t, err)
+			assert.Error(t, err, err)
 			done2 <- true
 		})
 	}

rules/lock/metrics.go

@@ -34,9 +34,9 @@ func (ml metricLocker) Lock(key string, options ...Option) (RuleLock, error) {
 }
 
 func (ml metricLocker) Unlock(options ...Option) error {
-	opts := buildOptions(options...)
 	err := ml.RuleLock.Unlock(options...)
 	if err != nil {
+		opts := buildOptions(options...)
 		ml.observeUnlockError(ml.lockerName, opts.method, opts.pattern)
 	}
 	return err

rules/watcher.go

@@ -1,6 +1,7 @@
 package rules
 
 import (
+	"errors"
 	"strings"
 	"time"
 
@@ -54,11 +55,14 @@ func (w *watcher) isStopped() bool {
 func (w *watcher) singleRun() {
 	key, value, err := w.kw.next()
 	if err != nil {
-		w.logger.Error("Watcher error", zap.Error(err))
 		if strings.Contains(err.Error(), "connection refused") {
 			w.logger.Info("Cluster unavailable; waiting one minute to retry")
 			time.Sleep(time.Minute)
 		} else {
+			// Watcher are always closed periodically, no need to log that
+			if !errors.Is(err, ErrWatcherClosing) {
+				w.logger.Error("Watcher error", zap.Error(err))
+			}
 			// Maximum logging rate is 1 per second.
 			time.Sleep(time.Second)
 		}