From 0319c33631551fc8fcf3559d7fe0e1bf6904d6a1 Mon Sep 17 00:00:00 2001 From: tangmingyou Date: Tue, 20 Feb 2024 15:22:50 +0800 Subject: [PATCH] gateway dispatch --- {benchmark => cmd/benchmark}/main.go | 12 +- {benchmark => cmd/benchmark}/prof/pprof.go | 0 cmd/gateway_ws/config.toml | 3 +- cmd/gateway_ws/main.go | 47 +++-- deploy_k8s/gateway_ws.yaml | 75 ++++++++ .../gateway_ws/gws_server/conn_handler.go | 179 +++++++++++------- internal/gateway_ws/gws_server/gws_conn.go | 1 + pkg/config/config.go | 1 + pkg/config/loader.go | 4 +- pkg/config/logger.go | 10 +- pkg/protocol/session/channel.go | 3 +- 11 files changed, 227 insertions(+), 108 deletions(-) rename {benchmark => cmd/benchmark}/main.go (96%) rename {benchmark => cmd/benchmark}/prof/pprof.go (100%) diff --git a/benchmark/main.go b/cmd/benchmark/main.go similarity index 96% rename from benchmark/main.go rename to cmd/benchmark/main.go index 9291fe3..46f1d07 100644 --- a/benchmark/main.go +++ b/cmd/benchmark/main.go @@ -34,6 +34,7 @@ import ( ) var ( + gatewayHttp = "http://192.168.110.41:7000" benchmarkMode = "deliver" // deliver/group mockUsers = 2000 eachUserSend = 100 @@ -41,7 +42,6 @@ var ( sendCounter int64 = 0 receiverCounter int64 = 0 //wsUrls = []string{"ws://192.168.110.36:7001", "ws://192.168.110.36:7003"} - gatewayHttp = "http://192.168.110.36:7000" httpClient *http.Client etcdClient *clientv3.Client seqId int32 @@ -52,7 +52,7 @@ var ( ) func init() { - config.InitLogger() + config.InitLogger(false) callbacks = make(map[int32]func(res any), 128) callbackMutex = &sync.Mutex{} @@ -72,10 +72,10 @@ func init() { func main() { runtime.GOMAXPROCS(runtime.NumCPU()) - flag.StringVar(&gatewayHttp, "gateway", "http://192.168.110.36:7000", "http gateway address") + flag.StringVar(&gatewayHttp, "gateway", "http://124.222.131.236:30830", "http gateway address") // 192.168.110.41:7000 flag.StringVar(&benchmarkMode, "mode", "deliver", "benchmark mode: deliver/group") - flag.IntVar(&mockUsers, "users", 10, "mock users") - flag.IntVar(&eachUserSend, "send", 10, "each user send msg count") + flag.IntVar(&mockUsers, "users", 1000, "mock users") + flag.IntVar(&eachUserSend, "send", 10000, "each user send msg count") flag.Parse() if benchmarkMode == "group" { @@ -223,7 +223,7 @@ func sendBatchChatMessage(ctx context.Context, uid string, conn *websocket.Conn, Receiver: receiverUid, Content: "hello", } - channel, err := send(conn, chat.Chat_ServiceDesc.ServiceName, "send", args) + channel, err := send(conn, chat.Chat_ServiceDesc.ServiceName, "Send", args) if err != nil { fmt.Println("send failed:", err) return diff --git a/benchmark/prof/pprof.go b/cmd/benchmark/prof/pprof.go similarity index 100% rename from benchmark/prof/pprof.go rename to cmd/benchmark/prof/pprof.go diff --git a/cmd/gateway_ws/config.toml b/cmd/gateway_ws/config.toml index 2d33c99..0a0dd75 100644 --- a/cmd/gateway_ws/config.toml +++ b/cmd/gateway_ws/config.toml @@ -1,11 +1,12 @@ [app] httpPort = 7001 -endpointAddress = "192.168.1.3:7001" +endpointAddress = "192.168.110.41:7001" subjectCacheTopic = "wsgate:subject:" subjectLrcExpiration = "10m" subjectLrcCleanupInterval = "5m" [grpc] +log = false address = ":7011" # postal cluster offset port +1000=8011 maxSendMsgSize = "8Mi" maxRecvMsgSize = "8Mi" diff --git a/cmd/gateway_ws/main.go b/cmd/gateway_ws/main.go index 4786c33..71ebed8 100644 --- a/cmd/gateway_ws/main.go +++ b/cmd/gateway_ws/main.go @@ -2,13 +2,11 @@ package main import ( "context" - "fmt" - "github.com/panjf2000/gnet/v2" clientv3 "go.etcd.io/etcd/client/v3" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" "runtime" - "sonet/internal/gateway_ws/gnet_server" + "sonet/internal/gateway_ws/gws_server" "sonet/internal/postal/group" "sonet/internal/postal/logic" "sonet/pkg/config" @@ -59,37 +57,38 @@ func main() { sessionStore := session.NewMapStore(128) groupStore := collect.NewConcurrentMap[string, *group.Group](128, func(k string) string { return k }) + ctx, cancel := context.WithCancel(context.Background()) + shutdown.AddHook(cancel) + // run websocket server //postalAddr := etcd.MustRegisterAddress(conf.Grpc.Address) //connHandler := server.NewConnHandler(postalAddr, grpcFactory, sessionStore, subjectStore) //httpServer := server.NewHttpServer(connHandler) - //gwsHandler := gws_server.NewGwsHandler(grpcFactory, sessionStore) - //httpServer := gws_server.NewGwsServer(gwsHandler) - //httpServer.Init() - //go func() { - // err = httpServer.Run(appConf.HttpPort) - // if err != nil { - // panic(err) - // } - //}() - - ctx, cancel := context.WithCancel(context.Background()) - shutdown.AddHook(cancel) - + gwsHandler := gws_server.NewGwsHandler(grpcFactory, sessionStore) + gwsHandler.Init(ctx) + httpServer := gws_server.NewGwsServer(gwsHandler) + httpServer.Init() go func() { - gnetWsServer := gnet_server.NewGnetWsServer(grpcFactory, sessionStore) - gnetWsServer.Init(ctx) - err := gnet.Run(gnetWsServer, - fmt.Sprintf("tcp://0.0.0.0:%d", appConf.HttpPort), - gnet.WithMulticore(true), - gnet.WithReusePort(true), - gnet.WithTicker(true), - ) + err = httpServer.Run(appConf.HttpPort) if err != nil { panic(err) } }() + //go func() { + // gnetWsServer := gnet_server.NewGnetWsServer(grpcFactory, sessionStore) + // gnetWsServer.Init(ctx) + // err := gnet.Run(gnetWsServer, + // fmt.Sprintf("tcp://0.0.0.0:%d", appConf.HttpPort), + // gnet.WithMulticore(true), + // gnet.WithReusePort(true), + // gnet.WithTicker(true), + // ) + // if err != nil { + // panic(err) + // } + //}() + // run postal cluster server postalPicker := deliver.NewPostalPicker(dis, grpc.WithTransportCredentials(insecure.NewCredentials())) if err = postalPicker.Init(ctx); err != nil { diff --git a/deploy_k8s/gateway_ws.yaml b/deploy_k8s/gateway_ws.yaml index 0b7ec5c..a059af3 100644 --- a/deploy_k8s/gateway_ws.yaml +++ b/deploy_k8s/gateway_ws.yaml @@ -144,3 +144,78 @@ spec: targetPort: 9100 selector: app: gateway-ws2 + + +--- +apiVersion: v1 +kind: Pod +metadata: + name: gateway-ws3 + namespace: sopod + labels: + app: gateway-ws3 +spec: + containers: + - name: gateway-ws3 + image: 10.0.16.5:3050/so_gateway_ws:1.0.0 + env: + - name: SO_ETCD.ENDPOINTS + value: etcd-svc:2379 + - name: SO_ETCD.USERNAME + value: "" + - name: SO_ETCD.PASSWORD + value: "" + - name: SO_PROMETHEUS.ENABLE + value: "false" + - name: SO_PROMETHEUS.PORT + value: "9100" + - name: SO_APP.HTTPPORT + value: "7003" + - name: SO_APP.ENDPOINTADDRESS + value: 124.222.131.236:30803 + - name: SO_REDIS.ADDR + value: "redis-svc:6379" + - name: SO_REDIS.PASSWORD + value: "" + - name: SO_NATS.URL + value: nats://nats-svc:4222 + imagePullPolicy: Always + ports: + - containerPort: 7003 + - containerPort: 30803 + - containerPort: 9100 + +--- +apiVersion: v1 +kind: Service +metadata: + name: gateway-ws3-svc + namespace: sopod + labels: + app: gateway-ws3-svc +spec: + type: NodePort + ports: + - name: gateway-ws3 + port: 7003 + targetPort: 7003 + nodePort: 30803 + selector: + app: gateway-ws3 + +--- +apiVersion: v1 +kind: Service +metadata: + name: gateway-ws3-metrics-svc + namespace: sopod + labels: + app: gateway-ws3-metrics-svc +spec: + type: ClusterIP + ports: + - name: metrics + port: 9100 + targetPort: 9100 + selector: + app: gateway-ws3 diff --git a/internal/gateway_ws/gws_server/conn_handler.go b/internal/gateway_ws/gws_server/conn_handler.go index 6aac344..e50e5c7 100644 --- a/internal/gateway_ws/gws_server/conn_handler.go +++ b/internal/gateway_ws/gws_server/conn_handler.go @@ -18,13 +18,18 @@ import ( const ( PingInterval = 5 * time.Second PingWait = 10 * time.Second - SessionUidKey = "uid" SessionGwsConnKey = "gws" ) +type soRequest struct { + conn *gwsConn + payload *protocol.Payload +} + type GwsHandler struct { grpcFactory *generic.GrpcGenericClientFactory sessionStore session.Store // 当前连接用户,内存缓存 + requests chan *soRequest } func NewGwsHandler( @@ -37,6 +42,93 @@ func NewGwsHandler( } } +func (c *GwsHandler) Init(ctx context.Context) { + c.requests = make(chan *soRequest, 1024*16) + + // todo 针对不同服务的协程池 + for i := 0; i < 32; i++ { + go c.dispatchRequest(ctx) + } +} + +// dispatchRequest 多个 goroutine 进行请求 dispatch +func (c *GwsHandler) dispatchRequest(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case request := <-c.requests: + func() { + conn, payload := request.conn, request.payload + header := payload.Header + service, method := header.Svc, header.Target + + // grpc generic call + ctx := context.Background() + ctx2, cancel2 := context.WithTimeout(ctx, time.Second*3) + defer cancel2() + grpcClient, err := c.grpcFactory.GetClient(ctx2, service) + if err != nil { + logger.Error("get grpc generic client error: ", err) + writeError(conn, header.SeqId, fmt.Sprintf("service %s not avaliable: %s", header.Svc, err.Error())) + return + } + + // put grpc request session + if conn.uid != "" { + ctx = session.PutSubject(ctx, session.NewRpcSubject(conn.uid)) + } + ctx3, cancel3 := context.WithTimeout(ctx, time.Second*10) + defer cancel3() + resp, err := grpcClient.InvokeUnary(ctx3, header.Target, payload.Body) + if err != nil { + writeError(conn, header.SeqId, unwrapRpcError(err.Error())) + logger.Error("grpc generic call error: ", err) + return + } + + // write response + if resp == nil { // proto.Empty + header.Target = "" + } else { + payload.Body, err = resp.Marshal() + if err != nil { + logger.Error("generic call response marshal error: ", err) + writeError(conn, header.SeqId, "server error") + return + } + header.Target = resp.XXX_MessageName() + } + + // auth verify success + if service == "Auth" && method == "Verify" { + var channel *session.Channel + conn.uid, channel, err = c.extractAuthVerify(conn, payload.Body) + if err != nil { + logger.Error("extractAuthVerify error: ", err) + writeError(conn, header.SeqId, "server error") + return + } + // channel todo dispatch on conn open + go c.dispatch(channel) + } + + header.Type = protocol.TypeResponse + resMessage, err := protocol.EncodeSo(payload) + if err != nil { + logger.Error("grpc generic call error: ", err) + return + } + err = conn.Write(resMessage) + if err != nil { + logger.Error("gws write response error: ", err) + return + } + }() + } + } +} + func (c *GwsHandler) OnOpen(socket *gws.Conn) { _ = socket.SetDeadline(time.Time{}) // time.Now().Add(PingInterval + PingWait) socket.Session().Store(SessionGwsConnKey, newGwsConn(socket)) @@ -46,13 +138,15 @@ func (c *GwsHandler) OnClose(socket *gws.Conn, err error) { if err != nil { logger.Error("gws conn close error: ", err) } - val, ok := socket.Session().Load(SessionUidKey) + val, ok := socket.Session().Load(SessionGwsConnKey) if !ok { return } - uid := val.(string) - c.sessionStore.Delete(uid) - logger.Info("subject offline: ", uid) + conn := val.(*gwsConn) + if conn.uid != "" { + c.sessionStore.Delete(conn.uid) + logger.Info("subject offline: ", conn.uid) + } } func (c *GwsHandler) OnPing(socket *gws.Conn, payload []byte) { @@ -70,6 +164,7 @@ func (c *GwsHandler) OnMessage(socket *gws.Conn, message *gws.Message) { } }() if message.Opcode != gws.OpcodeBinary { + logger.Info("receive websocket message type: ", message.Opcode) return } @@ -79,12 +174,7 @@ func (c *GwsHandler) OnMessage(socket *gws.Conn, message *gws.Message) { return } conn := val.(*gwsConn) - - val, authorized := socket.Session().Load(SessionUidKey) - var sessionUid string - if authorized { - sessionUid = val.(string) - } + authorized := conn.uid != "" messageBytes := message.Bytes() payload, err := protocol.DecodeSo(messageBytes) @@ -100,67 +190,12 @@ func (c *GwsHandler) OnMessage(socket *gws.Conn, message *gws.Message) { return } - // grpc generic call - ctx := context.Background() - ctx2, cancel2 := context.WithTimeout(ctx, time.Second*3) - defer cancel2() - grpcClient, err := c.grpcFactory.GetClient(ctx2, service) - if err != nil { - logger.Error("get grpc generic client error: ", err) - writeError(conn, header.SeqId, fmt.Sprintf("service %s not avaliable: %s", header.Svc, err.Error())) - return - } - // put session - if authorized { - ctx = session.PutSubject(ctx, session.NewRpcSubject(sessionUid)) - } - - ctx3, cancel3 := context.WithTimeout(ctx, time.Second*10) - defer cancel3() - resp, err := grpcClient.InvokeUnary(ctx3, header.Target, payload.Body) - if err != nil { - writeError(conn, header.SeqId, unwrapRpcError(err.Error())) - logger.Error("grpc generic call error: ", err) - return - } - - // write response - if resp == nil { // proto.Empty - header.Target = "" - } else { - payload.Body, err = resp.Marshal() - if err != nil { - logger.Error("generic call response marshal error: ", err) - writeError(conn, header.SeqId, "server error") - return - } - header.Target = resp.XXX_MessageName() - } - - // auth verify success - if service == "Auth" && method == "Verify" { - var channel *session.Channel - sessionUid, channel, err = c.extractAuthVerify(conn, payload.Body) - if err != nil { - logger.Error("extractAuthVerify error: ", err) - writeError(conn, header.SeqId, "server error") - return - } - socket.Session().Store(SessionUidKey, sessionUid) - // channel - go c.dispatch(channel) - } - - header.Type = protocol.TypeResponse - resMessage, err := protocol.EncodeSo(payload) - if err != nil { - logger.Error("grpc generic call error: ", err) - return - } - err = conn.Write(resMessage) - if err != nil { - logger.Error("gws write response error: ", err) - return + // payload -> processing channel + request := &soRequest{conn: conn, payload: payload} + select { + case c.requests <- request: + default: + writeError(conn, header.SeqId, "server busy") } } diff --git a/internal/gateway_ws/gws_server/gws_conn.go b/internal/gateway_ws/gws_server/gws_conn.go index 0e26135..e43dc91 100644 --- a/internal/gateway_ws/gws_server/gws_conn.go +++ b/internal/gateway_ws/gws_server/gws_conn.go @@ -5,6 +5,7 @@ import ( ) type gwsConn struct { + uid string conn *gws.Conn } diff --git a/pkg/config/config.go b/pkg/config/config.go index 53ae9ca..b691545 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -22,6 +22,7 @@ type Configuration struct { } type GrpcConfig struct { + Log bool Address string NoReflection bool MaxSendMsgSize string diff --git a/pkg/config/loader.go b/pkg/config/loader.go index 06112bb..86081ff 100644 --- a/pkg/config/loader.go +++ b/pkg/config/loader.go @@ -25,8 +25,6 @@ func parseConfPathFlag(confPath string) (filePath, fileName, confName, confType // appConf service custom config // return common service configuration func LoadConfig(appConf any, confPathArg ...string) *Configuration { - InitLogger() - confPath := "" confName := "config" confType := "toml" @@ -83,6 +81,8 @@ func LoadConfig(appConf any, confPathArg ...string) *Configuration { panic(err) } } + + InitLogger(conf.Grpc.Log) return conf } diff --git a/pkg/config/logger.go b/pkg/config/logger.go index 019a935..1058818 100644 --- a/pkg/config/logger.go +++ b/pkg/config/logger.go @@ -2,13 +2,19 @@ package config import ( "github.com/sirupsen/logrus" + "google.golang.org/grpc/grpclog" + "sonet/pkg/utils/logger" ) -func InitLogger() { +func InitLogger(grpcLog bool) { logrus.SetFormatter(&logrus.TextFormatter{ ForceColors: true, TimestampFormat: "2006-01-02 15:04:05", //时间格式 FullTimestamp: true, }) - // grpclog.SetLoggerV2(logger.Logger) + + // logging grpc + if grpcLog { + grpclog.SetLoggerV2(logger.Logger) + } } diff --git a/pkg/protocol/session/channel.go b/pkg/protocol/session/channel.go index 2b650cd..403f402 100644 --- a/pkg/protocol/session/channel.go +++ b/pkg/protocol/session/channel.go @@ -13,6 +13,7 @@ type NetConn interface { Write([]byte) error } +// Channel todo close channel type Channel struct { Uid string Conn NetConn @@ -25,7 +26,7 @@ func NewChannel(uid string, conn NetConn) *Channel { return &Channel{ Uid: uid, Conn: conn, - ch: make(chan *postal.Message, 16), + ch: make(chan *postal.Message, 32), lock: &sync.Mutex{}, } }