Method loop

pkg/ha/ha_tracker.go:404–481

Follows pattern used by ring for WatchKey.

(ctx context.Context)

// Follows pattern used by ring for WatchKey.
func (c *HATracker) loop(ctx context.Context) error {
	if !c.cfg.EnableHATracker {
		// don't do anything, but wait until asked to stop.
		<-ctx.Done()
		return nil
	}

	// Start cleanup loop. It will stop when context is done.
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func() {
		defer wg.Done()
		c.cleanupOldReplicasLoop(ctx)
	}()
	// Start periodic update of user replica group count.
	go func() {
		defer wg.Done()
		ticker := time.NewTicker(userReplicaGroupUpdateInterval)
		defer ticker.Stop()

		for {
			select {
			case <-ticker.C:
				c.updateUserReplicaGroupCount()
			case <-ctx.Done():
				return
			}
		}
	}()

	// The KVStore config we gave when creating c should have contained a prefix,
	// which would have given us a prefixed KVStore client. So, we can pass empty string here.
	c.client.WatchPrefix(ctx, "", func(key string, value any) bool {
		replica := value.(*ReplicaDesc)
		user, cluster, keyHasSeparator := strings.Cut(key, "/")

		// Valid key would look like cluster/replica, and a key without a / such as `ring` would be invalid.
		if !keyHasSeparator {
			return true
		}

		c.electedLock.Lock()
		defer c.electedLock.Unlock()

		if replica.DeletedAt > 0 {
			delete(c.elected, key)
			c.electedReplicaChanges.DeleteLabelValues(user, cluster)
			c.electedReplicaTimestamp.DeleteLabelValues(user, cluster)

			userClusters := c.replicaGroups[user]
			if userClusters != nil {
				delete(userClusters, cluster)
				if len(userClusters) == 0 {
					delete(c.replicaGroups, user)
				}
			}
			return true
		}