Method running

kv/memberlist/memberlist_client.go:654–726

(ctx context.Context)

var errFailedToJoinCluster = errors.New("failed to join memberlist cluster on startup")

func (m *KV) running(ctx context.Context) error {
	// The key notifications goroutine must be started as the very first thing, otherwise watch key notifications
	// will be delayed. In particular, it must be started before the memberlist cluster full-join procedure (below)
	// because it may take a long time to complete.
	if m.cfg.NotifyInterval > 0 {
		// Start delayed key notifications.
		notifTicker := time.NewTicker(m.cfg.NotifyInterval)
		defer notifTicker.Stop()
		go m.monitorKeyNotifications(ctx, notifTicker.C)
	}

	ok := m.joinMembersOnStartup(ctx)
	if !ok && m.cfg.AbortIfJoinFails {
		return errFailedToJoinCluster
	}

	// Start propagation delay tracker after joining the cluster, so that the first
	// WatchKey callback has the full cluster state and correctly skips pre-existing beacons.
	if m.cfg.PropagationDelayTracker.Enabled {
		m.propagationDelayTracker = NewPropagationDelayTracker(
			m,
			m.cfg.PropagationDelayTracker,
			m.memberlist.LocalNode().Name,
			m.logger,
			m.registerer,
		)
		if err := m.propagationDelayTracker.StartAsync(ctx); err != nil {
			level.Warn(m.logger).Log("msg", "failed to start propagation delay tracker", "err", err)
			m.propagationDelayTracker = nil
		}
	}

	var tickerChan <-chan time.Time
	if m.cfg.RejoinInterval > 0 && len(m.cfg.GetRejoinSeedNodes()) > 0 {
		// Use a random initial delay between 0 and RejoinInterval to uniformly
		// distribute rejoins across time when multiple processes start simultaneously.
		initialDelay := 1 + time.Duration(math_rand.Int63n(int64(m.cfg.RejoinInterval)))
		var stop func()
		stop, tickerChan = timeutil.NewVariableTicker(initialDelay, m.cfg.RejoinInterval)
		defer stop()
	}

	var obsoleteEntriesTickerChan <-chan time.Time
	if m.cfg.ObsoleteEntriesTimeout > 0 {
		obsoleteEntriesTicker := time.NewTicker(m.cfg.ObsoleteEntriesTimeout)
		defer obsoleteEntriesTicker.Stop()

		obsoleteEntriesTickerChan = obsoleteEntriesTicker.C
	}

	logger := log.With(m.logger, "phase", "periodic_rejoin")
	for {
		select {
		case <-tickerChan:
			const numAttempts = 1 // don't retry if resolution fails, we will try again next time
			reached, err := m.joinMembersWithRetries(ctx, m.cfg.GetRejoinSeedNodes(), numAttempts, logger)
			if err == nil {
				level.Info(logger).Log("msg", "re-joined memberlist cluster", "reached_nodes", reached)