Limit for time series read from disk (#3094)

* Limit for time series read from disk

There is additional memory costs when time series need to be loaded from
disk (in addition to the actual bytes being read). This new limit allows
docLimits to be higher so series already in memory can easily be served
without hitting limits.

src/cmd/services/m3dbnode/config/config_test.go

@@ -730,6 +730,7 @@ func TestConfiguration(t *testing.T) {
       meta_event_reporting_enabled: false
   limits:
     maxRecentlyQueriedSeriesDiskBytesRead: null
+    maxRecentlyQueriedSeriesDiskRead: null
     maxRecentlyQueriedSeriesBlocks: null
     maxOutstandingWriteRequests: 0
     maxOutstandingReadRequests: 0

src/cmd/services/m3dbnode/config/limits.go

@@ -29,6 +29,12 @@ type LimitsConfiguration struct {
 	// max is surpassed encounter an error.
 	MaxRecentlyQueriedSeriesDiskBytesRead *MaxRecentQueryResourceLimitConfiguration `yaml:"maxRecentlyQueriedSeriesDiskBytesRead"`
 
+	// MaxRecentlyQueriedSeriesDiskRead sets the upper limit on time series read from disk within a given lookback
+	// period. Queries which are issued while this max is surpassed encounter an error.
+	// This is the number of time series, which is different from the number of bytes controlled by
+	// MaxRecentlyQueriedSeriesDiskBytesRead.
+	MaxRecentlyQueriedSeriesDiskRead *MaxRecentQueryResourceLimitConfiguration `yaml:"maxRecentlyQueriedSeriesDiskRead"`
+
 	// MaxRecentlyQueriedSeriesBlocks sets the upper limit on time series blocks
 	// count within a given lookback period. Queries which are issued while this
 	// max is surpassed encounter an error.

src/dbnode/persist/fs/retriever.go

@@ -50,6 +50,7 @@ import (
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/pool"
 
+	"github.com/uber-go/tally"
 	"go.uber.org/zap"
 )
 
@@ -88,11 +89,12 @@ const (
 type blockRetriever struct {
 	sync.RWMutex
 
-	opts           BlockRetrieverOptions
-	fsOpts         Options
-	logger         *zap.Logger
-	queryLimits    limits.QueryLimits
-	bytesReadLimit limits.LookbackLimit
+	opts            BlockRetrieverOptions
+	fsOpts          Options
+	logger          *zap.Logger
+	queryLimits     limits.QueryLimits
+	bytesReadLimit  limits.LookbackLimit
+	seriesReadCount tally.Counter
 
 	newSeekerMgrFn newSeekerMgrFn
 
@@ -121,18 +123,21 @@ func NewBlockRetriever(
 		return nil, err
 	}
 
+	scope := fsOpts.InstrumentOptions().MetricsScope().SubScope("retriever")
+
 	return &blockRetriever{
-		opts:           opts,
-		fsOpts:         fsOpts,
-		logger:         fsOpts.InstrumentOptions().Logger(),
-		queryLimits:    opts.QueryLimits(),
-		bytesReadLimit: opts.QueryLimits().BytesReadLimit(),
-		newSeekerMgrFn: NewSeekerManager,
-		reqPool:        opts.RetrieveRequestPool(),
-		bytesPool:      opts.BytesPool(),
-		idPool:         opts.IdentifierPool(),
-		status:         blockRetrieverNotOpen,
-		notifyFetch:    make(chan struct{}, 1),
+		opts:            opts,
+		fsOpts:          fsOpts,
+		logger:          fsOpts.InstrumentOptions().Logger(),
+		queryLimits:     opts.QueryLimits(),
+		bytesReadLimit:  opts.QueryLimits().BytesReadLimit(),
+		seriesReadCount: scope.Counter("series-read"),
+		newSeekerMgrFn:  NewSeekerManager,
+		reqPool:         opts.RetrieveRequestPool(),
+		bytesPool:       opts.BytesPool(),
+		idPool:          opts.IdentifierPool(),
+		status:          blockRetrieverNotOpen,
+		notifyFetch:     make(chan struct{}, 1),
 		// We just close this channel when the fetchLoops should shutdown, so no
 		// buffering is required
 		fetchLoopsShouldShutdownCh: make(chan struct{}),
@@ -564,6 +569,11 @@ func (r *blockRetriever) streamRequest(
 	startTime time.Time,
 	nsCtx namespace.Context,
 ) (bool, error) {
+	req.resultWg.Add(1)
+	r.seriesReadCount.Inc(1)
+	if err := r.queryLimits.DiskSeriesReadLimit().Inc(1, req.source); err != nil {
+		return false, err
+	}
 	req.shard = shard
 
 	// NB(r): If the ID is a ident.BytesID then we can just hold
@@ -579,8 +589,6 @@ func (r *blockRetriever) streamRequest(
 	req.start = startTime
 	req.blockSize = r.blockSize
 
-	req.resultWg.Add(1)
-
 	// Ensure to finalize at the end of request
 	ctx.RegisterFinalizer(req)
 

src/dbnode/persist/fs/retriever_test.go

@@ -32,25 +32,29 @@ import (
 	"testing"
 	"time"
 
+	"github.com/uber-go/tally"
+
 	"github.com/m3db/m3/src/cluster/shard"
 	"github.com/m3db/m3/src/dbnode/digest"
+	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/persist"
 	"github.com/m3db/m3/src/dbnode/retention"
 	"github.com/m3db/m3/src/dbnode/sharding"
 	"github.com/m3db/m3/src/dbnode/storage/block"
 	"github.com/m3db/m3/src/dbnode/storage/index/convert"
+	"github.com/m3db/m3/src/dbnode/storage/limits"
 	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/dbnode/x/xio"
 	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/context"
 	"github.com/m3db/m3/src/x/ident"
+	"github.com/m3db/m3/src/x/instrument"
 	"github.com/m3db/m3/src/x/pool"
 	xsync "github.com/m3db/m3/src/x/sync"
 	xtime "github.com/m3db/m3/src/x/time"
 
 	"github.com/fortytw2/leaktest"
 	"github.com/golang/mock/gomock"
-	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -799,6 +803,38 @@ func TestBlockRetrieverHandlesSeekByIndexEntryErrors(t *testing.T) {
 	testBlockRetrieverHandlesSeekErrors(t, ctrl, mockSeeker)
 }
 
+func TestLimitSeriesReadFromDisk(t *testing.T) {
+	scope := tally.NewTestScope("test", nil)
+	limitOpts := limits.NewOptions().
+		SetInstrumentOptions(instrument.NewOptions().SetMetricsScope(scope)).
+		SetBytesReadLimitOpts(limits.DefaultLookbackLimitOptions()).
+		SetDocsLimitOpts(limits.DefaultLookbackLimitOptions()).
+		SetDiskSeriesReadLimitOpts(limits.LookbackLimitOptions{
+			Limit:    1,
+			Lookback: time.Second * 1,
+		})
+	queryLimits, err := limits.NewQueryLimits(limitOpts)
+	require.NoError(t, err)
+	opts := NewBlockRetrieverOptions().
+		SetBlockLeaseManager(&block.NoopLeaseManager{}).
+		SetQueryLimits(queryLimits)
+	publicRetriever, err := NewBlockRetriever(opts, NewOptions().
+		SetInstrumentOptions(instrument.NewOptions().SetMetricsScope(scope)))
+	require.NoError(t, err)
+	req := &retrieveRequest{}
+	retriever := publicRetriever.(*blockRetriever)
+	_, _ = retriever.streamRequest(context.NewContext(), req, 0, ident.StringID("id"), time.Now(), namespace.Context{})
+	_, err = retriever.streamRequest(context.NewContext(), req, 0, ident.StringID("id"), time.Now(), namespace.Context{})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "query aborted due to limit")
+
+	snapshot := scope.Snapshot()
+	seriesRead := snapshot.Counters()["test.retriever.series-read+"]
+	require.Equal(t, int64(2), seriesRead.Value())
+	seriesLimit := snapshot.Counters()["test.query-limit.exceeded+limit=disk-series-read"]
+	require.Equal(t, int64(1), seriesLimit.Value())
+}
+
 var errSeekErr = errors.New("some-error")
 
 func testBlockRetrieverHandlesSeekErrors(t *testing.T, ctrl *gomock.Controller, mockSeeker ConcurrentDataFileSetSeeker) {

src/dbnode/server/server.go

@@ -450,6 +450,7 @@ func Run(runOpts RunOptions) {
 	// Setup query stats tracking.
 	docsLimit := limits.DefaultLookbackLimitOptions()
 	bytesReadLimit := limits.DefaultLookbackLimitOptions()
+	diskSeriesReadLimit := limits.DefaultLookbackLimitOptions()
 	if limitConfig := runOpts.Config.Limits.MaxRecentlyQueriedSeriesBlocks; limitConfig != nil {
 		docsLimit.Limit = limitConfig.Value
 		docsLimit.Lookback = limitConfig.Lookback
@@ -458,9 +459,14 @@ func Run(runOpts RunOptions) {
 		bytesReadLimit.Limit = limitConfig.Value
 		bytesReadLimit.Lookback = limitConfig.Lookback
 	}
+	if limitConfig := runOpts.Config.Limits.MaxRecentlyQueriedSeriesDiskRead; limitConfig != nil {
+		diskSeriesReadLimit.Limit = limitConfig.Value
+		diskSeriesReadLimit.Lookback = limitConfig.Lookback
+	}
 	limitOpts := limits.NewOptions().
 		SetDocsLimitOpts(docsLimit).
 		SetBytesReadLimitOpts(bytesReadLimit).
+		SetDiskSeriesReadLimitOpts(diskSeriesReadLimit).
 		SetInstrumentOptions(iOpts)
 	if builder := opts.SourceLoggerBuilder(); builder != nil {
 		limitOpts = limitOpts.SetSourceLoggerBuilder(builder)

src/dbnode/storage/limits/noop_query_limits.go

@@ -44,6 +44,10 @@ func (q *noOpQueryLimits) BytesReadLimit() LookbackLimit {
 	return &noOpLookbackLimit{}
 }
 
+func (q *noOpQueryLimits) DiskSeriesReadLimit() LookbackLimit {
+	return &noOpLookbackLimit{}
+}
+
 func (q *noOpQueryLimits) AnyExceeded() error {
 	return nil
 }

src/dbnode/storage/limits/options.go

@@ -28,10 +28,11 @@ import (
 )
 
 type limitOpts struct {
-	iOpts               instrument.Options
-	docsLimitOpts       LookbackLimitOptions
-	bytesReadLimitOpts  LookbackLimitOptions
-	sourceLoggerBuilder SourceLoggerBuilder
+	iOpts                   instrument.Options
+	docsLimitOpts           LookbackLimitOptions
+	bytesReadLimitOpts      LookbackLimitOptions
+	diskSeriesReadLimitOpts LookbackLimitOptions
+	sourceLoggerBuilder     SourceLoggerBuilder
 }
 
 // NewOptions creates limit options with default values.
@@ -94,6 +95,18 @@ func (o *limitOpts) BytesReadLimitOpts() LookbackLimitOptions {
 	return o.bytesReadLimitOpts
 }
 
+// SetDiskSeriesReadLimitOpts sets the disk ts read limit options.
+func (o *limitOpts) SetDiskSeriesReadLimitOpts(value LookbackLimitOptions) Options {
+	opts := *o
+	opts.diskSeriesReadLimitOpts = value
+	return &opts
+}
+
+// DiskSeriesReadLimitOpts returns the disk ts read limit options.
+func (o *limitOpts) DiskSeriesReadLimitOpts() LookbackLimitOptions {
+	return o.diskSeriesReadLimitOpts
+}
+
 // SetSourceLoggerBuilder sets the source logger.
 func (o *limitOpts) SetSourceLoggerBuilder(value SourceLoggerBuilder) Options {
 	opts := *o

src/dbnode/storage/limits/query_limits.go

@@ -36,8 +36,9 @@ const (
 )
 
 type queryLimits struct {
-	docsLimit      *lookbackLimit
-	bytesReadLimit *lookbackLimit
+	docsLimit           *lookbackLimit
+	bytesReadLimit      *lookbackLimit
+	seriesDiskReadLimit *lookbackLimit
 }
 
 type lookbackLimit struct {
@@ -72,31 +73,30 @@ func DefaultLookbackLimitOptions() LookbackLimitOptions {
 }
 
 // NewQueryLimits returns a new query limits manager.
-func NewQueryLimits(
-	options Options,
-	// docsLimitOpts LookbackLimitOptions,
-	// bytesReadLimitOpts LookbackLimitOptions,
-	// instrumentOpts instrument.Options,
-) (QueryLimits, error) {
+func NewQueryLimits(options Options) (QueryLimits, error) {
 	if err := options.Validate(); err != nil {
 		return nil, err
 	}
 
 	var (
-		iOpts               = options.InstrumentOptions()
-		docsLimitOpts       = options.DocsLimitOpts()
-		bytesReadLimitOpts  = options.BytesReadLimitOpts()
-		sourceLoggerBuilder = options.SourceLoggerBuilder()
+		iOpts                   = options.InstrumentOptions()
+		docsLimitOpts           = options.DocsLimitOpts()
+		bytesReadLimitOpts      = options.BytesReadLimitOpts()
+		diskSeriesReadLimitOpts = options.DiskSeriesReadLimitOpts()
+		sourceLoggerBuilder     = options.SourceLoggerBuilder()
 
 		docsLimit = newLookbackLimit(
 			iOpts, docsLimitOpts, "docs-matched", sourceLoggerBuilder)
 		bytesReadLimit = newLookbackLimit(
 			iOpts, bytesReadLimitOpts, "disk-bytes-read", sourceLoggerBuilder)
+		seriesDiskReadLimit = newLookbackLimit(
+			iOpts, diskSeriesReadLimitOpts, "disk-series-read", sourceLoggerBuilder)
 	)
 
 	return &queryLimits{
-		docsLimit:      docsLimit,
-		bytesReadLimit: bytesReadLimit,
+		docsLimit:           docsLimit,
+		bytesReadLimit:      bytesReadLimit,
+		seriesDiskReadLimit: seriesDiskReadLimit,
 	}, nil
 }
 
@@ -145,20 +145,29 @@ func (q *queryLimits) BytesReadLimit() LookbackLimit {
 	return q.bytesReadLimit
 }
 
+func (q *queryLimits) DiskSeriesReadLimit() LookbackLimit {
+	return q.seriesDiskReadLimit
+}
+
 func (q *queryLimits) Start() {
 	q.docsLimit.start()
+	q.seriesDiskReadLimit.start()
 	q.bytesReadLimit.start()
 }
 
 func (q *queryLimits) Stop() {
 	q.docsLimit.stop()
+	q.seriesDiskReadLimit.stop()
 	q.bytesReadLimit.stop()
 }
 
 func (q *queryLimits) AnyExceeded() error {
 	if err := q.docsLimit.exceeded(); err != nil {
 		return err
 	}
+	if err := q.seriesDiskReadLimit.exceeded(); err != nil {
+		return err
+	}
 	return q.bytesReadLimit.exceeded()
 }
 

src/dbnode/storage/limits/query_limits_test.go

@@ -37,11 +37,13 @@ import (
 func testQueryLimitOptions(
 	docOpts LookbackLimitOptions,
 	bytesOpts LookbackLimitOptions,
+	seriesOpts LookbackLimitOptions,
 	iOpts instrument.Options,
 ) Options {
 	return NewOptions().
 		SetDocsLimitOpts(docOpts).
 		SetBytesReadLimitOpts(bytesOpts).
+		SetDiskSeriesReadLimitOpts(seriesOpts).
 		SetInstrumentOptions(iOpts)
 }
 
@@ -54,7 +56,11 @@ func TestQueryLimits(t *testing.T) {
 		Limit:    1,
 		Lookback: time.Second,
 	}
-	opts := testQueryLimitOptions(docOpts, bytesOpts, instrument.NewOptions())
+	seriesOpts := LookbackLimitOptions{
+		Limit:    1,
+		Lookback: time.Second,
+	}
+	opts := testQueryLimitOptions(docOpts, bytesOpts, seriesOpts, instrument.NewOptions())
 	queryLimits, err := NewQueryLimits(opts)
 	require.NoError(t, err)
 	require.NotNil(t, queryLimits)
@@ -69,7 +75,7 @@ func TestQueryLimits(t *testing.T) {
 	require.True(t, xerrors.IsInvalidParams(err))
 	require.True(t, IsQueryLimitExceededError(err))
 
-	opts = testQueryLimitOptions(docOpts, bytesOpts, instrument.NewOptions())
+	opts = testQueryLimitOptions(docOpts, bytesOpts, seriesOpts, instrument.NewOptions())
 	queryLimits, err = NewQueryLimits(opts)
 	require.NoError(t, err)
 	require.NotNil(t, queryLimits)
@@ -84,6 +90,22 @@ func TestQueryLimits(t *testing.T) {
 	require.Error(t, err)
 	require.True(t, xerrors.IsInvalidParams(err))
 	require.True(t, IsQueryLimitExceededError(err))
+
+	opts = testQueryLimitOptions(docOpts, bytesOpts, seriesOpts, instrument.NewOptions())
+	queryLimits, err = NewQueryLimits(opts)
+	require.NoError(t, err)
+	require.NotNil(t, queryLimits)
+
+	// No error yet.
+	err = queryLimits.AnyExceeded()
+	require.NoError(t, err)
+
+	// Limit from bytes.
+	require.Error(t, queryLimits.DiskSeriesReadLimit().Inc(2, nil))
+	err = queryLimits.AnyExceeded()
+	require.Error(t, err)
+	require.True(t, xerrors.IsInvalidParams(err))
+	require.True(t, IsQueryLimitExceededError(err))
 }
 
 func TestLookbackLimit(t *testing.T) {
@@ -298,7 +320,7 @@ func TestSourceLogger(t *testing.T) {
 		}
 
 		builder = &testBuilder{records: []testLoggerRecord{}}
-		opts    = testQueryLimitOptions(noLimit, noLimit, iOpts).
+		opts    = testQueryLimitOptions(noLimit, noLimit, noLimit, iOpts).
 			SetSourceLoggerBuilder(builder)
 	)