mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2025-02-09 15:27:11 +00:00
changes vmalert Querier with per rule querier (#1249)
* changes vmalert Querier with per rule querier it allows to changes some parametrs based on rule setting for instance - alert type, tenant for cluster version or event endpoint url.
This commit is contained in:
parent
421a92983a
commit
2eb8ef7b2b
16 changed files with 139 additions and 65 deletions
|
@ -29,6 +29,8 @@ type AlertingRule struct {
|
|||
GroupID uint64
|
||||
GroupName string
|
||||
|
||||
q datasource.Querier
|
||||
|
||||
// guard status fields
|
||||
mu sync.RWMutex
|
||||
// stores list of active alerts
|
||||
|
@ -49,7 +51,7 @@ type alertingRuleMetrics struct {
|
|||
active *gauge
|
||||
}
|
||||
|
||||
func newAlertingRule(group *Group, cfg config.Rule) *AlertingRule {
|
||||
func newAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule) *AlertingRule {
|
||||
ar := &AlertingRule{
|
||||
Type: cfg.Type,
|
||||
RuleID: cfg.ID,
|
||||
|
@ -60,6 +62,7 @@ func newAlertingRule(group *Group, cfg config.Rule) *AlertingRule {
|
|||
Annotations: cfg.Annotations,
|
||||
GroupID: group.ID(),
|
||||
GroupName: group.Name,
|
||||
q: qb.BuildWithParams(datasource.QuerierParams{DataSourceType: &cfg.Type}),
|
||||
alerts: make(map[uint64]*notifier.Alert),
|
||||
metrics: &alertingRuleMetrics{},
|
||||
}
|
||||
|
@ -121,8 +124,8 @@ func (ar *AlertingRule) ID() uint64 {
|
|||
|
||||
// Exec executes AlertingRule expression via the given Querier.
|
||||
// Based on the Querier results AlertingRule maintains notifier.Alerts
|
||||
func (ar *AlertingRule) Exec(ctx context.Context, q datasource.Querier, series bool) ([]prompbmarshal.TimeSeries, error) {
|
||||
qMetrics, err := q.Query(ctx, ar.Expr, ar.Type)
|
||||
func (ar *AlertingRule) Exec(ctx context.Context, series bool) ([]prompbmarshal.TimeSeries, error) {
|
||||
qMetrics, err := ar.q.Query(ctx, ar.Expr)
|
||||
ar.mu.Lock()
|
||||
defer ar.mu.Unlock()
|
||||
|
||||
|
@ -139,7 +142,7 @@ func (ar *AlertingRule) Exec(ctx context.Context, q datasource.Querier, series b
|
|||
}
|
||||
}
|
||||
|
||||
qFn := func(query string) ([]datasource.Metric, error) { return q.Query(ctx, query, ar.Type) }
|
||||
qFn := func(query string) ([]datasource.Metric, error) { return ar.q.Query(ctx, query) }
|
||||
updated := make(map[uint64]struct{})
|
||||
// update list of active alerts
|
||||
for _, m := range qMetrics {
|
||||
|
@ -407,7 +410,7 @@ func (ar *AlertingRule) Restore(ctx context.Context, q datasource.Querier, lookb
|
|||
return fmt.Errorf("querier is nil")
|
||||
}
|
||||
|
||||
qFn := func(query string) ([]datasource.Metric, error) { return q.Query(ctx, query, ar.Type) }
|
||||
qFn := func(query string) ([]datasource.Metric, error) { return ar.q.Query(ctx, query) }
|
||||
|
||||
// account for external labels in filter
|
||||
var labelsFilter string
|
||||
|
@ -420,7 +423,7 @@ func (ar *AlertingRule) Restore(ctx context.Context, q datasource.Querier, lookb
|
|||
// remote write protocol which is used for state persistence in vmalert.
|
||||
expr := fmt.Sprintf("last_over_time(%s{alertname=%q%s}[%ds])",
|
||||
alertForStateMetricName, ar.Name, labelsFilter, int(lookback.Seconds()))
|
||||
qMetrics, err := q.Query(ctx, expr, ar.Type)
|
||||
qMetrics, err := q.Query(ctx, expr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
|
@ -294,11 +294,12 @@ func TestAlertingRule_Exec(t *testing.T) {
|
|||
for _, tc := range testCases {
|
||||
t.Run(tc.rule.Name, func(t *testing.T) {
|
||||
fq := &fakeQuerier{}
|
||||
tc.rule.q = fq
|
||||
tc.rule.GroupID = fakeGroup.ID()
|
||||
for _, step := range tc.steps {
|
||||
fq.reset()
|
||||
fq.add(step...)
|
||||
if _, err := tc.rule.Exec(context.TODO(), fq, false); err != nil {
|
||||
if _, err := tc.rule.Exec(context.TODO(), false); err != nil {
|
||||
t.Fatalf("unexpected err: %s", err)
|
||||
}
|
||||
// artificial delay between applying steps
|
||||
|
@ -410,6 +411,7 @@ func TestAlertingRule_Restore(t *testing.T) {
|
|||
t.Run(tc.rule.Name, func(t *testing.T) {
|
||||
fq := &fakeQuerier{}
|
||||
tc.rule.GroupID = fakeGroup.ID()
|
||||
tc.rule.q = fq
|
||||
fq.add(tc.metrics...)
|
||||
if err := tc.rule.Restore(context.TODO(), fq, time.Hour, nil); err != nil {
|
||||
t.Fatalf("unexpected err: %s", err)
|
||||
|
@ -437,17 +439,18 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
|
|||
fq := &fakeQuerier{}
|
||||
ar := newTestAlertingRule("test", 0)
|
||||
ar.Labels = map[string]string{"job": "test"}
|
||||
ar.q = fq
|
||||
|
||||
// successful attempt
|
||||
fq.add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "bar"))
|
||||
_, err := ar.Exec(context.TODO(), fq, false)
|
||||
_, err := ar.Exec(context.TODO(), false)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// label `job` will collide with rule extra label and will make both time series equal
|
||||
fq.add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "baz"))
|
||||
_, err = ar.Exec(context.TODO(), fq, false)
|
||||
_, err = ar.Exec(context.TODO(), false)
|
||||
if !errors.Is(err, errDuplicate) {
|
||||
t.Fatalf("expected to have %s error; got %s", errDuplicate, err)
|
||||
}
|
||||
|
@ -456,7 +459,7 @@ func TestAlertingRule_Exec_Negative(t *testing.T) {
|
|||
|
||||
expErr := "connection reset by peer"
|
||||
fq.setErr(errors.New(expErr))
|
||||
_, err = ar.Exec(context.TODO(), fq, false)
|
||||
_, err = ar.Exec(context.TODO(), false)
|
||||
if err == nil {
|
||||
t.Fatalf("expected to get err; got nil")
|
||||
}
|
||||
|
@ -544,8 +547,9 @@ func TestAlertingRule_Template(t *testing.T) {
|
|||
t.Run(tc.rule.Name, func(t *testing.T) {
|
||||
fq := &fakeQuerier{}
|
||||
tc.rule.GroupID = fakeGroup.ID()
|
||||
tc.rule.q = fq
|
||||
fq.add(tc.metrics...)
|
||||
if _, err := tc.rule.Exec(context.TODO(), fq, false); err != nil {
|
||||
if _, err := tc.rule.Exec(context.TODO(), false); err != nil {
|
||||
t.Fatalf("unexpected err: %s", err)
|
||||
}
|
||||
for hash, expAlert := range tc.expAlerts {
|
||||
|
|
|
@ -4,11 +4,16 @@ import (
|
|||
"context"
|
||||
)
|
||||
|
||||
// QuerierBuilder builds Querier with given params.
|
||||
type QuerierBuilder interface {
|
||||
BuildWithParams(params QuerierParams) Querier
|
||||
}
|
||||
|
||||
// Querier interface wraps Query method which
|
||||
// executes given query and returns list of Metrics
|
||||
// as result
|
||||
type Querier interface {
|
||||
Query(ctx context.Context, query string, engine Type) ([]Metric, error)
|
||||
Query(ctx context.Context, query string) ([]Metric, error)
|
||||
}
|
||||
|
||||
// Metric is the basic entity which should be return by datasource
|
||||
|
|
|
@ -28,7 +28,7 @@ var (
|
|||
)
|
||||
|
||||
// Init creates a Querier from provided flag values.
|
||||
func Init() (Querier, error) {
|
||||
func Init() (QuerierBuilder, error) {
|
||||
if *addr == "" {
|
||||
return nil, fmt.Errorf("datasource.url is empty")
|
||||
}
|
||||
|
|
|
@ -81,6 +81,7 @@ type VMStorage struct {
|
|||
appendTypePrefix bool
|
||||
lookBack time.Duration
|
||||
queryStep time.Duration
|
||||
dataSourceType Type
|
||||
}
|
||||
|
||||
const queryPath = "/api/v1/query"
|
||||
|
@ -89,6 +90,38 @@ const graphitePath = "/render"
|
|||
const prometheusPrefix = "/prometheus"
|
||||
const graphitePrefix = "/graphite"
|
||||
|
||||
// QuerierParams params for Querier.
|
||||
type QuerierParams struct {
|
||||
DataSourceType *Type
|
||||
}
|
||||
|
||||
// Clone makes clone of VMStorage, shares http client.
|
||||
func (s *VMStorage) Clone() *VMStorage {
|
||||
return &VMStorage{
|
||||
c: s.c,
|
||||
datasourceURL: s.datasourceURL,
|
||||
basicAuthUser: s.basicAuthUser,
|
||||
basicAuthPass: s.basicAuthPass,
|
||||
lookBack: s.lookBack,
|
||||
queryStep: s.queryStep,
|
||||
appendTypePrefix: s.appendTypePrefix,
|
||||
dataSourceType: s.dataSourceType,
|
||||
}
|
||||
}
|
||||
|
||||
// ApplyParams - changes given querier params.
|
||||
func (s *VMStorage) ApplyParams(params QuerierParams) *VMStorage {
|
||||
if params.DataSourceType != nil {
|
||||
s.dataSourceType = *params.DataSourceType
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// BuildWithParams - implements interface.
|
||||
func (s *VMStorage) BuildWithParams(params QuerierParams) Querier {
|
||||
return s.Clone().ApplyParams(params)
|
||||
}
|
||||
|
||||
// NewVMStorage is a constructor for VMStorage
|
||||
func NewVMStorage(baseURL, basicAuthUser, basicAuthPass string, lookBack time.Duration, queryStep time.Duration, appendTypePrefix bool, c *http.Client) *VMStorage {
|
||||
return &VMStorage{
|
||||
|
@ -99,18 +132,19 @@ func NewVMStorage(baseURL, basicAuthUser, basicAuthPass string, lookBack time.Du
|
|||
appendTypePrefix: appendTypePrefix,
|
||||
lookBack: lookBack,
|
||||
queryStep: queryStep,
|
||||
dataSourceType: NewPrometheusType(),
|
||||
}
|
||||
}
|
||||
|
||||
// Query reads metrics from datasource by given query and type
|
||||
func (s *VMStorage) Query(ctx context.Context, query string, dataSourceType Type) ([]Metric, error) {
|
||||
switch dataSourceType.name {
|
||||
func (s *VMStorage) Query(ctx context.Context, query string) ([]Metric, error) {
|
||||
switch s.dataSourceType.name {
|
||||
case "", prometheusType:
|
||||
return s.queryDataSource(ctx, query, s.setPrometheusReqParams, parsePrometheusResponse)
|
||||
case graphiteType:
|
||||
return s.queryDataSource(ctx, query, s.setGraphiteReqParams, parseGraphiteResponse)
|
||||
default:
|
||||
return nil, fmt.Errorf("engine not found: %q", dataSourceType)
|
||||
return nil, fmt.Errorf("engine not found: %q", s.dataSourceType.name)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
@ -70,25 +70,25 @@ func TestVMSelectQuery(t *testing.T) {
|
|||
srv := httptest.NewServer(mux)
|
||||
defer srv.Close()
|
||||
am := NewVMStorage(srv.URL, basicAuthName, basicAuthPass, time.Minute, 0, false, srv.Client())
|
||||
if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected connection error got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected invalid response status error got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected response body error got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected error status got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected unknown status got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query, NewPrometheusType()); err == nil {
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected non-vector resultType error got nil")
|
||||
}
|
||||
m, err := am.Query(ctx, query, NewPrometheusType())
|
||||
m, err := am.Query(ctx, query)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected %s", err)
|
||||
}
|
||||
|
@ -106,7 +106,10 @@ func TestVMSelectQuery(t *testing.T) {
|
|||
m[0].Labels[0].Name != expected.Labels[0].Name {
|
||||
t.Fatalf("unexpected metric %+v want %+v", m[0], expected)
|
||||
}
|
||||
m, err = am.Query(ctx, queryRender, NewGraphiteType())
|
||||
|
||||
dst := NewGraphiteType()
|
||||
q := am.BuildWithParams(QuerierParams{&dst})
|
||||
m, err = q.Query(ctx, queryRender)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected %s", err)
|
||||
}
|
||||
|
|
|
@ -49,7 +49,7 @@ func newGroupMetrics(name, file string) *groupMetrics {
|
|||
return m
|
||||
}
|
||||
|
||||
func newGroup(cfg config.Group, defaultInterval time.Duration, labels map[string]string) *Group {
|
||||
func newGroup(cfg config.Group, qb datasource.QuerierBuilder, defaultInterval time.Duration, labels map[string]string) *Group {
|
||||
g := &Group{
|
||||
Type: cfg.Type,
|
||||
Name: cfg.Name,
|
||||
|
@ -81,17 +81,17 @@ func newGroup(cfg config.Group, defaultInterval time.Duration, labels map[string
|
|||
}
|
||||
r.Labels[k] = v
|
||||
}
|
||||
rules[i] = g.newRule(r)
|
||||
rules[i] = g.newRule(qb, r)
|
||||
}
|
||||
g.Rules = rules
|
||||
return g
|
||||
}
|
||||
|
||||
func (g *Group) newRule(rule config.Rule) Rule {
|
||||
func (g *Group) newRule(qb datasource.QuerierBuilder, rule config.Rule) Rule {
|
||||
if rule.Alert != "" {
|
||||
return newAlertingRule(g, rule)
|
||||
return newAlertingRule(qb, g, rule)
|
||||
}
|
||||
return newRecordingRule(g, rule)
|
||||
return newRecordingRule(qb, g, rule)
|
||||
}
|
||||
|
||||
// ID return unique group ID that consists of
|
||||
|
@ -106,7 +106,7 @@ func (g *Group) ID() uint64 {
|
|||
}
|
||||
|
||||
// Restore restores alerts state for group rules
|
||||
func (g *Group) Restore(ctx context.Context, q datasource.Querier, lookback time.Duration, labels map[string]string) error {
|
||||
func (g *Group) Restore(ctx context.Context, qb datasource.QuerierBuilder, lookback time.Duration, labels map[string]string) error {
|
||||
for _, rule := range g.Rules {
|
||||
rr, ok := rule.(*AlertingRule)
|
||||
if !ok {
|
||||
|
@ -115,6 +115,7 @@ func (g *Group) Restore(ctx context.Context, q datasource.Querier, lookback time
|
|||
if rr.For < 1 {
|
||||
continue
|
||||
}
|
||||
q := qb.BuildWithParams(datasource.QuerierParams{})
|
||||
if err := rr.Restore(ctx, q, lookback, labels); err != nil {
|
||||
return fmt.Errorf("error while restoring rule %q: %w", rule, err)
|
||||
}
|
||||
|
@ -189,7 +190,7 @@ func (g *Group) close() {
|
|||
|
||||
var skipRandSleepOnGroupStart bool
|
||||
|
||||
func (g *Group) start(ctx context.Context, querier datasource.Querier, nts []notifier.Notifier, rw *remotewrite.Client) {
|
||||
func (g *Group) start(ctx context.Context, nts []notifier.Notifier, rw *remotewrite.Client) {
|
||||
defer func() { close(g.finishedCh) }()
|
||||
|
||||
// Spread group rules evaluation over time in order to reduce load on VictoriaMetrics.
|
||||
|
@ -213,7 +214,7 @@ func (g *Group) start(ctx context.Context, querier datasource.Querier, nts []not
|
|||
}
|
||||
|
||||
logger.Infof("group %q started; interval=%v; concurrency=%d", g.Name, g.Interval, g.Concurrency)
|
||||
e := &executor{querier, nts, rw}
|
||||
e := &executor{nts, rw}
|
||||
t := time.NewTicker(g.Interval)
|
||||
defer t.Stop()
|
||||
for {
|
||||
|
@ -256,7 +257,6 @@ func (g *Group) start(ctx context.Context, querier datasource.Querier, nts []not
|
|||
}
|
||||
|
||||
type executor struct {
|
||||
querier datasource.Querier
|
||||
notifiers []notifier.Notifier
|
||||
rw *remotewrite.Client
|
||||
}
|
||||
|
@ -310,7 +310,7 @@ func (e *executor) exec(ctx context.Context, rule Rule, returnSeries bool, inter
|
|||
execDuration.UpdateDuration(execStart)
|
||||
}()
|
||||
|
||||
tss, err := rule.Exec(ctx, e.querier, returnSeries)
|
||||
tss, err := rule.Exec(ctx, returnSeries)
|
||||
if err != nil {
|
||||
execErrors.Inc()
|
||||
return fmt.Errorf("rule %q: failed to execute: %w", rule, err)
|
||||
|
|
|
@ -6,6 +6,8 @@ import (
|
|||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/notifier"
|
||||
)
|
||||
|
@ -105,20 +107,32 @@ func TestUpdateWith(t *testing.T) {
|
|||
{Record: "foo5"},
|
||||
},
|
||||
},
|
||||
{
|
||||
"update datasource type",
|
||||
[]config.Rule{
|
||||
{Alert: "foo1", Type: datasource.NewPrometheusType()},
|
||||
{Alert: "foo3", Type: datasource.NewGraphiteType()},
|
||||
},
|
||||
[]config.Rule{
|
||||
{Alert: "foo1", Type: datasource.NewGraphiteType()},
|
||||
{Alert: "foo10", Type: datasource.NewPrometheusType()},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
g := &Group{Name: "test"}
|
||||
qb := &fakeQuerier{}
|
||||
for _, r := range tc.currentRules {
|
||||
r.ID = config.HashRule(r)
|
||||
g.Rules = append(g.Rules, g.newRule(r))
|
||||
g.Rules = append(g.Rules, g.newRule(qb, r))
|
||||
}
|
||||
|
||||
ng := &Group{Name: "test"}
|
||||
for _, r := range tc.newRules {
|
||||
r.ID = config.HashRule(r)
|
||||
ng.Rules = append(ng.Rules, ng.newRule(r))
|
||||
ng.Rules = append(ng.Rules, ng.newRule(qb, r))
|
||||
}
|
||||
|
||||
err := g.updateWith(ng)
|
||||
|
@ -156,11 +170,11 @@ func TestGroupStart(t *testing.T) {
|
|||
t.Fatalf("failed to parse rules: %s", err)
|
||||
}
|
||||
const evalInterval = time.Millisecond
|
||||
g := newGroup(groups[0], evalInterval, map[string]string{"cluster": "east-1"})
|
||||
g.Concurrency = 2
|
||||
|
||||
fn := &fakeNotifier{}
|
||||
fs := &fakeQuerier{}
|
||||
fn := &fakeNotifier{}
|
||||
|
||||
g := newGroup(groups[0], fs, evalInterval, map[string]string{"cluster": "east-1"})
|
||||
g.Concurrency = 2
|
||||
|
||||
const inst1, inst2, job = "foo", "bar", "baz"
|
||||
m1 := metricWithLabels(t, "instance", inst1, "job", job)
|
||||
|
@ -195,7 +209,7 @@ func TestGroupStart(t *testing.T) {
|
|||
fs.add(m1)
|
||||
fs.add(m2)
|
||||
go func() {
|
||||
g.start(context.Background(), fs, []notifier.Notifier{fn}, nil)
|
||||
g.start(context.Background(), []notifier.Notifier{fn}, nil)
|
||||
close(finished)
|
||||
}()
|
||||
|
||||
|
|
|
@ -38,7 +38,11 @@ func (fq *fakeQuerier) add(metrics ...datasource.Metric) {
|
|||
fq.Unlock()
|
||||
}
|
||||
|
||||
func (fq *fakeQuerier) Query(_ context.Context, _ string, _ datasource.Type) ([]datasource.Metric, error) {
|
||||
func (fq *fakeQuerier) BuildWithParams(_ datasource.QuerierParams) datasource.Querier {
|
||||
return fq
|
||||
}
|
||||
|
||||
func (fq *fakeQuerier) Query(_ context.Context, _ string) ([]datasource.Metric, error) {
|
||||
fq.Lock()
|
||||
defer fq.Unlock()
|
||||
if fq.err != nil {
|
||||
|
@ -160,6 +164,9 @@ func compareAlertingRules(t *testing.T, a, b *AlertingRule) error {
|
|||
if !reflect.DeepEqual(a.Labels, b.Labels) {
|
||||
return fmt.Errorf("expected to have labels %#v; got %#v", a.Labels, b.Labels)
|
||||
}
|
||||
if a.Type.String() != b.Type.String() {
|
||||
return fmt.Errorf("expected to have Type %#v; got %#v", a.Type.String(), b.Type.String())
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
@ -140,10 +140,10 @@ func newManager(ctx context.Context) (*manager, error) {
|
|||
}
|
||||
|
||||
manager := &manager{
|
||||
groups: make(map[uint64]*Group),
|
||||
querier: q,
|
||||
notifiers: nts,
|
||||
labels: map[string]string{},
|
||||
groups: make(map[uint64]*Group),
|
||||
querierBuilder: q,
|
||||
notifiers: nts,
|
||||
labels: map[string]string{},
|
||||
}
|
||||
rw, err := remotewrite.Init(ctx)
|
||||
if err != nil {
|
||||
|
|
|
@ -15,11 +15,12 @@ import (
|
|||
|
||||
// manager controls group states
|
||||
type manager struct {
|
||||
querier datasource.Querier
|
||||
notifiers []notifier.Notifier
|
||||
querierBuilder datasource.QuerierBuilder
|
||||
notifiers []notifier.Notifier
|
||||
|
||||
rw *remotewrite.Client
|
||||
rr datasource.Querier
|
||||
// remote read builder.
|
||||
rr datasource.QuerierBuilder
|
||||
|
||||
wg sync.WaitGroup
|
||||
labels map[string]string
|
||||
|
@ -74,7 +75,7 @@ func (m *manager) startGroup(ctx context.Context, group *Group, restore bool) {
|
|||
m.wg.Add(1)
|
||||
id := group.ID()
|
||||
go func() {
|
||||
group.start(ctx, m.querier, m.notifiers, m.rw)
|
||||
group.start(ctx, m.notifiers, m.rw)
|
||||
m.wg.Done()
|
||||
}()
|
||||
m.groups[id] = group
|
||||
|
@ -89,7 +90,7 @@ func (m *manager) update(ctx context.Context, path []string, validateTpl, valida
|
|||
|
||||
groupsRegistry := make(map[uint64]*Group)
|
||||
for _, cfg := range groupsCfg {
|
||||
ng := newGroup(cfg, *evaluationInterval, m.labels)
|
||||
ng := newGroup(cfg, m.querierBuilder, *evaluationInterval, m.labels)
|
||||
groupsRegistry[ng.ID()] = ng
|
||||
}
|
||||
|
||||
|
|
|
@ -37,9 +37,9 @@ func TestManagerEmptyRulesDir(t *testing.T) {
|
|||
// Should be executed with -race flag
|
||||
func TestManagerUpdateConcurrent(t *testing.T) {
|
||||
m := &manager{
|
||||
groups: make(map[uint64]*Group),
|
||||
querier: &fakeQuerier{},
|
||||
notifiers: []notifier.Notifier{&fakeNotifier{}},
|
||||
groups: make(map[uint64]*Group),
|
||||
querierBuilder: &fakeQuerier{},
|
||||
notifiers: []notifier.Notifier{&fakeNotifier{}},
|
||||
}
|
||||
paths := []string{
|
||||
"config/testdata/dir/rules0-good.rules",
|
||||
|
@ -242,7 +242,7 @@ func TestManagerUpdate(t *testing.T) {
|
|||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.TODO())
|
||||
m := &manager{groups: make(map[uint64]*Group), querier: &fakeQuerier{}}
|
||||
m := &manager{groups: make(map[uint64]*Group), querierBuilder: &fakeQuerier{}}
|
||||
path := []string{tc.initPath}
|
||||
if err := m.update(ctx, path, true, true, false); err != nil {
|
||||
t.Fatalf("failed to complete initial rules update: %s", err)
|
||||
|
|
|
@ -25,6 +25,8 @@ type RecordingRule struct {
|
|||
Labels map[string]string
|
||||
GroupID uint64
|
||||
|
||||
q datasource.Querier
|
||||
|
||||
// guard status fields
|
||||
mu sync.RWMutex
|
||||
// stores last moment of time Exec was called
|
||||
|
@ -52,7 +54,7 @@ func (rr *RecordingRule) ID() uint64 {
|
|||
return rr.RuleID
|
||||
}
|
||||
|
||||
func newRecordingRule(group *Group, cfg config.Rule) *RecordingRule {
|
||||
func newRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule) *RecordingRule {
|
||||
rr := &RecordingRule{
|
||||
Type: cfg.Type,
|
||||
RuleID: cfg.ID,
|
||||
|
@ -61,6 +63,7 @@ func newRecordingRule(group *Group, cfg config.Rule) *RecordingRule {
|
|||
Labels: cfg.Labels,
|
||||
GroupID: group.ID(),
|
||||
metrics: &recordingRuleMetrics{},
|
||||
q: qb.BuildWithParams(datasource.QuerierParams{DataSourceType: &cfg.Type}),
|
||||
}
|
||||
|
||||
labels := fmt.Sprintf(`recording=%q, group=%q, id="%d"`, rr.Name, group.Name, rr.ID())
|
||||
|
@ -82,12 +85,12 @@ func (rr *RecordingRule) Close() {
|
|||
}
|
||||
|
||||
// Exec executes RecordingRule expression via the given Querier.
|
||||
func (rr *RecordingRule) Exec(ctx context.Context, q datasource.Querier, series bool) ([]prompbmarshal.TimeSeries, error) {
|
||||
func (rr *RecordingRule) Exec(ctx context.Context, series bool) ([]prompbmarshal.TimeSeries, error) {
|
||||
if !series {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
qMetrics, err := q.Query(ctx, rr.Expr, rr.Type)
|
||||
qMetrics, err := rr.q.Query(ctx, rr.Expr)
|
||||
rr.mu.Lock()
|
||||
defer rr.mu.Unlock()
|
||||
|
||||
|
|
|
@ -76,7 +76,8 @@ func TestRecoridngRule_ToTimeSeries(t *testing.T) {
|
|||
t.Run(tc.rule.Name, func(t *testing.T) {
|
||||
fq := &fakeQuerier{}
|
||||
fq.add(tc.metrics...)
|
||||
tss, err := tc.rule.Exec(context.TODO(), fq, true)
|
||||
tc.rule.q = fq
|
||||
tss, err := tc.rule.Exec(context.TODO(), true)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected Exec err: %s", err)
|
||||
}
|
||||
|
@ -95,8 +96,8 @@ func TestRecoridngRule_ToTimeSeriesNegative(t *testing.T) {
|
|||
fq := &fakeQuerier{}
|
||||
expErr := "connection reset by peer"
|
||||
fq.setErr(errors.New(expErr))
|
||||
|
||||
_, err := rr.Exec(context.TODO(), fq, true)
|
||||
rr.q = fq
|
||||
_, err := rr.Exec(context.TODO(), true)
|
||||
if err == nil {
|
||||
t.Fatalf("expected to get err; got nil")
|
||||
}
|
||||
|
@ -111,7 +112,7 @@ func TestRecoridngRule_ToTimeSeriesNegative(t *testing.T) {
|
|||
fq.add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "foo"))
|
||||
fq.add(metricWithValueAndLabels(t, 2, "__name__", "foo", "job", "bar"))
|
||||
|
||||
_, err = rr.Exec(context.TODO(), fq, true)
|
||||
_, err = rr.Exec(context.TODO(), true)
|
||||
if err == nil {
|
||||
t.Fatalf("expected to get err; got nil")
|
||||
}
|
||||
|
|
|
@ -26,7 +26,7 @@ var (
|
|||
|
||||
// Init creates a Querier from provided flag values.
|
||||
// Returns nil if addr flag wasn't set.
|
||||
func Init() (datasource.Querier, error) {
|
||||
func Init() (datasource.QuerierBuilder, error) {
|
||||
if *addr == "" {
|
||||
return nil, nil
|
||||
}
|
||||
|
|
|
@ -4,7 +4,6 @@ import (
|
|||
"context"
|
||||
"errors"
|
||||
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
|
||||
"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
|
||||
)
|
||||
|
||||
|
@ -18,7 +17,7 @@ type Rule interface {
|
|||
// Exec executes the rule with given context
|
||||
// and Querier. If returnSeries is true, Exec
|
||||
// may return TimeSeries as result of execution
|
||||
Exec(ctx context.Context, q datasource.Querier, returnSeries bool) ([]prompbmarshal.TimeSeries, error)
|
||||
Exec(ctx context.Context, returnSeries bool) ([]prompbmarshal.TimeSeries, error)
|
||||
// UpdateWith performs modification of current Rule
|
||||
// with fields of the given Rule.
|
||||
UpdateWith(Rule) error
|
||||
|
|
Loading…
Reference in a new issue