Merge

agent/agent.go

@@ -384,8 +384,12 @@ func (a *agentImpl) Kick(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	_, err = a.conn.Write(p)
-	return err
+	pWrite := pendingWrite{
+		ctx:  ctx,
+		data: p,
+	}
+	a.chSend <- pWrite
+	return nil
 }
 
 // SetLastAt sets the last at to now

app.go

@@ -342,17 +342,18 @@ func (app *App) Start() {
 				case s := <-sg:
 					logger.Log.Warn("got signal: ", s)
 					if s == syscall.SIGINT {
-						logger.Log.Warnf("Bypassing session draing due to SIGINT. %d sessions will be immediately terminated", app.sessionPool.GetSessionCount())
+						logger.Log.Warnf("Bypassing session draining due to SIGINT. %d sessions will be immediately terminated", app.sessionPool.GetSessionCount())
 					}
 					break loop
 				case <-timeoutTimer.C:
 					logger.Log.Warnf("Session drain has reached maximum timeout. %d sessions will be immediately terminated", app.sessionPool.GetSessionCount())
-				case <-time.After(app.config.Session.Drain.Timeout):
+					break loop
+				case <-time.After(app.config.Session.Drain.Period):
 					logger.Log.Infof("Waiting for all sessions to finish: %d sessions remaining...", app.sessionPool.GetSessionCount())
 				}
 			}
 		}
-		close(app.dieChan)
+		app.Shutdown()
 	}
 
 	logger.Log.Warn("server is stopping...")

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
 		}

cluster/nats_rpc_client_test.go

@@ -470,7 +470,7 @@ func TestNatsRPCClientCall(t *testing.T) {
 			ss.EXPECT().SetRequestInFlight(gomock.Any(), gomock.Any(), gomock.Any()).Times(2)
 
 			res, err := rpcClient.Call(context.Background(), protos.RPCType_Sys, rt, ss, msg, sv2)
-			assert.Equal(t, table.expected, res)
+			assert.Equal(t, table.expected.GetData(), res.GetData())
 			if table.err != nil {
 				assert.Error(t, err)
 				assert.Contains(t, err.Error(), table.err.Error())

config/config.go

@@ -596,6 +596,8 @@ func NewMemoryGroupConfig(conf *Config) *MemoryGroupConfig {
 type EtcdGroupServiceConfig struct {
 	DialTimeout        time.Duration `mapstructure:"dialtimeout"`
 	Endpoints          []string      `mapstructure:"endpoints"`
+	User               string        `mapstructure:"user"`
+	Pass               string        `mapstructure:"pass"`
 	Prefix             string        `mapstructure:"prefix"`
 	TransactionTimeout time.Duration `mapstructure:"transactiontimeout"`
 }
@@ -605,6 +607,8 @@ func NewDefaultEtcdGroupServiceConfig() *EtcdGroupServiceConfig {
 	return &EtcdGroupServiceConfig{
 		DialTimeout:        time.Duration(5 * time.Second),
 		Endpoints:          []string{"localhost:2379"},
+		User:               "",
+		Pass:               "",
 		Prefix:             "pitaya/",
 		TransactionTimeout: time.Duration(5 * time.Second),
 	}
@@ -623,6 +627,8 @@ func NewEtcdGroupServiceConfig(config *Config) *EtcdGroupServiceConfig {
 type ETCDBindingConfig struct {
 	DialTimeout time.Duration `mapstructure:"dialtimeout"`
 	Endpoints   []string      `mapstructure:"endpoints"`
+	User        string        `mapstructure:"user"`
+	Pass        string        `mapstructure:"pass"`
 	Prefix      string        `mapstructure:"prefix"`
 	LeaseTTL    time.Duration `mapstructure:"leasettl"`
 }
@@ -632,6 +638,8 @@ func NewDefaultETCDBindingConfig() *ETCDBindingConfig {
 	return &ETCDBindingConfig{
 		DialTimeout: time.Duration(5 * time.Second),
 		Endpoints:   []string{"localhost:2379"},
+		User:        "",
+		Pass:        "",
 		Prefix:      "pitaya/",
 		LeaseTTL:    time.Duration(5 * time.Hour),
 	}

config/viper_config.go

@@ -21,11 +21,12 @@
 package config
 
 import (
-	"github.com/mitchellh/mapstructure"
 	"reflect"
 	"strings"
 	"time"
 
+	"github.com/mitchellh/mapstructure"
+
 	"github.com/spf13/viper"
 )
 
@@ -91,6 +92,8 @@ func (c *Config) fillDefaultValues() {
 		"pitaya.cluster.rpc.server.nats.buffer.push":            natsRPCServerConfig.Buffer.Push,
 		"pitaya.cluster.sd.etcd.dialtimeout":                    etcdSDConfig.DialTimeout,
 		"pitaya.cluster.sd.etcd.endpoints":                      etcdSDConfig.Endpoints,
+		"pitaya.cluster.sd.etcd.user":                           etcdSDConfig.User,
+		"pitaya.cluster.sd.etcd.pass":                           etcdSDConfig.Pass,
 		"pitaya.cluster.sd.etcd.prefix":                         etcdSDConfig.Prefix,
 		"pitaya.cluster.sd.etcd.grantlease.maxretries":          etcdSDConfig.GrantLease.MaxRetries,
 		"pitaya.cluster.sd.etcd.grantlease.retryinterval":       etcdSDConfig.GrantLease.RetryInterval,
@@ -111,6 +114,8 @@ func (c *Config) fillDefaultValues() {
 		"pitaya.defaultpipelines.structvalidation.enabled": builderConfig.DefaultPipelines.StructValidation.Enabled,
 		"pitaya.groups.etcd.dialtimeout":                   etcdGroupServiceConfig.DialTimeout,
 		"pitaya.groups.etcd.endpoints":                     etcdGroupServiceConfig.Endpoints,
+		"pitaya.groups.etcd.user":                          etcdGroupServiceConfig.User,
+		"pitaya.groups.etcd.pass":                          etcdGroupServiceConfig.Pass,
 		"pitaya.groups.etcd.prefix":                        etcdGroupServiceConfig.Prefix,
 		"pitaya.groups.etcd.transactiontimeout":            etcdGroupServiceConfig.TransactionTimeout,
 		"pitaya.groups.memory.tickduration":                groupServiceConfig.TickDuration,
@@ -128,6 +133,8 @@ func (c *Config) fillDefaultValues() {
 		"pitaya.metrics.statsd.rate":                       statsdConfig.Statsd.Rate,
 		"pitaya.modules.bindingstorage.etcd.dialtimeout":   etcdBindingConfig.DialTimeout,
 		"pitaya.modules.bindingstorage.etcd.endpoints":     etcdBindingConfig.Endpoints,
+		"pitaya.modules.bindingstorage.etcd.user":          etcdBindingConfig.User,
+		"pitaya.modules.bindingstorage.etcd.pass":          etcdBindingConfig.Pass,
 		"pitaya.modules.bindingstorage.etcd.leasettl":      etcdBindingConfig.LeaseTTL,
 		"pitaya.modules.bindingstorage.etcd.prefix":        etcdBindingConfig.Prefix,
 		"pitaya.conn.ratelimiting.limit":                   rateLimitingConfig.Limit,

groups/etcd_group_service.go

@@ -6,12 +6,12 @@ import (
 	"sync"
 	"time"
 
-	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.etcd.io/etcd/client/v3/namespace"
-	"go.etcd.io/etcd/api/v3/mvccpb"
 	"github.com/topfreegames/pitaya/v2/config"
 	"github.com/topfreegames/pitaya/v2/constants"
 	"github.com/topfreegames/pitaya/v2/logger"
+	"go.etcd.io/etcd/api/v3/mvccpb"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/client/v3/namespace"
 )
 
 var (
@@ -54,6 +54,8 @@ func createBaseClient(config config.EtcdGroupServiceConfig) (*clientv3.Client, e
 	cli, err := clientv3.New(clientv3.Config{
 		Endpoints:   config.Endpoints,
 		DialTimeout: config.DialTimeout,
+		Password:    config.Pass,
+		Username:    config.User,
 	})
 	if err != nil {
 		return nil, err

helpers/helpers.go

@@ -64,7 +64,8 @@ func GetTestNatsServer(t *testing.T) *server.Server {
 // GetTestEtcd gets a test in memory etcd server
 func GetTestEtcd(t *testing.T) (*integration.ClusterV3, *clientv3.Client) {
 	t.Helper()
-	integration.BeforeTest(t)
+	// skip etcd go leak detection
+	integration.BeforeTest(t, integration.WithoutGoLeakDetection())
 	c := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
 	cli := c.RandClient()
 	return c, cli

modules/binding_storage.go

@@ -25,20 +25,22 @@ import (
 	"fmt"
 	"time"
 
-	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.etcd.io/etcd/client/v3/namespace"
 	"github.com/topfreegames/pitaya/v2/cluster"
 	"github.com/topfreegames/pitaya/v2/config"
 	"github.com/topfreegames/pitaya/v2/constants"
 	"github.com/topfreegames/pitaya/v2/logger"
 	"github.com/topfreegames/pitaya/v2/session"
+	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/client/v3/namespace"
 )
 
 // ETCDBindingStorage module that uses etcd to keep in which frontend server each user is bound
 type ETCDBindingStorage struct {
 	Base
 	cli             *clientv3.Client
 	etcdEndpoints   []string
+	etcdPassword    string
+	etcdUsername    string
 	etcdPrefix      string
 	etcdDialTimeout time.Duration
 	leaseTTL        time.Duration
@@ -57,6 +59,8 @@ func NewETCDBindingStorage(server *cluster.Server, sessionPool session.SessionPo
 	}
 	b.etcdDialTimeout = conf.DialTimeout
 	b.etcdEndpoints = conf.Endpoints
+	b.etcdPassword = conf.Pass
+	b.etcdUsername = conf.User
 	b.etcdPrefix = conf.Prefix
 	b.leaseTTL = conf.LeaseTTL
 	return b
@@ -159,6 +163,8 @@ func (b *ETCDBindingStorage) Init() error {
 		cli, err = clientv3.New(clientv3.Config{
 			Endpoints:   b.etcdEndpoints,
 			DialTimeout: b.etcdDialTimeout,
+			Username:    b.etcdUsername,
+			Password:    b.etcdPassword,
 		})
 		if err != nil {
 			return err