| 2978 | } |
| 2979 | |
| 2980 | func (i *Ingester) closeAllTSDB() { |
| 2981 | i.stoppedMtx.Lock() |
| 2982 | |
| 2983 | wg := &sync.WaitGroup{} |
| 2984 | wg.Add(len(i.TSDBState.dbs)) |
| 2985 | |
| 2986 | // Concurrently close all users TSDB |
| 2987 | for userID, userDB := range i.TSDBState.dbs { |
| 2988 | |
| 2989 | go func(db *userTSDB) { |
| 2990 | defer wg.Done() |
| 2991 | |
| 2992 | if err := db.Close(); err != nil { |
| 2993 | level.Warn(i.logger).Log("msg", "unable to close TSDB", "err", err, "user", userID) |
| 2994 | return |
| 2995 | } |
| 2996 | |
| 2997 | // Now that the TSDB has been closed, we should remove it from the |
| 2998 | // set of open ones. This lock acquisition doesn't deadlock with the |
| 2999 | // outer one, because the outer one is released as soon as all go |
| 3000 | // routines are started. |
| 3001 | i.stoppedMtx.Lock() |
| 3002 | delete(i.TSDBState.dbs, userID) |
| 3003 | i.stoppedMtx.Unlock() |
| 3004 | |
| 3005 | i.metrics.memUsers.Dec() |
| 3006 | i.metrics.activeSeriesPerUser.DeleteLabelValues(userID) |
| 3007 | i.metrics.activeNHSeriesPerUser.DeleteLabelValues(userID) |
| 3008 | }(userDB) |
| 3009 | } |
| 3010 | |
| 3011 | // Wait until all Close() completed |
| 3012 | i.stoppedMtx.Unlock() |
| 3013 | wg.Wait() |
| 3014 | } |
| 3015 | |
| 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. |