Add rootCAs support to Kafka & MQTT (#8236)

Fixes #8211

cmd/config-current.go

@@ -341,6 +341,9 @@ func (s *serverConfig) TestNotificationTargets() error {
 		if !v.Enable {
 			continue
 		}
+		if v.TLS.Enable {
+			v.TLS.RootCAs = globalRootCAs
+		}
 		t, err := target.NewKafkaTarget(k, v, GlobalServiceDoneCh)
 		if err != nil {
 			return fmt.Errorf("kafka(%s): %s", k, err.Error())
@@ -352,6 +355,7 @@ func (s *serverConfig) TestNotificationTargets() error {
 		if !v.Enable {
 			continue
 		}
+		v.RootCAs = globalRootCAs
 		t, err := target.NewMQTTTarget(k, v, GlobalServiceDoneCh)
 		if err != nil {
 			return fmt.Errorf("mqtt(%s): %s", k, err.Error())
@@ -690,6 +694,9 @@ func getNotificationTargets(config *serverConfig) *event.TargetList {
 
 	for id, args := range config.Notify.Kafka {
 		if args.Enable {
+			if args.TLS.Enable {
+				args.TLS.RootCAs = globalRootCAs
+			}
 			newTarget, err := target.NewKafkaTarget(id, args, GlobalServiceDoneCh)
 			if err != nil {
 				logger.LogIf(context.Background(), err)

pkg/event/target/kafka.go

@@ -18,6 +18,7 @@ package target
 
 import (
 	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"errors"
 	"net"
@@ -40,6 +41,7 @@ type KafkaArgs struct {
 	QueueLimit uint64      `json:"queueLimit"`
 	TLS        struct {
 		Enable     bool               `json:"enable"`
+		RootCAs    *x509.CertPool     `json:"-"`
 		SkipVerify bool               `json:"skipVerify"`
 		ClientAuth tls.ClientAuthType `json:"clientAuth"`
 	} `json:"tls"`
@@ -198,7 +200,9 @@ func NewKafkaTarget(id string, args KafkaArgs, doneCh <-chan struct{}) (*KafkaTa
 
 	config.Net.TLS.Enable = args.TLS.Enable
 	tlsConfig := &tls.Config{
-		ClientAuth: args.TLS.ClientAuth,
+		ClientAuth:         args.TLS.ClientAuth,
+		InsecureSkipVerify: args.TLS.SkipVerify,
+		RootCAs:            args.TLS.RootCAs,
 	}
 	config.Net.TLS.Config = tlsConfig