<commit_before>package mdata

import (
	"strconv"
	"sync"
	"time"

	"github.com/grafana/metrictank/mdata/cache"
	"github.com/raintank/schema"
	log "github.com/sirupsen/logrus"
)

// AggMetrics is an in-memory store of AggMetric objects
// note: they are keyed by MKey here because each
// AggMetric manages access to, and references of,
// their rollup archives themselves
type AggMetrics struct {
	store          Store
	cachePusher    cache.CachePusher
	dropFirstChunk bool
	sync.RWMutex
	Metrics        map[uint32]map[schema.Key]*AggMetric
	chunkMaxStale  uint32
	metricMaxStale uint32
	gcInterval     time.Duration
}

func NewAggMetrics(store Store, cachePusher cache.CachePusher, dropFirstChunk bool, chunkMaxStale, metricMaxStale uint32, gcInterval time.Duration) *AggMetrics {
	ms := AggMetrics{
		store:          store,
		cachePusher:    cachePusher,
		dropFirstChunk: dropFirstChunk,
		Metrics:        make(map[uint32]map[schema.Key]*AggMetric),
		chunkMaxStale:  chunkMaxStale,
		metricMaxStale: metricMaxStale,
		gcInterval:     gcInterval,
	}

	// gcInterval = 0 can be useful in tests
	if gcInterval > 0 {
		go ms.GC()
	}
	return &ms
}

// periodically scan chunks and close any that have not received data in a while
func (ms *AggMetrics) GC() {
	for {
		unix := time.Duration(time.Now().UnixNano())
		diff := ms.gcInterval - (unix % ms.gcInterval)
		time.Sleep(diff + time.Minute)
		log.Info("checking for stale chunks that need persisting.")
		now := uint32(time.Now().Unix())
		chunkMinTs := now - uint32(ms.chunkMaxStale)
		metricMinTs := now - uint32(ms.metricMaxStale)

		// as this is the only goroutine that can delete from ms.Metrics
		// we only need to lock long enough to get the list of orgs, then for each org
		// get the list of active metrics.
		// It doesn't matter if new orgs or metrics are added while we iterate these lists.
		ms.RLock()
		orgs := make([]uint32, 0, len(ms.Metrics))
		for o := range ms.Metrics {
			orgs = append(orgs, o)
		}
		ms.RUnlock()
		for _, org := range orgs {
			orgActiveMetrics := promActiveMetrics.WithLabelValues(strconv.Itoa(int(org)))
			keys := make([]schema.Key, 0, len(ms.Metrics[org]))
			ms.RLock()
			for k := range ms.Metrics[org] {
				keys = append(keys, k)
			}
			ms.RUnlock()
			for _, key := range keys {<commit_msg>Fix map read / write panic in aggmetrics<commit_after>package mdata

import (
	"strconv"
	"sync"
	"time"

	"github.com/grafana/metrictank/mdata/cache"
	"github.com/raintank/schema"
	log "github.com/sirupsen/logrus"
)

// AggMetrics is an in-memory store of AggMetric objects
// note: they are keyed by MKey here because each
// AggMetric manages access to, and references of,
// their rollup archives themselves
type AggMetrics struct {
	store          Store
	cachePusher    cache.CachePusher
	dropFirstChunk bool
	sync.RWMutex
	Metrics        map[uint32]map[schema.Key]*AggMetric
	chunkMaxStale  uint32
	metricMaxStale uint32
	gcInterval     time.Duration
}

func NewAggMetrics(store Store, cachePusher cache.CachePusher, dropFirstChunk bool, chunkMaxStale, metricMaxStale uint32, gcInterval time.Duration) *AggMetrics {
	ms := AggMetrics{
		store:          store,
		cachePusher:    cachePusher,
		dropFirstChunk: dropFirstChunk,
		Metrics:        make(map[uint32]map[schema.Key]*AggMetric),
		chunkMaxStale:  chunkMaxStale,
		metricMaxStale: metricMaxStale,
		gcInterval:     gcInterval,
	}

	// gcInterval = 0 can be useful in tests
	if gcInterval > 0 {
		go ms.GC()
	}
	return &ms
}

// periodically scan chunks and close any that have not received data in a while
func (ms *AggMetrics) GC() {
	for {
		unix := time.Duration(time.Now().UnixNano())
		diff := ms.gcInterval - (unix % ms.gcInterval)
		time.Sleep(diff + time.Minute)
		log.Info("checking for stale chunks that need persisting.")
		now := uint32(time.Now().Unix())
		chunkMinTs := now - uint32(ms.chunkMaxStale)
		metricMinTs := now - uint32(ms.metricMaxStale)

		// as this is the only goroutine that can delete from ms.Metrics
		// we only need to lock long enough to get the list of orgs, then for each org
		// get the list of active metrics.
		// It doesn't matter if new orgs or metrics are added while we iterate these lists.
		ms.RLock()
		orgs := make([]uint32, 0, len(ms.Metrics))
		for o := range ms.Metrics {
			orgs = append(orgs, o)
		}
		ms.RUnlock()
		for _, org := range orgs {
			orgActiveMetrics := promActiveMetrics.WithLabelValues(strconv.Itoa(int(org)))
			// need to acquire lock here, otherwise can run into panic because
			// GetOrCreate might be writing to ms.Metrics
			ms.RLock()
			keys := make([]schema.Key, 0, len(ms.Metrics[org]))
			for k := range ms.Metrics[org] {
				keys = append(keys, k)
			}
			ms.RUnlock()
			for _, key := range keys {
{'fraction_non_alphanumeric': 0.10611814345991562, 'fraction_numerical': 0.009282700421940928, 'mean_word_length': 3.563041385948027, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '2442283', 'n_tokens_mistral': 1734, 'n_tokens_neox': 1580, 'n_words': 475}