Bugfix/sd sigint (closes topfreegames#11) (topfreegames#34)

* Added a timeout for etcd's revoke function * Added graceful shutdown in case of etcd goes down * Update docs to include new configuration values and removed extra line in function
taoistcard · Aug 9, 2018 · 537a13a · 537a13a
1 parent c05b0ec
commit 537a13a
Show file tree

Hide file tree

Showing 8 changed files with 67 additions and 30 deletions.
diff --git a/Gopkg.lock b/Gopkg.lock
diff --git a/Gopkg.toml b/Gopkg.toml
@@ -51,7 +51,7 @@
 
 [[constraint]]
   name = "github.com/coreos/etcd"
-  version = "3.3.2"
+  version = "3.3.9"
 
 [[constraint]]
   name = "github.com/nats-io/go-nats"

diff --git a/app.go b/app.go
@@ -270,6 +270,7 @@ func startDefaultSD() {
 	app.serviceDiscovery, err = cluster.NewEtcdServiceDiscovery(
 		app.config,
 		app.server,
+		app.dieChan,
 	)
 	if err != nil {
 		logger.Log.Fatalf("error starting cluster service discovery component: %s", err.Error())

diff --git a/app_test.go b/app_test.go
@@ -77,7 +77,7 @@ func setup() {
 	initApp()
 	Configure(true, "testtype", Cluster, map[string]string{}, viper.New())
 
-	etcdSD, _ := cluster.NewEtcdServiceDiscovery(app.config, app.server)
+	etcdSD, _ := cluster.NewEtcdServiceDiscovery(app.config, app.server, app.dieChan)
 	typeOfetcdSD = reflect.TypeOf(etcdSD)
 
 	natsRPCServer, _ := cluster.NewNatsRPCServer(app.config, app.server, nil, app.dieChan)
@@ -183,7 +183,7 @@ func TestGetMetricsReporters(t *testing.T) {
 	assert.Equal(t, app.metricsReporters, GetMetricsReporters())
 }
 func TestGetServerByID(t *testing.T) {
-	r, err := cluster.NewEtcdServiceDiscovery(app.config, app.server)
+	r, err := cluster.NewEtcdServiceDiscovery(app.config, app.server, app.dieChan)
 	assert.NoError(t, err)
 	assert.NotNil(t, r)
 	SetServiceDiscoveryClient(r)
@@ -193,7 +193,7 @@ func TestGetServerByID(t *testing.T) {
 }
 
 func TestGetServersByType(t *testing.T) {
-	r, err := cluster.NewEtcdServiceDiscovery(app.config, app.server)
+	r, err := cluster.NewEtcdServiceDiscovery(app.config, app.server, app.dieChan)
 	assert.NoError(t, err)
 	assert.NotNil(t, r)
 	SetServiceDiscoveryClient(r)
@@ -232,7 +232,7 @@ func TestSetRPCClient(t *testing.T) {
 func TestSetServiceDiscovery(t *testing.T) {
 	initApp()
 	Configure(true, "testtype", Cluster, map[string]string{}, viper.New())
-	r, err := cluster.NewEtcdServiceDiscovery(app.config, app.server)
+	r, err := cluster.NewEtcdServiceDiscovery(app.config, app.server, app.dieChan)
 	assert.NoError(t, err)
 	assert.NotNil(t, r)
 	SetServiceDiscoveryClient(r)
@@ -399,7 +399,7 @@ func TestStartAndListenCluster(t *testing.T) {
 	initApp()
 	Configure(true, "testtype", Cluster, map[string]string{}, cfg)
 
-	etcdSD, err := cluster.NewEtcdServiceDiscovery(app.config, app.server, cli)
+	etcdSD, err := cluster.NewEtcdServiceDiscovery(app.config, app.server, app.dieChan, cli)
 	assert.NoError(t, err)
 	SetServiceDiscoveryClient(etcdSD)
 

diff --git a/cluster/etcd_service_discovery.go b/cluster/etcd_service_discovery.go
@@ -54,12 +54,16 @@ type etcdServiceDiscovery struct {
 	stopChan            chan bool
 	lastSyncTime        time.Time
 	listeners           []SDListener
+	revokeTimeout       time.Duration
+	bootstrapTimeout    time.Duration
+	appDieChan          chan bool
 }
 
 // NewEtcdServiceDiscovery ctor
 func NewEtcdServiceDiscovery(
 	config *config.Config,
 	server *Server,
+	appDieChan chan bool,
 	cli ...*clientv3.Client,
 ) (ServiceDiscovery, error) {
 	var client *clientv3.Client
@@ -72,6 +76,7 @@ func NewEtcdServiceDiscovery(
 		server:    server,
 		listeners: make([]SDListener, 0),
 		stopChan:  make(chan bool),
+		appDieChan: appDieChan,
 		cli:       client,
 	}
 
@@ -87,6 +92,8 @@ func (sd *etcdServiceDiscovery) configure() {
 	sd.heartbeatTTL = sd.config.GetDuration("pitaya.cluster.sd.etcd.heartbeat.ttl")
 	sd.logHeartbeat = sd.config.GetBool("pitaya.cluster.sd.etcd.heartbeat.log")
 	sd.syncServersInterval = sd.config.GetDuration("pitaya.cluster.sd.etcd.syncservers.interval")
+	sd.revokeTimeout = sd.config.GetDuration("pitaya.cluster.sd.etcd.revoke.timeout")
+	sd.bootstrapTimeout = sd.config.GetDuration("pitaya.cluster.sd.etcd.bootstrap.timeout")
 }
 
 func (sd *etcdServiceDiscovery) watchLeaseChan(c <-chan *clientv3.LeaseKeepAliveResponse) {
@@ -232,16 +239,28 @@ func (sd *etcdServiceDiscovery) GetServersByType(serverType string) (map[string]
 }
 
 func (sd *etcdServiceDiscovery) bootstrap() error {
-	err := sd.bootstrapLease()
-	if err != nil {
-		return err
-	}
-
-	err = sd.bootstrapServer(sd.server)
-	if err != nil {
+	c := make(chan error)
+	defer close(c)
+	go func() {
+		logger.Log.Infof("waiting for etcd connection")
+		err := sd.bootstrapLease()
+		if err != nil {
+			c <- err
+			return
+		}
+		err = sd.bootstrapServer(sd.server)
+		c <- err
+	}()
+	select {
+	case err := <-c:
 		return err
+	case <-time.After(sd.bootstrapTimeout):
+		logger.Log.Warn("timed out waiting for etcd connection")
+		if sd.appDieChan != nil {
+			sd.appDieChan <- true
+		}
+		return nil
 	}
-	return nil
 }
 
 // GetServer returns a server given it's id
@@ -369,12 +388,26 @@ func (sd *etcdServiceDiscovery) SyncServers() error {
 func (sd *etcdServiceDiscovery) Shutdown() error {
 	sd.running = false
 	close(sd.stopChan)
+	return sd.revoke()
+}
 
-	_, err := sd.cli.Revoke(context.TODO(), sd.leaseID)
-	if err != nil {
-		return err
+// revoke prevents Pitaya from crashing when etcd is not available
+func (sd *etcdServiceDiscovery) revoke()  error {
+	c := make(chan error)
+	defer close(c)
+	go func() {
+		logger.Log.Debug("waiting for etcd revoke")
+		_, err := sd.cli.Revoke(context.TODO(), sd.leaseID)
+		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
 	}
-	return nil
 }
 
 func (sd *etcdServiceDiscovery) addServer(sv *Server) {

diff --git a/cluster/etcd_service_discovery_test.go b/cluster/etcd_service_discovery_test.go
@@ -49,7 +49,8 @@ func getConfig(conf ...*viper.Viper) *config.Config {
 
 func getEtcdSD(t *testing.T, config *config.Config, server *Server, cli *clientv3.Client) *etcdServiceDiscovery {
 	t.Helper()
-	e, err := NewEtcdServiceDiscovery(config, server, cli)
+	appDieChan := make(chan bool)
+	e, err := NewEtcdServiceDiscovery(config, server, appDieChan, cli)
 	assert.NoError(t, err)
 	return e.(*etcdServiceDiscovery)
 }

diff --git a/config/config.go b/config/config.go
@@ -74,6 +74,8 @@ func (c *Config) fillDefaultValues() {
 		"pitaya.cluster.sd.etcd.endpoints":                      "localhost:2379",
 		"pitaya.cluster.sd.etcd.prefix":                         "pitaya/",
 		"pitaya.cluster.sd.etcd.heartbeat.ttl":                  "60s",
+		"pitaya.cluster.sd.etcd.bootstrap.timeout":              "60s",
+		"pitaya.cluster.sd.etcd.revoke.timeout":                 "5s",
 		"pitaya.cluster.sd.etcd.heartbeat.log":                  false,
 		"pitaya.cluster.sd.etcd.syncservers.interval":           "120s",
 		"pitaya.modules.bindingstorage.etcd.endpoints":          "localhost:2379",

diff --git a/docs/configuration.rst b/docs/configuration.rst
@@ -31,6 +31,14 @@ These configuration values configure service discovery for the default etcd serv
     - 60s
     - time.Time
     - Hearbeat interval for the etcd lease
+  * - pitaya.cluster.sd.etcd.bootstrap.timeout
+    - 60s
+    - time.Duration
+    - Connection timeout when waiting for etcd to become available
+  * - pitaya.cluster.sd.etcd.revoke.timeout
+    - 5s
+    - time.Duration
+    - Timeout for etcd's revoke function
   * - pitaya.cluster.sd.etcd.heartbeat.log
     - false
     - bool