From 3f8ba95beaf2804dada6a9945ed8952fe4daaf58 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 16 Sep 2019 16:31:00 +0000 Subject: [PATCH 1/4] refactor: pass msgpack flag into ReadBinary instead of a codec.Handle. This is a cleaner dependency. --- app/router.go | 12 +++++------- report/marshal.go | 35 +++++++++++++++++++++-------------- 2 files changed, 26 insertions(+), 21 deletions(-) diff --git a/app/router.go b/app/router.go index 3bf27060d..fb4956949 100644 --- a/app/router.go +++ b/app/router.go @@ -15,7 +15,6 @@ import ( "github.com/NYTimes/gziphandler" "github.com/gorilla/mux" log "github.com/sirupsen/logrus" - "github.com/ugorji/go/codec" "github.com/weaveworks/scope/common/hostname" "github.com/weaveworks/scope/common/xfer" @@ -128,19 +127,18 @@ func RegisterReportPostHandler(a Adder, router *mux.Router) { } contentType := r.Header.Get("Content-Type") - isMsgpack := strings.HasPrefix(contentType, "application/msgpack") - var handle codec.Handle + var isMsgpack bool switch { + case strings.HasPrefix(contentType, "application/msgpack"): + isMsgpack = true case strings.HasPrefix(contentType, "application/json"): - handle = &codec.JsonHandle{} - case isMsgpack: - handle = &codec.MsgpackHandle{} + isMsgpack = false default: respondWith(w, http.StatusBadRequest, fmt.Errorf("Unsupported Content-Type: %v", contentType)) return } - if err := rpt.ReadBinary(ctx, reader, gzipped, handle); err != nil { + if err := rpt.ReadBinary(ctx, reader, gzipped, isMsgpack); err != nil { respondWith(w, http.StatusBadRequest, err) return } diff --git a/report/marshal.go b/report/marshal.go index fcb32a350..2f51d16bf 100644 --- a/report/marshal.go +++ b/report/marshal.go @@ -74,9 +74,9 @@ var gzipWriterPool = &sync.Pool{ // ReadBinary reads bytes into a Report. // -// Will decompress the binary if gzipped is true, and will use the given -// codecHandle to decode it. -func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, codecHandle codec.Handle) error { +// Will decompress the binary if gzipped is true, and decode as +// msgpack if true, otherwise JSON +func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, msgpack bool) error { span, ctx := opentracing.StartSpanFromContext(ctx, "report.ReadBinary") defer span.Finish() var err error @@ -102,7 +102,7 @@ func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, co if err != nil { return err } - if err := rep.ReadBytes(buf.Bytes(), codecHandle); err != nil { + if err := rep.ReadBytes(buf.Bytes(), codecHandle(msgpack)); err != nil { return err } log.Debugf( @@ -118,7 +118,7 @@ func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, co // MakeFromBinary constructs a Report from a gzipped msgpack. func MakeFromBinary(ctx context.Context, r io.Reader) (*Report, error) { rep := MakeReport() - if err := rep.ReadBinary(ctx, r, true, &codec.MsgpackHandle{}); err != nil { + if err := rep.ReadBinary(ctx, r, true, true); err != nil { return nil, err } return &rep, nil @@ -166,12 +166,12 @@ func MakeFromFile(ctx context.Context, path string) (rpt Report, _ error) { } defer f.Close() - handle, gzipped, err := handlerFromFileType(path) + msgpack, gzipped, err := fileType(path) if err != nil { return rpt, err } - err = rpt.ReadBinary(ctx, f, gzipped, handle) + err = rpt.ReadBinary(ctx, f, gzipped, msgpack) return rpt, err } @@ -185,7 +185,7 @@ func (rep *Report) WriteToFile(path string) error { } defer f.Close() - handle, gzipped, err := handlerFromFileType(path) + msgpack, gzipped, err := fileType(path) if err != nil { return err } @@ -202,22 +202,29 @@ func (rep *Report) WriteToFile(path string) error { w = gzwriter } - return codec.NewEncoder(w, handle).Encode(rep) + return codec.NewEncoder(w, codecHandle(msgpack)).Encode(rep) } -func handlerFromFileType(path string) (codec.Handle, bool, error) { +func fileType(path string) (msgpack bool, gzipped bool, err error) { fileType := filepath.Ext(path) - gzipped := false + gzipped = false if fileType == ".gz" { gzipped = true fileType = filepath.Ext(strings.TrimSuffix(path, fileType)) } switch fileType { case ".json": - return &codec.JsonHandle{}, gzipped, nil + return false, gzipped, nil case ".msgpack": - return &codec.MsgpackHandle{}, gzipped, nil + return true, gzipped, nil default: - return nil, false, fmt.Errorf("Unsupported file extension: %v", fileType) + return false, false, fmt.Errorf("Unsupported file extension: %v", fileType) } } + +func codecHandle(msgpack bool) codec.Handle { + if msgpack { + return &codec.MsgpackHandle{} + } + return &codec.JsonHandle{} +} From 2bbd4a3f0d908d669d50614a28d8ec6b002467f2 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 16 Sep 2019 16:44:38 +0000 Subject: [PATCH 2/4] refactor: remove MakeFromBytes() function which is almost the same as ReadBinary() --- app/multitenant/memcache_client.go | 6 ++++-- report/marshal.go | 27 --------------------------- report/marshal_test.go | 10 ++++++---- 3 files changed, 10 insertions(+), 33 deletions(-) diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index a4443c342..5137e2d90 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -1,6 +1,7 @@ package multitenant import ( + "bytes" "fmt" "net" "sort" @@ -183,13 +184,14 @@ func (c *MemcacheClient) FetchReports(ctx context.Context, keys []string) (map[s continue } go func(key string) { - rep, err := report.MakeFromBytes(item.Value) + rep := report.MakeReport() + err := rep.ReadBinary(ctx, bytes.NewBuffer(item.Value), true, true) if err != nil { log.Warningf("Corrupt report in memcache %v: %v", key, err) ch <- result{key: key} return } - ch <- result{key: key, report: rep} + ch <- result{key: key, report: &rep} }(key) } diff --git a/report/marshal.go b/report/marshal.go index 2f51d16bf..044cf3750 100644 --- a/report/marshal.go +++ b/report/marshal.go @@ -129,33 +129,6 @@ func (rep *Report) ReadBytes(buf []byte, codecHandle codec.Handle) error { return codec.NewDecoderBytes(buf, codecHandle).Decode(&rep) } -// MakeFromBytes constructs a Report from a gzipped msgpack. -func MakeFromBytes(buf []byte) (*Report, error) { - compressedSize := len(buf) - r, err := gzip.NewReader(bytes.NewBuffer(buf)) - if err != nil { - return nil, err - } - buffer := bufferPool.Get().(*bytes.Buffer) - buffer.Reset() - defer bufferPool.Put(buffer) - uncompressedSize, err := buffer.ReadFrom(r) - if err != nil { - return nil, err - } - log.Debugf( - "Received report sizes: compressed %d bytes, uncompressed %d bytes (%.2f%%)", - compressedSize, - uncompressedSize, - float32(compressedSize)/float32(uncompressedSize)*100, - ) - rep := MakeReport() - if err := rep.ReadBytes(buffer.Bytes(), &codec.MsgpackHandle{}); err != nil { - return nil, err - } - return &rep, nil -} - // MakeFromFile construct a Report from a file, with the encoding // determined by the extension (".msgpack" or ".json", with an // optional ".gz"). diff --git a/report/marshal_test.go b/report/marshal_test.go index b79be7fe8..227f8e23d 100644 --- a/report/marshal_test.go +++ b/report/marshal_test.go @@ -1,6 +1,7 @@ package report_test import ( + "bytes" "context" "reflect" "testing" @@ -14,7 +15,7 @@ import ( func TestRoundtrip(t *testing.T) { r1 := report.MakeReport() buf, _ := r1.WriteBinary() - bytes := append([]byte{}, buf.Bytes()...) // copy the contents for later + original := append([]byte{}, buf.Bytes()...) // copy the contents for later r2, err := report.MakeFromBinary(context.Background(), buf) if err != nil { t.Error(err) @@ -22,12 +23,13 @@ func TestRoundtrip(t *testing.T) { if !reflect.DeepEqual(r1, *r2) { t.Errorf("%v != %v", r1, *r2) } - r3, err := report.MakeFromBytes(bytes) + r3 := report.MakeReport() + err = r3.ReadBinary(context.Background(), bytes.NewBuffer(original), true, true) if err != nil { t.Error(err) } - if !reflect.DeepEqual(r1, *r3) { - t.Errorf("%v != %v", r1, *r3) + if !reflect.DeepEqual(r1, r3) { + t.Errorf("%v != %v", r1, r3) } } From 5ffb563051a760d81f8ad226bff340bafd930671 Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 16 Sep 2019 16:51:46 +0000 Subject: [PATCH 3/4] refactor: eliminate ReadBytes() function that is only called once --- report/marshal.go | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/report/marshal.go b/report/marshal.go index 044cf3750..b67993d5f 100644 --- a/report/marshal.go +++ b/report/marshal.go @@ -102,7 +102,7 @@ func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, ms if err != nil { return err } - if err := rep.ReadBytes(buf.Bytes(), codecHandle(msgpack)); err != nil { + if err := codec.NewDecoderBytes(buf.Bytes(), codecHandle(msgpack)).Decode(&rep); err != nil { return err } log.Debugf( @@ -124,11 +124,6 @@ func MakeFromBinary(ctx context.Context, r io.Reader) (*Report, error) { return &rep, nil } -// ReadBytes reads bytes into a Report, using a codecHandle. -func (rep *Report) ReadBytes(buf []byte, codecHandle codec.Handle) error { - return codec.NewDecoderBytes(buf, codecHandle).Decode(&rep) -} - // MakeFromFile construct a Report from a file, with the encoding // determined by the extension (".msgpack" or ".json", with an // optional ".gz"). From 13af359bcfb463498d41d0ac3e72ae5d5e92f36c Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Mon, 23 Sep 2019 10:01:39 +0000 Subject: [PATCH 4/4] refactor: eliminate report.ReadBinary() in favour of MakeFromBinary() The signature of MakeFromFile changed to return a pointer for consistency. --- app/benchmark_internal_test.go | 2 +- app/collector.go | 2 +- app/multitenant/memcache_client.go | 5 ++--- app/multitenant/s3_client.go | 2 +- app/router.go | 6 +++--- report/marshal.go | 25 ++++++++----------------- report/marshal_test.go | 14 ++------------ 7 files changed, 18 insertions(+), 38 deletions(-) diff --git a/app/benchmark_internal_test.go b/app/benchmark_internal_test.go index 7a14f5ee9..499ce1ed8 100644 --- a/app/benchmark_internal_test.go +++ b/app/benchmark_internal_test.go @@ -34,7 +34,7 @@ func readReportFiles(b *testing.B, path string) []report.Report { if err != nil { return err } - reports = append(reports, rpt) + reports = append(reports, *rpt) return nil }); err != nil { b.Fatal(err) diff --git a/app/collector.go b/app/collector.go index 5d9deea9b..fbd21f7cd 100644 --- a/app/collector.go +++ b/app/collector.go @@ -282,7 +282,7 @@ func NewFileCollector(path string, window time.Duration) (Collector, error) { if err != nil { return err } - reports = append(reports, rpt) + reports = append(reports, *rpt) return nil }); err != nil { return nil, err diff --git a/app/multitenant/memcache_client.go b/app/multitenant/memcache_client.go index 5137e2d90..86a35c8c5 100644 --- a/app/multitenant/memcache_client.go +++ b/app/multitenant/memcache_client.go @@ -184,14 +184,13 @@ func (c *MemcacheClient) FetchReports(ctx context.Context, keys []string) (map[s continue } go func(key string) { - rep := report.MakeReport() - err := rep.ReadBinary(ctx, bytes.NewBuffer(item.Value), true, true) + rep, err := report.MakeFromBinary(ctx, bytes.NewBuffer(item.Value), true, true) if err != nil { log.Warningf("Corrupt report in memcache %v: %v", key, err) ch <- result{key: key} return } - ch <- result{key: key, report: &rep} + ch <- result{key: key, report: rep} }(key) } diff --git a/app/multitenant/s3_client.go b/app/multitenant/s3_client.go index 8f87109cd..250c7136e 100644 --- a/app/multitenant/s3_client.go +++ b/app/multitenant/s3_client.go @@ -87,7 +87,7 @@ func (store *S3Store) fetchReport(ctx context.Context, key string) (*report.Repo return nil, err } defer resp.Body.Close() - return report.MakeFromBinary(ctx, resp.Body) + return report.MakeFromBinary(ctx, resp.Body, true, true) } // StoreReportBytes stores a report. diff --git a/app/router.go b/app/router.go index fb4956949..fc507419c 100644 --- a/app/router.go +++ b/app/router.go @@ -116,7 +116,6 @@ func RegisterReportPostHandler(a Adder, router *mux.Router) { post := router.Methods("POST").Subrouter() post.HandleFunc("/api/report", requestContextDecorator(func(ctx context.Context, w http.ResponseWriter, r *http.Request) { var ( - rpt report.Report buf = &bytes.Buffer{} reader = io.TeeReader(r.Body, buf) ) @@ -138,7 +137,8 @@ func RegisterReportPostHandler(a Adder, router *mux.Router) { return } - if err := rpt.ReadBinary(ctx, reader, gzipped, isMsgpack); err != nil { + rpt, err := report.MakeFromBinary(ctx, reader, gzipped, isMsgpack) + if err != nil { respondWith(w, http.StatusBadRequest, err) return } @@ -148,7 +148,7 @@ func RegisterReportPostHandler(a Adder, router *mux.Router) { buf, _ = rpt.WriteBinary() } - if err := a.Add(ctx, rpt, buf.Bytes()); err != nil { + if err := a.Add(ctx, *rpt, buf.Bytes()); err != nil { log.Errorf("Error Adding report: %v", err) respondWith(w, http.StatusInternalServerError, err) return diff --git a/report/marshal.go b/report/marshal.go index b67993d5f..38d98253d 100644 --- a/report/marshal.go +++ b/report/marshal.go @@ -72,11 +72,11 @@ var gzipWriterPool = &sync.Pool{ New: func() interface{} { w, _ := gzip.NewWriterLevel(nil, gzip.DefaultCompression); return w }, } -// ReadBinary reads bytes into a Report. +// MakeFromBinary constructs a Report from binary data. // // Will decompress the binary if gzipped is true, and decode as // msgpack if true, otherwise JSON -func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, msgpack bool) error { +func MakeFromBinary(ctx context.Context, r io.Reader, gzipped bool, msgpack bool) (*Report, error) { span, ctx := opentracing.StartSpanFromContext(ctx, "report.ReadBinary") defer span.Finish() var err error @@ -91,7 +91,7 @@ func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, ms if gzipped { r, err = gzip.NewReader(r) if err != nil { - return err + return nil, err } } // Read everything into memory before decoding: it's faster @@ -100,10 +100,11 @@ func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, ms defer bufferPool.Put(buf) uncompressedSize, err := buf.ReadFrom(r) if err != nil { - return err + return nil, err } + rep := MakeReport() if err := codec.NewDecoderBytes(buf.Bytes(), codecHandle(msgpack)).Decode(&rep); err != nil { - return err + return nil, err } log.Debugf( "Received report sizes: compressed %d bytes, uncompressed %d bytes (%.2f%%)", @@ -112,22 +113,13 @@ func (rep *Report) ReadBinary(ctx context.Context, r io.Reader, gzipped bool, ms float32(compressedSize)/float32(uncompressedSize)*100, ) span.LogFields(otlog.Uint64("compressedSize", compressedSize), otlog.Int64("uncompressedSize", uncompressedSize)) - return nil -} - -// MakeFromBinary constructs a Report from a gzipped msgpack. -func MakeFromBinary(ctx context.Context, r io.Reader) (*Report, error) { - rep := MakeReport() - if err := rep.ReadBinary(ctx, r, true, true); err != nil { - return nil, err - } return &rep, nil } // MakeFromFile construct a Report from a file, with the encoding // determined by the extension (".msgpack" or ".json", with an // optional ".gz"). -func MakeFromFile(ctx context.Context, path string) (rpt Report, _ error) { +func MakeFromFile(ctx context.Context, path string) (rpt *Report, _ error) { f, err := os.Open(path) if err != nil { return rpt, err @@ -139,8 +131,7 @@ func MakeFromFile(ctx context.Context, path string) (rpt Report, _ error) { return rpt, err } - err = rpt.ReadBinary(ctx, f, gzipped, msgpack) - return rpt, err + return MakeFromBinary(ctx, f, gzipped, msgpack) } // WriteToFile writes a Report to a file. The encoding is determined diff --git a/report/marshal_test.go b/report/marshal_test.go index 227f8e23d..2926043b3 100644 --- a/report/marshal_test.go +++ b/report/marshal_test.go @@ -1,7 +1,6 @@ package report_test import ( - "bytes" "context" "reflect" "testing" @@ -15,22 +14,13 @@ import ( func TestRoundtrip(t *testing.T) { r1 := report.MakeReport() buf, _ := r1.WriteBinary() - original := append([]byte{}, buf.Bytes()...) // copy the contents for later - r2, err := report.MakeFromBinary(context.Background(), buf) + r2, err := report.MakeFromBinary(context.Background(), buf, true, true) if err != nil { t.Error(err) } if !reflect.DeepEqual(r1, *r2) { t.Errorf("%v != %v", r1, *r2) } - r3 := report.MakeReport() - err = r3.ReadBinary(context.Background(), bytes.NewBuffer(original), true, true) - if err != nil { - t.Error(err) - } - if !reflect.DeepEqual(r1, r3) { - t.Errorf("%v != %v", r1, r3) - } } // Create a Report for test purposes that contains about one of @@ -74,7 +64,7 @@ func makeTestReport() report.Report { func TestBiggerRoundtrip(t *testing.T) { r1 := makeTestReport() buf, _ := r1.WriteBinary() - r2, err := report.MakeFromBinary(context.Background(), buf) + r2, err := report.MakeFromBinary(context.Background(), buf, true, true) if err != nil { t.Error(err) }