|
|
|
@ -15,44 +15,26 @@
|
|
|
|
package main
|
|
|
|
package main
|
|
|
|
|
|
|
|
|
|
|
|
import (
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
|
|
|
|
"errors"
|
|
|
|
"errors"
|
|
|
|
"flag"
|
|
|
|
"flag"
|
|
|
|
"fmt"
|
|
|
|
"fmt"
|
|
|
|
"net"
|
|
|
|
|
|
|
|
"net/url"
|
|
|
|
|
|
|
|
"os"
|
|
|
|
"os"
|
|
|
|
"strings"
|
|
|
|
"strings"
|
|
|
|
"time"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
|
|
|
|
"github.com/IBM/sarama"
|
|
|
|
"github.com/OpenIMSDK/tools/component"
|
|
|
|
"github.com/OpenIMSDK/tools/errs"
|
|
|
|
"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/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"
|
|
|
|
"gopkg.in/yaml.v3"
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
const (
|
|
|
|
const (
|
|
|
|
// defaultCfgPath is the default path of the configuration file.
|
|
|
|
// defaultCfgPath is the default path of the configuration file.
|
|
|
|
defaultCfgPath = "../../../../../config/config.yaml"
|
|
|
|
defaultCfgPath = "../../../../../config/config.yaml"
|
|
|
|
minioHealthCheckDuration = 1
|
|
|
|
|
|
|
|
maxRetry = 300
|
|
|
|
maxRetry = 300
|
|
|
|
componentStartErrCode = 6000
|
|
|
|
componentStartErrCode = 6000
|
|
|
|
configErrCode = 6001
|
|
|
|
configErrCode = 6001
|
|
|
|
mongoConnTimeout = 30 * time.Second
|
|
|
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
const (
|
|
|
|
|
|
|
|
colorRed = 31
|
|
|
|
|
|
|
|
colorGreen = 32
|
|
|
|
|
|
|
|
colorYellow = 33
|
|
|
|
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
var (
|
|
|
|
var (
|
|
|
|
@ -103,16 +85,16 @@ func main() {
|
|
|
|
for _, check := range checks {
|
|
|
|
for _, check := range checks {
|
|
|
|
str, err := check.function()
|
|
|
|
str, err := check.function()
|
|
|
|
if err != nil {
|
|
|
|
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
|
|
|
|
allSuccess = false
|
|
|
|
break
|
|
|
|
break
|
|
|
|
} else {
|
|
|
|
} else {
|
|
|
|
successPrint(fmt.Sprintf("%s connected successfully, %s", check.name, str))
|
|
|
|
component.SuccessPrint(fmt.Sprintf("%s connected successfully, %s", check.name, str))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
if allSuccess {
|
|
|
|
if allSuccess {
|
|
|
|
successPrint("All components started successfully!")
|
|
|
|
component.SuccessPrint("All components started successfully!")
|
|
|
|
|
|
|
|
|
|
|
|
return
|
|
|
|
return
|
|
|
|
}
|
|
|
|
}
|
|
|
|
@ -120,19 +102,6 @@ func main() {
|
|
|
|
os.Exit(1)
|
|
|
|
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
|
|
|
|
// Helper function to get environment variable or default value
|
|
|
|
func getEnv(key, fallback string) string {
|
|
|
|
func getEnv(key, fallback string) string {
|
|
|
|
if value, exists := os.LookupEnv(key); exists {
|
|
|
|
if value, exists := os.LookupEnv(key); exists {
|
|
|
|
@ -143,45 +112,23 @@ func getEnv(key, fallback string) string {
|
|
|
|
|
|
|
|
|
|
|
|
// checkMongo checks the MongoDB connection without retries
|
|
|
|
// checkMongo checks the MongoDB connection without retries
|
|
|
|
func checkMongo() (string, error) {
|
|
|
|
func checkMongo() (string, error) {
|
|
|
|
uri := getEnv("MONGO_URI", buildMongoURI())
|
|
|
|
mongo := &component.Mongo{
|
|
|
|
|
|
|
|
Address: config.Config.Mongo.Address,
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), mongoConnTimeout)
|
|
|
|
Database: config.Config.Mongo.Database,
|
|
|
|
defer cancel()
|
|
|
|
Username: config.Config.Mongo.Username,
|
|
|
|
|
|
|
|
Password: config.Config.Mongo.Password,
|
|
|
|
str := "ths addr is:" + strings.Join(config.Config.Mongo.Address, ",")
|
|
|
|
MaxPoolSize: config.Config.Mongo.MaxPoolSize,
|
|
|
|
|
|
|
|
|
|
|
|
client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
defer client.Disconnect(context.Background())
|
|
|
|
uri, uriExist := os.LookupEnv("MONGO_URI")
|
|
|
|
|
|
|
|
if uriExist {
|
|
|
|
ctx, cancel = context.WithTimeout(context.Background(), mongoConnTimeout)
|
|
|
|
mongo.URL = uri
|
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if err = client.Ping(ctx, nil); err != nil {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
return str, nil
|
|
|
|
str, err := component.CheckMongo(mongo)
|
|
|
|
}
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return "", err
|
|
|
|
// 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)
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return fmt.Sprintf("mongodb://%s/%s?maxPoolSize=%d",
|
|
|
|
return str, nil
|
|
|
|
mongodbHosts, database, maxPoolSize)
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// checkMinio checks the MinIO connection
|
|
|
|
// checkMinio checks the MinIO connection
|
|
|
|
@ -193,50 +140,29 @@ func checkMinio() (string, error) {
|
|
|
|
|
|
|
|
|
|
|
|
// Prioritize environment variables
|
|
|
|
// Prioritize environment variables
|
|
|
|
endpoint := getEnv("MINIO_ENDPOINT", config.Config.Object.Minio.Endpoint)
|
|
|
|
endpoint := getEnv("MINIO_ENDPOINT", config.Config.Object.Minio.Endpoint)
|
|
|
|
accessKeyID := getEnv("MINIO_ACCESS_KEY_ID", config.Config.Object.Minio.AccessKeyID)
|
|
|
|
address, addressExist := os.LookupEnv("MINIO_ADDRESS")
|
|
|
|
secretAccessKey := getEnv("MINIO_SECRET_ACCESS_KEY", config.Config.Object.Minio.SecretAccessKey)
|
|
|
|
port, portExist := os.LookupEnv("MINIO_PORT")
|
|
|
|
useSSL := getEnv("MINIO_USE_SSL", "false") // Assuming SSL is not used by default
|
|
|
|
if portExist && addressExist {
|
|
|
|
|
|
|
|
endpoint = "http://" + address + ":" + port
|
|
|
|
if endpoint == "" || accessKeyID == "" || secretAccessKey == "" {
|
|
|
|
} else if !portExist && addressExist {
|
|
|
|
return "", ErrConfig.Wrap("MinIO configuration missing")
|
|
|
|
return "", errs.Wrap(errors.New("the MINIO_ADDRESS of minio is empty"))
|
|
|
|
}
|
|
|
|
} else if portExist && !addressExist {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errors.New("the MINIO_PORT of minio is empty"))
|
|
|
|
// Parse endpoint URL to determine if SSL is enabled
|
|
|
|
}
|
|
|
|
u, err := url.Parse(endpoint)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
minio := &component.Minio{
|
|
|
|
str := "the endpoint is:" + endpoint
|
|
|
|
ApiURL: config.Config.Object.ApiURL,
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
Endpoint: endpoint,
|
|
|
|
}
|
|
|
|
AccessKeyID: getEnv("MINIO_ACCESS_KEY_ID", config.Config.Object.Minio.AccessKeyID),
|
|
|
|
secure := u.Scheme == "https" || useSSL == "true"
|
|
|
|
SecretAccessKey: getEnv("MINIO_SECRET_ACCESS_KEY", config.Config.Object.Minio.SecretAccessKey),
|
|
|
|
|
|
|
|
SignEndpoint: config.Config.Object.Minio.SignEndpoint,
|
|
|
|
// Initialize MinIO client
|
|
|
|
UseSSL: getEnv("MINIO_USE_SSL", "false"),
|
|
|
|
minioClient, err := minio.New(u.Host, &minio.Options{
|
|
|
|
}
|
|
|
|
Creds: credentials.NewStaticV4(accessKeyID, secretAccessKey, ""),
|
|
|
|
|
|
|
|
Secure: secure,
|
|
|
|
str, err := component.CheckMinio(minio)
|
|
|
|
})
|
|
|
|
|
|
|
|
str := "ths addr is:" + u.Host
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
strs := fmt.Sprintf("%v;host:%s,accessKeyID:%s,secretAccessKey:%s,Secure:%v", err, u.Host, accessKeyID, secretAccessKey, secure)
|
|
|
|
return "", err
|
|
|
|
return "", errs.Wrap(err, strs)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Perform health check
|
|
|
|
|
|
|
|
cancel, err := minioClient.HealthCheck(time.Duration(minioHealthCheckDuration) * time.Second)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
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")
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
return str, nil
|
|
|
|
return str, nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@ -247,76 +173,65 @@ func checkRedis() (string, error) {
|
|
|
|
username := getEnv("REDIS_USERNAME", config.Config.Redis.Username)
|
|
|
|
username := getEnv("REDIS_USERNAME", config.Config.Redis.Username)
|
|
|
|
password := getEnv("REDIS_PASSWORD", config.Config.Redis.Password)
|
|
|
|
password := getEnv("REDIS_PASSWORD", config.Config.Redis.Password)
|
|
|
|
|
|
|
|
|
|
|
|
// Split address to handle multiple addresses for cluster setup
|
|
|
|
redis := &component.Redis{
|
|
|
|
redisAddresses := strings.Split(address, ",")
|
|
|
|
Address: 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,
|
|
|
|
Username: username,
|
|
|
|
Password: password,
|
|
|
|
Password: password,
|
|
|
|
})
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
|
|
|
// Use regular client for single address
|
|
|
|
|
|
|
|
redisClient = redis.NewClient(&redis.Options{
|
|
|
|
|
|
|
|
Addr: redisAddresses[0],
|
|
|
|
|
|
|
|
Username: username,
|
|
|
|
|
|
|
|
Password: password,
|
|
|
|
|
|
|
|
})
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
defer redisClient.Close()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Ping Redis to check connectivity
|
|
|
|
addr, addrExist := os.LookupEnv("REDIS_ADDRESS")
|
|
|
|
_, err := redisClient.Ping(context.Background()).Result()
|
|
|
|
port, portExist := os.LookupEnv("REDIS_PORT")
|
|
|
|
str := "the addr is:" + strings.Join(redisAddresses, ",")
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
if addrExist && portExist {
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
addresses := strings.Split(addr, ",")
|
|
|
|
|
|
|
|
for i, address := range addresses {
|
|
|
|
|
|
|
|
addresses[i] = address + ":" + port
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
redis.Address = addresses
|
|
|
|
|
|
|
|
} else if !addrExist && portExist {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errors.New("the REDIS_ADDRESS of minio is empty"))
|
|
|
|
|
|
|
|
} else if addrExist && !portExist {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errors.New("the REDIS_PORT of minio is empty"))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
str, err := component.CheckRedis(redis)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return "", err
|
|
|
|
|
|
|
|
}
|
|
|
|
return str, nil
|
|
|
|
return str, nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// checkZookeeper checks the Zookeeper connection
|
|
|
|
// checkZookeeper checks the Zookeeper connection
|
|
|
|
func checkZookeeper() (string, error) {
|
|
|
|
func checkZookeeper() (string, error) {
|
|
|
|
// Prioritize environment variables
|
|
|
|
// 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
|
|
|
|
address := getEnv("ZOOKEEPER_USERNAME", config.Config.Zookeeper.Username)
|
|
|
|
zookeeperAddresses := strings.Split(address, ",")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Connect to Zookeeper
|
|
|
|
zk := &component.Zookeeper{
|
|
|
|
str := "the addr is:" + address
|
|
|
|
Schema: getEnv("ZOOKEEPER_SCHEMA", "digest"),
|
|
|
|
c, eventChan, err := zk.Connect(zookeeperAddresses, time.Second) // Adjust the timeout as necessary
|
|
|
|
ZkAddr: strings.Split(address, ","),
|
|
|
|
if err != nil {
|
|
|
|
Username: getEnv("ZOOKEEPER_USERNAME", config.Config.Zookeeper.Username),
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
Password: getEnv("ZOOKEEPER_PASSWORD", config.Config.Zookeeper.Password),
|
|
|
|
}
|
|
|
|
|
|
|
|
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()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Set authentication if username and password are provided
|
|
|
|
address, addrExist := os.LookupEnv("ZOOKEEPER_ADDRESS")
|
|
|
|
if username != "" && password != "" {
|
|
|
|
port, portExist := os.LookupEnv("ZOOKEEPER_PORT")
|
|
|
|
if err := c.AddAuth(schema, []byte(username+":"+password)); err != nil {
|
|
|
|
if addrExist && portExist {
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
addresses := strings.Split(address, ",")
|
|
|
|
|
|
|
|
for i, addr := range addresses {
|
|
|
|
|
|
|
|
addresses[i] = addr + ":" + port
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
zk.ZkAddr = addresses
|
|
|
|
|
|
|
|
} else if !addrExist && portExist {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errors.New("the ZOOKEEPER_ADDRESS of minio is empty"))
|
|
|
|
|
|
|
|
} else if addrExist && !portExist {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errors.New("the ZOOKEEPER_PORT of minio is empty"))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
str, err := component.CheckZookeeper(zk)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
|
|
|
|
return "", err
|
|
|
|
|
|
|
|
}
|
|
|
|
return str, nil
|
|
|
|
return str, nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@ -327,26 +242,30 @@ func checkKafka() (string, error) {
|
|
|
|
password := getEnv("KAFKA_PASSWORD", config.Config.Kafka.Password)
|
|
|
|
password := getEnv("KAFKA_PASSWORD", config.Config.Kafka.Password)
|
|
|
|
address := getEnv("KAFKA_ADDRESS", strings.Join(config.Config.Kafka.Addr, ","))
|
|
|
|
address := getEnv("KAFKA_ADDRESS", strings.Join(config.Config.Kafka.Addr, ","))
|
|
|
|
|
|
|
|
|
|
|
|
// Split addresses to handle multiple Kafka brokers
|
|
|
|
kafka := &component.Kafka{
|
|
|
|
kafkaAddresses := strings.Split(address, ",")
|
|
|
|
Username: username,
|
|
|
|
|
|
|
|
Password: password,
|
|
|
|
|
|
|
|
Addr: strings.Split(address, ","),
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
// Configure Kafka client
|
|
|
|
address, addrExist := os.LookupEnv("KAFKA_ADDRESS")
|
|
|
|
cfg := sarama.NewConfig()
|
|
|
|
port, portExist := os.LookupEnv("KAFKA_PORT")
|
|
|
|
if username != "" && password != "" {
|
|
|
|
if addrExist && portExist {
|
|
|
|
cfg.Net.SASL.Enable = true
|
|
|
|
addresses := strings.Split(address, ",")
|
|
|
|
cfg.Net.SASL.User = username
|
|
|
|
for i, addr := range addresses {
|
|
|
|
cfg.Net.SASL.Password = password
|
|
|
|
addresses[i] = addr + ":" + port
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
kafka.Addr = addresses
|
|
|
|
|
|
|
|
} else if !addrExist && portExist {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errors.New("the KAFKA_ADDRESS of minio is empty"))
|
|
|
|
|
|
|
|
} else if addrExist && !portExist {
|
|
|
|
|
|
|
|
return "", errs.Wrap(errors.New("the KAFKA_PORT of minio is empty"))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
// Additional Kafka setup (e.g., TLS configuration) can be added here
|
|
|
|
|
|
|
|
// kafka.SetupTLSConfig(cfg)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Create Kafka client
|
|
|
|
str, kafkaClient, err := component.CheckKafka(kafka)
|
|
|
|
str := "the addr is:" + address
|
|
|
|
|
|
|
|
kafkaClient, err := sarama.NewClient(kafkaAddresses, cfg)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
if err != nil {
|
|
|
|
return "", errs.Wrap(errStr(err, str))
|
|
|
|
return "", err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
defer kafkaClient.Close()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
// Verify if necessary topics exist
|
|
|
|
// Verify if necessary topics exist
|
|
|
|
topics, err := kafkaClient.Topics()
|
|
|
|
topics, err := kafkaClient.Topics()
|
|
|
|
@ -378,23 +297,3 @@ func isTopicPresent(topic string, topics []string) bool {
|
|
|
|
}
|
|
|
|
}
|
|
|
|
return false
|
|
|
|
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 warningPrint(s string) {
|
|
|
|
|
|
|
|
colorPrint(colorYellow, "Warning: But %v", s)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func errStr(err error, str string) error {
|
|
|
|
|
|
|
|
return fmt.Errorf("%v;%s", err, str)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
luhaoling
2 years ago