Set console gateway timeout to match connection idle timeout value. (1538734f) · Commits · public-projects / Nakama

CHANGELOG.md

+4 −0

Original line number	Diff line number	Diff line
		@@ -4,6 +4,9 @@ All notable changes to this project are documented below.
		The format is based on [keep a changelog](http://keepachangelog.com) and this project uses [semantic versioning](http://semver.org).

		## [Unreleased]
		### Added
		- Strict validation of socket timeout configuration parameters.

		### Changed
		- Default maximum database connection lifetime is now 1 hour.
		- Improved parsing of client IP and port for incoming requests and socket connections.
		@@ -11,6 +14,7 @@ The format is based on [keep a changelog](http://keepachangelog.com) and this pr

		### Fixed
		- CRON expressions for leaderboard and tournament resets now allow concurrent processing.
		- Set console gateway timeout to match connection idle timeout value.

		## [2.4.0] - 2019-02-03
		### Added

server/api.go

+6 −4

Original line number	Diff line number	Diff line
		@@ -80,11 +80,14 @@ type ApiServer struct {
		}

		func StartApiServer(logger zap.Logger, startupLogger zap.Logger, db sql.DB, jsonpbMarshaler jsonpb.Marshaler, jsonpbUnmarshaler jsonpb.Unmarshaler, config Config, socialClient social.Client, leaderboardCache LeaderboardCache, leaderboardRankCache LeaderboardRankCache, sessionRegistry SessionRegistry, matchRegistry MatchRegistry, matchmaker Matchmaker, tracker Tracker, router MessageRouter, pipeline Pipeline, runtime Runtime) *ApiServer {
		var gatewayContextTimeoutMs string
		if config.GetSocket().IdleTimeoutMs > 500 {
		// Ensure the GRPC Gateway timeout is just under the idle timeout (if possible) to ensure it has priority.
		grpcRuntime.DefaultContextTimeout = time.Duration(config.GetSocket().IdleTimeoutMs-500) * time.Millisecond
		gatewayContextTimeoutMs = fmt.Sprintf("%vm", config.GetSocket().IdleTimeoutMs-500)
		} else {
		grpcRuntime.DefaultContextTimeout = time.Duration(config.GetSocket().IdleTimeoutMs) * time.Millisecond
		gatewayContextTimeoutMs = fmt.Sprintf("%vm", config.GetSocket().IdleTimeoutMs)
		}

		serverOpts := []grpc.ServerOption{
		@@ -194,10 +197,9 @@ func StartApiServer(logger zap.Logger, startupLogger zap.Logger, db *sql.DB, j
		handlerWithCompressResponse.ServeHTTP(w, r)
		})
		grpcGatewayRouter.NewRoute().HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Do not allow clients to set certain headers.
		// Currently disallowed headers:
		// "Grpc-Timeout"
		r.Header.Del("Grpc-Timeout")
		// Ensure some headers have required values.
		// Override any value set by the client if needed.
		r.Header.Set("Grpc-Timeout", gatewayContextTimeoutMs)

		// Allow GRPC Gateway to handle the request.
		handlerWithMaxBody.ServeHTTP(w, r)

server/config.go

+38 −6

Original line number	Diff line number	Diff line
		@@ -115,6 +115,18 @@ func CheckConfig(logger *zap.Logger, config Config) {
		if config.GetRuntime().HTTPKey == "" {
		logger.Fatal("Runtime HTTP key must be set", zap.String("param", "runtime.http_key"))
		}
		if config.GetConsole().MaxMessageSizeBytes < 1 {
		logger.Fatal("Console max message size bytes must be >= 1", zap.Int64("console.max_message_size_bytes", config.GetConsole().MaxMessageSizeBytes))
		}
		if config.GetConsole().ReadTimeoutMs < 1 {
		logger.Fatal("Console read timeout milliseconds must be >= 1", zap.Int("console.read_timeout_ms", config.GetConsole().ReadTimeoutMs))
		}
		if config.GetConsole().WriteTimeoutMs < 1 {
		logger.Fatal("Console write timeout milliseconds must be >= 1", zap.Int("console.write_timeout_ms", config.GetConsole().WriteTimeoutMs))
		}
		if config.GetConsole().IdleTimeoutMs < 1 {
		logger.Fatal("Console idle timeout milliseconds must be >= 1", zap.Int("console.idle_timeout_ms", config.GetConsole().IdleTimeoutMs))
		}
		if config.GetConsole().Username == "" {
		logger.Fatal("Console username must be set", zap.String("param", "console.username"))
		}
		@@ -124,6 +136,18 @@ func CheckConfig(logger *zap.Logger, config Config) {
		if p := config.GetSocket().Protocol; p != "tcp" && p != "tcp4" && p != "tcp6" {
		logger.Fatal("Socket protocol must be one of: tcp, tcp4, tcp6", zap.String("socket.protocol", config.GetSocket().Protocol))
		}
		if config.GetSocket().MaxMessageSizeBytes < 1 {
		logger.Fatal("Socket max message size bytes must be >= 1", zap.Int64("socket.max_message_size_bytes", config.GetSocket().MaxMessageSizeBytes))
		}
		if config.GetSocket().ReadTimeoutMs < 1 {
		logger.Fatal("Socket read timeout milliseconds must be >= 1", zap.Int("socket.read_timeout_ms", config.GetSocket().ReadTimeoutMs))
		}
		if config.GetSocket().WriteTimeoutMs < 1 {
		logger.Fatal("Socket write timeout milliseconds must be >= 1", zap.Int("socket.write_timeout_ms", config.GetSocket().WriteTimeoutMs))
		}
		if config.GetSocket().IdleTimeoutMs < 1 {
		logger.Fatal("Socket idle timeout milliseconds must be >= 1", zap.Int("socket.idle_timeout_ms", config.GetSocket().IdleTimeoutMs))
		}
		if config.GetSocket().PingPeriodMs >= config.GetSocket().PongWaitMs {
		logger.Fatal("Ping period value must be less than pong wait value", zap.Int("socket.ping_period_ms", config.GetSocket().PingPeriodMs), zap.Int("socket.pong_wait_ms", config.GetSocket().PongWaitMs))
		}
		@@ -524,6 +548,10 @@ func NewTrackerConfig() *TrackerConfig {
		// ConsoleConfig is configuration relevant to the embedded console.
		type ConsoleConfig struct {
		Port int `yaml:"port" json:"port" usage:"The port for accepting connections for the embedded console, listening on all interfaces."`
		MaxMessageSizeBytes int64 `yaml:"max_message_size_bytes" json:"max_message_size_bytes" usage:"Maximum amount of data in bytes allowed to be read from the client socket per message."`
		ReadTimeoutMs int `yaml:"read_timeout_ms" json:"read_timeout_ms" usage:"Maximum duration in milliseconds for reading the entire request."`
		WriteTimeoutMs int `yaml:"write_timeout_ms" json:"write_timeout_ms" usage:"Maximum duration in milliseconds before timing out writes of the response."`
		IdleTimeoutMs int `yaml:"idle_timeout_ms" json:"idle_timeout_ms" usage:"Maximum amount of time in milliseconds to wait for the next request when keep-alives are enabled."`
		Username string `yaml:"username" json:"username" usage:"Username for the embedded console. Default username is 'admin'."`
		Password string `yaml:"password" json:"password" usage:"Password for the embedded console. Default password is 'password'."`
		}
		@@ -532,6 +560,10 @@ type ConsoleConfig struct {
		func NewConsoleConfig() *ConsoleConfig {
		return &ConsoleConfig{
		Port: 7351,
		MaxMessageSizeBytes: 4096,
		ReadTimeoutMs: 10 * 1000,
		WriteTimeoutMs: 60 * 1000,
		IdleTimeoutMs: 300 * 1000,
		Username: "admin",
		Password: "password",
		}

server/console.go

+31 −8

Original line number	Diff line number	Diff line
		@@ -45,9 +45,17 @@ type ConsoleServer struct {
		}

		func StartConsoleServer(logger zap.Logger, startupLogger zap.Logger, config Config, db sql.DB) ConsoleServer {
		var gatewayContextTimeoutMs string
		if config.GetConsole().IdleTimeoutMs > 500 {
		// Ensure the GRPC Gateway timeout is just under the idle timeout (if possible) to ensure it has priority.
		gatewayContextTimeoutMs = fmt.Sprintf("%vm", config.GetConsole().IdleTimeoutMs-500)
		} else {
		gatewayContextTimeoutMs = fmt.Sprintf("%vm", config.GetConsole().IdleTimeoutMs)
		}

		serverOpts := []grpc.ServerOption{
		grpc.StatsHandler(&ocgrpc.ServerHandler{IsPublicEndpoint: true}),
		grpc.MaxRecvMsgSize(int(config.GetSocket().MaxMessageSizeBytes)),
		grpc.MaxRecvMsgSize(int(config.GetConsole().MaxMessageSizeBytes)),
		grpc.UnaryInterceptor(consoleInterceptorFunc(logger, config)),
		}
		grpcServer := grpc.NewServer(serverOpts...)
		@@ -77,7 +85,7 @@ func StartConsoleServer(logger zap.Logger, startupLogger zap.Logger, config Co
		dialAddr := fmt.Sprintf("127.0.0.1:%d", config.GetConsole().Port-3)
		dialOpts := []grpc.DialOption{
		//TODO (mo, zyro): Do we need to pass the statsHandler here as well?
		grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(int(config.GetSocket().MaxMessageSizeBytes))),
		grpc.WithDefaultCallOptions(grpc.MaxCallSendMsgSize(int(config.GetConsole().MaxMessageSizeBytes))),
		grpc.WithInsecure(),
		}

		@@ -100,9 +108,24 @@ func StartConsoleServer(logger zap.Logger, startupLogger zap.Logger, config Co
		grpcGatewayRouter.HandleFunc("/public/", func(w http.ResponseWriter, r *http.Request) {
		zpages.Handler.ServeHTTP(w, r)
		})
		// Enable compression on gateway responses.
		handlerWithGzip := handlers.CompressHandler(grpcGateway)
		grpcGatewayRouter.NewRoute().Handler(handlerWithGzip)

		// Enable max size check on requests coming arriving the gateway.
		// Enable compression on responses sent by the gateway.
		handlerWithCompressResponse := handlers.CompressHandler(grpcGateway)
		maxMessageSizeBytes := config.GetConsole().MaxMessageSizeBytes
		handlerWithMaxBody := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Check max body size before decompressing incoming request body.
		r.Body = http.MaxBytesReader(w, r.Body, maxMessageSizeBytes)
		handlerWithCompressResponse.ServeHTTP(w, r)
		})
		grpcGatewayRouter.NewRoute().HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Ensure some headers have required values.
		// Override any value set by the client if needed.
		r.Header.Set("Grpc-Timeout", gatewayContextTimeoutMs)

		// Allow GRPC Gateway to handle the request.
		handlerWithMaxBody.ServeHTTP(w, r)
		})

		// Enable CORS on all requests.
		CORSHeaders := handlers.AllowedHeaders([]string{"Authorization", "Content-Type", "User-Agent"})
		@@ -113,9 +136,9 @@ func StartConsoleServer(logger zap.Logger, startupLogger zap.Logger, config Co
		// Set up and start GRPC Gateway server.
		s.grpcGatewayServer = &http.Server{
		Addr: fmt.Sprintf(":%d", config.GetConsole().Port),
		ReadTimeout: time.Millisecond * time.Duration(int64(config.GetSocket().ReadTimeoutMs)),
		WriteTimeout: time.Millisecond * time.Duration(int64(config.GetSocket().WriteTimeoutMs)),
		IdleTimeout: time.Millisecond * time.Duration(int64(config.GetSocket().IdleTimeoutMs)),
		ReadTimeout: time.Millisecond * time.Duration(int64(config.GetConsole().ReadTimeoutMs)),
		WriteTimeout: time.Millisecond * time.Duration(int64(config.GetConsole().WriteTimeoutMs)),
		IdleTimeout: time.Millisecond * time.Duration(int64(config.GetConsole().IdleTimeoutMs)),
		Handler: handlerWithCORS,
		}