openExistingTSDB walks the user tsdb dir, and opens a tsdb for each user. This may start a WAL replay, so we limit the number of concurrently opening TSDB.
(ctx context.Context)
| 3016 | // openExistingTSDB walks the user tsdb dir, and opens a tsdb for each user. This may start a WAL replay, so we limit the number of |
| 3017 | // concurrently opening TSDB. |
| 3018 | func (i *Ingester) openExistingTSDB(ctx context.Context) error { |
| 3019 | level.Info(logutil.WithContext(ctx, i.logger)).Log("msg", "opening existing TSDBs") |
| 3020 | |
| 3021 | queue := make(chan string) |
| 3022 | group, groupCtx := errgroup.WithContext(ctx) |
| 3023 | |
| 3024 | // Create a pool of workers which will open existing TSDBs. |
| 3025 | for n := 0; n < i.cfg.BlocksStorageConfig.TSDB.MaxTSDBOpeningConcurrencyOnStartup; n++ { |
| 3026 | group.Go(func() error { |
| 3027 | for userID := range queue { |
| 3028 | startTime := time.Now() |
| 3029 | |
| 3030 | db, err := i.createTSDB(userID) |
| 3031 | if err != nil { |
| 3032 | level.Error(logutil.WithContext(ctx, i.logger)).Log("msg", "unable to open TSDB", "err", err, "user", userID) |
| 3033 | return errors.Wrapf(err, "unable to open TSDB for user %s", userID) |
| 3034 | } |
| 3035 | |
| 3036 | // Add the database to the map of user databases |
| 3037 | i.stoppedMtx.Lock() |
| 3038 | i.TSDBState.dbs[userID] = db |
| 3039 | i.stoppedMtx.Unlock() |
| 3040 | i.metrics.memUsers.Inc() |
| 3041 | |
| 3042 | i.TSDBState.walReplayTime.Observe(time.Since(startTime).Seconds()) |
| 3043 | } |
| 3044 | |
| 3045 | return nil |
| 3046 | }) |
| 3047 | } |
| 3048 | |
| 3049 | // Spawn a goroutine to find all users with a TSDB on the filesystem. |
| 3050 | group.Go(func() error { |
| 3051 | // Close the queue once filesystem walking is done. |
| 3052 | defer close(queue) |
| 3053 | |
| 3054 | walkErr := filepath.Walk(i.cfg.BlocksStorageConfig.TSDB.Dir, func(path string, info os.FileInfo, err error) error { |
| 3055 | if err != nil { |
| 3056 | // If the root directory doesn't exist, we're OK (not needed to be created upfront). |
| 3057 | if os.IsNotExist(err) && path == i.cfg.BlocksStorageConfig.TSDB.Dir { |
| 3058 | return filepath.SkipDir |
| 3059 | } |
| 3060 | |
| 3061 | level.Error(logutil.WithContext(ctx, i.logger)).Log("msg", "an error occurred while iterating the filesystem storing TSDBs", "path", path, "err", err) |
| 3062 | return errors.Wrapf(err, "an error occurred while iterating the filesystem storing TSDBs at %s", path) |
| 3063 | } |
| 3064 | |
| 3065 | // Skip root dir and all other files |
| 3066 | if path == i.cfg.BlocksStorageConfig.TSDB.Dir || !info.IsDir() { |
| 3067 | return nil |
| 3068 | } |
| 3069 | |
| 3070 | // Top level directories are assumed to be user TSDBs |
| 3071 | userID := info.Name() |
| 3072 | f, err := os.Open(path) |
| 3073 | if err != nil { |
| 3074 | level.Error(logutil.WithContext(ctx, i.logger)).Log("msg", "unable to open TSDB dir", "err", err, "user", userID, "path", path) |
| 3075 | return errors.Wrapf(err, "unable to open TSDB dir %s for user %s", path, userID) |