| 314 | } |
| 315 | |
| 316 | func (r *DefaultMultiTenantManager) getOrCreateNotifier(userID string, userManagerRegistry prometheus.Registerer) (*notifier.Manager, error) { |
| 317 | r.notifiersMtx.Lock() |
| 318 | defer r.notifiersMtx.Unlock() |
| 319 | |
| 320 | n, ok := r.notifiers[userID] |
| 321 | if ok { |
| 322 | // When there is a stale user, we stop the notifier but do not remove it |
| 323 | n.run() |
| 324 | return n.notifier, nil |
| 325 | } |
| 326 | |
| 327 | logger := log.With(r.logger, "user", userID) |
| 328 | |
| 329 | n = newRulerNotifier(¬ifier.Options{ |
| 330 | QueueCapacity: r.cfg.NotificationQueueCapacity, |
| 331 | Registerer: userManagerRegistry, |
| 332 | Do: func(ctx context.Context, client *http.Client, req *http.Request) (*http.Response, error) { |
| 333 | // Note: The passed-in context comes from the Prometheus notifier |
| 334 | // and does *not* contain the userID. So it needs to be added to the context |
| 335 | // here before using the context to inject the userID into the HTTP request. |
| 336 | ctx = user.InjectOrgID(ctx, userID) |
| 337 | if err := user.InjectOrgIDIntoHTTPRequest(ctx, req); err != nil { |
| 338 | return nil, err |
| 339 | } |
| 340 | // Jaeger complains the passed-in context has an invalid span ID, so start a new root span |
| 341 | sp := ot.GlobalTracer().StartSpan("notify", ot.Tag{Key: "organization", Value: userID}) |
| 342 | defer sp.Finish() |
| 343 | ctx = ot.ContextWithSpan(ctx, sp) |
| 344 | _ = ot.GlobalTracer().Inject(sp.Context(), ot.HTTPHeaders, ot.HTTPHeadersCarrier(req.Header)) |
| 345 | resp, err := ctxhttp.Do(ctx, client, req) |
| 346 | if err != nil { |
| 347 | level.Error(logger).Log("msg", "error occurred while sending alerts", "error", err) |
| 348 | return resp, err |
| 349 | } |
| 350 | defer resp.Body.Close() |
| 351 | if resp.StatusCode/100 != 2 { |
| 352 | bodyBytes, err := io.ReadAll(resp.Body) |
| 353 | if err != nil { |
| 354 | level.Error(logger).Log("msg", "error reading response body", "error", err, "response code", resp.StatusCode) |
| 355 | return resp, err |
| 356 | } |
| 357 | customErrorMessage := string(bodyBytes) |
| 358 | if len(customErrorMessage) >= 150 { |
| 359 | customErrorMessage = customErrorMessage[:150] |
| 360 | } |
| 361 | level.Error(logger).Log("msg", "error occurred sending notification", "error", customErrorMessage, "response code", resp.StatusCode) |
| 362 | } |
| 363 | return resp, err |
| 364 | }, |
| 365 | }, r.cfg.NameValidationScheme, logger, userManagerRegistry, r.notifiersDiscoveryMetrics) |
| 366 | |
| 367 | n.run() |
| 368 | |
| 369 | // This should never fail, unless there's a programming mistake. |
| 370 | if err := n.applyConfig(r.notifierCfg); err != nil { |
| 371 | return nil, err |
| 372 | } |
| 373 | |