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

@@ -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)
 	}
 }

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)

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())
 		}
 	}

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 {

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)
 }

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{

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
 }