Merge pull request #1682 from kinvolk/krnowak/plugin-controls

RFC: forwarding control requests to plugins

Makefile

@@ -67,7 +67,7 @@ $(SCOPE_EXE) $(RUNSVINIT) lint tests shell prog/static.go: $(SCOPE_BACKEND_BUILD
 		-v $(shell pwd)/.pkg:/go/pkg \
 		--net=host \
 		-e GOARCH -e GOOS -e CIRCLECI -e CIRCLE_BUILD_NUM -e CIRCLE_NODE_TOTAL \
-		-e CIRCLE_NODE_INDEX -e COVERDIR -e SLOW \
+		-e CIRCLE_NODE_INDEX -e COVERDIR -e SLOW -e TESTDIRS \
 		$(SCOPE_BACKEND_BUILD_IMAGE) SCOPE_VERSION=$(SCOPE_VERSION) GO_BUILD_INSTALL_DEPS=$(GO_BUILD_INSTALL_DEPS) $@
 
 else

app/collector.go

@@ -118,7 +118,9 @@ func (c *collector) Report(_ context.Context) (report.Report, error) {
 	}
 
 	c.clean()
-	return c.merger.Merge(c.reports), nil
+	rpt := c.merger.Merge(c.reports).Upgrade()
+	c.cached = &rpt
+	return rpt, nil
 }
 
 func (c *collector) clean() {

examples/plugins/README.md

@@ -1,45 +1,71 @@
 # Scope Probe Plugins
 
-Scope probe plugins let you insert your own custom metrics into Scope and get them displayed in the UI.
+Scope probe plugins let you insert your own custom metrics into Scope
+and get them displayed in the UI.
 
 <img src="../../imgs/plugin.png" width="800" alt="Scope Probe plugin screenshot" align="center">
 
-You can find some examples at the
-[the example plugins](https://github.com/weaveworks/scope/tree/master/examples/plugins)
+You can find some examples at the [the example
+plugins](https://github.com/weaveworks/scope/tree/master/examples/plugins)
 directory. We currently provide two examples:
-* A
-  [Python plugin](https://github.com/weaveworks/scope/tree/master/examples/plugins/http-requests)
-  using [bcc](http://iovisor.github.io/bcc/) to extract incoming HTTP request
-  rates per process, without any application-level instrumentation requirements and negligible performance toll (metrics are obtained in-kernel without any packet copying to userspace).
-  **Note:** This plugin needs a [recent kernel version with ebpf support](https://github.com/iovisor/bcc/blob/master/INSTALL.md#kernel-configuration). It will not compile on current [dlite](https://github.com/nlf/dlite) and boot2docker hosts.
-* A
-  [Go plugin](https://github.com/weaveworks/scope/tree/master/examples/plugins/iovisor),
-  using [iostat](https://en.wikipedia.org/wiki/Iostat) to provide host-level CPU IO wait
-  metrics.
+
+* A [Python
+  plugin](https://github.com/weaveworks/scope/tree/master/examples/plugins/http-requests)
+  using [bcc](http://iovisor.github.io/bcc/) to extract incoming HTTP
+  request rates per process, without any application-level
+  instrumentation requirements and negligible performance toll
+  (metrics are obtained in-kernel without any packet copying to
+  userspace).  **Note:** This plugin needs a [recent kernel version
+  with ebpf
+  support](https://github.com/iovisor/bcc/blob/master/INSTALL.md#kernel-configuration). It
+  will not compile on current [dlite](https://github.com/nlf/dlite)
+  and boot2docker hosts.
+* A [Go
+  plugin](https://github.com/weaveworks/scope/tree/master/examples/plugins/iowait),
+  using [iostat](https://en.wikipedia.org/wiki/Iostat) to provide
+  host-level CPU IO wait or idle metrics.
 
 The example plugins can be run by calling `make` in their directory.
 This will build the plugin, and immediately run it in the foreground.
 To run the plugin in the background, see the `Makefile` for examples
 of the `docker run ...` command.
 
-If the running plugin was picked up by Scope, you will see it in the list of `PLUGINS`
-in the bottom right of the UI.
+If the running plugin was picked up by Scope, you will see it in the
+list of `PLUGINS` in the bottom right of the UI.
 
+## Plugin ID
 
-## <a id="protocol"></a>Protocol
+Each plugin should have an unique ID. It is forbidden to change it
+during the plugin's lifetime. The scope probe will get the plugin's ID
+from the plugin's socket filename. For example, the socket named
+`my-plugin.sock`, the scope probe will deduce the ID as
+`my-plugin`. IDs can only contain alphanumeric sequences, optionally
+separated with a dash.
+
+## Plugin registration
 
 All plugins should listen for HTTP connections on a unix socket in the
-`/var/run/scope/plugins` directory. The scope probe will recursively scan that
-directory every 5 seconds, to look for sockets being added (or removed). It is
-also valid to put the plugin unix socket in a sub-directory, in case you want
-to apply some permissions, or store other information with the socket.
+`/var/run/scope/plugins` directory. The scope probe will recursively
+scan that directory every 5 seconds, to look for sockets being added
+(or removed). It is also valid to put the plugin unix socket in a
+sub-directory, in case you want to apply some permissions, or store
+other information with the socket.
 
-When a new plugin is detected, the scope probe will begin requesting
-reports from it via `GET /report`.
+## Protocol
 
-All plugin endpoints are expected to respond within 500ms, and respond in the JSON format.
+There are several interfaces a plugin may (or must) implement. Usually
+implementing an interface means handling specific requests. These
+requests are described below.
 
-### <a id="report"></a>Report
+### Reporter interface
+
+Plugins _must_ implement the reporter interface. Implementing this
+interface means listening for HTTP requests at `/report`.
+
+Add the "reporter" string to the `interfaces` field in the plugin
+specification.
+
+#### Report
 
 When the scope probe discovers a new plugin unix socket it will begin
 periodically making a `GET` request to the `/report` endpoint. The
@@ -69,16 +95,182 @@ For example:
 
 Note that the `Plugins` section includes exactly one plugin
 description. The plugin description fields are:
-`interfaces` including `reporter`.
 
-The fields are:
+* `id` is used to check for duplicate plugins. It is
+  required. Described in [the Plugin ID section](#plugin-id).
+* `label` is a human readable plugin label displayed in the UI. It is
+  required.
+* `description` is displayed in the UI.
+* `interfaces` is a list of interfaces which this plugin supports.  It
+  is required, and must contain at least `["reporter"]`.
+* `api_version` is used to ensure both the plugin and the scope probe
+  can speak to each other. It is required, and must match the probe.
 
-* `id` is used to check for duplicate plugins. It is required.
-* `label` is a human readable plugin label displayed in the UI. It is required.
-* `description` is displayed in the UI
-* `interfaces` is a list of interfaces which this plugin supports. It is required, and must equal `["reporter"]`.
-* `api_version` is used to ensure both the plugin and the scope probe can speak to each other. It is required, and must match the probe.
+You may notice a small chicken and egg problem - the plugin reports to
+the scope probe what interfaces it supports, but the scope probe can
+learn that only by doing a `GET /report` request which will be handled
+by the plugin if it implements the "reporter" interface. This is
+solved (or worked around) by requiring the plugin to always implements
+the "reporter" interface.
 
-### <a id="interfaces"></a>Interfaces
+### Controller interface
 
-Currently the only interface a plugin can fulfill is `reporter`.
+Plugins _may_ implement the controller interface. Implementing the
+controller interface means that the plugin can react to HTTP `POST`
+control requests sent by the app. The plugin will receive them only
+for controls it exposed in its reports. The requests will come to the
+`/control` endpoint.
+
+Add the "controller" string to the `interfaces` field in the plugin
+specification.
+
+#### Control
+
+The `POST` requests will have a JSON-encoded body with the following contents:
+
+```json
+{
+  "AppID": "some ID of an app",
+  "NodeID": "an ID of the node that had the control activated",
+  "Control": "the name of the activated control"
+}
+```
+
+The body of the response should also be a JSON-encoded data. Usually
+the body would be an empty JSON object (so, "{}" after
+serialization). If some error happens during handling the control,
+then the plugin can send a response with an `error` field set, for
+example:
+
+```json
+{
+  "error": "An error message here"
+}
+```
+
+Sometimes the control activation can make the control obsolete, so the
+plugin may want to hide it (for example, control for stopping the
+container should be hidden after the container is stopped). For this
+to work, the plugin can send a shortcut report by filling the
+`ShortcutReport` field in the response, like for example:
+
+```json
+{
+  "ShortcutReport": { body of the report here }
+}
+```
+
+##### How to expose controls
+
+Each topology in the report (be it host, pod, endpoint and so on) has
+a set of available controls a node in the topology may want to
+show. The following (rather artificial) example shows a topology with
+two controls (`ctrl-one` and `ctrl-two`) and two nodes, each having a
+different control from the two:
+
+```json
+{
+  "Host": {
+    "controls": {
+      "ctrl-one": {
+        "id": "ctrl-one",
+        "human": "Ctrl One",
+        "icon": "fa-futbol-o",
+        "rank": 1
+      },
+      "ctrl-two": {
+        "id": "ctrl-two",
+        "human": "Ctrl Two",
+        "icon": "fa-beer",
+        "rank": 2
+      }
+    },
+    "nodes": {
+      "host1": {
+        "latestControls": {
+          "ctrl-one": {
+            "timestamp": "2016-07-20T15:51:05Z01:00",
+            "value": {
+              "dead": false
+            }
+          }
+        }
+      },
+      "host2": {
+        "latestControls": {
+          "ctrl-two": {
+            "timestamp": "2016-07-20T15:51:05Z01:00",
+            "value": {
+              "dead": false
+            }
+          }
+        }
+      }
+    }
+  }
+}
+```
+
+When control "ctrl-one" is activated, the plugin will receive a
+request like:
+
+```json
+{
+  "AppID": "some ID of an app",
+  "NodeID": "host1",
+  "Control": "ctrl-one"
+}
+```
+
+A short note about the "icon" field of the topology control - the
+value for it can be taken from [Font Awesome
+Cheatsheet](http://fontawesome.io/cheatsheet/)
+
+##### Node naming
+
+Very often the controller plugin wants to add some controls to already
+existing nodes (like controls for network traffic management to nodes
+representing the running Docker container). To achieve that, it is
+important to make sure that the node ID in the plugin's report matches
+the ID of the node created by the probe. The ID is a
+semicolon-separated list of strings.
+
+For containers, images, hosts and others the ID is usually formatted
+as `${name};<${tag}>`. The `${name}` variable is usually a name of a
+thing the node represents, like an ID of the Docker container or the
+hostname. The `${tag}` denotes the type of the node. There is a fixed
+set of tags used by the probe:
+
+- host
+- container
+- container_image
+- pod
+- service
+- deployment
+- replica_set
+
+The examples of "tagged" node names:
+
+- The Docker container with full ID
+  2299a2ca59dfd821f367e689d5869c4e568272c2305701761888e1d79d7a6f51:
+  `2299a2ca59dfd821f367e689d5869c4e568272c2305701761888e1d79d7a6f51;<container>`
+- The Docker image with name `docker.io/alpine`:
+  `docker.io/alpine;<container_image>`
+- The host with name `example.com`: `example.com:<host>`
+
+The fixed set of tags listed above is not a complete set of names a
+node can have though. For example, nodes representing processes are
+have ID formatted as `${host};${pid}`. Probably the easiest ways to
+discover how the nodes are named are:
+
+- Read the code in
+  [report/id.go](https://github.com/weaveworks/scope/blob/master/report/id.go).
+- Browse the Weave Scope GUI, select some node and search for an `id`
+  key in the `nodeDetails` array in the address bar.
+  - For example in the
+    `http://localhost:4040/#!/state/{"controlPipe":null,"nodeDetails":[{"id":"example.com;<host>","label":"example.com","topologyId":"hosts"}],…`
+    URL, you can find the `example.com;<host>` which is an ID of the node
+    representing the host.
+  - Mentally substitute the `<SLASH>` with `/`. This can appear in
+    Docker image names, so `docker.io/alpine` in the address bar will
+    be `docker.io<SLASH>alpine`.

examples/plugins/http-requests/http-requests.py

@@ -123,7 +123,7 @@ class PluginRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):
             },
             'Plugins': [
               {
-                'id': 'http-requests',
+                'id': 'http_requests',
                 'label': 'HTTP Requests',
                 'description': 'Adds http request metrics to processes',
                 'interfaces': ['reporter'],

examples/plugins/iowait/Makefile

@@ -1,5 +1,6 @@
 .PHONY: run clean
 
+SUDO=$(shell docker info >/dev/null 2>&1 || echo "sudo -E")
 EXE=iowait
 IMAGE=weavescope-iowait-plugin
 UPTODATE=.$(EXE).uptodate
@@ -7,18 +8,18 @@ UPTODATE=.$(EXE).uptodate
 run: $(UPTODATE)
 	# --net=host gives us the remote hostname, in case we're being launched against a non-local docker host.
 	# We could also pass in the `-hostname=foo` flag, but that doesn't work against a remote docker host.
-	docker run --rm -it \
+	$(SUDO) docker run --rm -it \
 		--net=host \
 		-v /var/run/scope/plugins:/var/run/scope/plugins \
 		--name $(IMAGE) $(IMAGE)
 
 $(UPTODATE): $(EXE) Dockerfile
-	docker build -t $(IMAGE) .
+	$(SUDO) docker build -t $(IMAGE) .
 	touch $@
 
 $(EXE): main.go
-	docker run --rm -v "$$PWD":/usr/src/$(EXE) -w /usr/src/$(EXE) golang:1.6 go build -v
+	$(SUDO) docker run --rm -v "$$PWD":/usr/src/$(EXE) -w /usr/src/$(EXE) golang:1.6 go build -v
 
 clean:
 	- rm -rf $(UPTODATE) $(EXE)
-	- docker rmi $(IMAGE)
+	- $(SUDO) docker rmi $(IMAGE)

examples/plugins/iowait/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"encoding/json"
 	"flag"
 	"fmt"
 	"log"
@@ -11,6 +12,7 @@ import (
 	"os/signal"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -52,6 +54,7 @@ func main() {
 
 	plugin := &Plugin{HostID: *hostID}
 	http.HandleFunc("/report", plugin.Report)
+	http.HandleFunc("/control", plugin.Control)
 	if err := http.Serve(listener, nil); err != nil {
 		log.Printf("error: %v", err)
 	}
@@ -60,62 +63,308 @@ func main() {
 // Plugin groups the methods a plugin needs
 type Plugin struct {
 	HostID string
+
+	lock       sync.Mutex
+	iowaitMode bool
+}
+
+type request struct {
+	NodeID  string
+	Control string
+}
+
+type response struct {
+	ShortcutReport *report `json:"shortcutReport,omitempty"`
+}
+
+type report struct {
+	Host    topology
+	Plugins []pluginSpec
+}
+
+type topology struct {
+	Nodes           map[string]node           `json:"nodes"`
+	MetricTemplates map[string]metricTemplate `json:"metric_templates"`
+	Controls        map[string]control        `json:"controls"`
+}
+
+type node struct {
+	Metrics        map[string]metric       `json:"metrics"`
+	LatestControls map[string]controlEntry `json:"latestControls,omitempty"`
+}
+
+type metric struct {
+	Samples []sample `json:"samples,omitempty"`
+	Min     float64  `json:"min"`
+	Max     float64  `json:"max"`
+}
+
+type sample struct {
+	Date  time.Time `json:"date"`
+	Value float64   `json:"value"`
+}
+
+type controlEntry struct {
+	Timestamp time.Time   `json:"timestamp"`
+	Value     controlData `json:"value"`
+}
+
+type controlData struct {
+	Dead bool `json:"dead"`
+}
+
+type metricTemplate struct {
+	ID       string  `json:"id"`
+	Label    string  `json:"label,omitempty"`
+	Format   string  `json:"format,omitempty"`
+	Priority float64 `json:"priority,omitempty"`
+}
+
+type control struct {
+	ID    string `json:"id"`
+	Human string `json:"human"`
+	Icon  string `json:"icon"`
+	Rank  int    `json:"rank"`
+}
+
+type pluginSpec struct {
+	ID          string   `json:"id"`
+	Label       string   `json:"label"`
+	Description string   `json:"description,omitempty"`
+	Interfaces  []string `json:"interfaces"`
+	APIVersion  string   `json:"api_version,omitempty"`
+}
+
+func (p *Plugin) makeReport() (*report, error) {
+	metrics, err := p.metrics()
+	if err != nil {
+		return nil, err
+	}
+	rpt := &report{
+		Host: topology{
+			Nodes: map[string]node{
+				p.getTopologyHost(): {
+					Metrics:        metrics,
+					LatestControls: p.latestControls(),
+				},
+			},
+			MetricTemplates: p.metricTemplates(),
+			Controls:        p.controls(),
+		},
+		Plugins: []pluginSpec{
+			{
+				ID:          "iowait",
+				Label:       "iowait",
+				Description: "Adds a graph of CPU IO Wait to hosts",
+				Interfaces:  []string{"reporter", "controller"},
+				APIVersion:  "1",
+			},
+		},
+	}
+	return rpt, nil
+}
+
+func (p *Plugin) metrics() (map[string]metric, error) {
+	value, err := p.metricValue()
+	if err != nil {
+		return nil, err
+	}
+	id, _ := p.metricIDAndName()
+	metrics := map[string]metric{
+		id: {
+			Samples: []sample{
+				{
+					Date:  time.Now(),
+					Value: value,
+				},
+			},
+			Min: 0,
+			Max: 100,
+		},
+	}
+	return metrics, nil
+}
+
+func (p *Plugin) latestControls() map[string]controlEntry {
+	ts := time.Now()
+	ctrls := map[string]controlEntry{}
+	for _, details := range p.allControlDetails() {
+		ctrls[details.id] = controlEntry{
+			Timestamp: ts,
+			Value: controlData{
+				Dead: details.dead,
+			},
+		}
+	}
+	return ctrls
+}
+
+func (p *Plugin) metricTemplates() map[string]metricTemplate {
+	id, name := p.metricIDAndName()
+	return map[string]metricTemplate{
+		id: {
+			ID:       id,
+			Label:    name,
+			Format:   "percent",
+			Priority: 0.1,
+		},
+	}
+}
+
+func (p *Plugin) controls() map[string]control {
+	ctrls := map[string]control{}
+	for _, details := range p.allControlDetails() {
+		ctrls[details.id] = control{
+			ID:    details.id,
+			Human: details.human,
+			Icon:  details.icon,
+			Rank:  1,
+		}
+	}
+	return ctrls
 }
 
 // Report is called by scope when a new report is needed. It is part of the
 // "reporter" interface, which all plugins must implement.
 func (p *Plugin) Report(w http.ResponseWriter, r *http.Request) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
 	log.Println(r.URL.String())
-	now := time.Now()
-	nowISO := now.Format(time.RFC3339)
-	value, err := iowait()
+	rpt, err := p.makeReport()
 	if err != nil {
 		log.Printf("error: %v", err)
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
-	fmt.Fprintf(w, `{
-		 "Host": {
-			 "nodes": {
-				 %q: {
-					 "metrics": {
-						 "iowait": {
-							 "samples": [ {"date": %q, "value": %f} ],
-							 "min": 0,
-							 "max": 100
-						 }
-					 }
-				 }
-			 },
-			 "metric_templates": {
-				 "iowait": {
-					 "id":       "iowait",
-					 "label":    "IO Wait",
-					 "format":   "percent",
-					 "priority": 0.1
-				 }
-			 }
-		 },
-		 "Plugins": [
-			 {
-				 "id":          "iowait",
-				 "label":       "iowait",
-				 "description": "Adds a graph of CPU IO Wait to hosts",
-				 "interfaces":  ["reporter"],
-				 "api_version": "1"
-			 }
-		 ]
-	 }`, p.HostID+";<host>", nowISO, value)
+	raw, err := json.Marshal(*rpt)
 	if err != nil {
 		log.Printf("error: %v", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Write(raw)
+}
+
+// Control is called by scope when a control is activated. It is part
+// of the "controller" interface.
+func (p *Plugin) Control(w http.ResponseWriter, r *http.Request) {
+	p.lock.Lock()
+	defer p.lock.Unlock()
+	log.Println(r.URL.String())
+	xreq := request{}
+	err := json.NewDecoder(r.Body).Decode(&xreq)
+	if err != nil {
+		log.Printf("Bad request: %v", err)
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	thisNodeID := p.getTopologyHost()
+	if xreq.NodeID != thisNodeID {
+		log.Printf("Bad nodeID, expected %q, got %q", thisNodeID, xreq.NodeID)
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	expectedControlID, _, _ := p.controlDetails()
+	if expectedControlID != xreq.Control {
+		log.Printf("Bad control, expected %q, got %q", expectedControlID, xreq.Control)
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+	p.iowaitMode = !p.iowaitMode
+	rpt, err := p.makeReport()
+	if err != nil {
+		log.Printf("error: %v", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	res := response{ShortcutReport: rpt}
+	raw, err := json.Marshal(res)
+	if err != nil {
+		log.Printf("error: %v", err)
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.WriteHeader(http.StatusOK)
+	w.Write(raw)
+}
+
+func (p *Plugin) getTopologyHost() string {
+	return fmt.Sprintf("%s;<host>", p.HostID)
+}
+
+func (p *Plugin) metricIDAndName() (string, string) {
+	if p.iowaitMode {
+		return "iowait", "IO Wait"
+	}
+	return "idle", "Idle"
+}
+
+func (p *Plugin) metricValue() (float64, error) {
+	if p.iowaitMode {
+		return iowait()
+	}
+	return idle()
+}
+
+type controlDetails struct {
+	id    string
+	human string
+	icon  string
+	dead  bool
+}
+
+func (p *Plugin) allControlDetails() []controlDetails {
+	return []controlDetails{
+		{
+			id:    "switchToIdle",
+			human: "Switch to idle",
+			icon:  "fa-beer",
+			dead:  !p.iowaitMode,
+		},
+		{
+			id:    "switchToIOWait",
+			human: "Switch to IO wait",
+			icon:  "fa-hourglass",
+			dead:  p.iowaitMode,
+		},
 	}
 }
 
-// Get the latest iowait value
+func (p *Plugin) controlDetails() (string, string, string) {
+	for _, details := range p.allControlDetails() {
+		if !details.dead {
+			return details.id, details.human, details.icon
+		}
+	}
+	return "", "", ""
+}
+
 func iowait() (float64, error) {
+	return iostatValue(3)
+}
+
+func idle() (float64, error) {
+	return iostatValue(5)
+}
+
+func iostatValue(idx int) (float64, error) {
+	values, err := iostat()
+	if err != nil {
+		return 0, err
+	}
+	if idx >= len(values) {
+		return 0, fmt.Errorf("invalid iostat field index %d", idx)
+	}
+
+	return strconv.ParseFloat(values[idx], 64)
+}
+
+// Get the latest iostat values
+func iostat() ([]string, error) {
 	out, err := exec.Command("iostat", "-c").Output()
 	if err != nil {
-		return 0, fmt.Errorf("iowait: %v", err)
+		return nil, fmt.Errorf("iowait: %v", err)
 	}
 
 	// Linux 4.2.0-25-generic (a109563eab38)	04/01/16	_x86_64_(4 CPU)
@@ -124,13 +373,12 @@ func iowait() (float64, error) {
 	//	          2.37    0.00    1.58    0.01    0.00   96.04
 	lines := strings.Split(string(out), "\n")
 	if len(lines) < 4 {
-		return 0, fmt.Errorf("iowait: unexpected output: %q", out)
+		return nil, fmt.Errorf("iowait: unexpected output: %q", out)
 	}
 
 	values := strings.Fields(lines[3])
 	if len(values) != 6 {
-		return 0, fmt.Errorf("iowait: unexpected output: %q", out)
+		return nil, fmt.Errorf("iowait: unexpected output: %q", out)
 	}
-
-	return strconv.ParseFloat(values[3], 64)
+	return values, nil
 }

probe/controls/controls.go

@@ -6,33 +6,126 @@ import (
 	"github.com/weaveworks/scope/common/xfer"
 )
 
-var (
-	mtx      = sync.Mutex{}
-	handlers = map[string]xfer.ControlHandlerFunc{}
-)
+// HandlerRegistryBackend is an interface for storing control request
+// handlers.
+type HandlerRegistryBackend interface {
+	// Lock locks the backend, so the batch insertions or
+	// removals can be performed.
+	Lock()
+	// Unlock unlocks the registry.
+	Unlock()
+	// Register a new control handler under a given
+	// id. Implementations should not call Lock() or Unlock()
+	// here, it will be done by HandlerRegistry.
+	Register(control string, f xfer.ControlHandlerFunc)
+	// Rm deletes the handler for a given name. Implementations
+	// should not call Lock() or Unlock() here, it will be done by
+	// HandlerRegistry.
+	Rm(control string)
+	// Handler gets the handler for a control. Implementations
+	// should not call Lock() or Unlock() here, it will be done by
+	// HandlerRegistry.
+	Handler(control string) (xfer.ControlHandlerFunc, bool)
+}
+
+type defaultBackend struct {
+	handlers map[string]xfer.ControlHandlerFunc
+	mtx      sync.Mutex
+}
+
+// NewDefaultHandlerRegistryBackend creates a default backend for
+// handler registry.
+func NewDefaultHandlerRegistryBackend() HandlerRegistryBackend {
+	return &defaultBackend{
+		handlers: map[string]xfer.ControlHandlerFunc{},
+	}
+}
+
+// Lock locks the registry, so the batch insertions or
+// removals can be performed.
+func (b *defaultBackend) Lock() {
+	b.mtx.Lock()
+}
+
+// Unlock unlocks the registry.
+func (b *defaultBackend) Unlock() {
+	b.mtx.Unlock()
+}
+
+// Register a new control handler under a given id.
+func (b *defaultBackend) Register(control string, f xfer.ControlHandlerFunc) {
+	b.handlers[control] = f
+}
+
+// Rm deletes the handler for a given name.
+func (b *defaultBackend) Rm(control string) {
+	delete(b.handlers, control)
+}
+
+// Handler gets the handler for a control.
+func (b *defaultBackend) Handler(control string) (xfer.ControlHandlerFunc, bool) {
+	handler, ok := b.handlers[control]
+	return handler, ok
+}
+
+// HandlerRegistry uses backend for storing and retrieving control
+// requests handlers.
+type HandlerRegistry struct {
+	backend HandlerRegistryBackend
+}
+
+// NewDefaultHandlerRegistry creates a registry with a default
+// backend.
+func NewDefaultHandlerRegistry() *HandlerRegistry {
+	return NewHandlerRegistry(NewDefaultHandlerRegistryBackend())
+}
+
+// NewHandlerRegistry creates a registry with a custom backend.
+func NewHandlerRegistry(backend HandlerRegistryBackend) *HandlerRegistry {
+	return &HandlerRegistry{
+		backend: backend,
+	}
+}
+
+// Register registers a new control handler under a given name.
+func (r *HandlerRegistry) Register(control string, f xfer.ControlHandlerFunc) {
+	r.backend.Lock()
+	defer r.backend.Unlock()
+	r.backend.Register(control, f)
+}
+
+// Rm deletes the handler for a given name.
+func (r *HandlerRegistry) Rm(control string) {
+	r.backend.Lock()
+	defer r.backend.Unlock()
+	r.backend.Rm(control)
+}
+
+// Batch first deletes handlers for given names in toRemove then
+// registers new handlers for given names in toAdd.
+func (r *HandlerRegistry) Batch(toRemove []string, toAdd map[string]xfer.ControlHandlerFunc) {
+	r.backend.Lock()
+	defer r.backend.Unlock()
+	for _, control := range toRemove {
+		r.backend.Rm(control)
+	}
+	for control, handler := range toAdd {
+		r.backend.Register(control, handler)
+	}
+}
 
 // HandleControlRequest performs a control request.
-func HandleControlRequest(req xfer.Request) xfer.Response {
-	mtx.Lock()
-	handler, ok := handlers[req.Control]
-	mtx.Unlock()
+func (r *HandlerRegistry) HandleControlRequest(req xfer.Request) xfer.Response {
+	h, ok := r.handler(req.Control)
 	if !ok {
 		return xfer.ResponseErrorf("Control %q not recognised", req.Control)
 	}
 
-	return handler(req)
+	return h(req)
 }
 
-// Register a new control handler under a given id.
-func Register(control string, f xfer.ControlHandlerFunc) {
-	mtx.Lock()
-	defer mtx.Unlock()
-	handlers[control] = f
-}
-
-// Rm deletes the handler for a given name
-func Rm(control string) {
-	mtx.Lock()
-	defer mtx.Unlock()
-	delete(handlers, control)
+func (r *HandlerRegistry) handler(control string) (xfer.ControlHandlerFunc, bool) {
+	r.backend.Lock()
+	defer r.backend.Unlock()
+	return r.backend.Handler(control)
 }

probe/controls/controls_test.go

@@ -10,17 +10,18 @@ import (
 )
 
 func TestControls(t *testing.T) {
-	controls.Register("foo", func(req xfer.Request) xfer.Response {
+	registry := controls.NewDefaultHandlerRegistry()
+	registry.Register("foo", func(req xfer.Request) xfer.Response {
 		return xfer.Response{
 			Value: "bar",
 		}
 	})
-	defer controls.Rm("foo")
+	defer registry.Rm("foo")
 
 	want := xfer.Response{
 		Value: "bar",
 	}
-	have := controls.HandleControlRequest(xfer.Request{
+	have := registry.HandleControlRequest(xfer.Request{
 		Control: "foo",
 	})
 	if !reflect.DeepEqual(want, have) {
@@ -29,10 +30,11 @@ func TestControls(t *testing.T) {
 }
 
 func TestControlsNotFound(t *testing.T) {
+	registry := controls.NewDefaultHandlerRegistry()
 	want := xfer.Response{
 		Error: "Control \"baz\" not recognised",
 	}
-	have := controls.HandleControlRequest(xfer.Request{
+	have := registry.HandleControlRequest(xfer.Request{
 		Control: "baz",
 	})
 	if !reflect.DeepEqual(want, have) {

probe/docker/container.go

@@ -442,6 +442,22 @@ func (c *container) getBaseNode() report.Node {
 	return result
 }
 
+func (c *container) controlsMap() map[string]report.NodeControlData {
+	paused := c.container.State.Paused
+	running := !paused && c.container.State.Running
+	stopped := !paused && !running
+	return map[string]report.NodeControlData{
+		UnpauseContainer: {Dead: !paused},
+		RestartContainer: {Dead: !running},
+		StopContainer:    {Dead: !running},
+		PauseContainer:   {Dead: !running},
+		AttachContainer:  {Dead: !running},
+		ExecContainer:    {Dead: !running},
+		StartContainer:   {Dead: !stopped},
+		RemoveContainer:  {Dead: !stopped},
+	}
+}
+
 func (c *container) GetNode() report.Node {
 	c.RLock()
 	defer c.RUnlock()
@@ -450,11 +466,9 @@ func (c *container) GetNode() report.Node {
 		ContainerState:      c.StateString(),
 		ContainerStateHuman: c.State(),
 	}
-	controls := []string{}
+	controls := c.controlsMap()
 
-	if c.container.State.Paused {
-		controls = append(controls, UnpauseContainer)
-	} else if c.container.State.Running {
+	if !c.container.State.Paused && c.container.State.Running {
 		uptime := (mtime.Now().Sub(c.container.State.StartedAt) / time.Second) * time.Second
 		networkMode := ""
 		if c.container.HostConfig != nil {
@@ -463,13 +477,10 @@ func (c *container) GetNode() report.Node {
 		latest[ContainerUptime] = uptime.String()
 		latest[ContainerRestartCount] = strconv.Itoa(c.container.RestartCount)
 		latest[ContainerNetworkMode] = networkMode
-		controls = append(controls, RestartContainer, StopContainer, PauseContainer, AttachContainer, ExecContainer)
-	} else {
-		controls = append(controls, StartContainer, RemoveContainer)
 	}
 
 	result := c.baseNode.WithLatests(latest)
-	result = result.WithControls(controls...)
+	result = result.WithLatestControls(controls)
 	result = result.WithMetrics(c.metrics())
 	return result
 }

probe/docker/container_test.go

@@ -76,6 +76,16 @@ func TestContainer(t *testing.T) {
 	// Now see if we go them
 	{
 		uptime := (now.Sub(startTime) / time.Second) * time.Second
+		controls := map[string]report.NodeControlData{
+			docker.UnpauseContainer: {Dead: true},
+			docker.RestartContainer: {Dead: false},
+			docker.StopContainer:    {Dead: false},
+			docker.PauseContainer:   {Dead: false},
+			docker.AttachContainer:  {Dead: false},
+			docker.ExecContainer:    {Dead: false},
+			docker.StartContainer:   {Dead: true},
+			docker.RemoveContainer:  {Dead: true},
+		}
 		want := report.MakeNodeWith("ping;<container>", map[string]string{
 			"docker_container_command":     " ",
 			"docker_container_created":     "01 Jan 01 00:00 UTC",
@@ -87,11 +97,9 @@ func TestContainer(t *testing.T) {
 			"docker_container_state":       "running",
 			"docker_container_state_human": "Up 6 years",
 			"docker_container_uptime":      uptime.String(),
-		}).
-			WithControls(
-				docker.RestartContainer, docker.StopContainer, docker.PauseContainer,
-				docker.AttachContainer, docker.ExecContainer,
-			).WithMetrics(report.Metrics{
+		}).WithLatestControls(
+			controls,
+		).WithMetrics(report.Metrics{
 			"docker_cpu_total_usage": report.MakeMetric(nil),
 			"docker_memory_usage":    report.MakeSingletonMetric(now, 12345).WithMax(45678),
 		}).WithParents(report.EmptySets.
@@ -100,7 +108,7 @@ func TestContainer(t *testing.T) {
 
 		test.Poll(t, 100*time.Millisecond, want, func() interface{} {
 			node := c.GetNode()
-			node.Latest.ForEach(func(k, v string) {
+			node.Latest.ForEach(func(k string, _ time.Time, v string) {
 				if v == "0" || v == "" {
 					node.Latest = node.Latest.Delete(k)
 				}

probe/docker/controls.go

@@ -162,23 +162,29 @@ func captureContainerID(f func(string, xfer.Request) xfer.Response) func(xfer.Re
 }
 
 func (r *registry) registerControls() {
-	controls.Register(StopContainer, captureContainerID(r.stopContainer))
-	controls.Register(StartContainer, captureContainerID(r.startContainer))
-	controls.Register(RestartContainer, captureContainerID(r.restartContainer))
-	controls.Register(PauseContainer, captureContainerID(r.pauseContainer))
-	controls.Register(UnpauseContainer, captureContainerID(r.unpauseContainer))
-	controls.Register(RemoveContainer, captureContainerID(r.removeContainer))
-	controls.Register(AttachContainer, captureContainerID(r.attachContainer))
-	controls.Register(ExecContainer, captureContainerID(r.execContainer))
+	controls := map[string]xfer.ControlHandlerFunc{
+		StopContainer:    captureContainerID(r.stopContainer),
+		StartContainer:   captureContainerID(r.startContainer),
+		RestartContainer: captureContainerID(r.restartContainer),
+		PauseContainer:   captureContainerID(r.pauseContainer),
+		UnpauseContainer: captureContainerID(r.unpauseContainer),
+		RemoveContainer:  captureContainerID(r.removeContainer),
+		AttachContainer:  captureContainerID(r.attachContainer),
+		ExecContainer:    captureContainerID(r.execContainer),
+	}
+	r.handlerRegistry.Batch(nil, controls)
 }
 
 func (r *registry) deregisterControls() {
-	controls.Rm(StopContainer)
-	controls.Rm(StartContainer)
-	controls.Rm(RestartContainer)
-	controls.Rm(PauseContainer)
-	controls.Rm(UnpauseContainer)
-	controls.Rm(RemoveContainer)
-	controls.Rm(AttachContainer)
-	controls.Rm(ExecContainer)
+	controls := []string{
+		StopContainer,
+		StartContainer,
+		RestartContainer,
+		PauseContainer,
+		UnpauseContainer,
+		RemoveContainer,
+		AttachContainer,
+		ExecContainer,
+	}
+	r.handlerRegistry.Batch(controls, nil)
 }

probe/docker/controls_test.go

@@ -16,7 +16,8 @@ import (
 func TestControls(t *testing.T) {
 	mdc := newMockClient()
 	setupStubs(mdc, func() {
-		registry, _ := docker.NewRegistry(10*time.Second, nil, false, "")
+		hr := controls.NewDefaultHandlerRegistry()
+		registry, _ := docker.NewRegistry(10*time.Second, nil, false, "", hr)
 		defer registry.Stop()
 
 		for _, tc := range []struct{ command, result string }{
@@ -26,7 +27,7 @@ func TestControls(t *testing.T) {
 			{docker.PauseContainer, "paused"},
 			{docker.UnpauseContainer, "unpaused"},
 		} {
-			result := controls.HandleControlRequest(xfer.Request{
+			result := hr.HandleControlRequest(xfer.Request{
 				Control: tc.command,
 				NodeID:  report.MakeContainerNodeID("a1b2c3d4e5"),
 			})
@@ -56,7 +57,8 @@ func TestPipes(t *testing.T) {
 
 	mdc := newMockClient()
 	setupStubs(mdc, func() {
-		registry, _ := docker.NewRegistry(10*time.Second, nil, false, "")
+		hr := controls.NewDefaultHandlerRegistry()
+		registry, _ := docker.NewRegistry(10*time.Second, nil, false, "", hr)
 		defer registry.Stop()
 
 		test.Poll(t, 100*time.Millisecond, true, func() interface{} {
@@ -68,7 +70,7 @@ func TestPipes(t *testing.T) {
 			docker.AttachContainer,
 			docker.ExecContainer,
 		} {
-			result := controls.HandleControlRequest(xfer.Request{
+			result := hr.HandleControlRequest(xfer.Request{
 				Control: tc,
 				NodeID:  report.MakeContainerNodeID("ping"),
 			})

probe/docker/registry.go

@@ -52,12 +52,13 @@ type ContainerUpdateWatcher func(report.Node)
 
 type registry struct {
 	sync.RWMutex
-	quit         chan chan struct{}
-	interval     time.Duration
-	collectStats bool
-	client       Client
-	pipes        controls.PipeClient
-	hostID       string
+	quit            chan chan struct{}
+	interval        time.Duration
+	collectStats    bool
+	client          Client
+	pipes           controls.PipeClient
+	hostID          string
+	handlerRegistry *controls.HandlerRegistry
 
 	watchers        []ContainerUpdateWatcher
 	containers      *radix.Tree
@@ -91,7 +92,7 @@ func newDockerClient(endpoint string) (Client, error) {
 }
 
 // NewRegistry returns a usable Registry. Don't forget to Stop it.
-func NewRegistry(interval time.Duration, pipes controls.PipeClient, collectStats bool, hostID string) (Registry, error) {
+func NewRegistry(interval time.Duration, pipes controls.PipeClient, collectStats bool, hostID string, handlerRegistry *controls.HandlerRegistry) (Registry, error) {
 	client, err := NewDockerClientStub(endpoint)
 	if err != nil {
 		return nil, err
@@ -102,12 +103,13 @@ func NewRegistry(interval time.Duration, pipes controls.PipeClient, collectStats
 		containersByPID: map[int]Container{},
 		images:          map[string]docker_client.APIImages{},
 
-		client:       client,
-		pipes:        pipes,
-		interval:     interval,
-		collectStats: collectStats,
-		hostID:       hostID,
-		quit:         make(chan chan struct{}),
+		client:          client,
+		pipes:           pipes,
+		interval:        interval,
+		collectStats:    collectStats,
+		hostID:          hostID,
+		handlerRegistry: handlerRegistry,
+		quit:            make(chan chan struct{}),
 	}
 
 	r.registerControls()

probe/docker/registry_test.go

@@ -12,12 +12,19 @@ import (
 	client "github.com/fsouza/go-dockerclient"
 
 	"github.com/weaveworks/scope/common/mtime"
+	"github.com/weaveworks/scope/probe/controls"
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/test"
 	"github.com/weaveworks/scope/test/reflect"
 )
 
+func testRegistry() docker.Registry {
+	hr := controls.NewDefaultHandlerRegistry()
+	registry, _ := docker.NewRegistry(10*time.Second, nil, true, "", hr)
+	return registry
+}
+
 type mockContainer struct {
 	c *client.Container
 }
@@ -319,7 +326,7 @@ func allNetworks(r docker.Registry) []client.Network {
 func TestRegistry(t *testing.T) {
 	mdc := newMockClient()
 	setupStubs(mdc, func() {
-		registry, _ := docker.NewRegistry(10*time.Second, nil, true, "")
+		registry := testRegistry()
 		defer registry.Stop()
 		runtime.Gosched()
 
@@ -350,7 +357,7 @@ func TestRegistry(t *testing.T) {
 func TestLookupByPID(t *testing.T) {
 	mdc := newMockClient()
 	setupStubs(mdc, func() {
-		registry, _ := docker.NewRegistry(10*time.Second, nil, true, "")
+		registry := testRegistry()
 		defer registry.Stop()
 
 		want := docker.Container(&mockContainer{container1})
@@ -367,7 +374,7 @@ func TestLookupByPID(t *testing.T) {
 func TestRegistryEvents(t *testing.T) {
 	mdc := newMockClient()
 	setupStubs(mdc, func() {
-		registry, _ := docker.NewRegistry(10*time.Second, nil, true, "")
+		registry := testRegistry()
 		defer registry.Stop()
 		runtime.Gosched()
 
@@ -441,7 +448,7 @@ func TestRegistryDelete(t *testing.T) {
 
 	mdc := newMockClient()
 	setupStubs(mdc, func() {
-		registry, _ := docker.NewRegistry(10*time.Second, nil, true, "")
+		registry := testRegistry()
 		defer registry.Stop()
 		runtime.Gosched()
 

probe/host/controls.go

@@ -16,11 +16,11 @@ const (
 )
 
 func (r *Reporter) registerControls() {
-	controls.Register(ExecHost, r.execHost)
+	r.handlerRegistry.Register(ExecHost, r.execHost)
 }
 
-func (*Reporter) deregisterControls() {
-	controls.Rm(ExecHost)
+func (r *Reporter) deregisterControls() {
+	r.handlerRegistry.Rm(ExecHost)
 }
 
 func (r *Reporter) execHost(req xfer.Request) xfer.Response {

probe/host/reporter.go

@@ -52,24 +52,26 @@ var (
 
 // Reporter generates Reports containing the host topology.
 type Reporter struct {
-	hostID       string
-	hostName     string
-	probeID      string
-	version      string
-	pipes        controls.PipeClient
-	hostShellCmd []string
+	hostID          string
+	hostName        string
+	probeID         string
+	version         string
+	pipes           controls.PipeClient
+	hostShellCmd    []string
+	handlerRegistry *controls.HandlerRegistry
 }
 
 // NewReporter returns a Reporter which produces a report containing host
 // topology for this host.
-func NewReporter(hostID, hostName, probeID, version string, pipes controls.PipeClient) *Reporter {
+func NewReporter(hostID, hostName, probeID, version string, pipes controls.PipeClient, handlerRegistry *controls.HandlerRegistry) *Reporter {
 	r := &Reporter{
-		hostID:       hostID,
-		hostName:     hostName,
-		probeID:      probeID,
-		pipes:        pipes,
-		version:      version,
-		hostShellCmd: getHostShellCmd(),
+		hostID:          hostID,
+		hostName:        hostName,
+		probeID:         probeID,
+		pipes:           pipes,
+		version:         version,
+		hostShellCmd:    getHostShellCmd(),
+		handlerRegistry: handlerRegistry,
 	}
 	r.registerControls()
 	return r
@@ -143,7 +145,7 @@ func (r *Reporter) Report() (report.Report, error) {
 				Add(LocalNetworks, report.MakeStringSet(localCIDRs...)),
 			).
 			WithMetrics(metrics).
-			WithControls(ExecHost),
+			WithLatestActiveControls(ExecHost),
 	)
 
 	rep.Host.Controls.AddControl(report.Control{

probe/host/reporter_test.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/weaveworks/scope/common/mtime"
+	"github.com/weaveworks/scope/probe/controls"
 	"github.com/weaveworks/scope/probe/host"
 	"github.com/weaveworks/scope/report"
 )
@@ -55,7 +56,8 @@ func TestReporter(t *testing.T) {
 	host.GetMemoryUsageBytes = func() (float64, float64) { return 60.0, 100.0 }
 	host.GetLocalNetworks = func() ([]*net.IPNet, error) { return []*net.IPNet{ipnet}, nil }
 
-	rpt, err := host.NewReporter(hostID, hostname, "", "", nil).Report()
+	hr := controls.NewDefaultHandlerRegistry()
+	rpt, err := host.NewReporter(hostID, hostname, "", "", nil, hr).Report()
 	if err != nil {
 		t.Fatal(err)
 	}

probe/kubernetes/controls.go

@@ -144,15 +144,21 @@ func (r *Reporter) ScaleDown(req xfer.Request, resource, namespace, id string) x
 }
 
 func (r *Reporter) registerControls() {
-	controls.Register(GetLogs, r.CapturePod(r.GetLogs))
-	controls.Register(DeletePod, r.CapturePod(r.deletePod))
-	controls.Register(ScaleUp, r.CaptureResource(r.ScaleUp))
-	controls.Register(ScaleDown, r.CaptureResource(r.ScaleDown))
+	controls := map[string]xfer.ControlHandlerFunc{
+		GetLogs:   r.CapturePod(r.GetLogs),
+		DeletePod: r.CapturePod(r.deletePod),
+		ScaleUp:   r.CaptureResource(r.ScaleUp),
+		ScaleDown: r.CaptureResource(r.ScaleDown),
+	}
+	r.handlerRegistry.Batch(nil, controls)
 }
 
 func (r *Reporter) deregisterControls() {
-	controls.Rm(GetLogs)
-	controls.Rm(DeletePod)
-	controls.Rm(ScaleUp)
-	controls.Rm(ScaleDown)
+	controls := []string{
+		GetLogs,
+		DeletePod,
+		ScaleUp,
+		ScaleDown,
+	}
+	r.handlerRegistry.Batch(controls, nil)
 }

probe/kubernetes/deployment.go

@@ -55,5 +55,5 @@ func (d *deployment) GetNode(probeID string) report.Node {
 		UnavailableReplicas:   fmt.Sprint(d.Status.UnavailableReplicas),
 		Strategy:              string(d.Spec.Strategy.Type),
 		report.ControlProbeID: probeID,
-	}).WithControls(ScaleUp, ScaleDown)
+	}).WithLatestActiveControls(ScaleUp, ScaleDown)
 }

probe/kubernetes/pod.go

@@ -63,5 +63,5 @@ func (p *pod) GetNode(probeID string) report.Node {
 		report.ControlProbeID: probeID,
 	}).
 		WithParents(p.parents).
-		WithControls(GetLogs, DeletePod)
+		WithLatestActiveControls(GetLogs, DeletePod)
 }

probe/kubernetes/replica_set.go

@@ -59,5 +59,5 @@ func (r *replicaSet) GetNode(probeID string) report.Node {
 		DesiredReplicas:       fmt.Sprint(r.Spec.Replicas),
 		FullyLabeledReplicas:  fmt.Sprint(r.Status.FullyLabeledReplicas),
 		report.ControlProbeID: probeID,
-	}).WithParents(r.parents).WithControls(ScaleUp, ScaleDown)
+	}).WithParents(r.parents).WithLatestActiveControls(ScaleUp, ScaleDown)
 }

probe/kubernetes/replication_controller.go

@@ -50,5 +50,5 @@ func (r *replicationController) GetNode(probeID string) report.Node {
 		DesiredReplicas:       fmt.Sprint(r.Spec.Replicas),
 		FullyLabeledReplicas:  fmt.Sprint(r.Status.FullyLabeledReplicas),
 		report.ControlProbeID: probeID,
-	}).WithParents(r.parents).WithControls(ScaleUp, ScaleDown)
+	}).WithParents(r.parents).WithLatestActiveControls(ScaleUp, ScaleDown)
 }

probe/kubernetes/reporter.go

@@ -89,21 +89,23 @@ var (
 
 // Reporter generate Reports containing Container and ContainerImage topologies
 type Reporter struct {
-	client  Client
-	pipes   controls.PipeClient
-	probeID string
-	probe   *probe.Probe
-	hostID  string
+	client          Client
+	pipes           controls.PipeClient
+	probeID         string
+	probe           *probe.Probe
+	hostID          string
+	handlerRegistry *controls.HandlerRegistry
 }
 
 // NewReporter makes a new Reporter
-func NewReporter(client Client, pipes controls.PipeClient, probeID string, hostID string, probe *probe.Probe) *Reporter {
+func NewReporter(client Client, pipes controls.PipeClient, probeID string, hostID string, probe *probe.Probe, handlerRegistry *controls.HandlerRegistry) *Reporter {
 	reporter := &Reporter{
-		client:  client,
-		pipes:   pipes,
-		probeID: probeID,
-		probe:   probe,
-		hostID:  hostID,
+		client:          client,
+		pipes:           pipes,
+		probeID:         probeID,
+		probe:           probe,
+		hostID:          hostID,
+		handlerRegistry: handlerRegistry,
 	}
 	reporter.registerControls()
 	client.WatchPods(reporter.podEvent)

probe/kubernetes/reporter_test.go

@@ -12,6 +12,7 @@ import (
 	"k8s.io/kubernetes/pkg/types"
 
 	"github.com/weaveworks/scope/common/xfer"
+	"github.com/weaveworks/scope/probe/controls"
 	"github.com/weaveworks/scope/probe/docker"
 	"github.com/weaveworks/scope/probe/kubernetes"
 	"github.com/weaveworks/scope/report"
@@ -184,7 +185,8 @@ func TestReporter(t *testing.T) {
 	pod1ID := report.MakePodNodeID(pod1UID)
 	pod2ID := report.MakePodNodeID(pod2UID)
 	serviceID := report.MakeServiceNodeID(serviceUID)
-	rpt, _ := kubernetes.NewReporter(newMockClient(), nil, "", "foo", nil).Report()
+	hr := controls.NewDefaultHandlerRegistry()
+	rpt, _ := kubernetes.NewReporter(newMockClient(), nil, "", "foo", nil, hr).Report()
 
 	// Reporter should have added the following pods
 	for _, pod := range []struct {
@@ -244,7 +246,8 @@ func TestTagger(t *testing.T) {
 		docker.LabelPrefix + "io.kubernetes.pod.uid": "123456",
 	}))
 
-	rpt, err := kubernetes.NewReporter(newMockClient(), nil, "", "", nil).Tag(rpt)
+	hr := controls.NewDefaultHandlerRegistry()
+	rpt, err := kubernetes.NewReporter(newMockClient(), nil, "", "", nil, hr).Tag(rpt)
 	if err != nil {
 		t.Errorf("Unexpected error: %v", err)
 	}
@@ -272,7 +275,8 @@ func TestReporterGetLogs(t *testing.T) {
 
 	client := newMockClient()
 	pipes := mockPipeClient{}
-	reporter := kubernetes.NewReporter(client, pipes, "", "", nil)
+	hr := controls.NewDefaultHandlerRegistry()
+	reporter := kubernetes.NewReporter(client, pipes, "", "", nil, hr)
 
 	// Should error on invalid IDs
 	{

probe/plugins/registry.go

@@ -1,10 +1,13 @@
 package plugins
 
 import (
+	"bytes"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"path/filepath"
+	"regexp"
 	"sort"
 	"strings"
 	"sync"
@@ -19,6 +22,7 @@ import (
 	"github.com/weaveworks/scope/common/backoff"
 	"github.com/weaveworks/scope/common/fs"
 	"github.com/weaveworks/scope/common/xfer"
+	"github.com/weaveworks/scope/probe/controls"
 	"github.com/weaveworks/scope/report"
 )
 
@@ -27,6 +31,7 @@ var (
 	transport                 = makeUnixRoundTripper
 	maxResponseBytes    int64 = 50 * 1024 * 1024
 	errResponseTooLarge       = fmt.Errorf("response must be shorter than 50MB")
+	validPluginName           = regexp.MustCompile("^[A-Za-z0-9]+([-][A-Za-z0-9]+)*$")
 )
 
 const (
@@ -34,6 +39,11 @@ const (
 	scanningInterval = 5 * time.Second
 )
 
+// ReportPublisher is an interface for publishing reports immediately
+type ReportPublisher interface {
+	Publish(rpt report.Report)
+}
+
 // Registry maintains a list of available plugins by name.
 type Registry struct {
 	rootPath          string
@@ -43,11 +53,15 @@ type Registry struct {
 	lock              sync.RWMutex
 	context           context.Context
 	cancel            context.CancelFunc
+	controlsByPlugin  map[string]report.StringSet
+	pluginsByID       map[string]*Plugin
+	handlerRegistry   *controls.HandlerRegistry
+	publisher         ReportPublisher
 }
 
 // NewRegistry creates a new registry which watches the given dir root for new
 // plugins, and adds them.
-func NewRegistry(rootPath, apiVersion string, handshakeMetadata map[string]string) (*Registry, error) {
+func NewRegistry(rootPath, apiVersion string, handshakeMetadata map[string]string, handlerRegistry *controls.HandlerRegistry, publisher ReportPublisher) (*Registry, error) {
 	ctx, cancel := context.WithCancel(context.Background())
 	r := &Registry{
 		rootPath:          rootPath,
@@ -56,6 +70,10 @@ func NewRegistry(rootPath, apiVersion string, handshakeMetadata map[string]strin
 		pluginsBySocket:   map[string]*Plugin{},
 		context:           ctx,
 		cancel:            cancel,
+		controlsByPlugin:  map[string]report.StringSet{},
+		pluginsByID:       map[string]*Plugin{},
+		handlerRegistry:   handlerRegistry,
+		publisher:         publisher,
 	}
 	if err := r.scan(); err != nil {
 		r.Close()
@@ -90,11 +108,14 @@ func (r *Registry) scan() error {
 	}
 
 	r.lock.Lock()
+	defer r.lock.Unlock()
 	plugins := map[string]*Plugin{}
+	pluginsByID := map[string]*Plugin{}
 	// add (or keep) plugins which were found
 	for _, path := range sockets {
 		if plugin, ok := r.pluginsBySocket[path]; ok {
 			plugins[path] = plugin
+			pluginsByID[plugin.PluginSpec.ID] = plugin
 			continue
 		}
 		tr, err := transport(path, pluginTimeout)
@@ -103,18 +124,26 @@ func (r *Registry) scan() error {
 			continue
 		}
 		client := &http.Client{Transport: tr, Timeout: pluginTimeout}
-		plugins[path] = NewPlugin(r.context, path, client, r.apiVersion, r.handshakeMetadata)
+		plugin, err := NewPlugin(r.context, path, client, r.apiVersion, r.handshakeMetadata)
+		if err != nil {
+			log.Warningf("plugins: error loading plugin %s: %v", path, err)
+			continue
+		}
+		plugins[path] = plugin
+		pluginsByID[plugin.PluginSpec.ID] = plugin
 		log.Infof("plugins: added plugin %s", path)
 	}
 	// remove plugins which weren't found
+	pluginsToClose := map[string]*Plugin{}
 	for path, plugin := range r.pluginsBySocket {
 		if _, ok := plugins[path]; !ok {
-			plugin.Close()
+			pluginsToClose[plugin.PluginSpec.ID] = plugin
 			log.Infof("plugins: removed plugin %s", plugin.socket)
 		}
 	}
+	r.closePlugins(pluginsToClose)
 	r.pluginsBySocket = plugins
-	r.lock.Unlock()
+	r.pluginsByID = pluginsByID
 	return nil
 }
 
@@ -148,10 +177,10 @@ func (r *Registry) sockets(path string) ([]string, error) {
 	return result, nil
 }
 
-// ForEach walks through all the plugins running f for each one.
-func (r *Registry) ForEach(f func(p *Plugin)) {
-	r.lock.RLock()
-	defer r.lock.RUnlock()
+// forEach walks through all the plugins running f for each one.
+func (r *Registry) forEach(lock sync.Locker, f func(p *Plugin)) {
+	lock.Lock()
+	defer lock.Unlock()
 	paths := []string{}
 	for path := range r.pluginsBySocket {
 		paths = append(paths, path)
@@ -162,6 +191,11 @@ func (r *Registry) ForEach(f func(p *Plugin)) {
 	}
 }
 
+// ForEach walks through all the plugins running f for each one.
+func (r *Registry) ForEach(f func(p *Plugin)) {
+	r.forEach(r.lock.RLocker(), f)
+}
+
 // Implementers walks the available plugins fulfilling the given interface
 func (r *Registry) Implementers(iface string, f func(p *Plugin)) {
 	r.ForEach(func(p *Plugin) {
@@ -180,25 +214,156 @@ func (r *Registry) Name() string { return "plugins" }
 func (r *Registry) Report() (report.Report, error) {
 	rpt := report.MakeReport()
 	// All plugins are assumed to (and must) implement reporter
-	r.ForEach(func(plugin *Plugin) {
+	r.forEach(&r.lock, func(plugin *Plugin) {
 		pluginReport, err := plugin.Report()
 		if err != nil {
 			log.Errorf("plugins: %s: /report error: %v", plugin.socket, err)
 		}
+		if plugin.Implements("controller") {
+			r.updateAndRegisterControlsInReport(&pluginReport)
+		}
 		rpt = rpt.Merge(pluginReport)
 	})
 	return rpt, nil
 }
 
+func (r *Registry) updateAndRegisterControlsInReport(rpt *report.Report) {
+	key := rpt.Plugins.Keys()[0]
+	spec, _ := rpt.Plugins.Lookup(key)
+	pluginID := spec.ID
+	topologies := topologyPointers(rpt)
+	var newPluginControls []string
+	for _, topology := range topologies {
+		newPluginControls = append(newPluginControls, r.updateAndGetControlsInTopology(pluginID, topology)...)
+	}
+	r.updatePluginControls(pluginID, report.MakeStringSet(newPluginControls...))
+}
+
+func topologyPointers(rpt *report.Report) []*report.Topology {
+	// We cannot use rpt.Topologies(), because it makes a slice of
+	// topology copies and we need original locations to modify
+	// them.
+	return []*report.Topology{
+		&rpt.Endpoint,
+		&rpt.Process,
+		&rpt.Container,
+		&rpt.ContainerImage,
+		&rpt.Pod,
+		&rpt.Service,
+		&rpt.Deployment,
+		&rpt.ReplicaSet,
+		&rpt.Host,
+		&rpt.Overlay,
+	}
+}
+
+func (r *Registry) updateAndGetControlsInTopology(pluginID string, topology *report.Topology) []string {
+	var pluginControls []string
+	newControls := report.Controls{}
+	for controlID, control := range topology.Controls {
+		fakeID := fakeControlID(pluginID, controlID)
+		log.Debugf("plugins: replacing control %s with %s", controlID, fakeID)
+		control.ID = fakeID
+		newControls.AddControl(control)
+		pluginControls = append(pluginControls, controlID)
+	}
+	newNodes := report.Nodes{}
+	for name, node := range topology.Nodes {
+		log.Debugf("plugins: checking node controls in node %s of %s", name, topology.Label)
+		newNode := node.WithID(name)
+		newLatestControls := report.MakeNodeControlDataLatestMap()
+		node.LatestControls.ForEach(func(controlID string, ts time.Time, data report.NodeControlData) {
+			log.Debugf("plugins: got node control %s", controlID)
+			newControlID := ""
+			if _, found := topology.Controls[controlID]; !found {
+				log.Debugf("plugins: node control %s does not exist in topology controls", controlID)
+				newControlID = controlID
+			} else {
+				newControlID = fakeControlID(pluginID, controlID)
+				log.Debugf("plugins: will replace node control %s with %s", controlID, newControlID)
+			}
+			newLatestControls = newLatestControls.Set(newControlID, ts, data)
+		})
+		newNode.LatestControls = newLatestControls
+		newNodes[newNode.ID] = newNode
+	}
+	topology.Controls = newControls
+	topology.Nodes = newNodes
+	return pluginControls
+}
+
+func (r *Registry) updatePluginControls(pluginID string, newPluginControls report.StringSet) {
+	oldFakePluginControls := r.fakePluginControls(pluginID)
+	newFakePluginControls := map[string]xfer.ControlHandlerFunc{}
+	for _, controlID := range newPluginControls {
+		newFakePluginControls[fakeControlID(pluginID, controlID)] = r.pluginControlHandler
+	}
+	r.handlerRegistry.Batch(oldFakePluginControls, newFakePluginControls)
+	r.controlsByPlugin[pluginID] = newPluginControls
+}
+
+// PluginResponse is an extension of xfer.Response that allows plugins
+// to send the shortcut reports
+type PluginResponse struct {
+	xfer.Response
+	ShortcutReport *report.Report `json:"shortcutReport,omitempty"`
+}
+
+func (r *Registry) pluginControlHandler(req xfer.Request) xfer.Response {
+	pluginID, controlID := realPluginAndControlID(req.Control)
+	req.Control = controlID
+	r.lock.RLock()
+	defer r.lock.RUnlock()
+	if plugin, found := r.pluginsByID[pluginID]; found {
+		response := plugin.Control(req)
+		if response.ShortcutReport != nil {
+			r.updateAndRegisterControlsInReport(response.ShortcutReport)
+			response.ShortcutReport.Shortcut = true
+			r.publisher.Publish(*response.ShortcutReport)
+		}
+		return response.Response
+	}
+	return xfer.ResponseErrorf("plugin %s not found", pluginID)
+}
+
+func realPluginAndControlID(fakeID string) (string, string) {
+	parts := strings.SplitN(fakeID, "~", 2)
+	if len(parts) != 2 {
+		return "", fakeID
+	}
+	return parts[0], parts[1]
+}
+
 // Close shuts down the registry. It can still be used after this, but will be
 // out of date.
 func (r *Registry) Close() {
 	r.cancel()
 	r.lock.Lock()
 	defer r.lock.Unlock()
-	for _, plugin := range r.pluginsBySocket {
+	r.closePlugins(r.pluginsByID)
+}
+
+func (r *Registry) closePlugins(plugins map[string]*Plugin) {
+	var toRemove []string
+	for pluginID, plugin := range plugins {
+		toRemove = append(toRemove, r.fakePluginControls(pluginID)...)
+		delete(r.controlsByPlugin, pluginID)
 		plugin.Close()
 	}
+	r.handlerRegistry.Batch(toRemove, nil)
+}
+
+func (r *Registry) fakePluginControls(pluginID string) []string {
+	oldPluginControls := r.controlsByPlugin[pluginID]
+	var oldFakePluginControls []string
+	for _, controlID := range oldPluginControls {
+		oldFakePluginControls = append(oldFakePluginControls, fakeControlID(pluginID, controlID))
+	}
+	return oldFakePluginControls
+}
+
+func fakeControlID(pluginID, controlID string) string {
+	return fmt.Sprintf("%s~%s", pluginID, controlID)
 }
 
 // Plugin is the implementation of a plugin. It is responsible for doing the
@@ -216,16 +381,19 @@ type Plugin struct {
 
 // NewPlugin loads and initializes a new plugin. If client is nil,
 // http.DefaultClient will be used.
-func NewPlugin(ctx context.Context, socket string, client *http.Client, expectedAPIVersion string, handshakeMetadata map[string]string) *Plugin {
+func NewPlugin(ctx context.Context, socket string, client *http.Client, expectedAPIVersion string, handshakeMetadata map[string]string) (*Plugin, error) {
+	id := strings.TrimSuffix(filepath.Base(socket), filepath.Ext(socket))
+	if !validPluginName.MatchString(id) {
+		return nil, fmt.Errorf("invalid plugin id %q", id)
+	}
+
 	params := url.Values{}
 	for k, v := range handshakeMetadata {
 		params.Add(k, v)
 	}
 
-	id := strings.TrimSuffix(filepath.Base(socket), filepath.Ext(socket))
-
 	ctx, cancel := context.WithCancel(ctx)
-	return &Plugin{
+	plugin := &Plugin{
 		PluginSpec:         xfer.PluginSpec{ID: id, Label: id},
 		context:            ctx,
 		socket:             socket,
@@ -234,6 +402,7 @@ func NewPlugin(ctx context.Context, socket string, client *http.Client, expected
 		client:             client,
 		cancel:             cancel,
 	}
+	return plugin, nil
 }
 
 // Report gets the latest report from the plugin
@@ -257,29 +426,51 @@ func (p *Plugin) Report() (result report.Report, err error) {
 
 	key := result.Plugins.Keys()[0]
 	spec, _ := result.Plugins.Lookup(key)
+	if spec.ID != p.PluginSpec.ID {
+		return result, fmt.Errorf("plugin must not change its id (is %q, should be %q)", spec.ID, p.PluginSpec.ID)
+	}
 	p.PluginSpec = spec
 
-	foundReporter := false
-	for _, i := range spec.Interfaces {
-		if i == "reporter" {
-			foundReporter = true
-			break
-		}
-	}
 	switch {
 	case spec.APIVersion != p.expectedAPIVersion:
 		err = fmt.Errorf("incorrect API version: expected %q, got %q", p.expectedAPIVersion, spec.APIVersion)
-	case spec.ID == "":
-		err = fmt.Errorf("spec must contain an id")
 	case spec.Label == "":
 		err = fmt.Errorf("spec must contain a label")
-	case !foundReporter:
+	case !p.Implements("reporter"):
 		err = fmt.Errorf("spec must implement the \"reporter\" interface")
 	}
 
 	return result, err
 }
 
+// Control sends a control message to a plugin
+func (p *Plugin) Control(request xfer.Request) (res PluginResponse) {
+	var err error
+	defer func() {
+		p.setStatus(err)
+		if err != nil {
+			res = PluginResponse{Response: xfer.ResponseError(err)}
+		}
+	}()
+
+	if p.Implements("controller") {
+		err = p.post("/control", p.handshakeMetadata, request, &res)
+	} else {
+		err = fmt.Errorf("the %s plugin does not implement the controller interface", p.PluginSpec.Label)
+	}
+	return res
+}
+
+// Implements checks if the plugin implements the given interface
+func (p *Plugin) Implements(iface string) bool {
+	for _, i := range p.PluginSpec.Interfaces {
+		if i == iface {
+			return true
+		}
+	}
+	return false
+}
+
 func (p *Plugin) setStatus(err error) {
 	if err == nil {
 		p.Status = "ok"
@@ -296,11 +487,34 @@ func (p *Plugin) get(path string, params url.Values, result interface{}) error {
 	if err != nil {
 		return err
 	}
+	defer resp.Body.Close()
 	if resp.StatusCode != http.StatusOK {
 		return fmt.Errorf("plugin returned non-200 status code: %s", resp.Status)
 	}
+	return getResult(resp.Body, result)
+}
+
+func (p *Plugin) post(path string, params url.Values, data interface{}, result interface{}) error {
+	// Context here lets us either timeout req. or cancel it in Plugin.Close
+	ctx, cancel := context.WithTimeout(p.context, pluginTimeout)
+	defer cancel()
+	buf := &bytes.Buffer{}
+	if err := codec.NewEncoder(buf, &codec.JsonHandle{}).Encode(data); err != nil {
+		return fmt.Errorf("encoding error: %s", err)
+	}
+	resp, err := ctxhttp.Post(ctx, p.client, fmt.Sprintf("http://plugin%s?%s", path, params.Encode()), "application/json", buf)
+	if err != nil {
+		return err
+	}
 	defer resp.Body.Close()
-	err = codec.NewDecoder(MaxBytesReader(resp.Body, maxResponseBytes, errResponseTooLarge), &codec.JsonHandle{}).Decode(&result)
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("plugin returned non-200 status code: %s", resp.Status)
+	}
+	return getResult(resp.Body, result)
+}
+
+func getResult(body io.ReadCloser, result interface{}) error {
+	err := codec.NewDecoder(MaxBytesReader(body, maxResponseBytes, errResponseTooLarge), &codec.JsonHandle{}).Decode(&result)
 	if err == errResponseTooLarge {
 		return err
 	}

probe/plugins/registry_internal_test.go

@@ -1,6 +1,7 @@
 package plugins
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 	"net"
@@ -9,19 +10,33 @@ import (
 	"net/http/httputil"
 	"path/filepath"
 	"sort"
+	"sync"
 	"syscall"
 	"testing"
 	"time"
 
 	"github.com/paypal/ionet"
+	"github.com/ugorji/go/codec"
 
 	fs_hook "github.com/weaveworks/scope/common/fs"
 	"github.com/weaveworks/scope/common/xfer"
+	"github.com/weaveworks/scope/probe/controls"
+	"github.com/weaveworks/scope/report"
 	"github.com/weaveworks/scope/test"
 	"github.com/weaveworks/scope/test/fs"
 	"github.com/weaveworks/scope/test/reflect"
 )
 
+func testRegistry(t *testing.T, apiVersion string) *Registry {
+	handlerRegistry := controls.NewDefaultHandlerRegistry()
+	root := "/plugins"
+	r, err := NewRegistry(root, apiVersion, nil, handlerRegistry, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return r
+}
+
 func stubTransport(fn func(socket string, timeout time.Duration) (http.RoundTripper, error)) {
 	transport = fn
 }
@@ -158,16 +173,68 @@ func checkLoadedPluginIDs(t *testing.T, forEach iterator, expectedIDs []string)
 	}
 }
 
+type testResponse struct {
+	Status int
+	Body   string
+}
+
+type testResponseMap map[string]testResponse
+
+// mapStringHandler returns an http.Handler which just prints the given string for each path
+func mapStringHandler(responses testResponseMap) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if response, found := responses[r.URL.Path]; found {
+			w.WriteHeader(response.Status)
+			fmt.Fprint(w, response.Body)
+		} else {
+			http.NotFound(w, r)
+		}
+	})
+}
+
 // stringHandler returns an http.Handler which just prints the given string
 func stringHandler(status int, j string) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		if r.URL.Path != "/report" {
-			http.NotFound(w, r)
-			return
-		}
-		w.WriteHeader(status)
-		fmt.Fprint(w, j)
-	})
+	return mapStringHandler(testResponseMap{"/report": {status, j}})
+}
+
+type testHandlerRegistryBackend struct {
+	handlers map[string]xfer.ControlHandlerFunc
+	t        *testing.T
+	mtx      sync.Mutex
+}
+
+func newTestHandlerRegistryBackend(t *testing.T) *testHandlerRegistryBackend {
+	return &testHandlerRegistryBackend{
+		handlers: map[string]xfer.ControlHandlerFunc{},
+		t:        t,
+	}
+}
+
+// Lock locks the backend, so the batch insertions or removals can be
+// performed.
+func (b *testHandlerRegistryBackend) Lock() {
+	b.mtx.Lock()
+}
+
+// Unlock unlocks the backend.
+func (b *testHandlerRegistryBackend) Unlock() {
+	b.mtx.Unlock()
+}
+
+// Register a new control handler under a given id.
+func (b *testHandlerRegistryBackend) Register(control string, f xfer.ControlHandlerFunc) {
+	b.handlers[control] = f
+}
+
+// Rm deletes the handler for a given name.
+func (b *testHandlerRegistryBackend) Rm(control string) {
+	delete(b.handlers, control)
+}
+
+// Handler gets the handler for the given id.
+func (b *testHandlerRegistryBackend) Handler(control string) (xfer.ControlHandlerFunc, bool) {
+	handler, ok := b.handlers[control]
+	return handler, ok
 }
 
 func TestRegistryLoadsExistingPlugins(t *testing.T) {
@@ -181,11 +248,7 @@ func TestRegistryLoadsExistingPlugins(t *testing.T) {
 	)
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "1", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "1")
 	defer r.Close()
 
 	r.Report()
@@ -211,11 +274,7 @@ func TestRegistryLoadsExistingPluginsEvenWhenOneFails(t *testing.T) {
 	)
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "1", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "1")
 	defer r.Close()
 
 	r.Report()
@@ -239,11 +298,7 @@ func TestRegistryDiscoversNewPlugins(t *testing.T) {
 	mockFS := setup(t)
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "")
 	defer r.Close()
 
 	r.Report()
@@ -273,11 +328,7 @@ func TestRegistryRemovesPlugins(t *testing.T) {
 	mockFS := setup(t, plugin.file())
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "")
 	defer r.Close()
 
 	r.Report()
@@ -305,21 +356,24 @@ func TestRegistryUpdatesPluginsWhenTheyChange(t *testing.T) {
 	setup(t, plugin.file())
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "")
 	defer r.Close()
 
 	r.Report()
 	checkLoadedPluginIDs(t, r.ForEach, []string{"testPlugin"})
 
 	// Update the plugin. Just change what the handler will respond with.
-	resp = `{"Plugins":[{"id":"updatedPlugin","label":"updatedPlugin","interfaces":["reporter"]}]}`
+	resp = `{"Plugins":[{"id":"testPlugin","label":"updatedPlugin","interfaces":["reporter"]}]}`
 
 	r.Report()
-	checkLoadedPluginIDs(t, r.ForEach, []string{"updatedPlugin"})
+	checkLoadedPlugins(t, r.ForEach, []xfer.PluginSpec{
+		{
+			ID:         "testPlugin",
+			Label:      "updatedPlugin",
+			Interfaces: []string{"reporter"},
+			Status:     "ok",
+		},
+	})
 }
 
 func TestRegistryReturnsPluginsByInterface(t *testing.T) {
@@ -338,11 +392,7 @@ func TestRegistryReturnsPluginsByInterface(t *testing.T) {
 	)
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "")
 	defer r.Close()
 
 	r.Report()
@@ -367,11 +417,7 @@ func TestRegistryHandlesConflictingPlugins(t *testing.T) {
 	)
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "")
 	defer r.Close()
 
 	r.Report()
@@ -413,18 +459,34 @@ func TestRegistryRejectsErroneousPluginResponses(t *testing.T) {
 			Name:    "nonJSONResponseBody",
 			Handler: stringHandler(http.StatusOK, `notJSON`),
 		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "changedID",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"differentID","label":"changedID","interfaces":["reporter"]}]}`),
+		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "moreThanOnePlugin",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"moreThanOnePlugin","label":"moreThanOnePlugin","interfaces":["reporter"]}, {"id":"haha","label":"haha","interfaces":["reporter"]}]}`),
+		}.file(),
 	)
 	defer restore(t)
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "")
 	defer r.Close()
 
 	r.Report()
 	checkLoadedPlugins(t, r.ForEach, []xfer.PluginSpec{
+		{
+			ID:     "changedID",
+			Label:  "changedID",
+			Status: `error: plugin must not change its id (is "differentID", should be "changedID")`,
+		},
+		{
+			ID:     "moreThanOnePlugin",
+			Label:  "moreThanOnePlugin",
+			Status: `error: report must contain exactly one plugin (found 2)`,
+		},
 		{
 			ID:         "noInterface",
 			Label:      "noInterface",
@@ -489,11 +551,7 @@ func TestRegistryRejectsPluginResponsesWhichAreTooLarge(t *testing.T) {
 		restore(t)
 	}()
 
-	root := "/plugins"
-	r, err := NewRegistry(root, "", nil)
-	if err != nil {
-		t.Fatal(err)
-	}
+	r := testRegistry(t, "")
 	defer r.Close()
 
 	r.Report()
@@ -501,3 +559,289 @@ func TestRegistryRejectsPluginResponsesWhichAreTooLarge(t *testing.T) {
 		{ID: "foo", Label: "foo", Status: `error: response must be shorter than 50MB`},
 	})
 }
+
+func TestRegistryChecksForValidPluginIDs(t *testing.T) {
+	setup(
+		t,
+		mockPlugin{
+			t:       t,
+			Name:    "testPlugin",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"testPlugin","label":"testPlugin","interfaces":["reporter"],"api_version":"1"}]}`),
+		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "P-L-U-G-I-N",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"P-L-U-G-I-N","label":"testPlugin","interfaces":["reporter"],"api_version":"1"}]}`),
+		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "another-testPlugin",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"another-testPlugin","label":"testPlugin","interfaces":["reporter"],"api_version":"1"}]}`),
+		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "testPlugin!",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"testPlugin!","label":"testPlugin","interfaces":["reporter"],"api_version":"1"}]}`),
+		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "test~plugin",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"test~plugin","label":"testPlugin","interfaces":["reporter"],"api_version":"1"}]}`),
+		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "testPlugin-",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"testPlugin-","label":"testPlugin","interfaces":["reporter"],"api_version":"1"}]}`),
+		}.file(),
+		mockPlugin{
+			t:       t,
+			Name:    "-testPlugin",
+			Handler: stringHandler(http.StatusOK, `{"Plugins":[{"id":"-testPlugin","label":"testPlugin","interfaces":["reporter"],"api_version":"1"}]}`),
+		}.file(),
+	)
+	defer restore(t)
+
+	r := testRegistry(t, "1")
+	defer r.Close()
+
+	r.Report()
+	checkLoadedPluginIDs(t, r.ForEach, []string{"P-L-U-G-I-N", "another-testPlugin", "testPlugin"})
+}
+
+func checkControls(t *testing.T, topology report.Topology, expectedControls, expectedNodeControls []string, nodeID string) {
+	controlsSet := report.MakeStringSet(expectedControls...)
+	for _, id := range controlsSet {
+		control, found := topology.Controls[id]
+		if !found {
+			t.Fatalf("Could not find an expected control %s in topology %s", id, topology.Label)
+		}
+		if control.ID != id {
+			t.Fatalf("Control ID mismatch, expected %s, got %s", id, control.ID)
+		}
+	}
+	if len(controlsSet) != len(topology.Controls) {
+		t.Fatalf("Expected exactly %d controls in topology, got %d", len(controlsSet), len(topology.Controls))
+	}
+
+	node, found := topology.Nodes[nodeID]
+	if !found {
+		t.Fatalf("expected a node %s in a topology", nodeID)
+	}
+	actualNodeControls := []string{}
+	node.LatestControls.ForEach(func(controlID string, _ time.Time, _ report.NodeControlData) {
+		actualNodeControls = append(actualNodeControls, controlID)
+	})
+	nodeControlsSet := report.MakeStringSet(expectedNodeControls...)
+	actualNodeControlsSet := report.MakeStringSet(actualNodeControls...)
+	if !reflect.DeepEqual(nodeControlsSet, actualNodeControlsSet) {
+		t.Fatalf("node controls in node %s in topology %s are not equal:\n%s", nodeID, topology.Label, test.Diff(nodeControlsSet, actualNodeControlsSet))
+	}
+}
+
+func control(index int) (string, string) {
+	return fmt.Sprintf("ctrl%d", index), fmt.Sprintf("Ctrl %d", index)
+}
+
+func controlID(index int) string {
+	ID, _ := control(index)
+	return ID
+}
+
+func mustMarshal(value interface{}) string {
+	buf := &bytes.Buffer{}
+	codec.NewEncoder(buf, &codec.JsonHandle{}).MustEncode(value)
+	return buf.String()
+}
+
+func mustUnmarshal(r io.Reader, value interface{}) {
+	codec.NewDecoder(r, &codec.JsonHandle{}).MustDecode(value)
+}
+
+func topologyControls(indices []int) report.Controls {
+	var controls []report.Control
+	for _, index := range indices {
+		ID, name := control(index)
+		controls = append(controls, report.Control{
+			ID:    ID,
+			Human: name,
+			Icon:  "fa-at",
+			Rank:  index,
+		})
+	}
+	rptControls := report.Controls{}
+	rptControls.AddControls(controls)
+	return rptControls
+}
+
+func nodeControls(indices []int) []string {
+	var IDs []string
+	for _, index := range indices {
+		ID, _ := control(index)
+		IDs = append(IDs, ID)
+	}
+	return IDs
+}
+
+func topologyWithControls(label, nodeID string, controlIndices, nodeControlIndices []int) report.Topology {
+	topology := report.MakeTopology().WithLabel(label, "")
+	topology.Controls = topologyControls(controlIndices)
+	return topology.AddNode(report.MakeNode(nodeID).WithLatestActiveControls(nodeControls(nodeControlIndices)...))
+}
+
+func pluginSpec(ID string, interfaces ...string) xfer.PluginSpec {
+	return xfer.PluginSpec{
+		ID:         ID,
+		Label:      ID,
+		Interfaces: interfaces,
+		APIVersion: "1",
+	}
+}
+
+func testReport(topology report.Topology, spec xfer.PluginSpec) report.Report {
+	rpt := report.MakeReport()
+	set := false
+	f := func(t *report.Topology) {
+		if t.Label != topology.Label {
+			return
+		}
+		if set {
+			panic("Two topologies with the same label")
+		}
+		set = true
+		*t = t.Merge(topology)
+	}
+	rpt.WalkTopologies(f)
+	if !set {
+		panic(fmt.Sprintf("%s name is not a valid topology label", topology.Label))
+	}
+	rpt.Plugins = xfer.MakePluginSpecs(spec)
+	return rpt
+}
+
+func TestRegistryRewritesControlReports(t *testing.T) {
+	setup(
+		t,
+		mockPlugin{
+			t:    t,
+			Name: "testPlugin",
+			Handler: mapStringHandler(testResponseMap{
+				"/report":  {http.StatusOK, mustMarshal(testReport(topologyWithControls("pod", "node1", []int{1}, []int{1, 2}), pluginSpec("testPlugin", "reporter", "controller")))},
+				"/control": {http.StatusOK, mustMarshal(PluginResponse{})},
+			}),
+		}.file(),
+		mockPlugin{
+			t:    t,
+			Name: "testPluginReporterOnly",
+			Handler: mapStringHandler(testResponseMap{
+				"/report": {http.StatusOK, mustMarshal(testReport(topologyWithControls("host", "node1", []int{1}, []int{1, 2}), pluginSpec("testPluginReporterOnly", "reporter")))},
+			}),
+		}.file(),
+	)
+	defer restore(t)
+
+	r := testRegistry(t, "1")
+	defer r.Close()
+
+	rpt, err := r.Report()
+	if err != nil {
+		t.Fatal(err)
+	}
+	// in a Pod topology, ctrl1 should be faked, ctrl2 should be left intact
+	expectedPodControls := []string{fakeControlID("testPlugin", controlID(1))}
+	expectedPodNodeControls := []string{fakeControlID("testPlugin", controlID(1)), controlID(2)}
+	checkControls(t, rpt.Pod, expectedPodControls, expectedPodNodeControls, "node1")
+	// in a Host topology, controls should be kept untouched
+	expectedHostControls := []string{controlID(1)}
+	expectedHostNodeControls := []string{controlID(1), controlID(2)}
+	checkControls(t, rpt.Host, expectedHostControls, expectedHostNodeControls, "node1")
+}
+
+func TestRegistryRegistersHandlers(t *testing.T) {
+	setup(
+		t,
+		mockPlugin{
+			t:    t,
+			Name: "testPlugin",
+			Handler: mapStringHandler(testResponseMap{
+				"/report":  {http.StatusOK, mustMarshal(testReport(topologyWithControls("pod", "node1", []int{1}, []int{1, 2}), pluginSpec("testPlugin", "reporter", "controller")))},
+				"/control": {http.StatusOK, mustMarshal(PluginResponse{})},
+			}),
+		}.file(),
+		mockPlugin{
+			t:    t,
+			Name: "testPlugin2",
+			Handler: mapStringHandler(testResponseMap{
+				"/report":  {http.StatusOK, mustMarshal(testReport(topologyWithControls("pod", "node2", []int{1, 2}, []int{1}), pluginSpec("testPlugin2", "reporter", "controller")))},
+				"/control": {http.StatusOK, mustMarshal(PluginResponse{})},
+			}),
+		}.file(),
+	)
+	defer restore(t)
+
+	testBackend := newTestHandlerRegistryBackend(t)
+	handlerRegistry := controls.NewHandlerRegistry(testBackend)
+	root := "/plugins"
+	r, err := NewRegistry(root, "1", nil, handlerRegistry, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	r.Report()
+	expectedLen := 3
+	if len(testBackend.handlers) != expectedLen {
+		t.Fatalf("Expected %d registered handler, got %d", expectedLen, len(testBackend.handlers))
+	}
+	fakeIDs := []string{
+		fakeControlID("testPlugin", controlID(1)),
+		fakeControlID("testPlugin2", controlID(1)),
+		fakeControlID("testPlugin2", controlID(2)),
+	}
+	for _, fakeID := range fakeIDs {
+		if _, found := testBackend.Handler(fakeID); !found {
+			t.Fatalf("Expected to have a handler for %s", fakeID)
+		}
+	}
+}
+
+func TestRegistryHandlersCallPlugins(t *testing.T) {
+	setup(
+		t,
+		mockPlugin{
+			t:    t,
+			Name: "testPlugin",
+			Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				switch r.URL.Path {
+				case "/report":
+					w.WriteHeader(http.StatusOK)
+					rpt := mustMarshal(testReport(topologyWithControls("pod", "node1", []int{1}, []int{1}), pluginSpec("testPlugin", "reporter", "controller")))
+					fmt.Fprint(w, rpt)
+				case "/control":
+					xreq := xfer.Request{}
+					mustUnmarshal(r.Body, &xreq)
+					w.WriteHeader(http.StatusOK)
+					fmt.Fprint(w, mustMarshal(PluginResponse{Response: xfer.Response{Value: fmt.Sprintf("%s,%s", xreq.NodeID, xreq.Control)}}))
+				default:
+					http.NotFound(w, r)
+				}
+			}),
+		}.file(),
+	)
+	defer restore(t)
+
+	handlerRegistry := controls.NewDefaultHandlerRegistry()
+	root := "/plugins"
+	r, err := NewRegistry(root, "1", nil, handlerRegistry, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer r.Close()
+
+	r.Report()
+	fakeID := fakeControlID("testPlugin", controlID(1))
+	req := xfer.Request{NodeID: "node1", Control: fakeID}
+	res := handlerRegistry.HandleControlRequest(req)
+	if res.Value != fmt.Sprintf("node1,%s", controlID(1)) {
+		t.Fatalf("Got unexpected response: %#v", res)
+	}
+}

probe/probe.go

@@ -200,7 +200,7 @@ ForLoop:
 		}
 	}
 
-	if err := p.publisher.Publish(rpt); err != nil {
+	if err := p.publisher.Publish(rpt.BackwardCompatible()); err != nil {
 		log.Infof("publish: %v", err)
 	}
 }

prog/probe.go

@@ -113,10 +113,11 @@ func probeMain(flags probeFlags) {
 		ProbeID:      probeID,
 		Insecure:     flags.insecure,
 	}
+	handlerRegistry := controls.NewDefaultHandlerRegistry()
 	clientFactory := func(hostname, endpoint string) (appclient.AppClient, error) {
 		return appclient.NewAppClient(
 			probeConfig, hostname, endpoint,
-			xfer.ControlHandlerFunc(controls.HandleControlRequest),
+			xfer.ControlHandlerFunc(handlerRegistry.HandleControlRequest),
 		)
 	}
 	clients := appclient.NewMultiAppClient(clientFactory, flags.noControls)
@@ -131,7 +132,7 @@ func probeMain(flags probeFlags) {
 
 	p := probe.New(flags.spyInterval, flags.publishInterval, clients, flags.noControls)
 
-	hostReporter := host.NewReporter(hostID, hostName, probeID, version, clients)
+	hostReporter := host.NewReporter(hostID, hostName, probeID, version, clients, handlerRegistry)
 	defer hostReporter.Stop()
 	p.AddReporter(hostReporter)
 	p.AddTagger(probe.NewTopologyTagger(), host.NewTagger(hostID))
@@ -157,7 +158,7 @@ func probeMain(flags probeFlags) {
 				log.Errorf("Docker: problem with bridge %s: %v", flags.dockerBridge, err)
 			}
 		}
-		if registry, err := docker.NewRegistry(flags.dockerInterval, clients, true, hostID); err == nil {
+		if registry, err := docker.NewRegistry(flags.dockerInterval, clients, true, hostID, handlerRegistry); err == nil {
 			defer registry.Stop()
 			if flags.procEnabled {
 				p.AddTagger(docker.NewTagger(registry, processCache))
@@ -171,7 +172,7 @@ func probeMain(flags probeFlags) {
 	if flags.kubernetesEnabled {
 		if client, err := kubernetes.NewClient(flags.kubernetesAPI, flags.kubernetesInterval); err == nil {
 			defer client.Stop()
-			reporter := kubernetes.NewReporter(client, clients, probeID, hostID, p)
+			reporter := kubernetes.NewReporter(client, clients, probeID, hostID, p, handlerRegistry)
 			defer reporter.Stop()
 			p.AddReporter(reporter)
 			p.AddTagger(reporter)
@@ -205,6 +206,8 @@ func probeMain(flags probeFlags) {
 			"probe_id":    probeID,
 			"api_version": pluginAPIVersion,
 		},
+		handlerRegistry,
+		p,
 	)
 	if err != nil {
 		log.Errorf("plugins: problem loading: %v", err)

render/detailed/node.go

@@ -2,6 +2,7 @@ package detailed
 
 import (
 	"sort"
+	"time"
 
 	"github.com/ugorji/go/codec"
 
@@ -98,20 +99,22 @@ func controlsFor(topology report.Topology, nodeID string) []ControlInstance {
 	if !ok {
 		return result
 	}
-
-	for _, id := range node.Controls.Controls {
-		if control, ok := topology.Controls[id]; ok {
-			probeID, ok := node.Latest.Lookup(report.ControlProbeID)
-			if !ok {
-				continue
-			}
+	probeID, ok := node.Latest.Lookup(report.ControlProbeID)
+	if !ok {
+		return result
+	}
+	node.LatestControls.ForEach(func(controlID string, _ time.Time, data report.NodeControlData) {
+		if data.Dead {
+			return
+		}
+		if control, ok := topology.Controls[controlID]; ok {
 			result = append(result, ControlInstance{
 				ProbeID: probeID,
 				NodeID:  nodeID,
 				Control: control,
 			})
 		}
-	}
+	})
 	return result
 }
 

report/controls.go

@@ -122,3 +122,9 @@ func (NodeControls) MarshalJSON() ([]byte, error) {
 func (*NodeControls) UnmarshalJSON(b []byte) error {
 	panic("UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead")
 }
+
+// NodeControlData contains specific information about the control. It
+// is used as a Value field of LatestEntry in NodeControlDataLatestMap.
+type NodeControlData struct {
+	Dead bool `json:"dead"`
+}

report/latest_map.go

@@ -10,20 +10,27 @@ import (
 	"github.com/weaveworks/ps"
 )
 
-// LatestMap is a persitent map which support latest-win merges. We have to
-// embed ps.Map as its an interface.  LatestMaps are immutable.
+// LatestEntryDecoder is an interface for decoding the LatestEntry instances.
+type LatestEntryDecoder interface {
+	Decode(decoder *codec.Decoder, entry *LatestEntry)
+}
+
+// LatestMap is a persistent map which support latest-win merges. We
+// have to embed ps.Map as its interface. LatestMaps are immutable.
 type LatestMap struct {
 	ps.Map
+	decoder LatestEntryDecoder
 }
 
 // LatestEntry represents a timestamped value inside the LatestMap.
 type LatestEntry struct {
-	Timestamp time.Time `json:"timestamp"`
-	Value     string    `json:"value"`
+	Timestamp time.Time   `json:"timestamp"`
+	Value     interface{} `json:"value"`
 }
 
+// String returns the LatestEntry's string representation.
 func (e LatestEntry) String() string {
-	return fmt.Sprintf("\"%s\" (%s)", e.Value, e.Timestamp.String())
+	return fmt.Sprintf("%v (%s)", e.Value, e.Timestamp.String())
 }
 
 // Equal returns true if the supplied LatestEntry is equal to this one.
@@ -31,12 +38,9 @@ func (e LatestEntry) Equal(e2 LatestEntry) bool {
 	return e.Timestamp.Equal(e2.Timestamp) && e.Value == e2.Value
 }
 
-// EmptyLatestMap is an empty LatestMap.  Start with this.
-var EmptyLatestMap = LatestMap{ps.NewMap()}
-
-// MakeLatestMap makes an empty LatestMap
-func MakeLatestMap() LatestMap {
-	return EmptyLatestMap
+// MakeLatestMapWithDecoder makes an empty LatestMap holding custom values.
+func MakeLatestMapWithDecoder(decoder LatestEntryDecoder) LatestMap {
+	return LatestMap{ps.NewMap(), decoder}
 }
 
 // Copy is a noop, as LatestMaps are immutable.
@@ -44,7 +48,7 @@ func (m LatestMap) Copy() LatestMap {
 	return m
 }
 
-// Size returns the number of elements
+// Size returns the number of elements.
 func (m LatestMap) Size() int {
 	if m.Map == nil {
 		return 0
@@ -52,8 +56,9 @@ func (m LatestMap) Size() int {
 	return m.Map.Size()
 }
 
-// Merge produces a fresh LatestMap, container the kers from both inputs. When
-// both inputs container the same key, the latter value is used.
+// Merge produces a fresh StringLatestMap containing the keys from
+// both inputs. When both inputs contain the same key, the newer value
+// is used.
 func (m LatestMap) Merge(other LatestMap) LatestMap {
 	var (
 		mSize     = m.Size()
@@ -69,6 +74,9 @@ func (m LatestMap) Merge(other LatestMap) LatestMap {
 	case mSize < otherSize:
 		output, iter = iter, output
 	}
+	if m.decoder != other.decoder {
+		panic(fmt.Sprintf("Cannot merge maps with different entry value types, this has %#v, other has %#v", m.decoder, other.decoder))
+	}
 
 	iter.ForEach(func(key string, iterVal interface{}) {
 		if existingVal, ok := output.Lookup(key); ok {
@@ -80,58 +88,59 @@ func (m LatestMap) Merge(other LatestMap) LatestMap {
 		}
 	})
 
-	return LatestMap{output}
+	return LatestMap{output, m.decoder}
 }
 
 // Lookup the value for the given key.
-func (m LatestMap) Lookup(key string) (string, bool) {
+func (m LatestMap) Lookup(key string) (interface{}, bool) {
 	v, _, ok := m.LookupEntry(key)
 	return v, ok
 }
 
 // LookupEntry returns the raw entry for the given key.
-func (m LatestMap) LookupEntry(key string) (string, time.Time, bool) {
+func (m LatestMap) LookupEntry(key string) (interface{}, time.Time, bool) {
 	if m.Map == nil {
-		return "", time.Time{}, false
+		return nil, time.Time{}, false
 	}
 	value, ok := m.Map.Lookup(key)
 	if !ok {
-		return "", time.Time{}, false
+		return nil, time.Time{}, false
 	}
 	e := value.(LatestEntry)
 	return e.Value, e.Timestamp, true
 }
 
-// Set the value for the given key.
-func (m LatestMap) Set(key string, timestamp time.Time, value string) LatestMap {
+// Set sets the value for the given key.
+func (m LatestMap) Set(key string, timestamp time.Time, value interface{}) LatestMap {
 	if m.Map == nil {
-		m = EmptyLatestMap
+		m = MakeLatestMapWithDecoder(m.decoder)
 	}
-	return LatestMap{m.Map.Set(key, LatestEntry{timestamp, value})}
+	return LatestMap{m.Map.Set(key, LatestEntry{timestamp, value}), m.decoder}
 }
 
 // Delete the value for the given key.
 func (m LatestMap) Delete(key string) LatestMap {
 	if m.Map == nil {
-		m = EmptyLatestMap
+		m = MakeLatestMapWithDecoder(m.decoder)
 	}
-	return LatestMap{m.Map.Delete(key)}
+	return LatestMap{m.Map.Delete(key), m.decoder}
 }
 
-// ForEach executes f on each key value pair in the map
-func (m LatestMap) ForEach(fn func(k, v string)) {
+// ForEach executes fn on each key, timestamp, value triple in the map.
+func (m LatestMap) ForEach(fn func(k string, ts time.Time, v interface{})) {
 	if m.Map == nil {
 		return
 	}
 	m.Map.ForEach(func(key string, value interface{}) {
-		fn(key, value.(LatestEntry).Value)
+		fn(key, value.(LatestEntry).Timestamp, value.(LatestEntry).Value)
 	})
 }
 
+// String returns the LatestMap's string representation.
 func (m LatestMap) String() string {
 	keys := []string{}
 	if m.Map == nil {
-		m = EmptyLatestMap
+		m = MakeLatestMapWithDecoder(m.decoder)
 	}
 	for _, k := range m.Map.Keys() {
 		keys = append(keys, k)
@@ -147,7 +156,7 @@ func (m LatestMap) String() string {
 	return buf.String()
 }
 
-// DeepEqual tests equality with other LatestMap
+// DeepEqual tests equality with other LatestMap.
 func (m LatestMap) DeepEqual(n LatestMap) bool {
 	if m.Size() != n.Size() {
 		return false
@@ -155,7 +164,9 @@ func (m LatestMap) DeepEqual(n LatestMap) bool {
 	if m.Size() == 0 {
 		return true
 	}
-
+	if m.decoder != n.decoder {
+		panic(fmt.Sprintf("Cannot check equality of maps with different entry value types, this has %#v, other has %#v", m.decoder, n.decoder))
+	}
 	equal := true
 	m.Map.ForEach(func(k string, val interface{}) {
 		if otherValue, ok := n.Map.Lookup(k); !ok {
@@ -177,7 +188,7 @@ func (m LatestMap) toIntermediate() map[string]LatestEntry {
 	return intermediate
 }
 
-// CodecEncodeSelf implements codec.Selfer
+// CodecEncodeSelf implements codec.Selfer.
 func (m *LatestMap) CodecEncodeSelf(encoder *codec.Encoder) {
 	if m.Map != nil {
 		encoder.Encode(m.toIntermediate())
@@ -193,7 +204,7 @@ const (
 	containerMapEnd   = 4
 )
 
-// CodecDecodeSelf implements codec.Selfer
+// CodecDecodeSelf implements codec.Selfer.
 // This implementation does not use the intermediate form as that was a
 // performance issue; skipping it saved almost 10% CPU.  Note this means
 // we are using undocumented, internal APIs, which could break in the future.
@@ -201,7 +212,7 @@ const (
 func (m *LatestMap) CodecDecodeSelf(decoder *codec.Decoder) {
 	z, r := codec.GenHelperDecoder(decoder)
 	if r.TryDecodeAsNil() {
-		*m = LatestMap{}
+		*m = MakeLatestMapWithDecoder(m.decoder)
 		return
 	}
 
@@ -221,21 +232,21 @@ func (m *LatestMap) CodecDecodeSelf(decoder *codec.Decoder) {
 		var value LatestEntry
 		z.DecSendContainerState(containerMapValue)
 		if !r.TryDecodeAsNil() {
-			decoder.Decode(&value)
+			m.decoder.Decode(decoder, &value)
 		}
 
 		out = out.UnsafeMutableSet(key, value)
 	}
 	z.DecSendContainerState(containerMapEnd)
-	*m = LatestMap{out}
+	*m = LatestMap{out, m.decoder}
 }
 
-// MarshalJSON shouldn't be used, use CodecEncodeSelf instead
+// MarshalJSON shouldn't be used, use CodecEncodeSelf instead.
 func (LatestMap) MarshalJSON() ([]byte, error) {
 	panic("MarshalJSON shouldn't be used, use CodecEncodeSelf instead")
 }
 
-// UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead
+// UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead.
 func (*LatestMap) UnmarshalJSON(b []byte) error {
 	panic("UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead")
 }

report/latest_map_generated.go

@@ -0,0 +1,238 @@
+// Generated file, do not edit.
+// To regenerate, run ./tools/generate_latest_map ./report/latest_map_generated.go string NodeControlData
+
+package report
+
+import (
+	"time"
+
+	"github.com/ugorji/go/codec"
+)
+
+type wireStringLatestEntry struct {
+	Timestamp time.Time `json:"timestamp"`
+	Value     string    `json:"value"`
+}
+
+type stringLatestEntryDecoder struct{}
+
+func (d *stringLatestEntryDecoder) Decode(decoder *codec.Decoder, entry *LatestEntry) {
+	wire := wireStringLatestEntry{}
+	decoder.Decode(&wire)
+	entry.Timestamp = wire.Timestamp
+	entry.Value = wire.Value
+}
+
+// StringLatestEntryDecoder is an implementation of LatestEntryDecoder
+// that decodes the LatestEntry instances having a string value.
+var StringLatestEntryDecoder LatestEntryDecoder = &stringLatestEntryDecoder{}
+
+// StringLatestMap holds latest string instances.
+type StringLatestMap LatestMap
+
+// EmptyStringLatestMap is an empty StringLatestMap. Start with this.
+var EmptyStringLatestMap = (StringLatestMap)(MakeLatestMapWithDecoder(StringLatestEntryDecoder))
+
+// MakeStringLatestMap makes an empty StringLatestMap.
+func MakeStringLatestMap() StringLatestMap {
+	return EmptyStringLatestMap
+}
+
+// Copy is a noop, as StringLatestMaps are immutable.
+func (m StringLatestMap) Copy() StringLatestMap {
+	return (StringLatestMap)((LatestMap)(m).Copy())
+}
+
+// Size returns the number of elements.
+func (m StringLatestMap) Size() int {
+	return (LatestMap)(m).Size()
+}
+
+// Merge produces a fresh StringLatestMap containing the keys from both inputs.
+// When both inputs contain the same key, the newer value is used.
+func (m StringLatestMap) Merge(other StringLatestMap) StringLatestMap {
+	return (StringLatestMap)((LatestMap)(m).Merge((LatestMap)(other)))
+}
+
+// Lookup the value for the given key.
+func (m StringLatestMap) Lookup(key string) (string, bool) {
+	v, ok := (LatestMap)(m).Lookup(key)
+	if !ok {
+		var zero string
+		return zero, false
+	}
+	return v.(string), true
+}
+
+// LookupEntry returns the raw entry for the given key.
+func (m StringLatestMap) LookupEntry(key string) (string, time.Time, bool) {
+	v, timestamp, ok := (LatestMap)(m).LookupEntry(key)
+	if !ok {
+		var zero string
+		return zero, timestamp, false
+	}
+	return v.(string), timestamp, true
+}
+
+// Set the value for the given key.
+func (m StringLatestMap) Set(key string, timestamp time.Time, value string) StringLatestMap {
+	return (StringLatestMap)((LatestMap)(m).Set(key, timestamp, value))
+}
+
+// Delete the value for the given key.
+func (m StringLatestMap) Delete(key string) StringLatestMap {
+	return (StringLatestMap)((LatestMap)(m).Delete(key))
+}
+
+// ForEach executes fn on each key value pair in the map.
+func (m StringLatestMap) ForEach(fn func(k string, timestamp time.Time, v string)) {
+	(LatestMap)(m).ForEach(func(key string, ts time.Time, value interface{}) {
+		fn(key, ts, value.(string))
+	})
+}
+
+// String returns the StringLatestMap's string representation.
+func (m StringLatestMap) String() string {
+	return (LatestMap)(m).String()
+}
+
+// DeepEqual tests equality with other StringLatestMap.
+func (m StringLatestMap) DeepEqual(n StringLatestMap) bool {
+	return (LatestMap)(m).DeepEqual((LatestMap)(n))
+}
+
+// CodecEncodeSelf implements codec.Selfer.
+func (m *StringLatestMap) CodecEncodeSelf(encoder *codec.Encoder) {
+	(*LatestMap)(m).CodecEncodeSelf(encoder)
+}
+
+// CodecDecodeSelf implements codec.Selfer.
+func (m *StringLatestMap) CodecDecodeSelf(decoder *codec.Decoder) {
+	bm := (*LatestMap)(m)
+	bm.decoder = StringLatestEntryDecoder
+	bm.CodecDecodeSelf(decoder)
+}
+
+// MarshalJSON shouldn't be used, use CodecEncodeSelf instead.
+func (StringLatestMap) MarshalJSON() ([]byte, error) {
+	panic("MarshalJSON shouldn't be used, use CodecEncodeSelf instead")
+}
+
+// UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead.
+func (*StringLatestMap) UnmarshalJSON(b []byte) error {
+	panic("UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead")
+}
+
+type wireNodeControlDataLatestEntry struct {
+	Timestamp time.Time       `json:"timestamp"`
+	Value     NodeControlData `json:"value"`
+}
+
+type nodeControlDataLatestEntryDecoder struct{}
+
+func (d *nodeControlDataLatestEntryDecoder) Decode(decoder *codec.Decoder, entry *LatestEntry) {
+	wire := wireNodeControlDataLatestEntry{}
+	decoder.Decode(&wire)
+	entry.Timestamp = wire.Timestamp
+	entry.Value = wire.Value
+}
+
+// NodeControlDataLatestEntryDecoder is an implementation of LatestEntryDecoder
+// that decodes the LatestEntry instances having a NodeControlData value.
+var NodeControlDataLatestEntryDecoder LatestEntryDecoder = &nodeControlDataLatestEntryDecoder{}
+
+// NodeControlDataLatestMap holds latest NodeControlData instances.
+type NodeControlDataLatestMap LatestMap
+
+// EmptyNodeControlDataLatestMap is an empty NodeControlDataLatestMap. Start with this.
+var EmptyNodeControlDataLatestMap = (NodeControlDataLatestMap)(MakeLatestMapWithDecoder(NodeControlDataLatestEntryDecoder))
+
+// MakeNodeControlDataLatestMap makes an empty NodeControlDataLatestMap.
+func MakeNodeControlDataLatestMap() NodeControlDataLatestMap {
+	return EmptyNodeControlDataLatestMap
+}
+
+// Copy is a noop, as NodeControlDataLatestMaps are immutable.
+func (m NodeControlDataLatestMap) Copy() NodeControlDataLatestMap {
+	return (NodeControlDataLatestMap)((LatestMap)(m).Copy())
+}
+
+// Size returns the number of elements.
+func (m NodeControlDataLatestMap) Size() int {
+	return (LatestMap)(m).Size()
+}
+
+// Merge produces a fresh NodeControlDataLatestMap containing the keys from both inputs.
+// When both inputs contain the same key, the newer value is used.
+func (m NodeControlDataLatestMap) Merge(other NodeControlDataLatestMap) NodeControlDataLatestMap {
+	return (NodeControlDataLatestMap)((LatestMap)(m).Merge((LatestMap)(other)))
+}
+
+// Lookup the value for the given key.
+func (m NodeControlDataLatestMap) Lookup(key string) (NodeControlData, bool) {
+	v, ok := (LatestMap)(m).Lookup(key)
+	if !ok {
+		var zero NodeControlData
+		return zero, false
+	}
+	return v.(NodeControlData), true
+}
+
+// LookupEntry returns the raw entry for the given key.
+func (m NodeControlDataLatestMap) LookupEntry(key string) (NodeControlData, time.Time, bool) {
+	v, timestamp, ok := (LatestMap)(m).LookupEntry(key)
+	if !ok {
+		var zero NodeControlData
+		return zero, timestamp, false
+	}
+	return v.(NodeControlData), timestamp, true
+}
+
+// Set the value for the given key.
+func (m NodeControlDataLatestMap) Set(key string, timestamp time.Time, value NodeControlData) NodeControlDataLatestMap {
+	return (NodeControlDataLatestMap)((LatestMap)(m).Set(key, timestamp, value))
+}
+
+// Delete the value for the given key.
+func (m NodeControlDataLatestMap) Delete(key string) NodeControlDataLatestMap {
+	return (NodeControlDataLatestMap)((LatestMap)(m).Delete(key))
+}
+
+// ForEach executes fn on each key value pair in the map.
+func (m NodeControlDataLatestMap) ForEach(fn func(k string, timestamp time.Time, v NodeControlData)) {
+	(LatestMap)(m).ForEach(func(key string, ts time.Time, value interface{}) {
+		fn(key, ts, value.(NodeControlData))
+	})
+}
+
+// String returns the NodeControlDataLatestMap's string representation.
+func (m NodeControlDataLatestMap) String() string {
+	return (LatestMap)(m).String()
+}
+
+// DeepEqual tests equality with other NodeControlDataLatestMap.
+func (m NodeControlDataLatestMap) DeepEqual(n NodeControlDataLatestMap) bool {
+	return (LatestMap)(m).DeepEqual((LatestMap)(n))
+}
+
+// CodecEncodeSelf implements codec.Selfer.
+func (m *NodeControlDataLatestMap) CodecEncodeSelf(encoder *codec.Encoder) {
+	(*LatestMap)(m).CodecEncodeSelf(encoder)
+}
+
+// CodecDecodeSelf implements codec.Selfer.
+func (m *NodeControlDataLatestMap) CodecDecodeSelf(decoder *codec.Decoder) {
+	bm := (*LatestMap)(m)
+	bm.decoder = NodeControlDataLatestEntryDecoder
+	bm.CodecDecodeSelf(decoder)
+}
+
+// MarshalJSON shouldn't be used, use CodecEncodeSelf instead.
+func (NodeControlDataLatestMap) MarshalJSON() ([]byte, error) {
+	panic("MarshalJSON shouldn't be used, use CodecEncodeSelf instead")
+}
+
+// UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead.
+func (*NodeControlDataLatestMap) UnmarshalJSON(b []byte) error {
+	panic("UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead")
+}

report/latest_map_internal_test.go

@@ -14,7 +14,7 @@ import (
 
 func TestLatestMapAdd(t *testing.T) {
 	now := time.Now()
-	have := EmptyLatestMap.
+	have := EmptyStringLatestMap.
 		Set("foo", now.Add(-1), "Baz").
 		Set("foo", now, "Bar")
 	if v, ok := have.Lookup("foo"); !ok || v != "Bar" {
@@ -23,7 +23,7 @@ func TestLatestMapAdd(t *testing.T) {
 	if v, ok := have.Lookup("bar"); ok || v != "" {
 		t.Errorf("v != nil")
 	}
-	have.ForEach(func(k, v string) {
+	have.ForEach(func(k string, _ time.Time, v string) {
 		if k != "foo" || v != "Bar" {
 			t.Errorf("v != Bar")
 		}
@@ -33,7 +33,7 @@ func TestLatestMapAdd(t *testing.T) {
 func TestLatestMapLookupEntry(t *testing.T) {
 	now := time.Now()
 	entry := LatestEntry{Timestamp: now, Value: "Bar"}
-	have := EmptyLatestMap.Set("foo", entry.Timestamp, entry.Value)
+	have := EmptyStringLatestMap.Set("foo", entry.Timestamp, entry.Value.(string))
 	if got, timestamp, ok := have.LookupEntry("foo"); !ok || got != entry.Value || !timestamp.Equal(entry.Timestamp) {
 		t.Errorf("got: %#v %v != expected %#v", got, timestamp, entry)
 	}
@@ -44,7 +44,7 @@ func TestLatestMapLookupEntry(t *testing.T) {
 
 func TestLatestMapAddNil(t *testing.T) {
 	now := time.Now()
-	have := LatestMap{}.Set("foo", now, "Bar")
+	have := StringLatestMap{}.Set("foo", now, "Bar")
 	if v, ok := have.Lookup("foo"); !ok || v != "Bar" {
 		t.Errorf("v != Bar")
 	}
@@ -52,14 +52,14 @@ func TestLatestMapAddNil(t *testing.T) {
 
 func TestLatestMapDeepEquals(t *testing.T) {
 	now := time.Now()
-	want := EmptyLatestMap.
+	want := EmptyStringLatestMap.
 		Set("foo", now, "Bar")
-	have := EmptyLatestMap.
+	have := EmptyStringLatestMap.
 		Set("foo", now, "Bar")
 	if !reflect.DeepEqual(want, have) {
 		t.Errorf(test.Diff(want, have))
 	}
-	notequal := EmptyLatestMap.
+	notequal := EmptyStringLatestMap.
 		Set("foo", now, "Baz")
 	if reflect.DeepEqual(want, notequal) {
 		t.Errorf(test.Diff(want, have))
@@ -68,8 +68,8 @@ func TestLatestMapDeepEquals(t *testing.T) {
 
 func TestLatestMapDelete(t *testing.T) {
 	now := time.Now()
-	want := EmptyLatestMap
-	have := EmptyLatestMap.
+	want := EmptyStringLatestMap
+	have := EmptyStringLatestMap.
 		Set("foo", now, "Baz").
 		Delete("foo")
 	if !reflect.DeepEqual(want, have) {
@@ -78,54 +78,60 @@ func TestLatestMapDelete(t *testing.T) {
 }
 
 func TestLatestMapDeleteNil(t *testing.T) {
-	want := LatestMap{}
-	have := LatestMap{}.Delete("foo")
+	want := StringLatestMap{}
+	have := StringLatestMap{}.Delete("foo")
 	if !reflect.DeepEqual(want, have) {
 		t.Errorf(test.Diff(want, have))
 	}
 }
 
+func nilStringLatestMap() StringLatestMap {
+	m := EmptyStringLatestMap
+	m.Map = nil
+	return m
+}
+
 func TestLatestMapMerge(t *testing.T) {
 	now := time.Now()
 	then := now.Add(-1)
 
 	for name, c := range map[string]struct {
-		a, b, want LatestMap
+		a, b, want StringLatestMap
 	}{
 		"nils": {
-			a:    LatestMap{},
-			b:    LatestMap{},
-			want: LatestMap{},
+			a:    nilStringLatestMap(),
+			b:    nilStringLatestMap(),
+			want: nilStringLatestMap(),
 		},
 		"Empty a": {
-			a: EmptyLatestMap,
-			b: EmptyLatestMap.
+			a: EmptyStringLatestMap,
+			b: EmptyStringLatestMap.
 				Set("foo", now, "bar"),
-			want: EmptyLatestMap.
+			want: EmptyStringLatestMap.
 				Set("foo", now, "bar"),
 		},
 		"Empty b": {
-			a: EmptyLatestMap.
+			a: EmptyStringLatestMap.
 				Set("foo", now, "bar"),
-			b: EmptyLatestMap,
-			want: EmptyLatestMap.
+			b: EmptyStringLatestMap,
+			want: EmptyStringLatestMap.
 				Set("foo", now, "bar"),
 		},
 		"Disjoint a & b": {
-			a: EmptyLatestMap.
+			a: EmptyStringLatestMap.
 				Set("foo", now, "bar"),
-			b: EmptyLatestMap.
+			b: EmptyStringLatestMap.
 				Set("baz", now, "bop"),
-			want: EmptyLatestMap.
+			want: EmptyStringLatestMap.
 				Set("foo", now, "bar").
 				Set("baz", now, "bop"),
 		},
 		"Common a & b": {
-			a: EmptyLatestMap.
+			a: EmptyStringLatestMap.
 				Set("foo", now, "bar"),
-			b: EmptyLatestMap.
+			b: EmptyStringLatestMap.
 				Set("foo", then, "baz"),
-			want: EmptyLatestMap.
+			want: EmptyStringLatestMap.
 				Set("foo", now, "bar"),
 		},
 	} {
@@ -137,8 +143,8 @@ func TestLatestMapMerge(t *testing.T) {
 
 func BenchmarkLatestMapMerge(b *testing.B) {
 	var (
-		left  = EmptyLatestMap
-		right = EmptyLatestMap
+		left  = EmptyStringLatestMap
+		right = EmptyStringLatestMap
 		now   = time.Now()
 	)
 
@@ -159,7 +165,7 @@ func BenchmarkLatestMapMerge(b *testing.B) {
 
 func TestLatestMapEncoding(t *testing.T) {
 	now := time.Now()
-	want := EmptyLatestMap.
+	want := EmptyStringLatestMap.
 		Set("foo", now, "bar").
 		Set("bar", now, "baz")
 
@@ -171,7 +177,7 @@ func TestLatestMapEncoding(t *testing.T) {
 		encoder := codec.NewEncoder(buf, h)
 		want.CodecEncodeSelf(encoder)
 		decoder := codec.NewDecoder(buf, h)
-		have := EmptyLatestMap
+		have := EmptyStringLatestMap
 		have.CodecDecodeSelf(decoder)
 		if !reflect.DeepEqual(want, have) {
 			t.Error(test.Diff(want, have))
@@ -181,7 +187,7 @@ func TestLatestMapEncoding(t *testing.T) {
 }
 
 func TestLatestMapEncodingNil(t *testing.T) {
-	want := LatestMap{}
+	want := nilStringLatestMap()
 
 	for _, h := range []codec.Handle{
 		codec.Handle(&codec.MsgpackHandle{}),
@@ -191,7 +197,7 @@ func TestLatestMapEncodingNil(t *testing.T) {
 		encoder := codec.NewEncoder(buf, h)
 		want.CodecEncodeSelf(encoder)
 		decoder := codec.NewDecoder(buf, h)
-		have := EmptyLatestMap
+		have := EmptyStringLatestMap
 		have.CodecDecodeSelf(decoder)
 		if !reflect.DeepEqual(want, have) {
 			t.Error(test.Diff(want, have))
@@ -199,3 +205,31 @@ func TestLatestMapEncodingNil(t *testing.T) {
 	}
 
 }
+
+func TestLatestMapMergeEqualDecoderTypes(t *testing.T) {
+	defer func() {
+		if r := recover(); r != nil {
+			t.Error("Merging two maps with the same decoders should not panic")
+		}
+	}()
+	m1 := MakeStringLatestMap().Set("a", time.Now(), "bar")
+	m2 := MakeStringLatestMap().Set("b", time.Now(), "foo")
+	m1.Merge(m2)
+}
+
+type TestLatestEntryDecoder struct{}
+
+func (d *TestLatestEntryDecoder) Decode(decoder *codec.Decoder, entry *LatestEntry) {
+	decoder.Decode(entry)
+}
+
+func TestLatestMapMergeDifferentDecoderTypes(t *testing.T) {
+	defer func() {
+		if r := recover(); r == nil {
+			t.Error("Merging two maps with different decoders should panic")
+		}
+	}()
+	m1 := MakeStringLatestMap().Set("a", time.Now(), "bar")
+	m2 := ((StringLatestMap)(MakeLatestMapWithDecoder(&TestLatestEntryDecoder{}))).Set("b", time.Now(), "foo")
+	m1.Merge(m2)
+}

report/node.go

@@ -10,31 +10,33 @@ import (
 // given node in a given topology, along with the edges emanating from the
 // node and metadata about those edges.
 type Node struct {
-	ID        string        `json:"id,omitempty"`
-	Topology  string        `json:"topology,omitempty"`
-	Counters  Counters      `json:"counters,omitempty"`
-	Sets      Sets          `json:"sets,omitempty"`
-	Adjacency IDList        `json:"adjacency"`
-	Edges     EdgeMetadatas `json:"edges,omitempty"`
-	Controls  NodeControls  `json:"controls,omitempty"`
-	Latest    LatestMap     `json:"latest,omitempty"`
-	Metrics   Metrics       `json:"metrics,omitempty"`
-	Parents   Sets          `json:"parents,omitempty"`
-	Children  NodeSet       `json:"children,omitempty"`
+	ID             string                   `json:"id,omitempty"`
+	Topology       string                   `json:"topology,omitempty"`
+	Counters       Counters                 `json:"counters,omitempty"`
+	Sets           Sets                     `json:"sets,omitempty"`
+	Adjacency      IDList                   `json:"adjacency"`
+	Edges          EdgeMetadatas            `json:"edges,omitempty"`
+	Controls       NodeControls             `json:"controls,omitempty"`
+	LatestControls NodeControlDataLatestMap `json:"latestControls,omitempty"`
+	Latest         StringLatestMap          `json:"latest,omitempty"`
+	Metrics        Metrics                  `json:"metrics,omitempty"`
+	Parents        Sets                     `json:"parents,omitempty"`
+	Children       NodeSet                  `json:"children,omitempty"`
 }
 
 // MakeNode creates a new Node with no initial metadata.
 func MakeNode(id string) Node {
 	return Node{
-		ID:        id,
-		Counters:  EmptyCounters,
-		Sets:      EmptySets,
-		Adjacency: EmptyIDList,
-		Edges:     EmptyEdgeMetadatas,
-		Controls:  MakeNodeControls(),
-		Latest:    EmptyLatestMap,
-		Metrics:   Metrics{},
-		Parents:   EmptySets,
+		ID:             id,
+		Counters:       EmptyCounters,
+		Sets:           EmptySets,
+		Adjacency:      EmptyIDList,
+		Edges:          EmptyEdgeMetadatas,
+		Controls:       MakeNodeControls(),
+		LatestControls: EmptyNodeControlDataLatestMap,
+		Latest:         EmptyStringLatestMap,
+		Metrics:        Metrics{},
+		Parents:        EmptySets,
 	}
 }
 
@@ -136,6 +138,30 @@ func (n Node) WithControls(cs ...string) Node {
 	return n
 }
 
+// WithLatestActiveControls returns a fresh copy of n, with active controls cs added to LatestControls.
+func (n Node) WithLatestActiveControls(cs ...string) Node {
+	lcs := map[string]NodeControlData{}
+	for _, control := range cs {
+		lcs[control] = NodeControlData{}
+	}
+	return n.WithLatestControls(lcs)
+}
+
+// WithLatestControls returns a fresh copy of n, with lcs added to LatestControls.
+func (n Node) WithLatestControls(lcs map[string]NodeControlData) Node {
+	ts := mtime.Now()
+	for k, v := range lcs {
+		n.LatestControls = n.LatestControls.Set(k, ts, v)
+	}
+	return n
+}
+
+// WithLatestControl produces a new Node with control added to it
+func (n Node) WithLatestControl(control string, ts time.Time, data NodeControlData) Node {
+	n.LatestControls = n.LatestControls.Set(control, ts, data)
+	return n
+}
+
 // WithParents returns a fresh copy of n, with sets merged in.
 func (n Node) WithParents(parents Sets) Node {
 	n.Parents = n.Parents.Merge(parents)
@@ -174,16 +200,17 @@ func (n Node) Merge(other Node) Node {
 		panic("Cannot merge nodes with different topology types: " + topology + " != " + other.Topology)
 	}
 	return Node{
-		ID:        id,
-		Topology:  topology,
-		Counters:  n.Counters.Merge(other.Counters),
-		Sets:      n.Sets.Merge(other.Sets),
-		Adjacency: n.Adjacency.Merge(other.Adjacency),
-		Edges:     n.Edges.Merge(other.Edges),
-		Controls:  n.Controls.Merge(other.Controls),
-		Latest:    n.Latest.Merge(other.Latest),
-		Metrics:   n.Metrics.Merge(other.Metrics),
-		Parents:   n.Parents.Merge(other.Parents),
-		Children:  n.Children.Merge(other.Children),
+		ID:             id,
+		Topology:       topology,
+		Counters:       n.Counters.Merge(other.Counters),
+		Sets:           n.Sets.Merge(other.Sets),
+		Adjacency:      n.Adjacency.Merge(other.Adjacency),
+		Edges:          n.Edges.Merge(other.Edges),
+		Controls:       n.Controls.Merge(other.Controls),
+		LatestControls: n.LatestControls.Merge(other.LatestControls),
+		Latest:         n.Latest.Merge(other.Latest),
+		Metrics:        n.Metrics.Merge(other.Metrics),
+		Parents:        n.Parents.Merge(other.Parents),
+		Children:       n.Children.Merge(other.Children),
 	}
 }

report/report.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/weaveworks/scope/common/mtime"
 	"github.com/weaveworks/scope/common/xfer"
 )
 
@@ -254,6 +255,57 @@ func (r Report) Validate() error {
 	return nil
 }
 
+// Upgrade returns a new report based on a report received from the old probe.
+//
+// This for now creates node's LatestControls from Controls.
+func (r Report) Upgrade() Report {
+	cp := r.Copy()
+	ncd := NodeControlData{
+		Dead: false,
+	}
+	cp.WalkTopologies(func(topology *Topology) {
+		n := Nodes{}
+		for name, node := range topology.Nodes {
+			if node.LatestControls.Size() == 0 && len(node.Controls.Controls) > 0 {
+				for _, control := range node.Controls.Controls {
+					node.LatestControls = node.LatestControls.Set(control, node.Controls.Timestamp, ncd)
+				}
+			}
+			n[name] = node
+		}
+		topology.Nodes = n
+	})
+	return cp
+}
+
+// BackwardCompatible returns a new backward-compatible report.
+//
+// This for now creates node's Controls from LatestControls.
+func (r Report) BackwardCompatible() Report {
+	now := mtime.Now()
+	cp := r.Copy()
+	cp.WalkTopologies(func(topology *Topology) {
+		n := Nodes{}
+		for name, node := range topology.Nodes {
+			var controls []string
+			node.LatestControls.ForEach(func(k string, _ time.Time, v NodeControlData) {
+				if !v.Dead {
+					controls = append(controls, k)
+				}
+			})
+			if len(controls) > 0 {
+				node.Controls = NodeControls{
+					Timestamp: now,
+					Controls:  MakeStringSet(controls...),
+				}
+			}
+			n[name] = node
+		}
+		topology.Nodes = n
+	})
+	return cp
+}
+
 // Sampling describes how the packet data sources for this report were
 // sampled. It can be used to calculate effective sample rates. We can't
 // just put the rate here, because that can't be accurately merged. Counts

report/report_test.go

@@ -3,8 +3,12 @@ package report_test
 import (
 	"reflect"
 	"testing"
+	"time"
 
+	"github.com/weaveworks/scope/common/mtime"
 	"github.com/weaveworks/scope/report"
+	"github.com/weaveworks/scope/test"
+	s_reflect "github.com/weaveworks/scope/test/reflect"
 )
 
 func newu64(value uint64) *uint64 { return &value }
@@ -74,3 +78,46 @@ func TestNode(t *testing.T) {
 		}
 	}
 }
+
+func TestReportBackwardCompatibility(t *testing.T) {
+	mtime.NowForce(time.Now())
+	defer mtime.NowReset()
+	rpt := report.MakeReport()
+	controls := map[string]report.NodeControlData{
+		"dead": {
+			Dead: true,
+		},
+		"alive": {
+			Dead: false,
+		},
+	}
+	node := report.MakeNode("foo").WithLatestControls(controls)
+	expectedNode := node.WithControls("alive")
+	rpt.Pod.AddNode(node)
+	expected := report.MakeReport()
+	expected.Pod.AddNode(expectedNode)
+	got := rpt.BackwardCompatible()
+	if !s_reflect.DeepEqual(expected, got) {
+		t.Error(test.Diff(expected, got))
+	}
+}
+
+func TestReportUpgrade(t *testing.T) {
+	mtime.NowForce(time.Now())
+	defer mtime.NowReset()
+	node := report.MakeNode("foo").WithControls("alive")
+	controls := map[string]report.NodeControlData{
+		"alive": {
+			Dead: false,
+		},
+	}
+	expectedNode := node.WithLatestControls(controls)
+	rpt := report.MakeReport()
+	rpt.Pod.AddNode(node)
+	expected := report.MakeReport()
+	expected.Pod.AddNode(expectedNode)
+	got := rpt.Upgrade()
+	if !s_reflect.DeepEqual(expected, got) {
+		t.Error(test.Diff(expected, got))
+	}
+}

report/table.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 
 	log "github.com/Sirupsen/logrus"
 	"github.com/weaveworks/scope/common/mtime"
@@ -37,7 +38,7 @@ func (node Node) AddTable(prefix string, labels map[string]string) Node {
 func (node Node) ExtractTable(prefix string) (rows map[string]string, truncationCount int) {
 	rows = map[string]string{}
 	truncationCount = 0
-	node.Latest.ForEach(func(key, value string) {
+	node.Latest.ForEach(func(key string, _ time.Time, value string) {
 		if strings.HasPrefix(key, prefix) {
 			label := key[len(prefix):]
 			rows[label] = value

test/diff.go

@@ -5,15 +5,15 @@ import (
 	"github.com/pmezard/go-difflib/difflib"
 )
 
-func init() {
-	spew.Config.SortKeys = true // :\
-}
-
 // Diff diffs two arbitrary data structures, giving human-readable output.
 func Diff(want, have interface{}) string {
+	config := spew.NewDefaultConfig()
+	config.ContinueOnMethod = true
+	config.SortKeys = true
+	config.SpewKeys = true
 	text, _ := difflib.GetUnifiedDiffString(difflib.UnifiedDiff{
-		A:        difflib.SplitLines(spew.Sdump(want)),
-		B:        difflib.SplitLines(spew.Sdump(have)),
+		A:        difflib.SplitLines(config.Sdump(want)),
+		B:        difflib.SplitLines(config.Sdump(have)),
 		FromFile: "want",
 		ToFile:   "have",
 		Context:  3,

tools/generate_latest_map

@@ -0,0 +1,176 @@
+#!/usr/bin/env bash
+#
+# Generate concrete implementations of LatestMap.
+#
+# e.g.
+#     $ generate_latest_map ./report/out.go string NodeControlData ...
+#
+# Depends on:
+# - gofmt
+
+function generate_header {
+    local out_file="${1}"
+    local cmd="${2}"
+
+    cat << EOF >"${out_file}"
+    // Generated file, do not edit.
+    // To regenerate, run ${cmd}
+
+    package report
+
+    import (
+        "time"
+
+        "github.com/ugorji/go/codec"
+    )
+EOF
+}
+
+function generate_latest_map {
+    local out_file="$1"
+    local data_type="$2"
+    local uppercase_data_type="${data_type^}"
+    local lowercase_data_type="${data_type,}"
+    local wire_entry_type="wire${uppercase_data_type}LatestEntry"
+    local decoder_type="${lowercase_data_type}LatestEntryDecoder"
+    local iface_decoder_variable="${uppercase_data_type}LatestEntryDecoder"
+    local latest_map_type="${uppercase_data_type}LatestMap"
+    local empty_latest_map_variable="Empty${latest_map_type}"
+    local make_function="Make${latest_map_type}"
+
+    local json_timestamp='`json:"timestamp"`'
+    local json_value='`json:"value"`'
+
+    cat << EOF >>"${out_file}"
+    type ${wire_entry_type} struct {
+        Timestamp time.Time    ${json_timestamp}
+        Value     ${data_type} ${json_value}
+    }
+
+    type ${decoder_type} struct {}
+
+    func (d *${decoder_type}) Decode(decoder *codec.Decoder, entry *LatestEntry) {
+        wire := ${wire_entry_type}{}
+        decoder.Decode(&wire)
+        entry.Timestamp = wire.Timestamp
+        entry.Value = wire.Value
+    }
+
+    // ${iface_decoder_variable} is an implementation of LatestEntryDecoder
+    // that decodes the LatestEntry instances having a ${data_type} value.
+    var ${iface_decoder_variable} LatestEntryDecoder = &${decoder_type}{}
+
+    // ${latest_map_type} holds latest ${data_type} instances.
+    type ${latest_map_type} LatestMap
+
+    // ${empty_latest_map_variable} is an empty ${latest_map_type}. Start with this.
+    var ${empty_latest_map_variable} = (${latest_map_type})(MakeLatestMapWithDecoder(${iface_decoder_variable}))
+
+    // ${make_function} makes an empty ${latest_map_type}.
+    func ${make_function}() ${latest_map_type} {
+        return ${empty_latest_map_variable}
+    }
+
+    // Copy is a noop, as ${latest_map_type}s are immutable.
+    func (m ${latest_map_type}) Copy() ${latest_map_type} {
+        return (${latest_map_type})((LatestMap)(m).Copy())
+    }
+
+    // Size returns the number of elements.
+    func (m ${latest_map_type}) Size() int {
+        return (LatestMap)(m).Size()
+    }
+
+    // Merge produces a fresh ${latest_map_type} containing the keys from both inputs.
+    // When both inputs contain the same key, the newer value is used.
+    func (m ${latest_map_type}) Merge(other ${latest_map_type}) ${latest_map_type} {
+        return (${latest_map_type})((LatestMap)(m).Merge((LatestMap)(other)))
+    }
+
+    // Lookup the value for the given key.
+    func (m ${latest_map_type}) Lookup(key string) (${data_type}, bool) {
+        v, ok := (LatestMap)(m).Lookup(key)
+        if !ok {
+            var zero ${data_type}
+            return zero, false
+        }
+        return v.(${data_type}), true
+    }
+
+    // LookupEntry returns the raw entry for the given key.
+    func (m ${latest_map_type}) LookupEntry(key string) (${data_type}, time.Time, bool) {
+        v, timestamp, ok := (LatestMap)(m).LookupEntry(key)
+        if !ok {
+            var zero ${data_type}
+            return zero, timestamp, false
+        }
+        return v.(${data_type}), timestamp, true
+    }
+
+    // Set the value for the given key.
+    func (m ${latest_map_type}) Set(key string, timestamp time.Time, value ${data_type}) ${latest_map_type} {
+        return (${latest_map_type})((LatestMap)(m).Set(key, timestamp, value))
+    }
+
+    // Delete the value for the given key.
+    func (m ${latest_map_type}) Delete(key string) ${latest_map_type} {
+        return (${latest_map_type})((LatestMap)(m).Delete(key))
+    }
+
+    // ForEach executes fn on each key value pair in the map.
+    func (m ${latest_map_type}) ForEach(fn func(k string, timestamp time.Time, v ${data_type})) {
+        (LatestMap)(m).ForEach(func(key string, ts time.Time, value interface{}) {
+                fn(key, ts, value.(${data_type}))
+        })
+    }
+
+    // String returns the ${latest_map_type}'s string representation.
+    func (m ${latest_map_type}) String() string {
+        return (LatestMap)(m).String()
+    }
+
+    // DeepEqual tests equality with other ${latest_map_type}.
+    func (m ${latest_map_type}) DeepEqual(n ${latest_map_type}) bool {
+        return (LatestMap)(m).DeepEqual((LatestMap)(n))
+    }
+
+    // CodecEncodeSelf implements codec.Selfer.
+    func (m *${latest_map_type}) CodecEncodeSelf(encoder *codec.Encoder) {
+        (*LatestMap)(m).CodecEncodeSelf(encoder)
+    }
+
+    // CodecDecodeSelf implements codec.Selfer.
+    func (m *${latest_map_type}) CodecDecodeSelf(decoder *codec.Decoder) {
+        bm := (*LatestMap)(m)
+        bm.decoder = ${iface_decoder_variable}
+        bm.CodecDecodeSelf(decoder)
+    }
+
+    // MarshalJSON shouldn't be used, use CodecEncodeSelf instead.
+    func (${latest_map_type}) MarshalJSON() ([]byte, error) {
+        panic("MarshalJSON shouldn't be used, use CodecEncodeSelf instead")
+    }
+
+    // UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead.
+    func (*${latest_map_type}) UnmarshalJSON(b []byte) error {
+        panic("UnmarshalJSON shouldn't be used, use CodecDecodeSelf instead")
+    }
+EOF
+}
+
+if [ -z "${1}" ]; then
+    echo "No output file given"
+    exit 1
+fi
+
+out="${1}"
+outtmp="${out}.tmp"
+
+generate_header "${outtmp}" "${0} ${*}"
+shift
+for t in ${*}; do
+    generate_latest_map "${outtmp}" "${t}"
+done
+
+gofmt -s -w "${outtmp}"
+mv "${outtmp}" "${out}"

tools/test

@@ -47,8 +47,15 @@ fi
 
 fail=0
 
-# NB: Relies on paths being prefixed with './'.
-TESTDIRS=( $(git ls-files -- '*_test.go' | grep -vE '^(vendor|prog|experimental)/' | xargs -n1 dirname  | sort -u | sed -e 's|^|./|') )
+if [ -z "$TESTDIRS" ]; then
+    # NB: Relies on paths being prefixed with './'.
+    TESTDIRS=( $(git ls-files -- '*_test.go' | grep -vE '^(vendor|prog|experimental)/' | xargs -n1 dirname  | sort -u | sed -e 's|^|./|') )
+else
+    # TESTDIRS on the right side is not really an array variable, it
+    # is just a string with spaces, but it is written like that to
+    # shut up the shellcheck tool.
+    TESTDIRS=( $(for d in ${TESTDIRS[*]}; do echo "$d"; done) )
+fi
 
 # If running on circle, use the scheduler to work out what tests to run on what shard
 if [ -n "$CIRCLECI" ] && [ -z "$NO_SCHEDULER" ] && [ -x "$DIR/sched" ]; then