feat(cluster): cluster and gateway

Cargo.toml

@@ -36,6 +36,9 @@ prometheus = "0.13.3"
 lazy_static = "1.4.0"
 md5 = "0.7.0"
 chrono = "0.4"
+async-trait = "0.1"
+local-ip-address = "0.6"
+redis = { version = "0.25", features = ["tokio-comp", "cluster", "json"] }
 
 # cargo install cargo-deb
 # Reference:  https://github.com/kornelski/cargo-deb

config-dist.toml

@@ -35,3 +35,6 @@ urls = [
 # Values: off, error, warn, info, debug, trace
 # level = "warn"
 
+[cluster.storage]
+model = "RedisStandalone"
+addr = "redis://127.0.0.1:6379"

gateway/config.go

@@ -0,0 +1,25 @@
+package main
+
+import "github.com/BurntSushi/toml"
+
+type Config struct {
+	LoadBalancing string
+	ListenAddr    string
+	Model         string
+	Addr          string
+}
+
+func ParseConfig() *Config {
+	cfg := &Config{
+		LoadBalancing: "random",
+		ListenAddr:    ":8080",
+	}
+	_, err := toml.DecodeFile("config.toml", cfg)
+	if err != nil {
+		_, err := toml.DecodeFile("/etc/live777/gateway/config.toml", cfg)
+		if err != nil {
+			panic(err)
+		}
+	}
+	return cfg
+}

gateway/go.mod

@@ -0,0 +1,11 @@
+module github.com/binbat/live777/gateway
+
+go 1.22.1
+
+require (
+	github.com/BurntSushi/toml v1.3.2 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/gorilla/mux v1.8.1 // indirect
+	github.com/redis/go-redis/v9 v9.5.1 // indirect
+)

gateway/go.sum

@@ -0,0 +1,10 @@
+github.com/BurntSushi/toml v1.3.2 h1:o7IhLm0Msx3BaB+n3Ag7L8EVlByGnpq14C4YWiu/gL8=
+github.com/BurntSushi/toml v1.3.2/go.mod h1:CxXYINrC8qIiEnFrOxCa7Jy5BFHlXnUU2pbicEuybxQ=
+github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44=
+github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/gorilla/mux v1.8.1 h1:TuBL49tXwgrFYWhqrNgrUNEY92u81SPhu7sTdzQEiWY=
+github.com/gorilla/mux v1.8.1/go.mod h1:AKf9I4AEqPTmMytcMc0KkNouC66V3BtZ4qD5fmWSiMQ=
+github.com/redis/go-redis/v9 v9.5.1 h1:H1X4D3yHPaYrkL5X06Wh6xNVM/pX0Ft4RV0vMGvLBh8=
+github.com/redis/go-redis/v9 v9.5.1/go.mod h1:hdY0cQFCN4fnSYT6TkisLufl/4W5UIXyv0b/CLO2V2M=

gateway/load_balancing.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"math/rand"
+	"sync/atomic"
+)
+
+type LoadBalancing interface {
+	Next(context.Context, Storage) (*Node, error)
+}
+
+type RandomLoadBalancing struct{}
+
+func (r *RandomLoadBalancing) Next(ctx context.Context, s Storage) (*Node, error) {
+	nodes, err := storage.GetAllNode(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if len(nodes) == 0 {
+		return nil, errors.New("there are no nodes to provide services")
+	}
+	index := rand.Intn(len(nodes))
+	return &nodes[index], nil
+}
+
+type LocalPollingLoadBalancing struct {
+	offset uint64
+}
+
+func (l *LocalPollingLoadBalancing) Next(ctx context.Context, s Storage) (*Node, error) {
+	nodes, err := storage.GetAllNode(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if len(nodes) == 0 {
+		return nil, errors.New("there are no nodes to provide services")
+	}
+	offset := atomic.AddUint64(&l.offset, 1)
+	return &nodes[int(offset)%len(nodes)], nil
+}

gateway/main.go

@@ -0,0 +1,118 @@
+package main
+
+import (
+	"embed"
+	"encoding/json"
+	"fmt"
+	"io/fs"
+	"log"
+	"net/http"
+	"net/http/httputil"
+
+	"github.com/gorilla/mux"
+)
+
+var config *Config
+
+var storage Storage
+
+var loadBalancing LoadBalancing
+
+//go:embed assets
+var assets embed.FS
+
+func init() {
+	config = ParseConfig()
+	switch config.Model {
+	case "RedisStandalone":
+		storage, _ = NewRedisStandaloneStorage(config.Addr)
+	}
+	data, _ := json.Marshal(config)
+	log.Printf("config %s", data)
+	switch config.LoadBalancing {
+	case "random":
+		loadBalancing = &RandomLoadBalancing{}
+	case "localPolling":
+		loadBalancing = &LocalPollingLoadBalancing{offset: 0}
+	}
+	if storage == nil {
+		panic("storage is null,please check config")
+	}
+	if loadBalancing == nil {
+		panic("loadBalancing is null,please check config")
+	}
+}
+
+func main() {
+	assets, err := fs.Sub(assets, "assets")
+	if err != nil {
+		panic(err)
+	}
+	r := mux.NewRouter()
+	r.HandleFunc("/whip/{room}", whipHandler)
+	r.HandleFunc("/whep/{room}", proxyHandler)
+	r.HandleFunc("/resource/{room}/{session}", proxyHandler)
+	r.HandleFunc("/resource/{room}/{session}/layer", proxyHandler)
+	r.PathPrefix("/").Handler(http.StripPrefix("/", http.FileServer(http.FS(assets))))
+	r.Use(loggingMiddleware)
+	r.Use(mux.CORSMethodMiddleware(r))
+	panic(http.ListenAndServe(config.ListenAddr, r))
+}
+
+func loggingMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		header, _ := json.Marshal(r.Header)
+		log.Printf("%s %s %s", r.Method, r.RequestURI, header)
+		next.ServeHTTP(w, r)
+	})
+}
+
+func whipHandler(w http.ResponseWriter, r *http.Request) {
+	room := extractRequestRoom(r)
+	ownership, err := storage.GetRoomOwnership(r.Context(), room)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if ownership != nil {
+		http.Error(w, fmt.Sprintf("room has been used,node %s", ownership.Addr), http.StatusInternalServerError)
+		return
+	}
+	next, err := loadBalancing.Next(r.Context(), storage)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	doProxy(w, r, next.Addr)
+}
+
+func proxyHandler(w http.ResponseWriter, r *http.Request) {
+	room := extractRequestRoom(r)
+	ownership, err := storage.GetRoomOwnership(r.Context(), room)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	if ownership == nil {
+		http.Error(w, "the room does not exist", http.StatusNotFound)
+		return
+	}
+	doProxy(w, r, ownership.Addr)
+}
+
+func extractRequestRoom(r *http.Request) string {
+	vars := mux.Vars(r)
+	return vars["room"]
+}
+
+func doProxy(w http.ResponseWriter, r *http.Request, node string) {
+	log.Printf("request URI : %s, Handler Node : %s", r.RequestURI, node)
+	proxy := httputil.ReverseProxy{
+		Director: func(req *http.Request) {
+			req.URL.Scheme = "http"
+			req.URL.Host = node
+			req.Host = node
+		},
+	}
+	proxy.ServeHTTP(w, r)
+}

gateway/node.go

@@ -0,0 +1,6 @@
+package main
+
+type Node struct {
+	Addr     string
+	Metadata string
+}