fix: Refactoring the code for component detection (#1868)

* feat: add component check func

* fix: fix the outpu error

* fix: fix the stderr outpu

* fix: fix the component check func

* fix: fix the error

* fix: fix the output error

* fix: del the disruptions code

* fix the log output format

* fix: fix the tools version

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.31
+	github.com/OpenIMSDK/tools v0.0.32
 	github.com/bwmarrin/snowflake v0.3.0 // indirect
 	github.com/dtm-labs/rockscache v0.1.1
 	github.com/gin-gonic/gin v1.9.1

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.31 h1:fSrhcPTvHEMTSyrJZDupe730mL4nuhvSOUP/BaZiHaY=
+github.com/OpenIMSDK/tools v0.0.32 h1:b8KwtxXKZTsyyHUcZ4OtSo6s/vVXx4HjMuPxH7Kb7Gg=
-github.com/OpenIMSDK/tools v0.0.31/go.mod h1:wBfR5CYmEyvxl03QJbTkhz1CluK6J4/lX0lviu8JAjE=
+github.com/OpenIMSDK/tools v0.0.32/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=

pkg/common/db/cache/init_redis.go

@@ -77,9 +77,10 @@ func NewRedis() (redis.UniversalClient, error) {
 	defer cancel()
 	err = rdb.Ping(ctx).Err()
 	if err != nil {
-		return nil, errs.Wrap(fmt.Errorf("redis ping %w", err))
+		uriFormat := "address:%s, username:%s, password:%s, clusterMode:%t, enablePipeline:%t"
+		errMsg := fmt.Sprintf(uriFormat, config.Config.Redis.Address, config.Config.Redis.Username, config.Config.Redis.Password, config.Config.Redis.ClusterMode, config.Config.Redis.EnablePipeline)
+		return nil, errs.Wrap(err, errMsg)
 	}
-
 	redisClient = rdb
 	return rdb, err
 }

pkg/common/db/unrelation/mongo.go

@@ -61,9 +61,9 @@ func NewMongo() (*Mongo, error) {
 			time.Sleep(time.Second) // exponential backoff could be implemented here
 			continue
 		}
-		return nil, errs.Wrap(err)
+		return nil, errs.Wrap(err, uri)
 	}
-	return nil, errs.Wrap(err)
+	return nil, errs.Wrap(err, uri)
 }
 
 func buildMongoURI() string {

pkg/common/discoveryregister/zookeeper/zookeeper.go

@@ -15,6 +15,8 @@
 package zookeeper
 
 import (
+	"fmt"
+	"github.com/OpenIMSDK/tools/errs"
 	"os"
 	"strings"
 	"time"
@@ -33,7 +35,7 @@ func NewZookeeperDiscoveryRegister() (discoveryregistry.SvcDiscoveryRegistry, er
 	username := getEnv("ZOOKEEPER_USERNAME", config.Config.Zookeeper.Username)
 	password := getEnv("ZOOKEEPER_PASSWORD", config.Config.Zookeeper.Password)
 
-	return openkeeper.NewClient(
+	zk, err := openkeeper.NewClient(
 		zkAddr,
 		schema,
 		openkeeper.WithFreq(time.Hour),
@@ -42,6 +44,16 @@ func NewZookeeperDiscoveryRegister() (discoveryregistry.SvcDiscoveryRegistry, er
 		openkeeper.WithTimeout(10),
 		openkeeper.WithLogger(log.NewZkLogger()),
 	)
+	if err != nil {
+		uriFormat := "address:%s, username :%s, password :%s, schema:%s."
+		errInfo := fmt.Sprintf(uriFormat,
+			config.Config.Zookeeper.ZkAddr,
+			config.Config.Zookeeper.Username,
+			config.Config.Zookeeper.Password,
+			config.Config.Zookeeper.Schema)
+		return nil, errs.Wrap(err, errInfo)
+	}
+	return zk, nil
 }
 
 // getEnv returns the value of an environment variable if it exists, otherwise it returns the fallback value.

pkg/common/kafka/consumer_group.go

@@ -51,7 +51,7 @@ func NewMConsumerGroup(consumerConfig *MConsumerGroupConfig, topics, addrs []str
 	SetupTLSConfig(consumerGroupConfig)
 	consumerGroup, err := sarama.NewConsumerGroup(addrs, groupID, consumerGroupConfig)
 	if err != nil {
-		return nil, errs.Wrap(err, strings.Join(topics, ","), strings.Join(addrs, ","), groupID)
+		return nil, errs.Wrap(err, strings.Join(topics, ","), strings.Join(addrs, ","), groupID, config.Config.Kafka.Username, config.Config.Kafka.Password)
 	}
 	return &MConsumerGroup{
 		consumerGroup,

tools/component/component.go

@@ -15,9 +15,13 @@
 package main
 
 import (
-	"errors"
 	"flag"
 	"fmt"
+	"github.com/IBM/sarama"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/cache"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/db/unrelation"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister/zookeeper"
+	"github.com/openimsdk/open-im-server/v3/pkg/common/kafka"
 	"os"
 	"strings"
 	"time"
@@ -32,16 +36,12 @@ import (
 
 const (
 	// defaultCfgPath is the default path of the configuration file.
-	defaultCfgPath        = "../../../../../config/config.yaml"
+	defaultCfgPath = "../../../../../config/config.yaml"
-	maxRetry              = 300
+	maxRetry       = 300
-	componentStartErrCode = 6000
-	configErrCode         = 6001
 )
 
 var (
-	cfgPath           = flag.String("c", defaultCfgPath, "Path to the configuration file")
+	cfgPath = flag.String("c", defaultCfgPath, "Path to the configuration file")
-	ErrComponentStart = errs.NewCodeError(componentStartErrCode, "ComponentStartErr")
-	ErrConfig         = errs.NewCodeError(configErrCode, "Config file is incorrect")
 )
 
 func initCfg() error {
@@ -55,7 +55,7 @@ func initCfg() error {
 
 type checkFunc struct {
 	name     string
-	function func() (string, error)
+	function func() error
 }
 
 func main() {
@@ -67,11 +67,13 @@ func main() {
 		return
 	}
 
+	configGetEnv()
+
 	checks := []checkFunc{
 		//{name: "Mysql", function: checkMysql},
 		{name: "Mongo", function: checkMongo},
-		{name: "Minio", function: checkMinio},
 		{name: "Redis", function: checkRedis},
+		{name: "Minio", function: checkMinio},
 		{name: "Zookeeper", function: checkZookeeper},
 		{name: "Kafka", function: checkKafka},
 	}
@@ -82,165 +84,81 @@ func main() {
 		}
 		fmt.Printf("Checking components Round %v...\n", i+1)
 
+		var err error
 		allSuccess := true
 		for _, check := range checks {
-			str, err := check.function()
+			err = check.function()
 			if err != nil {
-				component.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 {
-				component.SuccessPrint(fmt.Sprintf("%s connected successfully, %s", check.name, str))
+				component.SuccessPrint(fmt.Sprintf("%s connected successfully", check.name))
 			}
 		}
 
 		if allSuccess {
 			component.SuccessPrint("All components started successfully!")
-
 			return
 		}
 	}
-	os.Exit(1)
-}
-
-// Helper function to get environment variable or default value
-func getEnv(key, fallback string) string {
-	if value, exists := os.LookupEnv(key); exists {
-		return value
-	}
-	return fallback
 }
 
 // checkMongo checks the MongoDB connection without retries
-func checkMongo() (string, error) {
+func checkMongo() error {
-	mongo := &component.Mongo{
+	_, err := unrelation.NewMongo()
-		Address:     config.Config.Mongo.Address,
+	return err
-		Database:    config.Config.Mongo.Database,
-		Username:    config.Config.Mongo.Username,
-		Password:    config.Config.Mongo.Password,
-		MaxPoolSize: config.Config.Mongo.MaxPoolSize,
-	}
-	uri, uriExist := os.LookupEnv("MONGO_URI")
-	if uriExist {
-		mongo.URL = uri
-	}
-
-	str, err := component.CheckMongo(mongo)
-	if err != nil {
-		return "", err
-	}
-	return str, nil
-}
-
-// checkMinio checks the MinIO connection
-func checkMinio() (string, error) {
-	// Check if MinIO is enabled
-	if config.Config.Object.Enable != "minio" {
-		return "", nil
-	}
-
-	endpoint, err := getMinioAddr("MINIO_ENDPOINT", "MINIO_ADDRESS", "MINIO_PORT", config.Config.Object.Minio.Endpoint)
-	if err != nil {
-		return "", err
-	}
-
-	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"),
-	}
-
-	str, err := component.CheckMinio(minio)
-	if err != nil {
-		return "", err
-	}
-	return str, nil
 }
 
 // checkRedis checks the Redis connection
-func checkRedis() (string, error) {
+func checkRedis() error {
-	// Prioritize environment variables
+	_, err := cache.NewRedis()
-	address := getEnv("REDIS_ADDRESS", strings.Join(config.Config.Redis.Address, ","))
+	return err
-	username := getEnv("REDIS_USERNAME", config.Config.Redis.Username)
+}
-	password := getEnv("REDIS_PASSWORD", config.Config.Redis.Password)
 
-	redis := &component.Redis{
+// checkMinio checks the MinIO connection
-		Address:  strings.Split(address, ","),
+func checkMinio() error {
-		Username: username,
-		Password: password,
-	}
 
-	addresses, err := getAddress("REDIS_ADDRESS", "REDIS_PORT", config.Config.Redis.Address)
+	// Check if MinIO is enabled
-	if err != nil {
+	if config.Config.Object.Enable != "minio" {
-		return "", err
+		return nil
 	}
-	redis.Address = addresses
+	minio := &component.Minio{
-
+		ApiURL:          config.Config.Object.ApiURL,
-	str, err := component.CheckRedis(redis)
+		Endpoint:        config.Config.Object.Minio.Endpoint,
-	if err != nil {
+		AccessKeyID:     config.Config.Object.Minio.AccessKeyID,
-		return "", err
+		SecretAccessKey: config.Config.Object.Minio.SecretAccessKey,
+		SignEndpoint:    config.Config.Object.Minio.SignEndpoint,
+		UseSSL:          getEnv("MINIO_USE_SSL", "false"),
 	}
-	return str, nil
+	_, err := component.CheckMinio(minio)
+	return err
 }
 
 // checkZookeeper checks the Zookeeper connection
-func checkZookeeper() (string, error) {
+func checkZookeeper() error {
-	// Prioritize environment variables
+	_, err := zookeeper.NewZookeeperDiscoveryRegister()
-
+	return err
-	address := getEnv("ZOOKEEPER_ADDRESS", strings.Join(config.Config.Zookeeper.ZkAddr, ","))
-
-	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),
-	}
-
-	addresses, err := getAddress("ZOOKEEPER_ADDRESS", "ZOOKEEPER_PORT", config.Config.Zookeeper.ZkAddr)
-	if err != nil {
-		return "", nil
-	}
-	zk.ZkAddr = addresses
-
-	str, err := component.CheckZookeeper(zk)
-	if err != nil {
-		return "", err
-	}
-	return str, nil
 }
 
 // checkKafka checks the Kafka connection
-func checkKafka() (string, error) {
+func checkKafka() error {
 	// Prioritize environment variables
-	username := getEnv("KAFKA_USERNAME", config.Config.Kafka.Username)
+	kafkaStu := &component.Kafka{
-	password := getEnv("KAFKA_PASSWORD", config.Config.Kafka.Password)
+		Username: config.Config.Kafka.Username,
-	address := getEnv("KAFKA_ADDRESS", strings.Join(config.Config.Kafka.Addr, ","))
+		Password: config.Config.Kafka.Password,
-
+		Addr:     config.Config.Kafka.Addr,
-	kafka := &component.Kafka{
-		Username: username,
-		Password: password,
-		Addr:     strings.Split(address, ","),
 	}
 
-	addresses, err := getAddress("KAFKA_ADDRESS", "KAFKA_PORT", config.Config.Kafka.Addr)
+	_, kafkaClient, err := component.CheckKafka(kafkaStu)
 	if err != nil {
-		return "", nil
+		return err
-	}
-	kafka.Addr = addresses
-
-	str, kafkaClient, err := component.CheckKafka(kafka)
-	if err != nil {
-		return "", err
 	}
 	defer kafkaClient.Close()
 
 	// Verify if necessary topics exist
 	topics, err := kafkaClient.Topics()
 	if err != nil {
-		return "", errs.Wrap(err)
+		return errs.Wrap(err)
 	}
 
 	requiredTopics := []string{
@@ -251,11 +169,38 @@ func checkKafka() (string, error) {
 
 	for _, requiredTopic := range requiredTopics {
 		if !isTopicPresent(requiredTopic, topics) {
-			return "", ErrComponentStart.Wrap(fmt.Sprintf("Kafka doesn't contain topic: %v", requiredTopic))
+			return errs.Wrap(err, fmt.Sprintf("Kafka doesn't contain topic: %v", requiredTopic))
 		}
 	}
 
-	return str, nil
+	_, 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)
+	if err != nil {
+		return err
+	}
+
+	_, err = kafka.NewMConsumerGroup(&kafka.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.MsgToMongo)
+	if err != nil {
+		return err
+	}
+
+	kafka.NewMConsumerGroup(&kafka.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)
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // isTopicPresent checks if a topic is present in the list of topics
@@ -268,42 +213,34 @@ func isTopicPresent(topic string, topics []string) bool {
 	return false
 }
 
-func getAddress(key1, key2 string, fallback []string) ([]string, error) {
+func configGetEnv() {
-	address, addrExist := os.LookupEnv(key1)
+	config.Config.Object.Minio.AccessKeyID = getEnv("MINIO_ACCESS_KEY_ID", config.Config.Object.Minio.AccessKeyID)
-	port, portExist := os.LookupEnv(key2)
+	config.Config.Object.Minio.SecretAccessKey = getEnv("MINIO_SECRET_ACCESS_KEY", config.Config.Object.Minio.SecretAccessKey)
-
+	config.Config.Mongo.Uri = getEnv("MONGO_URI", config.Config.Mongo.Uri)
-	if addrExist && portExist {
+	config.Config.Mongo.Username = getEnv("MONGO_OPENIM_USERNAME", config.Config.Mongo.Username)
-		addresses := strings.Split(address, ",")
+	config.Config.Mongo.Password = getEnv("MONGO_OPENIM_PASSWORD", config.Config.Mongo.Password)
-		for i, addr := range addresses {
+	config.Config.Kafka.Username = getEnv("KAFKA_USERNAME", config.Config.Kafka.Username)
-			addresses[i] = addr + ":" + port
+	config.Config.Kafka.Password = getEnv("KAFKA_PASSWORD", config.Config.Kafka.Password)
-		}
+	config.Config.Kafka.Addr = strings.Split(getEnv("KAFKA_ADDRESS", strings.Join(config.Config.Kafka.Addr, ",")), ",")
-		return addresses, nil
+	config.Config.Object.Minio.Endpoint = getMinioAddr("MINIO_ENDPOINT", "MINIO_ADDRESS", "MINIO_PORT", config.Config.Object.Minio.Endpoint)
-	} 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 getMinioAddr(key1, key2, key3, fallback string) (string, error) {
+func getMinioAddr(key1, key2, key3, fallback string) string {
 	// 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 endpoint
-		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
+	return endpoint
+}
+
+// Helper function to get environment variable or default value
+func getEnv(key, fallback string) string {
+	if value, exists := os.LookupEnv(key); exists {
+		return value
+	}
+	return fallback
 }