refactor(publisher): load the worker configurations from one same file (#193)

Signed-off-by: wuhuizuo <wuhuizuo@126.com>

Signed-off-by: wuhuizuo <wuhuizuo@126.com>

Author: wuhuizuo

publisher/README.md

@@ -33,3 +33,10 @@ go run ./cmd/publisher -config=config-publisher.yaml --debug --domain 0.0.0.0:80
 ```bash
 go run ./cmd/worker --config=config-worker.yaml --debug
 ```
+
+### Test with the client CLI
+
+```bash
+go run ./cmd/publisher-cli -url=http://localhost:8080 tiup request-to-publish -body '{ "artifact_url": "hub.pingcap.net/pingcap/tidb/package:master_linux_amd64" }'
+go run ./cmd/publisher-cli -url=http://localhost:8080 fileserver request-to-publish -body '{ "artifact_url": "hub.pingcap.net/pingcap/tidb/package:master_linux_amd64" }'
+```

publisher/cmd/worker/init.go

@@ -15,24 +15,23 @@ import (
 )
 
 // Load and parse configuration.
-func loadConfig(configFile string) (config.Worker, error) {
-	var config config.Worker
-	{
-		configData, err := os.ReadFile(configFile)
-		if err != nil {
-			return config, fmt.Errorf("error reading config file: %v", err)
-		}
-		if err := yaml.Unmarshal(configData, &config); err != nil {
-			return config, fmt.Errorf("error parsing config file: %v", err)
-		}
+func loadConfig(configFile string) (*config.Workers, error) {
+	configData, err := os.ReadFile(configFile)
+	if err != nil {
+		return nil, fmt.Errorf("error reading config file: %v", err)
+	}
+
+	var config config.Workers
+	if err := yaml.Unmarshal(configData, &config); err != nil {
+		return nil, fmt.Errorf("error parsing config file: %v", err)
 	}
-	return config, nil
+
+	return &config, nil
 }
 
-func initTiupWorkerFromConfig(configFile string) (*kafka.Reader, impl.Worker) {
-	config, err := loadConfig(configFile)
-	if err != nil {
-		log.Fatal().Err(err).Msg("load config failed")
+func initTiupWorkerFromConfig(config *config.Worker) (*kafka.Reader, impl.Worker) {
+	if config == nil {
+		return nil, nil
 	}
 
 	// Configure Redis client.
@@ -61,10 +60,9 @@ func initTiupWorkerFromConfig(configFile string) (*kafka.Reader, impl.Worker) {
 	return kafkaReader, worker
 }
 
-func initFsWorkerFromConfig(configFile string) (*kafka.Reader, impl.Worker) {
-	config, err := loadConfig(configFile)
-	if err != nil {
-		log.Fatal().Err(err).Msg("load config failed")
+func initFsWorkerFromConfig(config *config.Worker) (*kafka.Reader, impl.Worker) {
+	if config == nil {
+		return nil, nil
 	}
 
 	// Configure Redis client.
@@ -75,12 +73,9 @@ func initFsWorkerFromConfig(configFile string) (*kafka.Reader, impl.Worker) {
 		DB:       config.Redis.DB,
 	})
 
-	var worker impl.Worker
-	{
-		worker, err = impl.NewFsWorker(&log.Logger, redisClient, config.Options)
-		if err != nil {
-			log.Fatal().Err(err).Msg("Error creating tiup publishing worker")
-		}
+	worker, err := impl.NewFsWorker(&log.Logger, redisClient, config.Options)
+	if err != nil {
+		log.Fatal().Err(err).Msg("Error creating tiup publishing worker")
 	}
 
 	kafkaReader := kafka.NewReader(kafka.ReaderConfig{

publisher/cmd/worker/main.go

@@ -21,9 +21,8 @@ import (
 func main() {
 	// Parse command-line flags
 	var (
-		tiupConfigFile = flag.String("tiup-config", "config.yaml", "Path to config file")
-		fsConfigFile   = flag.String("fs-config", "config.yaml", "Path to config file")
-		dbgF           = flag.Bool("debug", false, "Enable debug mode")
+		configFile = flag.String("config", "config.yaml", "Path to config file")
+		dbgF       = flag.Bool("debug", false, "Enable debug mode")
 	)
 	flag.Parse()
 
@@ -35,8 +34,13 @@ func main() {
 		zerolog.SetGlobalLevel(zerolog.InfoLevel)
 	}
 
-	tiupPublishRequestKafkaReader, tiupWorker := initTiupWorkerFromConfig(*tiupConfigFile)
-	fsPublishRequestKafkaReader, fsWorker := initFsWorkerFromConfig(*fsConfigFile)
+	config, err := loadConfig(*configFile)
+	if err != nil {
+		log.Fatal().Err(err).Msg("load config failed")
+	}
+
+	tiupPublishRequestKafkaReader, tiupWorker := initTiupWorkerFromConfig(config.Tiup)
+	fsPublishRequestKafkaReader, fsWorker := initFsWorkerFromConfig(config.FileServer)
 
 	// Create channel used by both the signal handler and server goroutines
 	// to notify the main goroutine when to stop the server.
@@ -65,6 +69,15 @@ func main() {
 }
 
 func startWorker(ctx context.Context, wg *sync.WaitGroup, reader *kafka.Reader, worker impl.Worker) {
+	if reader == nil {
+		log.Warn().Msg("empty kafka reader, skip")
+		return
+	}
+	if worker == nil {
+		log.Warn().Msg("empty worker, skip")
+		return
+	}
+
 	(*wg).Add(1)
 	go func() {
 		defer (*wg).Done()

publisher/example/config/worker-example.fileserver.yaml

@@ -0,0 +1,36 @@
+tiup:
+  kafka:
+    brokers:
+      - example-bootstrap.kafka:9092
+    topic: example-topic
+    consumer_group: example-group
+
+  redis:
+    addr: "redis-server:6379"
+    db: 0
+    password: "redis_password"  
+
+  options:
+    mirror_url: http://tiup.mirror.site
+    lark_webhook_url: https://feishu.custom-bot-webhook # create and copy the url then paste here.
+    nightly_interval: 1h
+
+file_server:
+  kafka:
+    brokers:
+      - example-bootstrap.kafka:9092
+    topic: example-topic
+    consumer_group: example-group-fs
+
+  redis:
+    addr: "redis-server:6379"
+    db: 0
+    password: "redis_password"  
+
+  options:
+    lark_webhook_url: https://feishu.custom-bot-webhook # create and copy the url then paste here.
+    s3.endpoint: <endpoint>
+    s3.region: BEIJING
+    s3.bucket_name: <bucket-name>
+    s3.access_key: <access-key>
+    s3.secret_key: <secret-key>

publisher/example/config/worker-example.tiup.yaml

@@ -1,5 +1,10 @@
 package config
 
+type Workers struct {
+	Tiup       *Worker `yaml:"tiup,omitempty" json:"tiup,omitempty"`
+	FileServer *Worker `yaml:"file_server,omitempty" json:"file_server,omitempty"`
+}
+
 type Worker struct {
 	Kafka struct {
 		KafkaBasic    `yaml:",inline" json:",inline"`