diff --git a/Makefile b/Makefile index 170d64a2f..9b8fd2cc9 100644 --- a/Makefile +++ b/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 diff --git a/app/collector.go b/app/collector.go index deadce4a8..18b67bcc8 100644 --- a/app/collector.go +++ b/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() { diff --git a/examples/plugins/README.md b/examples/plugins/README.md index 4b425c5be..0dc0e2467 100644 --- a/examples/plugins/README.md +++ b/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. Scope Probe plugin screenshot -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 -## 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. -### 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. -### 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;` +- The Docker image with name `docker.io/alpine`: + `docker.io/alpine;` +- The host with name `example.com`: `example.com:` + +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;","label":"example.com","topologyId":"hosts"}],…` + URL, you can find the `example.com;` which is an ID of the node + representing the host. + - Mentally substitute the `` with `/`. This can appear in + Docker image names, so `docker.io/alpine` in the address bar will + be `docker.ioalpine`. diff --git a/examples/plugins/http-requests/http-requests.py b/examples/plugins/http-requests/http-requests.py index 866cb1fa9..348e6588a 100755 --- a/examples/plugins/http-requests/http-requests.py +++ b/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'], diff --git a/examples/plugins/iowait/Makefile b/examples/plugins/iowait/Makefile index 63040e03c..74f2fce50 100644 --- a/examples/plugins/iowait/Makefile +++ b/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) diff --git a/examples/plugins/iowait/main.go b/examples/plugins/iowait/main.go index a77b899fb..6f1230605 100644 --- a/examples/plugins/iowait/main.go +++ b/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+";", 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;", 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 } diff --git a/probe/controls/controls.go b/probe/controls/controls.go index 2d12daa69..446668b48 100644 --- a/probe/controls/controls.go +++ b/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) } diff --git a/probe/controls/controls_test.go b/probe/controls/controls_test.go index b7c153a6c..7bdf22690 100644 --- a/probe/controls/controls_test.go +++ b/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) { diff --git a/probe/docker/container.go b/probe/docker/container.go index c88288d7d..33f862e84 100644 --- a/probe/docker/container.go +++ b/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 } diff --git a/probe/docker/container_test.go b/probe/docker/container_test.go index 3d200a337..28c5c54fd 100644 --- a/probe/docker/container_test.go +++ b/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;", 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) } diff --git a/probe/docker/controls.go b/probe/docker/controls.go index 7d79c7156..5fc85a417 100644 --- a/probe/docker/controls.go +++ b/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) } diff --git a/probe/docker/controls_test.go b/probe/docker/controls_test.go index 97e3832b3..61b7baf43 100644 --- a/probe/docker/controls_test.go +++ b/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"), }) diff --git a/probe/docker/registry.go b/probe/docker/registry.go index a3afc9500..b751257c3 100644 --- a/probe/docker/registry.go +++ b/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() diff --git a/probe/docker/registry_test.go b/probe/docker/registry_test.go index cec900644..afeaf29b9 100644 --- a/probe/docker/registry_test.go +++ b/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() diff --git a/probe/host/controls.go b/probe/host/controls.go index 4e67e7a55..602678f16 100644 --- a/probe/host/controls.go +++ b/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 { diff --git a/probe/host/reporter.go b/probe/host/reporter.go index ce68a387d..7d0a2a5aa 100644 --- a/probe/host/reporter.go +++ b/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{ diff --git a/probe/host/reporter_test.go b/probe/host/reporter_test.go index bd1d2d823..3e2805775 100644 --- a/probe/host/reporter_test.go +++ b/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) } diff --git a/probe/kubernetes/controls.go b/probe/kubernetes/controls.go index 9eab252c9..b4ac690d7 100644 --- a/probe/kubernetes/controls.go +++ b/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) } diff --git a/probe/kubernetes/deployment.go b/probe/kubernetes/deployment.go index 965b812f3..77c785022 100644 --- a/probe/kubernetes/deployment.go +++ b/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) } diff --git a/probe/kubernetes/pod.go b/probe/kubernetes/pod.go index 52f53d761..c190d9684 100644 --- a/probe/kubernetes/pod.go +++ b/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) } diff --git a/probe/kubernetes/replica_set.go b/probe/kubernetes/replica_set.go index eafbebe69..745c5e992 100644 --- a/probe/kubernetes/replica_set.go +++ b/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) } diff --git a/probe/kubernetes/replication_controller.go b/probe/kubernetes/replication_controller.go index 14350f5a6..016bd4dde 100644 --- a/probe/kubernetes/replication_controller.go +++ b/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) } diff --git a/probe/kubernetes/reporter.go b/probe/kubernetes/reporter.go index 58d36b2f3..8dc4f55ad 100644 --- a/probe/kubernetes/reporter.go +++ b/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) diff --git a/probe/kubernetes/reporter_test.go b/probe/kubernetes/reporter_test.go index f9de82dc0..ad4d5c8b0 100644 --- a/probe/kubernetes/reporter_test.go +++ b/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 { diff --git a/probe/plugins/registry.go b/probe/plugins/registry.go index 80612a08d..c9915127f 100644 --- a/probe/plugins/registry.go +++ b/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 } diff --git a/probe/plugins/registry_internal_test.go b/probe/plugins/registry_internal_test.go index 6f60cc82b..a9f1a2262 100644 --- a/probe/plugins/registry_internal_test.go +++ b/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) + } +} diff --git a/probe/probe.go b/probe/probe.go index a9139fc4d..80d1bf458 100644 --- a/probe/probe.go +++ b/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) } } diff --git a/prog/probe.go b/prog/probe.go index 01c030136..e08741af6 100644 --- a/prog/probe.go +++ b/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) diff --git a/render/detailed/node.go b/render/detailed/node.go index 3824581da..6c821487c 100644 --- a/render/detailed/node.go +++ b/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 } diff --git a/report/controls.go b/report/controls.go index 9b0a89487..80dd2b062 100644 --- a/report/controls.go +++ b/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"` +} diff --git a/report/latest_map.go b/report/latest_map.go index d45e119c7..da767c1c7 100644 --- a/report/latest_map.go +++ b/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") } diff --git a/report/latest_map_generated.go b/report/latest_map_generated.go new file mode 100644 index 000000000..cd062bcb4 --- /dev/null +++ b/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") +} diff --git a/report/latest_map_internal_test.go b/report/latest_map_internal_test.go index b5072f8b2..286b265b0 100644 --- a/report/latest_map_internal_test.go +++ b/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) +} diff --git a/report/node.go b/report/node.go index 4ff2547f0..858eeaabd 100644 --- a/report/node.go +++ b/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), } } diff --git a/report/report.go b/report/report.go index a92dc4914..75c8e591b 100644 --- a/report/report.go +++ b/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 diff --git a/report/report_test.go b/report/report_test.go index 6db2c3184..95c10b685 100644 --- a/report/report_test.go +++ b/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)) + } +} diff --git a/report/table.go b/report/table.go index 10d4aa40c..feaec789d 100644 --- a/report/table.go +++ b/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 diff --git a/test/diff.go b/test/diff.go index d7542bb69..477ae98f9 100644 --- a/test/diff.go +++ b/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, diff --git a/tools/generate_latest_map b/tools/generate_latest_map new file mode 100755 index 000000000..366a718d4 --- /dev/null +++ b/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}" diff --git a/tools/test b/tools/test index f8d76c56b..4ee4ca1a4 100755 --- a/tools/test +++ b/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