diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..a81f46b --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +# alpine need CGO_ENABLED=0, sqlite need CGO_ENABLED=1 ... +# FROM alpine:3 +FROM ubuntu:22.10 +ARG APP +ENV LANG C.UTF-8 + +WORKDIR /opt + +COPY target/${APP}/${APP} /opt/go-app +COPY target/${APP}/config/ /opt/config/ + +RUN chmod +x /opt/go-app \ + && ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime \ + && echo 'Asia/Shanghai' > /etc/timezone + +CMD ["./go-app", "-conf=config/config.toml"] diff --git a/benchmark/main.go b/benchmark/main.go new file mode 100644 index 0000000..4fea675 --- /dev/null +++ b/benchmark/main.go @@ -0,0 +1,334 @@ +package main + +import ( + "context" + "encoding/base64" + "fmt" + "github.com/bytedance/sonic" + "github.com/gorilla/websocket" + "google.golang.org/protobuf/proto" + "math/rand" + "net" + "runtime" + "sonet/api/gen/auth" + "sonet/api/gen/chat" + "sonet/pkg/config" + "sonet/pkg/protocol" + "sonet/pkg/utils/logger" + "sonet/pkg/utils/security" + "sonet/pkg/utils/shutdown" + "strconv" + "sync" + "sync/atomic" + "time" +) + +var ( + mockUsers = 100 + eachUserSend = 100000 + mockNetUsers []*NetUser + sendCounter int64 = 0 + receiverCounter int64 = 0 + wsUrls = []string{"ws://192.168.110.36:7001", "ws://192.168.110.36:7003"} + seqId int32 + callbacks map[int32]func(res any) + callbackMutex *sync.Mutex + useMsSum int64 + useMsAvg int64 +) + +func init() { + config.InitLogger() + + callbacks = make(map[int32]func(res any), 128) + callbackMutex = &sync.Mutex{} +} + +func records(ctx context.Context) { + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + + var prevSendCounter, prevReceiverCounter int64 + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + // fmt.Printf("record: %d/%d, total: %d/%d \n", receiverCounter-prevReceiverCounter, sendCounter-prevSendCounter, receiverCounter, sendCounter) + logger.Infof("record: %d/%d, total: %d/%d, avg %dms\n", receiverCounter-prevReceiverCounter, sendCounter-prevSendCounter, receiverCounter, sendCounter, useMsAvg) + prevReceiverCounter = receiverCounter + prevSendCounter = sendCounter + } + } +} + +type NetUser struct { + Uid string + Conn *websocket.Conn +} + +func main() { + runtime.GOMAXPROCS(runtime.NumCPU()) + // go prof.StartPprof(":8888") + + ctx, cancel := context.WithCancel(context.Background()) + go records(ctx) + + mockNetUsers = make([]*NetUser, mockUsers) + // initial uids + for i := 0; i < mockUsers; i++ { + uid := strconv.Itoa(110000 + i) + conn, err := getConn() + if err != nil { + panic(err) + } + err = handleConn(ctx, uid, conn) + if err != nil { + panic(err) + } + mockNetUsers[i] = &NetUser{ + Uid: uid, + Conn: conn, + } + } + + for i := 0; i < mockUsers; i++ { + netUser := mockNetUsers[i] + go sendBatchChatMessage(ctx, netUser.Uid, netUser.Conn, eachUserSend) + } + + shutdown.Await() + cancel() + + logger.Infof("total receiver:%d, send:%d\n", receiverCounter, sendCounter) +} + +func sendBatchChatMessage(ctx context.Context, uid string, conn *websocket.Conn, count int) { + r := rand.New(rand.NewSource(time.Now().UnixMilli())) + for i := 0; i < count; i++ { + receiverUid := mockNetUsers[r.Intn(mockUsers)].Uid + args := &chat.ReqSend{ + Receiver: receiverUid, + Content: "hello", + } + channel, err := send(conn, chat.Chat_ServiceDesc.ServiceName, "send", args) + if err != nil { + fmt.Println("send failed:", err) + return + } + + res := <-channel // wait result + err, failed := res.(error) + if failed { + fmt.Println("send failed:", err) + } else { + // fmt.Println("send ok:", res) + } + } +} + +func getConn() (conn *websocket.Conn, err error) { + wsUrl := wsUrls[rand.Intn(len(wsUrls))] + conn, _, err = websocket.DefaultDialer.Dial(wsUrl+"/ws", nil) + if err != nil { + return + } + return +} + +func getToken(uid string) (token string, err error) { + // generate token + subject := &auth.Subject{Uid: uid, Username: uid, Time: time.Now().UnixMilli()} + bytes, err := sonic.Marshal(subject) + if err != nil { + return + } + keyBytes, err := base64.StdEncoding.DecodeString("9Nz3Y6DES3msAFndz4QsJAECUIFkf+KKaRa+jRnNALk=") + if err != nil { + return + } + encode, err := security.EncryptAesCBC(bytes, keyBytes) + if err != nil { + return + } + token = base64.URLEncoding.EncodeToString(encode) + return +} + +// handleConn listen and auth verify +func handleConn(ctx context.Context, uid string, conn *websocket.Conn) (err error) { + go listen(ctx, conn) + + // handshake + token, err := getToken(uid) + if err != nil { + return + } + args := &auth.ReqVerify{Token: token} + + channel, err := send(conn, auth.Auth_ServiceDesc.ServiceName, "Verify", args) + if err != nil { + return + } + res := <-channel + // fmt.Printf("%v\n", res) + if e, failed := res.(error); failed { + err = e + } + return +} + +func listen(ctx context.Context, conn *websocket.Conn) { + var e error + defer func() { + if e != nil { + fmt.Println("conn error: ", e) + } + }() + + go func() { + select { + case <-ctx.Done(): // finished close connection + _ = conn.Close() + } + }() + + for { + t, message, err := conn.ReadMessage() + if err != nil { + if opErr, ok := err.(*net.OpError); ok && opErr.Err.Error() == "use of closed network connection" { + return + } + + e = err + _ = conn.Close() + break + } + if t != websocket.BinaryMessage { + e = fmt.Errorf("unknown message type: %d\n", t) + continue + } + + payload, err := protocol.DecodeSo(message) + if err != nil { + logger.Error("decode message error: ", err) + continue + } + header := payload.Header + + if header.Type == 4 { + callbackMutex.Lock() + callback, ok := callbacks[header.SeqId] + if ok { + delete(callbacks, header.SeqId) + } + callbackMutex.Unlock() + if !ok { + e = fmt.Errorf("callback %d not found", header.SeqId) + return + } + callback(err) + continue + } + + // notice message + if header.Type == protocol.TypeNotice { + switch header.Target { + case "ChatMessage": + msg := &chat.ChatMessage{} + if err := proto.Unmarshal(payload.Body, msg); err != nil { + logger.Error("unmarshal proto message error: ", err) + return + } + //atomic.AddInt64(&receiverCounter, 1) + // fmt.Printf("reciever: %v\n", msg) + } + continue + } + + // rpc response + var msg proto.Message + switch header.Svc { + case auth.Auth_ServiceDesc.ServiceName: + switch header.Target { + case "Subject": + // fmt.Println("res verify...") + msg = &auth.Subject{} + } + case chat.Chat_ServiceDesc.ServiceName: + switch header.Target { + case "ResSend": + // fmt.Println("res send...") + msg = &chat.ResSend{} + } + } + if msg == nil { + logger.Warning("unknown rpc response: ", len(payload.Body), header) + continue + } + err = proto.Unmarshal(payload.Body, msg) + if e != nil { + logger.Error("unmarshal rcp res body error: ", payload, err) + return + } + + callbackMutex.Lock() + callback, ok := callbacks[header.SeqId] + if ok { + delete(callbacks, header.SeqId) + } + callbackMutex.Unlock() + if !ok { + e = fmt.Errorf("callback %d not found", header.SeqId) + return + } + callback(msg) + } +} + +func send(conn *websocket.Conn, svc, method string, reqArgs proto.Message) (channel chan any, err error) { + seq := atomic.AddInt32(&seqId, 1) + header := &protocol.Header{ + Magic: protocol.Magic, + Type: 1, + UrlType: 1, + SerializeType: 1, + SeqId: seq, + Svc: svc, + Target: method, + } + payload := &protocol.Payload{Header: header} + payload.Body, err = proto.Marshal(reqArgs) + if err != nil { + panic(err) + } + bytes, err := protocol.EncodeSo(payload) + if err != nil { + panic(err) + } + + begin := time.Now().UnixMilli() + // seq, await + channel = make(chan any) + + // ready callback + callbackMutex.Lock() + callbacks[seq] = func(res any) { + ms := time.Now().UnixMilli() - begin + msSum := atomic.AddInt64(&useMsSum, ms) + counter := atomic.AddInt64(&receiverCounter, 1) + atomic.StoreInt64(&useMsAvg, msSum/counter) + + channel <- res + } + callbackMutex.Unlock() + + // send message + err = conn.WriteMessage(websocket.BinaryMessage, bytes) + if err != nil { + return + } + atomic.AddInt64(&sendCounter, 1) + return +} diff --git a/benchmark/prof/pprof.go b/benchmark/prof/pprof.go new file mode 100644 index 0000000..cad510c --- /dev/null +++ b/benchmark/prof/pprof.go @@ -0,0 +1,14 @@ +package prof + +import ( + "log" + "net/http" + _ "net/http/pprof" +) + +func StartPprof(addr string) { + err := http.ListenAndServe(addr, nil) + if err != nil { + log.Fatal("http server start failed: ", err) + } +} diff --git a/build.bat b/build.bat new file mode 100644 index 0000000..09a8a3a --- /dev/null +++ b/build.bat @@ -0,0 +1,24 @@ +@echo off +@title build so app + +@REM build.bat gateway_ws 1.0.0 + +set APP=%1 +set DOCKER_IMAGE_VERSION=%2 +set GO111MODULE=on +set CGO_ENABLED=0 +set GOOS=linux +set GOARCH=amd64 +go build -o target/%APP%/%APP% cmd/%APP%/main.go +if exist cmd\%APP%\config.toml ( + if not exist target\%APP%\config ( + md target\%APP%\config + ) + copy cmd\%APP%\config.toml target\%APP%\config\ +) + +if not "%DOCKER_IMAGE_VERSION%"=="" ( + docker build -t so-%APP%:%DOCKER_IMAGE_VERSION% --build-arg APP=%APP% . +) + +echo %APP% build finished diff --git a/deploy_k8s/docker_compose.yml b/deploy_k8s/docker_compose.yml new file mode 100644 index 0000000..4388556 --- /dev/null +++ b/deploy_k8s/docker_compose.yml @@ -0,0 +1,28 @@ +version: '3' +services: + svr-gateway-ws-1: + image: so_gateway_ws:1.0.0 + network_mode: host + environment: + - SO_APP.HTTPPORT=7001 + - SO_APP.ENDPOINTADDRESS=192.168.110.36:7001 + - SO_GRPC.ADDRESS=:7011 + svr-gateway-ws-2: + image: so_gateway_ws:1.0.0 + network_mode: host + environment: + - SO_APP.HTTPPORT=7003 + - SO_APP.ENDPOINTADDRESS=192.168.110.36:7003 + - SO_GRPC.ADDRESS=:7012 + + svr-chat: + image: so_chat:1.0.0 + network_mode: host + + svr-auth: + image: so_auth:1.0.0 + network_mode: host + + svr-gatewat-http: + image: so_gateway_http:1.0.0 + network_mode: host diff --git a/pkg/config/loader.go b/pkg/config/loader.go index d972334..06112bb 100644 --- a/pkg/config/loader.go +++ b/pkg/config/loader.go @@ -1,6 +1,8 @@ package config import ( + "bytes" + "encoding/json" "errors" "flag" "github.com/spf13/viper" @@ -23,7 +25,7 @@ func parseConfPathFlag(confPath string) (filePath, fileName, confName, confType // appConf service custom config // return common service configuration func LoadConfig(appConf any, confPathArg ...string) *Configuration { - initLogger() + InitLogger() confPath := "" confName := "config" @@ -64,12 +66,23 @@ func LoadConfig(appConf any, confPathArg ...string) *Configuration { if err := viper.Unmarshal(conf); err != nil { panic(errors.New("viper unmarshal config failed: " + err.Error())) } + + // 读取appConf if appConf != nil && len(conf.App) > 0 { - if err := viper.UnmarshalKey("app", appConf); err != nil { - panic(errors.New("viper unmarshal app config failed: " + err.Error())) + app, err := json.Marshal(conf.App) + if err != nil { + panic(err) + } + // 不能 json.Unmarshal 类型转换问题, viper.UnmarshalKey 无法读取环境变量 + viper.Reset() + viper.SetConfigType("json") + if err = viper.ReadConfig(bytes.NewBuffer(app)); err != nil { + panic(err) + } + if err = viper.Unmarshal(appConf); err != nil { + panic(err) } } - return conf } diff --git a/pkg/config/logger.go b/pkg/config/logger.go index a4d38e2..9d9e437 100644 --- a/pkg/config/logger.go +++ b/pkg/config/logger.go @@ -6,7 +6,7 @@ import ( "sonet/pkg/utils/logger" ) -func initLogger() { +func InitLogger() { logrus.SetFormatter(&logrus.TextFormatter{ ForceColors: true, TimestampFormat: "2006-01-02 15:04:05", //时间格式 diff --git a/pkg/protocol/protocol.go b/pkg/protocol/protocol.go index 13fdf1f..f5cd7ea 100644 --- a/pkg/protocol/protocol.go +++ b/pkg/protocol/protocol.go @@ -28,7 +28,7 @@ type Header struct { Magic byte Type byte // 1 request, 2 response, 3 event, 4 error Status byte - UrlType byte // 4bit: svc,method url desc: 1 name, 2 number + UrlType byte // 4bit: svc,method url: 1 name, 2 number SerializeType byte // 4bit: body serialize: 1proto, 2json SeqId int32 SvcNo int32 // optional 1 @@ -44,6 +44,7 @@ type Payload struct { func DecodeSo(bytes []byte) (payload *Payload, err error) { header := &Header{} + payload = &Payload{Header: header} header.Magic = bytes[0] if header.Magic != Magic { @@ -80,7 +81,6 @@ func DecodeSo(bytes []byte) (payload *Payload, err error) { return } - payload = &Payload{Header: header} payload.Body = bytes[cursor:] return } diff --git a/pkg/utils/shutdown/signal.go b/pkg/utils/shutdown/signal.go index b8a0630..3e75cfa 100644 --- a/pkg/utils/shutdown/signal.go +++ b/pkg/utils/shutdown/signal.go @@ -45,7 +45,7 @@ func Await() { }() } - logger.Info("execute %d shutdown hook %d ok, %d failed\n", success+fail, success, fail) + logger.Infof("execute %d shutdown hook %d ok, %d failed\n", success+fail, success, fail) } //func Shutdown() {