| 427 | } |
| 428 | |
| 429 | func enableSharding(r *Ruler, ringStore kv.Client) error { |
| 430 | lifecyclerCfg, err := r.cfg.Ring.ToLifecyclerConfig(r.logger) |
| 431 | if err != nil { |
| 432 | return errors.Wrap(err, "failed to initialize ruler's lifecycler config") |
| 433 | } |
| 434 | |
| 435 | // Define lifecycler delegates in reverse order (last to be called defined first because they're |
| 436 | // chained via "next delegate"). |
| 437 | delegate := ring.BasicLifecyclerDelegate(r) |
| 438 | if !r.Config().Ring.KeepInstanceInTheRingOnShutdown { |
| 439 | delegate = ring.NewLeaveOnStoppingDelegate(delegate, r.logger) |
| 440 | } |
| 441 | delegate = ring.NewTokensPersistencyDelegate(r.cfg.Ring.TokensFilePath, ring.JOINING, delegate, r.logger) |
| 442 | delegate = ring.NewAutoForgetDelegate(r.cfg.Ring.HeartbeatTimeout*ringAutoForgetUnhealthyPeriods, delegate, r.logger) |
| 443 | |
| 444 | rulerRingName := "ruler" |
| 445 | r.lifecycler, err = ring.NewBasicLifecycler(lifecyclerCfg, rulerRingName, ringKey, ringStore, delegate, r.logger, prometheus.WrapRegistererWithPrefix("cortex_", r.registry)) |
| 446 | if err != nil { |
| 447 | return errors.Wrap(err, "failed to initialize ruler's lifecycler") |
| 448 | } |
| 449 | |
| 450 | r.ring, err = ring.NewWithStoreClientAndStrategy(r.cfg.Ring.ToRingConfig(), rulerRingName, ringKey, ringStore, ring.NewIgnoreUnhealthyInstancesReplicationStrategy(), prometheus.WrapRegistererWithPrefix("cortex_", r.registry), r.logger) |
| 451 | if err != nil { |
| 452 | return errors.Wrap(err, "failed to initialize ruler's ring") |
| 453 | } |
| 454 | |
| 455 | return nil |
| 456 | } |
| 457 | |
| 458 | func (r *Ruler) Logger() log.Logger { |
| 459 | return r.logger |