diff --git a/cmd/openim-api/main.go b/cmd/openim-api/main.go index f0b62e31f..59e0b7f9e 100644 --- a/cmd/openim-api/main.go +++ b/cmd/openim-api/main.go @@ -17,6 +17,7 @@ package main import ( "context" "fmt" + "github.com/OpenIMSDK/tools/errs" "net" "net/http" _ "net/http/pprof" @@ -28,8 +29,6 @@ import ( "github.com/OpenIMSDK/protocol/constant" "github.com/OpenIMSDK/tools/discoveryregistry" - "github.com/OpenIMSDK/tools/log" - "github.com/openimsdk/open-im-server/v3/internal/api" "github.com/openimsdk/open-im-server/v3/pkg/common/cmd" "github.com/openimsdk/open-im-server/v3/pkg/common/config" @@ -44,55 +43,42 @@ func main() { apiCmd.AddPortFlag() apiCmd.AddApi(run) if err := apiCmd.Execute(); err != nil { - log.ZError(context.Background(), "API command execution failed", err) panic(err.Error()) } } func run(port int, proPort int) error { - log.ZInfo(context.Background(), "Openim api port:", "port", port, "proPort", proPort) - if port == 0 || proPort == 0 { err := "port or proPort is empty:" + strconv.Itoa(port) + "," + strconv.Itoa(proPort) - log.ZError(context.Background(), err, nil) - return fmt.Errorf(err) + return errs.Wrap(fmt.Errorf(err)) } - rdb, err := cache.NewRedis() if err != nil { - log.ZError(context.Background(), "Failed to initialize Redis", err) return err } - log.ZInfo(context.Background(), "api start init discov client") var client discoveryregistry.SvcDiscoveryRegistry // Determine whether zk is passed according to whether it is a clustered deployment client, err = kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery) if err != nil { - log.ZError(context.Background(), "Failed to initialize discovery register", err) - return err + return errs.Wrap(err, "register discovery err") } if err = client.CreateRpcRootNodes(config.Config.GetServiceNames()); err != nil { - log.ZError(context.Background(), "Failed to create RPC root nodes", err) - return err + return errs.Wrap(err, "create rpc root nodes error") } - log.ZInfo(context.Background(), "api register public config to discov") if err = client.RegisterConf2Registry(constant.OpenIMCommonConfigKey, config.Config.EncodeConfig()); err != nil { - log.ZError(context.Background(), "Failed to register public config to discov", err) return err } - log.ZInfo(context.Background(), "api register public config to discov success") router := api.NewGinRouter(client, rdb) if config.Config.Prometheus.Enable { p := ginprom.NewPrometheus("app", prommetrics.GetGinCusMetrics("Api")) p.SetListenAddress(fmt.Sprintf(":%d", proPort)) p.Use(router) } - log.ZInfo(context.Background(), "api init router success") var address string if config.Config.Api.ListenIP != "" { @@ -100,13 +86,11 @@ func run(port int, proPort int) error { } else { address = net.JoinHostPort("0.0.0.0", strconv.Itoa(port)) } - log.ZInfo(context.Background(), "start api server", "address", address, "OpenIM version", config.Version) server := http.Server{Addr: address, Handler: router} go func() { err = server.ListenAndServe() if err != nil && err != http.ErrServerClosed { - log.ZError(context.Background(), "api run failed", err, "address", address) os.Exit(1) } }() @@ -120,7 +104,6 @@ func run(port int, proPort int) error { // graceful shutdown operation. if err := server.Shutdown(ctx); err != nil { - log.ZError(context.Background(), "failed to api-server shutdown", err) return err } diff --git a/docs/contrib/environment.md b/docs/contrib/environment.md index e18246afa..dc11a3c7b 100644 --- a/docs/contrib/environment.md +++ b/docs/contrib/environment.md @@ -305,6 +305,7 @@ Feel free to explore the MinIO documentation for more advanced configurations an This section involves setting up MongoDB, including its port, address, and credentials. + | Parameter | Example Value | Description | | -------------- | -------------- | ----------------------- | | MONGO_PORT | "27017" | Port used by MongoDB. | diff --git a/go.mod b/go.mod index c709e9ba4..a5fecb5ee 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,7 @@ go 1.19 require ( firebase.google.com/go v3.13.0+incompatible github.com/OpenIMSDK/protocol v0.0.48 - github.com/OpenIMSDK/tools v0.0.29 + github.com/OpenIMSDK/tools v0.0.31 github.com/bwmarrin/snowflake v0.3.0 // indirect github.com/dtm-labs/rockscache v0.1.1 github.com/gin-gonic/gin v1.9.1 @@ -31,7 +31,7 @@ require ( gopkg.in/yaml.v3 v3.0.1 ) -require github.com/google/uuid v1.3.1 +require github.com/google/uuid v1.5.0 require ( github.com/IBM/sarama v1.41.3 @@ -94,8 +94,8 @@ require ( github.com/jinzhu/now v1.1.5 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kevinburke/ssh_config v0.0.0-20190725054713-01f96b0aa0cd // indirect - github.com/klauspost/compress v1.16.7 // indirect - github.com/klauspost/cpuid/v2 v2.2.5 // indirect + github.com/klauspost/compress v1.17.4 // indirect + github.com/klauspost/cpuid/v2 v2.2.6 // indirect github.com/leodido/go-urn v1.2.4 // indirect github.com/lithammer/shortuuid v3.0.0+incompatible // indirect github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect @@ -129,7 +129,7 @@ require ( go.uber.org/atomic v1.7.0 // indirect go.uber.org/multierr v1.11.0 // indirect golang.org/x/arch v0.3.0 // indirect - golang.org/x/net v0.17.0 // indirect + golang.org/x/net v0.19.0 // indirect golang.org/x/oauth2 v0.13.0 // indirect golang.org/x/sys v0.15.0 // indirect golang.org/x/text v0.14.0 // indirect @@ -141,7 +141,7 @@ require ( google.golang.org/genproto/googleapis/rpc v0.0.0-20231012201019-e917dd12ba7a // indirect gopkg.in/src-d/go-billy.v4 v4.3.2 // indirect gopkg.in/warnings.v0 v0.1.2 // indirect - gorm.io/gorm v1.23.8 // indirect + gorm.io/gorm v1.25.4 // indirect stathat.com/c/consistent v1.0.0 // indirect ) diff --git a/go.sum b/go.sum index b7b40632f..fbd9366c3 100644 --- a/go.sum +++ b/go.sum @@ -20,8 +20,8 @@ github.com/IBM/sarama v1.41.3 h1:MWBEJ12vHC8coMjdEXFq/6ftO6DUZnQlFYcxtOJFa7c= github.com/IBM/sarama v1.41.3/go.mod h1:Xxho9HkHd4K/MDUo/T/sOqwtX/17D33++E9Wib6hUdQ= github.com/OpenIMSDK/protocol v0.0.48 h1:8MIMjyzJRsruYhVv2ZKArFiOveroaofDOb3dlAdgjsw= github.com/OpenIMSDK/protocol v0.0.48/go.mod h1:F25dFrwrIx3lkNoiuf6FkCfxuwf8L4Z8UIsdTHP/r0Y= -github.com/OpenIMSDK/tools v0.0.29 h1:NS4PEwYl9sX3SWsMjDOLVxMo3LcTWREMr+2cjzWjcqc= -github.com/OpenIMSDK/tools v0.0.29/go.mod h1:eg+q4A34Qmu73xkY0mt37FHGMCMfC6CtmOnm0kFEGFI= +github.com/OpenIMSDK/tools v0.0.31 h1:fSrhcPTvHEMTSyrJZDupe730mL4nuhvSOUP/BaZiHaY= +github.com/OpenIMSDK/tools v0.0.31/go.mod h1:wBfR5CYmEyvxl03QJbTkhz1CluK6J4/lX0lviu8JAjE= github.com/QcloudApi/qcloud_sign_golang v0.0.0-20141224014652-e4130a326409/go.mod h1:1pk82RBxDY/JZnPQrtqHlUFfCctgdorsd9M06fMynOM= github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7 h1:uSoVVbwJiQipAclBbw+8quDsfcvFjOpI5iCf4p/cqCs= github.com/alcortesm/tgz v0.0.0-20161220082320-9c5fe88206d7/go.mod h1:6zEj6s6u/ghQa61ZWa/C2Aw3RkjiTBOix7dkqa1VLIs= @@ -152,8 +152,8 @@ github.com/google/s2a-go v0.1.7 h1:60BLSyTrOV4/haCDW4zb1guZItoSq8foHCXrAnjBo/o= github.com/google/s2a-go v0.1.7/go.mod h1:50CgR4k1jNlWBu4UfS4AcfhVe1r6pdZPygJ3R8F0Qdw= github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4= -github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU= +github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/googleapis/enterprise-certificate-proxy v0.3.1 h1:SBWmZhjUDRorQxrN0nwzf+AHBxnbFjViHQS4P0yVpmQ= github.com/googleapis/enterprise-certificate-proxy v0.3.1/go.mod h1:VLSiSSBs/ksPL8kq3OBOQ6WRI2QnaFynd1DCjZ62+V0= github.com/googleapis/gax-go/v2 v2.12.0 h1:A+gCJKdRfqXkr+BIRGtZLibNXf0m1f9E4HG56etFpas= @@ -194,7 +194,6 @@ github.com/jinzhu/copier v0.3.5 h1:GlvfUwHk62RokgqVNvYsku0TATCF7bAHVwEXoBh3iJg= github.com/jinzhu/copier v0.3.5/go.mod h1:DfbEm0FYsaqBcKcFuvmOZb218JkPGtvSHsKg8S8hyyg= github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= -github.com/jinzhu/now v1.1.4/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= github.com/jonboulle/clockwork v0.4.0 h1:p4Cf1aMWXnXAUh8lVfewRBx1zaTSYKrKMF2g3ST4RZ4= @@ -205,12 +204,12 @@ github.com/kevinburke/ssh_config v0.0.0-20190725054713-01f96b0aa0cd/go.mod h1:CT github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= -github.com/klauspost/compress v1.16.7 h1:2mk3MPGNzKyxErAw8YaohYh69+pa4sIQSC0fPGCFR9I= -github.com/klauspost/compress v1.16.7/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE= +github.com/klauspost/compress v1.17.4 h1:Ej5ixsIri7BrIjBkRZLTo6ghwrEtHFk7ijlczPW4fZ4= +github.com/klauspost/compress v1.17.4/go.mod h1:/dCuZOvVtNoHsyb+cuJD3itjs3NbnF6KH9zAO4BDxPM= github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg= -github.com/klauspost/cpuid/v2 v2.2.5 h1:0E5MSMDEoAulmXNFquVs//DdoomxaoTY1kUhbc/qbZg= -github.com/klauspost/cpuid/v2 v2.2.5/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= +github.com/klauspost/cpuid/v2 v2.2.6 h1:ndNyv040zDGIDh8thGkXYjnFtiN02M1PVVF+JE/48xc= +github.com/klauspost/cpuid/v2 v2.2.6/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= @@ -402,8 +401,8 @@ golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qx golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs= -golang.org/x/net v0.17.0 h1:pVaXccu2ozPjCXewfr1S7xza/zcXTity9cCdXQYSjIM= -golang.org/x/net v0.17.0/go.mod h1:NxSsAGuq816PNPmqtQdLE42eU2Fs7NoRIZrHJAlaCOE= +golang.org/x/net v0.19.0 h1:zTwKpTd2XuCqf8huc7Fo2iSy+4RHPd10s4KzeTnVr1c= +golang.org/x/net v0.19.0/go.mod h1:CfAk/cbD4CthTvqiEl8NpboMuiuOYsAr/7NOjZJtv1U= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.13.0 h1:jDDenyj+WgFtmV3zYVoi8aE2BwtXFLWOA67ZfNWftiY= golang.org/x/oauth2 v0.13.0/go.mod h1:/JMhi4ZRXAf4HG9LiNmxvk+45+96RUlVThiH8FzNBn0= @@ -531,8 +530,8 @@ gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -gorm.io/gorm v1.23.8 h1:h8sGJ+biDgBA1AD1Ha9gFCx7h8npU7AsLdlkX0n2TpE= -gorm.io/gorm v1.23.8/go.mod h1:l2lP/RyAtc1ynaTjFksBde/O8v9oOGIApu2/xRitmZk= +gorm.io/gorm v1.25.4 h1:iyNd8fNAe8W9dvtlgeRI5zSVZPsq3OpcTu37cYcpCmw= +gorm.io/gorm v1.25.4/go.mod h1:L4uxeKpfBml98NYqVqwAdmV1a2nBtAec/cf3fpucW/k= gotest.tools v2.2.0+incompatible h1:VsBPFP1AI068pPrMxtb/S8Zkgf9xEmTLJjfM+P5UIEo= gotest.tools v2.2.0+incompatible/go.mod h1:DsYFclhRJ6vuDpmuTbkuFWG+y2sxOXAzmJt81HFBacw= honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4= diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go index 7d692662d..969761a9d 100644 --- a/internal/msgtransfer/init.go +++ b/internal/msgtransfer/init.go @@ -15,8 +15,10 @@ package msgtransfer import ( + "context" "errors" "fmt" + "github.com/OpenIMSDK/tools/errs" "log" "net/http" "sync" @@ -69,40 +71,47 @@ func StartTransfer(prometheusPort int) error { client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin"))) msgModel := cache.NewMsgCacheModel(rdb) msgDocModel := unrelation.NewMsgMongoDriver(mongo.GetDatabase()) - msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, msgModel) + msgDatabase, err := controller.NewCommonMsgDatabase(msgDocModel, msgModel) + if err != nil { + return err + } conversationRpcClient := rpcclient.NewConversationRpcClient(client) groupRpcClient := rpcclient.NewGroupRpcClient(client) - msgTransfer := NewMsgTransfer(msgDatabase, &conversationRpcClient, &groupRpcClient) + msgTransfer, err := NewMsgTransfer(msgDatabase, &conversationRpcClient, &groupRpcClient) + if err != nil { + return err + } return msgTransfer.Start(prometheusPort) } -func NewMsgTransfer(msgDatabase controller.CommonMsgDatabase, conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient) *MsgTransfer { - return &MsgTransfer{ - historyCH: NewOnlineHistoryRedisConsumerHandler(msgDatabase, conversationRpcClient, groupRpcClient), - historyMongoCH: NewOnlineHistoryMongoConsumerHandler(msgDatabase), +func NewMsgTransfer(msgDatabase controller.CommonMsgDatabase, conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient) (*MsgTransfer, error) { + historyCH, err := NewOnlineHistoryRedisConsumerHandler(msgDatabase, conversationRpcClient, groupRpcClient) + if err != nil { + return nil, err } + historyMongoCH, err := NewOnlineHistoryMongoConsumerHandler(msgDatabase) + if err != nil { + return nil, err + } + + return &MsgTransfer{ + historyCH: historyCH, + historyMongoCH: historyMongoCH, + }, nil } func (m *MsgTransfer) Start(prometheusPort int) error { + ctx := context.Background() var wg sync.WaitGroup wg.Add(1) fmt.Println("start msg transfer", "prometheusPort:", prometheusPort) if prometheusPort <= 0 { - return errors.New("prometheusPort not correct") - } - if config.Config.ChatPersistenceMysql { - // go m.persistentCH.persistentConsumerGroup.RegisterHandleAndConsumer(m.persistentCH) - } else { - fmt.Println("msg transfer not start mysql consumer") + return errs.Wrap(errors.New("prometheusPort not correct")) } - go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(m.historyCH) - go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(m.historyMongoCH) - // go m.modifyCH.modifyMsgConsumerGroup.RegisterHandleAndConsumer(m.modifyCH) - /*err := prome.StartPrometheusSrv(prometheusPort) - if err != nil { - return err - }*/ - //////////////////////////// + + go m.historyCH.historyConsumerGroup.RegisterHandleAndConsumer(ctx, m.historyCH) + go m.historyMongoCH.historyConsumerGroup.RegisterHandleAndConsumer(ctx, m.historyMongoCH) + if config.Config.Prometheus.Enable { reg := prometheus.NewRegistry() reg.MustRegister( diff --git a/internal/msgtransfer/online_history_msg_handler.go b/internal/msgtransfer/online_history_msg_handler.go index 127cede71..35af330c9 100644 --- a/internal/msgtransfer/online_history_msg_handler.go +++ b/internal/msgtransfer/online_history_msg_handler.go @@ -87,7 +87,7 @@ func NewOnlineHistoryRedisConsumerHandler( database controller.CommonMsgDatabase, conversationRpcClient *rpcclient.ConversationRpcClient, groupRpcClient *rpcclient.GroupRpcClient, -) *OnlineHistoryRedisConsumerHandler { +) (*OnlineHistoryRedisConsumerHandler, error) { var och OnlineHistoryRedisConsumerHandler och.msgDatabase = database och.msgDistributionCh = make(chan Cmd2Value) // no buffer channel @@ -98,14 +98,15 @@ func NewOnlineHistoryRedisConsumerHandler( } och.conversationRpcClient = conversationRpcClient och.groupRpcClient = groupRpcClient - och.historyConsumerGroup = kafka.NewMConsumerGroup(&kafka.MConsumerGroupConfig{ + var err error + och.historyConsumerGroup, err = kafka.NewMConsumerGroup(&kafka.MConsumerGroupConfig{ KafkaVersion: sarama.V2_0_0_0, OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false, }, []string{config.Config.Kafka.LatestMsgToRedis.Topic}, config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToRedis) // statistics.NewStatistics(&och.singleMsgSuccessCount, config.Config.ModuleName.MsgTransferName, fmt.Sprintf("%d // second singleMsgCount insert to mongo", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval) - return &och + return &och, err } func (och *OnlineHistoryRedisConsumerHandler) Run(channelID int) { diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index 8ef15fe72..6e6c4c819 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -34,16 +34,21 @@ type OnlineHistoryMongoConsumerHandler struct { msgDatabase controller.CommonMsgDatabase } -func NewOnlineHistoryMongoConsumerHandler(database controller.CommonMsgDatabase) *OnlineHistoryMongoConsumerHandler { +func NewOnlineHistoryMongoConsumerHandler(database controller.CommonMsgDatabase) (*OnlineHistoryMongoConsumerHandler, error) { + historyConsumerGroup, err := kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{ + KafkaVersion: sarama.V2_0_0_0, + OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false, + }, []string{config.Config.Kafka.MsgToMongo.Topic}, + config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMongo) + if err != nil { + return nil, err + } + mc := &OnlineHistoryMongoConsumerHandler{ - historyConsumerGroup: kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{ - KafkaVersion: sarama.V2_0_0_0, - OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false, - }, []string{config.Config.Kafka.MsgToMongo.Topic}, - config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToMongo), - msgDatabase: database, + historyConsumerGroup: historyConsumerGroup, + msgDatabase: database, } - return mc + return mc, nil } func (mc *OnlineHistoryMongoConsumerHandler) handleChatWs2Mongo( diff --git a/internal/push/consumer_init.go b/internal/push/consumer_init.go index b72c32bb1..ceab86165 100644 --- a/internal/push/consumer_init.go +++ b/internal/push/consumer_init.go @@ -14,19 +14,24 @@ package push +import "context" + type Consumer struct { pushCh ConsumerHandler successCount uint64 } -func NewConsumer(pusher *Pusher) *Consumer { - return &Consumer{ - pushCh: *NewConsumerHandler(pusher), +func NewConsumer(pusher *Pusher) (*Consumer, error) { + c, err := NewConsumerHandler(pusher) + if err != nil { + return nil, err } + return &Consumer{ + pushCh: *c, + }, nil } func (c *Consumer) Start() { - // statistics.NewStatistics(&c.successCount, config.Config.ModuleName.PushName, fmt.Sprintf("%d second push to - // msg_gateway count", constant.StatisticsTimeInterval), constant.StatisticsTimeInterval) - go c.pushCh.pushConsumerGroup.RegisterHandleAndConsumer(&c.pushCh) + + go c.pushCh.pushConsumerGroup.RegisterHandleAndConsumer(context.Background(), &c.pushCh) } diff --git a/internal/push/push_handler.go b/internal/push/push_handler.go index c91206ecc..19d42ebb9 100644 --- a/internal/push/push_handler.go +++ b/internal/push/push_handler.go @@ -35,15 +35,19 @@ type ConsumerHandler struct { pusher *Pusher } -func NewConsumerHandler(pusher *Pusher) *ConsumerHandler { +func NewConsumerHandler(pusher *Pusher) (*ConsumerHandler, error) { var consumerHandler ConsumerHandler consumerHandler.pusher = pusher - consumerHandler.pushConsumerGroup = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{ + var err error + consumerHandler.pushConsumerGroup, err = kfk.NewMConsumerGroup(&kfk.MConsumerGroupConfig{ KafkaVersion: sarama.V2_0_0_0, OffsetsInitial: sarama.OffsetNewest, IsReturnErr: false, }, []string{config.Config.Kafka.MsgToPush.Topic}, config.Config.Kafka.Addr, config.Config.Kafka.ConsumerGroupID.MsgToPush) - return &consumerHandler + if err != nil { + return nil, err + } + return &consumerHandler, nil } func (c *ConsumerHandler) handleMs2PsChat(ctx context.Context, msg []byte) { diff --git a/internal/push/push_rpc_server.go b/internal/push/push_rpc_server.go index 188ddc0e1..c1226ce6b 100644 --- a/internal/push/push_rpc_server.go +++ b/internal/push/push_rpc_server.go @@ -66,9 +66,12 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e pusher: pusher, }) }() + consumer, err := NewConsumer(pusher) + if err != nil { + return err + } go func() { defer wg.Done() - consumer := NewConsumer(pusher) consumer.Start() }() wg.Wait() diff --git a/internal/rpc/msg/server.go b/internal/rpc/msg/server.go index 88be287fd..fe1baa453 100644 --- a/internal/rpc/msg/server.go +++ b/internal/rpc/msg/server.go @@ -80,7 +80,10 @@ func Start(client discoveryregistry.SvcDiscoveryRegistry, server *grpc.Server) e userRpcClient := rpcclient.NewUserRpcClient(client) groupRpcClient := rpcclient.NewGroupRpcClient(client) friendRpcClient := rpcclient.NewFriendRpcClient(client) - msgDatabase := controller.NewCommonMsgDatabase(msgDocModel, cacheModel) + msgDatabase, err := controller.NewCommonMsgDatabase(msgDocModel, cacheModel) + if err != nil { + return err + } s := &msgServer{ Conversation: &conversationClient, User: &userRpcClient, diff --git a/internal/tools/cron_task.go b/internal/tools/cron_task.go index e22504bbb..cf2068d8e 100644 --- a/internal/tools/cron_task.go +++ b/internal/tools/cron_task.go @@ -17,6 +17,7 @@ package tools import ( "context" "fmt" + "github.com/OpenIMSDK/tools/errs" "os" "os/signal" "syscall" @@ -25,14 +26,13 @@ import ( "github.com/redis/go-redis/v9" "github.com/robfig/cron/v3" - "github.com/OpenIMSDK/tools/log" - "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/db/cache" ) func StartTask() error { fmt.Println("cron task start, config", config.Config.ChatRecordsClearTime) + msgTool, err := InitMsgTool() if err != nil { return err @@ -47,18 +47,16 @@ func StartTask() error { // register cron tasks var crontab = cron.New() - log.ZInfo(context.Background(), "start chatRecordsClearTime cron task", "cron config", config.Config.ChatRecordsClearTime) + fmt.Println("start chatRecordsClearTime cron task", "cron config", config.Config.ChatRecordsClearTime) _, err = crontab.AddFunc(config.Config.ChatRecordsClearTime, cronWrapFunc(rdb, "cron_clear_msg_and_fix_seq", msgTool.AllConversationClearMsgAndFixSeq)) if err != nil { - log.ZError(context.Background(), "start allConversationClearMsgAndFixSeq cron failed", err) - panic(err) + return errs.Wrap(err) } - log.ZInfo(context.Background(), "start msgDestruct cron task", "cron config", config.Config.MsgDestructTime) + fmt.Println("start msgDestruct cron task", "cron config", config.Config.MsgDestructTime) _, err = crontab.AddFunc(config.Config.MsgDestructTime, cronWrapFunc(rdb, "cron_conversations_destruct_msgs", msgTool.ConversationsDestructMsgs)) if err != nil { - log.ZError(context.Background(), "start conversationsDestructMsgs cron failed", err) - panic(err) + return errs.Wrap(err) } // start crontab diff --git a/internal/tools/msg.go b/internal/tools/msg.go index 30006670e..1ec1e03a2 100644 --- a/internal/tools/msg.go +++ b/internal/tools/msg.go @@ -84,7 +84,10 @@ func InitMsgTool() (*MsgTool, error) { if err != nil { return nil, err } - msgDatabase := controller.InitCommonMsgDatabase(rdb, mongo.GetDatabase()) + msgDatabase, err := controller.InitCommonMsgDatabase(rdb, mongo.GetDatabase()) + if err != nil { + return nil, err + } userMongoDB := unrelation.NewUserMongoDriver(mongo.GetDatabase()) ctxTx := tx.NewMongo(mongo.GetClient()) userDatabase := controller.NewUserDatabase( diff --git a/pkg/common/db/cache/init_redis.go b/pkg/common/db/cache/init_redis.go index 1308e9649..c593089e6 100644 --- a/pkg/common/db/cache/init_redis.go +++ b/pkg/common/db/cache/init_redis.go @@ -49,7 +49,7 @@ func NewRedis() (redis.UniversalClient, error) { overrideConfigFromEnv() if len(config.Config.Redis.Address) == 0 { - return nil, errors.New("redis address is empty") + return nil, errs.Wrap(errors.New("redis address is empty")) } specialerror.AddReplace(redis.Nil, errs.ErrRecordNotFound) var rdb redis.UniversalClient @@ -77,7 +77,7 @@ func NewRedis() (redis.UniversalClient, error) { defer cancel() err = rdb.Ping(ctx).Err() if err != nil { - return nil, fmt.Errorf("redis ping %w", err) + return nil, errs.Wrap(fmt.Errorf("redis ping %w", err)) } redisClient = rdb diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index dfff5c61d..b841a7d31 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -126,21 +126,32 @@ type CommonMsgDatabase interface { ConvertMsgsDocLen(ctx context.Context, conversationIDs []string) } -func NewCommonMsgDatabase(msgDocModel unrelationtb.MsgDocModelInterface, cacheModel cache.MsgModel) CommonMsgDatabase { +func NewCommonMsgDatabase(msgDocModel unrelationtb.MsgDocModelInterface, cacheModel cache.MsgModel) (CommonMsgDatabase, error) { + producerToRedis, err := kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.LatestMsgToRedis.Topic) + if err != nil { + return nil, err + } + producerToMongo, err := kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToMongo.Topic) + if err != nil { + return nil, err + } + producerToPush, err := kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToPush.Topic) + if err != nil { + return nil, err + } return &commonMsgDatabase{ msgDocDatabase: msgDocModel, cache: cacheModel, - producer: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.LatestMsgToRedis.Topic), - producerToMongo: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToMongo.Topic), - producerToPush: kafka.NewKafkaProducer(config.Config.Kafka.Addr, config.Config.Kafka.MsgToPush.Topic), - } + producer: producerToRedis, + producerToMongo: producerToMongo, + producerToPush: producerToPush, + }, nil } -func InitCommonMsgDatabase(rdb redis.UniversalClient, database *mongo.Database) CommonMsgDatabase { +func InitCommonMsgDatabase(rdb redis.UniversalClient, database *mongo.Database) (CommonMsgDatabase, error) { cacheModel := cache.NewMsgCacheModel(rdb) msgDocModel := unrelation.NewMsgMongoDriver(database) - CommonMsgDatabase := NewCommonMsgDatabase(msgDocModel, cacheModel) - return CommonMsgDatabase + return NewCommonMsgDatabase(msgDocModel, cacheModel) } type commonMsgDatabase struct { diff --git a/pkg/common/db/mgo/conversation.go b/pkg/common/db/mgo/conversation.go index 1614cfec5..0b8d597dc 100644 --- a/pkg/common/db/mgo/conversation.go +++ b/pkg/common/db/mgo/conversation.go @@ -16,6 +16,7 @@ package mgo import ( "context" + "github.com/OpenIMSDK/tools/errs" "time" "github.com/OpenIMSDK/protocol/constant" @@ -38,7 +39,7 @@ func NewConversationMongo(db *mongo.Database) (*ConversationMgo, error) { Options: options.Index().SetUnique(true), }) if err != nil { - return nil, err + return nil, errs.Wrap(err) } return &ConversationMgo{coll: coll}, nil } diff --git a/pkg/common/db/mgo/group.go b/pkg/common/db/mgo/group.go index a9c6d1eb8..9a4c660a6 100644 --- a/pkg/common/db/mgo/group.go +++ b/pkg/common/db/mgo/group.go @@ -16,6 +16,7 @@ package mgo import ( "context" + "github.com/OpenIMSDK/tools/errs" "time" "github.com/OpenIMSDK/tools/mgoutil" @@ -36,7 +37,7 @@ func NewGroupMongo(db *mongo.Database) (relation.GroupModelInterface, error) { Options: options.Index().SetUnique(true), }) if err != nil { - return nil, err + return nil, errs.Wrap(err) } return &GroupMgo{coll: coll}, nil } diff --git a/pkg/common/db/mgo/group_member.go b/pkg/common/db/mgo/group_member.go index 8e3dd1efa..fce79830e 100644 --- a/pkg/common/db/mgo/group_member.go +++ b/pkg/common/db/mgo/group_member.go @@ -16,6 +16,7 @@ package mgo import ( "context" + "github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/protocol/constant" "github.com/OpenIMSDK/tools/mgoutil" @@ -37,7 +38,7 @@ func NewGroupMember(db *mongo.Database) (relation.GroupMemberModelInterface, err Options: options.Index().SetUnique(true), }) if err != nil { - return nil, err + return nil, errs.Wrap(err) } return &GroupMemberMgo{coll: coll}, nil } diff --git a/pkg/common/db/mgo/group_request.go b/pkg/common/db/mgo/group_request.go index cb04d2308..dce0878ee 100644 --- a/pkg/common/db/mgo/group_request.go +++ b/pkg/common/db/mgo/group_request.go @@ -16,6 +16,7 @@ package mgo import ( "context" + "github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/tools/mgoutil" "github.com/OpenIMSDK/tools/pagination" @@ -36,7 +37,7 @@ func NewGroupRequestMgo(db *mongo.Database) (relation.GroupRequestModelInterface Options: options.Index().SetUnique(true), }) if err != nil { - return nil, err + return nil, errs.Wrap(err) } return &GroupRequestMgo{coll: coll}, nil } diff --git a/pkg/common/db/mgo/user.go b/pkg/common/db/mgo/user.go index 34a25ed08..2797bc53f 100644 --- a/pkg/common/db/mgo/user.go +++ b/pkg/common/db/mgo/user.go @@ -40,7 +40,7 @@ func NewUserMongo(db *mongo.Database) (relation.UserModelInterface, error) { Options: options.Index().SetUnique(true), }) if err != nil { - return nil, err + return nil, errs.Wrap(err) } return &UserMgo{coll: coll}, nil } diff --git a/pkg/common/db/unrelation/mongo.go b/pkg/common/db/unrelation/mongo.go index b8184d767..e07c5b7f4 100644 --- a/pkg/common/db/unrelation/mongo.go +++ b/pkg/common/db/unrelation/mongo.go @@ -27,8 +27,6 @@ import ( "github.com/OpenIMSDK/tools/errs" "github.com/OpenIMSDK/tools/mw/specialerror" - "github.com/OpenIMSDK/tools/utils" - "github.com/openimsdk/open-im-server/v3/pkg/common/config" "github.com/openimsdk/open-im-server/v3/pkg/common/db/table/unrelation" ) @@ -63,9 +61,9 @@ func NewMongo() (*Mongo, error) { time.Sleep(time.Second) // exponential backoff could be implemented here continue } - return nil, err + return nil, errs.Wrap(err) } - return nil, err + return nil, errs.Wrap(err) } func buildMongoURI() string { @@ -150,7 +148,7 @@ func (m *Mongo) createMongoIndex(collection string, isUnique bool, keys ...strin _, err := indexView.CreateOne(context.Background(), index, opts) if err != nil { - return utils.Wrap(err, "CreateIndex") + return errs.Wrap(err, "CreateIndex") } return nil } diff --git a/pkg/common/kafka/consumer_group.go b/pkg/common/kafka/consumer_group.go index 1eb7b522a..05006f582 100644 --- a/pkg/common/kafka/consumer_group.go +++ b/pkg/common/kafka/consumer_group.go @@ -16,6 +16,8 @@ package kafka import ( "context" + "github.com/OpenIMSDK/tools/errs" + "strings" "github.com/OpenIMSDK/tools/log" @@ -36,7 +38,7 @@ type MConsumerGroupConfig struct { IsReturnErr bool } -func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []string, groupID string) *MConsumerGroup { +func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []string, groupID string) (*MConsumerGroup, error) { consumerGroupConfig := sarama.NewConfig() consumerGroupConfig.Version = consumerConfig.KafkaVersion consumerGroupConfig.Consumer.Offsets.Initial = consumerConfig.OffsetsInitial @@ -49,26 +51,28 @@ func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []str SetupTLSConfig(consumerGroupConfig) consumerGroup, err := sarama.NewConsumerGroup(addrs, groupID, consumerGroupConfig) if err != nil { - panic(err.Error()) + return nil, errs.Wrap(err, strings.Join(topics, ","), strings.Join(addrs, ","), groupID) } return &MConsumerGroup{ consumerGroup, groupID, topics, - } + }, nil } func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) context.Context { return GetContextWithMQHeader(cMsg.Headers) } -func (mc *MConsumerGroup) RegisterHandleAndConsumer(handler sarama.ConsumerGroupHandler) { +func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler) { log.ZDebug(context.Background(), "register consumer group", "groupID", mc.groupID) - ctx := context.Background() for { err := mc.ConsumerGroup.Consume(ctx, mc.topics, handler) if err != nil { - panic(err.Error()) + log.ZWarn(ctx, "consume err", err, "topic", mc.topics, "groupID", mc.groupID) + } + if ctx.Err() != nil { + return } } } diff --git a/pkg/common/kafka/producer.go b/pkg/common/kafka/producer.go index 06b1e2b4c..b9f0b4656 100644 --- a/pkg/common/kafka/producer.go +++ b/pkg/common/kafka/producer.go @@ -18,6 +18,7 @@ import ( "bytes" "context" "errors" + "github.com/OpenIMSDK/tools/errs" "strings" "time" @@ -44,7 +45,7 @@ type Producer struct { } // NewKafkaProducer initializes a new Kafka producer. -func NewKafkaProducer(addr []string, topic string) *Producer { +func NewKafkaProducer(addr []string, topic string) (*Producer, error) { p := Producer{ addr: addr, topic: topic, @@ -87,17 +88,17 @@ func NewKafkaProducer(addr []string, topic string) *Producer { for i := 0; i <= maxRetry; i++ { p.producer, err = sarama.NewSyncProducer(p.addr, p.config) if err == nil { - return &p + return &p, nil } time.Sleep(1 * time.Second) // Wait before retrying } // Panic if unable to create producer after retries if err != nil { - panic("Failed to create Kafka producer: " + err.Error()) + return nil, errs.Wrap(errors.New("failed to create Kafka producer: " + err.Error())) } - return &p + return &p, nil } // configureProducerAck configures the producer's acknowledgement level. diff --git a/pkg/common/startrpc/start.go b/pkg/common/startrpc/start.go index 8295404d3..31fe4fdd5 100644 --- a/pkg/common/startrpc/start.go +++ b/pkg/common/startrpc/start.go @@ -17,6 +17,7 @@ package startrpc import ( "errors" "fmt" + "github.com/OpenIMSDK/tools/errs" "log" "net" "net/http" @@ -43,7 +44,6 @@ import ( "github.com/OpenIMSDK/tools/discoveryregistry" "github.com/OpenIMSDK/tools/mw" "github.com/OpenIMSDK/tools/network" - "github.com/OpenIMSDK/tools/utils" ) // Start rpc server. @@ -61,20 +61,20 @@ func Start( net.JoinHostPort(network.GetListenIP(config.Config.Rpc.ListenIP), strconv.Itoa(rpcPort)), ) if err != nil { - return err + return errs.Wrap(err, network.GetListenIP(config.Config.Rpc.ListenIP), strconv.Itoa(rpcPort)) } defer listener.Close() client, err := kdisc.NewDiscoveryRegister(config.Config.Envs.Discovery) if err != nil { - return utils.Wrap1(err) + return errs.Wrap(err) } defer client.Close() client.AddOption(mw.GrpcClient(), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, "round_robin"))) registerIP, err := network.GetRpcRegisterIP(config.Config.Rpc.RegisterIP) if err != nil { - return err + return errs.Wrap(err) } var reg *prometheus.Registry @@ -96,7 +96,7 @@ func Start( err = rpcFn(client, srv) if err != nil { - return utils.Wrap1(err) + return errs.Wrap(err) } err = client.Register( rpcRegisterName, @@ -105,7 +105,7 @@ func Start( grpc.WithTransportCredentials(insecure.NewCredentials()), ) if err != nil { - return utils.Wrap1(err) + return errs.Wrap(err) } var wg errgroup.Group @@ -123,7 +123,7 @@ func Start( }) wg.Go(func() error { - return utils.Wrap1(srv.Serve(listener)) + return errs.Wrap(srv.Serve(listener)) }) sigs := make(chan os.Signal, 1) @@ -146,7 +146,7 @@ func Start( return gerr case <-time.After(15 * time.Second): - return utils.Wrap1(errors.New("timeout exit")) + return errs.Wrap(errors.New("timeout exit")) } } diff --git a/pkg/rpcclient/third.go b/pkg/rpcclient/third.go index 48a537112..73d874005 100755 --- a/pkg/rpcclient/third.go +++ b/pkg/rpcclient/third.go @@ -20,7 +20,6 @@ import ( "github.com/minio/minio-go/v7" "github.com/minio/minio-go/v7/pkg/credentials" - "google.golang.org/grpc" "github.com/OpenIMSDK/protocol/third" diff --git a/tools/component/component.go b/tools/component/component.go index 220b845ee..823d8174b 100644 --- a/tools/component/component.go +++ b/tools/component/component.go @@ -15,44 +15,27 @@ package main import ( - "context" "errors" "flag" "fmt" - "net" - "net/url" "os" "strings" "time" - "github.com/IBM/sarama" + "github.com/OpenIMSDK/tools/component" "github.com/OpenIMSDK/tools/errs" - "github.com/go-zookeeper/zk" - "go.mongodb.org/mongo-driver/mongo" - "go.mongodb.org/mongo-driver/mongo/options" "github.com/openimsdk/open-im-server/v3/pkg/common/config" - "github.com/minio/minio-go/v7" - "github.com/minio/minio-go/v7/pkg/credentials" - "github.com/redis/go-redis/v9" "gopkg.in/yaml.v3" ) const ( // defaultCfgPath is the default path of the configuration file. - defaultCfgPath = "../../../../../config/config.yaml" - minioHealthCheckDuration = 1 - maxRetry = 300 - componentStartErrCode = 6000 - configErrCode = 6001 - mongoConnTimeout = 30 * time.Second -) - -const ( - colorRed = 31 - colorGreen = 32 - colorYellow = 33 + defaultCfgPath = "../../../../../config/config.yaml" + maxRetry = 300 + componentStartErrCode = 6000 + configErrCode = 6001 ) var ( @@ -103,16 +86,16 @@ func main() { for _, check := range checks { str, err := check.function() if err != nil { - errorPrint(fmt.Sprintf("Starting %s failed, %v", check.name, err)) + component.ErrorPrint(fmt.Sprintf("Starting %s failed, %v", check.name, err)) allSuccess = false break } else { - successPrint(fmt.Sprintf("%s connected successfully, %s", check.name, str)) + component.SuccessPrint(fmt.Sprintf("%s connected successfully, %s", check.name, str)) } } if allSuccess { - successPrint("All components started successfully!") + component.SuccessPrint("All components started successfully!") return } @@ -120,19 +103,6 @@ func main() { os.Exit(1) } -func exactIP(urll string) string { - u, _ := url.Parse(urll) - host, _, err := net.SplitHostPort(u.Host) - if err != nil { - host = u.Host - } - if strings.HasSuffix(host, ":") { - host = host[0 : len(host)-1] - } - - return host -} - // Helper function to get environment variable or default value func getEnv(key, fallback string) string { if value, exists := os.LookupEnv(key); exists { @@ -143,45 +113,23 @@ func getEnv(key, fallback string) string { // checkMongo checks the MongoDB connection without retries func checkMongo() (string, error) { - uri := getEnv("MONGO_URI", buildMongoURI()) - - ctx, cancel := context.WithTimeout(context.Background(), mongoConnTimeout) - defer cancel() - - str := "ths addr is:" + strings.Join(config.Config.Mongo.Address, ",") - - client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri)) - if err != nil { - return "", errs.Wrap(errStr(err, str)) + mongo := &component.Mongo{ + Address: config.Config.Mongo.Address, + Database: config.Config.Mongo.Database, + Username: config.Config.Mongo.Username, + Password: config.Config.Mongo.Password, + MaxPoolSize: config.Config.Mongo.MaxPoolSize, } - defer client.Disconnect(context.Background()) - - ctx, cancel = context.WithTimeout(context.Background(), mongoConnTimeout) - defer cancel() - - if err = client.Ping(ctx, nil); err != nil { - return "", errs.Wrap(errStr(err, str)) + uri, uriExist := os.LookupEnv("MONGO_URI") + if uriExist { + mongo.URL = uri } - return str, nil -} - -// buildMongoURI constructs the MongoDB URI using configuration settings -func buildMongoURI() string { - // Fallback to config if environment variables are not set - username := config.Config.Mongo.Username - password := config.Config.Mongo.Password - database := config.Config.Mongo.Database - maxPoolSize := config.Config.Mongo.MaxPoolSize - - mongodbHosts := strings.Join(config.Config.Mongo.Address, ",") - - if username != "" && password != "" { - return fmt.Sprintf("mongodb://%s:%s@%s/%s?maxPoolSize=%d", - username, password, mongodbHosts, database, maxPoolSize) + str, err := component.CheckMongo(mongo) + if err != nil { + return "", err } - return fmt.Sprintf("mongodb://%s/%s?maxPoolSize=%d", - mongodbHosts, database, maxPoolSize) + return str, nil } // checkMinio checks the MinIO connection @@ -191,52 +139,24 @@ func checkMinio() (string, error) { return "", nil } - // Prioritize environment variables - endpoint := getEnv("MINIO_ENDPOINT", config.Config.Object.Minio.Endpoint) - accessKeyID := getEnv("MINIO_ACCESS_KEY_ID", config.Config.Object.Minio.AccessKeyID) - secretAccessKey := getEnv("MINIO_SECRET_ACCESS_KEY", config.Config.Object.Minio.SecretAccessKey) - useSSL := getEnv("MINIO_USE_SSL", "false") // Assuming SSL is not used by default - - if endpoint == "" || accessKeyID == "" || secretAccessKey == "" { - return "", ErrConfig.Wrap("MinIO configuration missing") - } - - // Parse endpoint URL to determine if SSL is enabled - u, err := url.Parse(endpoint) + endpoint, err := getMinioAddr("MINIO_ENDPOINT", "MINIO_ADDRESS", "MINIO_PORT", config.Config.Object.Minio.Endpoint) if err != nil { - str := "the endpoint is:" + endpoint - return "", errs.Wrap(errStr(err, str)) - } - secure := u.Scheme == "https" || useSSL == "true" - - // Initialize MinIO client - minioClient, err := minio.New(u.Host, &minio.Options{ - Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""), - Secure: secure, - }) - str := "ths addr is:" + u.Host - if err != nil { - strs := fmt.Sprintf("%v;host:%s,accessKeyID:%s,secretAccessKey:%s,Secure:%v", err, u.Host, accessKeyID, secretAccessKey, secure) - return "", errs.Wrap(err, strs) + return "", err } - // Perform health check - cancel, err := minioClient.HealthCheck(time.Duration(minioHealthCheckDuration) * time.Second) - if err != nil { - return "", errs.Wrap(errStr(err, str)) + minio := &component.Minio{ + ApiURL: config.Config.Object.ApiURL, + Endpoint: endpoint, + AccessKeyID: getEnv("MINIO_ACCESS_KEY_ID", config.Config.Object.Minio.AccessKeyID), + SecretAccessKey: getEnv("MINIO_SECRET_ACCESS_KEY", config.Config.Object.Minio.SecretAccessKey), + SignEndpoint: config.Config.Object.Minio.SignEndpoint, + UseSSL: getEnv("MINIO_USE_SSL", "false"), } - defer cancel() - if minioClient.IsOffline() { - str := fmt.Sprintf("Minio server is offline;%s", str) - return "", ErrComponentStart.Wrap(str) - } - - // Check for localhost in API URL and Minio SignEndpoint - if exactIP(config.Config.Object.ApiURL) == "127.0.0.1" || exactIP(config.Config.Object.Minio.SignEndpoint) == "127.0.0.1" { - return "", ErrConfig.Wrap("apiURL or Minio SignEndpoint endpoint contain 127.0.0.1") + str, err := component.CheckMinio(minio) + if err != nil { + return "", err } - return str, nil } @@ -247,76 +167,48 @@ func checkRedis() (string, error) { username := getEnv("REDIS_USERNAME", config.Config.Redis.Username) password := getEnv("REDIS_PASSWORD", config.Config.Redis.Password) - // Split address to handle multiple addresses for cluster setup - redisAddresses := strings.Split(address, ",") - - var redisClient redis.UniversalClient - if len(redisAddresses) > 1 { - // Use cluster client for multiple addresses - redisClient = redis.NewClusterClient(&redis.ClusterOptions{ - Addrs: redisAddresses, - Username: username, - Password: password, - }) - } else { - // Use regular client for single address - redisClient = redis.NewClient(&redis.Options{ - Addr: redisAddresses[0], - Username: username, - Password: password, - }) + redis := &component.Redis{ + Address: strings.Split(address, ","), + Username: username, + Password: password, } - defer redisClient.Close() - // Ping Redis to check connectivity - _, err := redisClient.Ping(context.Background()).Result() - str := "the addr is:" + strings.Join(redisAddresses, ",") + addresses, err := getAddress("REDIS_ADDRESS", "REDIS_PORT", config.Config.Redis.Address) if err != nil { - return "", errs.Wrap(errStr(err, str)) + return "", err } + redis.Address = addresses + str, err := component.CheckRedis(redis) + if err != nil { + return "", err + } return str, nil } // checkZookeeper checks the Zookeeper connection func checkZookeeper() (string, error) { // Prioritize environment variables - schema := getEnv("ZOOKEEPER_SCHEMA", "digest") + address := getEnv("ZOOKEEPER_ADDRESS", strings.Join(config.Config.Zookeeper.ZkAddr, ",")) - username := getEnv("ZOOKEEPER_USERNAME", config.Config.Zookeeper.Username) - password := getEnv("ZOOKEEPER_PASSWORD", config.Config.Zookeeper.Password) - // Split addresses to handle multiple Zookeeper nodes - zookeeperAddresses := strings.Split(address, ",") + zk := &component.Zookeeper{ + Schema: getEnv("ZOOKEEPER_SCHEMA", "digest"), + ZkAddr: strings.Split(address, ","), + Username: getEnv("ZOOKEEPER_USERNAME", config.Config.Zookeeper.Username), + Password: getEnv("ZOOKEEPER_PASSWORD", config.Config.Zookeeper.Password), + } - // Connect to Zookeeper - str := "the addr is:" + address - c, eventChan, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary + addresses, err := getAddress("ZOOKEEPER_ADDRESS", "ZOOKEEPER_PORT", config.Config.Zookeeper.ZkAddr) if err != nil { - return "", errs.Wrap(errStr(err, str)) - } - timeout := time.After(5 * time.Second) - for { - select { - case event := <-eventChan: - if event.State == zk.StateConnected { - fmt.Println("Connected to Zookeeper") - goto Connected - } - case <-timeout: - return "", errs.Wrap(errors.New("timeout waiting for Zookeeper connection"), "Zookeeper Addr: "+strings.Join(config.Config.Zookeeper.ZkAddr, " ")) - } + return "", nil } -Connected: - defer c.Close() + zk.ZkAddr = addresses - // Set authentication if username and password are provided - if username != "" && password != "" { - if err := c.AddAuth(schema, []byte(username+":"+password)); err != nil { - return "", errs.Wrap(errStr(err, str)) - } + str, err := component.CheckZookeeper(zk) + if err != nil { + return "", err } - return str, nil } @@ -327,24 +219,21 @@ func checkKafka() (string, error) { password := getEnv("KAFKA_PASSWORD", config.Config.Kafka.Password) address := getEnv("KAFKA_ADDRESS", strings.Join(config.Config.Kafka.Addr, ",")) - // Split addresses to handle multiple Kafka brokers - kafkaAddresses := strings.Split(address, ",") + kafka := &component.Kafka{ + Username: username, + Password: password, + Addr: strings.Split(address, ","), + } - // Configure Kafka client - cfg := sarama.NewConfig() - if username != "" && password != "" { - cfg.Net.SASL.Enable = true - cfg.Net.SASL.User = username - cfg.Net.SASL.Password = password + addresses, err := getAddress("KAFKA_ADDRESS", "KAFKA_PORT", config.Config.Kafka.Addr) + if err != nil { + return "", nil } - // Additional Kafka setup (e.g., TLS configuration) can be added here - // kafka.SetupTLSConfig(cfg) + kafka.Addr = addresses - // Create Kafka client - str := "the addr is:" + address - kafkaClient, err := sarama.NewClient(kafkaAddresses, cfg) + str, kafkaClient, err := component.CheckKafka(kafka) if err != nil { - return "", errs.Wrap(errStr(err, str)) + return "", err } defer kafkaClient.Close() @@ -379,22 +268,42 @@ func isTopicPresent(topic string, topics []string) bool { return false } -func colorPrint(colorCode int, format string, a ...interface{}) { - fmt.Printf("\x1b[%dm%s\x1b[0m\n", colorCode, fmt.Sprintf(format, a...)) -} - -func errorPrint(s string) { - colorPrint(colorRed, "%v", s) -} - -func successPrint(s string) { - colorPrint(colorGreen, "%v", s) -} +func getAddress(key1, key2 string, fallback []string) ([]string, error) { + address, addrExist := os.LookupEnv(key1) + port, portExist := os.LookupEnv(key2) -func warningPrint(s string) { - colorPrint(colorYellow, "Warning: But %v", s) + if addrExist && portExist { + addresses := strings.Split(address, ",") + for i, addr := range addresses { + addresses[i] = addr + ":" + port + } + return addresses, nil + } else if !addrExist && portExist { + result := make([]string, len(config.Config.Redis.Address)) + for i, addr := range config.Config.Redis.Address { + add := strings.Split(addr, ":") + result[i] = add[0] + ":" + port + } + return result, nil + } else if addrExist && !portExist { + return nil, errs.Wrap(errors.New("the ZOOKEEPER_PORT of minio is empty")) + } + return fallback, nil } -func errStr(err error, str string) error { - return fmt.Errorf("%v;%s", err, str) +func getMinioAddr(key1, key2, key3, fallback string) (string, error) { + // Prioritize environment variables + endpoint := getEnv(key1, fallback) + address, addressExist := os.LookupEnv(key2) + port, portExist := os.LookupEnv(key3) + if portExist && addressExist { + endpoint = "http://" + address + ":" + port + } else if !portExist && addressExist { + return "", errs.Wrap(errors.New("the MINIO_PORT of minio is empty")) + } else if portExist && !addressExist { + arr := strings.Split(config.Config.Object.Minio.Endpoint, ":") + arr[2] = port + endpoint = strings.Join(arr, ":") + } + return endpoint, nil }