feature: kafka telegraf output options (#21494)

Co-authored-by: Qiu Jian <qiujian@yunionyun.com>

pkg/hostman/hostinfo/hostinfo.go

@@ -2415,6 +2415,7 @@ func (h *SHostInfo) OnCatalogChanged(catalog mcclient.KeystoneServiceCatalogV3)
 	telegraf := system_service.GetService("telegraf")
 	conf := map[string]interface{}{}
 	conf["hostname"] = h.getHostname()
+	conf["server_path"] = options.HostOptions.ServersPath
 	conf["tags"] = map[string]string{
 		"id":                                  h.HostId,
 		"host_id":                             h.HostId,
@@ -2433,8 +2434,22 @@ func (h *SHostInfo) OnCatalogChanged(catalog mcclient.KeystoneServiceCatalogV3)
 	conf["nics"] = h.getNicsTelegrafConf()
 	urls, _ := s.GetServiceURLs("kafka", defaultEndpointType)
 	if len(urls) > 0 {
-		conf["kafka"] = map[string]interface{}{"brokers": urls, "topic": "telegraf"}
+		kafkaConf := map[string]interface{}{
+			"brokers": urls,
+			"topic":   options.HostOptions.TelegrafKafkaOutputTopic,
+		}
+		if len(options.HostOptions.TelegrafKafkaOutputSaslUsername) > 0 {
+			kafkaConf["sasl_username"] = options.HostOptions.TelegrafKafkaOutputSaslUsername
+		}
+		if len(options.HostOptions.TelegrafKafkaOutputSaslPassword) > 0 {
+			kafkaConf["sasl_password"] = options.HostOptions.TelegrafKafkaOutputSaslPassword
+		}
+		if len(options.HostOptions.TelegrafKafkaOutputSaslMechanism) > 0 {
+			kafkaConf["sasl_mechanism"] = options.HostOptions.TelegrafKafkaOutputSaslMechanism
+		}
+		conf["kafka"] = kafkaConf
 	}
+
 	tsdb, _ := tsdb.GetDefaultServiceSource(s, defaultEndpointType)
 	if tsdb != nil && len(tsdb.URLs) > 0 {
 		conf[apis.SERVICE_TYPE_INFLUXDB] = map[string]interface{}{

pkg/hostman/options/options.go

@@ -53,6 +53,11 @@ type SHostBaseOptions struct {
 	ImageCacheCleanupPercentage int  `help:"The cleanup threshold ratio of image cache size v.s. total storage size" default:"12"`
 	ImageCacheCleanupOnStartup  bool `help:"Cleanup image cache on host startup" default:"false"`
 	ImageCacheCleanupDryRun     bool `help:"Dry run cleanup image cache" default:"false"`
+
+	TelegrafKafkaOutputTopic         string `json:"telegraf_kafka_output_topic" help:"telegraf kafka output topic"`
+	TelegrafKafkaOutputSaslUsername  string `json:"telegraf_kafka_output_sasl_username" help:"telegraf kafka output sasl_username"`
+	TelegrafKafkaOutputSaslPassword  string `json:"telegraf_kafka_output_sasl_password" help:"telegraf kafka output sasl_password"`
+	TelegrafKafkaOutputSaslMechanism string `json:"telegraf_kafka_output_sasl_mechanism" help:"telegraf kafka output sasl_mechanism"`
 }
 
 type SHostOptions struct {

pkg/hostman/system_service/telegraf.go

@@ -103,6 +103,38 @@ func (s *STelegraf) GetConfig(kwargs map[string]interface{}) string {
 		conf += "  timeout = \"30s\"\n"
 		conf += "\n"
 	}
+	/*
+	 *
+	 * [[outputs.kafka]]
+	 *   ## URLs of kafka brokers
+	 *   brokers = ["localhost:9092"]
+	 *   ## Kafka topic for producer messages
+	 *   topic = "telegraf"
+	 *   ## Optional SASL Config
+	 *   sasl_username = "kafka"
+	 *   sasl_password = "secret"
+	 *   ## Optional SASL:
+	 *   ## one of: OAUTHBEARER, PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI
+	 *   ## (defaults to PLAIN)
+	 *   sasl_mechanism = "PLAIN"
+	 */
+	if kafka, ok := kwargs["kafka"]; ok {
+		kafkaConf, _ := kafka.(map[string]interface{})
+		conf += "[[outputs.kafka]]\n"
+		for k := range kafkaConf {
+			if k == "brokers" {
+				brokers, _ := kafkaConf["brokers"].([]string)
+				for i := range brokers {
+					brokers[i] = fmt.Sprintf("\"%s\"", brokers[i])
+				}
+				conf += fmt.Sprintf("  brokers = [%s]\n", strings.Join(brokers, ", "))
+			} else {
+				conf += fmt.Sprintf("  %s = \"%s\"\n", k, kafkaConf[k])
+			}
+		}
+		conf += "\n"
+	}
+
 	conf += "[[inputs.cpu]]\n"
 	conf += "  percpu = false\n"
 	conf += "  totalcpu = true\n"
@@ -110,12 +142,30 @@ func (s *STelegraf) GetConfig(kwargs map[string]interface{}) string {
 	conf += "  report_active = true\n"
 	conf += "\n"
 	conf += "[[inputs.disk]]\n"
-	conf += "  ignore_mount_points = [\"/etc/telegraf\", \"/etc/hosts\", \"/etc/hostname\", \"/etc/resolv.conf\", \"/dev/termination-log\"]\n"
+	ignoreMountPoints := []string{
+		"/etc/telegraf",
+		"/etc/hosts",
+		"/etc/hostname",
+		"/etc/resolv.conf",
+		"/dev/termination-log",
+	}
+	for i := range ignoreMountPoints {
+		ignoreMountPoints[i] = fmt.Sprintf("%q", ignoreMountPoints[i])
+	}
+	ignorePathSegments := []string{
+		"/run/k3s/containerd/",
+	}
+	ignorePathSegments = append(ignorePathSegments, kwargs["server_path"].(string))
+	for i := range ignorePathSegments {
+		ignorePathSegments[i] = fmt.Sprintf("%q", ignorePathSegments[i])
+	}
+	conf += "  ignore_mount_points = [" + strings.Join(ignoreMountPoints, ", ") + "]\n"
+	conf += "  ignore_path_segments = [" + strings.Join(ignorePathSegments, ", ") + "]\n"
 	conf += "  ignore_fs = [\"tmpfs\", \"devtmpfs\", \"overlay\", \"squashfs\", \"iso9660\", \"rootfs\", \"hugetlbfs\", \"autofs\"]\n"
 	conf += "\n"
 	conf += "[[inputs.diskio]]\n"
 	conf += "  skip_serial_number = false\n"
-	conf += "  excludes = \"^nbd\"\n"
+	conf += "  excludes = \"^(nbd|loop)\"\n"
 	conf += "\n"
 	conf += "[[inputs.kernel]]\n"
 	conf += "\n"
@@ -157,6 +207,10 @@ func (s *STelegraf) GetConfig(kwargs map[string]interface{}) string {
 	}
 	conf += "[[inputs.netstat]]\n"
 	conf += "\n"
+	conf += "[[inputs.bond]]\n"
+	conf += "\n"
+	conf += "[[inputs.temp]]\n"
+	conf += "\n"
 	conf += "[[inputs.nstat]]\n"
 	conf += "\n"
 	conf += "[[inputs.ntpq]]\n"