Method getOrCreateTSDB

pkg/ingester/ingester.go:2776–2823

(userID string, force bool)

}

func (i *Ingester) getOrCreateTSDB(userID string, force bool) (*userTSDB, error) {
	db, err := i.getTSDB(userID)
	if db != nil {
		if err != nil {
			level.Warn(i.logger).Log("msg", "error getting user DB but userDB is not null", "err", err, "userID", userID)
		}
		return db, nil
	}

	i.stoppedMtx.Lock()
	defer i.stoppedMtx.Unlock()

	// Check again for DB in the event it was created in-between locks
	var ok bool
	db, ok = i.TSDBState.dbs[userID]
	if ok {
		return db, nil
	}

	// We're ready to create the TSDB, however we must be sure that the ingester
	// is in the ACTIVE state, otherwise it may conflict with the transfer in/out.
	// The TSDB is created when the first series is pushed and this shouldn't happen
	// to a non-ACTIVE ingester, however we want to protect from any bug, cause we
	// may have data loss or TSDB WAL corruption if the TSDB is created before/during
	// a transfer in occurs.
	if ingesterState := i.lifecycler.GetState(); !force && ingesterState != ring.ACTIVE {
		return nil, fmt.Errorf(errTSDBCreateIncompatibleState, ingesterState)
	}

	gl := i.getInstanceLimits()
	if gl != nil && gl.MaxInMemoryTenants > 0 {
		if users := int64(len(i.TSDBState.dbs)); users >= gl.MaxInMemoryTenants {
			return nil, errMaxUsersLimitReached
		}
	}

	// Create the database and a shipper for a user
	db, err = i.createTSDB(userID)
	if err != nil {
		return nil, err
	}

	// Add the db to list of user databases
	i.TSDBState.dbs[userID] = db
	i.metrics.memUsers.Inc()

	return db, nil
}

func (i *Ingester) blockChunkQuerierFunc(userId string) tsdb.BlockChunkQuerierFunc {
	return func(b tsdb.BlockReader, mint, maxt int64) (storage.ChunkQuerier, error) {