fix(lbagent,monitor): use victoria-metrics for telegraf and fix influxql translation (#24733)

Author: zexi

go.mod

@@ -83,7 +83,7 @@ require (
 	github.com/vmihailenco/msgpack v4.0.4+incompatible
 	github.com/xuri/excelize/v2 v2.7.1
 	github.com/zeebo/xxh3 v1.0.2
-	github.com/zexi/influxql-to-metricsql v0.1.1
+	github.com/zexi/influxql-to-metricsql v0.1.2
 	go.etcd.io/etcd/api/v3 v3.5.7
 	go.etcd.io/etcd/client/v3 v3.5.7
 	golang.org/x/crypto v0.41.0

go.sum

@@ -1203,8 +1203,8 @@ github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
 github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
 github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
 github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
-github.com/zexi/influxql-to-metricsql v0.1.1 h1:dXnVzlxVLuo88zGkDca7erEnmvprGWjzsBcs8qIoQGQ=
-github.com/zexi/influxql-to-metricsql v0.1.1/go.mod h1:JlC5FY+6De9ZPxG47G5DOgva8P9X1VaKS4ExzCmhSCc=
+github.com/zexi/influxql-to-metricsql v0.1.2 h1:Akkn1PLuNWjwt+Aqxf1fV+Gg+KUIgouVwtzm4XWy/5w=
+github.com/zexi/influxql-to-metricsql v0.1.2/go.mod h1:JlC5FY+6De9ZPxG47G5DOgva8P9X1VaKS4ExzCmhSCc=
 github.com/zexi/promql/v2 v2.12.1 h1:crHKpULdLLsBZ9b78Rg6qQkugzlk6BHeCj93tw/F5RU=
 github.com/zexi/promql/v2 v2.12.1/go.mod h1:2UtzWZGmth95n2qdIZWpJ5yQ0cE5hEvz0dGsajI1Sqg=
 go.etcd.io/bbolt v1.3.7 h1:j+zJOnnEjF/kyHlDDgGnVL/AIqIJPq8UoB2GSNfkUfQ=

pkg/cloudcommon/tsdb/tsdb.go

@@ -38,7 +38,7 @@ func NewTSDBServiceSource(t string, urls []string) *TSDBServiceSource {
 
 func GetDefaultServiceSource(s *mcclient.ClientSession, endpointType string) (*TSDBServiceSource, error) {
 	errs := []error{}
-	for _, sType := range []string{apis.SERVICE_TYPE_INFLUXDB, apis.SERVICE_TYPE_VICTORIA_METRICS} {
+	for _, sType := range []string{apis.SERVICE_TYPE_VICTORIA_METRICS, apis.SERVICE_TYPE_INFLUXDB} {
 		urls, err := s.GetServiceURLs(sType, endpointType, httputils.POST)
 		if err != nil {
 			errs = append(errs, errors.Wrapf(err, "get %s service type %q", endpointType, sType))

pkg/compute/models/loadbalanceragents.go

@@ -280,10 +280,12 @@ func (p *SLoadbalancerAgentParamsTelegraf) updateBy(pp *SLoadbalancerAgentParams
 }
 
 func (p *SLoadbalancerAgentParamsTelegraf) initDefault(data *jsonutils.JSONDict) {
-	if p.InfluxDbOutputUrl == "" {
+	{
 		baseOpts := &options.Options
 		u, _ := tsdb.GetDefaultServiceSourceURL(auth.GetAdminSession(context.Background(), baseOpts.Region), identity_apis.EndpointInterfacePublic)
-		p.InfluxDbOutputUrl = u
+		if u != "" {
+			p.InfluxDbOutputUrl = u
+		}
 		p.InfluxDbOutputUnsafeSsl = true
 	}
 	if p.HaproxyInputInterval == 0 {
@@ -910,6 +912,8 @@ listen stats
 	urls = ["{{ .telegraf.influx_db_output_url }}"]
 	database = "{{ .telegraf.influx_db_output_name }}"
 	insecure_skip_verify = {{ .telegraf.influx_db_output_unsafe_ssl }}
+	skip_database_creation = true
+	timeout = "30s"
 
 [[inputs.haproxy]]
 	interval = "{{ .telegraf.haproxy_input_interval }}s"

pkg/lbagent/haproxy.go

@@ -33,10 +33,13 @@ import (
 	"yunion.io/x/pkg/gotypes"
 
 	"yunion.io/x/onecloud/pkg/apis"
+	identity_apis "yunion.io/x/onecloud/pkg/apis/identity"
+	"yunion.io/x/onecloud/pkg/cloudcommon/tsdb"
 	"yunion.io/x/onecloud/pkg/hostman/hostinfo/hostconsts"
 	"yunion.io/x/onecloud/pkg/hostman/system_service"
 	agentmodels "yunion.io/x/onecloud/pkg/lbagent/models"
 	agentutils "yunion.io/x/onecloud/pkg/lbagent/utils"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
 	"yunion.io/x/onecloud/pkg/util/sysutils"
 )
 
@@ -187,6 +190,13 @@ func (h *HaproxyHelper) handleUseCorpusCmd(ctx context.Context, cmd *LbagentCmd)
 				return err
 			}
 		}
+		// refresh telegraf URL from service catalog
+		s := auth.GetAdminSession(ctx, h.opts.Region)
+		tsdbSrc, _ := tsdb.GetDefaultServiceSource(s, identity_apis.EndpointInterfacePublic)
+		if tsdbSrc != nil && len(tsdbSrc.URLs) > 0 {
+			agentParams.AgentModel.Params.Telegraf.InfluxDbOutputUrl = tsdbSrc.URLs[0]
+			agentParams.SetTelegrafParams("influx_db_output_url", tsdbSrc.URLs[0])
+		}
 		if agentParams.AgentModel.Params.Telegraf.InfluxDbOutputUrl != "" {
 			agentParams.SetTelegrafParams("haproxy_input_stats_socket", h.haproxyStatsSocketFile())
 			// telegraf config
@@ -198,7 +208,7 @@ func (h *HaproxyHelper) handleUseCorpusCmd(ctx context.Context, cmd *LbagentCmd)
 				p := filepath.Join(dir, "telegraf.conf")
 				err := os.WriteFile(p, d, agentutils.FileModeFile)
 				if err == nil {
-					err := h.reloadTelegraf(ctx, agentParams)
+					err := h.reloadTelegraf(ctx, agentParams, tsdbSrc)
 					if err != nil {
 						log.Errorf("reloading telegraf.conf failed: %s", err)
 					}
@@ -420,15 +430,15 @@ func (h *HaproxyHelper) telegrafPidFile() *agentutils.PidFile {
 	return pf
 }
 
-func (h *HaproxyHelper) reloadTelegraf(ctx context.Context, agentParams *agentmodels.AgentParams) error {
+func (h *HaproxyHelper) reloadTelegraf(ctx context.Context, agentParams *agentmodels.AgentParams, tsdbSrc *tsdb.TSDBServiceSource) error {
 	if h.opts.EnableRemoteExecutor {
-		return h.remoteReloadTelegraf(ctx, agentParams)
+		return h.remoteReloadTelegraf(ctx, agentParams, tsdbSrc)
 	} else {
 		return h.localReloadTelegraf(ctx)
 	}
 }
 
-func (h *HaproxyHelper) remoteReloadTelegraf(ctx context.Context, agentParams *agentmodels.AgentParams) error {
+func (h *HaproxyHelper) remoteReloadTelegraf(ctx context.Context, agentParams *agentmodels.AgentParams, tsdbSrc *tsdb.TSDBServiceSource) error {
 	telegraf := system_service.GetService("telegraf")
 	conf := map[string]interface{}{}
 	conf["hostname"] = h.getHostname()
@@ -442,7 +452,13 @@ func (h *HaproxyHelper) remoteReloadTelegraf(ctx context.Context, agentParams *a
 		hostconsts.TELEGRAF_TAG_KEY_HOST_TYPE: hostconsts.TELEGRAF_TAG_ONECLOUD_HOST_TYPE_LBAGENT,
 	}
 	conf["nics"] = h.getNicsTelegrafConf()
-	if len(agentParams.AgentModel.Params.Telegraf.InfluxDbOutputUrl) > 0 {
+	if tsdbSrc != nil && len(tsdbSrc.URLs) > 0 {
+		conf[apis.SERVICE_TYPE_INFLUXDB] = map[string]interface{}{
+			"url":       tsdbSrc.URLs,
+			"database":  agentParams.AgentModel.Params.Telegraf.InfluxDbOutputName,
+			"tsdb_type": tsdbSrc.Type,
+		}
+	} else if len(agentParams.AgentModel.Params.Telegraf.InfluxDbOutputUrl) > 0 {
 		conf[apis.SERVICE_TYPE_INFLUXDB] = map[string]interface{}{
 			"url": []string{
 				agentParams.AgentModel.Params.Telegraf.InfluxDbOutputUrl,
@@ -452,13 +468,13 @@ func (h *HaproxyHelper) remoteReloadTelegraf(ctx context.Context, agentParams *a
 	}
 	conf["haproxy"] = map[string]interface{}{
 		"interval":          agentParams.AgentModel.Params.Telegraf.HaproxyInputInterval,
-		"stats_socket_path": h.haproxyStatsSocketFile(),
+		"stats_socket_path": filepath.Join("/hostfs", h.haproxyStatsSocketFile()),
 	}
 	oldConf := telegraf.GetConf()
 	log.Debugf("old config: %s", oldConf)
 	log.Debugf("new config: %s", conf)
 	if gotypes.IsNil(oldConf) || !reflect.DeepEqual(oldConf, conf) {
-		log.Debugf("telegraf config: %s", conf)
+		log.Debugf("telegraf config: %s", telegraf.GetConfig(conf))
 		telegraf.SetConf(conf)
 		telegraf.BgReloadConf(conf)
 	}

vendor/github.com/zexi/influxql-to-metricsql/converter/translator/promql.go

@@ -1846,7 +1846,7 @@ github.com/yusufpapurcu/wmi
 # github.com/zeebo/xxh3 v1.0.2
 ## explicit; go 1.17
 github.com/zeebo/xxh3
-# github.com/zexi/influxql-to-metricsql v0.1.1
+# github.com/zexi/influxql-to-metricsql v0.1.2
 ## explicit; go 1.18
 github.com/zexi/influxql-to-metricsql/converter
 github.com/zexi/influxql-to-metricsql/converter/translator