func standardMetricTimestampNanos(windowStartNanos int64, resolution time.Duration) int64 {
	return windowStartNanos + resolution.Nanoseconds()
}
The timestamp of a standard metric is set to the end time boundary of the aggregation
window when the metric is flushed.
package aggregator
import (
	"container/list"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/m3db/m3aggregator/aggregator/handler"
	"github.com/m3db/m3aggregator/aggregator/handler/writer"
	"github.com/m3db/m3metrics/metric/aggregated"
	metricid "github.com/m3db/m3metrics/metric/id"
	"github.com/m3db/m3metrics/policy"
	"github.com/m3db/m3x/clock"
	"github.com/m3db/m3x/log"
	"github.com/uber-go/tally"
)
var (
	errListClosed  = errors.New("metric list is closed")
	errListsClosed = errors.New("metric lists are closed")
)
type metricList interface {
	Resolution() time.Duration
	Len() int
	PushBack(value metricElem) (*list.Element, error)
	Close()
}
type metricProcessingMetrics struct {
	metricConsumeSuccess tally.Counter
	metricConsumeErrors  tally.Counter
	metricDiscarded      tally.Counter
}
func newMetricProcessingMetrics(scope tally.Scope) metricProcessingMetrics {
	return metricProcessingMetrics{
		metricConsumeSuccess: scope.Counter("metric-consume-success"),
		metricConsumeErrors:  scope.Counter("metric-consume-errors"),
		metricDiscarded:      scope.Counter("metric-discarded"),
	}
}
type forwardedMetricProcessingMetrics struct {
	metricConsumed    tally.Counter
	metricDiscarded   tally.Counter
	onConsumedSuccess tally.Counter
	onConsumedErrors  tally.Counter
	onDiscarded       tally.Counter
}
func newForwardedMetricProcessingMetrics(scope tally.Scope) forwardedMetricProcessingMetrics {
	return forwardedMetricProcessingMetrics{
		metricConsumed:    scope.Counter("metric-consumed"),
		metricDiscarded:   scope.Counter("metric-discarded"),
		onConsumedSuccess: scope.Counter("on-consumed-success"),
		onConsumedErrors:  scope.Counter("on-consumed-errors"),
		onDiscarded:       scope.Counter("on-discarded"),
	}
}
type writerMetrics struct {
	flushSuccess tally.Counter
	flushErrors  tally.Counter
}
func newWriterMetrics(scope tally.Scope) writerMetrics {
	return writerMetrics{
		flushSuccess: scope.Counter("flush-success"),
		flushErrors:  scope.Counter("flush-errors"),
	}
}
type baseMetricListMetrics struct {
	flushLocal                  metricProcessingMetrics
	flushLocalWriter            writerMetrics
	flushForwarded              forwardedMetricProcessingMetrics
	flushForwardedWriter        writerMetrics
	flushElemCollected          tally.Counter
	flushDuration               tally.Timer
	flushBeforeCutover          tally.Counter
	flushBetweenCutoverCutoff   tally.Counter
	flushBetweenCutoffBufferEnd tally.Counter
	flushAfterBufferEnd         tally.Counter
	flushBeforeStale            tally.Counter
	flushBeforeDuration         tally.Timer
	discardBefore               tally.Counter
}
func newMetricListMetrics(scope tally.Scope) baseMetricListMetrics {
	flushScope := scope.SubScope("flush")
	flushBeforeScope := scope.SubScope("flush-before")
	flushLocalScope := flushScope.Tagged(map[string]string{"flush-type": "local"})
	flushLocalWriterScope := flushLocalScope.SubScope("writer")
	flushForwardedScope := flushScope.Tagged(map[string]string{"flush-type": "forwarded"})
	flushForwardedWriterScope := flushForwardedScope.SubScope("writer")
	return baseMetricListMetrics{
		flushLocal:                  newMetricProcessingMetrics(flushLocalScope),
		flushLocalWriter:            newWriterMetrics(flushLocalWriterScope),
		flushForwarded:              newForwardedMetricProcessingMetrics(flushForwardedScope),
		flushForwardedWriter:        newWriterMetrics(flushForwardedWriterScope),
		flushElemCollected:          flushScope.Counter("elem-collected"),
		flushDuration:               flushScope.Timer("duration"),
		flushBeforeCutover:          flushScope.Counter("before-cutover"),
		flushBetweenCutoverCutoff:   flushScope.Counter("between-cutover-cutoff"),
		flushBetweenCutoffBufferEnd: flushScope.Counter("between-cutoff-bufferend"),
		flushAfterBufferEnd:         flushScope.Counter("after-bufferend"),
		flushBeforeStale:            flushBeforeScope.Counter("stale"),
		flushBeforeDuration:         flushBeforeScope.Timer("duration"),
		discardBefore:               scope.Counter("discard-before"),
	}
}
type targetNanosFn func(nowNanos int64) int64
type flushBeforeFn func(beforeNanos int64, flushType flushType)
type baseMetricList struct {
	sync.RWMutex
	shard            uint32
	opts             Options
	nowFn            clock.NowFn
	timeLock         *sync.RWMutex
	flushHandler     handler.Handler
	localWriter      writer.Writer
	forwardedWriter  forwardedMetricWriter
	resolution       time.Duration
	targetNanosFn    targetNanosFn
	isEarlierThanFn  isEarlierThanFn
	timestampNanosFn timestampNanosFn
	closed           bool
	aggregations     *list.List
	lastFlushedNanos int64
	toCollect        []*list.Element
	metrics          baseMetricListMetrics
	flushBeforeFn               flushBeforeFn
	consumeLocalMetricFn        flushLocalMetricFn
	discardLocalMetricFn        flushLocalMetricFn
	consumeForwardedMetricFn    flushForwardedMetricFn
	discardForwardedMetricFn    flushForwardedMetricFn
	onForwardingElemConsumedFn  onForwardingElemFlushedFn
	onForwardingElemDiscardedFn onForwardingElemFlushedFn
}
func newBaseMetricList(
	shard uint32,
	resolution time.Duration,
	targetNanosFn targetNanosFn,
	isEarlierThanFn isEarlierThanFn,
	timestampNanosFn timestampNanosFn,
	opts Options,
) (*baseMetricList, error) {
	scope := opts.InstrumentOptions().MetricsScope().SubScope("list").Tagged(
		map[string]string{"resolution": resolution.String()},
	)
	flushHandler := opts.FlushHandler()
	localWriterScope := scope.Tagged(map[string]string{"writer-type": "local"}).SubScope("writer")
	localWriter, err := flushHandler.NewWriter(localWriterScope)
	if err != nil {
		return nil, err
	}
	forwardedWriterScope := scope.Tagged(map[string]string{"writer-type": "forwarded"}).SubScope("writer")
	forwardedWriter := newForwardedWriter(shard, opts.AdminClient(), forwardedWriterScope)
	l := &baseMetricList{
		shard:            shard,
		opts:             opts,
		nowFn:            opts.ClockOptions().NowFn(),
		timeLock:         opts.TimeLock(),
		flushHandler:     flushHandler,
		localWriter:      localWriter,
		forwardedWriter:  forwardedWriter,
		resolution:       resolution,
		targetNanosFn:    targetNanosFn,
		isEarlierThanFn:  isEarlierThanFn,
		timestampNanosFn: timestampNanosFn,
		aggregations:     list.New(),
		metrics:          newMetricListMetrics(scope),
	}
	l.flushBeforeFn = l.flushBefore
	l.consumeLocalMetricFn = l.consumeLocalMetric
	l.discardLocalMetricFn = l.discardLocalMetric
	l.consumeForwardedMetricFn = l.consumeForwardedMetric
	l.discardForwardedMetricFn = l.discardForwardedMetric
	l.onForwardingElemConsumedFn = l.onForwardingElemConsumed
	l.onForwardingElemDiscardedFn = l.onForwardingElemDiscarded
	return l, nil
}
func (l *baseMetricList) Shard() uint32                { return l.shard }
func (l *baseMetricList) Resolution() time.Duration    { return l.resolution }
func (l *baseMetricList) FlushInterval() time.Duration { return l.resolution }
func (l *baseMetricList) LastFlushedNanos() int64      { return atomic.LoadInt64(&l.lastFlushedNanos) }
func (l *baseMetricList) Len() int {
	l.RLock()
	numElems := l.aggregations.Len()
	l.RUnlock()
	return numElems
}
func (l *baseMetricList) PushBack(value metricElem) (*list.Element, error) {
	var (
		forwardedMetricType         = value.Type()
		forwardedID, hasForwardedID = value.ForwardedID()
		forwardedAggregationKey, _  = value.ForwardedAggregationKey()
	)
	l.Lock()
	if l.closed {
		l.Unlock()
		return nil, errListClosed
	}
	elem := l.aggregations.PushBack(value)
	if !hasForwardedID {
		l.Unlock()
		return elem, nil
	}
	writeForwardedFn, onForwardedWrittenFn, err := l.forwardedWriter.Register(
		forwardedMetricType,
		forwardedID,
		forwardedAggregationKey,
	)
	if err != nil {
		l.Unlock()
		return nil, err
	}
	value.SetForwardedCallbacks(writeForwardedFn, onForwardedWrittenFn)
	l.Unlock()
	return elem, nil
}
func (l *baseMetricList) Close() bool {
	l.Lock()
	defer l.Unlock()
	if l.closed {
		return false
	}
	l.localWriter.Close()
	l.forwardedWriter.Close()
	l.closed = true
	return true
}
func (l *baseMetricList) Flush(req flushRequest) {
	start := l.nowFn()
	defer func() {
		took := l.nowFn().Sub(start)
		l.metrics.flushDuration.Record(took)
	}()
	l.timeLock.Lock()
	nowNanos := l.nowFn().UnixNano()
	l.timeLock.Unlock()
	targetNanos := l.targetNanosFn(nowNanos)
	if targetNanos <= req.CutoverNanos {
		l.flushBeforeFn(targetNanos, discardType)
		l.metrics.flushBeforeCutover.Inc(1)
		return
	}
	if req.CutoverNanos > 0 {
		l.flushBeforeFn(req.CutoverNanos, discardType)
	}
	if targetNanos <= req.CutoffNanos {
		l.flushBeforeFn(targetNanos, consumeType)
		l.metrics.flushBetweenCutoverCutoff.Inc(1)
		return
	}
	l.flushBeforeFn(req.CutoffNanos, consumeType)
	bufferEndNanos := targetNanos - int64(req.BufferAfterCutoff)
	if bufferEndNanos <= req.CutoffNanos {
		l.metrics.flushBetweenCutoffBufferEnd.Inc(1)
		return
	}
	l.flushBeforeFn(bufferEndNanos, discardType)
	l.metrics.flushAfterBufferEnd.Inc(1)
}
func (l *baseMetricList) DiscardBefore(beforeNanos int64) {
	l.flushBeforeFn(beforeNanos, discardType)
	l.metrics.discardBefore.Inc(1)
}
func (l *baseMetricList) flushBefore(beforeNanos int64, flushType flushType) {
	if l.LastFlushedNanos() >= beforeNanos {
		l.metrics.flushBeforeStale.Inc(1)
		return
	}
	flushBeforeStart := l.nowFn()
	l.toCollect = l.toCollect[:0]
	flushLocalFn := l.consumeLocalMetricFn
	flushForwardedFn := l.consumeForwardedMetricFn
	onForwardedFlushedFn := l.onForwardingElemConsumedFn
	if flushType == discardType {
		flushLocalFn = l.discardLocalMetricFn
		flushForwardedFn = l.discardForwardedMetricFn
		onForwardedFlushedFn = l.onForwardingElemDiscardedFn
	}
	l.RLock()
	l.forwardedWriter.Prepare()
	for e := l.aggregations.Front(); e != nil; e = e.Next() {
		elem := e.Value.(metricElem)
		if elem.Consume(
			beforeNanos,
			l.isEarlierThanFn,
			l.timestampNanosFn,
			flushLocalFn,
			flushForwardedFn,
			onForwardedFlushedFn,
		) {
			l.toCollect = append(l.toCollect, e)
		}
	}
	l.RUnlock()
	if flushType == consumeType {
		if err := l.localWriter.Flush(); err != nil {
			l.metrics.flushLocalWriter.flushErrors.Inc(1)
		} else {
			l.metrics.flushLocalWriter.flushSuccess.Inc(1)
		}
		if l.forwardedWriter != nil {
			if err := l.forwardedWriter.Flush(); err != nil {
				l.metrics.flushForwardedWriter.flushErrors.Inc(1)
			} else {
				l.metrics.flushForwardedWriter.flushSuccess.Inc(1)
			}
		}
	}
	l.Lock()
	for _, e := range l.toCollect {
		elem := e.Value.(metricElem)
		if forwardedID, hasForwardedID := elem.ForwardedID(); hasForwardedID {
			forwardedType := elem.Type()
			forwardedAggregationKey, _ := elem.ForwardedAggregationKey()
			l.forwardedWriter.Unregister(forwardedType, forwardedID, forwardedAggregationKey)
		}
		elem.Close()
		e.Value = nil
		l.aggregations.Remove(e)
	}
	numCollected := len(l.toCollect)
	l.Unlock()
	atomic.StoreInt64(&l.lastFlushedNanos, beforeNanos)
	l.metrics.flushElemCollected.Inc(int64(numCollected))
	flushBeforeDuration := l.nowFn().Sub(flushBeforeStart)
	l.metrics.flushBeforeDuration.Record(flushBeforeDuration)
}
func (l *baseMetricList) consumeLocalMetric(
	idPrefix []byte,
	id metricid.RawID,
	idSuffix []byte,
	timeNanos int64,
	value float64,
	sp policy.StoragePolicy,
) {
	chunkedID := metricid.ChunkedID{
		Prefix: idPrefix,
		Data:   []byte(id),
		Suffix: idSuffix,
	}
	chunkedMetricWithPolicy := aggregated.ChunkedMetricWithStoragePolicy{
		ChunkedMetric: aggregated.ChunkedMetric{
			ChunkedID: chunkedID,
			TimeNanos: timeNanos,
			Value:     value,
		},
		StoragePolicy: sp,
	}
	if err := l.localWriter.Write(chunkedMetricWithPolicy); err != nil {
		l.metrics.flushLocal.metricConsumeErrors.Inc(1)
	} else {
		l.metrics.flushLocal.metricConsumeSuccess.Inc(1)
	}
}
func (l *baseMetricList) discardLocalMetric(
	idPrefix []byte,
	id metricid.RawID,
	idSuffix []byte,
	timeNanos int64,
	value float64,
	sp policy.StoragePolicy,
) {
	l.metrics.flushLocal.metricDiscarded.Inc(1)
}
func (l *baseMetricList) consumeForwardedMetric(
	writeFn writeForwardedMetricFn,
	aggregationKey aggregationKey,
	timeNanos int64,
	value float64,
) {
	writeFn(aggregationKey, timeNanos, value)
	l.metrics.flushForwarded.metricConsumed.Inc(1)
}
func (l *baseMetricList) discardForwardedMetric(
	writeFn writeForwardedMetricFn,
	aggregationKey aggregationKey,
	timeNanos int64,
	value float64,
) {
	l.metrics.flushForwarded.metricDiscarded.Inc(1)
}
func (l *baseMetricList) onForwardingElemConsumed(
	onForwardedWrittenFn onForwardedAggregationDoneFn,
	aggregationKey aggregationKey,
) {
	if err := onForwardedWrittenFn(aggregationKey); err != nil {
		l.metrics.flushForwarded.onConsumedErrors.Inc(1)
	} else {
		l.metrics.flushForwarded.onConsumedSuccess.Inc(1)
	}
}
func (l *baseMetricList) onForwardingElemDiscarded(
	onForwardedWrittenFn onForwardedAggregationDoneFn,
	aggregationKey aggregationKey,
) {
	l.metrics.flushForwarded.onDiscarded.Inc(1)
}
func standardMetricTargetNanos(nowNanos int64) int64 { return nowNanos }
func isStandardMetricEarlierThan(
	windowStartNanos int64,
	resolution time.Duration,
	targetNanos int64,
) bool {
	return windowStartNanos+resolution.Nanoseconds() <= targetNanos
}