diff --git a/common/backoff/backoff.go b/common/backoff/backoff.go index 4727a60cc..5b7b5005a 100644 --- a/common/backoff/backoff.go +++ b/common/backoff/backoff.go @@ -22,6 +22,7 @@ type Interface interface { Start() Stop() SetInitialBackoff(time.Duration) + SetMaxBackoff(time.Duration) } // New makes a new Interface @@ -40,6 +41,10 @@ func (b *backoff) SetInitialBackoff(d time.Duration) { b.initialBackoff = d } +func (b *backoff) SetMaxBackoff(d time.Duration) { + b.maxBackoff = d +} + // Stop the backoff, and waits for it to stop. func (b *backoff) Stop() { close(b.quit) diff --git a/probe/plugins/registry.go b/probe/plugins/registry.go index 7dc7b4d22..64967024c 100644 --- a/probe/plugins/registry.go +++ b/probe/plugins/registry.go @@ -2,6 +2,7 @@ package plugins import ( "fmt" + "io" "net/http" "net/url" "path/filepath" @@ -31,6 +32,9 @@ const ( pluginTimeout = 500 * time.Millisecond pluginRetry = 5 * time.Second pollingInterval = 5 * time.Second + + initialBackoff = 1 * time.Second + maxBackoff = 15 * time.Second ) // Registry maintains a list of available plugins by name. @@ -222,9 +226,17 @@ func NewPlugin(ctx context.Context, socket string, client *http.Client, expected ctx, cancel := context.WithCancel(ctx) p := &Plugin{context: ctx, socket: socket, client: client, cancel: cancel} f := p.handshake(ctx, expectedAPIVersion, params) - f() // try the first time synchronously - p.backoff = backoff.New(f, "plugin handshake") - go p.backoff.Start() + // try the first time synchronously (makes testing easier) + if ok, err := f(); !ok { + log.Errorf("Error plugin handshake, backing off 10s: %v", err) + p.backoff = backoff.New(f, "plugin handshake") + p.backoff.SetInitialBackoff(initialBackoff) + p.backoff.SetMaxBackoff(maxBackoff) + go func() { + time.Sleep(initialBackoff) + p.backoff.Start() + }() + } return p } @@ -274,20 +286,28 @@ func (p *Plugin) setStatus(err error) { } } -// TODO(paulbellamy): better error handling on wrong status codes func (p *Plugin) get(path string, params url.Values, 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() resp, err := ctxhttp.Get(ctx, p.client, fmt.Sprintf("unix://%s?%s", path, params.Encode())) if err != nil { return err } + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("plugin returned non-200 status code: %s", resp.Status) + } defer resp.Body.Close() - return codec.NewDecoder(resp.Body, &codec.JsonHandle{}).Decode(&result) + if err := codec.NewDecoder(io.LimitReader(resp.Body, 50*1024*1024), &codec.JsonHandle{}).Decode(&result); err != nil { + return fmt.Errorf("decoding error: %s", err) + } + return nil } // Close closes the client func (p *Plugin) Close() { - p.backoff.Stop() + if p.backoff != nil { + p.backoff.Stop() + } p.cancel() } diff --git a/probe/plugins/registry_internal_test.go b/probe/plugins/registry_internal_test.go index 590981dd0..72be8878b 100644 --- a/probe/plugins/registry_internal_test.go +++ b/probe/plugins/registry_internal_test.go @@ -78,7 +78,7 @@ func (p mockPlugin) file() fs.File { } resp := httptest.NewRecorder() p.Handler.ServeHTTP(resp, req) - fmt.Fprintf(outgoingW, "HTTP/1.1 200 OK\nContent-Length: %d\n\n%s", resp.Body.Len(), resp.Body.String()) + fmt.Fprintf(outgoingW, "HTTP/1.1 %d %s\nContent-Length: %d\n\n%s", resp.Code, http.StatusText(resp.Code), resp.Body.Len(), resp.Body.String()) if p.Requests != nil { p.Requests <- req } @@ -141,14 +141,15 @@ func checkLoadedPlugins(t *testing.T, forEach iterator, expectedIDs []string) { } // stringHandler returns an http.Handler which just prints the given string -func stringHandler(j string) http.Handler { +func stringHandler(status int, j string) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(status) fmt.Fprint(w, j) }) } func TestRegistryLoadsExistingPlugins(t *testing.T) { - setup(t, mockPlugin{t: t, Name: "testPlugin", Handler: stringHandler(`{"name":"testPlugin","interfaces":["reporter"],"api_version":"1"}`)}.file()) + setup(t, mockPlugin{t: t, Name: "testPlugin", Handler: stringHandler(http.StatusOK, `{"name":"testPlugin","interfaces":["reporter"],"api_version":"1"}`)}.file()) defer restore(t) root := "/plugins" @@ -166,9 +167,9 @@ func TestRegistryLoadsExistingPluginsEvenWhenOneFails(t *testing.T) { t, // TODO: This first one needs to fail fs.Dir("fail", - mockPlugin{t: t, Name: "aFailure", Handler: stringHandler(`{"name":"aFailure","interfaces":["reporter"],"api_version":"2"}`)}.file(), + mockPlugin{t: t, Name: "aFailure", Handler: stringHandler(http.StatusOK, `{"name":"aFailure","interfaces":["reporter"],"api_version":"2"}`)}.file(), ), - mockPlugin{t: t, Name: "testPlugin", Handler: stringHandler(`{"name":"testPlugin","interfaces":["reporter"],"api_version":"1"}`)}.file(), + mockPlugin{t: t, Name: "testPlugin", Handler: stringHandler(http.StatusOK, `{"name":"testPlugin","interfaces":["reporter"],"api_version":"1"}`)}.file(), ) defer restore(t) @@ -196,7 +197,12 @@ func TestRegistryDiscoversNewPlugins(t *testing.T) { checkLoadedPlugins(t, r.ForEach, []string{}) // Add the new plugin - plugin := mockPlugin{t: t, Name: "testPlugin", Requests: make(chan *http.Request), Handler: stringHandler(`{"name":"testPlugin","interfaces":["reporter"]}`)} + plugin := mockPlugin{ + t: t, + Name: "testPlugin", + Requests: make(chan *http.Request), + Handler: stringHandler(http.StatusOK, `{"name":"testPlugin","interfaces":["reporter"]}`), + } mockFS.Add(plugin.dir(), plugin.file()) if err := r.scan(); err != nil { t.Fatal(err) @@ -206,7 +212,12 @@ func TestRegistryDiscoversNewPlugins(t *testing.T) { } func TestRegistryRemovesPlugins(t *testing.T) { - plugin := mockPlugin{t: t, Name: "testPlugin", Requests: make(chan *http.Request), Handler: stringHandler(`{"name":"testPlugin","interfaces":["reporter"]}`)} + plugin := mockPlugin{ + t: t, + Name: "testPlugin", + Requests: make(chan *http.Request), + Handler: stringHandler(http.StatusOK, `{"name":"testPlugin","interfaces":["reporter"]}`), + } mockFS := setup(t, plugin.file()) defer restore(t) @@ -234,12 +245,12 @@ func TestRegistryReturnsPluginsByInterface(t *testing.T) { mockPlugin{ t: t, Name: "plugin1", - Handler: stringHandler(`{"name":"plugin1","interfaces":["reporter"]}`), + Handler: stringHandler(http.StatusOK, `{"name":"plugin1","interfaces":["reporter"]}`), }.file(), mockPlugin{ t: t, Name: "plugin2", - Handler: stringHandler(`{"name":"plugin2","interfaces":["other"]}`), + Handler: stringHandler(http.StatusOK, `{"name":"plugin2","interfaces":["other"]}`), }.file(), ) defer restore(t) @@ -262,12 +273,12 @@ func TestRegistryHandlesConflictingPlugins(t *testing.T) { mockPlugin{ t: t, Name: "plugin1", - Handler: stringHandler(`{"name":"plugin1","interfaces":["reporter"]}`), + Handler: stringHandler(http.StatusOK, `{"name":"plugin1","interfaces":["reporter"]}`), }.file(), mockPlugin{ t: t, Name: "plugin1", - Handler: stringHandler(`{"name":"plugin2","interfaces":["other"]}`), + Handler: stringHandler(http.StatusOK, `{"name":"plugin2","interfaces":["other"]}`), }.file(), ) defer restore(t) @@ -283,3 +294,35 @@ func TestRegistryHandlesConflictingPlugins(t *testing.T) { checkLoadedPlugins(t, r.ForEach, []string{"plugin2"}) checkLoadedPlugins(t, func(fn func(*Plugin)) { r.Implementers("other", fn) }, []string{"plugin2"}) } + +func TestRegistryRejectsErroneousPluginResponses(t *testing.T) { + setup( + t, + mockPlugin{ + t: t, + Name: "okPlugin", + Handler: stringHandler(http.StatusOK, `{"name":"okPlugin","interfaces":["reporter"]}`), + }.file(), + mockPlugin{ + t: t, + Name: "non200ResponseCode", + Handler: stringHandler(http.StatusInternalServerError, `{"name":"non200ResponseCode","interfaces":["reporter"]}`), + }.file(), + mockPlugin{ + t: t, + Name: "nonJSONResponseBody", + Handler: stringHandler(http.StatusOK, `notJSON`), + }.file(), + ) + defer restore(t) + + root := "/plugins" + r, err := NewRegistry(root, "", nil) + if err != nil { + t.Fatal(err) + } + defer r.Close() + + // Should have only okPlugin + checkLoadedPlugins(t, r.ForEach, []string{"", "", "okPlugin"}) +} diff --git a/render/detailed/connections.go b/render/detailed/connections.go index 8100a4a70..09203d83a 100644 --- a/render/detailed/connections.go +++ b/render/detailed/connections.go @@ -41,11 +41,11 @@ type ConnectionsSummary struct { // Connection is a row in the connections table. type Connection struct { - ID string `json:"id"` // ID of this element in the UI. Must be unique for a given ConnectionsSummary. - NodeID string `json:"nodeId"` // ID of a node in the topology. Optional, must be set if linkable is true. - Label string `json:"label"` - Linkable bool `json:"linkable"` - Metadata []MetadataRow `json:"metadata,omitempty"` + ID string `json:"id"` // ID of this element in the UI. Must be unique for a given ConnectionsSummary. + NodeID string `json:"nodeId"` // ID of a node in the topology. Optional, must be set if linkable is true. + Label string `json:"label"` + Linkable bool `json:"linkable"` + Metadata []report.MetadataRow `json:"metadata,omitempty"` } type connectionsByID []Connection