Add support for bulk listener WAF configuration

- Introduced new API endpoint and service method to enable/disable WAF for multiple listeners
- Updated router, controller, and service to support bulk listener WAF operations
- Added new request type `EnableListenerWafsReq` to handle multiple listener configurations
- Implemented logic to add or remove WAF configurations for specific listeners based on input

api/router.go

@@ -11,7 +11,7 @@ import (
 	"gorm.io/gorm"
 )
 
-func SetRouters(db *gorm.DB, clusterClientManager *utils.ClusterClientManager) *gin.Engine {
+func SetRouters(db *gorm.DB, clusterClientManager *utils.ClusterClientManager, gatewayUrl string) *gin.Engine {
 	var engine *gin.Engine
 
 	if !config.Conf.Debug {
@@ -61,7 +61,7 @@ func SetRouters(db *gorm.DB, clusterClientManager *utils.ClusterClientManager) *
 	// 	},
 	// 	// BearerToken: "1234567890",
 	// })
-	SetWafRouter(engine, clusterClientManager, db)
+	SetWafRouter(engine, clusterClientManager, db, gatewayUrl)
 
 	// 统一处理 404
 	engine.NoRoute(func(c *gin.Context) {

api/waf.go

@@ -7,10 +7,10 @@ import (
 	"gorm.io/gorm"
 )
 
-func SetWafRouter(e *gin.Engine, clusterClientManager *utils.ClusterClientManager, db *gorm.DB) {
+func SetWafRouter(e *gin.Engine, clusterClientManager *utils.ClusterClientManager, db *gorm.DB, gatewayUrl string) {
 	v1 := e.Group("v1/api/waf")
 
-	wafController := controller.NewWafController(clusterClientManager, db)
+	wafController := controller.NewWafController(clusterClientManager, db, gatewayUrl)
 	v1.GET("/:region_code/:namespace/:gateway_name", wafController.Waf)
 	// v1.POST("/gateway", wafController.GetWafGatewayInfo)
 	v1.POST("/", wafController.CreateWaf)
@@ -18,6 +18,7 @@ func SetWafRouter(e *gin.Engine, clusterClientManager *utils.ClusterClientManage
 	v1.PUT("rules", wafController.UpdateRule)
 	v1.PUT("listener/enable", wafController.EnableListenerWaf)
 	v1.PUT("gateway/enable", wafController.EnableGatewayWaf)
+	v1.PUT("listeners/enable", wafController.EnableListenerWafs)
 	v1.DELETE("listener/:region_code/:namespace/:gateway_name/:port", wafController.DeleteListenerWaf)
 	v1.DELETE("gateway/:region_code/:namespace/:gateway_name", wafController.DeleteGatewayWaf)
 	v1.POST("debug/savecatagory", wafController.SaveRuleCategoryToDB)

cmd/app/cmd.go

@@ -77,7 +77,7 @@ func NewRootCommand() *cobra.Command {
 				clusterClientManager.AddClient(regionConfig.RegionCode, client)
 			}
 
-			e := api.SetRouters(db, clusterClientManager)
+			e := api.SetRouters(db, clusterClientManager, config.GatewayUrl)
 			return e.Run(":8080")
 		},
 	}

cmd/app/config.go

@@ -19,6 +19,7 @@ type Config struct {
 	DBConfig      *DBConfig      `yaml:"db_config"`
 	RegionConfigs []RegionConfig `yaml:"region_configs"`
 	Debug         bool           `yaml:"debug"`
+	GatewayUrl    string         `yaml:"gateway_url"`
 }
 
 type DBConfig struct {

go.mod

@@ -52,7 +52,12 @@ require (
 	sigs.k8s.io/yaml v1.3.0 // indirect
 )
 
-require github.com/inconshreveable/mousetrap v1.1.0 // indirect
+require (
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
+	github.com/segmentio/kafka-go v0.4.40 // indirect
+)
 
 require (
 	github.com/bytedance/sonic v1.12.1 // indirect

go.sum

@@ -122,6 +122,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
+github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.8 h1:+StwCXwm9PdpiEkPyzBXIy+M9KUb4ODm0Zarf1kS5BM=
 github.com/klauspost/cpuid/v2 v2.2.8/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
@@ -167,6 +169,8 @@ github.com/onsi/gomega v0.0.0-20170829124025-dcabb60a477c/go.mod h1:C1qb7wdrVGGV
 github.com/onsi/gomega v1.27.4 h1:Z2AnStgsdSayCMDiCU42qIz+HLqEPcgiOCXjAU/w+8E=
 github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6Wq+LM=
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
+github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
+github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -177,6 +181,8 @@ github.com/rs/xid v1.4.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
 github.com/rs/zerolog v1.29.1 h1:cO+d60CHkknCbvzEWxP0S9K6KqyTjrCNUy1LdQLCGPc=
 github.com/rs/zerolog v1.29.1/go.mod h1:Le6ESbR7hc+DP6Lt1THiV8CQSdkkNrd3R0XbEgp3ZBU=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/segmentio/kafka-go v0.4.40 h1:sszW7c0/uyv7+VcTW5trx2ZC7kMWDTxuR/6Zn8U1bm8=
+github.com/segmentio/kafka-go v0.4.40/go.mod h1:naFEZc5MQKdeL3W6NkZIAn48Y6AazqjRFDhnXeg3h94=
 github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM=
 github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y=
 github.com/spf13/pflag v0.0.0-20170130214245-9ff6c6923cff/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
@@ -202,14 +208,19 @@ github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS
 github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
+github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 golang.org/x/arch v0.9.0 h1:ub9TgUInamJ8mrZIGlBG6/4TqWeMszd4N8lNorbrr6k=
 golang.org/x/arch v0.9.0/go.mod h1:FEVrYAQjsQXMVJ1nsMoVVXPZg6p2JE2mx8psSWTDQys=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20190611184440-5c40567a22f8/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw=
 golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
@@ -218,6 +229,7 @@ golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvx
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.17.0 h1:zY54UmvipHiNd+pm+m0x9KhZ9hl1/7QNMyxXbc6ICqA=
 golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -233,6 +245,8 @@ golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLL
 golang.org/x/net v0.0.0-20200324143707-d3edc9973b7e/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
 golang.org/x/net v0.0.0-20201110031124-69a78807bb2b/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
+golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
+golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.28.0 h1:a9JDOJc5GMUJ0+UDqmLT86WiEy7iWyIhz8gz8E4e5hE=
 golang.org/x/net v0.28.0/go.mod h1:yqtgsTWOOnlGLG9GFRrK3++bGOUEkNBoHZc8MEDWPNg=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -243,6 +257,7 @@ golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.10.0 h1:3NQrjDixjgGwUOCaF8w2+VYHv0Ve/vGYSbdkTa98gmQ=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
@@ -250,19 +265,27 @@ golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20190616124812-15dcb6c0061f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.24.0 h1:Twjiwq9dn6R1fQcyiK+wQyHWfaz/BJB+YIpzU/Cv3Xg=
 golang.org/x/sys v0.24.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
+golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU=
 golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
+golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
+golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
 golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4=
@@ -277,6 +300,7 @@ golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20200505023115-26f46d2f7ef8/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
+golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d h1:vU5i/LfpvrRCpgM/VPfJLg5KjxD3E+hfT1SH+d9zLwg=
 golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

internal/controller/waf.go

@@ -16,9 +16,9 @@ type WafController struct {
 	service service.Service
 }
 
-func NewWafController(clusterClientManager *utils.ClusterClientManager, db *gorm.DB) *WafController {
+func NewWafController(clusterClientManager *utils.ClusterClientManager, db *gorm.DB, gatewayUrl string) *WafController {
 	return &WafController{
-		service: service.NewWafService(clusterClientManager, db),
+		service: service.NewWafService(clusterClientManager, db, gatewayUrl),
 	}
 }
 
@@ -183,6 +183,24 @@ func (c *WafController) EnableGatewayWaf(ctx *gin.Context) {
 	utils.AssembleResponse(ctx, nil, nil)
 }
 
+func (c *WafController) EnableListenerWafs(ctx *gin.Context) {
+	ctx1, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	var req service.EnableListenerWafsReq
+	if err := ctx.BindJSON(&req); err != nil {
+		utils.AssembleResponse(ctx, nil, err)
+		return
+	}
+
+	err := c.service.EnableListenerWafs(ctx1, &req)
+	if err != nil {
+		utils.AssembleResponse(ctx, nil, err)
+		return
+	}
+	utils.AssembleResponse(ctx, nil, nil)
+}
+
 func (c *WafController) DeleteListenerWaf(ctx *gin.Context) {
 	ctx1, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()

internal/service/log_consumer.go

+package service
+
+import (
+	"context"
+
+	"github.com/rs/zerolog/log"
+	"github.com/segmentio/kafka-go"
+	"gorm.io/gorm"
+)
+
+type LogConsumerService struct {
+	consumer *kafka.Reader
+	db       *gorm.DB
+}
+
+func NewLogConsumerService(consumer *kafka.Reader, db *gorm.DB) *LogConsumerService {
+	return &LogConsumerService{
+		consumer: consumer,
+		db:       db,
+	}
+}
+
+func (s *LogConsumerService) Consume() {
+	for {
+		m, err := s.consumer.ReadMessage(context.Background())
+		if err != nil {
+			log.Error().Err(err).Msg("failed to read message")
+			continue
+		}
+		log.Info().Msgf("message: %s", string(m.Value))
+		go s.processMessage(m)
+	}
+}
+
+func (s *LogConsumerService) processMessage(m kafka.Message) {
+	log.Info().Msgf("processing message: %s", string(m.Value))
+}

internal/service/service.go

@@ -14,4 +14,5 @@ type Service interface {
 	EnableGatewayWaf(ctx context.Context, req *EnableGatewayWafReq) error
 	DeleteGatewayWaf(ctx context.Context, req *GatewateInfo) error
 	DeleteListenerWaf(ctx context.Context, req *DeleteListenerReq) error
+	EnableListenerWafs(ctx context.Context, req *EnableListenerWafsReq) error
 }

internal/service/types.go

@@ -228,6 +228,17 @@ type EnableGatewayWafReq struct {
 	Cookie string `json:"cookie"`
 }
 
+type ListenerWaf struct {
+	Hosts []string `json:"hosts"`
+	Port  int      `json:"port"`
+}
+
+type EnableListenerWafsReq struct {
+	GatewateInfo
+	Enable       bool          `json:"enable"`
+	ListenerWafs []ListenerWaf `json:"listener_wafs"`
+}
+
 type WafRule struct {
 	ID          int    `json:"id"`
 	Level       int    `json:"level"`

internal/service/waf.go

@@ -20,15 +20,17 @@ import (
 	"gopkg.in/yaml.v3"
 	"gorm.io/gorm"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 type wafService struct {
 	clusterClientManager *utils.ClusterClientManager
 	db                   *gorm.DB
+	gatewayUrl           string
 }
 
-func NewWafService(clusterClientManager *utils.ClusterClientManager, db *gorm.DB) Service {
-	return &wafService{clusterClientManager: clusterClientManager, db: db}
+func NewWafService(clusterClientManager *utils.ClusterClientManager, db *gorm.DB, gatewayUrl string) Service {
+	return &wafService{clusterClientManager: clusterClientManager, db: db, gatewayUrl: gatewayUrl}
 }
 
 func (s *wafService) GetWaf(ctx context.Context, regionCode, namespace, gatewayName string) (*WafService, error) {
@@ -517,7 +519,7 @@ func getGatewayNameFromCrn(crn string) string {
 	return parts[len(parts)-1]
 }
 
-func (s *wafService) ListListenerFromApiGateway(ctx context.Context, apiGatewayCrn string, regionCode string, cookie string) ([]GatewayRespListenerData, error) {
+func (s *wafService) listListenerFromApiGateway(ctx context.Context, apiGatewayCrn string, regionCode string, cookie string) ([]GatewayRespListenerData, error) {
 	body, err := json.Marshal(map[string]string{
 		"apigateway_crn": apiGatewayCrn,
 		"region_code":    regionCode,
@@ -525,7 +527,7 @@ func (s *wafService) ListListenerFromApiGateway(ctx context.Context, apiGatewayC
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal request body: %v", err)
 	}
-	request, err := http.NewRequest("POST", "https://csm.console.test.tg.unicom.local/apigatewaymng/listener/lf-tst7/list_listeners", bytes.NewBuffer(body))
+	request, err := http.NewRequestWithContext(ctx, "POST", "https://csm.console.test.tg.unicom.local/apigatewaymng/listener/lf-tst7/list_listeners", bytes.NewBuffer(body))
 	if err != nil {
 		return nil, fmt.Errorf("failed to create request: %v", err)
 	}
@@ -556,7 +558,7 @@ func (s *wafService) ListListenerFromApiGateway(ctx context.Context, apiGatewayC
 
 func (s *wafService) EnableGatewayWaf(ctx context.Context, req *EnableGatewayWafReq) error {
 	if req.Enable {
-		listeners, err := s.ListListenerFromApiGateway(ctx, req.ApiGatewayCrn, req.RegionCode, req.Cookie)
+		listeners, err := s.listListenerFromApiGateway(ctx, req.ApiGatewayCrn, req.RegionCode, req.Cookie)
 		if err != nil {
 			return fmt.Errorf("failed to get listener list: %v", err)
 		}
@@ -671,3 +673,72 @@ func (s *wafService) ListListenerWafStatus(ctx context.Context, req *GatewateInf
 
 	return listenerStatusList, nil
 }
+
+func (s *wafService) EnableListenerWafs(ctx context.Context, req *EnableListenerWafsReq) error {
+
+	client := s.clusterClientManager.GetClient(req.RegionCode)
+	if client == nil {
+		return fmt.Errorf("failed to get cluster client")
+	}
+
+	listenerList, err := client.WafV1alpha1().Services(req.Namespace).List(ctx, metav1.ListOptions{LabelSelector: fmt.Sprintf("apigateway_name=%s", req.GatewayName)})
+	if err != nil {
+		return err
+	}
+
+	portList := []int{}
+	for _, listener := range listenerList.Items {
+		n := strings.LastIndex(listener.Name, "-")
+		if n == -1 {
+			return fmt.Errorf("failed to get listener port: %v", listener.Name)
+		}
+		listenerPort := listener.Name[n+1:]
+		listenerPortInt, err := strconv.Atoi(listenerPort)
+		if err != nil {
+			return fmt.Errorf("failed to parse listener port: %v", err)
+		}
+		portList = append(portList, listenerPortInt)
+	}
+	currentPortSet := sets.NewInt(portList...)
+
+	desiredPortSet := sets.NewInt()
+	wafMap := map[int]ListenerWaf{}
+	for _, listener := range req.ListenerWafs {
+		desiredPortSet.Insert(listener.Port)
+		wafMap[listener.Port] = listener
+	}
+
+	// enable WAF for ports that are in the desired port set but not in the current port set
+	addingPortSet := desiredPortSet.Difference(currentPortSet)
+	for _, port := range addingPortSet.List() {
+		err := s.EnableListenerWaf(ctx, &EnableListenerWafReq{
+			GatewateInfo: GatewateInfo{
+				GatewayName: req.GatewayName,
+				Namespace:   req.Namespace,
+				RegionCode:  req.RegionCode,
+			},
+			Port:  port,
+			Hosts: wafMap[port].Hosts,
+		})
+		if err != nil {
+			return fmt.Errorf("failed to enable listener WAF: %v", err)
+		}
+	}
+
+	// delete WAF for ports that are not in the desired port set
+	deletingPortSet := currentPortSet.Difference(desiredPortSet)
+	for _, port := range deletingPortSet.List() {
+		err := s.DeleteListenerWaf(ctx, &DeleteListenerReq{
+			GatewateInfo: GatewateInfo{
+				GatewayName: req.GatewayName,
+				Namespace:   req.Namespace,
+				RegionCode:  req.RegionCode,
+			},
+			Port: port,
+		})
+		if err != nil {
+			return fmt.Errorf("failed to delete listener WAF: %v", err)
+		}
+	}
+	return nil
+}