From 96f51c47af0168526a7104b75807cf8cdc2714fa Mon Sep 17 00:00:00 2001 From: Bryan Boreham Date: Thu, 31 May 2018 11:05:07 +0000 Subject: [PATCH] probe: Eliminate Publisher interface from app_client Simplification: everything now implements Publish(Report), and we do away with writing/reading/writing in the MultiAppClient. --- extras/fixprobe/main.go | 10 +++++-- probe/appclient/app_client_internal_test.go | 9 +++---- probe/appclient/multi_client.go | 24 +++-------------- probe/appclient/multi_client_test.go | 5 ++-- probe/appclient/report_publisher.go | 30 +++++++++++++-------- probe/probe.go | 4 +-- probe/probe_internal_test.go | 11 +------- 7 files changed, 41 insertions(+), 52 deletions(-) diff --git a/extras/fixprobe/main.go b/extras/fixprobe/main.go index 5b99925cb..7dcc9a821 100644 --- a/extras/fixprobe/main.go +++ b/extras/fixprobe/main.go @@ -2,6 +2,8 @@ package main import ( + "bytes" + "compress/gzip" "flag" "fmt" "io/ioutil" @@ -60,8 +62,12 @@ func main() { log.Fatal(err) } - rp := appclient.NewReportPublisher(client, false) + buf := &bytes.Buffer{} + err = fixedReport.WriteBinary(buf, gzip.DefaultCompression) + if err != nil { + log.Fatal(err) + } for range time.Tick(*publishInterval) { - rp.Publish(fixedReport) + client.Publish(bytes.NewReader(buf.Bytes()), fixedReport.Shortcut) } } diff --git a/probe/appclient/app_client_internal_test.go b/probe/appclient/app_client_internal_test.go index 66bcae51d..210878b12 100644 --- a/probe/appclient/app_client_internal_test.go +++ b/probe/appclient/app_client_internal_test.go @@ -104,9 +104,9 @@ func TestAppClientPublish(t *testing.T) { defer p.Stop() // First few reports might be dropped as the client is spinning up. - rp := NewReportPublisher(p, false) for i := 0; i < 10; i++ { - if err := rp.Publish(rpt); err != nil { + buf, _ := serializeReport(rpt) + if err := p.Publish(buf, false); err != nil { t.Error(err) } time.Sleep(10 * time.Millisecond) @@ -204,15 +204,14 @@ func TestStop(t *testing.T) { t.Fatal(err) } - rp := NewReportPublisher(p, false) - // Make sure the app received our report and is stuck for done := false; !done; { select { case <-receivedReport: done = true default: - if err := rp.Publish(rpt); err != nil { + buf, _ := serializeReport(rpt) + if err := p.Publish(buf, false); err != nil { t.Error(err) } time.Sleep(10 * time.Millisecond) diff --git a/probe/appclient/multi_client.go b/probe/appclient/multi_client.go index c8187a5ed..e93b33674 100644 --- a/probe/appclient/multi_client.go +++ b/probe/appclient/multi_client.go @@ -4,8 +4,6 @@ import ( "bytes" "errors" "fmt" - "io" - "io/ioutil" "net/url" "strings" "sync" @@ -37,13 +35,6 @@ type clientTuple struct { AppClient } -// Publisher is something which can send a stream of data somewhere, probably -// to a remote collector. -type Publisher interface { - Publish(io.Reader, bool) error - Stop() -} - // MultiAppClient maintains a set of upstream apps, and ensures we have an // AppClient for each one. type MultiAppClient interface { @@ -51,7 +42,7 @@ type MultiAppClient interface { PipeConnection(appID, pipeID string, pipe xfer.Pipe) error PipeClose(appID, pipeID string) error Stop() - Publish(io.Reader, bool) error + ReportPublisher } // NewMultiAppClient creates a new MultiAppClient. @@ -165,25 +156,18 @@ func (c *multiClient) Stop() { // underlying publishers sequentially. To do that, it needs to drain the // reader, and recreate new readers for each publisher. Note that it will // publish to one endpoint for each unique ID. Failed publishes don't count. -func (c *multiClient) Publish(r io.Reader, shortcut bool) error { +func (c *multiClient) Publish(r report.Report) error { c.mtx.Lock() defer c.mtx.Unlock() - if len(c.clients) <= 1 { // optimisation - for _, c := range c.clients { - return c.Publish(r, shortcut) - } - return nil - } - - buf, err := ioutil.ReadAll(r) + buf, err := serializeReport(r) if err != nil { return err } errs := []string{} for _, c := range c.clients { - if err := c.Publish(bytes.NewReader(buf), shortcut); err != nil { + if err := c.Publish(bytes.NewReader(buf.Bytes()), r.Shortcut); err != nil { errs = append(errs, err.Error()) } } diff --git a/probe/appclient/multi_client_test.go b/probe/appclient/multi_client_test.go index e1789a710..770916557 100644 --- a/probe/appclient/multi_client_test.go +++ b/probe/appclient/multi_client_test.go @@ -1,7 +1,6 @@ package appclient_test import ( - "bytes" "io" "net/url" "runtime" @@ -9,6 +8,7 @@ import ( "github.com/weaveworks/scope/common/xfer" "github.com/weaveworks/scope/probe/appclient" + "github.com/weaveworks/scope/report" ) type mockClient struct { @@ -105,8 +105,9 @@ func TestMultiClientPublish(t *testing.T) { mp.Set("a", []url.URL{{Host: "a1"}, {Host: "a2"}}) mp.Set("b", []url.URL{{Host: "b2"}, {Host: "b3"}}) + rpt := report.MakeReport() for i := 1; i < 10; i++ { - if err := mp.Publish(&bytes.Buffer{}, false); err != nil { + if err := mp.Publish(rpt); err != nil { t.Error(err) } if want, have := 3*i, sum(); want != have { diff --git a/probe/appclient/report_publisher.go b/probe/appclient/report_publisher.go index 2f7bdd0b8..8d373f710 100644 --- a/probe/appclient/report_publisher.go +++ b/probe/appclient/report_publisher.go @@ -3,32 +3,40 @@ package appclient import ( "bytes" "compress/gzip" + "github.com/weaveworks/scope/report" ) -// A ReportPublisher uses a buffer pool to serialise reports, which it -// then passes to a publisher -type ReportPublisher struct { - publisher Publisher +// ReportPublisher publishes reports, probably to a remote collector. +type ReportPublisher interface { + Publish(r report.Report) error +} + +type reportPublisher struct { + publisher ReportPublisher noControls bool } // NewReportPublisher creates a new report publisher -func NewReportPublisher(publisher Publisher, noControls bool) *ReportPublisher { - return &ReportPublisher{ +func NewReportPublisher(publisher ReportPublisher, noControls bool) ReportPublisher { + return &reportPublisher{ publisher: publisher, noControls: noControls, } } -// Publish serialises and compresses a report, then passes it to a publisher -func (p *ReportPublisher) Publish(r report.Report) error { +func serializeReport(r report.Report) (*bytes.Buffer, error) { + buf := &bytes.Buffer{} + err := r.WriteBinary(buf, gzip.DefaultCompression) + return buf, err +} + +// Publish sanitises a report, then passes it to a publisher +func (p *reportPublisher) Publish(r report.Report) error { if p.noControls { r.WalkTopologies(func(t *report.Topology) { t.Controls = report.Controls{} }) } - buf := &bytes.Buffer{} - r.WriteBinary(buf, gzip.DefaultCompression) - return p.publisher.Publish(buf, r.Shortcut) + return p.publisher.Publish(r) } diff --git a/probe/probe.go b/probe/probe.go index a9139fc4d..d9c07bd9a 100644 --- a/probe/probe.go +++ b/probe/probe.go @@ -18,7 +18,7 @@ const ( // Probe sits there, generating and publishing reports. type Probe struct { spyInterval, publishInterval time.Duration - publisher *appclient.ReportPublisher + publisher appclient.ReportPublisher tickers []Ticker reporters []Reporter @@ -67,7 +67,7 @@ type Ticker interface { // New makes a new Probe. func New( spyInterval, publishInterval time.Duration, - publisher appclient.Publisher, + publisher appclient.ReportPublisher, noControls bool, ) *Probe { result := &Probe{ diff --git a/probe/probe_internal_test.go b/probe/probe_internal_test.go index af6404af6..adc5cd143 100644 --- a/probe/probe_internal_test.go +++ b/probe/probe_internal_test.go @@ -1,12 +1,9 @@ package probe import ( - "compress/gzip" - "io" "testing" "time" - "github.com/ugorji/go/codec" "github.com/weaveworks/common/mtime" "github.com/weaveworks/scope/report" "github.com/weaveworks/scope/test" @@ -53,13 +50,7 @@ type mockPublisher struct { have chan report.Report } -func (m mockPublisher) Publish(in io.Reader, shortcut bool) error { - var r report.Report - if reader, err := gzip.NewReader(in); err != nil { - return err - } else if err := codec.NewDecoder(reader, &codec.MsgpackHandle{}).Decode(&r); err != nil { - return err - } +func (m mockPublisher) Publish(r report.Report) error { m.have <- r return nil }