From 766fd8f4f148c7a735ca68655b4a0e780fa5d4a6 Mon Sep 17 00:00:00 2001 From: Patrick East Date: Tue, 28 Jan 2020 14:58:26 -0800 Subject: [PATCH] plugins/status: Update to report plugin statuses This adds in the status updates for the status plugin (reported to the plugin manager) as well as hooking the status plugin up as a listener for *all* plugin status updates to include in the status API events. Signed-off-by: Patrick East --- docs/content/management.md | 52 +++++++++++++++++++----------- plugins/status/plugin.go | 60 +++++++++++++++++++++++++++-------- plugins/status/plugin_test.go | 48 +++++++++++++++++++++++----- 3 files changed, 121 insertions(+), 39 deletions(-) diff --git a/docs/content/management.md b/docs/content/management.md index 666eb674b3..d5584de76a 100644 --- a/docs/content/management.md +++ b/docs/content/management.md @@ -430,9 +430,12 @@ OPA can periodically report status updates to remote HTTP servers. The updates contain status information for OPA itself as well as the [Bundles](#bundles) that have been downloaded and activated. -OPA sends status reports whenever bundles are downloaded and activated. If -the bundle download or activation fails for any reason, the status update -will include error information describing the failure. +OPA sends status reports whenever one of the following happens: + +* Bundles are downloaded and activated -- If the bundle download or activation fails for any reason, the status update + will include error information describing the failure. This includes Discovery bundles. +* A plugin state has changed -- All plugin status is reported, and an update to any plugin will + trigger a Status API report which contains the latest state. The status updates will include a set of labels that uniquely identify the OPA instance. OPA automatically includes an `id` value in the label set that @@ -457,23 +460,34 @@ on the agent, updates will be sent to `/status`. ```json { - "labels": { - "app": "my-example-app", - "id": "1780d507-aea2-45cc-ae50-fa153c8e4a5a", - "version": "{{< current_version >}}" + "labels": { + "app": "my-example-app", + "id": "1780d507-aea2-45cc-ae50-fa153c8e4a5a", + "version": "{{< current_version >}}" + }, + "bundles": { + "http/example/authz": { + "active_revision": "ABC", + "last_successful_download": "2018-01-01T00:00:00.000Z", + "last_successful_activation": "2018-01-01T00:00:00.000Z", + "metrics": { + "timer_rego_data_parse_ns": 12345, + "timer_rego_module_compile_ns": 12345, + "timer_rego_module_parse_ns": 12345 + } + } + }, + "plugins": { + "bundle": { + "state": "OK" }, - "bundles": { - "http/example/authz": { - "active_revision": "TODO", - "last_successful_download": "2018-01-01T00:00:00.000Z", - "last_successful_activation": "2018-01-01T00:00:00.000Z", - "metrics": { - "timer_rego_data_parse_ns": 12345, - "timer_rego_module_compile_ns": 12345, - "timer_rego_module_parse_ns": 12345 - } - } + "discovery": { + "state": "OK" }, + "status": { + "state": "OK" + } + }, "metrics": { "prometheus": { "go_gc_duration_seconds": { @@ -609,6 +623,8 @@ Status updates contain the following fields: | `discovery.active_revision` | `string` | Opaque revision identifier of the last successful discovery activation. | | `discovery.last_successful_download` | `string` | RFC3339 timestamp of last successful discovery bundle download. | | `discovery.last_successful_activation` | `string` | RFC3339 timestamp of last successful discovery bundle activation. | +| `plugins` | `object` | A set of objects describing the state of configured plugins in OPA's runtime. | +| `plugins[_].state | `string` | The state of each plugin. | | `metrics.prometheus` | `object` | Global performance metrics for the OPA instance. | If the bundle download or activation failed, the status update will contain diff --git a/plugins/status/plugin.go b/plugins/status/plugin.go index aef999dc79..29467c7e42 100644 --- a/plugins/status/plugin.go +++ b/plugins/status/plugin.go @@ -24,11 +24,12 @@ import ( // UpdateRequestV1 represents the status update message that OPA sends to // remote HTTP endpoints. type UpdateRequestV1 struct { - Labels map[string]string `json:"labels"` - Bundle *bundle.Status `json:"bundle,omitempty"` // Deprecated: Use bulk `bundles` status updates instead - Bundles map[string]*bundle.Status `json:"bundles,omitempty"` - Discovery *bundle.Status `json:"discovery,omitempty"` - Metrics map[string]interface{} `json:"metrics,omitempty"` + Labels map[string]string `json:"labels"` + Bundle *bundle.Status `json:"bundle,omitempty"` // Deprecated: Use bulk `bundles` status updates instead + Bundles map[string]*bundle.Status `json:"bundles,omitempty"` + Discovery *bundle.Status `json:"discovery,omitempty"` + Metrics map[string]interface{} `json:"metrics,omitempty"` + Plugins map[string]*plugins.Status `json:"plugins,omitempty"` } // Plugin implements status reporting. Updates can be triggered by the caller. @@ -44,6 +45,8 @@ type Plugin struct { stop chan chan struct{} reconfig chan interface{} metrics metrics.Metrics + lastPluginStatuses map[string]*plugins.Status + pluginStatusCh chan map[string]*plugins.Status } // Config contains configuration for the plugin. @@ -106,15 +109,20 @@ func ParseConfig(config []byte, services []string) (*Config, error) { // New returns a new Plugin with the given config. func New(parsedConfig *Config, manager *plugins.Manager) *Plugin { - return &Plugin{ - manager: manager, - config: *parsedConfig, - bundleCh: make(chan bundle.Status), - bulkBundleCh: make(chan map[string]*bundle.Status), - discoCh: make(chan bundle.Status), - stop: make(chan chan struct{}), - reconfig: make(chan interface{}), + p := &Plugin{ + manager: manager, + config: *parsedConfig, + bundleCh: make(chan bundle.Status), + bulkBundleCh: make(chan map[string]*bundle.Status), + discoCh: make(chan bundle.Status), + stop: make(chan chan struct{}), + reconfig: make(chan interface{}), + pluginStatusCh: make(chan map[string]*plugins.Status), } + + p.manager.UpdatePluginStatus(Name, &plugins.Status{State: plugins.StateNotReady}) + + return p } // WithMetrics sets the global metrics provider to be used by the plugin. @@ -137,16 +145,28 @@ func Lookup(manager *plugins.Manager) *Plugin { // Start starts the plugin. func (p *Plugin) Start(ctx context.Context) error { p.logInfo("Starting status reporter.") + go p.loop() + + // Setup a listener for plugin statuses, but only after starting the loop + // to prevent blocking threads pushing the plugin updates. + p.manager.RegisterPluginStatusListener(Name, p.UpdatePluginStatus) + + // Set the status plugin's status to OK now that everything is registered and + // the loop is running. This will trigger an update on the listener with the + // current status of all the other plugins too. + p.manager.UpdatePluginStatus(Name, &plugins.Status{State: plugins.StateOK}) return nil } // Stop stops the plugin. func (p *Plugin) Stop(ctx context.Context) { p.logInfo("Stopping status reporter.") + p.manager.UnregisterPluginStatusListener(Name) done := make(chan struct{}) p.stop <- done _ = <-done + p.manager.UpdatePluginStatus(Name, &plugins.Status{State: plugins.StateNotReady}) } // UpdateBundleStatus notifies the plugin that the policy bundle was updated. @@ -165,6 +185,11 @@ func (p *Plugin) UpdateDiscoveryStatus(status bundle.Status) { p.discoCh <- status } +// UpdatePluginStatus notifies the plugin that a plugin status was updated. +func (p *Plugin) UpdatePluginStatus(status map[string]*plugins.Status) { + p.pluginStatusCh <- status +} + // Reconfigure notifies the plugin with a new configuration. func (p *Plugin) Reconfigure(_ context.Context, config interface{}) { p.reconfig <- config @@ -176,6 +201,14 @@ func (p *Plugin) loop() { for { select { + case statuses := <-p.pluginStatusCh: + p.lastPluginStatuses = statuses + err := p.oneShot(ctx) + if err != nil { + p.logError("%v.", err) + } else { + p.logInfo("Status update sent successfully in response to plugin update.") + } case statuses := <-p.bulkBundleCh: p.lastBundleStatuses = statuses err := p.oneShot(ctx) @@ -219,6 +252,7 @@ func (p *Plugin) oneShot(ctx context.Context) error { Discovery: p.lastDiscoStatus, Bundle: p.lastBundleStatus, Bundles: p.lastBundleStatuses, + Plugins: p.lastPluginStatuses, } if p.metrics != nil { diff --git a/plugins/status/plugin_test.go b/plugins/status/plugin_test.go index bddf4a0805..fac28e5e95 100644 --- a/plugins/status/plugin_test.go +++ b/plugins/status/plugin_test.go @@ -40,9 +40,8 @@ func TestPluginStart(t *testing.T) { fixture.plugin.Start(ctx) defer fixture.plugin.Stop(ctx) - status := testStatus() - - fixture.plugin.UpdateBundleStatus(*status) + // Start will trigger a status update when the plugin state switches + // from "not ready" to "ok". result := <-fixture.server.ch exp := UpdateRequestV1{ @@ -51,12 +50,25 @@ func TestPluginStart(t *testing.T) { "app": "example-app", "version": version.Version, }, - Bundle: status, + Plugins: map[string]*plugins.Status{ + "status": {State: plugins.StateOK}, + }, } if !reflect.DeepEqual(result, exp) { t.Fatalf("Expected: %v but got: %v", exp, result) } + + status := testStatus() + + fixture.plugin.UpdateBundleStatus(*status) + result = <-fixture.server.ch + + exp.Bundle = status + + if !reflect.DeepEqual(result, exp) { + t.Fatalf("Expected: %v but got: %v", exp, result) + } } func TestPluginStartBulkUpdate(t *testing.T) { @@ -70,9 +82,8 @@ func TestPluginStartBulkUpdate(t *testing.T) { fixture.plugin.Start(ctx) defer fixture.plugin.Stop(ctx) - status := testStatus() - - fixture.plugin.BulkUpdateBundleStatus(map[string]*bundle.Status{status.Name: status}) + // Start will trigger a status update when the plugin state switches + // from "not ready" to "ok". result := <-fixture.server.ch exp := UpdateRequestV1{ @@ -81,9 +92,18 @@ func TestPluginStartBulkUpdate(t *testing.T) { "app": "example-app", "version": version.Version, }, - Bundles: map[string]*bundle.Status{status.Name: status}, + Plugins: map[string]*plugins.Status{ + "status": {State: plugins.StateOK}, + }, } + status := testStatus() + + fixture.plugin.BulkUpdateBundleStatus(map[string]*bundle.Status{status.Name: status}) + result = <-fixture.server.ch + + exp.Bundles = map[string]*bundle.Status{status.Name: status} + if !reflect.DeepEqual(result, exp) { t.Fatalf("Expected: %v but got: %v", exp, result) } @@ -100,6 +120,9 @@ func TestPluginStartBulkUpdateMultiple(t *testing.T) { fixture.plugin.Start(ctx) defer fixture.plugin.Stop(ctx) + // Ignore the plugin updating its status (tested elsewhere) + <-fixture.server.ch + statuses := map[string]*bundle.Status{} tDownload, _ := time.Parse("2018-01-01T00:00:00.0000000Z", time.RFC3339Nano) tActivate, _ := time.Parse("2018-01-01T00:00:01.0000000Z", time.RFC3339Nano) @@ -152,6 +175,9 @@ func TestPluginStartDiscovery(t *testing.T) { fixture.plugin.Start(ctx) defer fixture.plugin.Stop(ctx) + // Ignore the plugin updating its status (tested elsewhere) + <-fixture.server.ch + status := testStatus() fixture.plugin.UpdateDiscoveryStatus(*status) @@ -164,6 +190,9 @@ func TestPluginStartDiscovery(t *testing.T) { "version": version.Version, }, Discovery: status, + Plugins: map[string]*plugins.Status{ + "status": {State: plugins.StateOK}, + }, } if !reflect.DeepEqual(result, exp) { @@ -241,6 +270,9 @@ func TestMetrics(t *testing.T) { fixture.plugin.Start(ctx) defer fixture.plugin.Stop(ctx) + // Ignore the plugin updating its status (tested elsewhere) + <-fixture.server.ch + status := testStatus() fixture.plugin.BulkUpdateBundleStatus(map[string]*bundle.Status{"bundle": status})