diff --git a/pkg/hostman/hostinfo/hostinfo.go b/pkg/hostman/hostinfo/hostinfo.go index 5851d30ffb..d97e868aa2 100644 --- a/pkg/hostman/hostinfo/hostinfo.go +++ b/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{}{ diff --git a/pkg/hostman/options/options.go b/pkg/hostman/options/options.go index 3455d0bbf7..ef66316745 100644 --- a/pkg/hostman/options/options.go +++ b/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 { diff --git a/pkg/hostman/system_service/telegraf.go b/pkg/hostman/system_service/telegraf.go index d3e1264015..48c19250f9 100644 --- a/pkg/hostman/system_service/telegraf.go +++ b/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"