<commit_before>package mdata

import (
	"errors"
	"fmt"
	"math"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/grafana/metrictank/cluster"
	"github.com/grafana/metrictank/conf"
	"github.com/grafana/metrictank/consolidation"
	"github.com/grafana/metrictank/mdata/cache"
	"github.com/grafana/metrictank/mdata/chunk"
	mdataerrors "github.com/grafana/metrictank/mdata/errors"
	"github.com/grafana/metrictank/schema"
	log "github.com/sirupsen/logrus"
)

var ErrInvalidRange = errors.New("AggMetric: invalid range: from must be less than to")

// AggMetric takes in new values, updates the in-memory data and streams the points to aggregators
// it uses a circular buffer of chunks
// each chunk starts at their respective t0
// a t0 is a timestamp divisible by chunkSpan without a remainder (e.g. 2 hour boundaries)
// firstT0's data is held at index 0, indexes go up and wrap around from numChunks-1 to 0
// in addition, keep in mind that the last chunk is always a work in progress and not useable for aggregation
// AggMetric is concurrency-safe
type AggMetric struct {
	store       Store
	cachePusher cache.CachePusher
	sync.RWMutex
	key             schema.AMKey
	rob             *ReorderBuffer
	currentChunkPos int    // Chunks[CurrentChunkPos] is active. Others are finished. Only valid when len(chunks) > 0, e.g. when data has been written (excl ROB data)
	numChunks       uint32 // max size of the circular buffer
	chunkSpan       uint32 // span of individual chunks in seconds
	chunks          []*chunk.Chunk
	aggregators     []*Aggregator
	dropFirstChunk  bool
	ingestFromT0    uint32
	futureTolerance uint32
	ttl             uint32
	lastSaveStart   uint32 // last chunk T0 that was added to the write Queue.
	lastSaveFinish  uint32 // last chunk T0 successfully written to Cassandra.
	lastWrite       uint32 // wall clock time of when last point was successfully added (possibly to the ROB)
	firstTs         uint32 // timestamp of first point seen
}

// NewAggMetric creates a metric with given key, it retains the given number of chunks each chunkSpan seconds long
// it optionally also creates aggregations with the given settings
// the 0th retention is the native archive of this metric. if there's several others, we create aggregators, using agg.
// it's the callers responsibility to make sure agg is not nil in that case!
// If reorderWindow is greater than 0, a reorder buffer is enabled. In that case data points with duplicate timestamps
// the behavior is defined by reorderAllowUpdate
func NewAggMetric(store Store, cachePusher cache.CachePusher, key schema.AMKey, retentions conf.Retentions, reorderWindow, interval uint32, agg *conf.Aggregation, reorderAllowUpdate, dropFirstChunk bool, ingestFrom int64) *AggMetric {

	// note: during parsing of retentions, we assure there's at least 1.
	ret := retentions.Rets[0]

	m := AggMetric{
		cachePusher:     cachePusher,
		store:           store,
		key:             key,
		chunkSpan:       ret.ChunkSpan,
		numChunks:       ret.NumChunks,
		chunks:          make([]*chunk.Chunk, 0, ret.NumChunks),
		dropFirstChunk:  dropFirstChunk,
		futureTolerance: uint32(ret.MaxRetention()) * uint32(futureToleranceRatio) / 100,
		ttl:             uint32(ret.MaxRetention()),
		// we set LastWrite here to make sure a new Chunk doesn't get immediately
		// garbage collected right after creating it, before we can push to it.
		lastWrite: uint32(time.Now().Unix()),
	}
	if ingestFrom > 0 {
		// we only want to ingest data that will go into chunks with a t0 >= 'ingestFrom'.
		m.ingestFromT0 = AggBoundary(uint32(ingestFrom), ret.ChunkSpan)
	}
	if reorderWindow != 0 {
		m.rob = NewReorderBuffer(reorderWindow, interval, reorderAllowUpdate)
	}

	origSplits := strings.Split(retentions.Orig, ":")
	for i, ret := range retentions.Rets[1:] {
		retOrig := origSplits[i+1]
		m.aggregators = append(m.aggregators, NewAggregator(store, cachePusher, key, retOrig, ret, *agg, dropFirstChunk, ingestFrom))
	}

	return &m
}

// Sync the saved state of a chunk by its T0.
func (a *AggMetric) SyncChunkSaveState(ts uint32, sendPersist bool) ChunkSaveCallback {
	return func() {
		a.Lock()
		if ts > a.lastSaveFinish {
			a.lastSaveFinish = ts
		}
		if ts > a.lastSaveStart {
			a.lastSaveStart = ts
		}
		a.Unlock()
		log.Debugf("AM: metric %s at chunk T0=%d has been saved.", a.key, ts)
		if sendPersist {
			SendPersistMessage(a.key.String(), ts)
		}
	}
}

// Sync the saved state of a chunk by its T0.
func (a *AggMetric) SyncAggregatedChunkSaveState(ts uint32, consolidator consolidation.Consolidator, aggSpan uint32) {
	// no lock needed cause aggregators don't change at runtime
	for _, a := range a.aggregators {
		if a.span == aggSpan {
			switch consolidator {
			case consolidation.None:
				panic("cannot get an archive for no consolidation")
			case consolidation.Avg:
				panic("avg consolidator has no matching Archive(). you need sum and cnt")
			case consolidation.Cnt:
				if a.cntMetric != nil {
					a.cntMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Min:
				if a.minMetric != nil {
					a.minMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Max:
				if a.maxMetric != nil {
					a.maxMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Sum:
				if a.sumMetric != nil {
					a.sumMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Lst:
				if a.lstMetric != nil {
					a.lstMetric.SyncChunkSaveState(ts, false)()
				}
				return
			default:
				panic(fmt.Sprintf("internal error: no such consolidator %q with span %d", consolidator, aggSpan))
			}
		}
	}
}

func (a *AggMetric) GetAggregated(consolidator consolidation.Consolidator, aggSpan, from, to uint32) (Result, error) {
	// no lock needed cause aggregators don't change at runtime
	for _, a := range a.aggregators {
		if a.span == aggSpan {
			var agg *AggMetric
			switch consolidator {
			case consolidation.None:
				err := errors.New("internal error: AggMetric.GetAggregated(): cannot get an archive for no consolidation")
				log.Errorf("AM: %s", err.Error())
				badConsolidator.Inc()
				return Result{}, err
			case consolidation.Avg:
				err := errors.New("internal error: AggMetric.GetAggregated(): avg consolidator has no matching Archive(). you need sum and cnt")
				log.Errorf("AM: %s", err.Error())
				badConsolidator.Inc()
				return Result{}, err
			case consolidation.Cnt:
				agg = a.cntMetric
			case consolidation.Lst:
				agg = a.lstMetric
			case consolidation.Min:
				agg = a.minMetric
			case consolidation.Max:
				agg = a.maxMetric
			case consolidation.Sum:
				agg = a.sumMetric
			default:
				err := fmt.Errorf("internal error: AggMetric.GetAggregated(): unknown consolidator %q", consolidator)
				log.Errorf("AM: %s", err.Error())
				badConsolidator.Inc()
				return Result{}, err
			}
			if agg == nil {
				return Result{}, fmt.Errorf("Consolidator %q not configured", consolidator)
			}
			return agg.Get(from, to)
		}
	}
	err := fmt.Errorf("internal error: AggMetric.GetAggregated(): unknown aggSpan %d", aggSpan)
	log.Errorf("AM: %s", err.Error())
	badAggSpan.Inc()
	return Result{}, err
}

// Get all data between the requested time ranges. From is inclusive, to is exclusive. from <= x < to
// more data then what's requested may be included
// specifically, returns:
// * points from the ROB (if enabled)
// * iters from matching chunks
// * oldest point we have, so that if your query needs data before it, the caller knows when to query the store
func (a *AggMetric) Get(from, to uint32) (Result, error) {
	pre := time.Now()
	log.Debugf("AM: %s Get(): %d - %d (%s - %s) span:%ds", a.key, from, to, TS(from), TS(to), to-from-1)
	if from >= to {
		return Result{}, ErrInvalidRange
	}
	a.RLock()
	defer a.RUnlock()

	result := Result{
		Oldest: math.MaxInt32,
	}

	if a.rob != nil {
		result.Points = a.rob.Get()
		if len(result.Points) > 0 {
			result.Oldest = result.Points[0].Ts
			if result.Oldest <= from {
				return result, nil
			}
		}
	}

	if len(a.chunks) == 0 {
		// we dont have any data yet.
		log.Debugf("AM: %s Get(): no data for requested range.", a.key)
		return result, nil
	}

	newestChunk := a.chunks[a.currentChunkPos]

	if from >= newestChunk.Series.T0+a.chunkSpan {
		// request falls entirely ahead of the data we have
		// this can happen in a few cases:
		// * queries for the most recent data, but our ingestion has fallen behind.
		//   we don't want such a degradation to cause a storm of cassandra queries
		//   we should just return an oldest value that is <= from so we don't hit cassandra
		//   but it doesn't have to be the real oldest value, so do whatever is cheap.
		// * data got once written by another node, but has been re-sending (perhaps fixed)
		//   to this node, but not yet up to the point it was previously sent. so we're
		//   only aware of older data and not the newer data in cassandra. this is unlikely
		//   and it's better to not serve this scenario well in favor of the above case.
		//   seems like a fair tradeoff anyway that you have to refill all the way first.
		log.Debugf("AM: %s Get(): no data for requested range.", a.key)
		result.Oldest = from
		return result, nil
	}

	// get the oldest chunk we have.
	// eg if we have 5 chunks, N is the current chunk and n-4 is the oldest chunk.
	// -----------------------------
	// | n-4 | n-3 | n-2 | n-1 | n |  CurrentChunkPos = 4
	// -----------------------------
	// -----------------------------
	// | n | n-4 | n-3 | n-2 | n-1 |  CurrentChunkPos = 0
	// -----------------------------
	// -----------------------------
	// | n-2 | n-1 | n | n-4 | n-3 |  CurrentChunkPos = 2
	// -----------------------------
	oldestPos := a.currentChunkPos + 1
	if oldestPos >= len(a.chunks) {
		oldestPos = 0
	}

	oldestChunk := a.chunks[oldestPos]

	if to <= oldestChunk.Series.T0 {
		// the requested time range ends before any data we have.
		log.Debugf("AM: %s Get(): no data for requested range", a.key)
		if oldestChunk.First {
			result.Oldest = a.firstTs
		} else {
			result.Oldest = oldestChunk.Series.T0
		}
		return result, nil
	}

	// Find the oldest Chunk that the "from" ts falls in.  If from extends before the oldest
	// chunk, then we just use the oldest chunk.
	for from >= oldestChunk.Series.T0+a.chunkSpan {
		oldestPos++
		if oldestPos >= len(a.chunks) {
			oldestPos = 0
		}
		oldestChunk = a.chunks[oldestPos]
	}

	// find the newest Chunk that "to" falls in.  If "to" extends to after the newest data
	// then just return the newest chunk.
	// some examples to clarify this more. assume newestChunk.T0 is at 120, then
	// for a to of 121 -> data up to (incl) 120 -> stay at this chunk, it has a point we need
	// for a to of 120 -> data up to (incl) 119 -> use older chunk
	// for a to of 119 -> data up to (incl) 118 -> use older chunk
	newestPos := a.currentChunkPos
	for to <= newestChunk.Series.T0 {
		newestPos--
		if newestPos < 0 {
			newestPos += len(a.chunks)
		}
		newestChunk = a.chunks[newestPos]
	}

	// now just start at oldestPos and move through the Chunks circular Buffer to newestPos
	for {
		c := a.chunks[oldestPos]
		result.Iters = append(result.Iters, c.Series.Iter())

		if oldestPos == newestPos {
			break
		}

		oldestPos++
		if oldestPos >= len(a.chunks) {
			oldestPos = 0
		}
	}

	if oldestChunk.First {
		result.Oldest = a.firstTs
	} else {
		result.Oldest = oldestChunk.Series.T0
	}

	memToIterDuration.Value(time.Now().Sub(pre))
	return result, nil
}

// caller must hold lock
func (a *AggMetric) addAggregators(ts uint32, val float64) {
	for _, agg := range a.aggregators {
		if log.IsLevelEnabled(log.DebugLevel) {
			log.Debugf("AM: %s pushing %d,%f to aggregator %d", a.key, ts, val, agg.span)
		}
		agg.Add(ts, val)
	}
}

// pushToCache adds the chunk into the cache if it is hot
// caller must hold lock
func (a *AggMetric) pushToCache(c *chunk.Chunk) {
	if a.cachePusher == nil {
		return
	}
	// push into cache
	intervalHint := a.key.Archive.Span()

	itergen, err := chunk.NewIterGen(c.Series.T0, intervalHint, c.Encode(a.chunkSpan))
	if err != nil {
		log.Errorf("AM: %s failed to generate IterGen. this should never happen: %s", a.key, err)
	}
	go a.cachePusher.AddIfHot(a.key, 0, itergen)
}

// write a chunk to persistent storage.
// never persist a chunk that may receive further updates!
// (because the stores will read out chunk data on the unlocked chunk)
// caller must hold lock.
func (a *AggMetric) persist(pos int) {
	chunk := a.chunks[pos]
	pre := time.Now()

	if a.lastSaveStart >= chunk.Series.T0 {
		// this can happen if
		// a) there are 2 primary MT nodes both saving chunks to Cassandra
		// b) a primary failed and this node was promoted to be primary but metric consuming is lagging.
		// c) chunk was persisted by GC (stale) and then new data triggered another persist call
		// d) dropFirstChunk is enabled and this is the first chunk
		log.Debugf("AM: persist(): duplicate persist call for chunk.")
		return
	}

	// create an array of chunks that need to be sent to the writeQueue.
	pending := make([]*ChunkWriteRequest, 1)
	// add the current chunk to the list of chunks to send to the writeQueue
	cwr := NewChunkWriteRequest(
		a.SyncChunkSaveState(chunk.Series.T0, true),
		a.key,
		a.ttl,
		chunk.Series.T0,
		chunk.Encode(a.chunkSpan),
		time.Now(),
	)
	pending[0] = &cwr

	// if we recently became the primary, there may be older chunks
	// that the old primary did not save.  We should check for those
	// and save them.
	previousPos := pos - 1
	if previousPos < 0 {
		previousPos += len(a.chunks)
	}
	previousChunk := a.chunks[previousPos]
	for (previousChunk.Series.T0 < chunk.Series.T0) && (a.lastSaveStart < previousChunk.Series.T0) {
		log.Debugf("AM: persist(): old chunk needs saving. Adding %s:%d to writeQueue", a.key, previousChunk.Series.T0)
		cwr := NewChunkWriteRequest(
			a.SyncChunkSaveState(previousChunk.Series.T0, true),
			a.key,
			a.ttl,
			previousChunk.Series.T0,
			previousChunk.Encode(a.chunkSpan),
			time.Now(),
		)
		pending = append(pending, &cwr)
		previousPos--
		if previousPos < 0 {
			previousPos += len(a.chunks)
		}
		previousChunk = a.chunks[previousPos]
	}

	// Every chunk with a T0 <= this chunks' T0 is now either saved, or in the writeQueue.
	a.lastSaveStart = chunk.Series.T0

	log.Debugf("AM: persist(): sending %d chunks to write queue", len(pending))

	pendingChunk := len(pending) - 1

	// if the store blocks,
	// the calling function will block waiting for persist() to complete.
	// This is intended to put backpressure on our message handlers so
	// that they stop consuming messages, leaving them to buffer at
	// the message bus. The "pending" array of chunks are processed
	// last-to-first ensuring that older data is added to the store
	// before newer data.
	for pendingChunk >= 0 {
		log.Debugf("AM: persist(): sealing chunk %d/%d (%s:%d) and adding to write queue.", pendingChunk, len(pending), a.key, chunk.Series.T0)
		a.store.Add(pending[pendingChunk])
		pendingChunk--
	}
	persistDuration.Value(time.Now().Sub(pre))
	return
}

// don't ever call with a ts of 0, cause we use 0 to mean not initialized!
func (a *AggMetric) Add(ts uint32, val float64) {
	if ts < a.ingestFromT0 {
		// TODO: add metric to keep track of the # of points discarded
		if log.IsLevelEnabled(log.DebugLevel) {
			log.Debugf("AM: discarding metric <%d,%f>: does not belong to a chunk starting after ingest-from. First chunk considered starts at %d", ts, val, a.ingestFromT0)
		}
		// even if a point is too old for our raw data, it may not be too old for aggregated data
		// for example let's say a chunk starts at t0=3600 but we have 300-secondly aggregates
		// that mean the aggregators need data from 3301 and onwards, because we aggregate 3301-3600 into a point with ts=3600
		a.addAggregators(ts, val)
		return
	}

	// need to check if ts > futureTolerance to prevent that we reject a datapoint
	// because the ts value has wrapped around the uint32 boundary
	if ts > a.futureTolerance && int64(ts-a.futureTolerance) > time.Now().Unix() {
		discardedSampleTooFarAhead.Inc()
<commit_msg>add comment<commit_after>package mdata

import (
	"errors"
	"fmt"
	"math"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/grafana/metrictank/cluster"
	"github.com/grafana/metrictank/conf"
	"github.com/grafana/metrictank/consolidation"
	"github.com/grafana/metrictank/mdata/cache"
	"github.com/grafana/metrictank/mdata/chunk"
	mdataerrors "github.com/grafana/metrictank/mdata/errors"
	"github.com/grafana/metrictank/schema"
	log "github.com/sirupsen/logrus"
)

var ErrInvalidRange = errors.New("AggMetric: invalid range: from must be less than to")

// AggMetric takes in new values, updates the in-memory data and streams the points to aggregators
// it uses a circular buffer of chunks
// each chunk starts at their respective t0
// a t0 is a timestamp divisible by chunkSpan without a remainder (e.g. 2 hour boundaries)
// firstT0's data is held at index 0, indexes go up and wrap around from numChunks-1 to 0
// in addition, keep in mind that the last chunk is always a work in progress and not useable for aggregation
// AggMetric is concurrency-safe
type AggMetric struct {
	store       Store
	cachePusher cache.CachePusher
	sync.RWMutex
	key             schema.AMKey
	rob             *ReorderBuffer
	currentChunkPos int    // Chunks[CurrentChunkPos] is active. Others are finished. Only valid when len(chunks) > 0, e.g. when data has been written (excl ROB data)
	numChunks       uint32 // max size of the circular buffer
	chunkSpan       uint32 // span of individual chunks in seconds
	chunks          []*chunk.Chunk
	aggregators     []*Aggregator
	dropFirstChunk  bool
	ingestFromT0    uint32
	futureTolerance uint32
	ttl             uint32
	lastSaveStart   uint32 // last chunk T0 that was added to the write Queue.
	lastSaveFinish  uint32 // last chunk T0 successfully written to Cassandra.
	lastWrite       uint32 // wall clock time of when last point was successfully added (possibly to the ROB)
	firstTs         uint32 // timestamp of first point seen
}

// NewAggMetric creates a metric with given key, it retains the given number of chunks each chunkSpan seconds long
// it optionally also creates aggregations with the given settings
// the 0th retention is the native archive of this metric. if there's several others, we create aggregators, using agg.
// it's the callers responsibility to make sure agg is not nil in that case!
// If reorderWindow is greater than 0, a reorder buffer is enabled. In that case data points with duplicate timestamps
// the behavior is defined by reorderAllowUpdate
func NewAggMetric(store Store, cachePusher cache.CachePusher, key schema.AMKey, retentions conf.Retentions, reorderWindow, interval uint32, agg *conf.Aggregation, reorderAllowUpdate, dropFirstChunk bool, ingestFrom int64) *AggMetric {

	// note: during parsing of retentions, we assure there's at least 1.
	ret := retentions.Rets[0]

	m := AggMetric{
		cachePusher:     cachePusher,
		store:           store,
		key:             key,
		chunkSpan:       ret.ChunkSpan,
		numChunks:       ret.NumChunks,
		chunks:          make([]*chunk.Chunk, 0, ret.NumChunks),
		dropFirstChunk:  dropFirstChunk,
		futureTolerance: uint32(ret.MaxRetention()) * uint32(futureToleranceRatio) / 100,
		ttl:             uint32(ret.MaxRetention()),
		// we set LastWrite here to make sure a new Chunk doesn't get immediately
		// garbage collected right after creating it, before we can push to it.
		lastWrite: uint32(time.Now().Unix()),
	}
	if ingestFrom > 0 {
		// we only want to ingest data that will go into chunks with a t0 >= 'ingestFrom'.
		m.ingestFromT0 = AggBoundary(uint32(ingestFrom), ret.ChunkSpan)
	}
	if reorderWindow != 0 {
		m.rob = NewReorderBuffer(reorderWindow, interval, reorderAllowUpdate)
	}

	origSplits := strings.Split(retentions.Orig, ":")
	for i, ret := range retentions.Rets[1:] {
		retOrig := origSplits[i+1]
		m.aggregators = append(m.aggregators, NewAggregator(store, cachePusher, key, retOrig, ret, *agg, dropFirstChunk, ingestFrom))
	}

	return &m
}

// Sync the saved state of a chunk by its T0.
func (a *AggMetric) SyncChunkSaveState(ts uint32, sendPersist bool) ChunkSaveCallback {
	return func() {
		a.Lock()
		if ts > a.lastSaveFinish {
			a.lastSaveFinish = ts
		}
		if ts > a.lastSaveStart {
			a.lastSaveStart = ts
		}
		a.Unlock()
		log.Debugf("AM: metric %s at chunk T0=%d has been saved.", a.key, ts)
		if sendPersist {
			SendPersistMessage(a.key.String(), ts)
		}
	}
}

// Sync the saved state of a chunk by its T0.
func (a *AggMetric) SyncAggregatedChunkSaveState(ts uint32, consolidator consolidation.Consolidator, aggSpan uint32) {
	// no lock needed cause aggregators don't change at runtime
	for _, a := range a.aggregators {
		if a.span == aggSpan {
			switch consolidator {
			case consolidation.None:
				panic("cannot get an archive for no consolidation")
			case consolidation.Avg:
				panic("avg consolidator has no matching Archive(). you need sum and cnt")
			case consolidation.Cnt:
				if a.cntMetric != nil {
					a.cntMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Min:
				if a.minMetric != nil {
					a.minMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Max:
				if a.maxMetric != nil {
					a.maxMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Sum:
				if a.sumMetric != nil {
					a.sumMetric.SyncChunkSaveState(ts, false)()
				}
				return
			case consolidation.Lst:
				if a.lstMetric != nil {
					a.lstMetric.SyncChunkSaveState(ts, false)()
				}
				return
			default:
				panic(fmt.Sprintf("internal error: no such consolidator %q with span %d", consolidator, aggSpan))
			}
		}
	}
}

func (a *AggMetric) GetAggregated(consolidator consolidation.Consolidator, aggSpan, from, to uint32) (Result, error) {
	// no lock needed cause aggregators don't change at runtime
	for _, a := range a.aggregators {
		if a.span == aggSpan {
			var agg *AggMetric
			switch consolidator {
			case consolidation.None:
				err := errors.New("internal error: AggMetric.GetAggregated(): cannot get an archive for no consolidation")
				log.Errorf("AM: %s", err.Error())
				badConsolidator.Inc()
				return Result{}, err
			case consolidation.Avg:
				err := errors.New("internal error: AggMetric.GetAggregated(): avg consolidator has no matching Archive(). you need sum and cnt")
				log.Errorf("AM: %s", err.Error())
				badConsolidator.Inc()
				return Result{}, err
			case consolidation.Cnt:
				agg = a.cntMetric
			case consolidation.Lst:
				agg = a.lstMetric
			case consolidation.Min:
				agg = a.minMetric
			case consolidation.Max:
				agg = a.maxMetric
			case consolidation.Sum:
				agg = a.sumMetric
			default:
				err := fmt.Errorf("internal error: AggMetric.GetAggregated(): unknown consolidator %q", consolidator)
				log.Errorf("AM: %s", err.Error())
				badConsolidator.Inc()
				return Result{}, err
			}
			if agg == nil {
				return Result{}, fmt.Errorf("Consolidator %q not configured", consolidator)
			}
			return agg.Get(from, to)
		}
	}
	err := fmt.Errorf("internal error: AggMetric.GetAggregated(): unknown aggSpan %d", aggSpan)
	log.Errorf("AM: %s", err.Error())
	badAggSpan.Inc()
	return Result{}, err
}

// Get all data between the requested time ranges. From is inclusive, to is exclusive. from <= x < to
// more data then what's requested may be included
// specifically, returns:
// * points from the ROB (if enabled)
// * iters from matching chunks
// * oldest point we have, so that if your query needs data before it, the caller knows when to query the store
func (a *AggMetric) Get(from, to uint32) (Result, error) {
	pre := time.Now()
	log.Debugf("AM: %s Get(): %d - %d (%s - %s) span:%ds", a.key, from, to, TS(from), TS(to), to-from-1)
	if from >= to {
		return Result{}, ErrInvalidRange
	}
	a.RLock()
	defer a.RUnlock()

	result := Result{
		Oldest: math.MaxInt32,
	}

	if a.rob != nil {
		result.Points = a.rob.Get()
		if len(result.Points) > 0 {
			result.Oldest = result.Points[0].Ts
			if result.Oldest <= from {
				return result, nil
			}
		}
	}

	if len(a.chunks) == 0 {
		// we dont have any data yet.
		log.Debugf("AM: %s Get(): no data for requested range.", a.key)
		return result, nil
	}

	newestChunk := a.chunks[a.currentChunkPos]

	if from >= newestChunk.Series.T0+a.chunkSpan {
		// request falls entirely ahead of the data we have
		// this can happen in a few cases:
		// * queries for the most recent data, but our ingestion has fallen behind.
		//   we don't want such a degradation to cause a storm of cassandra queries
		//   we should just return an oldest value that is <= from so we don't hit cassandra
		//   but it doesn't have to be the real oldest value, so do whatever is cheap.
		// * data got once written by another node, but has been re-sending (perhaps fixed)
		//   to this node, but not yet up to the point it was previously sent. so we're
		//   only aware of older data and not the newer data in cassandra. this is unlikely
		//   and it's better to not serve this scenario well in favor of the above case.
		//   seems like a fair tradeoff anyway that you have to refill all the way first.
		log.Debugf("AM: %s Get(): no data for requested range.", a.key)
		result.Oldest = from
		return result, nil
	}

	// get the oldest chunk we have.
	// eg if we have 5 chunks, N is the current chunk and n-4 is the oldest chunk.
	// -----------------------------
	// | n-4 | n-3 | n-2 | n-1 | n |  CurrentChunkPos = 4
	// -----------------------------
	// -----------------------------
	// | n | n-4 | n-3 | n-2 | n-1 |  CurrentChunkPos = 0
	// -----------------------------
	// -----------------------------
	// | n-2 | n-1 | n | n-4 | n-3 |  CurrentChunkPos = 2
	// -----------------------------
	oldestPos := a.currentChunkPos + 1
	if oldestPos >= len(a.chunks) {
		oldestPos = 0
	}

	oldestChunk := a.chunks[oldestPos]

	if to <= oldestChunk.Series.T0 {
		// the requested time range ends before any data we have.
		log.Debugf("AM: %s Get(): no data for requested range", a.key)
		if oldestChunk.First {
			result.Oldest = a.firstTs
		} else {
			result.Oldest = oldestChunk.Series.T0
		}
		return result, nil
	}

	// Find the oldest Chunk that the "from" ts falls in.  If from extends before the oldest
	// chunk, then we just use the oldest chunk.
	for from >= oldestChunk.Series.T0+a.chunkSpan {
		oldestPos++
		if oldestPos >= len(a.chunks) {
			oldestPos = 0
		}
		oldestChunk = a.chunks[oldestPos]
	}

	// find the newest Chunk that "to" falls in.  If "to" extends to after the newest data
	// then just return the newest chunk.
	// some examples to clarify this more. assume newestChunk.T0 is at 120, then
	// for a to of 121 -> data up to (incl) 120 -> stay at this chunk, it has a point we need
	// for a to of 120 -> data up to (incl) 119 -> use older chunk
	// for a to of 119 -> data up to (incl) 118 -> use older chunk
	newestPos := a.currentChunkPos
	for to <= newestChunk.Series.T0 {
		newestPos--
		if newestPos < 0 {
			newestPos += len(a.chunks)
		}
		newestChunk = a.chunks[newestPos]
	}

	// now just start at oldestPos and move through the Chunks circular Buffer to newestPos
	for {
		c := a.chunks[oldestPos]
		result.Iters = append(result.Iters, c.Series.Iter())

		if oldestPos == newestPos {
			break
		}

		oldestPos++
		if oldestPos >= len(a.chunks) {
			oldestPos = 0
		}
	}

	if oldestChunk.First {
		result.Oldest = a.firstTs
	} else {
		result.Oldest = oldestChunk.Series.T0
	}

	memToIterDuration.Value(time.Now().Sub(pre))
	return result, nil
}

// caller must hold lock
func (a *AggMetric) addAggregators(ts uint32, val float64) {
	for _, agg := range a.aggregators {
		if log.IsLevelEnabled(log.DebugLevel) {
			log.Debugf("AM: %s pushing %d,%f to aggregator %d", a.key, ts, val, agg.span)
		}
		agg.Add(ts, val)
	}
}

// pushToCache adds the chunk into the cache if it is hot
// caller must hold lock
func (a *AggMetric) pushToCache(c *chunk.Chunk) {
	if a.cachePusher == nil {
		return
	}
	// push into cache
	intervalHint := a.key.Archive.Span()

	itergen, err := chunk.NewIterGen(c.Series.T0, intervalHint, c.Encode(a.chunkSpan))
	if err != nil {
		log.Errorf("AM: %s failed to generate IterGen. this should never happen: %s", a.key, err)
	}
	go a.cachePusher.AddIfHot(a.key, 0, itergen)
}

// write a chunk to persistent storage.
// never persist a chunk that may receive further updates!
// (because the stores will read out chunk data on the unlocked chunk)
// caller must hold lock.
func (a *AggMetric) persist(pos int) {
	chunk := a.chunks[pos]
	pre := time.Now()

	if a.lastSaveStart >= chunk.Series.T0 {
		// this can happen if
		// a) there are 2 primary MT nodes both saving chunks to Cassandra
		// b) a primary failed and this node was promoted to be primary but metric consuming is lagging.
		// c) chunk was persisted by GC (stale) and then new data triggered another persist call
		// d) dropFirstChunk is enabled and this is the first chunk
		log.Debugf("AM: persist(): duplicate persist call for chunk.")
		return
	}

	// create an array of chunks that need to be sent to the writeQueue.
	pending := make([]*ChunkWriteRequest, 1)
	// add the current chunk to the list of chunks to send to the writeQueue
	cwr := NewChunkWriteRequest(
		a.SyncChunkSaveState(chunk.Series.T0, true),
		a.key,
		a.ttl,
		chunk.Series.T0,
		chunk.Encode(a.chunkSpan),
		time.Now(),
	)
	pending[0] = &cwr

	// if we recently became the primary, there may be older chunks
	// that the old primary did not save.  We should check for those
	// and save them.
	previousPos := pos - 1
	if previousPos < 0 {
		previousPos += len(a.chunks)
	}
	previousChunk := a.chunks[previousPos]
	for (previousChunk.Series.T0 < chunk.Series.T0) && (a.lastSaveStart < previousChunk.Series.T0) {
		log.Debugf("AM: persist(): old chunk needs saving. Adding %s:%d to writeQueue", a.key, previousChunk.Series.T0)
		cwr := NewChunkWriteRequest(
			a.SyncChunkSaveState(previousChunk.Series.T0, true),
			a.key,
			a.ttl,
			previousChunk.Series.T0,
			previousChunk.Encode(a.chunkSpan),
			time.Now(),
		)
		pending = append(pending, &cwr)
		previousPos--
		if previousPos < 0 {
			previousPos += len(a.chunks)
		}
		previousChunk = a.chunks[previousPos]
	}

	// Every chunk with a T0 <= this chunks' T0 is now either saved, or in the writeQueue.
	a.lastSaveStart = chunk.Series.T0

	log.Debugf("AM: persist(): sending %d chunks to write queue", len(pending))

	pendingChunk := len(pending) - 1

	// if the store blocks,
	// the calling function will block waiting for persist() to complete.
	// This is intended to put backpressure on our message handlers so
	// that they stop consuming messages, leaving them to buffer at
	// the message bus. The "pending" array of chunks are processed
	// last-to-first ensuring that older data is added to the store
	// before newer data.
	for pendingChunk >= 0 {
		log.Debugf("AM: persist(): sealing chunk %d/%d (%s:%d) and adding to write queue.", pendingChunk, len(pending), a.key, chunk.Series.T0)
		a.store.Add(pending[pendingChunk])
		pendingChunk--
	}
	persistDuration.Value(time.Now().Sub(pre))
	return
}

// don't ever call with a ts of 0, cause we use 0 to mean not initialized!
func (a *AggMetric) Add(ts uint32, val float64) {
	if ts < a.ingestFromT0 {
		// TODO: add metric to keep track of the # of points discarded
		if log.IsLevelEnabled(log.DebugLevel) {
			log.Debugf("AM: discarding metric <%d,%f>: does not belong to a chunk starting after ingest-from. First chunk considered starts at %d", ts, val, a.ingestFromT0)
		}
		// even if a point is too old for our raw data, it may not be too old for aggregated data
		// for example let's say a chunk starts at t0=3600 but we have 300-secondly aggregates
		// that mean the aggregators need data from 3301 and onwards, because we aggregate 3301-3600 into a point with ts=3600
		a.addAggregators(ts, val)
		return
	}

	// need to check if ts > futureTolerance to prevent that we reject a datapoint
	// because the ts value has wrapped around the uint32 boundary
	if ts > a.futureTolerance && int64(ts-a.futureTolerance) > time.Now().Unix() {
		// we increase this counter in any case, even if the enforcement of the future tolerance
		// is disabled. this is useful to predict whether data points would get rejected once
		// the feature gets enabled.
		discardedSampleTooFarAhead.Inc()

{'fraction_non_alphanumeric': 0.11604884968852155, 'fraction_numerical': 0.011657312033553321, 'mean_word_length': 3.13504208110176, 'pattern_counts': {'":': 2, '<': 31, '<?xml version=': 0, '>': 51, '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': '2803337', 'n_tokens_mistral': 11480, 'n_tokens_neox': 10308, 'n_words': 3738}