Alerting: Rule backtesting with experimental UI (#115525)
* add function to convert StateTransition to LokiEntry * add QueryResultBuilder * update backtesting to produce result similar to historian * make shouldRecord public * filter out noop transitions * add experimental front-end * add new fields * move conversion of api model to AlertRule to validation * add extra labels * calculate tick timestamp using the same logic as in scheduler * implement correct logic of calculating first evaluation timestamp * add uid, group and folder uid they are needed for jitter strategy * add JitterOffsetInDuration and JitterStrategy.String() * add config `backtesting_max_evaluations` to [unified_alerting] (not documented for now) * remove obsolete tests * elevate permisisons for backtesting endpoint * move backtesting to separate dir
This commit is contained in:
@@ -161,7 +161,7 @@ func (api *API) RegisterAPIEndpoints(m *metrics.API) {
|
||||
authz: ruleAuthzService,
|
||||
evaluator: api.EvaluatorFactory,
|
||||
cfg: &api.Cfg.UnifiedAlerting,
|
||||
backtesting: backtesting.NewEngine(api.AppUrl, api.EvaluatorFactory, api.Tracer),
|
||||
backtesting: backtesting.NewEngine(api.AppUrl, api.EvaluatorFactory, api.Tracer, api.Cfg.UnifiedAlerting, api.FeatureManager),
|
||||
featureManager: api.FeatureManager,
|
||||
appUrl: api.AppUrl,
|
||||
tracer: api.Tracer,
|
||||
|
||||
@@ -34,7 +34,6 @@ import (
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/state"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/store"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/util"
|
||||
)
|
||||
|
||||
type folderService interface {
|
||||
@@ -230,54 +229,27 @@ func (srv TestingApiSrv) BacktestAlertRule(c *contextmodel.ReqContext, cmd apimo
|
||||
return ErrResp(http.StatusNotFound, nil, "Backgtesting API is not enabled")
|
||||
}
|
||||
|
||||
if cmd.From.After(cmd.To) {
|
||||
return ErrResp(400, nil, "From cannot be greater than To")
|
||||
}
|
||||
|
||||
noDataState, err := ngmodels.NoDataStateFromString(string(cmd.NoDataState))
|
||||
|
||||
rule, err := apivalidation.ValidateBacktestConfig(c.GetOrgID(), cmd, apivalidation.RuleLimitsFromConfig(srv.cfg, srv.featureManager))
|
||||
if err != nil {
|
||||
return ErrResp(400, err, "")
|
||||
}
|
||||
forInterval := time.Duration(cmd.For)
|
||||
if forInterval < 0 {
|
||||
return ErrResp(400, nil, "Bad For interval")
|
||||
return ErrResp(http.StatusBadRequest, err, "")
|
||||
}
|
||||
|
||||
intervalSeconds, err := apivalidation.ValidateInterval(time.Duration(cmd.Interval), srv.cfg.BaseInterval)
|
||||
if err != nil {
|
||||
return ErrResp(400, err, "")
|
||||
}
|
||||
|
||||
queries := AlertQueriesFromApiAlertQueries(cmd.Data)
|
||||
if err := srv.authz.AuthorizeDatasourceAccessForRule(c.Req.Context(), c.SignedInUser, &ngmodels.AlertRule{Data: queries}); err != nil {
|
||||
if err := srv.authz.AuthorizeDatasourceAccessForRule(c.Req.Context(), c.SignedInUser, rule); err != nil {
|
||||
return errorToResponse(err)
|
||||
}
|
||||
|
||||
rule := &ngmodels.AlertRule{
|
||||
// ID: 0,
|
||||
// Updated: time.Time{},
|
||||
// Version: 0,
|
||||
// NamespaceUID: "",
|
||||
// DashboardUID: nil,
|
||||
// PanelID: nil,
|
||||
// RuleGroup: "",
|
||||
// RuleGroupIndex: 0,
|
||||
// ExecErrState: "",
|
||||
Title: cmd.Title,
|
||||
// prefix backtesting- is to distinguish between executions of regular rule and backtesting in logs (like expression engine, evaluator, state manager etc)
|
||||
UID: "backtesting-" + util.GenerateShortUID(),
|
||||
OrgID: c.GetOrgID(),
|
||||
Condition: cmd.Condition,
|
||||
Data: queries,
|
||||
IntervalSeconds: intervalSeconds,
|
||||
NoDataState: noDataState,
|
||||
For: forInterval,
|
||||
Annotations: cmd.Annotations,
|
||||
Labels: cmd.Labels,
|
||||
// Fetch folder path for alert labels, fallback to "Backtesting" if not available
|
||||
var folderTitle string
|
||||
if cmd.NamespaceUID != "" {
|
||||
f, err := srv.folderService.GetNamespaceByUID(c.Req.Context(), cmd.NamespaceUID, c.OrgID, c.SignedInUser)
|
||||
if err != nil {
|
||||
srv.log.FromContext(c.Req.Context()).Warn("Failed to fetch folder path for alert labels", "error", err)
|
||||
} else {
|
||||
folderTitle = f.Fullpath
|
||||
}
|
||||
}
|
||||
|
||||
result, err := srv.backtesting.Test(c.Req.Context(), c.SignedInUser, rule, cmd.From, cmd.To)
|
||||
result, err := srv.backtesting.Test(c.Req.Context(), c.SignedInUser, rule, cmd.From, cmd.To, folderTitle)
|
||||
if err != nil {
|
||||
if errors.Is(err, backtesting.ErrInvalidInputData) {
|
||||
return ErrResp(400, err, "Failed to evaluate")
|
||||
@@ -285,9 +257,5 @@ func (srv TestingApiSrv) BacktestAlertRule(c *contextmodel.ReqContext, cmd apimo
|
||||
return ErrResp(500, err, "Failed to evaluate")
|
||||
}
|
||||
|
||||
body, err := data.FrameToJSON(result, data.IncludeAll)
|
||||
if err != nil {
|
||||
return ErrResp(500, err, "Failed to convert frame to JSON")
|
||||
}
|
||||
return response.JSON(http.StatusOK, body)
|
||||
return response.JSONStreaming(http.StatusOK, result)
|
||||
}
|
||||
|
||||
@@ -81,9 +81,15 @@ func (api *API) authorize(method, path string) web.Handler {
|
||||
// additional authorization is done in the request handler
|
||||
eval = ac.EvalPermission(ac.ActionAlertingRuleRead)
|
||||
// Grafana Rules Testing Paths
|
||||
case http.MethodPost + "/api/v1/rule/backtest":
|
||||
case http.MethodPost + "/api/v1/rule/backtest": // TODO (yuri) this should be protected by dedicated permission
|
||||
// additional authorization is done in the request handler
|
||||
eval = ac.EvalPermission(ac.ActionAlertingRuleRead)
|
||||
eval = ac.EvalAll(
|
||||
ac.EvalPermission(ac.ActionAlertingRuleRead),
|
||||
ac.EvalAny(
|
||||
ac.EvalPermission(ac.ActionAlertingRuleUpdate),
|
||||
ac.EvalPermission(ac.ActionAlertingRuleCreate),
|
||||
),
|
||||
)
|
||||
case http.MethodPost + "/api/v1/eval":
|
||||
// additional authorization is done in the request handler
|
||||
eval = ac.EvalPermission(ac.ActionAlertingRuleRead)
|
||||
|
||||
@@ -221,15 +221,21 @@ type BacktestConfig struct {
|
||||
To time.Time `json:"to"`
|
||||
Interval model.Duration `json:"interval,omitempty"`
|
||||
|
||||
Condition string `json:"condition"`
|
||||
Data []AlertQuery `json:"data"`
|
||||
For model.Duration `json:"for,omitempty"`
|
||||
Condition string `json:"condition"`
|
||||
Data []AlertQuery `json:"data"`
|
||||
For *model.Duration `json:"for,omitempty"`
|
||||
KeepFiringFor *model.Duration `json:"keep_firing_for,omitempty"`
|
||||
|
||||
Title string `json:"title"`
|
||||
Labels map[string]string `json:"labels,omitempty"`
|
||||
Annotations map[string]string `json:"annotations,omitempty"`
|
||||
Title string `json:"title"`
|
||||
Labels map[string]string `json:"labels,omitempty"`
|
||||
|
||||
NoDataState NoDataState `json:"no_data_state"`
|
||||
NoDataState NoDataState `json:"no_data_state"`
|
||||
ExecErrState ExecutionErrorState `json:"exec_err_state"`
|
||||
MissingSeriesEvalsToResolve *int64 `json:"missing_series_evals_to_resolve,omitempty"`
|
||||
|
||||
UID string `json:"uid,omitempty"`
|
||||
RuleGroup string `json:"rule_group,omitempty"`
|
||||
NamespaceUID string `json:"namespace_uid,omitempty"`
|
||||
}
|
||||
|
||||
// swagger:model
|
||||
|
||||
@@ -249,6 +249,21 @@ func ValidateCondition(condition string, queries []apimodels.AlertQuery, canPatc
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateGroupInterval(incoming prommodels.Duration, limits RuleLimits) (time.Duration, error) {
|
||||
interval := time.Duration(incoming)
|
||||
if interval == 0 {
|
||||
// if group interval is 0 (undefined) then we automatically fall back to the default interval
|
||||
interval = limits.DefaultRuleEvaluationInterval
|
||||
}
|
||||
|
||||
if interval < 0 || int64(interval.Seconds())%int64(limits.BaseInterval.Seconds()) != 0 {
|
||||
return 0, fmt.Errorf("rule evaluation interval (%d second) should be positive number that is multiple of the base interval of %d seconds", int64(interval.Seconds()), int64(limits.BaseInterval.Seconds()))
|
||||
}
|
||||
|
||||
// TODO should we validate that interval is >= cfg.MinInterval? Currently, we allow to save but fix the specified interval if it is < cfg.MinInterval
|
||||
return interval, nil
|
||||
}
|
||||
|
||||
func ValidateInterval(interval, baseInterval time.Duration) (int64, error) {
|
||||
intervalSeconds := int64(interval.Seconds())
|
||||
|
||||
@@ -336,18 +351,11 @@ func ValidateRuleGroup(
|
||||
return nil, fmt.Errorf("rule group name is too long. Max length is %d", store.AlertRuleMaxRuleGroupNameLength)
|
||||
}
|
||||
|
||||
interval := time.Duration(ruleGroupConfig.Interval)
|
||||
if interval == 0 {
|
||||
// if group interval is 0 (undefined) then we automatically fall back to the default interval
|
||||
interval = limits.DefaultRuleEvaluationInterval
|
||||
interval, err := validateGroupInterval(ruleGroupConfig.Interval, limits)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if interval < 0 || int64(interval.Seconds())%int64(limits.BaseInterval.Seconds()) != 0 {
|
||||
return nil, fmt.Errorf("rule evaluation interval (%d second) should be positive number that is multiple of the base interval of %d seconds", int64(interval.Seconds()), int64(limits.BaseInterval.Seconds()))
|
||||
}
|
||||
|
||||
// TODO should we validate that interval is >= cfg.MinInterval? Currently, we allow to save but fix the specified interval if it is < cfg.MinInterval
|
||||
|
||||
// If the rule group is reserved for no-group rules, we cannot have multiple rules in it.
|
||||
if isNoGroupRuleGroup && len(ruleGroupConfig.Rules) > 1 {
|
||||
return nil, fmt.Errorf("rule group %s is reserved for no-group rules and cannot be used for rule groups with multiple rules", ruleGroupConfig.Name)
|
||||
@@ -410,3 +418,32 @@ func ValidateNotificationSettings(n *apimodels.AlertRuleNotificationSettings) ([
|
||||
s,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func ValidateBacktestConfig(orgId int64, config apimodels.BacktestConfig, limits RuleLimits) (*ngmodels.AlertRule, error) {
|
||||
if config.From.After(config.To) {
|
||||
return nil, fmt.Errorf("invalid testing range: from %s must be before to %s", config.From, config.To)
|
||||
}
|
||||
|
||||
interval, err := validateGroupInterval(config.Interval, limits)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return ValidateRuleNode(&apimodels.PostableExtendedRuleNode{
|
||||
ApiRuleNode: &apimodels.ApiRuleNode{
|
||||
For: config.For,
|
||||
KeepFiringFor: config.KeepFiringFor,
|
||||
Labels: config.Labels,
|
||||
Annotations: nil,
|
||||
},
|
||||
GrafanaManagedAlert: &apimodels.PostableGrafanaRule{
|
||||
Title: config.Title,
|
||||
Condition: config.Condition,
|
||||
Data: config.Data,
|
||||
UID: config.UID,
|
||||
NoDataState: config.NoDataState,
|
||||
ExecErrState: config.ExecErrState,
|
||||
MissingSeriesEvalsToResolve: config.MissingSeriesEvalsToResolve,
|
||||
},
|
||||
}, config.RuleGroup, interval, orgId, config.NamespaceUID, limits)
|
||||
}
|
||||
|
||||
@@ -15,10 +15,16 @@ import (
|
||||
"github.com/grafana/grafana/pkg/apimachinery/identity"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/eval"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/models"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/schedule"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/schedule/ticker"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/state"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/state/historian"
|
||||
history_model "github.com/grafana/grafana/pkg/services/ngalert/state/historian/model"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/util"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -28,7 +34,7 @@ var (
|
||||
backtestingEvaluatorFactory = newBacktestingEvaluator
|
||||
)
|
||||
|
||||
type callbackFunc = func(evaluationIndex int, now time.Time, results eval.Results) error
|
||||
type callbackFunc = func(evaluationIndex int, now time.Time, results eval.Results) (bool, error)
|
||||
|
||||
type backtestingEvaluator interface {
|
||||
Eval(ctx context.Context, from time.Time, interval time.Duration, evaluations int, callback callbackFunc) error
|
||||
@@ -40,11 +46,17 @@ type stateManager interface {
|
||||
}
|
||||
|
||||
type Engine struct {
|
||||
evalFactory eval.EvaluatorFactory
|
||||
createStateManager func() stateManager
|
||||
evalFactory eval.EvaluatorFactory
|
||||
createStateManager func() stateManager
|
||||
disableGrafanaFolder bool
|
||||
featureToggles featuremgmt.FeatureToggles
|
||||
minInterval time.Duration
|
||||
baseInterval time.Duration
|
||||
jitterStrategy schedule.JitterStrategy
|
||||
maxEvaluations int
|
||||
}
|
||||
|
||||
func NewEngine(appUrl *url.URL, evalFactory eval.EvaluatorFactory, tracer tracing.Tracer) *Engine {
|
||||
func NewEngine(appUrl *url.URL, evalFactory eval.EvaluatorFactory, tracer tracing.Tracer, cfg setting.UnifiedAlertingSettings, toggles featuremgmt.FeatureToggles) *Engine {
|
||||
return &Engine{
|
||||
evalFactory: evalFactory,
|
||||
createStateManager: func() stateManager {
|
||||
@@ -60,74 +72,139 @@ func NewEngine(appUrl *url.URL, evalFactory eval.EvaluatorFactory, tracer tracin
|
||||
}
|
||||
return state.NewManager(cfg, state.NewNoopPersister())
|
||||
},
|
||||
disableGrafanaFolder: false,
|
||||
featureToggles: toggles,
|
||||
minInterval: cfg.MinInterval,
|
||||
baseInterval: cfg.BaseInterval,
|
||||
maxEvaluations: cfg.BacktestingMaxEvaluations,
|
||||
jitterStrategy: schedule.JitterStrategyFrom(cfg, toggles),
|
||||
}
|
||||
}
|
||||
|
||||
func (e *Engine) Test(ctx context.Context, user identity.Requester, rule *models.AlertRule, from, to time.Time) (*data.Frame, error) {
|
||||
ruleCtx := models.WithRuleKey(ctx, rule.GetKey())
|
||||
logger := logger.FromContext(ctx)
|
||||
|
||||
func (e *Engine) Test(ctx context.Context, user identity.Requester, rule *models.AlertRule, from, to time.Time, folderTitle string) (res *data.Frame, err error) {
|
||||
if rule == nil {
|
||||
return nil, fmt.Errorf("%w: rule is not defined", ErrInvalidInputData)
|
||||
}
|
||||
if !from.Before(to) {
|
||||
return nil, fmt.Errorf("%w: invalid interval of the backtesting [%d,%d]", ErrInvalidInputData, from.Unix(), to.Unix())
|
||||
return nil, fmt.Errorf("%w: invalid interval [%d,%d]", ErrInvalidInputData, from.Unix(), to.Unix())
|
||||
}
|
||||
if to.Sub(from).Seconds() < float64(rule.IntervalSeconds) {
|
||||
return nil, fmt.Errorf("%w: interval of the backtesting [%d,%d] is less than evaluation interval [%ds]", ErrInvalidInputData, from.Unix(), to.Unix(), rule.IntervalSeconds)
|
||||
|
||||
ruleCtx := models.WithRuleKey(ctx, rule.GetKey())
|
||||
logger := logger.FromContext(ruleCtx).New("backtesting", util.GenerateShortUID())
|
||||
|
||||
var warns []string
|
||||
if rule.GetInterval() < e.minInterval {
|
||||
logger.Warn("Interval adjusted to minimal interval", "originalInterval", rule.GetInterval(), "adjustedInterval", e.minInterval)
|
||||
rule = rule.Copy()
|
||||
rule.IntervalSeconds = int64(e.minInterval.Seconds())
|
||||
warns = append(warns, fmt.Sprintf("Interval adjusted to minimal interval %ds", rule.IntervalSeconds))
|
||||
}
|
||||
length := int(to.Sub(from).Seconds()) / int(rule.IntervalSeconds)
|
||||
|
||||
stateManager := e.createStateManager()
|
||||
effectiveStrategy := e.jitterStrategy
|
||||
if e.jitterStrategy == schedule.JitterByGroup && (rule.RuleGroup == "" || rule.NamespaceUID == "") ||
|
||||
e.jitterStrategy == schedule.JitterByRule && rule.UID == "" {
|
||||
logger.Warn(fmt.Sprintf("Jitter strategy is set to %s, but rule group or namespace is not set. Ignore jitter", e.jitterStrategy))
|
||||
warns = append(warns, fmt.Sprintf("Jitter strategy is set to %s, but rule group or namespace is not set. Ignore jitter. The results of testing will be different than real evaluations", e.jitterStrategy))
|
||||
effectiveStrategy = schedule.JitterNever
|
||||
}
|
||||
jitterOffset := schedule.JitterOffsetInDuration(rule, e.baseInterval, effectiveStrategy)
|
||||
firstEval, err := getFirstEvaluationTime(from, rule, e.baseInterval, jitterOffset)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%w: %s", ErrInvalidInputData, err)
|
||||
}
|
||||
|
||||
evaluator, err := backtestingEvaluatorFactory(ruleCtx, e.evalFactory, user, rule.GetEvalCondition().WithSource("backtesting"), &schedule.AlertingResultsFromRuleState{
|
||||
Manager: stateManager,
|
||||
Rule: rule,
|
||||
})
|
||||
evaluations := calculateNumberOfEvaluations(firstEval, to, rule.GetInterval())
|
||||
if e.maxEvaluations > 0 && evaluations > e.maxEvaluations {
|
||||
logger.Warn("Evaluations adjusted to maximal number", "originalEvaluations", evaluations, "adjustedEvaluations", e.maxEvaluations)
|
||||
warns = append(warns, fmt.Sprintf("Number of evaluations are adjusted to the limit of %d evaluations. Requested: %d", e.maxEvaluations, evaluations))
|
||||
evaluations = e.maxEvaluations
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
if err == nil {
|
||||
logger.Info("Rule testing finished successfully", "duration", time.Since(start))
|
||||
} else {
|
||||
logger.Error("Rule testing finished with error", "duration", time.Since(start), "error", err)
|
||||
}
|
||||
}()
|
||||
|
||||
stateMgr := e.createStateManager()
|
||||
|
||||
evaluator, err := backtestingEvaluatorFactory(ruleCtx,
|
||||
e.evalFactory,
|
||||
user,
|
||||
rule.GetEvalCondition().WithSource("backtesting"),
|
||||
&schedule.AlertingResultsFromRuleState{
|
||||
Manager: stateMgr,
|
||||
Rule: rule,
|
||||
},
|
||||
)
|
||||
if err != nil {
|
||||
return nil, errors.Join(ErrInvalidInputData, err)
|
||||
}
|
||||
|
||||
logger.Info("Start testing alert rule", "from", from, "to", to, "interval", rule.IntervalSeconds, "evaluations", length)
|
||||
logger.Info("Start testing alert rule", "from", from, "to", to, "interval", rule.GetInterval(), "firstTick", firstEval, "evaluations", evaluations, "jitterOffset", jitterOffset, "jitterStrategy", effectiveStrategy)
|
||||
|
||||
start := time.Now()
|
||||
var builder *historian.QueryResultBuilder
|
||||
|
||||
tsField := data.NewField("Time", nil, make([]time.Time, length))
|
||||
valueFields := make(map[data.Fingerprint]*data.Field)
|
||||
|
||||
err = evaluator.Eval(ruleCtx, from, time.Duration(rule.IntervalSeconds)*time.Second, length, func(idx int, currentTime time.Time, results eval.Results) error {
|
||||
if idx >= length {
|
||||
logger.Info("Unexpected evaluation. Skipping", "from", from, "to", to, "interval", rule.IntervalSeconds, "evaluationTime", currentTime, "evaluationIndex", idx, "expectedEvaluations", length)
|
||||
return nil
|
||||
}
|
||||
states := stateManager.ProcessEvalResults(ruleCtx, currentTime, rule, results, nil, nil)
|
||||
tsField.Set(idx, currentTime)
|
||||
for _, s := range states {
|
||||
field, ok := valueFields[s.CacheID]
|
||||
if !ok {
|
||||
field = data.NewField("", s.Labels, make([]*string, length))
|
||||
valueFields[s.CacheID] = field
|
||||
}
|
||||
if s.State.State != eval.NoData { // set nil if NoData
|
||||
value := s.State.State.String()
|
||||
if s.StateReason != "" {
|
||||
value += " (" + s.StateReason + ")"
|
||||
}
|
||||
field.Set(idx, &value)
|
||||
continue
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
fields := make([]*data.Field, 0, len(valueFields)+1)
|
||||
fields = append(fields, tsField)
|
||||
for _, f := range valueFields {
|
||||
fields = append(fields, f)
|
||||
ruleMeta := history_model.RuleMeta{
|
||||
ID: rule.ID,
|
||||
OrgID: rule.OrgID,
|
||||
UID: rule.UID,
|
||||
Title: rule.Title,
|
||||
Group: rule.RuleGroup,
|
||||
NamespaceUID: rule.NamespaceUID,
|
||||
// DashboardUID: "",
|
||||
// PanelID: 0,
|
||||
Condition: rule.Condition,
|
||||
}
|
||||
result := data.NewFrame("Testing results", fields...)
|
||||
|
||||
labels := map[string]string{
|
||||
historian.OrgIDLabel: fmt.Sprint(ruleMeta.OrgID),
|
||||
historian.GroupLabel: fmt.Sprint(ruleMeta.Group),
|
||||
historian.FolderUIDLabel: fmt.Sprint(rule.NamespaceUID),
|
||||
}
|
||||
labelsBytes, err := json.Marshal(labels)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
logger.Info("Rule testing finished successfully", "duration", time.Since(start))
|
||||
return result, nil
|
||||
|
||||
// Ensure fallback if empty string is passed
|
||||
if folderTitle == "" {
|
||||
folderTitle = "Backtesting"
|
||||
}
|
||||
extraLabels := state.GetRuleExtraLabels(logger, rule, folderTitle, !e.disableGrafanaFolder, e.featureToggles)
|
||||
|
||||
processFn := func(idx int, currentTime time.Time, results eval.Results) (bool, error) {
|
||||
// init the builder. Do the best guess for the size of the result
|
||||
if builder == nil {
|
||||
builder = historian.NewQueryResultBuilder(evaluations * len(results))
|
||||
for _, warn := range warns {
|
||||
builder.AddWarn(warn)
|
||||
}
|
||||
}
|
||||
states := stateMgr.ProcessEvalResults(ruleCtx, currentTime, rule, results, extraLabels, nil)
|
||||
for _, s := range states {
|
||||
if !historian.ShouldRecord(s) {
|
||||
continue
|
||||
}
|
||||
entry := historian.StateTransitionToLokiEntry(ruleMeta, s)
|
||||
err := builder.AddRow(currentTime, entry, labelsBytes)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
}
|
||||
return idx <= evaluations, nil
|
||||
}
|
||||
|
||||
err = evaluator.Eval(ruleCtx, firstEval, rule.GetInterval(), evaluations, processFn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if builder == nil {
|
||||
return nil, errors.New("no results were produced")
|
||||
}
|
||||
return builder.ToFrame(), nil
|
||||
}
|
||||
|
||||
func newBacktestingEvaluator(ctx context.Context, evalFactory eval.EvaluatorFactory, user identity.Requester, condition models.Condition, reader eval.AlertingResultsReader) (backtestingEvaluator, error) {
|
||||
@@ -173,3 +250,53 @@ type NoopImageService struct{}
|
||||
func (s *NoopImageService) NewImage(_ context.Context, _ *models.AlertRule) (*models.Image, error) {
|
||||
return &models.Image{}, nil
|
||||
}
|
||||
|
||||
func getNextEvaluationTime(currentTime time.Time, rule *models.AlertRule, baseInterval time.Duration, jitterOffset time.Duration) (time.Time, error) {
|
||||
if rule.IntervalSeconds%int64(baseInterval.Seconds()) != 0 {
|
||||
return time.Time{}, fmt.Errorf("interval %ds is not divisible by base interval %ds", rule.IntervalSeconds, int64(baseInterval.Seconds()))
|
||||
}
|
||||
|
||||
freq := rule.IntervalSeconds / int64(baseInterval.Seconds())
|
||||
|
||||
firstTickNum := currentTime.Unix() / int64(baseInterval.Seconds())
|
||||
|
||||
jitterOffsetTicks := int64(jitterOffset / baseInterval)
|
||||
|
||||
firstEvalTickNum := firstTickNum + (jitterOffsetTicks-(firstTickNum%freq)+freq)%freq
|
||||
|
||||
return time.Unix(firstEvalTickNum*int64(baseInterval.Seconds()), 0), nil
|
||||
}
|
||||
|
||||
func getFirstEvaluationTime(from time.Time, rule *models.AlertRule, baseInterval time.Duration, jitterOffset time.Duration) (time.Time, error) {
|
||||
// Now calculate the time of the tick the same way as in the scheduler
|
||||
firstTick := ticker.GetStartTick(from, baseInterval)
|
||||
|
||||
// calculate time of the first evaluation that is at or after the first tick
|
||||
firstEval, err := getNextEvaluationTime(firstTick, rule, baseInterval, jitterOffset)
|
||||
if err != nil {
|
||||
return time.Time{}, err
|
||||
}
|
||||
|
||||
// Ensure firstEval is at or after from
|
||||
// Calculate how many intervals to skip to get past 'from'
|
||||
if firstEval.Before(from) {
|
||||
diff := from.Sub(firstEval)
|
||||
interval := rule.GetInterval()
|
||||
// Ceiling division: how many intervals needed to cover the difference
|
||||
intervalsToAdd := (diff + interval - 1) / interval
|
||||
firstEval = firstEval.Add(interval * intervalsToAdd)
|
||||
}
|
||||
|
||||
return firstEval, nil
|
||||
}
|
||||
|
||||
func calculateNumberOfEvaluations(firstEval, to time.Time, interval time.Duration) int {
|
||||
var evaluations int
|
||||
if to.After(firstEval) {
|
||||
evaluations = int(to.Sub(firstEval).Seconds()) / int(interval.Seconds())
|
||||
}
|
||||
if evaluations == 0 {
|
||||
evaluations = 1
|
||||
}
|
||||
return evaluations
|
||||
}
|
||||
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -14,9 +13,11 @@ import (
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apimachinery/identity"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/eval"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/eval/eval_mocks"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/models"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/schedule"
|
||||
"github.com/grafana/grafana/pkg/services/ngalert/state"
|
||||
"github.com/grafana/grafana/pkg/util"
|
||||
)
|
||||
@@ -158,16 +159,6 @@ func TestNewBacktestingEvaluator(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestEvaluatorTest(t *testing.T) {
|
||||
states := []eval.State{eval.Normal, eval.Alerting, eval.Pending}
|
||||
generateState := func(prefix string) *state.State {
|
||||
labels := models.GenerateAlertLabels(rand.Intn(5)+1, prefix+"-")
|
||||
return &state.State{
|
||||
CacheID: labels.Fingerprint(),
|
||||
Labels: labels,
|
||||
State: states[rand.Intn(len(states))],
|
||||
}
|
||||
}
|
||||
|
||||
randomResultCallback := func(now time.Time) (eval.Results, error) {
|
||||
return eval.GenerateResults(rand.Intn(5)+1, eval.ResultGen()), nil
|
||||
}
|
||||
@@ -189,84 +180,17 @@ func TestEvaluatorTest(t *testing.T) {
|
||||
createStateManager: func() stateManager {
|
||||
return manager
|
||||
},
|
||||
disableGrafanaFolder: false,
|
||||
featureToggles: featuremgmt.WithFeatures(),
|
||||
minInterval: 1 * time.Second,
|
||||
baseInterval: 1 * time.Second,
|
||||
jitterStrategy: schedule.JitterNever,
|
||||
maxEvaluations: 10000,
|
||||
}
|
||||
gen := models.RuleGen
|
||||
rule := gen.With(gen.WithInterval(time.Second)).GenerateRef()
|
||||
ruleInterval := time.Duration(rule.IntervalSeconds) * time.Second
|
||||
|
||||
t.Run("should return data frame in specific format", func(t *testing.T) {
|
||||
from := time.Unix(0, 0)
|
||||
to := from.Add(5 * ruleInterval)
|
||||
allStates := [...]eval.State{eval.Normal, eval.Alerting, eval.Pending, eval.NoData, eval.Error}
|
||||
|
||||
var states []state.StateTransition
|
||||
|
||||
for _, s := range allStates {
|
||||
labels := models.GenerateAlertLabels(rand.Intn(5)+1, s.String()+"-")
|
||||
states = append(states, state.StateTransition{
|
||||
State: &state.State{
|
||||
CacheID: labels.Fingerprint(),
|
||||
Labels: labels,
|
||||
State: s,
|
||||
StateReason: util.GenerateShortUID(),
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
manager.stateCallback = func(now time.Time) []state.StateTransition {
|
||||
return states
|
||||
}
|
||||
|
||||
frame, err := engine.Test(context.Background(), nil, rule, from, to)
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Len(t, frame.Fields, len(states)+1) // +1 - timestamp
|
||||
|
||||
t.Run("should contain field Time", func(t *testing.T) {
|
||||
timestampField, _ := frame.FieldByName("Time")
|
||||
require.NotNil(t, timestampField, "frame does not contain field 'Time'")
|
||||
require.Equal(t, data.FieldTypeTime, timestampField.Type())
|
||||
})
|
||||
|
||||
fieldByState := make(map[data.Fingerprint]*data.Field, len(states))
|
||||
|
||||
t.Run("should contain a field per state", func(t *testing.T) {
|
||||
for _, s := range states {
|
||||
var f *data.Field
|
||||
for _, field := range frame.Fields {
|
||||
if field.Labels.String() == s.Labels.String() {
|
||||
f = field
|
||||
break
|
||||
}
|
||||
}
|
||||
require.NotNilf(t, f, "Cannot find a field by state labels")
|
||||
fieldByState[s.CacheID] = f
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("should be populated with correct values", func(t *testing.T) {
|
||||
timestampField, _ := frame.FieldByName("Time")
|
||||
expectedLength := timestampField.Len()
|
||||
for _, field := range frame.Fields {
|
||||
require.Equalf(t, expectedLength, field.Len(), "Field %s should have the size %d", field.Name, expectedLength)
|
||||
}
|
||||
for i := 0; i < expectedLength; i++ {
|
||||
expectedTime := from.Add(time.Duration(int64(i)*rule.IntervalSeconds) * time.Second)
|
||||
require.Equal(t, expectedTime, timestampField.At(i).(time.Time))
|
||||
for _, s := range states {
|
||||
f := fieldByState[s.CacheID]
|
||||
if s.State.State == eval.NoData {
|
||||
require.Nil(t, f.At(i))
|
||||
} else {
|
||||
v := f.At(i).(*string)
|
||||
require.NotNilf(t, v, "Field [%s] value at index %d should not be nil", s.CacheID, i)
|
||||
require.Equal(t, fmt.Sprintf("%s (%s)", s.State.State, s.StateReason), *v)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("should not fail if 'to-from' is not times of interval", func(t *testing.T) {
|
||||
from := time.Unix(0, 0)
|
||||
to := from.Add(5 * ruleInterval)
|
||||
@@ -287,84 +211,26 @@ func TestEvaluatorTest(t *testing.T) {
|
||||
return states
|
||||
}
|
||||
|
||||
frame, err := engine.Test(context.Background(), nil, rule, from, to)
|
||||
frame, err := engine.Test(context.Background(), nil, rule, from, to, "")
|
||||
require.NoError(t, err)
|
||||
expectedLen := frame.Rows()
|
||||
for i := 0; i < 100; i++ {
|
||||
jitter := time.Duration(rand.Int63n(ruleInterval.Milliseconds())) * time.Millisecond
|
||||
frame, err = engine.Test(context.Background(), nil, rule, from, to.Add(jitter))
|
||||
frame, err = engine.Test(context.Background(), nil, rule, from, to.Add(jitter), "")
|
||||
require.NoError(t, err)
|
||||
require.Equalf(t, expectedLen, frame.Rows(), "jitter %v caused result to be different that base-line", jitter)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("should backfill field with nulls if a new dimension created in the middle", func(t *testing.T) {
|
||||
from := time.Unix(0, 0)
|
||||
|
||||
state1 := state.StateTransition{
|
||||
State: generateState("1"),
|
||||
}
|
||||
state2 := state.StateTransition{
|
||||
State: generateState("2"),
|
||||
}
|
||||
state3 := state.StateTransition{
|
||||
State: generateState("3"),
|
||||
}
|
||||
stateByTime := map[time.Time][]state.StateTransition{
|
||||
from: {state1, state2},
|
||||
from.Add(1 * ruleInterval): {state1, state2},
|
||||
from.Add(2 * ruleInterval): {state1, state2},
|
||||
from.Add(3 * ruleInterval): {state1, state2, state3},
|
||||
from.Add(4 * ruleInterval): {state1, state2, state3},
|
||||
}
|
||||
to := from.Add(time.Duration(len(stateByTime)) * ruleInterval)
|
||||
|
||||
manager.stateCallback = func(now time.Time) []state.StateTransition {
|
||||
return stateByTime[now]
|
||||
}
|
||||
|
||||
frame, err := engine.Test(context.Background(), nil, rule, from, to)
|
||||
require.NoError(t, err)
|
||||
|
||||
var field3 *data.Field
|
||||
for _, field := range frame.Fields {
|
||||
if field.Labels.String() == state3.Labels.String() {
|
||||
field3 = field
|
||||
break
|
||||
}
|
||||
}
|
||||
require.NotNilf(t, field3, "Result for state 3 was not found")
|
||||
require.Equalf(t, len(stateByTime), field3.Len(), "State3 result has unexpected number of values")
|
||||
|
||||
idx := 0
|
||||
for curTime, states := range stateByTime {
|
||||
value := field3.At(idx).(*string)
|
||||
if len(states) == 2 {
|
||||
require.Nilf(t, value, "The result should be nil if state3 was not available for time %v", curTime)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("should fail", func(t *testing.T) {
|
||||
manager.stateCallback = func(now time.Time) []state.StateTransition {
|
||||
return nil
|
||||
}
|
||||
|
||||
t.Run("when interval is not correct", func(t *testing.T) {
|
||||
from := time.Now()
|
||||
t.Run("when from=to", func(t *testing.T) {
|
||||
to := from
|
||||
_, err := engine.Test(context.Background(), nil, rule, from, to)
|
||||
require.ErrorIs(t, err, ErrInvalidInputData)
|
||||
})
|
||||
t.Run("when from > to", func(t *testing.T) {
|
||||
to := from.Add(-ruleInterval)
|
||||
_, err := engine.Test(context.Background(), nil, rule, from, to)
|
||||
require.ErrorIs(t, err, ErrInvalidInputData)
|
||||
})
|
||||
t.Run("when to-from < interval", func(t *testing.T) {
|
||||
to := from.Add(ruleInterval).Add(-time.Millisecond)
|
||||
_, err := engine.Test(context.Background(), nil, rule, from, to)
|
||||
_, err := engine.Test(context.Background(), nil, rule, from, to, "")
|
||||
require.ErrorIs(t, err, ErrInvalidInputData)
|
||||
})
|
||||
})
|
||||
@@ -376,7 +242,7 @@ func TestEvaluatorTest(t *testing.T) {
|
||||
}
|
||||
from := time.Now()
|
||||
to := from.Add(ruleInterval)
|
||||
_, err := engine.Test(context.Background(), nil, rule, from, to)
|
||||
_, err := engine.Test(context.Background(), nil, rule, from, to, "")
|
||||
require.ErrorIs(t, err, expectedError)
|
||||
})
|
||||
})
|
||||
@@ -404,10 +270,188 @@ func (f *fakeBacktestingEvaluator) Eval(_ context.Context, from time.Time, inter
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = callback(idx, now, results)
|
||||
c, err := callback(idx, now, results)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !c {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func TestGetNextEvaluationTime(t *testing.T) {
|
||||
baseInterval := 10 * time.Second
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
ruleInterval int64
|
||||
currentTimestamp int64
|
||||
jitterOffset time.Duration
|
||||
expectError bool
|
||||
expectedNext int64
|
||||
}{
|
||||
{
|
||||
name: "interval not divisible by base interval",
|
||||
ruleInterval: 15,
|
||||
currentTimestamp: 0,
|
||||
jitterOffset: 0,
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from tick 0",
|
||||
ruleInterval: 20,
|
||||
currentTimestamp: 0,
|
||||
jitterOffset: 0,
|
||||
expectedNext: 0,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from tick 1",
|
||||
ruleInterval: 20,
|
||||
currentTimestamp: 10,
|
||||
jitterOffset: 0,
|
||||
expectedNext: 20,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from tick 2",
|
||||
ruleInterval: 20,
|
||||
currentTimestamp: 20,
|
||||
jitterOffset: 0,
|
||||
expectedNext: 20,
|
||||
},
|
||||
{
|
||||
name: "with 20s jitter - from tick 0",
|
||||
ruleInterval: 60,
|
||||
currentTimestamp: 0,
|
||||
jitterOffset: 20 * time.Second,
|
||||
expectedNext: 20,
|
||||
},
|
||||
{
|
||||
name: "with 20s jitter - from tick 2",
|
||||
ruleInterval: 60,
|
||||
currentTimestamp: 20,
|
||||
jitterOffset: 20 * time.Second,
|
||||
expectedNext: 20,
|
||||
},
|
||||
{
|
||||
name: "with 20s jitter - from tick 3",
|
||||
ruleInterval: 60,
|
||||
currentTimestamp: 30,
|
||||
jitterOffset: 20 * time.Second,
|
||||
expectedNext: 80,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
rule := &models.AlertRule{IntervalSeconds: tc.ruleInterval}
|
||||
currentTime := time.Unix(tc.currentTimestamp, 0)
|
||||
result, err := getNextEvaluationTime(currentTime, rule, baseInterval, tc.jitterOffset)
|
||||
|
||||
if tc.expectError {
|
||||
require.Error(t, err)
|
||||
require.Contains(t, err.Error(), "is not divisible by base interval")
|
||||
return
|
||||
}
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, tc.expectedNext, result.Unix())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetFirstEvaluationTime(t *testing.T) {
|
||||
baseInterval := 10 * time.Second
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
ruleInterval int64
|
||||
fromUnix int64
|
||||
jitterOffset time.Duration
|
||||
expectError bool
|
||||
expectedUnix int64
|
||||
}{
|
||||
{
|
||||
name: "interval not divisible by base interval",
|
||||
ruleInterval: 15,
|
||||
fromUnix: 0,
|
||||
jitterOffset: 0,
|
||||
expectError: true,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from at tick 0",
|
||||
ruleInterval: 20,
|
||||
fromUnix: 0,
|
||||
jitterOffset: 0,
|
||||
expectedUnix: 0,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from at tick 1",
|
||||
ruleInterval: 20,
|
||||
fromUnix: 10,
|
||||
jitterOffset: 0,
|
||||
expectedUnix: 20,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from before first tick",
|
||||
ruleInterval: 20,
|
||||
fromUnix: 5,
|
||||
jitterOffset: 0,
|
||||
expectedUnix: 20,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from after first aligned tick",
|
||||
ruleInterval: 20,
|
||||
fromUnix: 25,
|
||||
jitterOffset: 0,
|
||||
expectedUnix: 40,
|
||||
},
|
||||
{
|
||||
name: "no jitter - from at tick boundary",
|
||||
ruleInterval: 10,
|
||||
fromUnix: 10,
|
||||
jitterOffset: 0,
|
||||
expectedUnix: 10,
|
||||
},
|
||||
{
|
||||
name: "with 20s jitter - from epoch",
|
||||
ruleInterval: 60,
|
||||
fromUnix: 0,
|
||||
jitterOffset: 20 * time.Second,
|
||||
expectedUnix: 20,
|
||||
},
|
||||
{
|
||||
name: "with 20s jitter - from 70s",
|
||||
ruleInterval: 60,
|
||||
fromUnix: 70,
|
||||
jitterOffset: 20 * time.Second,
|
||||
expectedUnix: 80,
|
||||
},
|
||||
{
|
||||
name: "with 50s jitter - from 25s",
|
||||
ruleInterval: 60,
|
||||
fromUnix: 25,
|
||||
jitterOffset: 50 * time.Second,
|
||||
expectedUnix: 50,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
rule := &models.AlertRule{IntervalSeconds: tc.ruleInterval}
|
||||
from := time.Unix(tc.fromUnix, 0)
|
||||
result, err := getFirstEvaluationTime(from, rule, baseInterval, tc.jitterOffset)
|
||||
|
||||
if tc.expectError {
|
||||
require.Error(t, err)
|
||||
require.Contains(t, err.Error(), "is not divisible by base interval")
|
||||
return
|
||||
}
|
||||
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, tc.expectedUnix, result.Unix())
|
||||
require.GreaterOrEqual(t, result.Unix(), from.Unix(), "first eval should be at or after from")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -85,10 +85,13 @@ func (d *dataEvaluator) Eval(_ context.Context, from time.Time, interval time.Du
|
||||
EvaluatedAt: now,
|
||||
})
|
||||
}
|
||||
err := callback(i, now, result)
|
||||
cont, err := callback(i, now, result)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !cont {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -100,11 +100,11 @@ func TestDataEvaluator_Eval(t *testing.T) {
|
||||
|
||||
resultsCount := int(to.Sub(from).Seconds() / interval.Seconds())
|
||||
|
||||
err = evaluator.Eval(context.Background(), from, time.Second, resultsCount, func(idx int, now time.Time, res eval.Results) error {
|
||||
err = evaluator.Eval(context.Background(), from, time.Second, resultsCount, func(idx int, now time.Time, res eval.Results) (bool, error) {
|
||||
r = append(r, results{
|
||||
now, res,
|
||||
})
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
@@ -164,11 +164,11 @@ func TestDataEvaluator_Eval(t *testing.T) {
|
||||
size := to.Sub(from).Milliseconds() / interval.Milliseconds()
|
||||
r := make([]results, 0, size)
|
||||
|
||||
err = evaluator.Eval(context.Background(), from, interval, int(size), func(idx int, now time.Time, res eval.Results) error {
|
||||
err = evaluator.Eval(context.Background(), from, interval, int(size), func(idx int, now time.Time, res eval.Results) (bool, error) {
|
||||
r = append(r, results{
|
||||
now, res,
|
||||
})
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
|
||||
currentRowIdx := 0
|
||||
@@ -195,11 +195,11 @@ func TestDataEvaluator_Eval(t *testing.T) {
|
||||
size := int(to.Sub(from).Seconds() / interval.Seconds())
|
||||
r := make([]results, 0, size)
|
||||
|
||||
err = evaluator.Eval(context.Background(), from, interval, size, func(idx int, now time.Time, res eval.Results) error {
|
||||
err = evaluator.Eval(context.Background(), from, interval, size, func(idx int, now time.Time, res eval.Results) (bool, error) {
|
||||
r = append(r, results{
|
||||
now, res,
|
||||
})
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
|
||||
currentRowIdx := 0
|
||||
@@ -230,11 +230,11 @@ func TestDataEvaluator_Eval(t *testing.T) {
|
||||
t.Run("should be noData until the frame interval", func(t *testing.T) {
|
||||
newFrom := from.Add(-10 * time.Second)
|
||||
r := make([]results, 0, int(to.Sub(newFrom).Seconds()))
|
||||
err = evaluator.Eval(context.Background(), newFrom, time.Second, cap(r), func(idx int, now time.Time, res eval.Results) error {
|
||||
err = evaluator.Eval(context.Background(), newFrom, time.Second, cap(r), func(idx int, now time.Time, res eval.Results) (bool, error) {
|
||||
r = append(r, results{
|
||||
now, res,
|
||||
})
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
|
||||
rowIdx := 0
|
||||
@@ -258,11 +258,11 @@ func TestDataEvaluator_Eval(t *testing.T) {
|
||||
t.Run("should be the last value after the frame interval", func(t *testing.T) {
|
||||
newTo := to.Add(10 * time.Second)
|
||||
r := make([]results, 0, int(newTo.Sub(from).Seconds()))
|
||||
err = evaluator.Eval(context.Background(), from, time.Second, cap(r), func(idx int, now time.Time, res eval.Results) error {
|
||||
err = evaluator.Eval(context.Background(), from, time.Second, cap(r), func(idx int, now time.Time, res eval.Results) (bool, error) {
|
||||
r = append(r, results{
|
||||
now, res,
|
||||
})
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
|
||||
rowIdx := 0
|
||||
@@ -282,12 +282,21 @@ func TestDataEvaluator_Eval(t *testing.T) {
|
||||
})
|
||||
t.Run("should stop if callback error", func(t *testing.T) {
|
||||
expectedError := errors.New("error")
|
||||
err = evaluator.Eval(context.Background(), from, time.Second, 6, func(idx int, now time.Time, res eval.Results) error {
|
||||
err = evaluator.Eval(context.Background(), from, time.Second, 6, func(idx int, now time.Time, res eval.Results) (bool, error) {
|
||||
if idx == 5 {
|
||||
return expectedError
|
||||
return false, expectedError
|
||||
}
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
require.ErrorIs(t, err, expectedError)
|
||||
})
|
||||
t.Run("should stop if callback does not want to continue", func(t *testing.T) {
|
||||
evaluated := 0
|
||||
err = evaluator.Eval(context.Background(), from, time.Second, 6, func(idx int, now time.Time, res eval.Results) (bool, error) {
|
||||
evaluated++
|
||||
return evaluated < 2, nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 2, evaluated)
|
||||
})
|
||||
}
|
||||
|
||||
@@ -18,10 +18,13 @@ func (d *queryEvaluator) Eval(ctx context.Context, from time.Time, interval time
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
err = callback(idx, now, results)
|
||||
cont, err := callback(idx, now, results)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !cont {
|
||||
break
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -31,9 +31,9 @@ func TestQueryEvaluator_Eval(t *testing.T) {
|
||||
|
||||
intervals := make([]time.Time, times)
|
||||
|
||||
err := evaluator.Eval(ctx, from, interval, times, func(idx int, now time.Time, results eval.Results) error {
|
||||
err := evaluator.Eval(ctx, from, interval, times, func(idx int, now time.Time, results eval.Results) (bool, error) {
|
||||
intervals[idx] = now
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, intervals, times)
|
||||
@@ -49,7 +49,7 @@ func TestQueryEvaluator_Eval(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("should stop evaluation if error", func(t *testing.T) {
|
||||
t.Run("should stop evaluation", func(t *testing.T) {
|
||||
t.Run("when evaluation fails", func(t *testing.T) {
|
||||
m := &eval_mocks.ConditionEvaluatorMock{}
|
||||
expectedResults := eval.Results{}
|
||||
@@ -62,9 +62,9 @@ func TestQueryEvaluator_Eval(t *testing.T) {
|
||||
|
||||
intervals := make([]time.Time, 0, times)
|
||||
|
||||
err := evaluator.Eval(ctx, from, interval, times, func(idx int, now time.Time, results eval.Results) error {
|
||||
err := evaluator.Eval(ctx, from, interval, times, func(idx int, now time.Time, results eval.Results) (bool, error) {
|
||||
intervals = append(intervals, now)
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
require.ErrorIs(t, err, expectedError)
|
||||
require.Len(t, intervals, 3)
|
||||
@@ -81,14 +81,31 @@ func TestQueryEvaluator_Eval(t *testing.T) {
|
||||
|
||||
intervals := make([]time.Time, 0, times)
|
||||
|
||||
err := evaluator.Eval(ctx, from, interval, times, func(idx int, now time.Time, results eval.Results) error {
|
||||
err := evaluator.Eval(ctx, from, interval, times, func(idx int, now time.Time, results eval.Results) (bool, error) {
|
||||
if len(intervals) > 3 {
|
||||
return expectedError
|
||||
return false, expectedError
|
||||
}
|
||||
intervals = append(intervals, now)
|
||||
return nil
|
||||
return true, nil
|
||||
})
|
||||
require.ErrorIs(t, err, expectedError)
|
||||
})
|
||||
|
||||
t.Run("when callback does not want to continue", func(t *testing.T) {
|
||||
m := &eval_mocks.ConditionEvaluatorMock{}
|
||||
expectedResults := eval.Results{}
|
||||
m.EXPECT().Evaluate(mock.Anything, mock.Anything).Return(expectedResults, nil)
|
||||
evaluator := queryEvaluator{
|
||||
eval: m,
|
||||
}
|
||||
|
||||
evaluated := 0
|
||||
err := evaluator.Eval(ctx, from, interval, times, func(idx int, now time.Time, results eval.Results) (bool, error) {
|
||||
evaluated++
|
||||
return evaluated <= 2, nil
|
||||
})
|
||||
require.NoError(t, err, nil)
|
||||
require.Equal(t, 3, evaluated)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
@@ -480,6 +480,10 @@ func (alertRule *AlertRule) GetPanelID() int64 {
|
||||
return -1
|
||||
}
|
||||
|
||||
func (alertRule *AlertRule) GetInterval() time.Duration {
|
||||
return time.Duration(alertRule.IntervalSeconds) * time.Second
|
||||
}
|
||||
|
||||
type LabelOption func(map[string]string)
|
||||
|
||||
func WithoutInternalLabels() LabelOption {
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/data"
|
||||
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
ngmodels "github.com/grafana/grafana/pkg/services/ngalert/models"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
@@ -13,6 +14,10 @@ import (
|
||||
// JitterStrategy represents a modifier to alert rule timing that affects how evaluations are distributed.
|
||||
type JitterStrategy int
|
||||
|
||||
func (s JitterStrategy) String() string {
|
||||
return [...]string{"never", "by group", "by rule"}[s]
|
||||
}
|
||||
|
||||
const (
|
||||
JitterNever JitterStrategy = iota
|
||||
JitterByGroup
|
||||
@@ -57,6 +62,11 @@ func jitterOffsetInTicks(r *ngmodels.AlertRule, baseInterval time.Duration, stra
|
||||
return res
|
||||
}
|
||||
|
||||
// JitterOffsetInDuration gives the jitter offset for a rule, in terms of a duration relative to its interval and a base interval.
|
||||
func JitterOffsetInDuration(r *ngmodels.AlertRule, baseInterval time.Duration, strategy JitterStrategy) time.Duration {
|
||||
return time.Duration(jitterOffsetInTicks(r, baseInterval, strategy)) * baseInterval
|
||||
}
|
||||
|
||||
func jitterHash(r *ngmodels.AlertRule, strategy JitterStrategy) uint64 {
|
||||
ls := data.Labels{
|
||||
"name": r.RuleGroup,
|
||||
|
||||
@@ -44,7 +44,11 @@ func New(c clock.Clock, interval time.Duration, metric *Metrics, logger log.Logg
|
||||
}
|
||||
|
||||
func getStartTick(clk clock.Clock, interval time.Duration) time.Time {
|
||||
nano := clk.Now().UnixNano()
|
||||
return GetStartTick(clk.Now(), interval)
|
||||
}
|
||||
|
||||
func GetStartTick(t time.Time, interval time.Duration) time.Time {
|
||||
nano := t.UnixNano()
|
||||
return time.Unix(0, nano-(nano%interval.Nanoseconds()))
|
||||
}
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ import (
|
||||
|
||||
const StateHistoryWriteTimeout = time.Minute
|
||||
|
||||
func shouldRecord(transition state.StateTransition) bool {
|
||||
func ShouldRecord(transition state.StateTransition) bool {
|
||||
if !transition.Changed() {
|
||||
return false
|
||||
}
|
||||
@@ -35,9 +35,9 @@ func shouldRecord(transition state.StateTransition) bool {
|
||||
}
|
||||
|
||||
// ShouldRecordAnnotation returns true if an annotation should be created for a given state transition.
|
||||
// This is stricter than shouldRecord to avoid cluttering panels with state transitions.
|
||||
// This is stricter than ShouldRecord to avoid cluttering panels with state transitions.
|
||||
func ShouldRecordAnnotation(t state.StateTransition) bool {
|
||||
if !shouldRecord(t) {
|
||||
if !ShouldRecord(t) {
|
||||
return false
|
||||
}
|
||||
|
||||
|
||||
@@ -92,7 +92,7 @@ func TestShouldRecord(t *testing.T) {
|
||||
}
|
||||
|
||||
t.Run(fmt.Sprintf("%s -> %s should be %v", trans.PreviousFormatted(), trans.Formatted(), !ok), func(t *testing.T) {
|
||||
require.Equal(t, !ok, shouldRecord(trans))
|
||||
require.Equal(t, !ok, ShouldRecord(trans))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -41,6 +41,69 @@ const (
|
||||
dfLabels = "labels"
|
||||
)
|
||||
|
||||
// QueryResultBuilder is a builder for a data frame that represents query results from Loki.
|
||||
// It contains three fields: time (timestamp), line (JSON data), and labels (JSON labels).
|
||||
type QueryResultBuilder struct {
|
||||
frame *data.Frame
|
||||
}
|
||||
|
||||
// NewQueryResultBuilder creates a new QueryResultBuilder with the specified capacity.
|
||||
// The capacity is used to pre-allocate the underlying slices for better performance.
|
||||
func NewQueryResultBuilder(capacity int) *QueryResultBuilder {
|
||||
frame := data.NewFrame("states")
|
||||
lbls := data.Labels(map[string]string{})
|
||||
|
||||
// We represent state history as a single merged history, that roughly corresponds to what you get in the Grafana Explore tab when querying Loki directly.
|
||||
// The format is composed of the following vectors:
|
||||
// 1. `time` - timestamp - when the transition happened
|
||||
// 2. `line` - JSON - the full data of the transition
|
||||
// 3. `labels` - JSON - the labels associated with that state transition
|
||||
times := make([]time.Time, 0, capacity)
|
||||
lines := make([]json.RawMessage, 0, capacity)
|
||||
labels := make([]json.RawMessage, 0, capacity)
|
||||
|
||||
frame.Fields = append(frame.Fields, data.NewField(dfTime, lbls, times))
|
||||
frame.Fields = append(frame.Fields, data.NewField(dfLine, lbls, lines))
|
||||
frame.Fields = append(frame.Fields, data.NewField(dfLabels, lbls, labels))
|
||||
|
||||
return &QueryResultBuilder{frame: frame}
|
||||
}
|
||||
|
||||
func (qr QueryResultBuilder) AddRowRaw(timestamp time.Time, line json.RawMessage, labels json.RawMessage) {
|
||||
frame := qr.frame
|
||||
frame.Fields[0].Append(timestamp)
|
||||
frame.Fields[1].Append(line)
|
||||
frame.Fields[2].Append(labels)
|
||||
}
|
||||
|
||||
func (qr QueryResultBuilder) AddRow(timestamp time.Time, line LokiEntry, labels json.RawMessage) error {
|
||||
lineBytes, err := json.Marshal(line)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
qr.AddRowRaw(timestamp, lineBytes, labels)
|
||||
return nil
|
||||
}
|
||||
|
||||
// ToFrame converts the QueryResultBuilder back to a data.Frame.
|
||||
func (qr QueryResultBuilder) ToFrame() *data.Frame {
|
||||
return qr.frame
|
||||
}
|
||||
|
||||
func (qr QueryResultBuilder) AddWarn(s string) {
|
||||
m := qr.frame.Meta
|
||||
if m == nil {
|
||||
m = &data.FrameMeta{}
|
||||
qr.frame.SetMeta(m)
|
||||
}
|
||||
m.Notices = append(m.Notices, data.Notice{
|
||||
Severity: data.NoticeSeverityWarning,
|
||||
Text: s,
|
||||
Link: "",
|
||||
Inspect: 0,
|
||||
})
|
||||
}
|
||||
|
||||
const (
|
||||
StateHistoryLabelKey = "from"
|
||||
StateHistoryLabelValue = "state-history"
|
||||
@@ -191,20 +254,7 @@ func (h RemoteLokiBackend) merge(res []lokiclient.Stream, folderUIDToFilter []st
|
||||
totalLen += len(arr.Values)
|
||||
}
|
||||
|
||||
// Create a new slice to store the merged elements.
|
||||
frame := data.NewFrame("states")
|
||||
|
||||
// We merge all series into a single linear history.
|
||||
lbls := data.Labels(map[string]string{})
|
||||
|
||||
// We represent state history as a single merged history, that roughly corresponds to what you get in the Grafana Explore tab when querying Loki directly.
|
||||
// The format is composed of the following vectors:
|
||||
// 1. `time` - timestamp - when the transition happened
|
||||
// 2. `line` - JSON - the full data of the transition
|
||||
// 3. `labels` - JSON - the labels associated with that state transition
|
||||
times := make([]time.Time, 0, totalLen)
|
||||
lines := make([]json.RawMessage, 0, totalLen)
|
||||
labels := make([]json.RawMessage, 0, totalLen)
|
||||
queryResult := NewQueryResultBuilder(totalLen)
|
||||
|
||||
// Initialize a slice of pointers to the current position in each array.
|
||||
pointers := make([]int, len(res))
|
||||
@@ -259,17 +309,10 @@ func (h RemoteLokiBackend) merge(res []lokiclient.Stream, folderUIDToFilter []st
|
||||
pointers[minElStreamIdx]++
|
||||
continue
|
||||
}
|
||||
times = append(times, time.Unix(0, tsNano))
|
||||
labels = append(labels, lblsJson)
|
||||
lines = append(lines, json.RawMessage(entryBytes))
|
||||
queryResult.AddRowRaw(time.Unix(0, tsNano), entryBytes, lblsJson)
|
||||
pointers[minElStreamIdx]++
|
||||
}
|
||||
|
||||
frame.Fields = append(frame.Fields, data.NewField(dfTime, lbls, times))
|
||||
frame.Fields = append(frame.Fields, data.NewField(dfLine, lbls, lines))
|
||||
frame.Fields = append(frame.Fields, data.NewField(dfLabels, lbls, labels))
|
||||
|
||||
return frame, nil
|
||||
return queryResult.ToFrame(), nil
|
||||
}
|
||||
|
||||
func StatesToStream(rule history_model.RuleMeta, states []state.StateTransition, externalLabels map[string]string, logger log.Logger) lokiclient.Stream {
|
||||
@@ -282,28 +325,11 @@ func StatesToStream(rule history_model.RuleMeta, states []state.StateTransition,
|
||||
|
||||
samples := make([]lokiclient.Sample, 0, len(states))
|
||||
for _, state := range states {
|
||||
if !shouldRecord(state) {
|
||||
if !ShouldRecord(state) {
|
||||
continue
|
||||
}
|
||||
|
||||
sanitizedLabels := removePrivateLabels(state.Labels)
|
||||
entry := LokiEntry{
|
||||
SchemaVersion: 1,
|
||||
Previous: state.PreviousFormatted(),
|
||||
Current: state.Formatted(),
|
||||
Values: valuesAsDataBlob(state.State),
|
||||
Condition: rule.Condition,
|
||||
DashboardUID: rule.DashboardUID,
|
||||
PanelID: rule.PanelID,
|
||||
Fingerprint: labelFingerprint(sanitizedLabels),
|
||||
RuleTitle: rule.Title,
|
||||
RuleID: rule.ID,
|
||||
RuleUID: rule.UID,
|
||||
InstanceLabels: sanitizedLabels,
|
||||
}
|
||||
if state.State.State == eval.Error {
|
||||
entry.Error = state.Error.Error()
|
||||
}
|
||||
entry := StateTransitionToLokiEntry(rule, state)
|
||||
|
||||
jsn, err := json.Marshal(entry)
|
||||
if err != nil {
|
||||
@@ -324,6 +350,28 @@ func StatesToStream(rule history_model.RuleMeta, states []state.StateTransition,
|
||||
}
|
||||
}
|
||||
|
||||
func StateTransitionToLokiEntry(rule history_model.RuleMeta, state state.StateTransition) LokiEntry {
|
||||
sanitizedLabels := removePrivateLabels(state.Labels)
|
||||
entry := LokiEntry{
|
||||
SchemaVersion: 1,
|
||||
Previous: state.PreviousFormatted(),
|
||||
Current: state.Formatted(),
|
||||
Values: valuesAsDataBlob(state.State),
|
||||
Condition: rule.Condition,
|
||||
DashboardUID: rule.DashboardUID,
|
||||
PanelID: rule.PanelID,
|
||||
Fingerprint: labelFingerprint(sanitizedLabels),
|
||||
RuleTitle: rule.Title,
|
||||
RuleID: rule.ID,
|
||||
RuleUID: rule.UID,
|
||||
InstanceLabels: sanitizedLabels,
|
||||
}
|
||||
if state.State.State == eval.Error && state.Error != nil {
|
||||
entry.Error = state.Error.Error()
|
||||
}
|
||||
return entry
|
||||
}
|
||||
|
||||
func (h *RemoteLokiBackend) recordStreams(ctx context.Context, stream lokiclient.Stream, logger log.Logger) error {
|
||||
if err := h.client.Push(ctx, []lokiclient.Stream{stream}); err != nil {
|
||||
return err
|
||||
|
||||
Reference in New Issue
Block a user