func (e *CounterElem) AddValue(timestamp time.Time, value float64) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	lockedAgg.aggregation.Add(value)
	lockedAgg.Unlock()
	return nil
}
AddValue adds a metric value at a given timestamp.
package aggregator
import (
	"fmt"
	"math"
	"sync"
	"time"
	maggregation "github.com/m3db/m3metrics/aggregation"
	"github.com/m3db/m3metrics/metric/id"
	"github.com/m3db/m3metrics/metric/unaggregated"
	"github.com/m3db/m3metrics/pipeline/applied"
	"github.com/m3db/m3metrics/policy"
	"github.com/m3db/m3metrics/transformation"
	"github.com/willf/bitset"
)
type lockedCounterAggregation struct {
	sync.Mutex
	closed      bool
	sourcesSeen *bitset.BitSet
	aggregation counterAggregation
}
type timedCounter struct {
	startAtNanos int64  
	lockedAgg    *lockedCounterAggregation
}
func (ta *timedCounter) Reset() {
	ta.startAtNanos = 0
	ta.lockedAgg = nil
}
type CounterElem struct {
	elemBase
	counterElemBase
	values              []timedCounter  
	toConsume           []timedCounter  
	lastConsumedAtNanos int64           
	lastConsumedValues  []float64       
}
func NewCounterElem(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
	opts Options,
) (*CounterElem, error) {
	e := &CounterElem{
		elemBase: newElemBase(opts),
		values:   make([]timedCounter, 0, defaultNumAggregations),  
	}
	if err := e.ResetSetData(id, sp, aggTypes, pipeline, numForwardedTimes, idPrefixSuffixType); err != nil {
		return nil, err
	}
	return e, nil
}
func MustNewCounterElem(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
	opts Options,
) *CounterElem {
	elem, err := NewCounterElem(id, sp, aggTypes, pipeline, numForwardedTimes, idPrefixSuffixType, opts)
	if err != nil {
		panic(fmt.Errorf("unable to create element: %v", err))
	}
	return elem
}
func (e *CounterElem) ResetSetData(
	id id.RawID,
	sp policy.StoragePolicy,
	aggTypes maggregation.Types,
	pipeline applied.Pipeline,
	numForwardedTimes int,
	idPrefixSuffixType IDPrefixSuffixType,
) error {
	useDefaultAggregation := aggTypes.IsDefault()
	if useDefaultAggregation {
		aggTypes = e.DefaultAggregationTypes(e.aggTypesOpts)
	}
	if err := e.elemBase.resetSetData(id, sp, aggTypes, useDefaultAggregation, pipeline, numForwardedTimes, idPrefixSuffixType); err != nil {
		return err
	}
	if err := e.counterElemBase.ResetSetData(e.aggTypesOpts, aggTypes, useDefaultAggregation); err != nil {
		return err
	}
	if !e.parsedPipeline.HasDerivativeTransform {
		return nil
	}
	numAggTypes := len(e.aggTypes)
	if cap(e.lastConsumedValues) < numAggTypes {
		e.lastConsumedValues = make([]float64, numAggTypes)
	}
	e.lastConsumedValues = e.lastConsumedValues[:numAggTypes]
	for i := 0; i < len(e.lastConsumedValues); i++ {
		e.lastConsumedValues[i] = nan
	}
	return nil
}
func (e *CounterElem) AddUnion(timestamp time.Time, mu unaggregated.MetricUnion) error {
	alignedStart := timestamp.Truncate(e.sp.Resolution().Window).UnixNano()
	lockedAgg, err := e.findOrCreate(alignedStart, createAggregationOptions{})
	if err != nil {
		return err
	}
	lockedAgg.Lock()
	if lockedAgg.closed {
		lockedAgg.Unlock()
		return errAggregationClosed
	}
	lockedAgg.aggregation.AddUnion(mu)
	lockedAgg.Unlock()
	return nil
}