rules/manager.go
func (g *Group) Eval(ctx context.Context, ts time.Time) {
	for i, rule := range g.rules {
		select {
		case <-g.done:
			return
		default:
		}

		func(i int, rule Rule) {
			sp, ctx := opentracing.StartSpanFromContext(ctx, "rule")
			sp.SetTag("name", rule.Name())
			defer func(t time.Time) {
				sp.Finish()

				since := time.Since(t)
				g.metrics.evalDuration.Observe(since.Seconds())
				rule.SetEvaluationDuration(since)
				rule.SetEvaluationTimestamp(t)
			}(time.Now())

			g.metrics.evalTotal.Inc()

			vector, err := rule.Eval(ctx, ts, g.opts.QueryFunc, g.opts.ExternalURL)
			if err != nil {
				// Canceled queries are intentional termination of queries. This normally
				// happens on shutdown and thus we skip logging of any errors here.
				if _, ok := err.(promql.ErrQueryCanceled); !ok {
					level.Warn(g.logger).Log("msg", "Evaluating rule failed", "rule", rule, "err", err)
				}
				g.metrics.evalFailures.Inc()
				return
			}

			if ar, ok := rule.(*AlertingRule); ok {
				ar.sendAlerts(ctx, ts, g.opts.ResendDelay, g.interval, g.opts.NotifyFunc)
			}
			var (
				numOutOfOrder = 0
				numDuplicates = 0
			)

			app, err := g.opts.Appendable.Appender()
			if err != nil {
				level.Warn(g.logger).Log("msg", "creating appender failed", "err", err)
				return
			}

			seriesReturned := make(map[string]labels.Labels, len(g.seriesInPreviousEval[i]))
			for _, s := range vector {
				if _, err := app.Add(s.Metric, s.T, s.V); err != nil {
					switch err {
					case storage.ErrOutOfOrderSample:
						numOutOfOrder++
						level.Debug(g.logger).Log("msg", "Rule evaluation result discarded", "err", err, "sample", s)
					case storage.ErrDuplicateSampleForTimestamp:
						numDuplicates++
						level.Debug(g.logger).Log("msg", "Rule evaluation result discarded", "err", err, "sample", s)
					default:
						level.Warn(g.logger).Log("msg", "Rule evaluation result discarded", "err", err, "sample", s)
					}
				} else {
					seriesReturned[s.Metric.String()] = s.Metric
				}
			}
			if numOutOfOrder > 0 {
				level.Warn(g.logger).Log("msg", "Error on ingesting out-of-order result from rule evaluation", "numDropped", numOutOfOrder)
			}
			if numDuplicates > 0 {
				level.Warn(g.logger).Log("msg", "Error on ingesting results from rule evaluation with different value but same timestamp", "numDropped", numDuplicates)
			}

			for metric, lset := range g.seriesInPreviousEval[i] {
				if _, ok := seriesReturned[metric]; !ok {
					// Series no longer exposed, mark it stale.
					_, err = app.Add(lset, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN))
					switch err {
					case nil:
					case storage.ErrOutOfOrderSample, storage.ErrDuplicateSampleForTimestamp:
						// Do not count these in logging, as this is expected if series
						// is exposed from a different rule.
					default:
						level.Warn(g.logger).Log("msg", "adding stale sample failed", "sample", metric, "err", err)
					}
				}
			}
			if err := app.Commit(); err != nil {
				level.Warn(g.logger).Log("msg", "rule sample appending failed", "err", err)
			} else {
				g.seriesInPreviousEval[i] = seriesReturned
			}
		}(i, rule)
	}
}
func (g *Group) Eval(ctx context.Context, ts time.Time) {
	for i, rule := range g.rules {
		select {
		case <-g.done:
			return
		default:
		}

		func(i int, rule Rule) {
			sp, ctx := opentracing.StartSpanFromContext(ctx, "rule")
			sp.SetTag("name", rule.Name())
			defer func(t time.Time) {
				sp.Finish()

				since := time.Since(t)
				g.metrics.evalDuration.Observe(since.Seconds())
				rule.SetEvaluationDuration(since)
				rule.SetEvaluationTimestamp(t)
			}(time.Now())

			g.metrics.evalTotal.Inc()

			vector, err := rule.Eval(ctx, ts, g.opts.QueryFunc, g.opts.ExternalURL)
			if err != nil {
				// Canceled queries are intentional termination of queries. This normally
				// happens on shutdown and thus we skip logging of any errors here.
				if _, ok := err.(promql.ErrQueryCanceled); !ok {
					level.Warn(g.logger).Log("msg", "Evaluating rule failed", "rule", rule, "err", err)
				}
				g.metrics.evalFailures.Inc()
				return
			}

			if ar, ok := rule.(*AlertingRule); ok {
				ar.sendAlerts(ctx, ts, g.opts.ResendDelay, g.interval, g.opts.NotifyFunc)
			}
			var (
				numOutOfOrder = 0
				numDuplicates = 0
			)

			app, err := g.opts.Appendable.Appender()
			if err != nil {
				level.Warn(g.logger).Log("msg", "creating appender failed", "err", err)
				return
			}

			seriesReturned := make(map[string]labels.Labels, len(g.seriesInPreviousEval[i]))
			for _, s := range vector {
				if _, err := app.Add(s.Metric, s.T, s.V); err != nil {
					switch err {
					case storage.ErrOutOfOrderSample:
						numOutOfOrder++
						level.Debug(g.logger).Log("msg", "Rule evaluation result discarded", "err", err, "sample", s)
					case storage.ErrDuplicateSampleForTimestamp:
						numDuplicates++
						level.Debug(g.logger).Log("msg", "Rule evaluation result discarded", "err", err, "sample", s)
					default:
						level.Warn(g.logger).Log("msg", "Rule evaluation result discarded", "err", err, "sample", s)
					}
				} else {
					seriesReturned[s.Metric.String()] = s.Metric
				}
			}
			if numOutOfOrder > 0 {
				level.Warn(g.logger).Log("msg", "Error on ingesting out-of-order result from rule evaluation", "numDropped", numOutOfOrder)
			}
			if numDuplicates > 0 {
				level.Warn(g.logger).Log("msg", "Error on ingesting results from rule evaluation with different value but same timestamp", "numDropped", numDuplicates)
			}

			for metric, lset := range g.seriesInPreviousEval[i] {
				if _, ok := seriesReturned[metric]; !ok {
					// Series no longer exposed, mark it stale.
					_, err = app.Add(lset, timestamp.FromTime(ts), math.Float64frombits(value.StaleNaN))
					switch err {
					case nil:
					case storage.ErrOutOfOrderSample, storage.ErrDuplicateSampleForTimestamp:
						// Do not count these in logging, as this is expected if series
						// is exposed from a different rule.
					default:
						level.Warn(g.logger).Log("msg", "adding stale sample failed", "sample", metric, "err", err)
					}
				}
			}
			if err := app.Commit(); err != nil {
				level.Warn(g.logger).Log("msg", "rule sample appending failed", "err", err)
			} else {
				g.seriesInPreviousEval[i] = seriesReturned
			}
		}(i, rule)
	}
}
func ( g * Group ) Eval ( ctx context . Context , ts time . Time ) { for i , rule := range g . rules { select { case <- g . done : return default : } func ( i int , rule Rule ) { sp , ctx := opentracing . StartSpanFromContext ( ctx , " " ) sp . SetTag ( " " , rule . Name ( ) ) defer func ( t time . Time ) { sp . Finish ( ) since := time . Since ( t ) g . metrics . evalDuration . Observe ( since . Seconds ( ) ) rule . SetEvaluationDuration ( since ) rule . SetEvaluationTimestamp ( t ) } ( time . Now ( ) ) g . metrics . evalTotal . Inc ( ) vector , err := rule . Eval ( ctx , ts , g . opts . QueryFunc , g . opts . ExternalURL ) if err != nil { // Canceled queries are intentional termination of queries. This normally // happens on shutdown and thus we skip logging of any errors here. if _ , ok := err . ( promql . ErrQueryCanceled ) ; ! ok { level . Warn ( g . logger ) . Log ( " " , " " , " " , rule , " " , err ) } g . metrics . evalFailures . Inc ( ) return } if ar , ok := rule . ( * AlertingRule ) ; ok { ar . sendAlerts ( ctx , ts , g . opts . ResendDelay , g . interval , g . opts . NotifyFunc ) } var ( numOutOfOrder = 0 numDuplicates = 0 ) app , err := g . opts . Appendable . Appender ( ) if err != nil { level . Warn ( g . logger ) . Log ( " " , " " , " " , err ) return } seriesReturned := make ( map [ string ] labels . Labels , len ( g . seriesInPreviousEval [ i ] ) ) for _ , s := range vector { if _ , err := app . Add ( s . Metric , s . T , s . V ) ; err != nil { switch err { case storage . ErrOutOfOrderSample : numOutOfOrder ++ level . Debug ( g . logger ) . Log ( " " , " " , " " , err , " " , s ) case storage . ErrDuplicateSampleForTimestamp : numDuplicates ++ level . Debug ( g . logger ) . Log ( " " , " " , " " , err , " " , s ) default : level . Warn ( g . logger ) . Log ( " " , " " , " " , err , " " , s ) } } else { seriesReturned [ s . Metric . String ( ) ] = s . Metric } } if numOutOfOrder > 0 { level . Warn ( g . logger ) . Log ( " " , " " , " " , numOutOfOrder ) } if numDuplicates > 0 { level . Warn ( g . logger ) . Log ( " " , " " , " " , numDuplicates ) } for metric , lset := range g . seriesInPreviousEval [ i ] { if _ , ok := seriesReturned [ metric ] ; ! ok { // Series no longer exposed, mark it stale. _ , err = app . Add ( lset , timestamp . FromTime ( ts ) , math . Float64frombits ( value . StaleNaN ) ) switch err { case nil : case storage . ErrOutOfOrderSample , storage . ErrDuplicateSampleForTimestamp : // Do not count these in logging, as this is expected if series // is exposed from a different rule. default : level . Warn ( g . logger ) . Log ( " " , " " , " " , metric , " " , err ) } } } if err := app . Commit ( ) ; err != nil { level . Warn ( g . logger ) . Log ( " " , " " , " " , err ) } else { g . seriesInPreviousEval [ i ] = seriesReturned } } ( i , rule ) } }
// Eval runs a single evaluation cycle in which all rules are evaluated sequentially.
Eval runs a single evaluation cycle in which all rules are evaluated sequentially .