Enhance LogConsumerService to include region name in log processing, updating...

Enhance LogConsumerService to include region name in log processing, updating configuration structure and initialization to support new regionName parameter.

cmd/api-server/app/cmd.go

@@ -46,7 +46,7 @@ func NewRootCommand() *cobra.Command {
 				ESTimeout:     10 * time.Second,
 			}, esClient)
 			esStore.Init()
-			logConsumerService := service.NewLogConsumerService(nil, esStore, config.KafkaConfig)
+			logConsumerService := service.NewLogConsumerService(nil, esStore, config.KafkaConfig, config.RegionConfig.RegionName)
 			go logConsumerService.Consume()
 			return e.Run(":8080")
 		},

cmd/api-server/config/config.go

@@ -10,11 +10,12 @@ import (
 
 type RegionConfig struct {
 	RegionCode string `json:"region_code"`
+	RegionName string `json:"region_name"`
 	ApiServer  string `json:"api_server"`
 }
 
 type Config struct {
-	RegionConfigs       []RegionConfig              `json:"region_configs"`
+	RegionConfig        RegionConfig                `json:"region_config"`
 	Debug               bool                        `json:"debug"`
 	SSOUrl              string                      `json:"sso_url"`
 	ElasticsearchConfig *config.ElasticsearchConfig `json:"elasticsearch_config"`

internal/service/log_consumer.go

@@ -112,12 +112,13 @@ var AttackClassDespMap = map[string]AttackClassDesp{
 // }
 
 type LogConsumerService struct {
-	consumer *kafka.Reader
-	db       *gorm.DB
-	esStore  *esStore.ESStore
+	consumer   *kafka.Reader
+	db         *gorm.DB
+	esStore    *esStore.ESStore
+	regionName string
 }
 
-func NewLogConsumerService(db *gorm.DB, esStore *esStore.ESStore, config *config.KafkaConfig) *LogConsumerService {
+func NewLogConsumerService(db *gorm.DB, esStore *esStore.ESStore, config *config.KafkaConfig, regionName string) *LogConsumerService {
 	// mechanism, _, err := getSASLMechanismByEnv()
 	// if err != nil {
 	// 	log.Fatal().Err(err).Msg("failed to get SASL mechanism")
@@ -140,9 +141,10 @@ func NewLogConsumerService(db *gorm.DB, esStore *esStore.ESStore, config *config
 	consumer := NewKafkaReader(config)
 
 	return &LogConsumerService{
-		consumer: consumer,
-		db:       db,
-		esStore:  esStore,
+		consumer:   consumer,
+		db:         db,
+		esStore:    esStore,
+		regionName: regionName,
 	}
 }
 
@@ -199,7 +201,7 @@ func (s *LogConsumerService) genWafDetectionSignal(wafDetectionMessage model.Waf
 			"cluster": {
 				Kind: "cluster",
 				ID:   wafDetectionMessage.WafDetectionMessageBasic.ClusterKey,
-				Name: wafDetectionMessage.WafDetectionMessageBasic.ClusterKey,
+				Name: s.regionName,
 			},
 			"namespace": {
 				Kind: "namespace",
@@ -209,7 +211,7 @@ func (s *LogConsumerService) genWafDetectionSignal(wafDetectionMessage model.Waf
 			"resource": {
 				Kind: "resource",
 				ID:   "",
-				Name: fmt.Sprintf("%s(%s)", wafDetectionMessage.WafDetectionMessageBasic.ResName, wafDetectionMessage.WafDetectionMessageBasic.ResKind),
+				Name: fmt.Sprintf("%s(%s)", wafDetectionMessage.WafDetectionMessageBasic.ResName, "Deployment"),
 			},
 		},
 		Severity: serverityFromAttackAction(attackedLog.Action),
@@ -263,7 +265,7 @@ func (s *LogConsumerService) genWafDetectionEvent(wafDetectionMessage model.WafD
 				{
 					Kind: "cluster",
 					ID:   wafDetectionMessage.WafDetectionMessageBasic.ClusterKey,
-					Name: wafDetectionMessage.WafDetectionMessageBasic.ClusterKey,
+					Name: s.regionName,
 				},
 			},
 			"namespace": {
@@ -296,7 +298,7 @@ func (s *LogConsumerService) genWafDetectionEvent(wafDetectionMessage model.WafD
 				"resource": {
 					Kind: "resource",
 					ID:   "",
-					Name: fmt.Sprintf("%s(%s)", wafDetectionMessage.WafDetectionMessageBasic.ResName, wafDetectionMessage.WafDetectionMessageBasic.ResKind),
+					Name: fmt.Sprintf("%s(%s)", wafDetectionMessage.WafDetectionMessageBasic.ResName, "Deployment"),
 				},
 			},
 		},