From c4efb263ea8b1b87fb9044af4f5ebc1ba99ff0e0 Mon Sep 17 00:00:00 2001 From: Ashutosh Narkar Date: Mon, 6 Mar 2023 22:24:44 -0800 Subject: [PATCH] plugins: Surface decision log errors via status API Currently errors encountered during decision log uploads are logged. This change now includes these errors as part of status updates which can be useful for the control plane to determine if OPA is having any issues while processing, uploading logs etc. Fixes: #5637 Signed-off-by: Ashutosh Narkar --- docs/content/management-status.md | 54 ++++++----- plugins/logs/plugin.go | 20 +++- plugins/logs/plugin_test.go | 153 ++++++++++++++++++++++++++++++ plugins/logs/status/status.go | 58 +++++++++++ plugins/status/plugin.go | 75 +++++++++------ plugins/status/plugin_test.go | 45 +++++++++ 6 files changed, 351 insertions(+), 54 deletions(-) create mode 100644 plugins/logs/status/status.go diff --git a/docs/content/management-status.md b/docs/content/management-status.md index 2950af0e49..bfe2111d67 100644 --- a/docs/content/management-status.md +++ b/docs/content/management-status.md @@ -221,32 +221,36 @@ on the agent, updates will be sent to `/status`. Status updates contain the following fields: -| Field | Type | Description | -| --- | --- | --- | -| `labels` | `object` | Set of key-value pairs that uniquely identify the OPA instance. | -| `bundles` | `object` | Set of objects describing the status for each bundle configured with OPA. | -| `bundles[_].name` | `string` | Name of bundle that the OPA instance is configured to download. | -| `bundles[_].active_revision` | `string` | Opaque revision identifier of the last successful activation. | -| `bundles[_].last_request` | `string` | RFC3339 timestamp of last bundle request. This timestamp should be >= to the successful request timestamp in normal operation. | -| `bundles[_].last_successful_request` | `string` | RFC3339 timestamp of last successful bundle request. This timestamp should be >= to the successful download timestamp in normal operation. | -| `bundles[_].last_successful_download` | `string` | RFC3339 timestamp of last successful bundle download. | -| `bundles[_].last_successful_activation` | `string` | RFC3339 timestamp of last successful bundle activation. | -| `bundles[_].metrics` | `object` | Metrics from the last update of the bundle. | -| `bundles[_].code` | `string` | If present, indicates error(s) occurred activating this bundle. | -| `bundles[_].message` | `string` | Human readable messages describing the error(s). | -| `bundles[_].http_code` | `number` | If present, indicates an erroneous HTTP status code that OPA received downloading this bundle. | -| `bundles[_].errors` | `array` | Collection of detailed parse or compile errors that occurred during activation of this bundle. | -| `bundles[_].size` | `number` | Bundle size, in bytes | -| `bundles[_].type` | `string` | Bundle type, either `snapshot` or `delta` | -| `discovery.name` | `string` | Name of discovery bundle that the OPA instance is configured to download. | -| `discovery.active_revision` | `string` | Opaque revision identifier of the last successful discovery activation. | -| `discovery.last_request` | `string` | RFC3339 timestamp of last discovery bundle request. This timestamp should be >= to the successful request timestamp in normal operation. | +| Field | Type | Description | +|---------------------| --- |------------------------------------------------------------------------------------------------------------------------------------------------------| +| `labels` | `object` | Set of key-value pairs that uniquely identify the OPA instance. | +| `bundles` | `object` | Set of objects describing the status for each bundle configured with OPA. | +| `bundles[_].name` | `string` | Name of bundle that the OPA instance is configured to download. | +| `bundles[_].active_revision` | `string` | Opaque revision identifier of the last successful activation. | +| `bundles[_].last_request` | `string` | RFC3339 timestamp of last bundle request. This timestamp should be >= to the successful request timestamp in normal operation. | +| `bundles[_].last_successful_request` | `string` | RFC3339 timestamp of last successful bundle request. This timestamp should be >= to the successful download timestamp in normal operation. | +| `bundles[_].last_successful_download` | `string` | RFC3339 timestamp of last successful bundle download. | +| `bundles[_].last_successful_activation` | `string` | RFC3339 timestamp of last successful bundle activation. | +| `bundles[_].metrics` | `object` | Metrics from the last update of the bundle. | +| `bundles[_].code` | `string` | If present, indicates error(s) occurred activating this bundle. | +| `bundles[_].message` | `string` | Human readable messages describing the error(s). | +| `bundles[_].http_code` | `number` | If present, indicates an erroneous HTTP status code that OPA received downloading this bundle. | +| `bundles[_].errors` | `array` | Collection of detailed parse or compile errors that occurred during activation of this bundle. | +| `bundles[_].size` | `number` | Bundle size, in bytes | +| `bundles[_].type` | `string` | Bundle type, either `snapshot` or `delta` | +| `discovery.name` | `string` | Name of discovery bundle that the OPA instance is configured to download. | +| `discovery.active_revision` | `string` | Opaque revision identifier of the last successful discovery activation. | +| `discovery.last_request` | `string` | RFC3339 timestamp of last discovery bundle request. This timestamp should be >= to the successful request timestamp in normal operation. | | `discovery.last_successful_request` | `string` | RFC3339 timestamp of last successful discovery bundle request. This timestamp should be >= to the successful download timestamp in normal operation. | -| `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. | +| `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. | +| `decision_logs.code` | `string` | If present, indicates error(s) occurred during decision log upload event. | +| `decision_logs.message` | `string` | Human readable messages describing the error(s). | +| `decision_logs.http_code` | `number` | If present, indicates an erroneous HTTP status code that OPA received during a decision log upload event. | +| `decision_logs.metrics` | `object` | Metrics from the last decision log upload event. | +| `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 discovery bundle download or activation failed, the status update will contain the following additional fields. diff --git a/plugins/logs/plugin.go b/plugins/logs/plugin.go index 093fe586a3..f83365a4ca 100644 --- a/plugins/logs/plugin.go +++ b/plugins/logs/plugin.go @@ -11,7 +11,6 @@ import ( "fmt" "math" "math/rand" - "net/http" "net/url" "reflect" "strings" @@ -25,7 +24,9 @@ import ( "github.com/open-policy-agent/opa/logging" "github.com/open-policy-agent/opa/metrics" "github.com/open-policy-agent/opa/plugins" + lstat "github.com/open-policy-agent/opa/plugins/logs/status" "github.com/open-policy-agent/opa/plugins/rest" + "github.com/open-policy-agent/opa/plugins/status" "github.com/open-policy-agent/opa/rego" "github.com/open-policy-agent/opa/server" "github.com/open-policy-agent/opa/storage" @@ -406,6 +407,7 @@ type Plugin struct { limiter *rate.Limiter metrics metrics.Metrics logger logging.Logger + status *lstat.Status } type reconfigure struct { @@ -496,6 +498,7 @@ func New(parsedConfig *Config, manager *plugins.Manager) *Plugin { enc: newChunkEncoder(*parsedConfig.Reporting.UploadSizeLimitBytes), reconfig: make(chan reconfigure), logger: manager.Logger().WithFields(map[string]interface{}{"plugin": Name}), + status: &lstat.Status{}, } if parsedConfig.Reporting.MaxDecisionsPerSecond != nil { @@ -767,6 +770,19 @@ func (p *Plugin) loop() { func (p *Plugin) doOneShot(ctx context.Context) error { uploaded, err := p.oneShot(ctx) + p.mtx.Lock() + defer p.mtx.Unlock() + + p.status.SetError(err) + + if p.metrics != nil { + p.status.Metrics = p.metrics + } + + if s := status.Lookup(p.manager); s != nil { + s.UpdateDecisionLogsStatus(*p.status) + } + if err != nil { p.logger.Error("%v.", err) } else if uploaded { @@ -1036,7 +1052,7 @@ func uploadChunk(ctx context.Context, client rest.Client, uploadPath string, dat defer util.Close(resp) if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return fmt.Errorf("log upload failed, server replied with HTTP %v %v", resp.StatusCode, http.StatusText(resp.StatusCode)) + return lstat.HTTPError{StatusCode: resp.StatusCode} } return nil diff --git a/plugins/logs/plugin_test.go b/plugins/logs/plugin_test.go index cd6091ba90..0a1134157a 100644 --- a/plugins/logs/plugin_test.go +++ b/plugins/logs/plugin_test.go @@ -323,6 +323,10 @@ func TestPluginStartSameInput(t *testing.T) { if !reflect.DeepEqual(chunk3[expLen3-1], exp) { t.Fatalf("Expected %+v but got %+v", exp, chunk3[expLen3-1]) } + + if fixture.plugin.status.Code != "" { + t.Fatal("expected no error in status update") + } } func TestPluginStartChangingInputValues(t *testing.T) { @@ -750,6 +754,155 @@ func TestPluginRateLimitFloat(t *testing.T) { compareLogEvent(t, chunk[0], exp) } +func TestPluginStatusUpdateHTTPError(t *testing.T) { + ctx := context.Background() + + fixture := newTestFixture(t, testFixtureOptions{ReportingUploadSizeLimitBytes: 300}) + defer fixture.server.stop() + + fixture.server.ch = make(chan []EventV1, 3) + + var input interface{} = map[string]interface{}{"method": "GET"} + var result1 interface{} = false + + _ = fixture.plugin.Log(ctx, logServerInfo("abc", input, result1)) + _ = fixture.plugin.Log(ctx, logServerInfo("def", input, result1)) + _ = fixture.plugin.Log(ctx, logServerInfo("ghi", input, result1)) + + bufLen := fixture.plugin.buffer.Len() + if bufLen < 1 { + t.Fatal("Expected buffer length of at least 1") + } + + fixture.server.expCode = 500 + err := fixture.plugin.doOneShot(ctx) + if err == nil { + t.Fatal("Expected error") + } + + <-fixture.server.ch + + if fixture.plugin.buffer.Len() < bufLen { + t.Fatal("Expected buffer to be preserved") + } + + if fixture.plugin.status.HTTPCode != "500" { + t.Fatal("expected http_code to be 500 instead of ", fixture.plugin.status.HTTPCode) + } + + msg := "log upload failed, server replied with HTTP 500 Internal Server Error" + if fixture.plugin.status.Message != msg { + t.Fatalf("expected status message to be %v instead of %v", msg, fixture.plugin.status.Message) + } +} + +func TestPluginStatusUpdate(t *testing.T) { + ctx := context.Background() + testLogger := test.New() + + ts, err := time.Parse(time.RFC3339Nano, "2018-01-01T12:00:00.123456Z") + if err != nil { + panic(err) + } + + numDecisions := 1 // 1 decision per second + fixture := newTestFixture(t, testFixtureOptions{ + ConsoleLogger: testLogger, + ReportingMaxDecisionsPerSecond: float64(numDecisions), + ReportingUploadSizeLimitBytes: 300, + }) + defer fixture.server.stop() + + fixture.server.ch = make(chan []EventV1, 1) + + fixture.plugin.metrics = metrics.New() + + var input interface{} = map[string]interface{}{"method": "GET"} + var result interface{} = false + + event1 := &server.Info{ + DecisionID: "abc", + Path: "foo/bar", + Input: &input, + Results: &result, + RemoteAddr: "test-1", + Timestamp: ts, + } + + event2 := &server.Info{ + DecisionID: "def", + Path: "foo/baz", + Input: &input, + Results: &result, + RemoteAddr: "test-2", + Timestamp: ts, + } + + event3 := &server.Info{ + DecisionID: "ghi", + Path: "foo/aux", + Input: &input, + Results: &result, + RemoteAddr: "test-3", + Timestamp: ts, + } + + _ = fixture.plugin.Log(ctx, event1) // event 1 should be written into the encoder + + fixture.plugin.mtx.Lock() + if fixture.plugin.enc.bytesWritten == 0 { + t.Fatal("Expected event to be written into the encoder") + } + fixture.plugin.mtx.Unlock() + + // Create a status plugin that logs to console + pluginConfig := []byte(`{ + "console": true, + }`) + + config, _ := status.ParseConfig(pluginConfig, fixture.manager.Services(), nil) + p := status.New(config, fixture.manager).WithMetrics(fixture.plugin.metrics) + + fixture.manager.Register(status.Name, p) + if err := fixture.manager.Start(ctx); err != nil { + t.Fatal(err) + } + + _ = fixture.plugin.Log(ctx, event2) // event 2 should not be written into the encoder as rate limit exceeded + _ = fixture.plugin.Log(ctx, event3) // event 3 should not be written into the encoder as rate limit exceeded + + // Trigger a status update + fixture.server.expCode = 200 + err = fixture.plugin.doOneShot(ctx) + if err != nil { + t.Fatal("Unexpected error") + } + + <-fixture.server.ch + + // Give the logger / console some time to process and print the events + time.Sleep(10 * time.Millisecond) + p.Stop(ctx) + + entries := testLogger.Entries() + if len(entries) == 0 { + t.Fatal("Expected log entries but got none") + } + + // Pick the last entry as it should have the decision log update + e := entries[len(entries)-1] + + if _, ok := e.Fields["decision_logs"]; !ok { + t.Fatal("Expected decision_log status update") + } + + exp := map[string]interface{}{"metrics": map[string]interface{}{"counter_decision_logs_dropped": json.Number("2")}} + + if !reflect.DeepEqual(e.Fields["decision_logs"], exp) { + t.Fatalf("Expected %v but got %v", exp, e.Fields["decision_logs"]) + } +} + func TestPluginRateLimitRequeue(t *testing.T) { ctx := context.Background() diff --git a/plugins/logs/status/status.go b/plugins/logs/status/status.go new file mode 100644 index 0000000000..4a9b24f7bb --- /dev/null +++ b/plugins/logs/status/status.go @@ -0,0 +1,58 @@ +// Copyright 2023 The OPA Authors. All rights reserved. +// Use of this source code is governed by an Apache2 +// license that can be found in the LICENSE file. + +package status + +import ( + "encoding/json" + "errors" + "fmt" + "net/http" + "strconv" + + "github.com/open-policy-agent/opa/metrics" +) + +const ( + errCode = "decision_log_error" +) + +// Status represents the status of processing a decision log. +type Status struct { + Code string `json:"code,omitempty"` + Message string `json:"message,omitempty"` + HTTPCode json.Number `json:"http_code,omitempty"` + Metrics metrics.Metrics `json:"metrics,omitempty"` +} + +// SetError updates the status object to reflect a failure to upload or +// process a log. If err is nil, the error status is cleared. +func (s *Status) SetError(err error) { + var httpError HTTPError + + switch { + case err == nil: + s.Code = "" + s.HTTPCode = "" + s.Message = "" + + case errors.As(err, &httpError): + s.Code = errCode + s.HTTPCode = json.Number(strconv.Itoa(httpError.StatusCode)) + s.Message = err.Error() + + default: + s.Code = errCode + s.HTTPCode = "" + s.Message = err.Error() + } +} + +type HTTPError struct { + StatusCode int +} + +func (e HTTPError) Error() string { + return fmt.Sprintf("log upload failed, server replied with HTTP %v %v", e.StatusCode, http.StatusText(e.StatusCode)) +} diff --git a/plugins/status/plugin.go b/plugins/status/plugin.go index 5c84adf578..2e4f2f36b4 100644 --- a/plugins/status/plugin.go +++ b/plugins/status/plugin.go @@ -12,6 +12,7 @@ import ( "net/http" "reflect" + lstat "github.com/open-policy-agent/opa/plugins/logs/status" prom "github.com/prometheus/client_golang/prometheus" "github.com/open-policy-agent/opa/logging" @@ -31,32 +32,35 @@ type Logger interface { // 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"` - Plugins map[string]*plugins.Status `json:"plugins,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"` + DecisionLogs *lstat.Status `json:"decision_logs,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. type Plugin struct { - manager *plugins.Manager - config Config - bundleCh chan bundle.Status // Deprecated: Use bulk bundle status updates instead - lastBundleStatus *bundle.Status // Deprecated: Use bulk bundle status updates instead - bulkBundleCh chan map[string]*bundle.Status - lastBundleStatuses map[string]*bundle.Status - discoCh chan bundle.Status - lastDiscoStatus *bundle.Status - pluginStatusCh chan map[string]*plugins.Status - lastPluginStatuses map[string]*plugins.Status - queryCh chan chan *UpdateRequestV1 - stop chan chan struct{} - reconfig chan interface{} - metrics metrics.Metrics - logger logging.Logger - trigger chan trigger + manager *plugins.Manager + config Config + bundleCh chan bundle.Status // Deprecated: Use bulk bundle status updates instead + lastBundleStatus *bundle.Status // Deprecated: Use bulk bundle status updates instead + bulkBundleCh chan map[string]*bundle.Status + lastBundleStatuses map[string]*bundle.Status + discoCh chan bundle.Status + lastDiscoStatus *bundle.Status + pluginStatusCh chan map[string]*plugins.Status + decisionLogsCh chan lstat.Status + lastDecisionLogsStatus *lstat.Status + lastPluginStatuses map[string]*plugins.Status + queryCh chan chan *UpdateRequestV1 + stop chan chan struct{} + reconfig chan interface{} + metrics metrics.Metrics + logger logging.Logger + trigger chan trigger } // Config contains configuration for the plugin. @@ -192,6 +196,7 @@ func New(parsedConfig *Config, manager *plugins.Manager) *Plugin { bundleCh: make(chan bundle.Status), bulkBundleCh: make(chan map[string]*bundle.Status), discoCh: make(chan bundle.Status), + decisionLogsCh: make(chan lstat.Status), stop: make(chan chan struct{}), reconfig: make(chan interface{}), pluginStatusCh: make(chan map[string]*plugins.Status), @@ -279,6 +284,11 @@ func (p *Plugin) UpdateDiscoveryStatus(status bundle.Status) { p.discoCh <- status } +// UpdateDecisionLogsStatus notifies the plugin that status of a decision log upload event. +func (p *Plugin) UpdateDecisionLogsStatus(status lstat.Status) { + p.decisionLogsCh <- status +} + // UpdatePluginStatus notifies the plugin that a plugin status was updated. func (p *Plugin) UpdatePluginStatus(status map[string]*plugins.Status) { p.pluginStatusCh <- status @@ -358,6 +368,16 @@ func (p *Plugin) loop() { p.logger.Info("Status update sent successfully in response to discovery update.") } } + case status := <-p.decisionLogsCh: + p.lastDecisionLogsStatus = &status + if *p.config.Trigger == plugins.TriggerPeriodic { + err := p.oneShot(ctx) + if err != nil { + p.logger.Error("%v.", err) + } else { + p.logger.Info("Status update sent successfully in response to discovery update.") + } + } case newConfig := <-p.reconfig: p.reconfigure(newConfig) case respCh := <-p.queryCh: @@ -437,11 +457,12 @@ func (p *Plugin) reconfigure(config interface{}) { func (p *Plugin) snapshot() *UpdateRequestV1 { s := &UpdateRequestV1{ - Labels: p.manager.Labels(), - Discovery: p.lastDiscoStatus, - Bundle: p.lastBundleStatus, - Bundles: p.lastBundleStatuses, - Plugins: p.lastPluginStatuses, + Labels: p.manager.Labels(), + Discovery: p.lastDiscoStatus, + DecisionLogs: p.lastDecisionLogsStatus, + 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 0f5e220783..06344d8480 100644 --- a/plugins/status/plugin_test.go +++ b/plugins/status/plugin_test.go @@ -23,6 +23,8 @@ import ( inmem "github.com/open-policy-agent/opa/storage/inmem/test" "github.com/open-policy-agent/opa/util" "github.com/open-policy-agent/opa/version" + + lstat "github.com/open-policy-agent/opa/plugins/logs/status" ) func TestMain(m *testing.M) { @@ -464,6 +466,49 @@ func TestPluginStartDiscovery(t *testing.T) { } } +func TestPluginStartDecisionLogs(t *testing.T) { + + fixture := newTestFixture(t, nil) + fixture.server.ch = make(chan UpdateRequestV1) + defer fixture.server.stop() + + ctx := context.Background() + + err := fixture.plugin.Start(ctx) + if err != nil { + t.Fatal(err) + } + defer fixture.plugin.Stop(ctx) + + // Ignore the plugin updating its status (tested elsewhere) + <-fixture.server.ch + + status := &lstat.Status{ + Code: "decision_log_error", + Message: "Upload Failed", + HTTPCode: "400", + } + + fixture.plugin.UpdateDecisionLogsStatus(*status) + result := <-fixture.server.ch + + exp := UpdateRequestV1{ + Labels: map[string]string{ + "id": "test-instance-id", + "app": "example-app", + "version": version.Version, + }, + DecisionLogs: status, + Plugins: map[string]*plugins.Status{ + "status": {State: plugins.StateOK}, + }, + } + + if !reflect.DeepEqual(result, exp) { + t.Fatalf("Expected: %+v but got: %+v", exp, result) + } +} + func TestPluginBadAuth(t *testing.T) { fixture := newTestFixture(t, nil) ctx := context.Background()