Method userIndexUpdateLoop

pkg/compactor/compactor.go:1212–1243

(ctx context.Context)

}

func (c *Compactor) userIndexUpdateLoop(ctx context.Context) {
	// Hardcode ID to check which compactor owns updating user index.
	userID := users.UserIndexCompressedFilename
	// Align with clean up interval.
	ticker := time.NewTicker(util.DurationWithJitter(c.storageCfg.UsersScanner.UpdateInterval, 0.1))
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			level.Error(c.logger).Log("msg", "context timeout, exit user index update loop", "err", ctx.Err())
			return
		case <-ticker.C:
			owned, err := c.ownUser(userID, true)
			if err != nil {
				level.Error(c.logger).Log("msg", "failed to check if compactor owns updating user index", "err", err)
				// Wait for next interval. Worst case, the user index scanner will fallback to list strategy.
				continue
			}
			if !owned {
				continue
			}
			start := time.Now()
			if err := c.userIndexUpdater.UpdateUserIndex(ctx); err != nil {
				level.Error(c.logger).Log("msg", "failed to update user index", "err", err)
				// Wait for next interval. Worst case, the user index scanner will fallback to list strategy.
				continue
			}
			level.Info(c.logger).Log("msg", "successfully updated user index", "duration_ms", time.Since(start).Milliseconds())
		}
	}
}

const compactorMetaPrefix = "compactor-meta-"