From e4c75e6223647308d3c8f5d99256e1dad13c39b4 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Thu, 14 Jul 2016 18:06:10 +0100 Subject: [PATCH 1/9] Test for marshalling --- report/marshal_test.go | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 report/marshal_test.go diff --git a/report/marshal_test.go b/report/marshal_test.go new file mode 100644 index 000000000..8d9865862 --- /dev/null +++ b/report/marshal_test.go @@ -0,0 +1,22 @@ +package report_test + +import ( + "bytes" + "reflect" + "testing" + + "github.com/weaveworks/scope/report" +) + +func TestRoundtrip(t *testing.T) { + var buf bytes.Buffer + r1 := report.MakeReport() + r1.WriteBinary(&buf) + r2, err := report.MakeFromBinary(&buf) + if err != nil { + t.Error(err) + } + if !reflect.DeepEqual(r1, *r2) { + t.Errorf("%v != %v", r1, *r2) + } +} From 2bfd6d7eb77d14988bc8880f8f8c7ede48f84f9e Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Thu, 14 Jul 2016 18:35:42 +0100 Subject: [PATCH 2/9] Parametrize compression level --- app/multitenant/aws_collector.go | 3 ++- probe/appclient/report_publisher.go | 3 ++- report/marshal.go | 4 ++-- report/marshal_test.go | 29 ++++++++++++++++++++++++++++- 4 files changed, 34 insertions(+), 5 deletions(-) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index 2365f8c55..70450c675 100644 --- a/app/multitenant/aws_collector.go +++ b/app/multitenant/aws_collector.go @@ -2,6 +2,7 @@ package multitenant import ( "bytes" + "compress/gzip" "crypto/md5" "fmt" "io" @@ -346,7 +347,7 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { // first, encode the report into a buffer and record its size var buf bytes.Buffer - rep.WriteBinary(&buf) + rep.WriteBinary(&buf, gzip.BestCompression) reportSizeHistogram.Observe(float64(buf.Len())) // second, put the report on s3 diff --git a/probe/appclient/report_publisher.go b/probe/appclient/report_publisher.go index 890d22c95..8d426d853 100644 --- a/probe/appclient/report_publisher.go +++ b/probe/appclient/report_publisher.go @@ -2,6 +2,7 @@ package appclient import ( "bytes" + "compress/gzip" "github.com/weaveworks/scope/report" ) @@ -28,6 +29,6 @@ func (p *ReportPublisher) Publish(r report.Report) error { }) } buf := &bytes.Buffer{} - r.WriteBinary(buf) + r.WriteBinary(buf, gzip.BestCompression) return p.publisher.Publish(buf) } diff --git a/report/marshal.go b/report/marshal.go index 2cafd0568..e362102f4 100644 --- a/report/marshal.go +++ b/report/marshal.go @@ -9,8 +9,8 @@ import ( ) // WriteBinary writes a Report as a gzipped msgpack. -func (rep Report) WriteBinary(w io.Writer) error { - gzwriter, err := gzip.NewWriterLevel(w, gzip.BestCompression) +func (rep Report) WriteBinary(w io.Writer, compressionLevel int) error { + gzwriter, err := gzip.NewWriterLevel(w, compressionLevel) if err != nil { return err } diff --git a/report/marshal_test.go b/report/marshal_test.go index 8d9865862..f779c4743 100644 --- a/report/marshal_test.go +++ b/report/marshal_test.go @@ -2,6 +2,7 @@ package report_test import ( "bytes" + "compress/gzip" "reflect" "testing" @@ -11,7 +12,7 @@ import ( func TestRoundtrip(t *testing.T) { var buf bytes.Buffer r1 := report.MakeReport() - r1.WriteBinary(&buf) + r1.WriteBinary(&buf, gzip.BestCompression) r2, err := report.MakeFromBinary(&buf) if err != nil { t.Error(err) @@ -20,3 +21,29 @@ func TestRoundtrip(t *testing.T) { t.Errorf("%v != %v", r1, *r2) } } + +func TestRoundtripNoCompression(t *testing.T) { + // Make sure that we can use our standard routines for decompressing + // something with '0' level compression. + var buf bytes.Buffer + r1 := report.MakeReport() + r1.WriteBinary(&buf, 0) + r2, err := report.MakeFromBinary(&buf) + if err != nil { + t.Error(err) + } + if !reflect.DeepEqual(r1, *r2) { + t.Errorf("%v != %v", r1, *r2) + } +} + +func TestMoreCompressionMeansSmaller(t *testing.T) { + // Make sure that 0 level compression actually does compress less. + var buf1, buf2 bytes.Buffer + r := report.MakeReport() + r.WriteBinary(&buf1, gzip.BestCompression) + r.WriteBinary(&buf2, 0) + if buf1.Len() >= buf2.Len() { + t.Errorf("Compression doesn't change size: %v >= %v", buf1.Len(), buf2.Len()) + } +} From 60e14c1dc2df0abb4c6b9553e409c753e83873c4 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Thu, 14 Jul 2016 18:46:50 +0100 Subject: [PATCH 3/9] Plumb through an option for compression --- app/multitenant/memcache_client.go | 11 ++++++----- prog/app.go | 16 +++++++++------- prog/main.go | 23 +++++++++++++---------- 3 files changed, 28 insertions(+), 22 deletions(-) diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index fe4ec8f5d..9809f950c 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -58,11 +58,12 @@ type MemcacheClient struct { // MemcacheConfig defines how a MemcacheClient should be constructed. type MemcacheConfig struct { - Host string - Service string - Timeout time.Duration - UpdateInterval time.Duration - Expiration time.Duration + Host string + Service string + Timeout time.Duration + UpdateInterval time.Duration + Expiration time.Duration + CompressionLevel int } // NewMemcacheClient creates a new MemcacheClient that gets its server list diff --git a/prog/app.go b/prog/app.go index 0642676e8..e444db4ad 100644 --- a/prog/app.go +++ b/prog/app.go @@ -81,7 +81,7 @@ func awsConfigFromURL(url *url.URL) (*aws.Config, error) { return config, nil } -func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHostname, memcachedHostname string, memcachedTimeout time.Duration, memcachedService string, memcachedExpiration, window time.Duration, createTables bool) (app.Collector, error) { +func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHostname, memcachedHostname string, memcachedTimeout time.Duration, memcachedService string, memcachedExpiration time.Duration, memcachedCompressionLevel int, window time.Duration, createTables bool) (app.Collector, error) { if collectorURL == "local" { return app.NewCollector(window), nil } @@ -114,11 +114,12 @@ func collectorFactory(userIDer multitenant.UserIDer, collectorURL, s3URL, natsHo if memcachedHostname != "" { memcacheClient = multitenant.NewMemcacheClient( multitenant.MemcacheConfig{ - Host: memcachedHostname, - Timeout: memcachedTimeout, - Expiration: memcachedExpiration, - UpdateInterval: memcacheUpdateInterval, - Service: memcachedService, + Host: memcachedHostname, + Timeout: memcachedTimeout, + Expiration: memcachedExpiration, + UpdateInterval: memcacheUpdateInterval, + Service: memcachedService, + CompressionLevel: memcachedCompressionLevel, }, ) } @@ -214,7 +215,8 @@ func appMain(flags appFlags) { collector, err := collectorFactory( userIDer, flags.collectorURL, flags.s3URL, flags.natsHostname, flags.memcachedHostname, - flags.memcachedTimeout, flags.memcachedService, flags.memcachedExpiration, flags.window, flags.awsCreateTables) + flags.memcachedTimeout, flags.memcachedService, flags.memcachedExpiration, flags.memcachedCompressionLevel, + flags.window, flags.awsCreateTables) if err != nil { log.Fatalf("Error creating collector: %v", err) return diff --git a/prog/main.go b/prog/main.go index 7e9aed1f6..4998ed465 100644 --- a/prog/main.go +++ b/prog/main.go @@ -1,6 +1,7 @@ package main import ( + "compress/gzip" "flag" "fmt" "net" @@ -100,16 +101,17 @@ type appFlags struct { containerName string dockerEndpoint string - collectorURL string - s3URL string - controlRouterURL string - pipeRouterURL string - natsHostname string - memcachedHostname string - memcachedTimeout time.Duration - memcachedService string - memcachedExpiration time.Duration - userIDHeader string + collectorURL string + s3URL string + controlRouterURL string + pipeRouterURL string + natsHostname string + memcachedHostname string + memcachedTimeout time.Duration + memcachedService string + memcachedExpiration time.Duration + memcachedCompressionLevel int + userIDHeader string awsCreateTables bool consulInf string @@ -195,6 +197,7 @@ func main() { flag.DurationVar(&flags.app.memcachedTimeout, "app.memcached.timeout", 100*time.Millisecond, "Maximum time to wait before giving up on memcached requests.") flag.DurationVar(&flags.app.memcachedExpiration, "app.memcached.expiration", 2*15*time.Second, "How long reports stay in the memcache.") flag.StringVar(&flags.app.memcachedService, "app.memcached.service", "memcached", "SRV service used to discover memcache servers.") + flag.IntVar(&flags.app.memcachedCompressionLevel, "app.memcached.compression", gzip.DefaultCompression, "How much to compress reports stored in memcached.") flag.StringVar(&flags.app.userIDHeader, "app.userid.header", "", "HTTP header to use as userid") flag.BoolVar(&flags.app.awsCreateTables, "app.aws.create.tables", false, "Create the tables in DynamoDB") From 270a55060fa034da2445bda2dd2cbace9b0da3bc Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Fri, 15 Jul 2016 09:42:45 +0100 Subject: [PATCH 4/9] Add StoreReport methods to stores Not sure if we'll use them. --- app/multitenant/memcache_client.go | 8 ++++++++ app/multitenant/s3_client.go | 8 ++++++++ 2 files changed, 16 insertions(+) diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 9809f950c..ee8e0e07e 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -2,6 +2,7 @@ package multitenant import ( "bytes" + "compress/gzip" "fmt" "net" "sort" @@ -202,6 +203,13 @@ func (c *MemcacheClient) FetchReports(keys []string) (map[string]report.Report, return reports, missing, nil } +// StoreReport serializes and stores a report. +func (c *MemcacheClient) StoreReport(key string, report *report.Report) error { + var buf bytes.Buffer + report.WriteBinary(&buf, gzip.BestCompression) + return c.StoreBytes(key, buf.Bytes()) +} + // StoreBytes stores a report, expecting the report to be serialized already. func (c *MemcacheClient) StoreBytes(key string, content []byte) error { return instrument.TimeRequestHistogramStatus("Put", memcacheRequestDuration, memcacheStatusCode, func() error { diff --git a/app/multitenant/s3_client.go b/app/multitenant/s3_client.go index 94deef5f3..6d90bda74 100644 --- a/app/multitenant/s3_client.go +++ b/app/multitenant/s3_client.go @@ -2,6 +2,7 @@ package multitenant import ( "bytes" + "compress/gzip" "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/aws/session" @@ -84,6 +85,13 @@ func (store *S3Store) fetchReport(key string) (*report.Report, error) { return report.MakeFromBinary(resp.Body) } +// StoreReport serializes and stores a report. +func (store *S3Store) StoreReport(key string, report *report.Report) error { + var buf bytes.Buffer + report.WriteBinary(&buf, gzip.BestCompression) + return store.StoreBytes(key, buf.Bytes()) +} + // StoreBytes stores a report in S3, expecting the report to be serialized // already. func (store *S3Store) StoreBytes(key string, content []byte) error { From 0058229687446f5eb1c44ae7d372f30957411846 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Fri, 15 Jul 2016 09:53:27 +0100 Subject: [PATCH 5/9] Extract functions for calculating keys Not so much for re-use as to help jml understand what's going on --- app/multitenant/aws_collector.go | 26 ++++++++++++++++++++------ 1 file changed, 20 insertions(+), 6 deletions(-) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index 70450c675..0f7d893dc 100644 --- a/app/multitenant/aws_collector.go +++ b/app/multitenant/aws_collector.go @@ -339,6 +339,22 @@ func (c *awsCollector) Report(ctx context.Context) (report.Report, error) { return c.merger.Merge(reports), nil } +// calculateDynamoKeys generates the row & column keys for Dynamo. +func calculateDynamoKeys(userid string, now time.Time) (string, string) { + rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(now.UnixNano()/time.Hour.Nanoseconds(), 10)) + colKey := strconv.FormatInt(now.UnixNano(), 10) + return rowKey, colKey +} + +// calculateReportKey determines the key we should use for a report. +func calculateReportKey(rowKey, colKey string) (string, error) { + rowKeyHash := md5.New() + if _, err := io.WriteString(rowKeyHash, rowKey); err != nil { + return "", err + } + return fmt.Sprintf("%x/%s", rowKeyHash.Sum(nil), colKey), nil +} + func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { userid, err := c.userIDer(ctx) if err != nil { @@ -351,14 +367,12 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { reportSizeHistogram.Observe(float64(buf.Len())) // second, put the report on s3 - now := time.Now() - rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(now.UnixNano()/time.Hour.Nanoseconds(), 10)) - colKey := strconv.FormatInt(now.UnixNano(), 10) - rowKeyHash := md5.New() - if _, err := io.WriteString(rowKeyHash, rowKey); err != nil { + rowKey, colKey := calculateDynamoKeys(userid, time.Now()) + s3Key, err := calculateReportKey(rowKey, colKey) + if err != nil { return err } - s3Key := fmt.Sprintf("%x/%s", rowKeyHash.Sum(nil), colKey) + err = c.s3.StoreBytes(s3Key, buf.Bytes()) if err != nil { return err From 46dfeb627d25f54468f9b37f52162e577f59bfa0 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Fri, 15 Jul 2016 10:00:44 +0100 Subject: [PATCH 6/9] Call it reportKey --- app/multitenant/aws_collector.go | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index 0f7d893dc..571522c0f 100644 --- a/app/multitenant/aws_collector.go +++ b/app/multitenant/aws_collector.go @@ -368,30 +368,30 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { // second, put the report on s3 rowKey, colKey := calculateDynamoKeys(userid, time.Now()) - s3Key, err := calculateReportKey(rowKey, colKey) + reportKey, err := calculateReportKey(rowKey, colKey) if err != nil { return err } - err = c.s3.StoreBytes(s3Key, buf.Bytes()) + err = c.s3.StoreBytes(reportKey, buf.Bytes()) if err != nil { return err } // third, put it in memcache if c.memcache != nil { - err = c.memcache.StoreBytes(s3Key, buf.Bytes()) + err = c.memcache.StoreBytes(reportKey, buf.Bytes()) if err != nil { // NOTE: We don't abort here because failing to store in memcache // doesn't actually break anything else -- it's just an // optimization. - log.Warningf("Could not store %v in memcache: %v", s3Key, err) + log.Warningf("Could not store %v in memcache: %v", reportKey, err) } } // fourth, put the key in dynamodb dynamoValueSize.WithLabelValues("PutItem"). - Add(float64(len(s3Key))) + Add(float64(len(reportKey))) var resp *dynamodb.PutItemOutput err = instrument.TimeRequestHistogram("PutItem", dynamoRequestDuration, func() error { @@ -406,7 +406,7 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { N: aws.String(colKey), }, reportField: { - S: aws.String(s3Key), + S: aws.String(reportKey), }, }, ReturnConsumedCapacity: aws.String(dynamodb.ReturnConsumedCapacityTotal), @@ -422,7 +422,7 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { } if rep.Shortcut && c.nats != nil { - err := c.nats.Publish(userid, []byte(s3Key)) + err := c.nats.Publish(userid, []byte(reportKey)) natsRequests.WithLabelValues("Publish", instrument.ErrorCode(err)).Add(1) if err != nil { log.Errorf("Error sending shortcut report: %v", err) From 1fd8a5fb881c2826913722b32bf92cafb271e97d Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Fri, 15 Jul 2016 10:05:55 +0100 Subject: [PATCH 7/9] Use `StoreReport` in main AWS routine --- app/multitenant/aws_collector.go | 14 ++++---------- app/multitenant/memcache_client.go | 5 +++-- app/multitenant/s3_client.go | 7 +++++-- 3 files changed, 12 insertions(+), 14 deletions(-) diff --git a/app/multitenant/aws_collector.go b/app/multitenant/aws_collector.go index 571522c0f..346b7ebb4 100644 --- a/app/multitenant/aws_collector.go +++ b/app/multitenant/aws_collector.go @@ -1,8 +1,6 @@ package multitenant import ( - "bytes" - "compress/gzip" "crypto/md5" "fmt" "io" @@ -361,26 +359,22 @@ func (c *awsCollector) Add(ctx context.Context, rep report.Report) error { return err } - // first, encode the report into a buffer and record its size - var buf bytes.Buffer - rep.WriteBinary(&buf, gzip.BestCompression) - reportSizeHistogram.Observe(float64(buf.Len())) - - // second, put the report on s3 + // first, put the report on s3 rowKey, colKey := calculateDynamoKeys(userid, time.Now()) reportKey, err := calculateReportKey(rowKey, colKey) if err != nil { return err } - err = c.s3.StoreBytes(reportKey, buf.Bytes()) + reportSize, err := c.s3.StoreReport(reportKey, &rep) if err != nil { return err } + reportSizeHistogram.Observe(float64(reportSize)) // third, put it in memcache if c.memcache != nil { - err = c.memcache.StoreBytes(reportKey, buf.Bytes()) + _, err = c.memcache.StoreReport(reportKey, &rep) if err != nil { // NOTE: We don't abort here because failing to store in memcache // doesn't actually break anything else -- it's just an diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index ee8e0e07e..3539b39fb 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -204,10 +204,11 @@ func (c *MemcacheClient) FetchReports(keys []string) (map[string]report.Report, } // StoreReport serializes and stores a report. -func (c *MemcacheClient) StoreReport(key string, report *report.Report) error { +func (c *MemcacheClient) StoreReport(key string, report *report.Report) (int, error) { var buf bytes.Buffer report.WriteBinary(&buf, gzip.BestCompression) - return c.StoreBytes(key, buf.Bytes()) + err := c.StoreBytes(key, buf.Bytes()) + return buf.Len(), err } // StoreBytes stores a report, expecting the report to be serialized already. diff --git a/app/multitenant/s3_client.go b/app/multitenant/s3_client.go index 6d90bda74..48fff4f19 100644 --- a/app/multitenant/s3_client.go +++ b/app/multitenant/s3_client.go @@ -86,10 +86,13 @@ func (store *S3Store) fetchReport(key string) (*report.Report, error) { } // StoreReport serializes and stores a report. -func (store *S3Store) StoreReport(key string, report *report.Report) error { +// +// Returns the size of the report. This only equals bytes written if err is nil. +func (store *S3Store) StoreReport(key string, report *report.Report) (int, error) { var buf bytes.Buffer report.WriteBinary(&buf, gzip.BestCompression) - return store.StoreBytes(key, buf.Bytes()) + err := store.StoreBytes(key, buf.Bytes()) + return buf.Len(), err } // StoreBytes stores a report in S3, expecting the report to be serialized From bbd75ddd24b87629c5c7348e6a0e9e54f4ea96d1 Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Fri, 15 Jul 2016 10:46:12 +0100 Subject: [PATCH 8/9] Use memcache compression level from config --- app/multitenant/memcache_client.go | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 3539b39fb..732592e99 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -2,7 +2,6 @@ package multitenant import ( "bytes" - "compress/gzip" "fmt" "net" "sort" @@ -47,11 +46,12 @@ func init() { // MemcacheClient is a memcache client that gets its server list from SRV // records, and periodically updates that ServerList. type MemcacheClient struct { - client *memcache.Client - serverList *memcache.ServerList - expiration int32 - hostname string - service string + client *memcache.Client + serverList *memcache.ServerList + expiration int32 + hostname string + service string + compressionLevel int quit chan struct{} wait sync.WaitGroup @@ -75,12 +75,13 @@ func NewMemcacheClient(config MemcacheConfig) *MemcacheClient { client.Timeout = config.Timeout newClient := &MemcacheClient{ - client: client, - serverList: &servers, - expiration: int32(config.Expiration.Seconds()), - hostname: config.Host, - service: config.Service, - quit: make(chan struct{}), + client: client, + serverList: &servers, + expiration: int32(config.Expiration.Seconds()), + hostname: config.Host, + service: config.Service, + compressionLevel: config.CompressionLevel, + quit: make(chan struct{}), } err := newClient.updateMemcacheServers() if err != nil { @@ -206,7 +207,7 @@ func (c *MemcacheClient) FetchReports(keys []string) (map[string]report.Report, // StoreReport serializes and stores a report. func (c *MemcacheClient) StoreReport(key string, report *report.Report) (int, error) { var buf bytes.Buffer - report.WriteBinary(&buf, gzip.BestCompression) + report.WriteBinary(&buf, c.compressionLevel) err := c.StoreBytes(key, buf.Bytes()) return buf.Len(), err } From a3648f0c891bebe16ca7ba14156e7522d508529b Mon Sep 17 00:00:00 2001 From: Jonathan Lange Date: Fri, 15 Jul 2016 12:58:27 +0100 Subject: [PATCH 9/9] Inline StoreBytes --- app/multitenant/memcache_client.go | 11 +++-------- app/multitenant/s3_client.go | 12 +++--------- 2 files changed, 6 insertions(+), 17 deletions(-) diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 732592e99..3b9db267e 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -208,14 +208,9 @@ func (c *MemcacheClient) FetchReports(keys []string) (map[string]report.Report, func (c *MemcacheClient) StoreReport(key string, report *report.Report) (int, error) { var buf bytes.Buffer report.WriteBinary(&buf, c.compressionLevel) - err := c.StoreBytes(key, buf.Bytes()) - return buf.Len(), err -} - -// StoreBytes stores a report, expecting the report to be serialized already. -func (c *MemcacheClient) StoreBytes(key string, content []byte) error { - return instrument.TimeRequestHistogramStatus("Put", memcacheRequestDuration, memcacheStatusCode, func() error { - item := memcache.Item{Key: key, Value: content, Expiration: c.expiration} + err := instrument.TimeRequestHistogramStatus("Put", memcacheRequestDuration, memcacheStatusCode, func() error { + item := memcache.Item{Key: key, Value: buf.Bytes(), Expiration: c.expiration} return c.client.Set(&item) }) + return buf.Len(), err } diff --git a/app/multitenant/s3_client.go b/app/multitenant/s3_client.go index 48fff4f19..9e1aaaeac 100644 --- a/app/multitenant/s3_client.go +++ b/app/multitenant/s3_client.go @@ -91,19 +91,13 @@ func (store *S3Store) fetchReport(key string) (*report.Report, error) { func (store *S3Store) StoreReport(key string, report *report.Report) (int, error) { var buf bytes.Buffer report.WriteBinary(&buf, gzip.BestCompression) - err := store.StoreBytes(key, buf.Bytes()) - return buf.Len(), err -} - -// StoreBytes stores a report in S3, expecting the report to be serialized -// already. -func (store *S3Store) StoreBytes(key string, content []byte) error { - return instrument.TimeRequestHistogram("Put", s3RequestDuration, func() error { + err := instrument.TimeRequestHistogram("Put", s3RequestDuration, func() error { _, err := store.s3.PutObject(&s3.PutObjectInput{ - Body: bytes.NewReader(content), + Body: bytes.NewReader(buf.Bytes()), Bucket: aws.String(store.bucketName), Key: aws.String(key), }) return err }) + return buf.Len(), err }