[dbnode] Gracefully handle reads including documents with stale index…

… state

In db nodes, state is kept for cheap lookups to determine if a series is
indexed for a specific block start. This state exists on the document in the
index and in the shard object. On concurrent writes to the same time series,
it's possible for these two sources to get out of sync. Basically, the
state associated with the document on the index is a different instance
than the one that gets placed in the shard's lookup map. This can
result in issues where we don't accurately track that a series in
indexed for a specific block start as only the entry in the lookup
map will get updated in the future.

To resolve this, we introduce reconciliation methods when querying these
objects. If it's noticed on read that we're querying index state that
never got inserted into the shard lookup map, then we query the shard
for the appropriate entry and use the results on that object.

src/dbnode/integration/index_block_orphaned_entry_test.go

@@ -0,0 +1,207 @@
+// +build integration
+//
+// Copyright (c) 2021  Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package integration
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/m3db/m3/src/dbnode/client"
+	"github.com/m3db/m3/src/dbnode/namespace"
+	"github.com/m3db/m3/src/dbnode/persist/fs"
+	"github.com/m3db/m3/src/dbnode/storage"
+	"github.com/m3db/m3/src/dbnode/storage/index/compaction"
+	xclock "github.com/m3db/m3/src/x/clock"
+	"github.com/m3db/m3/src/x/ident"
+	xsync "github.com/m3db/m3/src/x/sync"
+	xtime "github.com/m3db/m3/src/x/time"
+)
+
+const (
+	numTestSeries     = 5
+	concurrentWorkers = 25
+	writesPerWorker   = 5
+	blockSize         = 2 * time.Hour
+)
+
+func TestIndexBlockOrphanedEntry(t *testing.T) {
+	setup := generateTestSetup(t)
+	defer setup.Close()
+
+	// Start the server
+	log := setup.StorageOpts().InstrumentOptions().Logger()
+	require.NoError(t, setup.StartServer())
+
+	// Stop the server
+	defer func() {
+		assert.NoError(t, setup.StopServer())
+		log.Debug("server is now down")
+	}()
+
+	client := setup.M3DBClient()
+	session, err := client.DefaultSession()
+	require.NoError(t, err)
+
+	// Write concurrent metrics to generate multiple entries for the same series
+	ids := make([]ident.ID, 0, numTestSeries)
+	for i := 0; i < numTestSeries; i++ {
+		fooID := ident.StringID(fmt.Sprintf("foo.%v", i))
+		ids = append(ids, fooID)
+
+		writeConcurrentMetrics(t, setup, session, fooID)
+	}
+
+	// Write metrics for a different series to push current foreground segment
+	// to the background. After this, all documents for foo.X exist in background segments
+	barID := ident.StringID("bar")
+	writeConcurrentMetrics(t, setup, session, barID)
+
+	// Fast-forward to a block rotation
+	newBlock := xtime.Now().Truncate(blockSize).Add(blockSize)
+	newCurrentTime := newBlock.Add(30 * time.Minute) // Add extra to account for buffer past
+	setup.SetNowFn(newCurrentTime)
+
+	// Wait for flush
+	log.Info("waiting for block rotation to complete")
+	nsID := setup.Namespaces()[0].ID()
+	found := xclock.WaitUntil(func() bool {
+		filesets, err := fs.IndexFileSetsAt(setup.FilePathPrefix(), nsID, newBlock.Add(-blockSize))
+		require.NoError(t, err)
+		return len(filesets) == 1
+	}, 30*time.Second)
+	require.True(t, found)
+
+	// Do post-block rotation writes
+	for _, id := range ids {
+		writeMetric(t, session, nsID, id, newCurrentTime, 999.0)
+	}
+	writeMetric(t, session, nsID, barID, newCurrentTime, 999.0)
+
+	// Foreground segments should be in the background again which means updated index entry
+	// is now behind the orphaned entry so index reads should fail.
+	log.Info("waiting for metrics to be indexed")
+	var (
+		missing string
+		ok      bool
+	)
+	found = xclock.WaitUntil(func() bool {
+		for _, id := range ids {
+			ok, err = isIndexedCheckedWithTime(
+				t, session, nsID, id, genTags(id), newCurrentTime,
+			)
+			if !ok {
+				missing = id.String()
+				return false
+			}
+		}
+		return true
+	}, 30*time.Second)
+	assert.True(t, found, fmt.Sprintf("series %s never indexed\n", missing))
+	assert.NoError(t, err)
+}
+
+func writeConcurrentMetrics(
+	t *testing.T,
+	setup TestSetup,
+	session client.Session,
+	seriesID ident.ID,
+) {
+	var wg sync.WaitGroup
+	nowFn := setup.DB().Options().ClockOptions().NowFn()
+
+	workerPool := xsync.NewWorkerPool(concurrentWorkers)
+	workerPool.Init()
+
+	mdID := setup.Namespaces()[0].ID()
+	for i := 0; i < concurrentWorkers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			for j := 0; j < writesPerWorker; j++ {
+				j := j
+				wg.Add(1)
+				workerPool.Go(func() {
+					defer wg.Done()
+					writeMetric(t, session, mdID, seriesID, xtime.ToUnixNano(nowFn()), float64(j))
+				})
+			}
+		}()
+	}
+
+	wg.Wait()
+}
+
+func genTags(seriesID ident.ID) ident.TagsIterator {
+	return ident.NewTagsIterator(ident.NewTags(ident.StringTag("tagName", seriesID.String())))
+}
+
+func writeMetric(
+	t *testing.T,
+	session client.Session,
+	nsID ident.ID,
+	seriesID ident.ID,
+	timestamp xtime.UnixNano,
+	value float64,
+) {
+	err := session.WriteTagged(nsID, seriesID, genTags(seriesID),
+		timestamp, value, xtime.Second, nil)
+	require.NoError(t, err)
+}
+
+func generateTestSetup(t *testing.T) TestSetup {
+	md, err := namespace.NewMetadata(testNamespaces[0],
+		namespace.NewOptions().
+			SetRetentionOptions(DefaultIntegrationTestRetentionOpts).
+			SetIndexOptions(namespace.NewIndexOptions().SetEnabled(true)))
+	require.NoError(t, err)
+
+	testOpts := NewTestOptions(t).
+		SetNamespaces([]namespace.Metadata{md}).
+		SetWriteNewSeriesAsync(true)
+	testSetup, err := NewTestSetup(t, testOpts, nil,
+		func(s storage.Options) storage.Options {
+			s = s.SetCoreFn(func() int {
+				return rand.Intn(4) //nolint:gosec
+			})
+			compactionOpts := s.IndexOptions().ForegroundCompactionPlannerOptions()
+			compactionOpts.Levels = []compaction.Level{
+				{
+					MinSizeInclusive: 0,
+					MaxSizeExclusive: 1,
+				},
+			}
+			return s.SetIndexOptions(
+				s.IndexOptions().
+					SetForegroundCompactionPlannerOptions(compactionOpts))
+		})
+	require.NoError(t, err)
+
+	return testSetup
+}

src/dbnode/integration/index_helpers.go

@@ -291,13 +291,30 @@ func isIndexed(t *testing.T, s client.Session, ns ident.ID, id ident.ID, tags id
 	return result
 }
 
-func isIndexedChecked(t *testing.T, s client.Session, ns ident.ID, id ident.ID, tags ident.TagIterator) (bool, error) {
+func isIndexedChecked(
+	t *testing.T,
+	s client.Session,
+	ns ident.ID,
+	id ident.ID,
+	tags ident.TagIterator,
+) (bool, error) {
+	return isIndexedCheckedWithTime(t, s, ns, id, tags, xtime.Now())
+}
+
+func isIndexedCheckedWithTime(
+	t *testing.T,
+	s client.Session,
+	ns ident.ID,
+	id ident.ID,
+	tags ident.TagIterator,
+	queryTime xtime.UnixNano,
+) (bool, error) {
 	q := newQuery(t, tags)
 	iter, _, err := s.FetchTaggedIDs(ContextWithDefaultTimeout(), ns,
 		index.Query{Query: q},
 		index.QueryOptions{
-			StartInclusive: xtime.Now(),
-			EndExclusive:   xtime.Now(),
+			StartInclusive: queryTime,
+			EndExclusive:   queryTime.Add(time.Nanosecond),
 			SeriesLimit:    10,
 		})
 	if err != nil {

src/dbnode/storage/entry.go

@@ -141,13 +141,43 @@ func (entry *Entry) IndexedForBlockStart(indexBlockStart xtime.UnixNano) bool {
 	return isIndexed
 }
 
-// IndexedRange returns minimum and maximum blockStart values covered by index entry.
+// ReconciledIndexedForBlockStart returns true if the blockStart has been indexed.
+// Attempts to retrieve the most recent index entry from the shard if this entry
+// was never inserted there. If there is an error, returns the value for this entry.
+func (entry *Entry) ReconciledIndexedForBlockStart(indexBlockStart xtime.UnixNano) bool {
+	if entry.insertTime.Load() > 0 {
+		return entry.IndexedForBlockStart(indexBlockStart)
+	}
+	e, _, err := entry.Shard.TryRetrieveSeriesAndIncrementReaderWriterCount(entry.ID)
+	if err != nil || e == nil {
+		return entry.IndexedForBlockStart(indexBlockStart)
+	}
+	defer e.DecrementReaderWriterCount()
+
+	return e.IndexedForBlockStart(indexBlockStart)
+}
+
+// ReconciledIndexedRange returns minimum and maximum blockStart values covered by index entry.
 // The range is inclusive. Note that there may be uncovered gaps within the range.
+// Attempts to retrieve the most recent index entry from the shard if this entry
+// was never inserted there. If there is an error, returns the range on this entry.
 // Returns (0, 0) for an empty range.
-func (entry *Entry) IndexedRange() (xtime.UnixNano, xtime.UnixNano) {
-	entry.reverseIndex.RLock()
-	min, max := entry.reverseIndex.indexedRangeWithRLock()
-	entry.reverseIndex.RUnlock()
+func (entry *Entry) ReconciledIndexedRange() (xtime.UnixNano, xtime.UnixNano) {
+	var min, max xtime.UnixNano
+	if entry.insertTime.Load() > 0 {
+		min, max = entry.reverseIndex.indexedRangeWithRLock()
+	} else {
+		e, _, err := entry.Shard.TryRetrieveSeriesAndIncrementReaderWriterCount(entry.ID)
+		if err != nil || e == nil {
+			return entry.reverseIndex.indexedRangeWithRLock()
+		}
+		defer e.DecrementReaderWriterCount()
+
+		e.reverseIndex.RLock()
+		min, max = e.reverseIndex.indexedRangeWithRLock()
+		e.reverseIndex.RUnlock()
+	}
+
 	return min, max
 }
 

src/dbnode/storage/entry_blackbox_test.go

@@ -171,10 +171,22 @@ func TestEntryIndexedRange(t *testing.T) {
 	ctrl := xtest.NewController(t)
 	defer ctrl.Finish()
 
-	entry := NewEntry(NewEntryOptions{Series: newMockSeries(ctrl)})
+	opts := DefaultTestOptions()
+	ctx := opts.ContextPool().Get()
+	defer ctx.Close()
+
+	shard := testDatabaseShard(t, opts)
+	defer func() {
+		require.NoError(t, shard.Close())
+	}()
+
+	entry := NewEntry(NewEntryOptions{
+		Shard:  shard,
+		Series: newMockSeries(ctrl),
+	})
 
 	assertRange := func(expectedMin, expectedMax xtime.UnixNano) {
-		min, max := entry.IndexedRange()
+		min, max := entry.ReconciledIndexedRange()
 		assert.Equal(t, expectedMin, min)
 		assert.Equal(t, expectedMax, max)
 	}

src/dbnode/storage/index/block.go

@@ -547,7 +547,7 @@ func (b *block) queryWithSpan(
 				inBlock                bool
 				currentBlock           = opts.StartInclusive.Truncate(b.blockSize)
 				endExclusive           = opts.EndExclusive
-				minIndexed, maxIndexed = md.OnIndexSeries.IndexedRange()
+				minIndexed, maxIndexed = md.OnIndexSeries.ReconciledIndexedRange()
 			)
 
 			if maxIndexed == 0 {
@@ -566,7 +566,7 @@ func (b *block) queryWithSpan(
 			}
 
 			for !inBlock && currentBlock.Before(endExclusive) {
-				inBlock = md.OnIndexSeries.IndexedForBlockStart(currentBlock)
+				inBlock = md.OnIndexSeries.ReconciledIndexedForBlockStart(currentBlock)
 				currentBlock = currentBlock.Add(b.blockSize)
 			}
 

src/dbnode/storage/index/block_bench_test.go

@@ -134,3 +134,9 @@ func (m mockOnIndexSeries) NeedsIndexGarbageCollected() bool           { return
 func (m mockOnIndexSeries) IndexedRange() (xtime.UnixNano, xtime.UnixNano) {
 	return 0, 0
 }
+
+func (m mockOnIndexSeries) ReconciledIndexedRange() (xtime.UnixNano, xtime.UnixNano) {
+	return 0, 0
+}
+
+func (m mockOnIndexSeries) ReconciledIndexedForBlockStart(_ xtime.UnixNano) bool { return false }