| 355 | type ManagerFactory func(ctx context.Context, userID string, notifier *notifier.Manager, logger log.Logger, frontendPool *client.Pool, reg prometheus.Registerer) (RulesManager, error) |
| 356 | |
| 357 | func DefaultTenantManagerFactory(cfg Config, p Pusher, q storage.Queryable, engine promql.QueryEngine, overrides RulesLimits, evalMetrics *RuleEvalMetrics, reg prometheus.Registerer) ManagerFactory { |
| 358 | // Wrap errors returned by Queryable to our wrapper, so that we can distinguish between those errors |
| 359 | // and errors returned by PromQL engine. Errors from Queryable can be either caused by user (limits) or internal errors. |
| 360 | // Errors from PromQL are always "user" errors. |
| 361 | q = querier.NewErrorTranslateQueryableWithFn(q, WrapQueryableErrors) |
| 362 | return func(ctx context.Context, userID string, notifier *notifier.Manager, logger log.Logger, frontendPool *client.Pool, reg prometheus.Registerer) (RulesManager, error) { |
| 363 | qfeClient, err := resolveFrontendClient(cfg.FrontendAddress, frontendPool) |
| 364 | if err != nil { |
| 365 | return nil, err |
| 366 | } |
| 367 | |
| 368 | if qfeClient == nil && engine == nil { |
| 369 | return nil, fmt.Errorf("neither engine nor frontend client is configured for user %s", userID) |
| 370 | } |
| 371 | |
| 372 | queryFunc := buildQueryFunc(engine, qfeClient, q, overrides, userID, cfg, evalMetrics, logger) |
| 373 | // We let the Prometheus rules manager control the context so that there is a chance |
| 374 | // for graceful shutdown of rules that are still in execution even in case the cortex context is canceled. |
| 375 | prometheusContext := user.InjectOrgID(context.WithoutCancel(ctx), userID) |
| 376 | |
| 377 | return rules.NewManager(&rules.ManagerOptions{ |
| 378 | Appendable: NewPusherAppendable(p, userID, overrides, |
| 379 | evalMetrics.TotalWritesVec.WithLabelValues(userID), |
| 380 | evalMetrics.FailedWritesVec.WithLabelValues(userID)), |
| 381 | Queryable: q, |
| 382 | QueryFunc: queryFunc, |
| 383 | Context: prometheusContext, |
| 384 | ExternalURL: cfg.ExternalURL.URL, |
| 385 | NotifyFunc: SendAlerts(notifier, cfg.ExternalURL.URL.String()), |
| 386 | Logger: util_log.GoKitLogToSlog(log.With(logger, "user", userID)), |
| 387 | Registerer: reg, |
| 388 | OutageTolerance: cfg.OutageTolerance, |
| 389 | ForGracePeriod: cfg.ForGracePeriod, |
| 390 | ResendDelay: cfg.ResendDelay, |
| 391 | ConcurrentEvalsEnabled: cfg.ConcurrentEvalsEnabled, |
| 392 | MaxConcurrentEvals: cfg.MaxConcurrentEvals, |
| 393 | DefaultRuleQueryOffset: func() time.Duration { |
| 394 | return overrides.RulerQueryOffset(userID) |
| 395 | }, |
| 396 | RestoreNewRuleGroups: cfg.EnableSharding, |
| 397 | }), nil |
| 398 | } |
| 399 | } |
| 400 | |
| 401 | func resolveFrontendClient(addr string, pool *client.Pool) (*frontendClient, error) { |
| 402 | if addr == "" { |