feat: token expiration

Makefile

@@ -33,8 +33,8 @@ image: build
 .PHONY: install
 install:
 	@kubectl apply -k deploy
-	@kubectl rollout restart deployment.apps/spiracle
-	@kubectl rollout status deployment.apps/spiracle
+	@kubectl rollout restart statefulset.apps/spiracle
+	@kubectl rollout status statefulset.apps/spiracle
 
 .PHONY: deploy
 deploy: image install
@@ -45,8 +45,8 @@ clean:
 
 .PHONY: install-sample
 install-sample:
-	@kubectl apply -f deploy/samples/roomingress1.yaml
+	@kubectl apply -f deploy/samples/roomingress.yaml
 
 .PHONY: clean-sample
 clean-sample:
-	@kubectl delete -f deploy/samples/roomingress1.yaml
+	@kubectl delete -f deploy/samples/roomingress.yaml

api/v1/roomingress.go

@@ -15,41 +15,52 @@ func init() {
 
 type RoomIngressPlayer struct {
 	//+kubebuilder:validation:MinLength=1
-	Id    string `json:"id"`
-	//+kubebuilder:validation:Minimum=1
+	Id string `json:"id"`
+	//+kubebuilder:validation:Minimum=0
 	//+kubebuilder:validation:Maximum=4294967295
 	Token int64 `json:"token"`
 }
 
 type RoomIngressRoom struct {
 	//+kubebuilder:validation:MinLength=1
-	Id       string `json:"id,omitempty"`
+	Id string `json:"id,omitempty"`
 	//+kubebuilder:validation:MinLength=1
-	Server   string `json:"server,omitempty"`
+	Server string `json:"server,omitempty"`
 	//+kubebuilder:validation:MinLength=1
-	Upstream string `json:"upstream,omitempty"`
-	//+kubebuilder:validation:UniqueItems=true
+	Upstream string              `json:"upstream,omitempty"`
 	Players  []RoomIngressPlayer `json:"players"`
 }
 
 type RoomIngressSpec struct {
 	//+kubebuilder:validation:MinItems=1
-	//+kubebuilder:validation:UniqueItems=true
 	Rooms []RoomIngressRoom `json:"rooms,omitempty"`
 }
 
+// +kubebuilder:validation:Enum=Success;Pending;Failure;Expired;Retry
+type PlayerStatus string
+
+const PlayerStatusSuccess PlayerStatus = "Success"
+const PlayerStatusPending PlayerStatus = "Pending"
+const PlayerStatusFailure PlayerStatus = "Failure"
+const PlayerStatusExpired PlayerStatus = "Expired"
+const PlayerStatusRetry   PlayerStatus = "Retry"
+
 type RoomIngressPlayerStatus struct {
-	Id    string `json:"id"`
-	//+kubebuilder:validation:Minimum=1
+	Id string `json:"id"`
+	//+kubebuilder:validation:Minimum=0
 	//+kubebuilder:validation:Maximum=4294967295
-	Token int64 `json:"token"`
+	Token     int64        `json:"token"`
+	Timestamp metav1.Time  `json:"timestamp"`
+	Expire    metav1.Time  `json:"expire"`
+	Status    PlayerStatus `json:"status"`
+	Detail    string       `json:"detail"`
 }
 
 type RoomIngressRoomStatus struct {
-	Id      string `json:"id,omitempty"`
-	Server   string `json:"server,omitempty"`
-	Upstream string `json:"upstream,omitempty"`
-	Players []RoomIngressPlayerStatus `json:"players,omitempty"`
+	Id       string                    `json:"id,omitempty"`
+	Server   string                    `json:"server,omitempty"`
+	Upstream string                    `json:"upstream,omitempty"`
+	Players  []RoomIngressPlayerStatus `json:"players,omitempty"`
 }
 
 type RoomIngressStatus struct {
@@ -58,7 +69,7 @@ type RoomIngressStatus struct {
 
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
-//+kubebuilder:printcolumn:name="Server",type=string,JSONPath=`.spec.rules[0].server`
+//+kubebuilder:printcolumn:name="Server",type=string,JSONPath=`.spec.rooms[0].server`
 //+kubebuilder:resource:shortName="ring"
 type RoomIngress struct {
 	metav1.TypeMeta   `json:",inline"`

cmd/spiracle/controller.go

@@ -14,10 +14,10 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	manager "sigs.k8s.io/controller-runtime/pkg/manager"
 
 	v1 "github.com/LilithGames/spiracle/api/v1"
 	"github.com/LilithGames/spiracle/controllers"
-	"github.com/LilithGames/spiracle/infra/db"
 	"github.com/LilithGames/spiracle/repos"
 	//+kubebuilder:scaffold:imports
 )
@@ -34,19 +34,13 @@ func init() {
 	//+kubebuilder:scaffold:scheme
 }
 
-func controller(ctx context.Context) {
+func controller(ctx context.Context) manager.Manager {
 	metricsAddr := ":8080"
 	enableLeaderElection := true
 	probeAddr := ":8081"
 
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&zap.Options{})))
 
-	client, err := db.ProvideClient(ctx, db.ClientLocalConfig())
-	if err != nil {
-		setupLog.Error(err, "unable to connect local db")
-		os.Exit(1)
-	}
-
 	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
 		Scheme:                 scheme,
 		MetricsBindAddress:     metricsAddr,
@@ -60,10 +54,10 @@ func controller(ctx context.Context) {
 		os.Exit(1)
 	}
 	reconciler := &controllers.RoomIngressReconciler{
-		Client:  mgr.GetClient(),
-		Scheme:  mgr.GetScheme(),
-		Log:     ctrl.Log,
-		Routers: repos.NewClientRouterRepo(client),
+		Client:    mgr.GetClient(),
+		Scheme:    mgr.GetScheme(),
+		Log:       ctrl.Log,
+		TokenRepo: repos.NewTsTokenRepo(),
 	}
 	if err := reconciler.SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "RoomIngress")
@@ -79,8 +73,6 @@ func controller(ctx context.Context) {
 		os.Exit(1)
 	}
 	setupLog.Info("starting manager")
-	if err := mgr.Start(ctx); err != nil {
-		setupLog.Error(err, "problem running manager")
-		os.Exit(1)
-	}
+	mgr.GetFieldIndexer().IndexField(ctx, &v1.RoomIngress{}, "indexToken", repos.BuildIndexToken)
+	return mgr
 }

cmd/spiracle/db.go

@@ -4,23 +4,27 @@ import (
 	"context"
 	"log"
 	"time"
+	"sync"
 
 	"github.com/LilithGames/spiracle/infra/db"
 	"github.com/buraksezer/olric"
 )
 
-func database(ctx context.Context) *olric.Olric {
+func database(ctx context.Context, wg *sync.WaitGroup) *olric.Olric {
 	conf := db.ServerClusterConfig()
 	server, err := db.ProvideServer(ctx, conf)
 	if err != nil {
-		log.Fatalln("start db err", err)
+		log.Fatalln("[ERROR] Olric start db err", err)
 	}
 	go func() {
+		defer wg.Done()
 		<-ctx.Done()
 		sctx, _ := context.WithTimeout(context.TODO(), time.Second*10)
 		if err := server.Shutdown(sctx); err != nil {
-			log.Fatalln("shutdown db err", err)
+			log.Println("[ERROR] Olric shutdown err", err)
+			return
 		}
+		log.Println("[INFO] Olric is shutdown gracefully")
 	}()
 	return server
 }

cmd/spiracle/main.go

@@ -1,13 +1,26 @@
 package main
 
 import (
+	"sync"
+	"log"
 	ctrl "sigs.k8s.io/controller-runtime"
+	"github.com/LilithGames/spiracle/config"
 )
 
 func main() {
 	ctx := ctrl.SetupSignalHandler()
-	db := database(ctx)
-	go spiracle(ctx, db)
-	go controller(ctx)
-	<-ctx.Done()
+	conf, err := config.Load("config.yaml")
+	if err != nil {
+		log.Fatalln("load config err: ", err)
+	}
+	wg := &sync.WaitGroup{}
+	wg.Add(2)
+	db := database(ctx, wg)
+	mgr := controller(ctx)
+	go spiracle(ctx, conf, wg, db, mgr)
+	if err := mgr.Start(ctx); err != nil {
+		log.Println(err, "[ERROR] controller stop err")
+		return
+	}
+	wg.Wait()
 }

cmd/spiracle/proxy.go

@@ -3,23 +3,38 @@ package main
 import (
 	"context"
 	"log"
+	"sync"
+	"fmt"
 
+	manager "sigs.k8s.io/controller-runtime/pkg/manager"
 	"github.com/buraksezer/olric"
 	"github.com/LilithGames/spiracle/proxy"
 	"github.com/LilithGames/spiracle/services/roomproxy"
+	"github.com/LilithGames/spiracle/repos"
+	"github.com/LilithGames/spiracle/config"
 )
 
-func spiracle(ctx context.Context, db *olric.Olric) {
+func spiracle(ctx context.Context, conf *config.Config, wg *sync.WaitGroup, db *olric.Olric, mgr manager.Manager) {
+	defer wg.Done()
 	s := &proxy.Statd{}
-	// go s.Tick()
+	if conf.RoomProxy.Debug {
+		go s.Tick()
+	}
 	// maxproc
 	ctx = proxy.WithStatd(ctx, s)
-	rp, err := roomproxy.NewRoomProxy(ctx, "roomproxy", roomproxy.RoomProxyDb(db))
-	if err != nil {
-		log.Fatalln("create roomproxy err", err)
-	}
-	server := proxy.NewServer("0.0.0.0:4321", rp)
-	if err := server.Run(ctx); err != nil {
-		log.Println("roomproxy server err", err)
+	for _, s := range conf.RoomProxy.Servers {
+		rp, err := roomproxy.NewRoomProxy(ctx, s.Name, roomproxy.RoomProxyDb(db), roomproxy.RoomProxyRouterRepo(repos.NewK8sRouterRepo(mgr.GetClient())))
+		if err != nil {
+			log.Fatalln("create roomproxy err", err)
+		}
+		hostport := fmt.Sprintf("%s:%d", s.Host, s.Port)
+		server := proxy.NewServer(hostport, rp)
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			if err := server.Run(ctx); err != nil {
+				log.Println("roomproxy server err", err, "name", s.Name)
+			}
+		}()
 	}
 }

config.yaml

@@ -1,6 +1,9 @@
 roomproxy:
+  debug: false
   servers:
     - name: dev
+      host: 0.0.0.0
       port: 4000
     - name: alpha
+      host: 0.0.0.0
       port: 5000

config/config.go

@@ -4,6 +4,7 @@ import "github.com/jinzhu/configor"
 
 type Config struct {
 	RoomProxy struct {
+		Debug bool
 		Servers []struct {
 			Name string
 			Host string `default:"0.0.0.0"`