AggMetrics.ingestAfter: simplify and generalise by using a map orgId -> timestamp

Author: fkaleo

api/ccache_test.go

@@ -33,7 +33,7 @@ func newSrv(delSeries, delArchives int) (*Server, *cache.MockCache) {
 	mockCache := cache.NewMockCache()
 	mockCache.DelMetricSeries = delSeries
 	mockCache.DelMetricArchives = delArchives
-	metrics := mdata.NewAggMetrics(store, mockCache, false, 0, 0, 0, 0, 0)
+	metrics := mdata.NewAggMetrics(store, mockCache, false, nil, 0, 0, 0)
 	srv.BindMemoryStore(metrics)
 	srv.BindCache(mockCache)
 

api/dataprocessor_test.go

@@ -346,7 +346,7 @@ func TestGetSeriesFixed(t *testing.T) {
 	mdata.SetSingleAgg(conf.Avg, conf.Min, conf.Max)
 	mdata.SetSingleSchema(conf.NewRetentionMT(10, 100, 600, 10, 0))
 
-	metrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, 0, 0, 0, 0, 0)
+	metrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, nil, 0, 0, 0)
 	srv, _ := NewServer()
 	srv.BindBackendStore(store)
 	srv.BindMemoryStore(metrics)
@@ -501,7 +501,7 @@ func TestGetSeriesCachedStore(t *testing.T) {
 	store := mdata.NewMockStore()
 	srv.BindBackendStore(store)
 
-	metrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, 0, 0, 0, 0, 0)
+	metrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, nil, 0, 0, 0)
 	srv.BindMemoryStore(metrics)
 	metric := test.GetAMKey(1)
 
@@ -687,7 +687,7 @@ func TestGetSeriesAggMetrics(t *testing.T) {
 	cluster.Init("default", "test", time.Now(), "http", 6060)
 	store := mdata.NewMockStore()
 
-	metrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, 0, 0, 0, 0, 0)
+	metrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, nil, 0, 0, 0)
 	srv, _ := NewServer()
 	srv.BindBackendStore(store)
 	srv.BindMemoryStore(metrics)

cmd/metrictank/metrictank.go

@@ -302,8 +302,10 @@ func main() {
 	if ingestAfterTimestamp > 0 {
 		log.Infof("Will only ingest data points for org id %d belonging to chunks starting after %s", ingestAfterOrgID, time.Unix(ingestAfterTimestamp, 0))
 	}
+	ingestAfter := make(map[uint32]int64)
+	ingestAfter[ingestAfterOrgID] = ingestAfterTimestamp
 	if inputEnabled {
-		metrics = mdata.NewAggMetrics(store, ccache, *dropFirstChunk, ingestAfterOrgID, ingestAfterTimestamp, chunkMaxStale, metricMaxStale, gcInterval)
+		metrics = mdata.NewAggMetrics(store, ccache, *dropFirstChunk, ingestAfter, chunkMaxStale, metricMaxStale, gcInterval)
 	}
 
 	/***********************************

input/input_test.go

@@ -195,7 +195,7 @@ func getDefaultHandler(t *testing.T) (DefaultHandler, idx.MetricIndex, func()) {
 	}
 
 	mdata.Schemas = conf.NewSchemas(nil)
-	metrics := mdata.NewAggMetrics(nil, nil, false, 0, 0, 3600, 7200, 3600)
+	metrics := mdata.NewAggMetrics(nil, nil, false, nil, 3600, 7200, 3600)
 	return NewDefaultHandler(metrics, index, "test"), index, reset
 }
 
@@ -207,7 +207,7 @@ func BenchmarkProcessMetricDataUniqueMetrics(b *testing.B) {
 	mdata.SetSingleSchema(conf.NewRetentionMT(10, 10000, 600, 10, 0))
 	mdata.SetSingleAgg(conf.Avg, conf.Min, conf.Max)
 
-	aggmetrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, 0, 0, 800, 8000, 0)
+	aggmetrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, nil, 800, 8000, 0)
 	metricIndex := memory.New()
 	metricIndex.Init()
 	defer metricIndex.Stop()
@@ -247,7 +247,7 @@ func BenchmarkProcessMetricDataSameMetric(b *testing.B) {
 	mdata.SetSingleSchema(conf.NewRetentionMT(10, 10000, 600, 10, 0))
 	mdata.SetSingleAgg(conf.Avg, conf.Min, conf.Max)
 
-	aggmetrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, 0, 0, 800, 8000, 0)
+	aggmetrics := mdata.NewAggMetrics(store, &cache.MockCache{}, false, nil, 800, 8000, 0)
 	metricIndex := memory.New()
 	metricIndex.Init()
 	defer metricIndex.Stop()

mdata/aggmetrics.go

@@ -18,10 +18,7 @@ type AggMetrics struct {
 	store          Store
 	cachePusher    cache.CachePusher
 	dropFirstChunk bool
-	ingestAfter    struct {
-		orgID     uint32
-		timestamp int64
-	}
+	ingestAfter    map[uint32]int64
 	chunkMaxStale  uint32
 	metricMaxStale uint32
 	gcInterval     time.Duration
@@ -30,18 +27,12 @@ type AggMetrics struct {
 	Metrics map[uint32]map[schema.Key]*AggMetric
 }
 
-func NewAggMetrics(store Store, cachePusher cache.CachePusher, dropFirstChunk bool, ingestAfterOrgID uint32, ingestAfterTimestamp int64, chunkMaxStale, metricMaxStale uint32, gcInterval time.Duration) *AggMetrics {
+func NewAggMetrics(store Store, cachePusher cache.CachePusher, dropFirstChunk bool, ingestAfter map[uint32]int64, chunkMaxStale, metricMaxStale uint32, gcInterval time.Duration) *AggMetrics {
 	ms := AggMetrics{
 		store:          store,
 		cachePusher:    cachePusher,
 		dropFirstChunk: dropFirstChunk,
-		ingestAfter: struct {
-			orgID     uint32
-			timestamp int64
-		}{
-			orgID:     ingestAfterOrgID,
-			timestamp: ingestAfterTimestamp,
-		},
+		ingestAfter:    ingestAfter,
 		Metrics:        make(map[uint32]map[schema.Key]*AggMetric),
 		chunkMaxStale:  chunkMaxStale,
 		metricMaxStale: metricMaxStale,
@@ -173,10 +164,7 @@ func (ms *AggMetrics) GetOrCreate(key schema.MKey, schemaId, aggId uint16, inter
 		ms.Unlock()
 		return m
 	}
-	ingestAfter := int64(0)
-	if key.Org == ms.ingestAfter.orgID {
-		ingestAfter = ms.ingestAfter.timestamp
-	}
+	ingestAfter := ms.ingestAfter[key.Org]
 	m = NewAggMetric(ms.store, ms.cachePusher, k, confSchema.Retentions, confSchema.ReorderWindow, interval, &agg, ms.dropFirstChunk, ingestAfter)
 	ms.Metrics[key.Org][key.Key] = m
 	active := len(ms.Metrics[key.Org])