fix: update the component output format

pull/1585/head
luhaoling 2 years ago
parent 87610568ae
commit 27ea09eacd

@ -72,7 +72,7 @@ func initCfg() error {
type checkFunc struct { type checkFunc struct {
name string name string
function func() error function func() (string, error)
} }
func main() { func main() {
@ -101,13 +101,13 @@ func main() {
allSuccess := true allSuccess := true
for _, check := range checks { for _, check := range checks {
err := check.function() str, err := check.function()
if err != nil { if err != nil {
errorPrint(fmt.Sprintf("Starting %s failed: %v", check.name, err)) errorPrint(fmt.Sprintf("Starting %s failed: %v", check.name, err))
allSuccess = false allSuccess = false
break break
} else { } else {
successPrint(fmt.Sprintf("%s starts successfully", check.name)) successPrint(fmt.Sprintf("%s connected successfully, the connected address is: %s", check.name, str))
} }
} }
@ -142,21 +142,22 @@ func getEnv(key, fallback string) string {
} }
// checkMongo checks the MongoDB connection // checkMongo checks the MongoDB connection
func checkMongo() error { func checkMongo() (string, error) {
// Use environment variables or fallback to config // Use environment variables or fallback to config
uri := getEnv("MONGO_URI", buildMongoURI()) uri := getEnv("MONGO_URI", buildMongoURI())
client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(uri)) client, err := mongo.Connect(context.TODO(), options.Client().ApplyURI(uri))
str := "addr:" + uri
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
defer client.Disconnect(context.TODO()) defer client.Disconnect(context.TODO())
if err = client.Ping(context.TODO(), nil); err != nil { if err = client.Ping(context.TODO(), nil); err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
return nil return str, nil
} }
// buildMongoURI constructs the MongoDB URI using configuration settings // buildMongoURI constructs the MongoDB URI using configuration settings
@ -178,10 +179,10 @@ func buildMongoURI() string {
} }
// checkMinio checks the MinIO connection // checkMinio checks the MinIO connection
func checkMinio() error { func checkMinio() (string, error) {
// Check if MinIO is enabled // Check if MinIO is enabled
if config.Config.Object.Enable != "minio" { if config.Config.Object.Enable != "minio" {
return nil return "", nil
} }
// Prioritize environment variables // Prioritize environment variables
@ -191,13 +192,13 @@ func checkMinio() error {
useSSL := getEnv("MINIO_USE_SSL", "false") // Assuming SSL is not used by default useSSL := getEnv("MINIO_USE_SSL", "false") // Assuming SSL is not used by default
if endpoint == "" || accessKeyID == "" || secretAccessKey == "" { if endpoint == "" || accessKeyID == "" || secretAccessKey == "" {
return ErrConfig.Wrap("MinIO configuration missing") return "", ErrConfig.Wrap("MinIO configuration missing")
} }
// Parse endpoint URL to determine if SSL is enabled // Parse endpoint URL to determine if SSL is enabled
u, err := url.Parse(endpoint) u, err := url.Parse(endpoint)
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err)
} }
secure := u.Scheme == "https" || useSSL == "true" secure := u.Scheme == "https" || useSSL == "true"
@ -206,31 +207,32 @@ func checkMinio() error {
Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""), Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
Secure: secure, Secure: secure,
}) })
str := "addr:" + u.Host
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
// Perform health check // Perform health check
cancel, err := minioClient.HealthCheck(time.Duration(minioHealthCheckDuration) * time.Second) cancel, err := minioClient.HealthCheck(time.Duration(minioHealthCheckDuration) * time.Second)
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
defer cancel() defer cancel()
if minioClient.IsOffline() { if minioClient.IsOffline() {
return ErrComponentStart.Wrap("Minio server is offline") return "", ErrComponentStart.Wrap("Minio server is offline", str)
} }
// Check for localhost in API URL and Minio SignEndpoint // 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" { 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") return "", ErrConfig.Wrap("apiURL or Minio SignEndpoint endpoint contain 127.0.0.1")
} }
return nil return str, nil
} }
// checkRedis checks the Redis connection // checkRedis checks the Redis connection
func checkRedis() error { func checkRedis() (string, error) {
// Prioritize environment variables // Prioritize environment variables
address := getEnv("REDIS_ADDRESS", strings.Join(config.Config.Redis.Address, ",")) address := getEnv("REDIS_ADDRESS", strings.Join(config.Config.Redis.Address, ","))
username := getEnv("REDIS_USERNAME", config.Config.Redis.Username) username := getEnv("REDIS_USERNAME", config.Config.Redis.Username)
@ -259,15 +261,16 @@ func checkRedis() error {
// Ping Redis to check connectivity // Ping Redis to check connectivity
_, err := redisClient.Ping(context.Background()).Result() _, err := redisClient.Ping(context.Background()).Result()
str := "address:" + strings.Join(redisAddresses, ",")
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
return nil return str, nil
} }
// checkZookeeper checks the Zookeeper connection // checkZookeeper checks the Zookeeper connection
func checkZookeeper() error { func checkZookeeper() (string, error) {
// Prioritize environment variables // Prioritize environment variables
schema := getEnv("ZOOKEEPER_SCHEMA", "digest") schema := getEnv("ZOOKEEPER_SCHEMA", "digest")
address := getEnv("ZOOKEEPER_ADDRESS", strings.Join(config.Config.Zookeeper.ZkAddr, ",")) address := getEnv("ZOOKEEPER_ADDRESS", strings.Join(config.Config.Zookeeper.ZkAddr, ","))
@ -278,30 +281,31 @@ func checkZookeeper() error {
zookeeperAddresses := strings.Split(address, ",") zookeeperAddresses := strings.Split(address, ",")
// Connect to Zookeeper // Connect to Zookeeper
str := "addr:" + address
c, _, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary c, _, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
defer c.Close() defer c.Close()
// Set authentication if username and password are provided // Set authentication if username and password are provided
if username != "" && password != "" { if username != "" && password != "" {
if err := c.AddAuth(schema, []byte(username+":"+password)); err != nil { if err := c.AddAuth(schema, []byte(username+":"+password)); err != nil {
return errs.Wrap(err) return "", errs.Wrap(err)
} }
} }
// Check if Zookeeper is reachable // Check if Zookeeper is reachable
_, _, err = c.Get("/") _, _, err = c.Get("/")
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
return nil return str, nil
} }
// checkKafka checks the Kafka connection // checkKafka checks the Kafka connection
func checkKafka() error { func checkKafka() (string, error) {
// Prioritize environment variables // Prioritize environment variables
username := getEnv("KAFKA_USERNAME", config.Config.Kafka.Username) username := getEnv("KAFKA_USERNAME", config.Config.Kafka.Username)
password := getEnv("KAFKA_PASSWORD", config.Config.Kafka.Password) password := getEnv("KAFKA_PASSWORD", config.Config.Kafka.Password)
@ -321,16 +325,17 @@ func checkKafka() error {
// kafka.SetupTLSConfig(cfg) // kafka.SetupTLSConfig(cfg)
// Create Kafka client // Create Kafka client
str := "addr:" + address
kafkaClient, err := sarama.NewClient(kafkaAddresses, cfg) kafkaClient, err := sarama.NewClient(kafkaAddresses, cfg)
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err, str)
} }
defer kafkaClient.Close() defer kafkaClient.Close()
// Verify if necessary topics exist // Verify if necessary topics exist
topics, err := kafkaClient.Topics() topics, err := kafkaClient.Topics()
if err != nil { if err != nil {
return errs.Wrap(err) return "", errs.Wrap(err)
} }
requiredTopics := []string{ requiredTopics := []string{
@ -341,11 +346,11 @@ func checkKafka() error {
for _, requiredTopic := range requiredTopics { for _, requiredTopic := range requiredTopics {
if !isTopicPresent(requiredTopic, topics) { if !isTopicPresent(requiredTopic, topics) {
return ErrComponentStart.Wrap(fmt.Sprintf("Kafka doesn't contain topic: %v", requiredTopic)) return "", ErrComponentStart.Wrap(fmt.Sprintf("Kafka doesn't contain topic: %v", requiredTopic))
} }
} }
return nil return str, nil
} }
// isTopicPresent checks if a topic is present in the list of topics // isTopicPresent checks if a topic is present in the list of topics

Loading…
Cancel
Save