robskillington · Nov 4, 2022 · Nov 4, 2022 · Nov 4, 2022 · Nov 4, 2022 · Nov 4, 2022
diff --git a/go.mod b/go.mod
@@ -27,6 +27,7 @@ require (
 	github.com/klauspost/compress v1.14.2
 	github.com/leanovate/gopter v0.2.8
 	github.com/lightstep/lightstep-tracer-go v0.18.1
+	github.com/lithammer/fuzzysearch v1.1.5
 	github.com/m3db/bitset v2.0.0+incompatible
 	github.com/m3db/bloom/v4 v4.0.0-20200901140942-52efb8544fe9
 	github.com/m3db/prometheus_client_golang v1.12.8

diff --git a/go.sum b/go.sum
@@ -1043,6 +1043,8 @@ github.com/lightstep/lightstep-tracer-go v0.18.1 h1:vi1F1IQ8N7hNWytK9DpJsUfQhGuN
 github.com/lightstep/lightstep-tracer-go v0.18.1/go.mod h1:jlF1pusYV4pidLvZ+XD0UBX0ZE6WURAspgAczcDHrL4=
 github.com/linode/linodego v1.1.0 h1:ZiFVUptlzuExtUbHZtXiN7I0dAOFQAyirBKb/6/n9n4=
 github.com/linode/linodego v1.1.0/go.mod h1:x/7+BoaKd4unViBmS2umdjYyVAmpFtBtEXZ0wou7FYQ=
+github.com/lithammer/fuzzysearch v1.1.5 h1:Ag7aKU08wp0R9QCfF4GoGST9HbmAIeLP7xwMrOBEp1c=
+github.com/lithammer/fuzzysearch v1.1.5/go.mod h1:1R1LRNk7yKid1BaQkmuLQaHruxcC4HmAH30Dh61Ih1Q=
 github.com/lyft/protoc-gen-star v0.5.1/go.mod h1:9toiA3cC7z5uVbODF7kEQ91Xn7XNFkVUl+SrEe+ZORU=
 github.com/lyft/protoc-gen-star v0.5.3/go.mod h1:V0xaHgaf5oCCqmcxYcWiDfTiKsZsRc87/1qhoTACD8w=
 github.com/m3db/bitset v2.0.0+incompatible h1:wMgri1Z2QSwJ8K/7ZuV7vE4feLOT7EofVC8RakIOybI=

diff --git a/src/cmd/services/m3query/config/config.go b/src/cmd/services/m3query/config/config.go
@@ -448,6 +448,15 @@ type PerQueryLimitsConfiguration struct {
 	// service.
 	MaxFetchedRange time.Duration `yaml:"maxFetchedRange"`
 
+	// ReturnedSeriesLimit limits the number of time series returned.
+	ReturnedSeriesLimit int `yaml:"returnedSeriesLimit"`
+
+	// ReturnedDatapointsLimit limits the number of datapoints returned.
+	ReturnedDatapointsLimit int `yaml:"returnedDatapointsLimit"`
+
+	// ReturnedSeriesMetadataLimit limits the number of absolute metrics metadata returned.
+	ReturnedSeriesMetadataLimit int `yaml:"returnedSeriesMetadataLimit"`
+
 	// RequireExhaustive results in an error if the query exceeds any limit.
 	RequireExhaustive *bool `yaml:"requireExhaustive"`
 
@@ -481,12 +490,15 @@ func (l *PerQueryLimitsConfiguration) AsFetchOptionsBuilderLimitsOptions() handl
 	}
 
 	return handleroptions.FetchOptionsBuilderLimitsOptions{
-		SeriesLimit:            seriesLimit,
-		InstanceMultiple:       l.InstanceMultiple,
-		DocsLimit:              docsLimit,
-		RangeLimit:             l.MaxFetchedRange,
-		RequireExhaustive:      requireExhaustive,
-		MaxMetricMetadataStats: maxMetricMetadataStats,
+		SeriesLimit:                 seriesLimit,
+		InstanceMultiple:            l.InstanceMultiple,
+		DocsLimit:                   docsLimit,
+		RangeLimit:                  l.MaxFetchedRange,
+		RequireExhaustive:           requireExhaustive,
+		MaxMetricMetadataStats:      maxMetricMetadataStats,
+		ReturnedSeriesLimit:         l.ReturnedSeriesLimit,
+		ReturnedDatapointsLimit:     l.ReturnedDatapointsLimit,
+		ReturnedSeriesMetadataLimit: l.ReturnedSeriesMetadataLimit,
 	}
 }
 

diff --git a/src/dbnode/integration/graphite_find_test.go b/src/dbnode/integration/graphite_find_test.go
@@ -33,6 +33,7 @@ import (
 	"reflect"
 	"runtime"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"testing"
@@ -49,6 +50,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/retention"
 	graphitehandler "github.com/m3db/m3/src/query/api/v1/handler/graphite"
 	"github.com/m3db/m3/src/query/graphite/graphite"
+	"github.com/m3db/m3/src/x/headers"
 	"github.com/m3db/m3/src/x/ident"
 	xhttp "github.com/m3db/m3/src/x/net/http"
 	xsync "github.com/m3db/m3/src/x/sync"
@@ -59,12 +61,14 @@ type testGraphiteFindDatasetSize uint
 
 const (
 	smallDatasetSize testGraphiteFindDatasetSize = iota
+	mediumDatasetSize
 	largeDatasetSize
 )
 
 type testGraphiteFindOptions struct {
 	checkConcurrency int
 	datasetSize      testGraphiteFindDatasetSize
+	checkLimit       bool
 }
 
 func TestGraphiteFindSequential(t *testing.T) {
@@ -73,7 +77,7 @@ func TestGraphiteFindSequential(t *testing.T) {
 	// forever) encounters errors running on CI.
 	testGraphiteFind(t, testGraphiteFindOptions{
 		checkConcurrency: 1,
-		datasetSize:      smallDatasetSize,
+		datasetSize:      mediumDatasetSize,
 	})
 }
 
@@ -86,6 +90,14 @@ func TestGraphiteFindParallel(t *testing.T) {
 	})
 }
 
+func TestGraphiteFindLimits(t *testing.T) {
+	testGraphiteFind(t, testGraphiteFindOptions{
+		checkConcurrency: 1,
+		datasetSize:      smallDatasetSize,
+		checkLimit:       true,
+	})
+}
+
 func testGraphiteFind(tt *testing.T, testOpts testGraphiteFindOptions) {
 	if testing.Short() {
 		tt.SkipNow() // Just skip if we're doing a short run
@@ -95,7 +107,7 @@ func testGraphiteFind(tt *testing.T, testOpts testGraphiteFindOptions) {
 	// by using a TestingT that panics when FailNow is called.
 	t := xtest.FailNowPanicsTestingT(tt)
 
-	const queryConfigYAML = `
+	queryConfigYAML := `
 listenAddress: 127.0.0.1:7201
 
 logging:
@@ -121,6 +133,19 @@ local:
       resolution: 1m
 `
 
+	if testOpts.checkLimit {
+		queryConfigYAML += `
+carbon:
+  limitsFind:
+    perQuery:
+      maxFetchedSeries: 10
+      instanceMultiple: 2
+      maxFetchedRange: 2h
+      requireExhaustive: false
+      returnedSeriesMetadataLimit: 8
+`
+	}
+
 	var (
 		blockSize       = 2 * time.Hour
 		retentionPeriod = 6 * blockSize
@@ -170,6 +195,10 @@ local:
 	)
 	switch testOpts.datasetSize {
 	case smallDatasetSize:
+		levels = 2
+		entriesPerLevelMin = 12
+		entriesPerLevelMax = 15
+	case mediumDatasetSize:
 		levels = 4
 		entriesPerLevelMin = 5
 		entriesPerLevelMax = 7
@@ -362,6 +391,11 @@ local:
 			http.MethodGet, url, nil)
 		require.NoError(t, err)
 
+		// Ensure that when the limit test runs we don't apply limit
+		// for this specific request (due to this being verification check).
+		req.Header.Set(headers.LimitMaxSeriesHeader, strconv.Itoa(1000))
+		req.Header.Set(headers.LimitMaxReturnedSeriesMetadataHeader, strconv.Itoa(1000))
+
 		res, err := httpClient.Do(req)
 		if err != nil {
 			return r, nil, err
@@ -417,6 +451,8 @@ local:
 		zap.Uint64("numSeriesChecking", numSeriesChecking))
 	parallelVerifyFindQueries(rootNode, 0)
 
+	testGraphiteFindLimit(t, setup, log)
+
 	// Wait for execution.
 	wg.Wait()
 
@@ -427,6 +463,40 @@ local:
 	}
 }
 
+func testGraphiteFindLimit(
+	t require.TestingT,
+	setup TestSetup,
+	log *zap.Logger,
+) {
+	params := make(url.Values)
+	params.Set("query", "lvl00_entry00_dir.*")
+
+	url := fmt.Sprintf("http://%s%s?%s", setup.QueryAddress(),
+		graphitehandler.FindURL, params.Encode())
+
+	req, err := http.NewRequestWithContext(context.Background(),
+		http.MethodGet, url, nil)
+	require.NoError(t, err)
+
+	res, err := http.DefaultClient.Do(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, res.StatusCode)
+
+	log.Info("find with limit applied response headers", zap.Any("headers", res.Header))
+
+	defer res.Body.Close()
+
+	var results graphiteFindResults
+	require.NoError(t, json.NewDecoder(res.Body).Decode(&results))
+
+	// Make sure was limited at server side.
+	assert.Equal(t, headers.LimitHeaderSeriesLimitApplied, res.Header.Get(headers.LimitHeader))
+
+	// Also make sure applied the results limit client side in query handler.
+	assert.NotEmpty(t, res.Header.Get(headers.ReturnedMetadataLimitedHeader))
+	assert.Equal(t, 8, len(results))
+}
+
 type graphiteFindResults []graphiteFindResult
 
 type graphiteFindResult struct {

diff --git a/src/query/api/v1/handler/graphite/find.go b/src/query/api/v1/handler/graphite/find.go
@@ -22,8 +22,11 @@ package graphite
 
 import (
 	"errors"
+	"io/ioutil"
+	"math"
 	"net/http"
 	"sort"
+	"strings"
 	"sync"
 
 	"github.com/m3db/m3/src/query/api/v1/handler/prometheus/handleroptions"
@@ -37,6 +40,7 @@ import (
 	"github.com/m3db/m3/src/x/instrument"
 	xhttp "github.com/m3db/m3/src/x/net/http"
 
+	"github.com/lithammer/fuzzysearch/fuzzy"
 	"go.uber.org/zap"
 )
 
@@ -116,16 +120,37 @@ func mergeTags(
 	return tagMap, nil
 }
 
-func findResultsSorted(prefix string, tagMap map[string]nodeDescriptor) []findResult {
+func findResultsSorted(search, prefix string, tagMap map[string]nodeDescriptor) []findResult {
 	results := make([]findResult, 0, len(tagMap))
+	words := make([]string, 0, len(tagMap))
 	for name, node := range tagMap {
 		results = append(results, findResult{
 			id:   prefix + name,
 			name: name,
 			node: node,
 		})
+		words = append(words, name)
 	}
+
+	var ranks fuzzy.Ranks
+	match := strings.Trim(search, "* ")
+	if match != "" {
+		ranks = fuzzy.RankFindFold(strings.ToLower(search), words)
+
+		// Set any unmatches to the bottom by making the distance max int.
+		for i := range ranks {
+			if ranks[i].Distance == -1 {
+				ranks[i].Distance = math.MaxUint32
+			}
+		}
+	}
+
 	sort.Slice(results, func(i, j int) bool {
+		if len(ranks) > 0 && ranks[i].Distance != ranks[j].Distance {
+			// If the ranks are different, sort by rank.
+			return ranks[i].Distance < ranks[j].Distance
+		}
+		// Otherwise sort by name then attributes.
 		if results[i].id != results[j].id {
 			return results[i].id < results[j].id
 		}
@@ -212,20 +237,38 @@ func (h *grahiteFindHandler) ServeHTTP(
 		prefix += "."
 	}
 
-	results := findResultsSorted(prefix, seenMap)
+	search := graphite.ExtractNthMetricPart(raw, graphite.CountMetricParts(raw)-1)
+	results := findResultsSorted(search, prefix, seenMap)
+	if err := handleroptions.AddDBResultResponseHeaders(w, meta, opts); err != nil {
+		logger.Error("unable to render find header", zap.Error(err))
+		xhttp.WriteError(w, err)
+		return
+	}
 
-	err = handleroptions.AddDBResultResponseHeaders(w, meta, opts)
+	// First write out results to zero output to check if will limit
+	// results and if so then write the header about truncation if occurred.
+	var (
+		noopWriter = ioutil.Discard
+		renderOpts = findResultsOptions{
+			includeBothExpandableAndLeaf: h.graphiteStorageOpts.FindResultsIncludeBothExpandableAndLeaf,
+			returnedSeriesMetadataLimit:  opts.ReturnedSeriesMetadataLimit,
+		}
+	)
+	limited, err := findResultsJSON(noopWriter, results, renderOpts)
 	if err != nil {
-		logger.Error("unable to render find header", zap.Error(err))
+		logger.Error("unable to render list tags results", zap.Error(err))
 		xhttp.WriteError(w, err)
 		return
 	}
 
-	// TODO: Support multiple result types
-	resultOpts := findResultsOptions{
-		includeBothExpandableAndLeaf: h.graphiteStorageOpts.FindResultsIncludeBothExpandableAndLeaf,
+	if err := handleroptions.AddReturnedLimitResponseHeaders(w, nil, limited); err != nil {
+		logger.Error("unable to write returned limit response headers", zap.Error(err))
+		xhttp.WriteError(w, err)
+		return
 	}
-	if err := findResultsJSON(w, results, resultOpts); err != nil {
+
+	_, err = findResultsJSON(w, results, renderOpts)
+	if err != nil {
 		logger.Error("unable to render find results", zap.Error(err))
 	}
 }
diff --git a/src/query/api/v1/handler/graphite/find_parser.go b/src/query/api/v1/handler/graphite/find_parser.go
@@ -27,6 +27,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/m3db/m3/src/query/api/v1/handler/prometheus/handleroptions"
 	"github.com/m3db/m3/src/query/errors"
 	"github.com/m3db/m3/src/query/graphite/graphite"
 	graphitestorage "github.com/m3db/m3/src/query/graphite/storage"
@@ -174,22 +175,24 @@ func parseFindParamsToQueries(r *http.Request) (
 
 type findResultsOptions struct {
 	includeBothExpandableAndLeaf bool
+	returnedSeriesMetadataLimit  int
 }
 
 func findResultsJSON(
 	w io.Writer,
 	results []findResult,
 	opts findResultsOptions,
-) error {
+) (*handleroptions.ReturnedMetadataLimited, error) {
+	meta := &handleroptions.ReturnedMetadataLimited{}
 	jw := json.NewWriter(w)
 	jw.BeginArray()
 
 	for _, result := range results {
-		writeFindNodeResultJSON(jw, result, opts)
+		writeFindNodeResultJSON(jw, result, opts, meta)
 	}
 
 	jw.EndArray()
-	return jw.Close()
+	return meta, jw.Close()
 }
 
 type findResult struct {
@@ -207,6 +210,7 @@ func writeFindNodeResultJSON(
 	jw json.Writer,
 	result findResult,
 	opts findResultsOptions,
+	meta *handleroptions.ReturnedMetadataLimited,
 ) {
 	descriptor := result.node
 
@@ -218,11 +222,11 @@ func writeFindNodeResultJSON(
 			descriptor.hasChildren &&
 			opts.includeBothExpandableAndLeaf)
 	if includeLeafNode {
-		writeFindResultJSON(jw, result.id, result.name, false)
+		writeFindResultJSON(jw, result.id, result.name, false, opts, meta)
 	}
 
 	if descriptor.hasChildren {
-		writeFindResultJSON(jw, result.id, result.name, true)
+		writeFindResultJSON(jw, result.id, result.name, true, opts, meta)
 	}
 }
 
@@ -231,7 +235,17 @@ func writeFindResultJSON(
 	id string,
 	value string,
 	hasChildren bool,
+	opts findResultsOptions,
+	meta *handleroptions.ReturnedMetadataLimited,
 ) {
+	meta.TotalResults++
+	if opts.returnedSeriesMetadataLimit > 0 && meta.Results >= opts.returnedSeriesMetadataLimit {
+		meta.Limited = true
+		return
+	}
+
+	meta.Results++
+
 	leaf := 1
 	if hasChildren {
 		leaf = 0