Unverified Commit 6470d811 authored by Simon Esposito's avatar Simon Esposito Committed by GitHub
Browse files

Add storage engine index (#1056) 

Introduce storage index runtime functions:
* RegisterStorageIndex
* RegisterStorageIndexFilter
* StorageIndexList
parent 83ac9c4e

CHANGELOG.md

+1 −0
Original line number Diff line number Diff line
@@ -7,6 +7,7 @@ The format is based on [keep a changelog](http://keepachangelog.com) and this pr 
### Added

- Introduce pagination for console API leaderboard and tournament listing endpoint.

- Introduce pagination for devconsole leaderboard view.

- Add storage object indexing support and related runtime functions.



### Changed

- Better formatting for graphed values in devconsole status view.

go.mod

+1 −1
Original line number Diff line number Diff line
@@ -13,7 +13,7 @@ require ( 
	github.com/gorilla/mux v1.8.0

	github.com/gorilla/websocket v1.5.0

	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0

	github.com/heroiclabs/nakama-common v0.0.0-20230713130524-38774b285b66

	github.com/heroiclabs/nakama-common v1.27.1-0.20230717184507-dff09d7c8047

	github.com/jackc/pgconn v1.14.0

	github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa

	github.com/jackc/pgtype v1.14.0

go.sum

+2 −2
Original line number Diff line number Diff line
@@ -291,8 +291,8 @@ github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO 
github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0mNTz8vQ=

github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=

github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=

github.com/heroiclabs/nakama-common v0.0.0-20230713130524-38774b285b66 h1:hcZM9fsheM4xprMyszVL5n+YLl2MBGlhRIrYUV0/tg0=

github.com/heroiclabs/nakama-common v0.0.0-20230713130524-38774b285b66/go.mod h1:Os8XeXGvHAap/p6M/8fQ3gle4eEXDGRQmoRNcPQTjXs=

github.com/heroiclabs/nakama-common v1.27.1-0.20230717184507-dff09d7c8047 h1:BnNhnDQBeQyPigo2F162XwYYP7RxrEgxMJx992SLK+g=

github.com/heroiclabs/nakama-common v1.27.1-0.20230717184507-dff09d7c8047/go.mod h1:Os8XeXGvHAap/p6M/8fQ3gle4eEXDGRQmoRNcPQTjXs=

github.com/huandu/xstrings v1.3.1/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=

github.com/huandu/xstrings v1.3.2/go.mod h1:y5/lhBue+AyNmUVz9RLU9xbLR0o4KIIExikq4ovT0aE=

github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=

main.go

+15 −14
Original line number Diff line number Diff line
@@ -16,11 +16,8 @@ package main 


import (

	"context"

	cryptoRand "crypto/rand"

	"encoding/binary"

	"flag"

	"fmt"

	"math/rand"

	"net/http"

	"net/url"

	"os"
@@ -104,14 +101,6 @@ func main() { 
	startupLogger.Info("Node", zap.String("name", config.GetName()), zap.String("version", semver), zap.String("runtime", runtime.Version()), zap.Int("cpu", runtime.NumCPU()), zap.Int("proc", runtime.GOMAXPROCS(0)))

	startupLogger.Info("Data directory", zap.String("path", config.GetDataDir()))



	// Initialize the global random with strongly seed.

	var seed int64

	if err := binary.Read(cryptoRand.Reader, binary.BigEndian, &seed); err != nil {

		startupLogger.Warn("Failed to get strongly random seed, fallback to a less random one.", zap.Error(err))

		seed = time.Now().UnixNano()

	}

	rand.Seed(seed)



	redactedAddresses := make([]string, 0, 1)

	for _, address := range config.GetDatabase().Addresses {

		rawURL := fmt.Sprintf("postgres://%s", address)
@@ -153,7 +142,12 @@ func main() { 
	tracker.SetMatchJoinListener(matchRegistry.Join)

	tracker.SetMatchLeaveListener(matchRegistry.Leave)

	streamManager := server.NewLocalStreamManager(config, sessionRegistry, tracker)

	runtime, runtimeInfo, err := server.NewRuntime(ctx, logger, startupLogger, db, jsonpbMarshaler, jsonpbUnmarshaler, config, version, socialClient, leaderboardCache, leaderboardRankCache, leaderboardScheduler, sessionRegistry, sessionCache, statusRegistry, matchRegistry, tracker, metrics, streamManager, router)



	storageIndex, err := server.NewLocalStorageIndex(logger, db)

	if err != nil {

		logger.Fatal("Failed to initialize storage index", zap.Error(err))

	}

	runtime, runtimeInfo, err := server.NewRuntime(ctx, logger, startupLogger, db, jsonpbMarshaler, jsonpbUnmarshaler, config, version, socialClient, leaderboardCache, leaderboardRankCache, leaderboardScheduler, sessionRegistry, sessionCache, statusRegistry, matchRegistry, tracker, metrics, streamManager, router, storageIndex)

	if err != nil {

		startupLogger.Fatal("Failed initializing runtime modules", zap.Error(err))

	}
@@ -162,14 +156,21 @@ func main() { 
	tracker.SetPartyJoinListener(partyRegistry.Join)

	tracker.SetPartyLeaveListener(partyRegistry.Leave)



	storageIndex.RegisterFilters(runtime)

	go func() {

		if err = storageIndex.Load(ctx); err != nil {

			logger.Error("Failed to load storage index entries from database", zap.Error(err))

		}

	}()



	leaderboardScheduler.Start(runtime)

	googleRefundScheduler.Start(runtime)



	pipeline := server.NewPipeline(logger, config, db, jsonpbMarshaler, jsonpbUnmarshaler, sessionRegistry, statusRegistry, matchRegistry, partyRegistry, matchmaker, tracker, router, runtime)

	statusHandler := server.NewLocalStatusHandler(logger, sessionRegistry, matchRegistry, tracker, metrics, config.GetName())



	apiServer := server.StartApiServer(logger, startupLogger, db, jsonpbMarshaler, jsonpbUnmarshaler, config, version, socialClient, leaderboardCache, leaderboardRankCache, sessionRegistry, sessionCache, statusRegistry, matchRegistry, matchmaker, tracker, router, streamManager, metrics, pipeline, runtime)

	consoleServer := server.StartConsoleServer(logger, startupLogger, db, config, tracker, router, streamManager, metrics, sessionRegistry, sessionCache, consoleSessionCache, loginAttemptCache, statusRegistry, statusHandler, runtimeInfo, matchRegistry, configWarnings, semver, leaderboardCache, leaderboardRankCache, leaderboardScheduler, apiServer, runtime, cookie)

	apiServer := server.StartApiServer(logger, startupLogger, db, jsonpbMarshaler, jsonpbUnmarshaler, config, version, socialClient, storageIndex, leaderboardCache, leaderboardRankCache, sessionRegistry, sessionCache, statusRegistry, matchRegistry, matchmaker, tracker, router, streamManager, metrics, pipeline, runtime)

	consoleServer := server.StartConsoleServer(logger, startupLogger, db, config, tracker, router, streamManager, metrics, sessionRegistry, sessionCache, consoleSessionCache, loginAttemptCache, statusRegistry, statusHandler, runtimeInfo, matchRegistry, configWarnings, semver, leaderboardCache, leaderboardRankCache, leaderboardScheduler, storageIndex, apiServer, runtime, cookie)



	gaenabled := len(os.Getenv("NAKAMA_TELEMETRY")) < 1

	console.UIFS.Nt = !gaenabled

server/api.go

+3 −1
Original line number Diff line number Diff line
@@ -69,6 +69,7 @@ type ApiServer struct { 
	config               Config

	version              string

	socialClient         *social.Client

	storageIndex         StorageIndex

	leaderboardCache     LeaderboardCache

	leaderboardRankCache LeaderboardRankCache

	sessionCache         SessionCache
@@ -84,7 +85,7 @@ type ApiServer struct { 
	grpcGatewayServer    *http.Server

}



func StartApiServer(logger *zap.Logger, startupLogger *zap.Logger, db *sql.DB, protojsonMarshaler *protojson.MarshalOptions, protojsonUnmarshaler *protojson.UnmarshalOptions, config Config, version string, socialClient *social.Client, leaderboardCache LeaderboardCache, leaderboardRankCache LeaderboardRankCache, sessionRegistry SessionRegistry, sessionCache SessionCache, statusRegistry *StatusRegistry, matchRegistry MatchRegistry, matchmaker Matchmaker, tracker Tracker, router MessageRouter, streamManager StreamManager, metrics Metrics, pipeline *Pipeline, runtime *Runtime) *ApiServer {

func StartApiServer(logger *zap.Logger, startupLogger *zap.Logger, db *sql.DB, protojsonMarshaler *protojson.MarshalOptions, protojsonUnmarshaler *protojson.UnmarshalOptions, config Config, version string, socialClient *social.Client, storageIndex StorageIndex, leaderboardCache LeaderboardCache, leaderboardRankCache LeaderboardRankCache, sessionRegistry SessionRegistry, sessionCache SessionCache, statusRegistry *StatusRegistry, matchRegistry MatchRegistry, matchmaker Matchmaker, tracker Tracker, router MessageRouter, streamManager StreamManager, metrics Metrics, 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.
@@ -119,6 +120,7 @@ func StartApiServer(logger *zap.Logger, startupLogger *zap.Logger, db *sql.DB, p 
		socialClient:         socialClient,

		leaderboardCache:     leaderboardCache,

		leaderboardRankCache: leaderboardRankCache,

		storageIndex:         storageIndex,

		sessionCache:         sessionCache,

		sessionRegistry:      sessionRegistry,

		statusRegistry:       statusRegistry,