Multi-tenant backends for app.

Add DynamoDB based collector - Store compressed reports in dynamodb Add SQS based control router. - Uses a queue per probe and a queue per UI for control requests & responses. Add Consul-based, horizontally-scalable, multi-tenant pipe router. - Uses consul to coordinate each end of pipe connections replicas of a pipe service.
2026-07-28 01:31:17 +00:00 · 2016-03-09 13:47:02 +00:00
parent b61c5c0255
commit 9885706402
14 changed files with 1432 additions and 31 deletions
--- a/app/multitenant/consul_pipe_router.go
+++ b/app/multitenant/consul_pipe_router.go
@@ -0,0 +1,605 @@
+package multitenant
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"sync"
+	"time"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/gorilla/mux"
+	"github.com/gorilla/websocket"
+	consul "github.com/hashicorp/consul/api"
+	"golang.org/x/net/context"
+
+	"github.com/weaveworks/scope/app"
+	"github.com/weaveworks/scope/common/mtime"
+	"github.com/weaveworks/scope/common/network"
+	"github.com/weaveworks/scope/common/xfer"
+)
+
+const (
+	gcInterval       = 30 * time.Second // we check all the pipes every 30s
+	pipeTimeout      = 1 * time.Minute  // pipes are closed when a client hasn't been connected for 1 minute
+	gcTimeout        = 10 * time.Minute // after another 10 minutes, tombstoned pipes are forgotten
+	longPollDuration = 10 * time.Second
+
+	privateAPIPort = 4444
+)
+
+var (
+	queryOptions = &consul.QueryOptions{
+		RequireConsistent: true,
+	}
+	writeOptions = &consul.WriteOptions{}
+	wsDialer     = &websocket.Dialer{}
+)
+
+// TODO deal with garbage collection
+type consulPipe struct {
+	CreatedAt, DeletedAt time.Time
+	UIEnd, ProbeEnd      string // Addrs where each end is connected
+	UIRef, ProbeRef      int    // Ref counts
+}
+
+func (c *consulPipe) toBytes() ([]byte, error) {
+	buf := bytes.Buffer{}
+	if err := json.NewEncoder(&buf).Encode(c); err != nil {
+		return nil, err
+	}
+	return buf.Bytes(), nil
+}
+
+func (c *consulPipe) fromBytes(bs []byte) error {
+	return json.NewDecoder(bytes.NewReader(bs)).Decode(c)
+}
+
+func (c *consulPipe) setEnd(e app.End, addr string) {
+	if e == app.UIEnd {
+		c.UIEnd = addr
+	} else {
+		c.ProbeEnd = addr
+	}
+}
+
+func (c *consulPipe) end(e app.End) string {
+	if e == app.UIEnd {
+		return c.UIEnd
+	}
+	return c.ProbeEnd
+}
+
+func (c *consulPipe) otherEnd(e app.End) string {
+	if e == app.UIEnd {
+		return c.ProbeEnd
+	}
+	return c.UIEnd
+}
+
+func (c *consulPipe) eitherEndFor(addr string) bool {
+	return c.end(app.UIEnd) == addr || c.end(app.ProbeEnd) == addr
+}
+
+func (c *consulPipe) incr(e app.End) int {
+	if e == app.UIEnd {
+		c.UIRef++
+		return c.UIRef
+	}
+	c.ProbeRef++
+	return c.ProbeRef
+}
+
+func (c *consulPipe) decr(e app.End) int {
+	if e == app.UIEnd {
+		c.UIRef--
+		return c.UIRef
+	}
+	c.ProbeRef--
+	return c.ProbeRef
+}
+
+type consulPipeRouter struct {
+	prefix    string
+	advertise string // Address of this pipe router to advertise in consul
+	client    *consul.Client
+	userIDer  UserIDer
+
+	mtx     sync.Mutex
+	cond    *sync.Cond
+	pipes   map[string]xfer.Pipe // Active pipes
+	bridges map[string]*bridgeConnection
+
+	// Used by Stop()
+	quit chan struct{}
+	wait sync.WaitGroup
+}
+
+// NewConsulPipeRouter returns a new consul based router
+func NewConsulPipeRouter(addr, prefix, inf string, userIDer UserIDer) (app.PipeRouter, error) {
+	advertise, err := network.GetFirstAddressOf(inf)
+	if err != nil {
+		return nil, err
+	}
+	client, err := consul.NewClient(&consul.Config{
+		Address: addr,
+		Scheme:  "http",
+	})
+	if err != nil {
+		return nil, err
+	}
+	pipeRouter := &consulPipeRouter{
+		prefix:    prefix,
+		advertise: advertise,
+		client:    client,
+		userIDer:  userIDer,
+
+		pipes:   map[string]xfer.Pipe{},
+		bridges: map[string]*bridgeConnection{},
+		quit:    make(chan struct{}),
+	}
+	pipeRouter.cond = sync.NewCond(&pipeRouter.mtx)
+	pipeRouter.wait.Add(1)
+	go pipeRouter.watchAll()
+	go pipeRouter.privateAPI()
+	return pipeRouter, nil
+}
+
+func (pr *consulPipeRouter) Stop() {
+	close(pr.quit)
+	pr.wait.Wait()
+}
+
+// watchAll listens to all pipe updates from consul.
+// This is effectively a distributed, consistent actor routine.
+// All state changes for this pipe router happen in this loop,
+// and all the methods are implemented as CAS's on consul, to
+// trigger an event in this loop.
+func (pr *consulPipeRouter) watchAll() {
+	var (
+		index = uint64(0)
+		kv    = pr.client.KV()
+	)
+	for {
+		select {
+		case <-pr.quit:
+			pr.wait.Done()
+			return
+		default:
+		}
+
+		kvps, meta, err := kv.List(pr.prefix, &consul.QueryOptions{
+			RequireConsistent: true,
+			WaitIndex:         index,
+			WaitTime:          longPollDuration,
+		})
+		if err != nil {
+			log.Errorf("Error getting path %s: %v", pr.prefix, err)
+			continue
+		}
+		if index == meta.LastIndex {
+			continue
+		}
+		index = meta.LastIndex
+
+		for _, kvp := range kvps {
+			//log.Infof("Got background update to %s (%d)", kvp.Key, index)
+
+			cp := consulPipe{}
+			if err := cp.fromBytes(kvp.Value); err != nil {
+				log.Errorf("Error deserialising pipe %s: %s", kvp.Key, err)
+				continue
+			}
+
+			pr.handlePipeUpdate(kvp.Key, cp)
+		}
+	}
+}
+
+func (pr *consulPipeRouter) handlePipeUpdate(key string, cp consulPipe) {
+	log.Infof("Got update to pipe %s", key)
+
+	// 1. If this pipe is closed, or we're not one of the ends, we
+	//    should ensure our local pipe (and bridge) is closed.
+	if !cp.DeletedAt.IsZero() || !cp.eitherEndFor(pr.advertise) {
+		log.Infof("Pipe %s not in use on this node.", key)
+
+		pr.mtx.Lock()
+		pipe, ok := pr.pipes[key]
+		delete(pr.pipes, key)
+		pr.mtx.Unlock()
+
+		// These could block, so don't hold the locks.
+		if ok {
+			pipe.Close()
+		}
+
+		bridge, ok := pr.bridges[key]
+		delete(pr.bridges, key)
+		if ok {
+			bridge.stop()
+		}
+		return
+	}
+
+	if !cp.eitherEndFor(pr.advertise) {
+		return
+	}
+
+	// 2. If this pipe if for us, we should have a pipe for it.
+	pr.mtx.Lock()
+	pipe, ok := pr.pipes[key]
+	if !ok {
+		pipe = xfer.NewPipe()
+		pr.pipes[key] = pipe
+	}
+	pr.mtx.Unlock()
+
+	// 3. Ensure there is a bridging connection for this pipe.
+	//    Semantics are the owner of the UIEnd connects to the owner of the ProbeEnd
+	//  NB no need to hold lock for pr.bridged, this is the only place we use it.
+	shouldBridge := cp.DeletedAt.IsZero() &&
+		cp.end(app.UIEnd) != cp.end(app.ProbeEnd) &&
+		cp.end(app.UIEnd) == pr.advertise &&
+		cp.end(app.ProbeEnd) != ""
+	bridge, ok := pr.bridges[key]
+
+	// If we shouldn't be bridging but are, or we should be bridging but are pointing
+	// at the wrong place, stop the current bridge.
+	if (!shouldBridge && ok) || (shouldBridge && ok && bridge.addr != cp.end(app.ProbeEnd)) {
+		log.Infof("Stopping bridge connection for %s", key)
+		delete(pr.bridges, key)
+		bridge.stop()
+		ok = false
+	}
+
+	// If we should be bridging and are not, start a new bridge
+	if shouldBridge && !ok {
+		log.Infof("Starting bridge connection for %s", key)
+		bridge = newBridgeConnection(key, cp.end(app.ProbeEnd), pipe)
+		pr.bridges[key] = bridge
+	}
+}
+
+func (pr *consulPipeRouter) privateAPI() {
+	router := mux.NewRouter()
+	router.Methods("GET").
+		MatcherFunc(app.URLMatcher("/private/api/pipe/{key}")).
+		HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			key := mux.Vars(r)["key"]
+			pr.mtx.Lock()
+			pipe, ok := pr.pipes[key]
+			pr.mtx.Unlock()
+			if !ok {
+				http.NotFound(w, r)
+				return
+			}
+
+			conn, err := xfer.Upgrade(w, r, nil)
+			if err != nil {
+				log.Errorf("Error upgrading pipe %s websocket: %v", key, err)
+				return
+			}
+			defer conn.Close()
+
+			end, _ := pipe.Ends()
+			if err := pipe.CopyToWebsocket(end, conn); err != nil && !xfer.IsExpectedWSCloseError(err) {
+				log.Printf("Error copying to pipe %s websocket: %v", key, err)
+			}
+		})
+
+	addr := fmt.Sprintf("%s:%d", pr.advertise, privateAPIPort)
+	log.Infof("Serving private API on endpoint %s.", addr)
+	log.Infof("Private API terminated: %v", http.ListenAndServe(addr, router))
+}
+
+// Atomically modify a pipe in a callback.
+// If pipe doesn't exist you'll get nil in callback.
+func (pr *consulPipeRouter) get(key string) (*consulPipe, error) {
+	var (
+		kv   = pr.client.KV()
+		pipe consulPipe
+	)
+	kvp, _, err := kv.Get(key, queryOptions)
+	if err != nil {
+		return nil, err
+	}
+	if kvp == nil {
+		return nil, nil
+	}
+	if err := pipe.fromBytes(kvp.Value); err != nil {
+		return nil, err
+	}
+	return &pipe, nil
+}
+
+// Atomically modify a pipe in a callback.
+// If pipe doesn't exist you'll get nil in callback.
+func (pr *consulPipeRouter) cas(key string, f func(*consulPipe) (*consulPipe, bool, error)) (*consulPipe, error) {
+	var (
+		index   = uint64(0)
+		kv      = pr.client.KV()
+		pipe    *consulPipe
+		retries = 10
+		retry   = true
+	)
+	for i := 0; i < retries; i++ {
+		kvp, _, err := kv.Get(key, queryOptions)
+		if err != nil {
+			log.Errorf("Error getting %s: %v", key, err)
+			continue
+		}
+		if kvp != nil {
+			pipe = &consulPipe{}
+			if err := pipe.fromBytes(kvp.Value); err != nil {
+				log.Errorf("Error deserialising pipe %s: %v", key, err)
+				continue
+			}
+			index = kvp.ModifyIndex // if it doesn't exist, it will be 0
+		}
+
+		if pipe, retry, err = f(pipe); err != nil {
+			log.Errorf("Error CASing pipe %s: %v", key, err)
+			if !retry {
+				return nil, err
+			}
+			continue
+		}
+
+		if pipe == nil {
+			panic("Callback must instantiate pipe!")
+		}
+
+		value, err := pipe.toBytes()
+		if err != nil {
+			log.Errorf("Error serialising pipe %s: %v", key, err)
+			continue
+		}
+		ok, _, err := kv.CAS(&consul.KVPair{
+			Key:         key,
+			Value:       value,
+			ModifyIndex: index,
+		}, writeOptions)
+		if err != nil {
+			log.Errorf("Error CASing pipe %s: %v", key, err)
+			continue
+		}
+		if !ok {
+			log.Errorf("Error CASing pipe %s, trying again %d", key, index)
+			continue
+		}
+		return pipe, nil
+	}
+	return nil, fmt.Errorf("Failed to aquire pipe")
+}
+
+// Watch a given pipe and trigger a callback when it changes.
+// if callback returns false or error, exit (with the error).
+func (pr *consulPipeRouter) watch(key string, deadline time.Time, f func(*consulPipe) (bool, error)) (*consulPipe, error) {
+	var (
+		index = uint64(0)
+		kv    = pr.client.KV()
+	)
+	for deadline.After(mtime.Now()) {
+		// Poll waiting for the entry to get updated
+		kvp, meta, err := kv.Get(key, &consul.QueryOptions{
+			RequireConsistent: true,
+			WaitIndex:         index,
+			WaitTime:          longPollDuration,
+		})
+		if err != nil {
+			return nil, fmt.Errorf("Error getting %s: %v", key, err)
+		}
+		if kvp == nil {
+			return nil, fmt.Errorf("Pipe %s unexpectedly deleted!", key)
+		}
+
+		pipe := &consulPipe{}
+		pipe.fromBytes(kvp.Value)
+		if ok, err := f(pipe); !ok {
+			return pipe, nil
+		} else if err != nil {
+			return pipe, err
+		}
+
+		index = meta.LastIndex
+	}
+	return nil, fmt.Errorf("Timed out waiting on %s", key)
+}
+
+// negotiate tries to ensure the given end of the given pipe
+// is 'owned' by this pipe service replica in consul.
+func (pr *consulPipeRouter) negotiate(key string, e app.End) (*consulPipe, error) {
+	// First try and establish a record with us owning one end
+	_, err := pr.cas(key, func(p *consulPipe) (*consulPipe, bool, error) {
+		if p == nil {
+			p = &consulPipe{
+				CreatedAt: mtime.Now(),
+			}
+		}
+		if !p.DeletedAt.IsZero() {
+			return nil, false, fmt.Errorf("Pipe %s has been deleted", key)
+		}
+		end := p.end(e)
+		if end != "" && end != pr.advertise {
+			return nil, true, fmt.Errorf("Error: Pipe %s has existing connection to %s", key, end)
+		}
+		p.setEnd(e, pr.advertise)
+		p.incr(e)
+		return p, false, nil
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// Next wait for other end to connect
+	// at this point we 'own' one end
+	pipe, err := pr.watch(key, mtime.Now().Add(10*longPollDuration), func(p *consulPipe) (bool, error) {
+		return p.otherEnd(e) == "", nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return pipe, nil
+}
+
+func (pr *consulPipeRouter) Exists(ctx context.Context, id string) (bool, error) {
+	userID, err := pr.userIDer(ctx)
+	if err != nil {
+		return false, err
+	}
+	key := fmt.Sprintf("%s%s-%s", pr.prefix, userID, id)
+	consulPipe, err := pr.get(key)
+	if err != nil {
+		return false, err
+	}
+	return consulPipe == nil || consulPipe.DeletedAt.IsZero(), nil
+}
+
+func (pr *consulPipeRouter) Get(ctx context.Context, id string, e app.End) (xfer.Pipe, io.ReadWriter, error) {
+	userID, err := pr.userIDer(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+	key := fmt.Sprintf("%s%s-%s", pr.prefix, userID, id)
+	log.Infof("Get %s:%s", key, e)
+
+	// first check we own the pipe in consul
+	_, err = pr.negotiate(key, e)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// next see if we already have a active pipe
+	pr.mtx.Lock()
+	pipe, ok := pr.pipes[key]
+	if !ok {
+		pipe = xfer.NewPipe()
+		pr.pipes[key] = pipe
+	}
+	pr.mtx.Unlock()
+
+	myEnd, _ := pipe.Ends()
+	if e == app.ProbeEnd {
+		_, myEnd = pipe.Ends()
+	}
+	return pipe, myEnd, nil
+}
+
+func (pr *consulPipeRouter) Release(ctx context.Context, id string, e app.End) error {
+	userID, err := pr.userIDer(ctx)
+	if err != nil {
+		return err
+	}
+	key := fmt.Sprintf("%s%s-%s", pr.prefix, userID, id)
+	log.Infof("Release %s:%s", key, e)
+
+	// atomically clear my end of the pipe in consul
+	_, err = pr.cas(key, func(p *consulPipe) (*consulPipe, bool, error) {
+		if p == nil {
+			return nil, false, fmt.Errorf("Pipe %s not found", id)
+		}
+		if p.end(e) != pr.advertise {
+			return nil, false, fmt.Errorf("Pipe %s not owned by us!", id)
+		}
+		refs := p.decr(e)
+		if refs == 0 {
+			p.setEnd(e, "")
+		}
+		return p, true, nil
+	})
+	return err
+}
+
+func (pr *consulPipeRouter) Delete(ctx context.Context, id string) error {
+	userID, err := pr.userIDer(ctx)
+	if err != nil {
+		return err
+	}
+	key := fmt.Sprintf("%s%s-%s", pr.prefix, userID, id)
+	log.Infof("Delete %s", key)
+
+	_, err = pr.cas(key, func(p *consulPipe) (*consulPipe, bool, error) {
+		if p == nil {
+			return nil, false, fmt.Errorf("Pipe %s not found", id)
+		}
+		p.DeletedAt = mtime.Now()
+		return p, false, nil
+	})
+	return err
+}
+
+type bridgeConnection struct {
+	key  string
+	addr string // address to connect to
+	pipe xfer.Pipe
+
+	mtx     sync.Mutex
+	conn    xfer.Websocket
+	stopped bool
+	wait    sync.WaitGroup
+}
+
+func newBridgeConnection(key, addr string, pipe xfer.Pipe) *bridgeConnection {
+	result := &bridgeConnection{
+		key:  key,
+		addr: addr,
+		pipe: pipe,
+	}
+	result.wait.Add(1)
+	go result.loop()
+	return result
+}
+
+func (bc *bridgeConnection) stop() {
+	bc.mtx.Lock()
+	bc.stopped = true
+	if bc.conn != nil {
+		bc.conn.Close()
+	}
+	bc.mtx.Unlock()
+	bc.wait.Wait()
+}
+
+func (bc *bridgeConnection) loop() {
+	log.Infof("Making bridge connection for pipe %s to %s", bc.key, bc.addr)
+	defer bc.wait.Done()
+	defer log.Infof("Stopping bridge connection for pipe %s to %s", bc.key, bc.addr)
+
+	_, end := bc.pipe.Ends()
+	url := fmt.Sprintf("ws://%s:%d/private/api/pipe/%s", bc.addr, privateAPIPort, url.QueryEscape(bc.key))
+
+	for {
+		bc.mtx.Lock()
+		bc.conn = nil
+		if bc.stopped {
+			bc.mtx.Unlock()
+			return
+		}
+		bc.mtx.Unlock()
+
+		// connect to other pipes instance
+		conn, _, err := xfer.DialWS(wsDialer, url, http.Header{})
+		if err != nil {
+			log.Errorf("Error connecting to %s: %v", url, err)
+			time.Sleep(time.Second) // TODO backoff
+			continue
+		}
+
+		bc.mtx.Lock()
+		if bc.stopped {
+			bc.mtx.Unlock()
+			conn.Close()
+			return
+		}
+		bc.conn = conn
+		bc.mtx.Unlock()
+
+		if err := bc.pipe.CopyToWebsocket(end, conn); err != nil && !xfer.IsExpectedWSCloseError(err) {
+			log.Printf("Error copying to pipe %s websocket: %v", bc.key, err)
+		}
+	}
+}
--- a/app/multitenant/dynamo_collector.go
+++ b/app/multitenant/dynamo_collector.go
@@ -0,0 +1,211 @@
+package multitenant
+
+import (
+	"bytes"
+	"compress/gzip"
+	"fmt"
+	"strconv"
+	"time"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/ugorji/go/codec"
+	"golang.org/x/net/context"
+
+	"github.com/weaveworks/scope/app"
+	"github.com/weaveworks/scope/report"
+)
+
+const (
+	tableName   = "reports"
+	hourField   = "hour"
+	tsField     = "ts"
+	reportField = "report"
+)
+
+// DynamoDBCollector is a Collector which can also CreateTables
+type DynamoDBCollector interface {
+	app.Collector
+	CreateTables() error
+}
+
+type dynamoDBCollector struct {
+	userIDer UserIDer
+	db       *dynamodb.DynamoDB
+}
+
+// NewDynamoDBCollector the reaper of souls
+// https://github.com/aws/aws-sdk-go/wiki/common-examples
+func NewDynamoDBCollector(url, region string, creds *credentials.Credentials, userIDer UserIDer) DynamoDBCollector {
+	return &dynamoDBCollector{
+		db: dynamodb.New(session.New(aws.NewConfig().
+			WithEndpoint(url).
+			WithRegion(region).
+			WithCredentials(creds))),
+		userIDer: userIDer,
+	}
+}
+
+// CreateDynamoDBTables creates the required tables in dynamodb
+func (c *dynamoDBCollector) CreateTables() error {
+	// see if tableName exists
+	resp, err := c.db.ListTables(&dynamodb.ListTablesInput{
+		Limit: aws.Int64(10),
+	})
+	if err != nil {
+		return err
+	}
+	for _, s := range resp.TableNames {
+		if *s == tableName {
+			return nil
+		}
+	}
+
+	params := &dynamodb.CreateTableInput{
+		TableName: aws.String(tableName),
+		AttributeDefinitions: []*dynamodb.AttributeDefinition{
+			{
+				AttributeName: aws.String(hourField),
+				AttributeType: aws.String("N"),
+			},
+			{
+				AttributeName: aws.String(tsField),
+				AttributeType: aws.String("N"),
+			},
+			// Don't need to specify non-key attributes in schema
+			//{
+			//	AttributeName: aws.String(reportField),
+			//	AttributeType: aws.String("B"),
+			//},
+		},
+		KeySchema: []*dynamodb.KeySchemaElement{
+			{
+				AttributeName: aws.String(hourField),
+				KeyType:       aws.String("HASH"),
+			},
+			{
+				AttributeName: aws.String(tsField),
+				KeyType:       aws.String("RANGE"),
+			},
+		},
+		ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
+			ReadCapacityUnits:  aws.Int64(10),
+			WriteCapacityUnits: aws.Int64(5),
+		},
+	}
+	log.Infof("Creating table %s", tableName)
+	_, err = c.db.CreateTable(params)
+	return err
+}
+
+func (c *dynamoDBCollector) getRows(userid string, row int64, start, end time.Time, input report.Report) (report.Report, error) {
+	rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(row, 10))
+	resp, err := c.db.Query(&dynamodb.QueryInput{
+		TableName: aws.String(tableName),
+		KeyConditions: map[string]*dynamodb.Condition{
+			hourField: {
+				AttributeValueList: []*dynamodb.AttributeValue{
+					{N: aws.String(rowKey)},
+				},
+				ComparisonOperator: aws.String("EQ"),
+			},
+			tsField: {
+				AttributeValueList: []*dynamodb.AttributeValue{
+					{N: aws.String(strconv.FormatInt(start.UnixNano(), 10))},
+					{N: aws.String(strconv.FormatInt(end.UnixNano(), 10))},
+				},
+				ComparisonOperator: aws.String("BETWEEN"),
+			},
+		},
+	})
+	if err != nil {
+		return report.MakeReport(), err
+	}
+	result := input
+	for _, item := range resp.Items {
+		b := item[reportField].B
+		if b == nil {
+			log.Errorf("Empty row!")
+			continue
+		}
+		buf := bytes.NewBuffer(b)
+		reader, err := gzip.NewReader(buf)
+		if err != nil {
+			log.Errorf("Error gunzipping report: %v", err)
+			continue
+		}
+		rep := report.MakeReport()
+		if err := codec.NewDecoder(reader, &codec.MsgpackHandle{}).Decode(&rep); err != nil {
+			log.Errorf("Failed to decode report: %v", err)
+			continue
+		}
+		result = result.Merge(rep)
+	}
+	return result, nil
+}
+
+func (c *dynamoDBCollector) Report(ctx context.Context) (report.Report, error) {
+	var (
+		now              = time.Now()
+		start            = now.Add(-15 * time.Second)
+		rowStart, rowEnd = start.UnixNano() / time.Hour.Nanoseconds(), now.UnixNano() / time.Hour.Nanoseconds()
+		result           = report.MakeReport()
+		userid, err      = c.userIDer(ctx)
+	)
+	if err != nil {
+		return report.MakeReport(), err
+	}
+
+	// Queries will only every span 2 rows max.
+	if rowStart != rowEnd {
+		if result, err = c.getRows(userid, rowStart, start, now, result); err != nil {
+			return report.MakeReport(), nil
+		}
+	}
+	if result, err = c.getRows(userid, rowEnd, start, now, result); err != nil {
+		return report.MakeReport(), nil
+	}
+	return result, nil
+}
+
+func (c *dynamoDBCollector) Add(ctx context.Context, rep report.Report) error {
+	userid, err := c.userIDer(ctx)
+	if err != nil {
+		return err
+	}
+
+	var buf bytes.Buffer
+	writer := gzip.NewWriter(&buf)
+	if err := codec.NewEncoder(writer, &codec.MsgpackHandle{}).Encode(&rep); err != nil {
+		return err
+	}
+	writer.Close()
+
+	now := time.Now()
+	rowKey := fmt.Sprintf("%s-%s", userid, strconv.FormatInt(now.UnixNano()/time.Hour.Nanoseconds(), 10))
+	_, err = c.db.PutItem(&dynamodb.PutItemInput{
+		TableName: aws.String(tableName),
+		Item: map[string]*dynamodb.AttributeValue{
+			hourField: {
+				N: aws.String(rowKey),
+			},
+			tsField: {
+				N: aws.String(strconv.FormatInt(now.UnixNano(), 10)),
+			},
+			reportField: {
+				B: buf.Bytes(),
+			},
+		},
+	})
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *dynamoDBCollector) WaitOn(context.Context, chan struct{}) {}
+
+func (c *dynamoDBCollector) UnWait(context.Context, chan struct{}) {}
--- a/app/multitenant/sqs_control_router.go
+++ b/app/multitenant/sqs_control_router.go
@@ -0,0 +1,330 @@
+package multitenant
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"math/rand"
+	"sync"
+	"time"
+
+	log "github.com/Sirupsen/logrus"
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/sqs"
+	"golang.org/x/net/context"
+
+	"github.com/weaveworks/scope/app"
+	"github.com/weaveworks/scope/common/xfer"
+)
+
+var (
+	longPollTime = aws.Int64(10)
+	rpcTimeout   = time.Minute
+)
+
+// sqsControlRouter:
+// Creates a queue for every probe that connects to it, and a queue for
+// responses back to it.  When it recieves a request, posts it to the
+// probe queue.  When probe recieves a request, handles it and posts the
+// response back to the response queue.
+type sqsControlRouter struct {
+	service  *sqs.SQS
+	queueURL *string
+	userIDer UserIDer
+
+	mtx          sync.Mutex
+	responses    map[string]chan xfer.Response
+	probeWorkers map[int64]*probeWorker
+}
+
+type sqsRequestMessage struct {
+	ID               string
+	Request          xfer.Request
+	ResponseQueueURL string
+}
+
+type sqsResponseMessage struct {
+	ID       string
+	Response xfer.Response
+}
+
+// NewSQSControlRouter the harbinger of death
+func NewSQSControlRouter(url, region string, creds *credentials.Credentials, userIDer UserIDer) app.ControlRouter {
+	result := &sqsControlRouter{
+		service: sqs.New(session.New(aws.NewConfig().
+			WithEndpoint(url).
+			WithRegion(region).
+			WithCredentials(creds))),
+		queueURL:     nil,
+		userIDer:     userIDer,
+		responses:    map[string]chan xfer.Response{},
+		probeWorkers: map[int64]*probeWorker{},
+	}
+	go result.loop()
+	return result
+}
+
+func (cr *sqsControlRouter) Stop() error {
+	return nil
+}
+
+func (cr *sqsControlRouter) setQueueURL(url *string) {
+	cr.mtx.Lock()
+	defer cr.mtx.Unlock()
+	cr.queueURL = url
+}
+
+func (cr *sqsControlRouter) getQueueURL() *string {
+	cr.mtx.Lock()
+	defer cr.mtx.Unlock()
+	return cr.queueURL
+}
+
+func (cr *sqsControlRouter) getOrCreateQueue(name string) (*string, error) {
+	getQueueURLRes, err := cr.service.GetQueueUrl(&sqs.GetQueueUrlInput{
+		QueueName: aws.String(name),
+	})
+	if err == nil {
+		return getQueueURLRes.QueueUrl, nil
+	}
+
+	createQueueRes, err := cr.service.CreateQueue(&sqs.CreateQueueInput{
+		QueueName: aws.String(name),
+	})
+	if err != nil {
+		return nil, err
+	}
+	return createQueueRes.QueueUrl, nil
+}
+
+func (cr *sqsControlRouter) loop() {
+	for {
+		// This app has a random id and uses this as a return path for all responses from probes.
+		name := fmt.Sprintf("control-app-%d", rand.Int63())
+		queueURL, err := cr.getOrCreateQueue(name)
+		if err != nil {
+			log.Errorf("Failed to create queue: %v", err)
+			time.Sleep(1 * time.Second)
+			continue
+		}
+		cr.setQueueURL(queueURL)
+		break
+	}
+
+	for {
+		res, err := cr.service.ReceiveMessage(&sqs.ReceiveMessageInput{
+			QueueUrl:        cr.queueURL,
+			WaitTimeSeconds: longPollTime,
+		})
+		if err != nil {
+			log.Errorf("Error recieving message from %s: %v", *cr.queueURL, err)
+			continue
+		}
+		if len(res.Messages) == 0 {
+			continue
+		}
+		cr.handleResponses(res)
+		if err := cr.deleteMessages(cr.queueURL, res.Messages); err != nil {
+			log.Errorf("Error deleting message from %s: %v", *cr.queueURL, err)
+		}
+	}
+}
+
+func (cr *sqsControlRouter) deleteMessages(queueURL *string, messages []*sqs.Message) error {
+	entries := []*sqs.DeleteMessageBatchRequestEntry{}
+	for _, message := range messages {
+		entries = append(entries, &sqs.DeleteMessageBatchRequestEntry{
+			ReceiptHandle: message.ReceiptHandle,
+			Id:            message.MessageId,
+		})
+	}
+	_, err := cr.service.DeleteMessageBatch(&sqs.DeleteMessageBatchInput{
+		QueueUrl: queueURL,
+		Entries:  entries,
+	})
+	return err
+}
+
+func (cr *sqsControlRouter) handleResponses(res *sqs.ReceiveMessageOutput) {
+	sqsResponses := []sqsResponseMessage{}
+	for _, message := range res.Messages {
+		var sqsResponse sqsResponseMessage
+		if err := json.NewDecoder(bytes.NewBufferString(*message.Body)).Decode(&sqsResponse); err != nil {
+			log.Errorf("Error decoding message: %v", err)
+			continue
+		}
+		sqsResponses = append(sqsResponses, sqsResponse)
+	}
+
+	for _, sqsResponse := range sqsResponses {
+		cr.mtx.Lock()
+		waiter, ok := cr.responses[sqsResponse.ID]
+		cr.mtx.Unlock()
+
+		if !ok {
+			log.Errorf("Dropping response %s - no one waiting for it!", sqsResponse.ID)
+			continue
+		}
+		waiter <- sqsResponse.Response
+	}
+}
+
+func (cr *sqsControlRouter) sendMessage(queueURL *string, message interface{}) error {
+	buf := bytes.Buffer{}
+	if err := json.NewEncoder(&buf).Encode(message); err != nil {
+		return err
+	}
+	log.Infof("sendMessage to %s: %s", *queueURL, buf.String())
+	_, err := cr.service.SendMessage(&sqs.SendMessageInput{
+		QueueUrl:    queueURL,
+		MessageBody: aws.String(buf.String()),
+	})
+	return err
+}
+
+func (cr *sqsControlRouter) Handle(ctx context.Context, probeID string, req xfer.Request) (xfer.Response, error) {
+	// Make sure we know the users
+	userID, err := cr.userIDer(ctx)
+	if err != nil {
+		return xfer.Response{}, err
+	}
+
+	// Get the queue url for the local (control app) queue, and for the probe.
+	queueURL := cr.getQueueURL()
+	if queueURL == nil {
+		return xfer.Response{}, fmt.Errorf("No SQS queue yet!")
+	}
+	probeQueueName := fmt.Sprintf("probe-%s-%s", userID, probeID)
+	probeQueueURL, err := cr.service.GetQueueUrl(&sqs.GetQueueUrlInput{
+		QueueName: aws.String(probeQueueName),
+	})
+	if err != nil {
+		return xfer.Response{}, err
+	}
+
+	// Wait for a response befor we send the request, to prevent races
+	id := fmt.Sprintf("request-%s-%d", userID, rand.Int63())
+	waiter := make(chan xfer.Response, 1)
+	cr.mtx.Lock()
+	cr.responses[id] = waiter
+	cr.mtx.Unlock()
+	defer func() {
+		cr.mtx.Lock()
+		delete(cr.responses, id)
+		cr.mtx.Unlock()
+	}()
+
+	// Next, send the request to that queue
+	if err := cr.sendMessage(probeQueueURL.QueueUrl, sqsRequestMessage{
+		ID:               id,
+		Request:          req,
+		ResponseQueueURL: *queueURL,
+	}); err != nil {
+		return xfer.Response{}, err
+	}
+
+	// Finally, wait for a response on our queue
+	select {
+	case response := <-waiter:
+		return response, nil
+	case <-time.After(rpcTimeout):
+		return xfer.Response{}, fmt.Errorf("Request timedout.")
+	}
+}
+
+func (cr *sqsControlRouter) Register(ctx context.Context, probeID string, handler xfer.ControlHandlerFunc) (int64, error) {
+	userID, err := cr.userIDer(ctx)
+	if err != nil {
+		return 0, err
+	}
+
+	name := fmt.Sprintf("probe-%s-%s", userID, probeID)
+	queueURL, err := cr.getOrCreateQueue(name)
+	if err != nil {
+		return 0, err
+	}
+
+	pwID := rand.Int63()
+	pw := &probeWorker{
+		router:   cr,
+		queueURL: queueURL,
+		handler:  handler,
+		quit:     make(chan struct{}),
+	}
+	pw.done.Add(1)
+	go pw.loop()
+
+	cr.mtx.Lock()
+	defer cr.mtx.Unlock()
+	cr.probeWorkers[pwID] = pw
+	return pwID, nil
+}
+
+func (cr *sqsControlRouter) Deregister(_ context.Context, probeID string, id int64) error {
+	cr.mtx.Lock()
+	pw, ok := cr.probeWorkers[id]
+	delete(cr.probeWorkers, id)
+	cr.mtx.Unlock()
+	if ok {
+		pw.stop()
+	}
+	return nil
+}
+
+type probeWorker struct {
+	router   *sqsControlRouter
+	queueURL *string
+	handler  xfer.ControlHandlerFunc
+	quit     chan struct{}
+	done     sync.WaitGroup
+}
+
+func (pw *probeWorker) stop() {
+	close(pw.quit)
+	pw.done.Wait()
+}
+
+func (pw *probeWorker) loop() {
+	for {
+		// have we been stopped?
+		select {
+		case <-pw.quit:
+			return
+		default:
+		}
+
+		res, err := pw.router.service.ReceiveMessage(&sqs.ReceiveMessageInput{
+			QueueUrl:        pw.queueURL,
+			WaitTimeSeconds: longPollTime,
+		})
+		if err != nil {
+			log.Errorf("Error recieving message: %v", err)
+			continue
+		}
+		if len(res.Messages) == 0 {
+			continue
+		}
+
+		// TODO do we need to parallelise the handling of requests?
+		for _, message := range res.Messages {
+			var sqsRequest sqsRequestMessage
+			if err := json.NewDecoder(bytes.NewBufferString(*message.Body)).Decode(&sqsRequest); err != nil {
+				log.Errorf("Error decoding message from: %v", err)
+				continue
+			}
+
+			if err := pw.router.sendMessage(&sqsRequest.ResponseQueueURL, sqsResponseMessage{
+				ID:       sqsRequest.ID,
+				Response: pw.handler(sqsRequest.Request),
+			}); err != nil {
+				log.Errorf("Error sending response: %v", err)
+			}
+		}
+
+		if err := pw.router.deleteMessages(pw.queueURL, res.Messages); err != nil {
+			log.Errorf("Error deleting message from %s: %v", *pw.queueURL, err)
+		}
+	}
+}
--- a/app/multitenant/user.go
+++ b/app/multitenant/user.go
@@ -0,0 +1,37 @@
+package multitenant
+
+import (
+	"fmt"
+	"net/http"
+
+	"golang.org/x/net/context"
+
+	"github.com/weaveworks/scope/app"
+)
+
+// ErrNotFound should be returned by a UserIDer when it fails to ID the
+// user for a request.
+var ErrNotFound = fmt.Errorf("User ID not found")
+
+// UserIDer identifies users given a request context.
+type UserIDer func(context.Context) (string, error)
+
+// UserIDHeader returns a UserIDer which a header by the supplied key.
+func UserIDHeader(headerName string) UserIDer {
+	return func(ctx context.Context) (string, error) {
+		request, ok := ctx.Value(app.RequestCtxKey).(*http.Request)
+		if !ok || request == nil {
+			return "", ErrNotFound
+		}
+		userID := request.Header.Get(headerName)
+		if userID == "" {
+			return "", ErrNotFound
+		}
+		return userID, nil
+	}
+}
+
+// NoopUserIDer always returns the empty user ID.
+func NoopUserIDer(context.Context) (string, error) {
+	return "", nil
+}
--- a/common/network/interface.go
+++ b/common/network/interface.go
@@ -0,0 +1,29 @@
+package network
+
+import (
+	"fmt"
+	"net"
+)
+
+// GetFirstAddressOf returns the first address of the supplied interface name.
+func GetFirstAddressOf(name string) (string, error) {
+	inf, err := net.InterfaceByName(name)
+	if err != nil {
+		return "", err
+	}
+
+	addrs, err := inf.Addrs()
+	if err != nil {
+		return "", err
+	}
+	if len(addrs) <= 0 {
+		return "", fmt.Errorf("No address found for %s", name)
+	}
+
+	switch v := addrs[0].(type) {
+	case *net.IPNet:
+		return v.IP.String(), nil
+	default:
+		return "", fmt.Errorf("No address found for %s", name)
+	}
+}
--- a/common/xfer/pipes.go
+++ b/common/xfer/pipes.go
@@ -93,7 +93,7 @@ func (p *pipe) CopyToWebsocket(end io.ReadWriter, conn Websocket) error {
 	p.mtx.Unlock()
 	defer p.wg.Done()

-	errors := make(chan error, 1)
+	errors := make(chan error, 2)

 	// Read-from-UI loop
 	go func() {
--- a/experimental/multitenant/.gitignore
+++ b/experimental/multitenant/.gitignore
@@ -0,0 +1,4 @@
+collection/collection
+query/query
+control/control
+static/static
--- a/experimental/multitenant/Makefile
+++ b/experimental/multitenant/Makefile
@@ -0,0 +1,11 @@
+BUILD_IN_CONTAINER=true
+
+all: .frontend.uptodate
+
+.frontend.uptodate: frontend/*
+	docker build -t weaveworks/scope-frontend frontend/
+	touch $@
+
+clean:
+	go clean ./..
+	rm -f .*.uptodate
--- a/experimental/multitenant/frontend/Dockerfile
+++ b/experimental/multitenant/frontend/Dockerfile
@@ -0,0 +1,12 @@
+FROM ubuntu
+MAINTAINER Weaveworks Inc <help@weave.works>
+RUN apt-get update && \
+  apt-get install -y nginx && \
+  rm -rf /var/lib/apt/lists/*
+RUN rm /etc/nginx/sites-available/default && \
+  ln -sf /dev/stdout /var/log/nginx/access.log && \
+  ln -sf /dev/stderr /var/log/nginx/error.log
+COPY default.conf /etc/nginx/conf.d/
+COPY api.json /home/weave/
+EXPOSE 80
+CMD ["nginx", "-g", "daemon off;"]
--- a/experimental/multitenant/frontend/api.json
+++ b/experimental/multitenant/frontend/api.json
@@ -0,0 +1 @@
+{"id": "sccopeservice", "version": "Multimagic"}
--- a/experimental/multitenant/frontend/default.conf
+++ b/experimental/multitenant/frontend/default.conf
@@ -0,0 +1,51 @@
+server {
+    # Listen on port 80 insecurely until the probes all support https
+    listen 80;
+    resolver dns.weave.local;
+
+    # topology (from UI) websockets
+    location ~ ^/api/topology/[^/]+/ws {
+        proxy_pass http://query.weave.local$request_uri;
+        proxy_http_version 1.1;
+        proxy_set_header Upgrade $http_upgrade;
+        proxy_set_header Connection "upgrade";
+    }
+
+    # control (from probe) websockets
+    location = /api/control/ws {
+        proxy_pass http://controls.weave.local$request_uri;
+        proxy_http_version 1.1;
+        proxy_set_header Upgrade $http_upgrade;
+        proxy_set_header Connection "upgrade";
+    }
+
+    # control (from probe & UI) websockets
+    location ~ ^/api/pipe/[^/]+(/probe)? {
+        proxy_pass http://pipes.weave.local$request_uri;
+        proxy_http_version 1.1;
+        proxy_set_header Upgrade $http_upgrade;
+        proxy_set_header Connection "upgrade";
+    }
+
+    location = /api/report {
+        proxy_pass http://collection.weave.local$request_uri;
+    }
+
+    location /api/topology {
+        proxy_pass http://query.weave.local$request_uri;
+    }
+
+    location /api/control {
+        proxy_pass http://controls.weave.local$request_uri;
+    }
+
+    # Static version file
+    location = /api {
+        alias /home/weave/api.json;
+    }
+
+    # The rest will be served by the collection server (any one can do it)
+    location / {
+        proxy_pass http://collection.weave.local$request_uri;
+    }
+}
--- a/experimental/multitenant/run.sh
+++ b/experimental/multitenant/run.sh
@@ -0,0 +1,62 @@
+#!/bin/bash
+
+set -eux
+
+eval $(weave env)
+CHECKPOINT_DISABLE=true
+
+start_container() {
+    local replicas=$1
+    local image=$2
+    local basename=$3
+    shift 3
+    local hostname=${basename}.weave.local
+
+    local docker_args=
+    while [ "$#" -gt 0 ]; do
+        case "$1" in
+            --)
+            shift
+            break
+            ;;
+            *)
+            docker_args="${docker_args} $1"
+            shift
+            ;;
+        esac
+    done
+    local container_args="$@"
+
+    for i in $(seq ${replicas}); do
+        if docker inspect ${basename}${i} >/dev/null 2>&1; then
+            docker rm -f ${basename}${i}
+        fi
+        docker run -d -e CHECKPOINT_DISABLE --name=${basename}${i} --hostname=${hostname} \
+            ${docker_args} ${image} ${container_args}
+    done
+}
+
+# These are the infrastructure bits - do not use these containers in production
+start_container 1 deangiberson/aws-dynamodb-local dynamodb
+start_container 1 pakohan/elasticmq sqs
+start_container 1 progrium/consul consul -p 8400:8400 -p 8500:8500 -p 8600:53/udp -- -server -bootstrap -ui-dir /ui
+
+# These are the micro services
+common_args="--no-probe --app.weave.addr= --app.http.address=:80"
+aws_args="--app.aws.region=us-east-1 --app.aws.id=abc --app.aws.secret=123 --app.aws.token=xyz"
+start_container 2 weaveworks/scope collection -- ${common_args} --app.collector=dynamodb \
+                                                 ${aws_args} \
+                                                 --app.aws.dynamodb=http://dynamodb.weave.local:8000 \
+                                                 --app.aws.create.tables=true
+start_container 2 weaveworks/scope query      -- ${common_args} --app.collector=dynamodb \
+                                                 ${aws_args} \
+                                                 --app.aws.dynamodb=http://dynamodb.weave.local:8000
+start_container 2 weaveworks/scope controls   -- ${common_args} --app.control.router=sqs \
+                                                 ${aws_args} \
+                                                 --app.aws.sqs=http://sqs.weave.local:9324
+start_container 2 weaveworks/scope pipes      -- ${common_args} --app.pipe.router=consul \
+                                                 --app.consul.addr=consul.weave.local:8500 --app.consul.inf=ethwe \
+                                                 --app.consul.prefix=pipes/
+
+# And we bring it all together with a reverse proxy
+start_container 1 weaveworks/scope-frontend frontend --add-host=dns.weave.local:$(weave docker-bridge-ip) --publish=4040:80
--- a/prog/app.go
+++ b/prog/app.go
@@ -9,23 +9,25 @@ import (
 	"time"

 	log "github.com/Sirupsen/logrus"
+	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/gorilla/mux"
 	"github.com/weaveworks/go-checkpoint"
 	"github.com/weaveworks/weave/common"

 	"github.com/weaveworks/scope/app"
+	"github.com/weaveworks/scope/app/multitenant"
 	"github.com/weaveworks/scope/common/weave"
 	"github.com/weaveworks/scope/common/xfer"
 	"github.com/weaveworks/scope/probe/docker"
 )

 // Router creates the mux for all the various app components.
-func router(c app.Collector) http.Handler {
+func router(collector app.Collector, controlRouter app.ControlRouter, pipeRouter app.PipeRouter) http.Handler {
 	router := mux.NewRouter()
-	app.RegisterReportPostHandler(c, router)
-	app.RegisterControlRoutes(router, app.NewLocalControlRouter())
-	app.RegisterPipeRoutes(router, app.NewLocalPipeRouter())
-	return app.TopologyHandler(c, router, http.FileServer(FS(false)))
+	app.RegisterReportPostHandler(collector, router)
+	app.RegisterControlRoutes(router, controlRouter)
+	app.RegisterPipeRoutes(router, pipeRouter)
+	return app.TopologyHandler(collector, router, http.FileServer(FS(false)))
 }

 // Main runs the app
@@ -40,14 +42,81 @@ func appMain() {
 		weaveHostname  = flag.String("weave.hostname", app.DefaultHostname, "Hostname to advertise in WeaveDNS")
 		containerName  = flag.String("container.name", app.DefaultContainerName, "Name of this container (to lookup container ID)")
 		dockerEndpoint = flag.String("docker", app.DefaultDockerEndpoint, "Location of docker endpoint (to lookup container ID)")
+
+		collectorType     = flag.String("collector", "local", "Collector to use (local of dynamodb)")
+		controlRouterType = flag.String("control.router", "local", "Control router to use (local or sqs)")
+		pipeRouterType    = flag.String("pipe.router", "local", "Pipe router to use (local)")
+		userIDHeader      = flag.String("userid.header", "", "HTTP header to use as userid")
+
+		awsDynamoDB     = flag.String("aws.dynamodb", "", "URL of DynamoDB instance")
+		awsCreateTables = flag.Bool("aws.create.tables", false, "Create the tables in DynamoDB")
+		awsSQS          = flag.String("aws.sqs", "", "URL of SQS instance")
+		awsRegion       = flag.String("aws.region", "", "AWS Region")
+		awsID           = flag.String("aws.id", "", "AWS Account ID")
+		awsSecret       = flag.String("aws.secret", "", "AWS Account Secret")
+		awsToken        = flag.String("aws.token", "", "AWS Account Token")
+
+		consulPrefix = flag.String("consul.prefix", "", "Prefix for keys in consul")
+		consulAddr   = flag.String("consul.addr", "", "Address of consul instance")
+		consulInf    = flag.String("consul.inf", "", "The interface who's address I should advertise myself under in consul")
 	)
 	flag.Parse()

 	setLogLevel(*logLevel)
 	setLogFormatter(*logPrefix)

-	defer log.Info("app exiting")
+	// Do we need a user IDer?
+	var userIDer = multitenant.NoopUserIDer
+	if *userIDHeader != "" {
+		userIDer = multitenant.UserIDHeader(*userIDHeader)
+	}

+	// Create a collector
+	var collector app.Collector
+	if *collectorType == "local" {
+		collector = app.NewCollector(*window)
+	} else if *collectorType == "dynamodb" {
+		creds := credentials.NewStaticCredentials(*awsID, *awsSecret, *awsToken)
+		dynamoCollector := multitenant.NewDynamoDBCollector(*awsDynamoDB, *awsRegion, creds, userIDer)
+		collector = dynamoCollector
+		if *awsCreateTables {
+			if err := dynamoCollector.CreateTables(); err != nil {
+				log.Fatalf("Error createing DynamoDB tables: %v", err)
+			}
+		}
+	} else {
+		log.Fatalf("Invalid collector '%s'", *collectorType)
+		return
+	}
+
+	// Create a control router
+	var controlRouter app.ControlRouter
+	if *controlRouterType == "local" {
+		controlRouter = app.NewLocalControlRouter()
+	} else if *controlRouterType == "sqs" {
+		creds := credentials.NewStaticCredentials(*awsID, *awsSecret, *awsToken)
+		controlRouter = multitenant.NewSQSControlRouter(*awsSQS, *awsRegion, creds, userIDer)
+	} else {
+		log.Fatalf("Invalid control router '%s'", *controlRouterType)
+		return
+	}
+
+	// Create a pipe router
+	var pipeRouter app.PipeRouter
+	if *pipeRouterType == "local" {
+		pipeRouter = app.NewLocalPipeRouter()
+	} else if *pipeRouterType == "consul" {
+		var err error
+		pipeRouter, err = multitenant.NewConsulPipeRouter(*consulAddr, *consulPrefix, *consulInf, userIDer)
+		if err != nil {
+			log.Fatalf("Error createing consul pipe router: %v", err)
+		}
+	} else {
+		log.Fatalf("Invalid pipe router '%s'", *pipeRouterType)
+		return
+	}
+
+	defer log.Info("app exiting")
 	rand.Seed(time.Now().UnixNano())
 	app.UniqueID = strconv.FormatInt(rand.Int63(), 16)
 	app.Version = version
@@ -79,7 +148,7 @@ func appMain() {
 		}
 	}

-	handler := router(app.NewCollector(*window))
+	handler := router(collector, controlRouter, pipeRouter)
 	go func() {
 		log.Infof("listening on %s", *listen)
 		log.Info(http.ListenAndServe(*listen, handler))
--- a/prog/probe.go
+++ b/prog/probe.go
@@ -18,6 +18,7 @@ import (
 	"github.com/weaveworks/weave/common"

 	"github.com/weaveworks/scope/common/hostname"
+	"github.com/weaveworks/scope/common/network"
 	"github.com/weaveworks/scope/common/sanitize"
 	"github.com/weaveworks/scope/common/weave"
 	"github.com/weaveworks/scope/common/xfer"
@@ -176,7 +177,7 @@ func probeMain() {
 		p.AddTagger(weave)
 		p.AddReporter(weave)

-		dockerBridgeIP, err := getFirstAddressOf(*dockerBridge)
+		dockerBridgeIP, err := network.GetFirstAddressOf(*dockerBridge)
 		if err != nil {
 			log.Println("Error getting docker bridge ip:", err)
 		} else {
@@ -199,25 +200,3 @@ func probeMain() {

 	common.SignalHandlerLoop()
 }
-
-func getFirstAddressOf(name string) (string, error) {
-	inf, err := net.InterfaceByName(name)
-	if err != nil {
-		return "", err
-	}
-
-	addrs, err := inf.Addrs()
-	if err != nil {
-		return "", err
-	}
-	if len(addrs) <= 0 {
-		return "", fmt.Errorf("No address found for %s", name)
-	}
-
-	switch v := addrs[0].(type) {
-	case *net.IPNet:
-		return v.IP.String(), nil
-	default:
-		return "", fmt.Errorf("No address found for %s", name)
-	}
-}
				`@@ -0,0 +1 @@`
				`{"id": "sccopeservice", "version": "Multimagic"}`