Add option to print probe reports to stdout, for debugging

app/router.go

@@ -121,13 +121,13 @@ func RegisterReportPostHandler(a Adder, router *mux.Router) {
 	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)
+			buf    = &bytes.Buffer{}
+			reader = io.TeeReader(r.Body, buf)
 		)
 
 		gzipped := strings.Contains(r.Header.Get("Content-Encoding"), "gzip")
 		if !gzipped {
-			reader = io.TeeReader(r.Body, gzip.NewWriter(&buf))
+			reader = io.TeeReader(r.Body, gzip.NewWriter(buf))
 		}
 
 		contentType := r.Header.Get("Content-Type")
@@ -150,8 +150,7 @@ func RegisterReportPostHandler(a Adder, router *mux.Router) {
 
 		// a.Add(..., buf) assumes buf is gzip'd msgpack
 		if !isMsgpack {
-			buf = bytes.Buffer{}
-			rpt.WriteBinary(&buf, gzip.DefaultCompression)
+			buf, _ = rpt.WriteBinary()
 		}
 
 		if err := a.Add(ctx, rpt, buf.Bytes()); err != nil {

extras/fixprobe/main.go

@@ -2,6 +2,7 @@
 package main
 
 import (
+	"bytes"
 	"flag"
 	"fmt"
 	"io/ioutil"
@@ -60,8 +61,11 @@ func main() {
 		log.Fatal(err)
 	}
 
-	rp := appclient.NewReportPublisher(client, false)
+	buf, err := fixedReport.WriteBinary()
+	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, _ := rpt.WriteBinary()
+		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, _ := rpt.WriteBinary()
+			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,21 +35,14 @@ 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 {
 	Set(hostname string, urls []url.URL)
 	PipeConnection(appID, pipeID string, pipe xfer.Pipe) error
 	PipeClose(appID, pipeID string) error
 	Stop()
-	Publish(io.Reader, bool) error
+	Publish(r report.Report) error
 }
 
 // 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 := r.WriteBinary()
 	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,14 +1,14 @@
 package appclient_test
 
 import (
-	"bytes"
 	"io"
 	"net/url"
 	"runtime"
 	"testing"
 
 	"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/controls/pipes.go

@@ -54,3 +54,12 @@ func (p *pipe) Close() error {
 	}
 	return err2
 }
+
+// DummyPipeClient implements PipeClient when running the probe in debugging mode
+type DummyPipeClient struct{}
+
+// PipeConnection implements controls.PipeClient
+func (DummyPipeClient) PipeConnection(appID, pipeID string, pipe xfer.Pipe) error { return nil }
+
+// PipeClose implements controls.PipeClient
+func (DummyPipeClient) PipeClose(appID, pipeID string) error { return nil }

probe/probe.go

@@ -7,18 +7,23 @@ import (
 	log "github.com/Sirupsen/logrus"
 	"github.com/armon/go-metrics"
 
-	"github.com/weaveworks/scope/probe/appclient"
 	"github.com/weaveworks/scope/report"
 )
 
 const (
 	reportBufferSize = 16
 )
 
+// ReportPublisher publishes reports, probably to a remote collector.
+type ReportPublisher interface {
+	Publish(r report.Report) error
+}
+
 // Probe sits there, generating and publishing reports.
 type Probe struct {
 	spyInterval, publishInterval time.Duration
-	publisher                    *appclient.ReportPublisher
+	publisher                    ReportPublisher
+	noControls                   bool
 
 	tickers   []Ticker
 	reporters []Reporter
@@ -67,13 +72,14 @@ type Ticker interface {
 // New makes a new Probe.
 func New(
 	spyInterval, publishInterval time.Duration,
-	publisher appclient.Publisher,
+	publisher ReportPublisher,
 	noControls bool,
 ) *Probe {
 	result := &Probe{
 		spyInterval:     spyInterval,
 		publishInterval: publishInterval,
-		publisher:       appclient.NewReportPublisher(publisher, noControls),
+		publisher:       publisher,
+		noControls:      noControls,
 		quit:            make(chan struct{}),
 		spiedReports:    make(chan report.Report, reportBufferSize),
 		shortcutReports: make(chan report.Report, reportBufferSize),
@@ -200,6 +206,11 @@ ForLoop:
 		}
 	}
 
+	if p.noControls {
+		rpt.WalkTopologies(func(t *report.Topology) {
+			t.Controls = report.Controls{}
+		})
+	}
 	if err := p.publisher.Publish(rpt); err != nil {
 		log.Infof("publish: %v", err)
 	}

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
 }

prog/main.go

@@ -93,6 +93,7 @@ type flags struct {
 }
 
 type probeFlags struct {
+	printOnStdout          bool
 	token                  string
 	httpListen             string
 	publishInterval        time.Duration
@@ -274,6 +275,7 @@ func setupFlags(flags *flags) {
 	flag.Bool("app-only", false, "Only run the app.")
 
 	// Probe flags
+	flag.BoolVar(&flags.probe.printOnStdout, "probe.publish.stdout", false, "Print reports on stdout instead of sending to app, for debugging")
 	flag.StringVar(&flags.probe.token, serviceTokenFlag, "", "Token to authenticate with cloud.weave.works")
 	flag.StringVar(&flags.probe.token, probeTokenFlag, "", "Token to authenticate with cloud.weave.works")
 	flag.StringVar(&flags.probe.httpListen, "probe.http.listen", "", "listen address for HTTP profiling and instrumentation server")