opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-08-31 01:35:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
Zexi Li
2026-04-27 16:42:08 +08:00
committed by GitHub
parent e6ba085e37
commit 986d0fbd8d
7 changed files with 183 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
+2 -2
View File
@@ -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
+1 -1
View File
@@ -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
+6 -2
View File
@@ -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
+23 -7
View File
@@ -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
Generated Vendored
+149 -8
View File
@@ -42,6 +42,7 @@ type promQL struct {
groupByWildcard bool
timeRange *influxql.TimeRange
fieldIsWildcard bool
fieldIsRegex bool
measurement string
labelsVisitor *labelsVisitor
}
@@ -103,9 +104,17 @@ func (m *promQL) translateField(s *influxql.SelectStatement, field *influxql.Fie
if err != nil {
return nil, errors.Wrap(err, "get matchers")
}
m.fieldIsRegex = false
if !m.fieldIsWildcard {
nameMatcher, _ := labels.NewMatcher(labels.MatchEqual, labels.MetricName, metricName)
matchers = append(matchers, nameMatcher)
if isRegexMetricName(metricName) {
m.fieldIsRegex = true
regexPattern := trimRegexDelimiters(metricName)
nameMatcher, _ := labels.NewMatcher(labels.MatchRegexp, labels.MetricName, regexPattern)
matchers = append(matchers, nameMatcher)
} else {
nameMatcher, _ := labels.NewMatcher(labels.MatchEqual, labels.MetricName, metricName)
matchers = append(matchers, nameMatcher)
}
}
lookbehindWin, groups, err := m.getGroups(s.Dimensions)
@@ -122,6 +131,12 @@ func (m *promQL) translateField(s *influxql.SelectStatement, field *influxql.Fie
if err != nil {
return nil, errors.Wrap(err, "generate expression")
}
if binExpr, ok := field.Expr.(*influxql.BinaryExpr); ok {
expr, err = wrapBinaryExpr(binExpr, expr)
if err != nil {
return nil, errors.Wrap(err, "wrap binary expression")
}
}
return newFieldResult(metricName, aggrOps, expr), nil
}
@@ -129,6 +144,7 @@ func (m *promQL) translate(s *influxql.SelectStatement) (string, error) {
exprs := make([]*fieldResult, 0)
var resultExpr promql.Expr
for _, field := range s.Fields {
m.labelsVisitor = newLabelsVisitor()
expr, err := m.translateField(s, field)
if err != nil {
return "", errors.Wrapf(err, "translate field %s", field)
@@ -246,7 +262,7 @@ func (m promQL) generateExpr(
LabelMatchers: ls,
Range: time.Duration(dur),
}
if !m.fieldIsWildcard {
if !m.fieldIsWildcard && !m.fieldIsRegex {
ms.Name = metricName
}
result = ms
@@ -254,7 +270,7 @@ func (m promQL) generateExpr(
vs := &promql.VectorSelector{
LabelMatchers: ls,
}
if !m.fieldIsWildcard {
if !m.fieldIsWildcard && !m.fieldIsRegex {
vs.Name = metricName
}
result = vs
@@ -450,11 +466,29 @@ func getAggrOperator(op *influxql.Call) ([]*AggrOperator, error) {
}
func getAggrOperators(field *influxql.Field) ([]*AggrOperator, error) {
aggrOp, ok := field.Expr.(*influxql.Call)
if !ok {
return nil, nil
switch expr := field.Expr.(type) {
case *influxql.Call:
return getAggrOperator(expr)
case *influxql.BinaryExpr:
return getBinaryExprAggrOperators(expr)
}
return getAggrOperator(aggrOp)
return nil, nil
}
func getBinaryExprAggrOperators(expr *influxql.BinaryExpr) ([]*AggrOperator, error) {
if call, ok := expr.LHS.(*influxql.Call); ok {
return getAggrOperator(call)
}
if call, ok := expr.RHS.(*influxql.Call); ok {
return getAggrOperator(call)
}
if binExpr, ok := expr.LHS.(*influxql.BinaryExpr); ok {
return getBinaryExprAggrOperators(binExpr)
}
if binExpr, ok := expr.RHS.(*influxql.BinaryExpr); ok {
return getBinaryExprAggrOperators(binExpr)
}
return nil, nil
}
func getMetricName(sources influxql.Sources, field *influxql.Field) (string, error) {
@@ -477,6 +511,8 @@ func getMetricName(sources influxql.Sources, field *influxql.Field) (string, err
fieldName = expr.Val
case *influxql.Call:
fieldName, err = getCallVariable(expr)
case *influxql.BinaryExpr:
fieldName, err = getBinaryExprVariable(expr)
default:
return "", errors.Errorf("field.Expr %#v is not supported", expr)
}
@@ -492,6 +528,26 @@ var (
ErrVariableIsWildcard = errors.New("variable field is wildcard")
)
// isRegexMetricName checks if a metric name contains a regex field pattern.
// In InfluxQL, regex fields are delimited by forward slashes, e.g., measurement_/pattern/
func isRegexMetricName(metricName string) bool {
idx := strings.Index(metricName, "/")
return idx >= 0 && strings.HasSuffix(metricName, "/")
}
// trimRegexDelimiters converts a metric name like "haproxy_/d(req|con)/" to "haproxy_d(req|con)"
func trimRegexDelimiters(metricName string) string {
idx := strings.Index(metricName, "/")
if idx < 0 {
return metricName
}
prefix := metricName[:idx]
regexPart := metricName[idx:]
regexPart = strings.TrimPrefix(regexPart, "/")
regexPart = strings.TrimSuffix(regexPart, "/")
return prefix + regexPart
}
func getCallVariable(c *influxql.Call) (string, error) {
if len(c.Args) != 1 && !MUL_ARGS_AGGREGATOR.Has(c.Name) {
return "", errors.Errorf("length of call %q args %#v != 1", c.Name, c.Args)
@@ -509,6 +565,91 @@ func getCallVariable(c *influxql.Call) (string, error) {
return c.Args[0].String(), nil
}
func getBinaryExprVariable(expr *influxql.BinaryExpr) (string, error) {
switch lhs := expr.LHS.(type) {
case *influxql.Call:
return getCallVariable(lhs)
case *influxql.VarRef:
return lhs.Val, nil
case *influxql.BinaryExpr:
return getBinaryExprVariable(lhs)
}
switch rhs := expr.RHS.(type) {
case *influxql.Call:
return getCallVariable(rhs)
case *influxql.VarRef:
return rhs.Val, nil
case *influxql.BinaryExpr:
return getBinaryExprVariable(rhs)
}
return "", errors.Errorf("BinaryExpr %#v doesn't contain a Call or VarRef", expr)
}
func influxqlOpToPromqlOp(op influxql.Token) (promql.ItemType, error) {
switch op {
case influxql.ADD:
return promql.ItemADD, nil
case influxql.SUB:
return promql.ItemSUB, nil
case influxql.MUL:
return promql.ItemMUL, nil
case influxql.DIV:
return promql.ItemDIV, nil
case influxql.MOD:
return promql.ItemMOD, nil
default:
return 0, errors.Errorf("unsupported influxql binary operator: %s", op)
}
}
func influxqlLiteralToPromqlExpr(expr influxql.Expr) (promql.Expr, error) {
switch v := expr.(type) {
case *influxql.IntegerLiteral:
return &promql.NumberLiteral{Val: float64(v.Val)}, nil
case *influxql.NumberLiteral:
return &promql.NumberLiteral{Val: v.Val}, nil
default:
return nil, errors.Errorf("unsupported literal type %T in binary expression", expr)
}
}
func wrapBinaryExpr(binExpr *influxql.BinaryExpr, innerExpr promql.Expr) (promql.Expr, error) {
promOp, err := influxqlOpToPromqlOp(binExpr.Op)
if err != nil {
return nil, err
}
// Determine which side is the Call (already translated as innerExpr)
// and which side is the literal
var literalExpr promql.Expr
var lhs, rhs promql.Expr
switch binExpr.LHS.(type) {
case *influxql.Call, *influxql.BinaryExpr:
// LHS is the call side, RHS should be the literal
literalExpr, err = influxqlLiteralToPromqlExpr(binExpr.RHS)
if err != nil {
return nil, err
}
lhs = innerExpr
rhs = literalExpr
default:
// RHS is the call side, LHS should be the literal
literalExpr, err = influxqlLiteralToPromqlExpr(binExpr.LHS)
if err != nil {
return nil, err
}
lhs = literalExpr
rhs = innerExpr
}
return &promql.BinaryExpr{
Op: promOp,
LHS: lhs,
RHS: rhs,
}, nil
}
type labelsVisitor struct {
err error
labels []*labels.Matcher
vendor/modules.txt
Vendored
+1 -1
View File
@@ -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