Prevent read on TSDB once closeAllTSDB function has been called

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## master / unreleased
 
 * [CHANGE] Querier / ruler: Change `-querier.max-fetched-chunks-per-query` configuration to limit to maximum number of chunks that can be fetched in a single query. The number of chunks fetched by ingesters AND long-term storare combined should not exceed the value configured on `-querier.max-fetched-chunks-per-query`. #4260
+* [BUGFIX] Ingester: Prevent any reads on TSDBs when the ingester is stopping. #4304
 
 ## 1.10.0-rc.0 / 2021-06-28
 

pkg/ingester/ingester_v2.go

@@ -992,6 +992,10 @@ func (i *Ingester) v2Query(ctx context.Context, req *client.QueryRequest) (*clie
 
 	i.metrics.queries.Inc()
 
+	if i.checkIfAllTSDBClosing() {
+		return &client.QueryResponse{}, nil
+	}
+
 	db := i.getTSDB(userID)
 	if db == nil {
 		return &client.QueryResponse{}, nil
@@ -1048,6 +1052,10 @@ func (i *Ingester) v2QueryExemplars(ctx context.Context, req *client.ExemplarQue
 
 	i.metrics.queries.Inc()
 
+	if i.checkIfAllTSDBClosing() {
+		return &client.ExemplarQueryResponse{}, nil
+	}
+
 	db := i.getTSDB(userID)
 	if db == nil {
 		return &client.ExemplarQueryResponse{}, nil
@@ -1093,6 +1101,10 @@ func (i *Ingester) v2LabelValues(ctx context.Context, req *client.LabelValuesReq
 		return nil, err
 	}
 
+	if i.checkIfAllTSDBClosing() {
+		return &client.LabelValuesResponse{}, nil
+	}
+
 	db := i.getTSDB(userID)
 	if db == nil {
 		return &client.LabelValuesResponse{}, nil
@@ -1125,6 +1137,10 @@ func (i *Ingester) v2LabelNames(ctx context.Context, req *client.LabelNamesReque
 		return nil, err
 	}
 
+	if i.checkIfAllTSDBClosing() {
+		return &client.LabelNamesResponse{}, nil
+	}
+
 	db := i.getTSDB(userID)
 	if db == nil {
 		return &client.LabelNamesResponse{}, nil
@@ -1157,6 +1173,10 @@ func (i *Ingester) v2MetricsForLabelMatchers(ctx context.Context, req *client.Me
 		return nil, err
 	}
 
+	if i.checkIfAllTSDBClosing() {
+		return &client.MetricsForLabelMatchersResponse{}, nil
+	}
+
 	db := i.getTSDB(userID)
 	if db == nil {
 		return &client.MetricsForLabelMatchersResponse{}, nil
@@ -1224,6 +1244,10 @@ func (i *Ingester) v2UserStats(ctx context.Context, req *client.UserStatsRequest
 		return nil, err
 	}
 
+	if i.checkIfAllTSDBClosing() {
+		return &client.UserStatsResponse{}, nil
+	}
+
 	db := i.getTSDB(userID)
 	if db == nil {
 		return &client.UserStatsResponse{}, nil
@@ -1236,6 +1260,10 @@ func (i *Ingester) v2AllUserStats(ctx context.Context, req *client.UserStatsRequ
 	i.userStatesMtx.RLock()
 	defer i.userStatesMtx.RUnlock()
 
+	if i.checkIfAllTSDBClosing() {
+		return &client.UsersStatsResponse{}, nil
+	}
+
 	users := i.TSDBState.dbs
 
 	response := &client.UsersStatsResponse{
@@ -1280,6 +1308,10 @@ func (i *Ingester) v2QueryStream(req *client.QueryRequest, stream client.Ingeste
 
 	i.metrics.queries.Inc()
 
+	if i.checkIfAllTSDBClosing() {
+		return nil
+	}
+
 	db := i.getTSDB(userID)
 	if db == nil {
 		return nil
@@ -1787,6 +1819,11 @@ func (i *Ingester) getMemorySeriesMetric() float64 {
 	defer i.userStatesMtx.RUnlock()
 
 	count := uint64(0)
+
+	if i.checkIfAllTSDBClosing() {
+		return 0
+	}
+
 	for _, db := range i.TSDBState.dbs {
 		count += db.Head().NumSeries()
 	}
@@ -2245,3 +2282,11 @@ func (i *Ingester) getInstanceLimits() *InstanceLimits {
 
 	return l
 }
+
+func (i *Ingester) checkIfAllTSDBClosing() bool {
+	if i.State() == services.Stopping {
+		level.Debug(i.logger).Log("msg", "TSDB is unavailable, as the Ingester is in the process of stopping and closing all TSDB")
+		return true
+	}
+	return false
+}