func (c *timeSeriesCollectionType) LookupBatch(
	timestamp float64, mlist metrics.List) (
	fetched map[*timeSeriesType]interface{},
	newOnes, notFetched []*timeSeriesType,
	fetchedTimeStamps map[*timestampSeriesType]float64,
	newTs []*timestampSeriesType,
	notFetchedTimeStamps []*timestampSeriesType,
	err error) {
	if err = metrics.VerifyList(mlist); err != nil {
		return
	}
	c.lock.Lock()
	defer c.lock.Unlock()
	if !c.active {
		err = ErrInactive
		return
	}
	valueByMetric := make(map[*MetricInfo]interface{})
	timestampByGroupId := make(map[int]float64)
	groupIds := make(map[int]bool)
	fetched = make(map[*timeSeriesType]interface{})
	fetchedTimeStamps = make(map[*timestampSeriesType]float64)
	mlen := mlist.Len()
	for i := 0; i < mlen; i++ {
		var avalue metrics.Value
		mlist.Index(i, &avalue)
		kind, subType := types.FromGoValueWithSubType(avalue.Value)
		id := c.metricInfoStore.Register(&avalue, kind, subType)
		if kind == types.Dist {
			distributionRollOvers := c.distributionRollOversByPath[avalue.Path]
			if distributionRollOvers == nil {
				distributionRollOvers = &distributionRollOverType{}
				c.distributionRollOversByPath[avalue.Path] = distributionRollOvers
			}
			distribution := avalue.Value.(*messages.Distribution)
			rollOverCount := distributionRollOvers.UpdateAndFetchRollOverCount(
				id.Ranges(), distribution.Generation)
			currentDistributionTotals := &DistributionTotals{
				Counts:        distExtractCounts(distribution),
				Sum:           distribution.Sum,
				RollOverCount: rollOverCount,
			}
			valueByMetric[id] = currentDistributionTotals
		} else {
			valueByMetric[id] = avalue.Value
		}
		groupIds[id.GroupId()] = true
		if !avalue.TimeStamp.IsZero() {
			timestampByGroupId[id.GroupId()] = duration.TimeToFloat(avalue.TimeStamp)
		}
	}
	for groupId := range groupIds {
		if _, ok := timestampByGroupId[groupId]; !ok {
			timestampByGroupId[groupId] = timestamp
		}
	}
	for id, series := range c.timeSeries {
		if _, ok := valueByMetric[id]; !ok {
			notFetched = append(notFetched, series)
		}
	}
	for id, value := range valueByMetric {
		if c.timeSeries[id] == nil {
			thisTs := timestampByGroupId[id.GroupId()]
			c.timeSeries[id] = newTimeSeriesType(
				id, thisTs, value, c.metrics)
			newOnes = append(newOnes, c.timeSeries[id])
		} else {
			fetched[c.timeSeries[id]] = value
		}
	}
	for groupId, series := range c.timestampSeries {
		if _, ok := timestampByGroupId[groupId]; !ok {
			notFetchedTimeStamps = append(
				notFetchedTimeStamps, series)
		}
	}
	for groupId, ts := range timestampByGroupId {
		if c.timestampSeries[groupId] == nil {
			c.timestampSeries[groupId] = newTimeStampSeriesType(
				groupId, ts, c.metrics)
			newTs = append(newTs, c.timestampSeries[groupId])
		} else {
			fetchedTimeStamps[c.timestampSeries[groupId]] = ts
		}
	}
	return
}
LookupBatch looks up all the metrics in one go and returns the
following:
fetched: timeSeries already in this collection keyed by Metric.
values must be added to these manually.
newOnes: timeSeries just added as a result of this lookup. Since these
are new, the first value added automatically.
notFetched: timeSeries in this collection but not fetched. These
are the time series that should be marked inactive.
ok is true if metrics can be added to this instance or false if this
instance is inactive and closed for new metrics.
package store
import (
	"github.com/Symantec/scotty/metrics"
	"github.com/Symantec/scotty/tsdb"
	"github.com/Symantec/tricorder/go/tricorder/duration"
	"github.com/Symantec/tricorder/go/tricorder/messages"
	"github.com/Symantec/tricorder/go/tricorder/types"
	"strings"
	"sync"
)
type metricInfoStoreType struct {
	ByInfo      map[MetricInfo]*MetricInfo
	ByName      map[string][]*MetricInfo
	rangesCache rangesCacheType
}
func (m *metricInfoStoreType) Init() {
	m.ByInfo = make(map[MetricInfo]*MetricInfo)
	m.ByName = make(map[string][]*MetricInfo)
	m.rangesCache.Init()
}
func (m *metricInfoStoreType) Register(
	metric *metrics.Value, kind, subType types.Type) (
	result *MetricInfo) {
	if kind == types.Unknown {
		panic("Got Unknown type")
	}
	if kind.UsesSubType() && subType == types.Unknown {
		panic("Got unknown sub-type when it is required")
	}
	var ranges *Ranges
	var isNotCumulative bool
	if kind == types.Dist {
		distribution := metric.Value.(*messages.Distribution)
		isNotCumulative = distribution.IsNotCumulative
		upperLimitSlice := distExtractUpperLimits(distribution)
		ranges = m.rangesCache.Get(metric.Path, upperLimitSlice)
	}
	infoStruct := MetricInfo{
		path:            metric.Path,
		description:     metric.Description,
		unit:            metric.Unit,
		kind:            kind,
		subType:         subType,
		ranges:          ranges,
		isNotCumulative: isNotCumulative,
		groupId:         metric.GroupId}
	result, alreadyExists := m.ByInfo[infoStruct]
	if alreadyExists {
		return
	}
	result = &infoStruct
	m.ByInfo[infoStruct] = result
	m.ByName[infoStruct.path] = append(m.ByName[infoStruct.path], result)
	return
}
type timeSeriesCollectionType struct {
	applicationId interface{}
	metrics       *storeMetricsType
	statusChangeLock sync.Mutex
	lock                        sync.Mutex
	timeSeries                  map[*MetricInfo]*timeSeriesType
	timestampSeries             map[int]*timestampSeriesType
	metricInfoStore             metricInfoStoreType
	active                      bool
	iterators                   map[string]*namedIteratorDataType
	distributionRollOversByPath map[string]*distributionRollOverType
}
func newTimeSeriesCollectionType(
	app interface{},
	metrics *storeMetricsType) *timeSeriesCollectionType {
	result := &timeSeriesCollectionType{
		applicationId:               app,
		metrics:                     metrics,
		timeSeries:                  make(map[*MetricInfo]*timeSeriesType),
		timestampSeries:             make(map[int]*timestampSeriesType),
		active:                      true,
		iterators:                   make(map[string]*namedIteratorDataType),
		distributionRollOversByPath: make(map[string]*distributionRollOverType),
	}
	result.metricInfoStore.Init()
	return result
}
func (c *timeSeriesCollectionType) NamedIteratorInfo(name string) (
	startTimes map[int]float64,
	completed map[*MetricInfo]float64,
	timestampSeries []*timestampSeriesType) {
	c.lock.Lock()
	defer c.lock.Unlock()
	timestampSeries = c.tsAllTimeStamps()
	snapshot := c.iterators[name]
	if snapshot != nil {
		startTimes = snapshot.startTimeStamps
		completed = snapshot.completed
	}
	return
}
func (c *timeSeriesCollectionType) NewNamedIterator(
	name string,
	maxFrames int,
	strategy MetricGroupingStrategy) (NamedIterator, IteratorData) {
	startTimes,
		completed,
		timestampSeries := c.NamedIteratorInfo(name)
	allTimeSeries := c.TsAll()
	timesByGroup := make(map[int][]float64, len(timestampSeries))
	var skipped bool
	for _, series := range timestampSeries {
		groupId := series.GroupId()
		timesByGroup[groupId] = series.FindAfter(
			startTimes[groupId], maxFrames)
		if startTimes[groupId] > 0.0 && len(timesByGroup[groupId]) > 0 && timesByGroup[groupId][0] == series.Earliest() {
			skipped = true
		}
	}
	strategy.orderedPartition(allTimeSeries)
	result := &namedIteratorType{
		name:                 name,
		timeSeriesCollection: c,
		startTimeStamps:      startTimes,
		timestamps:           timesByGroup,
		completed:            copyCompleted(completed),
		timeSeries:           allTimeSeries,
		strategy:             strategy,
	}
	timeLeft, percentCaughtUp := timeLeft(timestampSeries, timesByGroup)
	return result, IteratorData{timeLeft, percentCaughtUp, skipped}
}
func timeLeft(
	timestampSeries []*timestampSeriesType,
	timesByGroup map[int][]float64) (float64, FloatVar) {
	result := 0.0
	percentCaughtUpSum := 0.0
	percentCaughtUpCount := uint64(0)
	for _, series := range timestampSeries {
		times := timesByGroup[series.GroupId()]
		timesLen := len(times)
		if timesLen > 0 {
			current := series.Latest() - times[timesLen-1]
			if current > result {
				result = current
			}
			earliest := series.Earliest()
			latest := series.Latest()
			diff := latest - earliest
			if diff > 0.0 {
				percentCaughtUpSum += (times[0] - earliest) / diff * 100.0
				percentCaughtUpCount++
			}
		}
	}
	return result, FloatVar{Sum: percentCaughtUpSum, Count: percentCaughtUpCount}
}
func (c *timeSeriesCollectionType) NewNamedIteratorRollUp(
	name string,
	duration float64,
	maxFrames int,
	strategy MetricGroupingStrategy) (
	NamedIterator, IteratorData) {
	startTimes,
		completed,
		timestampSeries := c.NamedIteratorInfo(name)
	allTimeSeries := c.TsAll()
	timesByGroup := make(map[int][]float64, len(timestampSeries))
	var skipped bool
	for _, series := range timestampSeries {
		groupId := series.GroupId()
		nextTimes := series.FindAfter(startTimes[groupId], 0)
		if startTimes[groupId] > 0.0 && len(nextTimes) > 0 && nextTimes[0] == series.Earliest() {
			skipped = true
		}
		chopForRollUp(duration, maxFrames, &nextTimes)
		timesByGroup[groupId] = nextTimes
	}
	strategy.orderedPartition(allTimeSeries)
	result := &rollUpNamedIteratorType{
		namedIteratorType: &namedIteratorType{
			name:                 name,
			timeSeriesCollection: c,
			startTimeStamps:      startTimes,
			timestamps:           timesByGroup,
			completed:            copyCompleted(completed),
			timeSeries:           allTimeSeries,
			strategy:             strategy,
		},
		strategy: strategy,
		interval: duration,
	}
	timeLeft, percentCaughtUp := timeLeft(timestampSeries, timesByGroup)
	return result, IteratorData{timeLeft, percentCaughtUp, skipped}
}
func (c *timeSeriesCollectionType) StartAtBeginning(names []string) {
	c.lock.Lock()
	defer c.lock.Unlock()
	for _, name := range names {
		delete(c.iterators, name)
	}
}
func (c *timeSeriesCollectionType) SetIteratorTo(destName, srcName string) {
	if destName == srcName {
		return
	}
	c.lock.Lock()
	defer c.lock.Unlock()
	srcProgress, ok := c.iterators[srcName]
	if !ok {
		delete(c.iterators, destName)
	} else {
		c.iterators[destName] = srcProgress
	}
}
func (c *timeSeriesCollectionType) SaveProgress(
	name string, progress *namedIteratorDataType) {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.iterators[name] = progress
}
func (c *timeSeriesCollectionType) tsAll() (
	result []*timeSeriesType) {
	for _, ts := range c.timeSeries {
		result = append(result, ts)
	}
	return
}
func (c *timeSeriesCollectionType) TsAll() []*timeSeriesType {
	c.lock.Lock()
	defer c.lock.Unlock()
	return c.tsAll()
}
func (c *timeSeriesCollectionType) tsAllTimeStamps() (
	result []*timestampSeriesType) {
	for _, ts := range c.timestampSeries {
		result = append(result, ts)
	}
	return
}
func (c *timeSeriesCollectionType) TsAllTimeStamps() []*timestampSeriesType {
	c.lock.Lock()
	defer c.lock.Unlock()
	return c.tsAllTimeStamps()
}
func (c *timeSeriesCollectionType) TsAllAndTimeStampsMarkingInactive() (
	valueSeries []*timeSeriesType,
	timestampSeries []*timestampSeriesType,
	markedInactive bool) {
	c.lock.Lock()
	defer c.lock.Unlock()
	if !c.active {
		return
	}
	c.active = false
	return c.tsAll(), c.tsAllTimeStamps(), true
}
func (c *timeSeriesCollectionType) TsByName(name string) (
	result []*timeSeriesType) {
	c.lock.Lock()
	defer c.lock.Unlock()
	infoList := c.metricInfoStore.ByName[name]
	for _, info := range infoList {
		result = append(result, c.timeSeries[info])
	}
	return
}
func (c *timeSeriesCollectionType) TsAndTimeStampsByName(name string) (
	result []*timeSeriesType, resultTs map[int]*timestampSeriesType) {
	c.lock.Lock()
	defer c.lock.Unlock()
	infoList := c.metricInfoStore.ByName[name]
	resultTs = make(map[int]*timestampSeriesType)
	for _, info := range infoList {
		ts := c.timeSeries[info]
		groupId := ts.id.GroupId()
		result = append(result, ts)
		resultTs[groupId] = c.timestampSeries[groupId]
	}
	return
}
func (c *timeSeriesCollectionType) TsByPrefix(prefix string) (
	result []*timeSeriesType) {
	c.lock.Lock()
	defer c.lock.Unlock()
	for _, info := range c.metricInfoStore.ByInfo {
		if strings.HasPrefix(info.Path(), prefix) {
			result = append(result, c.timeSeries[info])
		}
	}
	return
}
func (c *timeSeriesCollectionType) TsAndTimeStampsByPrefix(prefix string) (
	result []*timeSeriesType, resultTs map[int]*timestampSeriesType) {
	c.lock.Lock()
	defer c.lock.Unlock()
	resultTs = make(map[int]*timestampSeriesType)
	for _, info := range c.metricInfoStore.ByInfo {
		if strings.HasPrefix(info.Path(), prefix) {
			ts := c.timeSeries[info]
			groupId := ts.id.GroupId()
			result = append(result, ts)
			resultTs[groupId] = c.timestampSeries[groupId]
		}
	}
	return
}