diff --git a/api/postal.proto b/api/postal.proto index a239b4e..4207227 100644 --- a/api/postal.proto +++ b/api/postal.proto @@ -88,23 +88,23 @@ message ResEndpoint { // 集群之间接口调用，重定向消息... service PostalCluster { // socket不在当前节点，重新投递消息 - rpc Redirect(ReqRedirect) returns(ResDeliver); + // rpc Redirect(ReqRedirect) returns(ResDeliver); - rpc ReDeliver(ReqReDeliver) returns(ResDeliver); + rpc Redeliver(ReqRedeliver) returns(ResDeliver); } -message ReqReDeliver { - string receiver = 1; +message ReqRedeliver { + repeated string receivers = 1; Message msg = 2; int32 offset = 7; // consistent hash node offset, <0逆时针, >0顺时针, =0丢弃包 repeated int64 nodes = 8; // [ip:port,] 出现重复节点丢弃包 } -message ReqRedirect { - int32 ttl = 1; // 投递一次减一, 到0丢弃 - int32 redirectMethod = 2; // 10 deliver, 11 deliverBatch, 12 deliverGroup - // bool sync = 7; // 是否同步阻塞等待投递结果,否则放到channel顺序投递 - ReqDeliver deliver = 10; - ReqDeliverBatch deliverBatch = 11; - ReqDeliverGroup deliverGroup = 12; -} +//message ReqRedirect { +// int32 ttl = 1; // 投递一次减一, 到0丢弃 +// int32 redirectMethod = 2; // 10 deliver, 11 deliverBatch, 12 deliverGroup +// // bool sync = 7; // 是否同步阻塞等待投递结果,否则放到channel顺序投递 +// ReqDeliver deliver = 10; +// ReqDeliverBatch deliverBatch = 11; +// ReqDeliverGroup deliverGroup = 12; +//} diff --git a/cmd/gateway_ws/config.toml b/cmd/gateway_ws/config.toml index 5cfdfd1..8443775 100644 --- a/cmd/gateway_ws/config.toml +++ b/cmd/gateway_ws/config.toml @@ -27,12 +27,6 @@ endpoints = ["124.222.131.236:3279"] username = "root" password = "sopod@etcd" -[redis] -Addr = "124.222.131.236:3379" -Password = "sopod@redis#" -DB = 1 -MinIdleConns = 3 - [nats] Url = "nats://nats.sopod@124.222.131.236:3222" RetryOnFailedConnect = true diff --git a/cmd/gateway_ws/main.go b/cmd/gateway_ws/main.go index e9889d6..391b832 100644 --- a/cmd/gateway_ws/main.go +++ b/cmd/gateway_ws/main.go @@ -1,8 +1,7 @@ package main import ( - "github.com/nats-io/nats.go" - "github.com/redis/go-redis/v9" + "context" clientv3 "go.etcd.io/etcd/client/v3" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" @@ -14,11 +13,10 @@ import ( "sonet/pkg/grpc/discovery" "sonet/pkg/grpc/discovery/etcd" "sonet/pkg/grpc/generic" - "sonet/pkg/plugins/cache" "sonet/pkg/plugins/mq" + "sonet/pkg/protocol/deliver" "sonet/pkg/protocol/session" "sonet/pkg/utils/collect" - "sonet/pkg/utils/conver" "sonet/pkg/utils/shutdown" ) @@ -43,11 +41,6 @@ func main() { } shutdown.AddHook(func() { _ = etcdClient.Close() }, shutdown.WithOrderBack()) - subjectStore, producer, err := getSubjectMultiCache(appConf, conf.Redis, conf.Nats) - if err != nil { - panic(err) - } - dis := discovery.NewEtcdDiscovery(etcdClient) etcdResolver, err := dis.Resolver() if err != nil { @@ -67,7 +60,7 @@ func main() { postalAddr := etcd.MustRegisterAddress(conf.Grpc.Address) //connHandler := server.NewConnHandler(postalAddr, grpcFactory, sessionStore, subjectStore) //httpServer := server.NewHttpServer(connHandler) - gwsHandler := gws_server.NewGwsHandler(postalAddr, grpcFactory, sessionStore, subjectStore) + gwsHandler := gws_server.NewGwsHandler(postalAddr, grpcFactory, sessionStore) httpServer := gws_server.NewGwsServer(gwsHandler) httpServer.Init() @@ -78,20 +71,32 @@ func main() { } }() - // run postal server + // run postal cluster server + ctx, cancel := context.WithCancel(context.Background()) + shutdown.AddHook(cancel) + postalPicker := deliver.NewPostalPicker(dis, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err = postalPicker.Init(ctx); err != nil { + panic(err) + } clientFactory := client.NewGrpcDirectClientFactory(grpc.WithTransportCredentials(insecure.NewCredentials())) - postalServer := logic.NewPostalServer(appConf.EndpointAddress, sessionStore, groupStore, subjectStore, clientFactory, producer) + postalClusterServer := logic.NewPostalClusterServer(sessionStore, postalPicker, clientFactory) go func() { - err = postalServer.Run(conf.Grpc, dis) + err := postalClusterServer.Run(conf.Grpc.Address, config.GetGrpcOptions(conf.Grpc)...) if err != nil { panic(err) } }() - // run postal cluster server - postalClusterServer := logic.NewPostalClusterServer(postalServer, nil) + // run postal server + // nats mq + producer, err := mq.NewNatsProducer(conf.Nats) + if err != nil { + panic(err) + } + shutdown.AddHook(producer.Stop) + postalServer := logic.NewPostalServer(appConf.EndpointAddress, sessionStore, groupStore, postalClusterServer, producer) go func() { - err := postalClusterServer.Run(conf.Grpc.Address, config.GetGrpcOptions(conf.Grpc)...) + err = postalServer.Run(conf.Grpc, dis) if err != nil { panic(err) } @@ -99,34 +104,3 @@ func main() { shutdown.Await() } - -func getSubjectMultiCache(appConf *GatewayWsConfig, redisOptions redis.Options, natsOptions nats.Options) (*cache.LocalRemoteCache, mq.Producer, error) { - // initial cache... - rdb := redis.NewClient(&redisOptions) - subjectRedisCache := cache.NewRedisCache(appConf.SubjectCacheTopic, rdb) - shutdown.AddHook(func() { _ = rdb.Close() }) - - // nats mq - producer, err := mq.NewNatsProducer(natsOptions) - if err != nil { - return nil, nil, err - } - shutdown.AddHook(func() { producer.Stop() }) - consumer, err := mq.NewNatsConsumer(natsOptions) - if err != nil { - return nil, nil, err - } - shutdown.AddHook(func() { consumer.Stop() }) - - // 多级缓存 - subjectLrcOpts := cache.LocalRemoteCacheOptions{ - Topic: appConf.SubjectCacheTopic, - LocalExpiration: conver.MustParseDuration(appConf.SubjectLrcExpiration), - CleanupInterval: conver.MustParseDuration(appConf.SubjectLrcCleanupInterval), - Remote: subjectRedisCache, - Producer: producer, - Consumer: consumer, - } - subjectLrc, err := cache.NewLocalRemoteCache(subjectLrcOpts) - return subjectLrc, producer, err -} diff --git a/internal/gateway_ws/gws_server/conn_handler.go b/internal/gateway_ws/gws_server/conn_handler.go index 3b4a941..c1ef18b 100644 --- a/internal/gateway_ws/gws_server/conn_handler.go +++ b/internal/gateway_ws/gws_server/conn_handler.go @@ -8,7 +8,6 @@ import ( "regexp" "sonet/api/gen/auth" "sonet/pkg/grpc/generic" - "sonet/pkg/plugins/cache" "sonet/pkg/protocol" "sonet/pkg/protocol/session" "sonet/pkg/utils/logger" @@ -26,18 +25,16 @@ type GwsHandler struct { postalServerAddress string grpcFactory *generic.GrpcGenericClientFactory sessionStore session.Store // 当前连接用户,内存缓存 - subjectStore cache.MultiLevelCache } func NewGwsHandler(postalServerAddress string, grpcFactory *generic.GrpcGenericClientFactory, sessionStore session.Store, - subjectStore cache.MultiLevelCache) *GwsHandler { +) *GwsHandler { return &GwsHandler{ grpcFactory: grpcFactory, postalServerAddress: postalServerAddress, sessionStore: sessionStore, - subjectStore: subjectStore, } } @@ -56,10 +53,6 @@ func (c *GwsHandler) OnClose(socket *gws.Conn, err error) { } uid := val.(string) c.sessionStore.Delete(uid) - // todo offline event - if err = c.subjectStore.Del(context.Background(), uid); err != nil { - logger.Error("del offline subject store error: ", err) - } logger.Info("subject offline: ", uid) } @@ -179,22 +172,10 @@ func (c *GwsHandler) extractAuthVerify(conn *gwsConn, resp []byte) (sessionUid s } // 设置连接身份信息 sessionUid = authSubject.Uid - gateSubject := &session.GateSubject{ - Uid: authSubject.Uid, - Online: 1, - Time: time.Now().UnixMilli(), - Gate: c.postalServerAddress, - } - // 存储session - err = c.subjectStore.Set(context.Background(), sessionUid, gateSubject) // store to redis cluster cache - if err != nil { - return - } // 关闭旧的链接 oldChannel, ok := c.sessionStore.Load(sessionUid) if ok { - // TODO nats offline / force load subject target cluster call offline oldChannel.Conn.(*gwsConn).conn.WriteClose(1000, nil) logger.Infof("close subject old conn: %s\n", sessionUid) } diff --git a/internal/postal/logic/errors.go b/internal/postal/logic/errors.go new file mode 100644 index 0000000..e1e50f5 --- /dev/null +++ b/internal/postal/logic/errors.go @@ -0,0 +1,8 @@ +package logic + +import "errors" + +var ( + errRedeliverTtlEnd = errors.New("redeliver ttl ended") + errRedeliverOffsetNodeRepeat = errors.New("redeliver offset node repeat") +) diff --git a/internal/postal/logic/postal_cluster_server.go b/internal/postal/logic/postal_cluster_server.go index 7a95611..efebeaa 100644 --- a/internal/postal/logic/postal_cluster_server.go +++ b/internal/postal/logic/postal_cluster_server.go @@ -2,11 +2,11 @@ package logic import ( "context" - "errors" "fmt" "google.golang.org/grpc" "net" "sonet/api/gen/postal" + "sonet/pkg/grpc/client" "sonet/pkg/protocol/deliver" "sonet/pkg/protocol/session" "sonet/pkg/utils/logger" @@ -28,24 +28,23 @@ func PostalAddr2Cluster(postalServerAddr string) (string, error) { return fmt.Sprintf("%s:%d", addr.IP.String(), addr.Port+PostalClusterPortOffset), nil } -var ( - methodRedirectDeliver int32 = 10 - methodRedirectDeliverBatch int32 = 11 - methodRedirectDeliverGroup int32 = 12 -) - // PostalClusterServer 用于postal server集群间, 消息转发通信 type PostalClusterServer struct { postal.UnimplementedPostalClusterServer - postalServer postal.PostalServer // 当前节点 postal server - postalPicker *deliver.PostalPicker - sessionStore session.Store + sessionStore session.Store + postalPicker *deliver.PostalPicker + clientFactory *client.GrpcDirectClientFactory // todo 清理过期连接 } -func NewPostalClusterServer(postalServer postal.PostalServer, postalPicker *deliver.PostalPicker) *PostalClusterServer { +func NewPostalClusterServer( + sessionStore session.Store, + postalPicker *deliver.PostalPicker, + clientFactory *client.GrpcDirectClientFactory, +) *PostalClusterServer { return &PostalClusterServer{ - postalServer: postalServer, - postalPicker: postalPicker, + sessionStore: sessionStore, + postalPicker: postalPicker, + clientFactory: clientFactory, } } @@ -69,58 +68,72 @@ func (s *PostalClusterServer) Run(postalAddr string, opts ...grpc.ServerOption) return } -func (s *PostalClusterServer) Redirect(ctx context.Context, req *postal.ReqRedirect) (*postal.ResDeliver, error) { - req.Ttl -= 1 - switch req.RedirectMethod { - case methodRedirectDeliver: - return s.postalServer.Deliver(ctx, req.Deliver) - case methodRedirectDeliverBatch: - return s.postalServer.DeliverBatch(ctx, req.DeliverBatch) - case methodRedirectDeliverGroup: - return s.postalServer.DeliverGroup(ctx, req.DeliverGroup) - } - return nil, errors.New(fmt.Sprintf("not found redirect type %d", req.RedirectMethod)) -} - -func (s *PostalClusterServer) ReDeliver(ctx context.Context, req *postal.ReqReDeliver) (res *postal.ResDeliver, err error) { - channel, ok := s.sessionStore.Load(req.Receiver) - if ok { - bytes, err := encodeDeliverMessage(req.Msg) - if err != nil { - return +func (s *PostalClusterServer) Redeliver(ctx context.Context, req *postal.ReqRedeliver) (res *postal.ResDeliver, err error) { + bytes, err := encodeDeliverMessage(req.Msg) + if err != nil { + return + } + var redelivers []string + for _, receiver := range req.Receivers { + channel, ok := s.sessionStore.Load(receiver) + if !ok { + redelivers = append(redelivers, receiver) + continue } err = channel.Conn.Write(bytes) if err != nil { - return + logger.Errorf("redeliver channel %s write error: %v", receiver, err) + continue } + } + if len(redelivers) == 0 { res = &postal.ResDeliver{Ok: true} return } + req.Receivers = redelivers // 重新投递 var next int next, req.Offset = convergeOffset(req.Offset) if next == 0 || req.Offset == 0 { // 重投次数结束 + err = errRedeliverTtlEnd return } - node, _, err := s.postalPicker.PickOffsetNode(req.Receiver, next) + + node, _, err := s.postalPicker.PickOffsetNode(redelivers[0], next) if err != nil { return } - addr, err := nets.Address2i64(node.GetAddr()) + addr := node.GetAddr() + addrI64, err := nets.Address2i64(addr) if err != nil { return } for _, nodeI64 := range req.Nodes { - if addr == nodeI64 { + if addrI64 == nodeI64 { // 重复投递节点, 结束投递 + err = errRedeliverOffsetNodeRepeat return } } - req.Nodes = append(req.Nodes, addr) + // 再次投递 + req.Nodes = append(req.Nodes, addrI64) + // postal集群中转发消息 + clusterAddr, err := PostalAddr2Cluster(addr) + if err != nil { + logger.Errorf("parse postal server addr error: %s", addr, err) + return + } + conn, err := s.clientFactory.GetConn(ctx, clusterAddr) + if err != nil { + logger.Error("get postal cluster conn error: ", err) + return nil, err + } + clusterClient := postal.NewPostalClusterClient(conn) + res, err = clusterClient.Redeliver(ctx, req) return } diff --git a/internal/postal/logic/postal_server.go b/internal/postal/logic/postal_server.go index c206aa4..ba0cc77 100644 --- a/internal/postal/logic/postal_server.go +++ b/internal/postal/logic/postal_server.go @@ -10,12 +10,9 @@ import ( "sonet/api/gen/postal" "sonet/internal/postal/group" "sonet/pkg/config" - "sonet/pkg/grpc/client" "sonet/pkg/grpc/discovery" "sonet/pkg/grpc/interceptor" - "sonet/pkg/plugins/cache" "sonet/pkg/plugins/mq" - "sonet/pkg/protocol" "sonet/pkg/protocol/event" "sonet/pkg/protocol/session" "sonet/pkg/utils/collect" @@ -25,29 +22,26 @@ import ( type PostalServer struct { postal.UnimplementedPostalServer - endpointAddress string // websocket 前端连接地址 - broadcastAddress string // postal server 在注册中心注册的地址, 其他服务可直连访问 - sessionStore session.Store // k:uid 在线用户conn存储 - groupStore *collect.ConcurrentMap[string, *group.Group] // k:gid 在线用户 groups 存储 - subjectStore cache.MultiLevelCache // online subject address cache, ws gateway集群所有在线用户存储 - clientFactory *client.GrpcDirectClientFactory - producer mq.Producer + endpointAddress string // websocket 前端连接地址 + broadcastAddress string // postal server 在注册中心注册的地址, 其他服务可直连访问 + sessionStore session.Store // k:uid 在线用户conn存储 + groupStore *collect.ConcurrentMap[string, *group.Group] // k:gid 在线用户 groups 存储 + postalClusterServer *PostalClusterServer + producer mq.Producer } func NewPostalServer( endpointAddress string, sessionStore session.Store, groupStore *collect.ConcurrentMap[string, *group.Group], - subjectStore cache.MultiLevelCache, - clientFactory *client.GrpcDirectClientFactory, + postalClusterServer *PostalClusterServer, producer mq.Producer) *PostalServer { return &PostalServer{ - endpointAddress: endpointAddress, - sessionStore: sessionStore, - groupStore: groupStore, - subjectStore: subjectStore, - clientFactory: clientFactory, - producer: producer, + endpointAddress: endpointAddress, + sessionStore: sessionStore, + groupStore: groupStore, + postalClusterServer: postalClusterServer, + producer: producer, } } @@ -134,55 +128,6 @@ func (s *PostalServer) processSessionStoreEvent(ctx context.Context) { } } -func (s *PostalServer) deliverMessage(msg *postal.Message, conn session.NetConn) error { - header := &protocol.Header{ - Magic: protocol.Magic, - Type: protocol.TypeNotice, - UrlType: 1, - SerializeType: 1, - Svc: msg.Svc, - Target: msg.Msg, - } - payload := &protocol.Payload{Header: header, Body: msg.Body} - bytes, err := protocol.EncodeSo(payload) - if err != nil { - logger.Error("req deliver encode notice error: ", err) - return err - } - return conn.Write(bytes) -} - -// receiverGates 找receiver在集群内哪些其他节点 -func (s *PostalServer) clusterGateReceivers(ctx context.Context, receivers []string) (gateReceivers map[string][]string, offline []string) { - gateReceivers = make(map[string][]string) - for _, receiver := range receivers { - subject := &session.GateSubject{} - err := s.subjectStore.Load(ctx, receiver, subject) - if err != nil { - offline = append(offline, receiver) - if err != cache.NotExists { - logger.Error("load from subject store error: ", err) - offline = append(offline, receiver) - } - continue - } - - if subject.Gate == s.broadcastAddress { - offline = append(offline, receiver) - // 清除失效缓存 - err := s.subjectStore.Del(ctx, receiver) - if err != nil { - logger.Error("del subject store error: ", receiver, err) - } - continue - } - - // put receiver gate addr - gateReceivers[subject.Gate] = append(gateReceivers[subject.Gate], receiver) - } - return -} - func (s *PostalServer) Deliver(ctx context.Context, req *postal.ReqDeliver) (res *postal.ResDeliver, err error) { receiver, ok := s.sessionStore.Load(req.Receiver) if ok { @@ -199,40 +144,20 @@ func (s *PostalServer) Deliver(ctx context.Context, req *postal.ReqDeliver) (res res = &postal.ResDeliver{Ok: true} return } - // todo 向一致性 hash 下一个节点传递 - // todo picker reqCluster offset +-3 - - // 用户连接不在当前gateway - gateReceivers, offline := s.clusterGateReceivers(ctx, []string{req.Receiver}) - if offline != nil && len(offline) > 0 { - res = &postal.ResDeliver{Code: int32(postal.DeliverResult_ReceiverOffline.Number())} - return - } - - for postalAddr := range gateReceivers { - // gateway集群中转发消息 - clusterAddr, err := PostalAddr2Cluster(postalAddr) - if err != nil { - logger.Errorf("parse postal server addr error: %s", postalAddr, err) - continue - } - conn, err := s.clientFactory.GetConn(context.Background(), clusterAddr) - if err != nil { - logger.Error("get postal cluster conn error: ", err) - return nil, err - } - clusterClient := postal.NewPostalClusterClient(conn) - - reqRedirect := &postal.ReqRedirect{ - Ttl: 3, // TODO 转发n次就丢弃 - RedirectMethod: methodRedirectDeliver, - Deliver: req, + // 向一致性 hash 下一个节点传递 + // picker reqCluster offset +3: + for _, offset := range []int32{3, -3} { + res, err = s.postalClusterServer.Redeliver(ctx, &postal.ReqRedeliver{ + Receivers: []string{req.Receiver}, + Msg: req.Msg, + Offset: offset, + }) + if err == nil { + res = &postal.ResDeliver{Ok: true} + return } - resp, err := clusterClient.Redirect(ctx, reqRedirect) - return resp, err } - - return nil, nil + return } func (s *PostalServer) DeliverBatch(ctx context.Context, req *postal.ReqDeliverBatch) (res *postal.ResDeliver, err error) { @@ -241,11 +166,11 @@ func (s *PostalServer) DeliverBatch(ctx context.Context, req *postal.ReqDeliverB return } - var redirectReceivers []string + var redeliverReceivers []string for _, receiverId := range req.Receivers { receiver, ok := s.sessionStore.Load(receiverId) if !ok { - redirectReceivers = append(redirectReceivers, receiverId) + redeliverReceivers = append(redeliverReceivers, receiverId) continue } err = receiver.Conn.Write(bytes) @@ -254,44 +179,23 @@ func (s *PostalServer) DeliverBatch(ctx context.Context, req *postal.ReqDeliverB } } - if len(redirectReceivers) == 0 { + if len(redeliverReceivers) == 0 { return &postal.ResDeliver{Ok: true}, nil } - // todo 向一致性 hash 下个节点传递 - gateReceivers, offline := s.clusterGateReceivers(ctx, redirectReceivers) - if len(offline) > 0 { - logger.Warning("offline redirect receivers: ", offline) - } - - if len(gateReceivers) > 0 { - for postalAddr, receivers := range gateReceivers { - // gateway集群中转发消息 - clusterAddr, err := PostalAddr2Cluster(postalAddr) - if err != nil { - logger.Errorf("parse postal server addr error: %s", postalAddr, err) - continue - } - conn, err := s.clientFactory.GetConn(ctx, clusterAddr) - if err != nil { - logger.Error("get postal cluster conn error: ", err) - continue - } - clusterClient := postal.NewPostalClusterClient(conn) - - req.Receivers = receivers - reqRedirect := &postal.ReqRedirect{ - Ttl: 3, // TODO 转发n次就丢弃 - RedirectMethod: methodRedirectDeliverBatch, - DeliverBatch: req, - } - _, err = clusterClient.Redirect(ctx, reqRedirect) - if err != nil { - logger.Error("redirect batch error: ", err) - } + // 向一致性 hash 下一个节点传递 + for _, offset := range []int32{3, -3} { + res, err = s.postalClusterServer.Redeliver(ctx, &postal.ReqRedeliver{ + Receivers: redeliverReceivers, + Msg: req.Msg, + Offset: offset, + }) + if err == nil { + res = &postal.ResDeliver{Ok: true} + return } } - return &postal.ResDeliver{Ok: true}, nil + return } // DeliverGroup postal broadcast