Don't update lastUpdateTime if TSDB isn't updated. (#3727)

Signed-off-by: Peter Štibraný <peter.stibrany@grafana.com>

Author: pstibrany

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## master / unreleased
 
+* [CHANGE] Ingester: don't update internal "last updated" timestamp of TSDB if tenant only sends invalid samples. This affects how "idle" time is computed. #3727
 
 ## 1.7.0 in progress
 

pkg/ingester/ingester_v2.go

@@ -767,7 +767,10 @@ func (i *Ingester) v2Push(ctx context.Context, req *client.WriteRequest) (*clien
 	}
 	i.TSDBState.appenderCommitDuration.Observe(time.Since(startCommit).Seconds())
 
-	db.setLastUpdate(time.Now())
+	// If only invalid samples are pushed, don't change "last update", as TSDB was not modified.
+	if succeededSamplesCount > 0 {
+		db.setLastUpdate(time.Now())
+	}
 
 	// Increment metrics only if the samples have been successfully committed.
 	// If the code didn't reach this point, it means that we returned an error

pkg/ingester/ingester_v2_test.go

@@ -1884,6 +1884,47 @@ func TestIngester_closeAndDeleteUserTSDBIfIdle_shouldNotCloseTSDBIfShippingIsInP
 	assert.Equal(t, tsdbNotActive, i.closeAndDeleteUserTSDBIfIdle(userID))
 }
 
+func TestIngester_idleCloseEmptyTSDB(t *testing.T) {
+	ctx := context.Background()
+	cfg := defaultIngesterTestConfig()
+	cfg.BlocksStorageConfig.TSDB.ShipInterval = 1 * time.Minute
+	cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval = 1 * time.Minute
+	cfg.BlocksStorageConfig.TSDB.CloseIdleTSDBTimeout = 0 // Will not run the loop, but will allow us to close any TSDB fast.
+
+	// Create ingester
+	i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, nil)
+	require.NoError(t, err)
+	defer cleanup()
+
+	require.NoError(t, services.StartAndAwaitRunning(ctx, i))
+	defer services.StopAndAwaitTerminated(ctx, i) //nolint:errcheck
+
+	// Wait until it's ACTIVE
+	test.Poll(t, 1*time.Second, ring.ACTIVE, func() interface{} {
+		return i.lifecycler.GetState()
+	})
+
+	db, err := i.getOrCreateTSDB(userID, true)
+	require.NoError(t, err)
+	require.NotNil(t, db)
+
+	// Run compaction and shipping.
+	i.compactBlocks(context.Background(), true)
+	i.shipBlocks(context.Background())
+
+	// Make sure we can close completely empty TSDB without problems.
+	require.Equal(t, tsdbIdleClosed, i.closeAndDeleteUserTSDBIfIdle(userID))
+
+	// Verify that it was closed.
+	db = i.getTSDB(userID)
+	require.Nil(t, db)
+
+	// And we can recreate it again, if needed.
+	db, err = i.getOrCreateTSDB(userID, true)
+	require.NoError(t, err)
+	require.NotNil(t, db)
+}
+
 type shipperMock struct {
 	mock.Mock
 }
@@ -1894,6 +1935,51 @@ func (m *shipperMock) Sync(ctx context.Context) (uploaded int, err error) {
 	return args.Int(0), args.Error(1)
 }
 
+func TestIngester_invalidSamplesDontChangeLastUpdateTime(t *testing.T) {
+	cfg := defaultIngesterTestConfig()
+	cfg.LifecyclerConfig.JoinAfter = 0
+
+	// Create ingester
+	i, cleanup, err := newIngesterMockWithTSDBStorage(cfg, nil)
+	require.NoError(t, err)
+	defer cleanup()
+
+	require.NoError(t, services.StartAndAwaitRunning(context.Background(), i))
+	defer services.StopAndAwaitTerminated(context.Background(), i) //nolint:errcheck
+
+	// Wait until it's ACTIVE
+	test.Poll(t, 1*time.Second, ring.ACTIVE, func() interface{} {
+		return i.lifecycler.GetState()
+	})
+
+	ctx := user.InjectOrgID(context.Background(), userID)
+	sampleTimestamp := int64(model.Now())
+
+	{
+		req, _, _ := mockWriteRequest(labels.Labels{{Name: labels.MetricName, Value: "test"}}, 0, sampleTimestamp)
+		_, err = i.v2Push(ctx, req)
+		require.NoError(t, err)
+	}
+
+	db := i.getTSDB(userID)
+	lastUpdate := db.lastUpdate.Load()
+
+	// Wait until 1 second passes.
+	test.Poll(t, 1*time.Second, time.Now().Unix()+1, func() interface{} {
+		return time.Now().Unix()
+	})
+
+	// Push another sample to the same metric and timestamp, with different value. We expect to get error.
+	{
+		req, _, _ := mockWriteRequest(labels.Labels{{Name: labels.MetricName, Value: "test"}}, 1, sampleTimestamp)
+		_, err = i.v2Push(ctx, req)
+		require.Error(t, err)
+	}
+
+	// Make sure last update hasn't changed.
+	require.Equal(t, lastUpdate, db.lastUpdate.Load())
+}
+
 func TestIngester_flushing(t *testing.T) {
 	for name, tc := range map[string]struct {
 		setupIngester func(cfg *Config)