|
|
|
@ -16,6 +16,7 @@ package main
|
|
|
|
|
|
|
|
|
|
|
|
import (
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"context"
|
|
|
|
|
|
|
|
"errors"
|
|
|
|
"flag"
|
|
|
|
"flag"
|
|
|
|
"fmt"
|
|
|
|
"fmt"
|
|
|
|
"net"
|
|
|
|
"net"
|
|
|
|
@ -72,7 +73,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 +102,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, %s", check.name, str))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@ -142,21 +143,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 := "ths addr is:" + strings.Join(config.Config.Mongo.Address, ",")
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return errs.Wrap(err)
|
|
|
|
return "", errs.Wrap(errStr(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(errStr(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 +180,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 +193,14 @@ 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)
|
|
|
|
str := "the endpoint is:" + endpoint
|
|
|
|
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
secure := u.Scheme == "https" || useSSL == "true"
|
|
|
|
secure := u.Scheme == "https" || useSSL == "true"
|
|
|
|
|
|
|
|
|
|
|
|
@ -206,31 +209,34 @@ func checkMinio() error {
|
|
|
|
Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
|
|
|
|
Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
|
|
|
|
Secure: secure,
|
|
|
|
Secure: secure,
|
|
|
|
})
|
|
|
|
})
|
|
|
|
|
|
|
|
str := "ths addr is:" + u.Host
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return errs.Wrap(err)
|
|
|
|
strs := fmt.Sprintf("%v;host:%s,accessKeyID:%s,secretAccessKey:%s,Secure:%v", err, u.Host, accessKeyID, secretAccessKey, secure)
|
|
|
|
|
|
|
|
return "", errs.Wrap(err, strs)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// 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(errStr(err, str))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
defer cancel()
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
|
|
|
|
if minioClient.IsOffline() {
|
|
|
|
if minioClient.IsOffline() {
|
|
|
|
return ErrComponentStart.Wrap("Minio server is offline")
|
|
|
|
str := fmt.Sprintf("Minio server is offline;%s", str)
|
|
|
|
|
|
|
|
return "", ErrComponentStart.Wrap(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 +265,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 := "the addr is:" + strings.Join(redisAddresses, ",")
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return errs.Wrap(err)
|
|
|
|
return "", errs.Wrap(errStr(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 +285,38 @@ func checkZookeeper() error {
|
|
|
|
zookeeperAddresses := strings.Split(address, ",")
|
|
|
|
zookeeperAddresses := strings.Split(address, ",")
|
|
|
|
|
|
|
|
|
|
|
|
// Connect to Zookeeper
|
|
|
|
// Connect to Zookeeper
|
|
|
|
c, _, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary
|
|
|
|
str := "the addr is:" + address
|
|
|
|
|
|
|
|
c, eventChan, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return errs.Wrap(err)
|
|
|
|
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, " "))
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
Connected:
|
|
|
|
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(errStr(err, str))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// Check if Zookeeper is reachable
|
|
|
|
return str, nil
|
|
|
|
_, _, err = c.Get("/")
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return errs.Wrap(err)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return 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 +336,17 @@ func checkKafka() error {
|
|
|
|
// kafka.SetupTLSConfig(cfg)
|
|
|
|
// kafka.SetupTLSConfig(cfg)
|
|
|
|
|
|
|
|
|
|
|
|
// Create Kafka client
|
|
|
|
// Create Kafka client
|
|
|
|
|
|
|
|
str := "the addr is:" + 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(errStr(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 +357,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
|
|
|
|
@ -373,3 +389,7 @@ func successPrint(s string) {
|
|
|
|
func warningPrint(s string) {
|
|
|
|
func warningPrint(s string) {
|
|
|
|
colorPrint(colorYellow, "Warning: But %v", s)
|
|
|
|
colorPrint(colorYellow, "Warning: But %v", s)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func errStr(err error, str string) error {
|
|
|
|
|
|
|
|
return fmt.Errorf("%v;%s", err, str)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
AndrewZuo01
2 years ago