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()