From 22d72514ff491bdcfd35d5977abe5d26139c9a5a Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Sun, 21 Jun 2020 15:08:08 +0000 Subject: [PATCH] WIP apply timestamp from index when fetching reports --- app/multitenant/aws_collector.go | 16 ++++++++-------- app/multitenant/memcache_client.go | 19 +++++++++++-------- app/multitenant/s3_client.go | 17 +++++++++++------ 3 files changed, 30 insertions(+), 22 deletions(-) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index dd630ec68..22089cd22 100644 --- a/app/multitenant/aws_collector.go +++ b/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) } diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 86a35c8c5..92f1fca5e 100644 --- a/app/multitenant/memcache_client.go +++ b/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) } diff --git a/app/multitenant/s3_client.go b/app/multitenant/s3_client.go index 250c7136e..09b3f6249 100644 --- a/app/multitenant/s3_client.go +++ b/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) {