Refactor app for multitenancy

app/api_report.go

@@ -2,12 +2,13 @@ package app
 
 import (
 	"net/http"
+
+	"golang.org/x/net/context"
 )
 
 // Raw report handler
-func makeRawReportHandler(rep Reporter) func(http.ResponseWriter, *http.Request) {
-	return func(w http.ResponseWriter, r *http.Request) {
-		// r.ParseForm()
-		respondWith(w, http.StatusOK, rep.Report())
+func makeRawReportHandler(rep Reporter) CtxHandlerFunc {
+	return func(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+		respondWith(w, http.StatusOK, rep.Report(ctx))
 	}
 }

app/api_topologies.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"github.com/gorilla/mux"
+	"golang.org/x/net/context"
 
 	"github.com/weaveworks/scope/render"
 	"github.com/weaveworks/scope/report"
@@ -192,9 +193,9 @@ func (r *registry) walk(f func(APITopologyDesc)) {
 }
 
 // makeTopologyList returns a handler that yields an APITopologyList.
-func (r *registry) makeTopologyList(rep Reporter) func(w http.ResponseWriter, r *http.Request) {
-	return func(w http.ResponseWriter, req *http.Request) {
-		topologies := r.renderTopologies(rep.Report(), req)
+func (r *registry) makeTopologyList(rep Reporter) CtxHandlerFunc {
+	return func(ctx context.Context, w http.ResponseWriter, req *http.Request) {
+		topologies := r.renderTopologies(rep.Report(ctx), req)
 		respondWith(w, http.StatusOK, topologies)
 	}
 }
@@ -252,27 +253,27 @@ func renderedForRequest(r *http.Request, topology APITopologyDesc) render.Render
 	return renderer
 }
 
-type reportRenderHandler func(Reporter, render.Renderer, http.ResponseWriter, *http.Request)
+type reportRenderHandler func(context.Context, Reporter, render.Renderer, http.ResponseWriter, *http.Request)
 
-func (r *registry) captureRenderer(rep Reporter, f reportRenderHandler) http.HandlerFunc {
-	return func(w http.ResponseWriter, req *http.Request) {
+func (r *registry) captureRenderer(rep Reporter, f reportRenderHandler) CtxHandlerFunc {
+	return func(ctx context.Context, w http.ResponseWriter, req *http.Request) {
 		topology, ok := r.get(mux.Vars(req)["topology"])
 		if !ok {
 			http.NotFound(w, req)
 			return
 		}
 		renderer := renderedForRequest(req, topology)
-		f(rep, renderer, w, req)
+		f(ctx, rep, renderer, w, req)
 	}
 }
 
-func (r *registry) captureRendererWithoutFilters(rep Reporter, topologyID string, f reportRenderHandler) http.HandlerFunc {
-	return func(w http.ResponseWriter, req *http.Request) {
+func (r *registry) captureRendererWithoutFilters(rep Reporter, topologyID string, f reportRenderHandler) CtxHandlerFunc {
+	return func(ctx context.Context, w http.ResponseWriter, req *http.Request) {
 		topology, ok := r.get(topologyID)
 		if !ok {
 			http.NotFound(w, req)
 			return
 		}
-		f(rep, topology.renderer, w, req)
+		f(ctx, rep, topology.renderer, w, req)
 	}
 }

app/api_topology.go

@@ -6,6 +6,7 @@ import (
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/gorilla/websocket"
+	"golang.org/x/net/context"
 
 	"github.com/weaveworks/scope/common/xfer"
 	"github.com/weaveworks/scope/render"
@@ -28,14 +29,14 @@ type APINode struct {
 }
 
 // Full topology.
-func handleTopology(rep Reporter, renderer render.Renderer, w http.ResponseWriter, r *http.Request) {
+func handleTopology(ctx context.Context, rep Reporter, renderer render.Renderer, w http.ResponseWriter, r *http.Request) {
 	respondWith(w, http.StatusOK, APITopology{
-		Nodes: renderer.Render(rep.Report()).Prune(),
+		Nodes: renderer.Render(rep.Report(ctx)).Prune(),
 	})
 }
 
 // Websocket for the full topology. This route overlaps with the next.
-func handleWs(rep Reporter, renderer render.Renderer, w http.ResponseWriter, r *http.Request) {
+func handleWs(ctx context.Context, rep Reporter, renderer render.Renderer, w http.ResponseWriter, r *http.Request) {
 	if err := r.ParseForm(); err != nil {
 		respondWith(w, http.StatusInternalServerError, err.Error())
 		return
@@ -48,15 +49,15 @@ func handleWs(rep Reporter, renderer render.Renderer, w http.ResponseWriter, r *
 			return
 		}
 	}
-	handleWebsocket(w, r, rep, renderer, loop)
+	handleWebsocket(ctx, w, r, rep, renderer, loop)
 }
 
 // Individual nodes.
-func handleNode(nodeID string) func(Reporter, render.Renderer, http.ResponseWriter, *http.Request) {
-	return func(rep Reporter, renderer render.Renderer, w http.ResponseWriter, r *http.Request) {
+func handleNode(nodeID string) func(context.Context, Reporter, render.Renderer, http.ResponseWriter, *http.Request) {
+	return func(ctx context.Context, rep Reporter, renderer render.Renderer, w http.ResponseWriter, r *http.Request) {
 		var (
-			rpt      = rep.Report()
-			node, ok = renderer.Render(rep.Report())[nodeID]
+			rpt      = rep.Report(ctx)
+			node, ok = renderer.Render(rep.Report(ctx))[nodeID]
 		)
 		if !ok {
 			http.NotFound(w, r)
@@ -71,6 +72,7 @@ var upgrader = websocket.Upgrader{
 }
 
 func handleWebsocket(
+	ctx context.Context,
 	w http.ResponseWriter,
 	r *http.Request,
 	rep Reporter,
@@ -88,6 +90,7 @@ func handleWebsocket(
 	go func(c *websocket.Conn) {
 		for { // just discard everything the browser sends
 			if _, _, err := c.NextReader(); err != nil {
+				log.Println("err:", err)
 				close(quit)
 				break
 			}
@@ -99,15 +102,16 @@ func handleWebsocket(
 		tick         = time.Tick(loop)
 		wait         = make(chan struct{}, 1)
 	)
-	rep.WaitOn(wait)
-	defer rep.UnWait(wait)
+	rep.WaitOn(ctx, wait)
+	defer rep.UnWait(ctx, wait)
 
 	for {
-		newTopo := renderer.Render(rep.Report()).Prune()
+		newTopo := renderer.Render(rep.Report(ctx)).Prune()
 		diff := render.TopoDiff(previousTopo, newTopo)
 		previousTopo = newTopo
 
 		if err := conn.SetWriteDeadline(time.Now().Add(websocketTimeout)); err != nil {
+			log.Println("err:", err)
 			return
 		}
 

app/collector.go

@@ -6,22 +6,23 @@ import (
 	"time"
 
 	"github.com/spaolacci/murmur3"
+	"golang.org/x/net/context"
 
 	"github.com/weaveworks/scope/report"
 )
 
 // Reporter is something that can produce reports on demand. It's a convenient
 // interface for parts of the app, and several experimental components.
 type Reporter interface {
-	Report() report.Report
-	WaitOn(chan struct{})
-	UnWait(chan struct{})
+	Report(context.Context) report.Report
+	WaitOn(context.Context, chan struct{})
+	UnWait(context.Context, chan struct{})
 }
 
 // Adder is something that can accept reports. It's a convenient interface for
 // parts of the app, and several experimental components.
 type Adder interface {
-	Add(report.Report)
+	Add(context.Context, report.Report)
 }
 
 // A Collector is a Reporter and an Adder
@@ -45,13 +46,13 @@ type waitableCondition struct {
 	waiters map[chan struct{}]struct{}
 }
 
-func (wc *waitableCondition) WaitOn(waiter chan struct{}) {
+func (wc *waitableCondition) WaitOn(_ context.Context, waiter chan struct{}) {
 	wc.Lock()
 	wc.waiters[waiter] = struct{}{}
 	wc.Unlock()
 }
 
-func (wc *waitableCondition) UnWait(waiter chan struct{}) {
+func (wc *waitableCondition) UnWait(_ context.Context, waiter chan struct{}) {
 	wc.Lock()
 	delete(wc.waiters, waiter)
 	wc.Unlock()
@@ -82,7 +83,7 @@ func NewCollector(window time.Duration) Collector {
 var now = time.Now
 
 // Add adds a report to the collector's internal state. It implements Adder.
-func (c *collector) Add(rpt report.Report) {
+func (c *collector) Add(_ context.Context, rpt report.Report) {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 	c.reports = append(c.reports, timestampReport{now(), rpt})
@@ -95,7 +96,7 @@ func (c *collector) Add(rpt report.Report) {
 
 // Report returns a merged report over all added reports. It implements
 // Reporter.
-func (c *collector) Report() report.Report {
+func (c *collector) Report(_ context.Context) report.Report {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 

app/collector_test.go

@@ -4,13 +4,16 @@ import (
 	"testing"
 	"time"
 
+	"golang.org/x/net/context"
+
 	"github.com/weaveworks/scope/app"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/test"
 	"github.com/weaveworks/scope/test/reflect"
 )
 
 func TestCollector(t *testing.T) {
+	ctx := context.Background()
 	window := time.Millisecond
 	c := app.NewCollector(window)
 
@@ -20,32 +23,33 @@ func TestCollector(t *testing.T) {
 	r2 := report.MakeReport()
 	r2.Endpoint.AddNode("bar", report.MakeNode())
 
-	if want, have := report.MakeReport(), c.Report(); !reflect.DeepEqual(want, have) {
+	if want, have := report.MakeReport(), c.Report(ctx); !reflect.DeepEqual(want, have) {
 		t.Error(test.Diff(want, have))
 	}
 
-	c.Add(r1)
-	if want, have := r1, c.Report(); !reflect.DeepEqual(want, have) {
+	c.Add(ctx, r1)
+	if want, have := r1, c.Report(ctx); !reflect.DeepEqual(want, have) {
 		t.Error(test.Diff(want, have))
 	}
 
-	c.Add(r2)
+	c.Add(ctx, r2)
 
 	merged := report.MakeReport()
 	merged = merged.Merge(r1)
 	merged = merged.Merge(r2)
-	if want, have := merged, c.Report(); !reflect.DeepEqual(want, have) {
+	if want, have := merged, c.Report(ctx); !reflect.DeepEqual(want, have) {
 		t.Error(test.Diff(want, have))
 	}
 }
 
 func TestCollectorWait(t *testing.T) {
+	ctx := context.Background()
 	window := time.Millisecond
 	c := app.NewCollector(window)
 
 	waiter := make(chan struct{}, 1)
-	c.WaitOn(waiter)
-	defer c.UnWait(waiter)
+	c.WaitOn(ctx, waiter)
+	defer c.UnWait(ctx, waiter)
 	c.(interface {
 		Broadcast()
 	}).Broadcast()

app/control_router.go

@@ -0,0 +1,69 @@
+package app
+
+import (
+	"fmt"
+	"math/rand"
+	"sync"
+
+	"golang.org/x/net/context"
+
+	"github.com/weaveworks/scope/common/xfer"
+)
+
+// ControlRouter is a thing that can route control requests and responses
+// between the UI and a probe.
+type ControlRouter interface {
+	Handle(ctx context.Context, probeID string, req xfer.Request) (xfer.Response, error)
+	Register(ctx context.Context, probeID string, handler xfer.ControlHandlerFunc) (int64, error)
+	Deregister(ctx context.Context, probeID string, id int64) error
+}
+
+// NewLocalControlRouter creates a new ControlRouter that does everything
+// locally, in memory.
+func NewLocalControlRouter() ControlRouter {
+	return &localControlRouter{
+		probes: map[string]probe{},
+	}
+}
+
+type localControlRouter struct {
+	sync.Mutex
+	probes map[string]probe
+}
+
+type probe struct {
+	id      int64
+	handler xfer.ControlHandlerFunc
+}
+
+func (l *localControlRouter) Handle(_ context.Context, probeID string, req xfer.Request) (xfer.Response, error) {
+	l.Lock()
+	probe, ok := l.probes[probeID]
+	l.Unlock()
+	if !ok {
+		return xfer.Response{}, fmt.Errorf("Probe %s is not connected right now...", probeID)
+	}
+	return probe.handler(req), nil
+}
+
+func (l *localControlRouter) Register(_ context.Context, probeID string, handler xfer.ControlHandlerFunc) (int64, error) {
+	l.Lock()
+	defer l.Unlock()
+	id := rand.Int63()
+	l.probes[probeID] = probe{
+		id:      id,
+		handler: handler,
+	}
+	return id, nil
+}
+
+func (l *localControlRouter) Deregister(_ context.Context, probeID string, id int64) error {
+	l.Lock()
+	defer l.Unlock()
+	// NB probe might have reconnected in the mean time, need to ensure we do not
+	// delete new connection!  Also, it might have connected then deleted itself!
+	if l.probes[probeID].id == id {
+		delete(l.probes, probeID)
+	}
+	return nil
+}