WIP apply timestamp from index when fetching reports

app/multitenant/aws_collector.go

@@ -127,7 +127,7 @@ type AWSCollector interface {
 
 // ReportStore is a thing that we can get reports from.
 type ReportStore interface {
-	FetchReports(context.Context, []string) (map[string]report.Report, []string, error)
+	FetchReports(context.Context, []keyInfo) (map[string]report.Report, []keyInfo, error)
 }
 
 // AWSCollectorConfig has everything we need to make an AWS collector.
@@ -411,7 +411,7 @@ func (c *awsCollector) getReportKeys(ctx context.Context, userid string, start,
 	return reportKeys, nil
 }
 
-func (c *awsCollector) getReports(ctx context.Context, userid string, reportKeys []string) ([]report.Report, error) {
+func (c *awsCollector) getReports(ctx context.Context, userid string, reportKeys []keyInfo) ([]report.Report, error) {
 	missing := reportKeys
 
 	stores := []ReportStore{c.inProcess}
@@ -508,7 +508,7 @@ func (c *awsCollector) Report(ctx context.Context, timestamp time.Time) (report.
 // Fetch a merged report either from cache or from store which we put in cache
 func (c *awsCollector) reportForQuantum(ctx context.Context, userid string, reportKeys []keyInfo, start int64) (report.Report, error) {
 	key := fmt.Sprintf("%s:%d", userid, start)
-	cached, _, err := c.inProcess.FetchReports(ctx, []string{key})
+	cached, _, err := c.inProcess.FetchReports(ctx, []keyInfo{{key: key, ts: start}})
 	if len(cached) == 1 {
 		return cached[key], nil
 	}
@@ -523,10 +523,10 @@ func (c *awsCollector) reportForQuantum(ctx context.Context, userid string, repo
 
 // Find the keys relating to this time period then fetch from memcached and/or S3
 func (c *awsCollector) reportsForKeysInRange(ctx context.Context, userid string, reportKeys []keyInfo, start, end int64) ([]report.Report, error) {
-	var keys []string
+	var keys []keyInfo
 	for _, k := range reportKeys {
 		if k.ts >= start && k.ts < end {
-			keys = append(keys, k.key)
+			keys = append(keys, k)
 		}
 	}
 	if span := opentracing.SpanFromContext(ctx); span != nil {
@@ -793,13 +793,13 @@ func newInProcessStore(size int, expiration time.Duration) inProcessStore {
 }
 
 // FetchReports retrieves the given reports from the store.
-func (c inProcessStore) FetchReports(_ context.Context, keys []string) (map[string]report.Report, []string, error) {
+func (c inProcessStore) FetchReports(_ context.Context, keys []keyInfo) (map[string]report.Report, []keyInfo, error) {
 	found := map[string]report.Report{}
-	missing := []string{}
+	missing := []keyInfo{}
 	for _, key := range keys {
 		rpt, err := c.cache.Get(key)
 		if err == nil {
-			found[key] = rpt.(report.Report)
+			found[key.key] = rpt.(report.Report)
 		} else {
 			missing = append(missing, key)
 		}

app/multitenant/memcache_client.go

@@ -155,14 +155,18 @@ func memcacheStatusCode(err error) string {
 }
 
 // FetchReports gets reports from memcache.
-func (c *MemcacheClient) FetchReports(ctx context.Context, keys []string) (map[string]report.Report, []string, error) {
+func (c *MemcacheClient) FetchReports(ctx context.Context, keys []keyInfo) (map[string]report.Report, []keyInfo, error) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, "Memcache.FetchReports")
 	defer span.Finish()
 	defer memcacheRequests.Add(float64(len(keys)))
 	var found map[string]*memcache.Item
+	mkeys := make([]string, len(keys))
+	for i, k := range keys {
+		mkeys[i] = k.key
+	}
 	err := instrument.TimeRequestHistogramStatus(ctx, "Memcache.GetMulti", memcacheRequestDuration, memcacheStatusCode, func(_ context.Context) error {
 		var err error
-		found, err = c.client.GetMulti(keys)
+		found, err = c.client.GetMulti(mkeys)
 		return err
 	})
 	span.LogFields(otlog.Int("keys", len(keys)), otlog.Int("hits", len(found)))
@@ -172,18 +176,18 @@ func (c *MemcacheClient) FetchReports(ctx context.Context, keys []string) (map[s
 
 	// Decode all the reports in parallel.
 	type result struct {
-		key    string
+		key    keyInfo
 		report *report.Report
 	}
 	ch := make(chan result, len(keys))
-	var missing []string
+	var missing []keyInfo
 	for _, key := range keys {
-		item, ok := found[key]
+		item, ok := found[key.key]
 		if !ok {
 			missing = append(missing, key)
 			continue
 		}
-		go func(key string) {
+		go func(key keyInfo) {
 			rep, err := report.MakeFromBinary(ctx, bytes.NewBuffer(item.Value), true, true)
 			if err != nil {
 				log.Warningf("Corrupt report in memcache %v: %v", key, err)
@@ -201,12 +205,11 @@ func (c *MemcacheClient) FetchReports(ctx context.Context, keys []string) (map[s
 		if r.report == nil {
 			missing = append(missing, r.key)
 		} else {
-			reports[r.key] = *r.report
+			reports[r.key.key] = *r.report
 		}
 	}
 
 	if len(missing) > 0 {
-		sort.Strings(missing)
 		log.Warningf("Missing %d reports from memcache: %v", len(missing), missing)
 	}
 

app/multitenant/s3_client.go

@@ -3,6 +3,7 @@ package multitenant
 import (
 	"bytes"
 	"sync"
+	"time"
 
 	"context"
 	"github.com/aws/aws-sdk-go/aws"
@@ -45,7 +46,7 @@ func NewS3Client(config *aws.Config, bucketName string) S3Store {
 }
 
 // FetchReports fetches multiple reports in parallel from S3.
-func (store *S3Store) FetchReports(ctx context.Context, keys []string) (map[string]report.Report, []string, error) {
+func (store *S3Store) FetchReports(ctx context.Context, keys []keyInfo) (map[string]report.Report, []keyInfo, error) {
 	type result struct {
 		key    string
 		report *report.Report
@@ -55,9 +56,13 @@ func (store *S3Store) FetchReports(ctx context.Context, keys []string) (map[stri
 	ch := make(chan result, len(keys))
 
 	for _, key := range keys {
-		go func(key string) {
-			r := result{key: key}
-			r.report, r.err = store.fetchReport(ctx, key)
+		go func(key keyInfo) {
+			r := result{key: key.key}
+			r.report, r.err = store.fetchReport(ctx, key.key)
+			// If the report doesn't have a timestamp, add the one from the index
+			if r.report.TS.IsZero() {
+				r.report.TS = time.Unix(key.ts/1000, (key.ts%1000)*1000000)
+			}
 			ch <- r
 		}(key)
 	}
@@ -66,11 +71,11 @@ func (store *S3Store) FetchReports(ctx context.Context, keys []string) (map[stri
 	for range keys {
 		r := <-ch
 		if r.err != nil {
-			return nil, []string{}, r.err
+			return nil, []keyInfo{}, r.err
 		}
 		reports[r.key] = *r.report
 	}
-	return reports, []string{}, nil
+	return reports, []keyInfo{}, nil
 }
 
 func (store *S3Store) fetchReport(ctx context.Context, key string) (*report.Report, error) {