init

app/vmalert/config/config.go

@@ -16,6 +16,8 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
 )
 
+var DefaultGroupType string
+
 // Group contains list of Rules grouped into
 // entity with one name and evaluation interval
 type Group struct {
@@ -60,7 +62,7 @@ func (g *Group) UnmarshalYAML(unmarshal func(any) error) error {
 	}
 	// change default value to prometheus datasource.
 	if g.Type.Get() == "" {
-		g.Type.Set(NewPrometheusType())
+		g.Type = NewRawType(DefaultGroupType)
 	}
 
 	h := md5.New()

app/vmalert/config/testdata/rules/vlog/rules0-good.rules

@@ -0,0 +1,12 @@
+groups:
+  - name: vlogrule
+    interval: 5s
+    rules:
+      - alert: rule1
+        expr: "_time:5m | stats by (path) count() if (level:error) as errors,  count() as total | math errors / total as errors_percentage | fields errors_percentage,path"
+        labels:
+          label: bar
+      - record: rule2
+        expr: "_time:5h | stats by (_stream, path) count() as b"
+        labels:
+          label: bar

app/vmalert/config/types.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmselect/graphiteql"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/metricsql"
 )
 
@@ -27,6 +28,12 @@ func NewGraphiteType() Type {
 	}
 }
 
+func NewVlogsType() Type {
+	return Type{
+		Name: "vlogs",
+	}
+}
+
 // NewRawType returns datasource type from raw string
 // without validation.
 func NewRawType(d string) Type {
@@ -62,6 +69,10 @@ func (t *Type) ValidateExpr(expr string) error {
 		if _, err := metricsql.Parse(expr); err != nil {
 			return fmt.Errorf("bad prometheus expr: %q, err: %w", expr, err)
 		}
+	case "vlogs":
+		if _, err := logstorage.ParseQuery(expr); err != nil {
+			return fmt.Errorf("bad vlogs expr: %q, err: %w", expr, err)
+		}
 	default:
 		return fmt.Errorf("unknown datasource type=%q", t.Name)
 	}
@@ -78,7 +89,7 @@ func (t *Type) UnmarshalYAML(unmarshal func(any) error) error {
 		s = "prometheus"
 	}
 	switch s {
-	case "graphite", "prometheus":
+	case "graphite", "prometheus", "vlogs":
 	default:
 		return fmt.Errorf("unknown datasource type=%q, want %q or %q", s, "prometheus", "graphite")
 	}

app/vmalert/datasource/init.go

@@ -64,7 +64,7 @@ var (
 		`If true, disables HTTP keep-alive and will only use the connection to the server for a single HTTP request.`)
 	roundDigits = flag.Int("datasource.roundDigits", 0, `Adds "round_digits" GET param to datasource requests. `+
 		`In VM "round_digits" limits the number of digits after the decimal point in response values.`)
-)
+	)
 
 // InitSecretFlags must be called after flag.Parse and before any logging
 func InitSecretFlags() {
@@ -139,7 +139,7 @@ func Init(extraParams url.Values) (QuerierBuilder, error) {
 		datasourceURL:    strings.TrimSuffix(*addr, "/"),
 		appendTypePrefix: *appendTypePrefix,
 		queryStep:        *queryStep,
-		dataSourceType:   datasourcePrometheus,
+		dataSourceType:   toDatasourceType(""),
 		extraParams:      extraParams,
 	}, nil
 }

app/vmalert/datasource/vlog_api.go

@@ -0,0 +1,50 @@
+package datasource
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+)
+
+func (s *VMStorage) setVlogsInstantReqParams(r *http.Request, query string, timestamp time.Time) {
+	if !*disablePathAppend {
+		r.URL.Path += "/select/logsql/stats_query"
+	}
+	q := r.URL.Query()
+	q.Set("time", timestamp.Format(time.RFC3339))
+	r.URL.RawQuery = q.Encode()
+	s.setPrometheusReqParams(r, query)
+}
+
+func (s *VMStorage) setVlogsRangeReqParams(r *http.Request, query string, start, end time.Time) {
+	if !*disablePathAppend {
+		r.URL.Path += "/select/logsql/stats_query_range"
+	}
+	q := r.URL.Query()
+	q.Add("start", start.Format(time.RFC3339))
+	q.Add("end", end.Format(time.RFC3339))
+	if s.evaluationInterval > 0 { // set step as evaluationInterval by default
+		// always convert to seconds to keep compatibility with older
+		// Prometheus versions. See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1943
+		q.Set("step", fmt.Sprintf("%ds", int(s.evaluationInterval.Seconds())))
+	}
+	r.URL.RawQuery = q.Encode()
+	s.setPrometheusReqParams(r, query)
+}
+
+func parseVlogsResponse(req *http.Request, resp *http.Response) (res Result, err error) {
+	res, err = parsePrometheusResponse(req, resp)
+	if err != nil {
+		return Result{}, err
+	}
+	for i := range res.Data {
+		m := &res.Data[i]
+		for j := range m.Labels {
+			if m.Labels[j].Name == "__name__" {
+				m.Labels[j].Name = "stats_function"
+				continue
+			}
+		}
+	}
+	return
+}

app/vmalert/datasource/vm.go

@@ -19,14 +19,15 @@ type datasourceType string
 
 const (
 	datasourcePrometheus datasourceType = "prometheus"
+	datasourceVlogs      datasourceType = "vlogs"
 	datasourceGraphite   datasourceType = "graphite"
 )
 
 func toDatasourceType(s string) datasourceType {
-	if s == string(datasourceGraphite) {
+	if s == "" {
-		return datasourceGraphite
-	}
 		return datasourcePrometheus
+	}
+	return datasourceType(s)
 }
 
 // VMStorage represents vmstorage entity with ability to read and write metrics
@@ -86,7 +87,9 @@ func (s *VMStorage) Clone() *VMStorage {
 
 // ApplyParams - changes given querier params.
 func (s *VMStorage) ApplyParams(params QuerierParams) *VMStorage {
+	if params.DataSourceType != "" {
 		s.dataSourceType = toDatasourceType(params.DataSourceType)
+	}
 	s.evaluationInterval = params.EvaluationInterval
 	if params.QueryParams != nil {
 		if s.extraParams == nil {
@@ -158,9 +161,14 @@ func (s *VMStorage) Query(ctx context.Context, query string, ts time.Time) (Resu
 	}
 
 	// Process the received response.
-	parseFn := parsePrometheusResponse
+	var parseFn func(req *http.Request, resp *http.Response) (Result, error)
-	if s.dataSourceType != datasourcePrometheus {
+	switch s.dataSourceType {
+		case datasourcePrometheus:
+			parseFn = parsePrometheusResponse
+		case datasourceGraphite:
 			parseFn = parseGraphiteResponse
+		case datasourceVlogs:
+			parseFn = parseVlogsResponse
 	}
 	result, err := parseFn(req, resp)
 	_ = resp.Body.Close()
@@ -171,7 +179,7 @@ func (s *VMStorage) Query(ctx context.Context, query string, ts time.Time) (Resu
 // For Prometheus type see https://prometheus.io/docs/prometheus/latest/querying/api/#range-queries
 // Graphite type isn't supported.
 func (s *VMStorage) QueryRange(ctx context.Context, query string, start, end time.Time) (res Result, err error) {
-	if s.dataSourceType != datasourcePrometheus {
+	if s.dataSourceType == datasourceGraphite {
 		return res, fmt.Errorf("%q is not supported for QueryRange", s.dataSourceType)
 	}
 	if start.IsZero() {
@@ -247,6 +255,8 @@ func (s *VMStorage) newQueryRequest(ctx context.Context, query string, ts time.T
 		s.setPrometheusInstantReqParams(req, query, ts)
 	case datasourceGraphite:
 		s.setGraphiteReqParams(req, query)
+	case datasourceVlogs:
+		s.setVlogsInstantReqParams(req, query, ts)
 	default:
 		logger.Panicf("BUG: engine not found: %q", s.dataSourceType)
 	}

app/vmalert/main.go

@@ -81,6 +81,7 @@ absolute path to all .tpl files in root.
 	remoteReadIgnoreRestoreErrors = flag.Bool("remoteRead.ignoreRestoreErrors", true, "Whether to ignore errors from remote storage when restoring alerts state on startup. DEPRECATED - this flag has no effect and will be removed in the next releases.")
 
 	dryRun = flag.Bool("dryRun", false, "Whether to check only config files without running vmalert. The rules file are validated. The -rule flag must be specified.")
+	defaultRuleType = flag.String("defaultRuleType", "prometheus", "")
 )
 
 var alertURLGeneratorFn notifier.AlertURLGenerator
@@ -106,6 +107,7 @@ func main() {
 		logger.Fatalf("failed to parse %q: %s", *ruleTemplatesPath, err)
 	}
 
+	config.DefaultGroupType = *defaultRuleType
 	if *dryRun {
 		groups, err := config.Parse(*rulePath, notifier.ValidateTemplates, true)
 		if err != nil {