Skip to content

Make sure unregistered ingester joining the ring after WAL replay #6277

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 10 commits into from
Dec 5, 2024
Merged

Make sure unregistered ingester joining the ring after WAL replay #6277

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
04b470f
Make sure ingester is active when joining the ring
alexqyle Oct 17, 2024
2a780f3
Merge branch 'cortexproject:master' into ingester-join-ring-on-start
alexqyle Oct 29, 2024
eab6616
Merge branch 'master' into ingester-join-ring-on-start
alexqyle Nov 5, 2024
d87cab9
Merge commit 'e1c3d79a4f3ef2662a72d782a1290a017d14e380' into ingester…
alexqyle Nov 7, 2024
d6c9cf3
tokens from file would be ignored if instance was not in the ring whe…
alexqyle Dec 3, 2024
fb10504
Skip CAS if instance was not in the ring on start and delay heartbeat…
alexqyle Dec 4, 2024
913e056
Fix compactor tests
alexqyle Dec 4, 2024
1b19602
Fixed existing token not loaded issue and added unit test
alexqyle Dec 4, 2024
3a1145d
update changelog
alexqyle Dec 5, 2024
4278600
Merge branch 'master' into ingester-join-ring-on-start
alexqyle Dec 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions pkg/ingester/ingester.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -713,7 +713,7 @@ func New(cfg Config, limits *validation.Overrides, registerer prometheus.Registe
}, i.getOldestUnshippedBlockMetric)
}

i.lifecycler, err = ring.NewLifecycler(cfg.LifecyclerConfig, i, "ingester", RingKey, false, cfg.BlocksStorageConfig.TSDB.FlushBlocksOnShutdown, logger, prometheus.WrapRegistererWithPrefix("cortex_", registerer))
i.lifecycler, err = ring.NewLifecycler(cfg.LifecyclerConfig, i, "ingester", RingKey, true, cfg.BlocksStorageConfig.TSDB.FlushBlocksOnShutdown, logger, prometheus.WrapRegistererWithPrefix("cortex_", registerer))
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -799,6 +799,13 @@ func (i *Ingester) startingV2ForFlusher(ctx context.Context) error {
}

func (i *Ingester) starting(ctx context.Context) error {
if err := i.openExistingTSDB(ctx); err != nil {
// Try to rollback and close opened TSDBs before halting the ingester.
i.closeAllTSDB()

return errors.Wrap(err, "opening existing TSDBs")
}

// Important: we want to keep lifecycler running until we ask it to stop, so we need to give it independent context
if err := i.lifecycler.StartAsync(context.Background()); err != nil {
return errors.Wrap(err, "failed to start lifecycler")
Expand All @@ -807,13 +814,6 @@ func (i *Ingester) starting(ctx context.Context) error {
return errors.Wrap(err, "failed to start lifecycler")
}

if err := i.openExistingTSDB(ctx); err != nil {
// Try to rollback and close opened TSDBs before halting the ingester.
i.closeAllTSDB()

return errors.Wrap(err, "opening existing TSDBs")
}

i.lifecycler.Join()

// let's start the rest of subservices via manager
Expand Down
3 changes: 3 additions & 0 deletions pkg/ring/lifecycler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -453,6 +453,8 @@ func (i *Lifecycler) loop(ctx context.Context) error {
return errors.Wrapf(err, "failed to join the ring %s", i.RingName)
}

level.Info(i.logger).Log("msg", "finished init ring", "ring", i.RingName, "state", i.GetState())

// We do various period tasks
var autoJoinAfter <-chan time.Time
var observeChan <-chan time.Time
Expand Down Expand Up @@ -480,6 +482,7 @@ func (i *Lifecycler) loop(ctx context.Context) error {
select {
case <-i.autojoinChan:
autoJoinAfter = time.After(i.cfg.JoinAfter)
level.Info(i.logger).Log("msg", "will do auto-joining after timeout", "timeout", i.cfg.JoinAfter, "state", i.GetState())
case <-autoJoinAfter:
if joined {
continue
Expand Down
Loading