[dbnode] additional query limit on bytes read

go.mod

@@ -54,6 +54,7 @@ require (
 	github.com/m3db/bloom/v4 v4.0.0-20200901140942-52efb8544fe9
 	github.com/m3db/build-tools v0.0.0-20181013000606-edd1bdd1df8a
 	github.com/m3db/m3x v0.0.0-20190408051622-ebf3c7b94afd // indirect
+	github.com/m3db/prometheus v2.1.0+incompatible
 	github.com/m3db/prometheus_client_golang v0.8.1
 	github.com/m3db/prometheus_client_model v0.0.0-20180517145114-8b2299a4bf7d
 	github.com/m3db/prometheus_common v0.0.0-20180517030744-25aaa3dff79b

go.sum

@@ -497,6 +497,8 @@ github.com/m3db/build-tools v0.0.0-20181013000606-edd1bdd1df8a h1:CwsSHIJLeCESKd
 github.com/m3db/build-tools v0.0.0-20181013000606-edd1bdd1df8a/go.mod h1:Pk9AtZeKuCO2xcAth0gxwzRNFv4lV26GPSx4I6A7DQ8=
 github.com/m3db/m3x v0.0.0-20190408051622-ebf3c7b94afd h1:wzLBtXzxZM9b6IXwLSRE5crynocLTCuRDpGDaOJzyuI=
 github.com/m3db/m3x v0.0.0-20190408051622-ebf3c7b94afd/go.mod h1:zLbcVb352e3Jsg62A6zzEhZ1gumeFsiamTqDs9ZmZrs=
+github.com/m3db/prometheus v2.1.0+incompatible h1:Wie+/s10WAoOYOoU1PHoj4F9xaAFrI5hug4kqd8FDow=
+github.com/m3db/prometheus v2.1.0+incompatible/go.mod h1:y3O2uC8UcDL9w/sDA2m1sI6t/UKkvOGsqTw4YuSXYwk=
 github.com/m3db/prometheus_client_golang v0.8.1 h1:t7w/tcFws81JL1j5sqmpqcOyQOpH4RDOmIe3A3fdN3w=
 github.com/m3db/prometheus_client_golang v0.8.1/go.mod h1:8R/f1xYhXWq59KD/mbRqoBulXejss7vYtYzWmruNUwI=
 github.com/m3db/prometheus_client_model v0.0.0-20180517145114-8b2299a4bf7d h1:BtNPRz2kmh42OAVYjsOBjty2wE0FaIh7aPVtFNPcF48=

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

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

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

@@ -27,7 +27,12 @@ type LimitsConfiguration struct {
 	// 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.
-	MaxRecentlyQueriedSeriesBlocks *MaxRecentlyQueriedSeriesBlocksConfiguration `yaml:"maxRecentlyQueriedSeriesBlocks"`
+	MaxRecentlyQueriedSeriesBlocks *MaxRecentQueryResourceLimitConfiguration `yaml:"maxRecentlyQueriedSeriesBlocks"`
+
+	// MaxRecentlyQueriedSeriesBytesRead sets the upper limit on time series bytes
+	// read from disk within a given lookback period. Queries which are issued while this
+	// max is surpassed encounter an error.
+	MaxRecentlyQueriedSeriesBytesRead *MaxRecentQueryResourceLimitConfiguration `yaml:"maxRecentlyQueriedSeriesBytesRead"`
 
 	// MaxOutstandingWriteRequests controls the maximum number of outstanding write requests
 	// that the server will allow before it begins rejecting requests. Note that this value
@@ -55,14 +60,12 @@ type LimitsConfiguration struct {
 	MaxEncodersPerBlock int `yaml:"maxEncodersPerBlock" validate:"min=0"`
 }
 
-// MaxRecentlyQueriedSeriesBlocksConfiguration 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.
-type MaxRecentlyQueriedSeriesBlocksConfiguration struct {
-	// Value sets the max recently queried time series blocks for the given
-	// time window.
+// MaxRecentQueryResourceLimitConfiguration sets an upper limit on resources consumed by all queries
+// globally within a dbnode per some lookback period of time. Once exceeded, queries within that period
+// of time will be abandoned.
+type MaxRecentQueryResourceLimitConfiguration struct {
+	// Value sets the max value for the resource limit.
 	Value int64 `yaml:"value" validate:"min=0"`
-	// Lookback is the period to time window the max value of time series
-	// blocks allowed to be queried.
+	// Lookback is the period in which a given resource limit is enforced.
 	Lookback time.Duration `yaml:"lookback" validate:"min=0"`
 }

src/dbnode/persist/fs/retriever.go

@@ -288,6 +288,13 @@ func (r *blockRetriever) fetchBatch(
 	reqs []*retrieveRequest,
 	seekerResources ReusableSeekerResources,
 ) {
+	if err := r.opts.QueryLimits().AnyExceeded(); err != nil {
+		for _, req := range reqs {
+			req.onError(err)
+		}
+		return
+	}
+
 	// Resolve the seeker from the seeker mgr
 	seeker, err := seekerMgr.Borrow(shard, blockStart)
 	if err != nil {
@@ -299,13 +306,25 @@ func (r *blockRetriever) fetchBatch(
 
 	// Sort the requests by offset into the file before seeking
 	// to ensure all seeks are in ascending order
+	var limitErr error
 	for _, req := range reqs {
+		if limitErr != nil {
+			req.onError(err)
+			continue
+		}
+
 		entry, err := seeker.SeekIndexEntry(req.id, seekerResources)
 		if err != nil && err != errSeekIDNotFound {
 			req.onError(err)
 			continue
 		}
 
+		if err := r.opts.QueryLimits().BytesReadLimit().Inc(int(entry.Size)); err != nil {
+			req.onError(err)
+			limitErr = err
+			continue
+		}
+
 		if err == errSeekIDNotFound {
 			req.notFound = true
 		}

src/dbnode/persist/fs/retriever_options.go

@@ -25,6 +25,7 @@ import (
 	"runtime"
 
 	"github.com/m3db/m3/src/dbnode/storage/block"
+	"github.com/m3db/m3/src/dbnode/storage/limits"
 	"github.com/m3db/m3/src/dbnode/x/xio"
 	"github.com/m3db/m3/src/x/ident"
 	"github.com/m3db/m3/src/x/pool"
@@ -43,6 +44,7 @@ type blockRetrieverOptions struct {
 	fetchConcurrency  int
 	identifierPool    ident.Pool
 	blockLeaseManager block.LeaseManager
+	queryLimits       limits.QueryLimits
 }
 
 // NewBlockRetrieverOptions creates a new set of block retriever options
@@ -65,6 +67,7 @@ func NewBlockRetrieverOptions() BlockRetrieverOptions {
 		bytesPool:        bytesPool,
 		fetchConcurrency: defaultFetchConcurrency,
 		identifierPool:   ident.NewPool(bytesPool, ident.PoolOptions{}),
+		queryLimits:      limits.NoOpQueryLimits(),
 	}
 
 	return o
@@ -126,3 +129,13 @@ func (o *blockRetrieverOptions) SetBlockLeaseManager(leaseMgr block.LeaseManager
 func (o *blockRetrieverOptions) BlockLeaseManager() block.LeaseManager {
 	return o.blockLeaseManager
 }
+
+func (o *blockRetrieverOptions) SetQueryLimits(value limits.QueryLimits) BlockRetrieverOptions {
+	opts := *o
+	opts.queryLimits = value
+	return &opts
+}
+
+func (o *blockRetrieverOptions) QueryLimits() limits.QueryLimits {
+	return o.queryLimits
+}

src/dbnode/persist/fs/types.go

@@ -34,6 +34,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/sharding"
 	"github.com/m3db/m3/src/dbnode/storage/block"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
+	"github.com/m3db/m3/src/dbnode/storage/limits"
 	"github.com/m3db/m3/src/dbnode/x/xio"
 	"github.com/m3db/m3/src/m3ninx/doc"
 	"github.com/m3db/m3/src/m3ninx/index/segment/fst"
@@ -548,6 +549,12 @@ type BlockRetrieverOptions interface {
 
 	// BlockLeaseManager returns the block leaser.
 	BlockLeaseManager() block.LeaseManager
+
+	// SetQueryLimits sets query limits.
+	SetQueryLimits(value limits.QueryLimits) BlockRetrieverOptions
+
+	// QueryLimits returns the query limits.
+	QueryLimits() limits.QueryLimits
 }
 
 // ForEachRemainingFn is the function that is run on each of the remaining

src/dbnode/server/server.go

@@ -65,8 +65,8 @@ import (
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
 	"github.com/m3db/m3/src/dbnode/storage/cluster"
 	"github.com/m3db/m3/src/dbnode/storage/index"
+	"github.com/m3db/m3/src/dbnode/storage/limits"
 	"github.com/m3db/m3/src/dbnode/storage/series"
-	"github.com/m3db/m3/src/dbnode/storage/stats"
 	"github.com/m3db/m3/src/dbnode/topology"
 	"github.com/m3db/m3/src/dbnode/ts"
 	"github.com/m3db/m3/src/dbnode/ts/writes"
@@ -138,9 +138,6 @@ type RunOptions struct {
 	// interrupt and shutdown the server.
 	InterruptCh <-chan error
 
-	// QueryStatsTrackerFn returns a tracker for tracking query stats.
-	QueryStatsTrackerFn func(instrument.Options, stats.QueryStatsOptions) stats.QueryStatsTracker
-
 	// CustomOptions are custom options to apply to the session.
 	CustomOptions []client.CustomAdminOption
 
@@ -410,27 +407,22 @@ func Run(runOpts RunOptions) {
 	defer stopReporting()
 
 	// Setup query stats tracking.
-	statsOpts := stats.QueryStatsOptions{
-		Lookback: stats.DefaultLookback,
-	}
-	if max := runOpts.Config.Limits.MaxRecentlyQueriedSeriesBlocks; max != nil {
-		statsOpts = stats.QueryStatsOptions{
-			MaxDocs:  max.Value,
-			Lookback: max.Lookback,
-		}
+	docsLimit := limits.DefaultLookbackLimitOptions()
+	bytesReadLimit := limits.DefaultLookbackLimitOptions()
+	if limitConfig := runOpts.Config.Limits.MaxRecentlyQueriedSeriesBlocks; limitConfig != nil {
+		docsLimit.Limit = limitConfig.Value
+		docsLimit.Lookback = limitConfig.Lookback
 	}
-	if err := statsOpts.Validate(); err != nil {
-		logger.Fatal("could not construct query stats options from config", zap.Error(err))
+	if limitConfig := runOpts.Config.Limits.MaxRecentlyQueriedSeriesBytesRead; limitConfig != nil {
+		bytesReadLimit.Limit = limitConfig.Value
+		bytesReadLimit.Lookback = limitConfig.Lookback
 	}
-
-	tracker := stats.DefaultQueryStatsTracker(iopts, statsOpts)
-	if runOpts.QueryStatsTrackerFn != nil {
-		tracker = runOpts.QueryStatsTrackerFn(iopts, statsOpts)
+	queryLimits, err := limits.NewQueryLimits(iopts, docsLimit, bytesReadLimit)
+	if err != nil {
+		logger.Fatal("could not construct docs query limits from config", zap.Error(err))
 	}
-
-	queryStats := stats.NewQueryStats(tracker)
-	queryStats.Start()
-	defer queryStats.Stop()
+	queryLimits.Start()
+	defer queryLimits.Stop()
 
 	// FOLLOWUP(prateek): remove this once we have the runtime options<->index wiring done
 	indexOpts := opts.IndexOptions()
@@ -445,7 +437,7 @@ func Run(runOpts RunOptions) {
 			CacheTerms:  plCacheConfig.CacheTermsOrDefault(),
 		}).
 		SetMmapReporter(mmapReporter).
-		SetQueryStats(queryStats)
+		SetQueryLimits(queryLimits)
 	opts = opts.SetIndexOptions(indexOpts)
 
 	if tick := cfg.Tick; tick != nil {
@@ -559,7 +551,8 @@ func Run(runOpts RunOptions) {
 			SetBytesPool(opts.BytesPool()).
 			SetRetrieveRequestPool(opts.RetrieveRequestPool()).
 			SetIdentifierPool(opts.IdentifierPool()).
-			SetBlockLeaseManager(blockLeaseManager)
+			SetBlockLeaseManager(blockLeaseManager).
+			SetQueryLimits(queryLimits)
 		if blockRetrieveCfg := cfg.BlockRetrieve; blockRetrieveCfg != nil {
 			retrieverOpts = retrieverOpts.
 				SetFetchConcurrency(blockRetrieveCfg.FetchConcurrency)

src/dbnode/storage/database.go

@@ -826,6 +826,12 @@ func (d *db) QueryIDs(
 	}
 	defer sp.Finish()
 
+	// Check if exceeding query limits at very beginning of
+	// query path to abandon as early as possible.
+	if err := d.opts.IndexOptions().QueryLimits().AnyExceeded(); err != nil {
+		return index.QueryResult{}, err
+	}
+
 	n, err := d.namespaceFor(namespace)
 	if err != nil {
 		sp.LogFields(opentracinglog.Error(err))

src/dbnode/storage/index/block.go

@@ -30,7 +30,7 @@ import (
 
 	"github.com/m3db/m3/src/dbnode/namespace"
 	"github.com/m3db/m3/src/dbnode/storage/bootstrap/result"
-	"github.com/m3db/m3/src/dbnode/storage/stats"
+	"github.com/m3db/m3/src/dbnode/storage/limits"
 	"github.com/m3db/m3/src/dbnode/tracepoint"
 	"github.com/m3db/m3/src/m3ninx/doc"
 	m3ninxindex "github.com/m3db/m3/src/m3ninx/index"
@@ -141,7 +141,7 @@ type block struct {
 	blockOpts                       BlockOptions
 	nsMD                            namespace.Metadata
 	namespaceRuntimeOptsMgr         namespace.RuntimeOptionsManager
-	queryStats                      stats.QueryStats
+	queryLimits                     limits.QueryLimits
 
 	metrics blockMetrics
 	logger  *zap.Logger
@@ -249,7 +249,7 @@ func NewBlock(
 		namespaceRuntimeOptsMgr:         namespaceRuntimeOptsMgr,
 		metrics:                         newBlockMetrics(scope),
 		logger:                          iopts.Logger(),
-		queryStats:                      opts.QueryStats(),
+		queryLimits:                     opts.QueryLimits(),
 	}
 	b.newFieldsAndTermsIteratorFn = newFieldsAndTermsIterator
 	b.newExecutorWithRLockFn = b.executorWithRLock
@@ -533,7 +533,7 @@ func (b *block) addQueryResults(
 	batch []doc.Document,
 ) ([]doc.Document, int, int, error) {
 	// update recently queried docs to monitor memory.
-	if err := b.queryStats.Update(len(batch)); err != nil {
+	if err := b.queryLimits.DocsLimit().Inc(len(batch)); err != nil {
 		return batch, 0, 0, err
 	}
 
@@ -794,7 +794,7 @@ func (b *block) addAggregateResults(
 	batch []AggregateResultsEntry,
 ) ([]AggregateResultsEntry, int, int, error) {
 	// update recently queried docs to monitor memory.
-	if err := b.queryStats.Update(len(batch)); err != nil {
+	if err := b.queryLimits.DocsLimit().Inc(len(batch)); err != nil {
 		return batch, 0, 0, err
 	}