Merge pull request #409 from gibertoni/fix/panic-on-etcd-revoke

Fix panic caused by ETCD client timeout on Revoke()

cluster/etcd_service_discovery.go

@@ -27,12 +27,13 @@ import (
 	"strings"
 	"sync"
 	"time"
+
 	"github.com/topfreegames/pitaya/v2/config"
 	"github.com/topfreegames/pitaya/v2/constants"
 	"github.com/topfreegames/pitaya/v2/logger"
 	"github.com/topfreegames/pitaya/v2/util"
-	clientv3 "go.etcd.io/etcd/client/v3"
 	logutil "go.etcd.io/etcd/client/pkg/v3/logutil"
+	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/namespace"
 	"google.golang.org/grpc"
 )
@@ -81,14 +82,14 @@ func NewEtcdServiceDiscovery(
 		client = cli[0]
 	}
 	sd := &etcdServiceDiscovery{
-		running:         false,
-		server:          server,
-		serverMapByType: make(map[string]map[string]*Server),
-		listeners:       make([]SDListener, 0),
-		stopChan:        make(chan bool),
-		stopLeaseChan:   make(chan bool),
-		appDieChan:      appDieChan,
-		cli:             client,
+		running:            false,
+		server:             server,
+		serverMapByType:    make(map[string]map[string]*Server),
+		listeners:          make([]SDListener, 0),
+		stopChan:           make(chan bool),
+		stopLeaseChan:      make(chan bool),
+		appDieChan:         appDieChan,
+		cli:                client,
 		syncServersRunning: make(chan bool),
 	}
 
@@ -300,7 +301,7 @@ func (sd *etcdServiceDiscovery) GetServersByType(serverType string) (map[string]
 		// Create a new map to avoid concurrent read and write access to the
 		// map, this also prevents accidental changes to the list of servers
 		// kept by the service discovery.
-		ret := make(map[string]*Server,len(sd.serverMapByType[serverType]))
+		ret := make(map[string]*Server, len(sd.serverMapByType[serverType]))
 		for k, v := range sd.serverMapByType[serverType] {
 			ret[k] = v
 		}
@@ -615,16 +616,15 @@ func (sd *etcdServiceDiscovery) revoke() error {
 	go func() {
 		defer close(c)
 		logger.Log.Debug("waiting for etcd revoke")
-		_, err := sd.cli.Revoke(context.TODO(), sd.leaseID)
+		ctx, cancel := context.WithTimeout(context.Background(), sd.revokeTimeout)
+		_, err := sd.cli.Revoke(ctx, sd.leaseID)
+		cancel()
 		c <- err
 		logger.Log.Debug("finished waiting for etcd revoke")
 	}()
 	select {
 	case err := <-c:
 		return err // completed normally
-	case <-time.After(sd.revokeTimeout):
-		logger.Log.Warn("timed out waiting for etcd revoke")
-		return nil // timed out
 	}
 }