<commit_before>// 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 limits

import (
	"fmt"
	"sync"
	"time"

	"github.com/uber-go/tally"
	"go.uber.org/atomic"
	"go.uber.org/zap"

	xerrors "github.com/m3db/m3/src/x/errors"
	"github.com/m3db/m3/src/x/instrument"
)

const (
	disabledLimitValue = 0
	defaultLookback    = time.Second * 15
)

type queryLimits struct {
	docsLimit           *lookbackLimit
	bytesReadLimit      *lookbackLimit
	aggregatedDocsLimit *lookbackLimit
}

type lookbackLimit struct {
	name      string
	started   bool
	options   LookbackLimitOptions
	metrics   lookbackLimitMetrics
	logger    *zap.Logger
	recent    *atomic.Int64
	stopCh    chan struct{}
	stoppedCh chan struct{}
	lock      sync.RWMutex
	iOpts     instrument.Options
}

type lookbackLimitMetrics struct {
	optionsLimit    tally.Gauge
	optionsLookback tally.Gauge
	recentCount     tally.Gauge
	recentMax       tally.Gauge
	total           tally.Counter
	exceeded        tally.Counter

	sourceLogger SourceLogger
}

var (
	_ QueryLimits   = (*queryLimits)(nil)
	_ LookbackLimit = (*lookbackLimit)(nil)
)

// DefaultLookbackLimitOptions returns a new query limits manager.
func DefaultLookbackLimitOptions() LookbackLimitOptions {
	return LookbackLimitOptions{
		// Default to no limit.
		Limit:    disabledLimitValue,
		Lookback: defaultLookback,
	}
}

// NewQueryLimits returns a new query limits manager.
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()
		aggDocsLimitOpts    = options.AggregateDocsLimitOpts()
		sourceLoggerBuilder = options.SourceLoggerBuilder()

		docsMatched      = "docs-matched"
		bytesRead        = "disk-bytes-read"
		aggregateMatched = "aggregate-matched"
		docsLimit        = newLookbackLimit(
			iOpts, docsLimitOpts, docsMatched, docsMatched,
			sourceLoggerBuilder, map[string]string{"type": "fetch"})
		bytesReadLimit = newLookbackLimit(
			iOpts, bytesReadLimitOpts, bytesRead, bytesRead,
			sourceLoggerBuilder, nil)

		aggregatedDocsLimit = newLookbackLimit(
			iOpts, aggDocsLimitOpts, docsMatched, aggregateMatched,
			sourceLoggerBuilder, map[string]string{"type": "aggregate"})
	)

	return &queryLimits{
		docsLimit:           docsLimit,
		bytesReadLimit:      bytesReadLimit,
		aggregatedDocsLimit: aggregatedDocsLimit,
	}, nil
}

// NewLookbackLimit returns a new lookback limit.
func NewLookbackLimit(
	instrumentOpts instrument.Options,
	opts LookbackLimitOptions,
	name string,
	sourceLoggerBuilder SourceLoggerBuilder,
	tags map[string]string,
) LookbackLimit {
	return newLookbackLimit(instrumentOpts, opts, name, name, sourceLoggerBuilder, tags)
}

func newLookbackLimit(
	instrumentOpts instrument.Options,
	opts LookbackLimitOptions,
	metricName string,
	name string,
	sourceLoggerBuilder SourceLoggerBuilder,
	tags map[string]string,
) *lookbackLimit {
	metrics := newLookbackLimitMetrics(
		instrumentOpts,
		metricName,
		sourceLoggerBuilder,
		tags,
	)

	return &lookbackLimit{
		name:      name,
		options:   opts,
		metrics:   metrics,
		logger:    instrumentOpts.Logger(),
		recent:    atomic.NewInt64(0),
		stopCh:    make(chan struct{}),
		stoppedCh: make(chan struct{}),
		iOpts:     instrumentOpts,
	}
}

func newLookbackLimitMetrics(
	instrumentOpts instrument.Options,
	name string,
	sourceLoggerBuilder SourceLoggerBuilder,
	tags map[string]string,
) lookbackLimitMetrics {
	scope := instrumentOpts.
		MetricsScope().
		SubScope("query-limit")

	if tags != nil {
		scope = scope.Tagged(tags)
	}

	return lookbackLimitMetrics{
		optionsLimit:    scope.Gauge(fmt.Sprintf("current-limit%s", name)),
		optionsLookback: scope.Gauge(fmt.Sprintf("current-lookback-%s", name)),
		recentCount:     scope.Gauge(fmt.Sprintf("recent-count-%s", name)),
		recentMax:       scope.Gauge(fmt.Sprintf("recent-max-%s", name)),
		total:           scope.Counter(fmt.Sprintf("total-%s", name)),
		exceeded:        scope.Tagged(map[string]string{"limit": name}).Counter("exceeded"),

		// nb: no need to provide query-limits subscope to source logger,
		// as it's not directly related to limits.
		sourceLogger: sourceLoggerBuilder.NewSourceLogger(name, instrumentOpts),
	}
}

func (q *queryLimits) FetchDocsLimit() LookbackLimit {<commit_msg>[instrumentation] Clean up some metric names (#3241)<commit_after>// 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 limits

import (
	"fmt"
	"sync"
	"time"

	"github.com/uber-go/tally"
	"go.uber.org/atomic"
	"go.uber.org/zap"

	xerrors "github.com/m3db/m3/src/x/errors"
	"github.com/m3db/m3/src/x/instrument"
)

const (
	disabledLimitValue = 0
	defaultLookback    = time.Second * 15
)

type queryLimits struct {
	docsLimit           *lookbackLimit
	bytesReadLimit      *lookbackLimit
	aggregatedDocsLimit *lookbackLimit
}

type lookbackLimit struct {
	name      string
	started   bool
	options   LookbackLimitOptions
	metrics   lookbackLimitMetrics
	logger    *zap.Logger
	recent    *atomic.Int64
	stopCh    chan struct{}
	stoppedCh chan struct{}
	lock      sync.RWMutex
	iOpts     instrument.Options
}

type lookbackLimitMetrics struct {
	optionsLimit    tally.Gauge
	optionsLookback tally.Gauge
	recentCount     tally.Gauge
	recentMax       tally.Gauge
	total           tally.Counter
	exceeded        tally.Counter

	sourceLogger SourceLogger
}

var (
	_ QueryLimits   = (*queryLimits)(nil)
	_ LookbackLimit = (*lookbackLimit)(nil)
)

// DefaultLookbackLimitOptions returns a new query limits manager.
func DefaultLookbackLimitOptions() LookbackLimitOptions {
	return LookbackLimitOptions{
		// Default to no limit.
		Limit:    disabledLimitValue,
		Lookback: defaultLookback,
	}
}

// NewQueryLimits returns a new query limits manager.
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()
		aggDocsLimitOpts    = options.AggregateDocsLimitOpts()
		sourceLoggerBuilder = options.SourceLoggerBuilder()

		docsMatched      = "docs-matched"
		bytesRead        = "disk-bytes-read"
		aggregateMatched = "aggregate-matched"
		docsLimit        = newLookbackLimit(
			iOpts, docsLimitOpts, docsMatched, docsMatched,
			sourceLoggerBuilder, map[string]string{"type": "fetch"})
		bytesReadLimit = newLookbackLimit(
			iOpts, bytesReadLimitOpts, bytesRead, bytesRead,
			sourceLoggerBuilder, nil)

		aggregatedDocsLimit = newLookbackLimit(
			iOpts, aggDocsLimitOpts, docsMatched, aggregateMatched,
			sourceLoggerBuilder, map[string]string{"type": "aggregate"})
	)

	return &queryLimits{
		docsLimit:           docsLimit,
		bytesReadLimit:      bytesReadLimit,
		aggregatedDocsLimit: aggregatedDocsLimit,
	}, nil
}

// NewLookbackLimit returns a new lookback limit.
func NewLookbackLimit(
	instrumentOpts instrument.Options,
	opts LookbackLimitOptions,
	name string,
	sourceLoggerBuilder SourceLoggerBuilder,
	tags map[string]string,
) LookbackLimit {
	return newLookbackLimit(instrumentOpts, opts, name, name, sourceLoggerBuilder, tags)
}

func newLookbackLimit(
	instrumentOpts instrument.Options,
	opts LookbackLimitOptions,
	metricName string,
	name string,
	sourceLoggerBuilder SourceLoggerBuilder,
	tags map[string]string,
) *lookbackLimit {
	metrics := newLookbackLimitMetrics(
		instrumentOpts,
		metricName,
		sourceLoggerBuilder,
		tags,
	)

	return &lookbackLimit{
		name:      name,
		options:   opts,
		metrics:   metrics,
		logger:    instrumentOpts.Logger(),
		recent:    atomic.NewInt64(0),
		stopCh:    make(chan struct{}),
		stoppedCh: make(chan struct{}),
		iOpts:     instrumentOpts,
	}
}

func newLookbackLimitMetrics(
	instrumentOpts instrument.Options,
	name string,
	sourceLoggerBuilder SourceLoggerBuilder,
	tags map[string]string,
) lookbackLimitMetrics {
	loggerScope := instrumentOpts.MetricsScope()
	if tags != nil {
		loggerScope = loggerScope.Tagged(tags)
	}

	var (
		loggerOpts  = instrumentOpts.SetMetricsScope(loggerScope)
		metricScope = loggerScope.SubScope("query-limit")
	)

	return lookbackLimitMetrics{
		optionsLimit:    metricScope.Gauge(fmt.Sprintf("current-limit-%s", name)),
		optionsLookback: metricScope.Gauge(fmt.Sprintf("current-lookback-%s", name)),
		recentCount:     metricScope.Gauge(fmt.Sprintf("recent-count-%s", name)),
		recentMax:       metricScope.Gauge(fmt.Sprintf("recent-max-%s", name)),
		total:           metricScope.Counter(fmt.Sprintf("total-%s", name)),
		exceeded:        metricScope.Tagged(map[string]string{"limit": name}).Counter("exceeded"),

		sourceLogger: sourceLoggerBuilder.NewSourceLogger(name, loggerOpts),
	}
}

func (q *queryLimits) FetchDocsLimit() LookbackLimit {
{'fraction_non_alphanumeric': 0.09134571816946081, 'fraction_numerical': 0.0032623470774807433, 'mean_word_length': 4.190968955785513, 'pattern_counts': {'":': 6, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1202515', 'n_tokens_mistral': 3509, 'n_tokens_neox': 3324, 'n_words': 926}