Merge pull request #18897 from zexi/monitor-vm-top

fix(monitor): update vendor to support victoriametrics topk func

go.mod

@@ -65,7 +65,7 @@ require (
 	github.com/vishvananda/netns v0.0.0-20211101163701-50045581ed74
 	github.com/vmihailenco/msgpack v4.0.4+incompatible
 	github.com/xuri/excelize/v2 v2.7.1
-	github.com/zexi/influxql-to-promql v0.0.5
+	github.com/zexi/influxql-to-metricsql v0.0.6
 	go.etcd.io/etcd/api/v3 v3.5.0
 	go.etcd.io/etcd/client/v3 v3.5.0
 	golang.org/x/crypto v0.8.0

go.sum

@@ -745,8 +745,8 @@ github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
 github.com/yusufpapurcu/wmi v1.2.2 h1:KBNDSne4vP5mbSWnJbO+51IMOXJB67QiYCSBrubbPRg=
 github.com/yusufpapurcu/wmi v1.2.2/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
-github.com/zexi/influxql-to-promql v0.0.5 h1:4gRTqrof31O2Ku443RoTh20AadJG4sXbREn8i892WQo=
-github.com/zexi/influxql-to-promql v0.0.5/go.mod h1:a+S9SfEZ/aVyBM5abA7UCQmp7vD8Q9o+GfkQhk5mM58=
+github.com/zexi/influxql-to-metricsql v0.0.6 h1:E16T4oqgjIJtSNVvhGGHnw+pmY3yGz2iRsmmmSVuOpY=
+github.com/zexi/influxql-to-metricsql v0.0.6/go.mod h1:PyRRM+3Zrzzig6J4okYLeSv+/d+5GaL5ccBaUKQABNs=
 go.etcd.io/etcd/api/v3 v3.5.0 h1:GsV3S+OfZEOCNXdtNkBSR7kgLobAa/SO6tCxRa0GAYw=
 go.etcd.io/etcd/api/v3 v3.5.0/go.mod h1:cbVKeC6lCfl7j/8jBhAK6aIYO9XOjdptoxU/nLQcPvs=
 go.etcd.io/etcd/client/pkg/v3 v3.5.0 h1:2aQv6F436YnN7I4VbI8PPYrBhu+SmrTaADcf8Mi/6PU=

pkg/monitor/tsdb/driver/victoriametrics/result_parser.go

@@ -21,7 +21,7 @@ import (
 	"strings"
 
 	"github.com/influxdata/promql/v2/pkg/labels"
-	"github.com/zexi/influxql-to-promql/converter/translator"
+	"github.com/zexi/influxql-to-metricsql/converter/translator"
 
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"

pkg/monitor/tsdb/driver/victoriametrics/vm.go

@@ -23,8 +23,8 @@ import (
 
 	"github.com/influxdata/influxql"
 	"github.com/influxdata/promql/v2/pkg/labels"
-	"github.com/zexi/influxql-to-promql/converter"
-	"github.com/zexi/influxql-to-promql/converter/translator"
+	"github.com/zexi/influxql-to-metricsql/converter"
+	"github.com/zexi/influxql-to-metricsql/converter/translator"
 
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"

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

@@ -7,7 +7,7 @@ import (
 	"github.com/influxdata/influxql"
 	"github.com/pkg/errors"
 
-	"github.com/zexi/influxql-to-promql/converter/translator"
+	"github.com/zexi/influxql-to-metricsql/converter/translator"
 )
 
 type Converter interface {

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

@@ -3,6 +3,7 @@ package translator
 import (
 	"fmt"
 	"log"
+	"strconv"
 	"strings"
 	"time"
 
@@ -15,6 +16,10 @@ import (
 
 const UNION_RESULT_NAME = "__union_result__"
 
+const (
+	CALL_TOP = "top"
+)
+
 type promQL struct {
 	groupByWildcard bool
 	timeRange       *influxql.TimeRange
@@ -40,11 +45,11 @@ func (m *promQL) GetTimeRange() *influxql.TimeRange {
 
 type fieldResult struct {
 	metricName string
-	aggrOps    []string
+	aggrOps    []*AggrOperator
 	expr       promql.Expr
 }
 
-func newFieldResult(metricName string, ops []string, expr promql.Expr) *fieldResult {
+func newFieldResult(metricName string, ops []*AggrOperator, expr promql.Expr) *fieldResult {
 	return &fieldResult{
 		metricName: metricName,
 		aggrOps:    ops,
@@ -128,7 +133,11 @@ func unionFieldsExpr(exprs []*fieldResult) promql.Expr {
 		expr := exprs[i]
 		setValue := expr.metricName
 		if len(expr.aggrOps) > 0 {
-			setValue = fmt.Sprintf("%s_%s", strings.Join(expr.aggrOps, "_"), expr.metricName)
+			opsNames := make([]string, len(expr.aggrOps))
+			for i := range expr.aggrOps {
+				opsNames[i] = expr.aggrOps[i].Name
+			}
+			setValue = fmt.Sprintf("%s_%s", strings.Join(opsNames, "_"), expr.metricName)
 		}
 		result[i] = &promql.Call{
 			Func: &promql.Function{
@@ -191,7 +200,7 @@ func (m promQL) generateExpr(
 	metricName string,
 	ls []*labels.Matcher,
 	lookbehindWindow string,
-	aggrOps []string,
+	aggrOps []*AggrOperator,
 	groups []string) (promql.Expr, error) {
 	//fmt.Printf("=====name: %s, labels: %#v, lookbehindWindow: %q, aggrOps: %#v, groups: %#v\n", metricName, ls, lookbehindWindow, aggrOps, groups)
 	for _, l := range ls {
@@ -256,25 +265,29 @@ func (m promQL) formatExpr(expr promql.Expr) string {
 }
 
 func newAggrExpr(name string, argType promql.ValueType, returnType promql.ValueType, restExpr promql.Expr) promql.Expr {
+	return newAggrExprWithArgs(name, []promql.ValueType{argType}, returnType, promql.Expressions{restExpr})
+}
+
+func newAggrExprWithArgs(name string, args []promql.ValueType, returnType promql.ValueType, restExprs promql.Expressions) promql.Expr {
 	return &promql.Call{
 		Func: &promql.Function{
 			Name:       name,
-			ArgTypes:   []promql.ValueType{argType},
+			ArgTypes:   args,
 			Variadic:   0,
 			ReturnType: returnType,
 		},
-		Args: promql.Expressions{restExpr},
+		Args: restExprs,
 	}
 }
 
-func getAggrExpr(ops []string, expr promql.Expr) promql.Expr {
+func getAggrExpr(ops []*AggrOperator, expr promql.Expr) promql.Expr {
 	if len(ops) == 0 {
 		return expr
 	}
 	aggrOp := ops[0]
 	restOps := ops[1:]
 	restExpr := getAggrExpr(restOps, expr)
-	switch aggrOp {
+	switch aggrOp.Name {
 	case "abs":
 		// https://prometheus.io/docs/prometheus/latest/querying/functions/#abs
 		expr = newAggrExpr("abs", promql.ValueTypeVector, promql.ValueTypeVector, restExpr)
@@ -309,15 +322,46 @@ func getAggrExpr(ops []string, expr promql.Expr) promql.Expr {
 		expr = newAggrExpr("integrate", promql.ValueTypeMatrix, promql.ValueTypeVector, restExpr)
 	case "distinct":
 		expr = newAggrExpr("distinct", promql.ValueTypeMatrix, promql.ValueTypeVector, restExpr)
+	case CALL_TOP:
+		expr = newAggrExprWithArgs("topk_avg",
+			[]promql.ValueType{
+				promql.ValueTypeString,
+				promql.ValueTypeMatrix,
+			}, promql.ValueTypeVector,
+			promql.Expressions{
+				aggrOp.Args[0],
+				restExpr})
 	}
 	return expr
 }
 
-func getAggrOperator(op *influxql.Call) ([]string, error) {
-	if len(op.Args) != 1 {
-		return nil, errors.Errorf("not supported operator: %s with args: %#v", op.String(), op.Args)
+type AggrOperator struct {
+	Name string
+	Args promql.Expressions
+}
+
+func newAggrOperatorByName(name string) *AggrOperator {
+	return &AggrOperator{
+		Name: name,
 	}
-	ret := []string{op.Name}
+}
+
+func getAggrOperator(op *influxql.Call) ([]*AggrOperator, error) {
+	if len(op.Args) != 1 && op.Name != CALL_TOP {
+		return nil, errors.Errorf("not supported aggregator: %s with args: %#v", op.String(), op.Args)
+	}
+	aggOp := newAggrOperatorByName(op.Name)
+	if op.Name == CALL_TOP {
+		topNumStr := op.Args[len(op.Args)-1].String()
+		topNum, err := strconv.Atoi(topNumStr)
+		if err != nil {
+			return nil, errors.Wrapf(err, "parse top aggregator: %s", op)
+		}
+		aggOp.Args = promql.Expressions{
+			&promql.NumberLiteral{Val: float64(topNum)},
+		}
+	}
+	ret := []*AggrOperator{aggOp}
 	args, ok := op.Args[0].(*influxql.Call)
 	if !ok {
 		return ret, nil
@@ -330,7 +374,7 @@ func getAggrOperator(op *influxql.Call) ([]string, error) {
 	return ret, nil
 }
 
-func getAggrOperators(field *influxql.Field) ([]string, error) {
+func getAggrOperators(field *influxql.Field) ([]*AggrOperator, error) {
 	aggrOp, ok := field.Expr.(*influxql.Call)
 	if !ok {
 		return nil, nil
@@ -374,8 +418,8 @@ var (
 )
 
 func getCallVariable(c *influxql.Call) (string, error) {
-	if len(c.Args) != 1 {
-		return "", errors.Errorf("length of args %#v != 1", c.Args)
+	if len(c.Args) != 1 && c.Name != CALL_TOP {
+		return "", errors.Errorf("length of call %q args %#v != 1", c.Name, c.Args)
 	}
 	switch args := c.Args[0].(type) {
 	case *influxql.VarRef:

vendor/modules.txt

@@ -956,10 +956,10 @@ github.com/xuri/nfp
 # github.com/yusufpapurcu/wmi v1.2.2
 ## explicit; go 1.16
 github.com/yusufpapurcu/wmi
-# github.com/zexi/influxql-to-promql v0.0.5
+# github.com/zexi/influxql-to-metricsql v0.0.6
 ## explicit; go 1.18
-github.com/zexi/influxql-to-promql/converter
-github.com/zexi/influxql-to-promql/converter/translator
+github.com/zexi/influxql-to-metricsql/converter
+github.com/zexi/influxql-to-metricsql/converter/translator
 # go.etcd.io/etcd/api/v3 v3.5.0
 ## explicit; go 1.16
 go.etcd.io/etcd/api/v3/authpb