Method waitStableTokens

ring/basic_lifecycler.go:373–401

(ctx context.Context, period time.Duration)

}

func (l *BasicLifecycler) waitStableTokens(ctx context.Context, period time.Duration) error {
	heartbeatTickerStop, heartbeatTickerChan := newDisableableTicker(l.cfg.HeartbeatPeriod)
	defer heartbeatTickerStop()

	// The first observation will occur after the specified period.
	level.Info(l.logger).Log("msg", "waiting stable tokens", "ring", l.ringName)
	observeChan := time.After(period)

	for {
		select {
		case <-observeChan:
			if !l.verifyTokens(ctx) {
				// The verification has failed
				level.Info(l.logger).Log("msg", "tokens verification failed, keep observing", "ring", l.ringName)
				observeChan = time.After(period)
				break
			}

			level.Info(l.logger).Log("msg", "tokens verification succeeded", "ring", l.ringName)
			return nil

		case <-heartbeatTickerChan:
			l.heartbeat(ctx)

		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

// Verifies that tokens that this instance has registered to the ring still belong to it.
// Gossiping ring may change the ownership of tokens in case of conflicts.