diff --git a/config/config.go b/config/config.go index 8834ce7782..9926df211b 100644 --- a/config/config.go +++ b/config/config.go @@ -27,6 +27,13 @@ type Config struct { Plugins map[string]json.RawMessage `json:"plugins"` DefaultDecision *string `json:"default_decision"` DefaultAuthorizationDecision *string `json:"default_authorization_decision"` + MetricsProvider *MetricsProviderConfig `json:"metrics_provider"` +} + +// MetricsProviderConfig represents metrics_provider config section +type MetricsProviderConfig struct { + Name string `json:"name"` + Config json.RawMessage `json:"config"` } // ParseConfig returns a valid Config object with defaults injected. The id @@ -86,6 +93,10 @@ func (c *Config) validateAndInjectDefaults(id string) error { c.Labels["id"] = id c.Labels["version"] = version.Version + if c.MetricsProvider == nil { + c.MetricsProvider = &MetricsProviderConfig{Name: "prometheus"} + } + return nil } diff --git a/docs/content/configuration.md b/docs/content/configuration.md index ec85808140..826e03ae8d 100644 --- a/docs/content/configuration.md +++ b/docs/content/configuration.md @@ -52,6 +52,9 @@ status: service: acmecorp default_decision: /http/example/authz/allow + +metrics_provider: + name: prometheus ``` ## Environment Variable Substitution @@ -343,6 +346,7 @@ server provenance, etc. | --- | --- | --- | --- | | `status.service` | `string` | Yes | Name of service to use to contact remote server. | | `status.partition_name` | `string` | No | Path segment to include in status updates. | +| `status.include_metrics` | `boolean` | (default: `false`) | Include Prometheus metrics in status updates. | ## Decision Logs @@ -367,3 +371,15 @@ server provenance, etc. | `discovery.decision` | `string` | No (default: value of `discovery.name` configuration field) | Name of the OPA query that will be used to calculate the configuration | | `discovery.polling.min_delay_seconds` | `int64` | No (default: `60`) | Minimum amount of time to wait between configuration downloads. | | `discovery.polling.max_delay_seconds` | `int64` | No (default: `120`) | Maximum amount of time to wait between configuration downloads. | + + +## Metrics provider + +| Field | Type | Required | Description | +| --- | --- | --- | --- | +| `metrics_provider.name` | `string` | No | Name of the metrics provider to use. | +| `metrics_provider.config` | `object` | No | Provider-specific configuration (not used as of now). | + +Available metrics providers: +* `prometheus` (default) +* (empty string): do not collect metrics diff --git a/docs/content/status.md b/docs/content/status.md index b71fdae6a5..aea0ed24fb 100644 --- a/docs/content/status.md +++ b/docs/content/status.md @@ -46,6 +46,75 @@ on the agent, updates will be sent to `/status`. "last_successful_download": "2018-01-01T00:00:00.000Z", "last_successful_activation": "2018-01-01T00:00:00.000Z" } + }, + "metrics": { + "prometheus": [ + { + "help": "A summary of the GC invocation durations.", + "name": "go_gc_duration_seconds", + "type": 2, + "metric": [ + { + "summary": { + "quantile": [ + { + "quantile": 0, + "value": 0.000044358 + }, + { + "quantile": 0.25, + "value": 0.000045003 + }, + { + "quantile": 0.5, + "value": 0.000049726 + }, + { + "quantile": 0.75, + "value": 0.000219553 + }, + { + "quantile": 1, + "value": 0.000219553 + } + ], + "sample_count": 4, + "sample_sum": 0.00035864 + } + } + ] + }, + { + "help": "Number of goroutines that currently exist.", + "name": "go_goroutines", + "type": 1, + "metric": [ + { + "gauge": { + "value": 11 + } + } + ] + }, + { + "help": "Information about the Go environment.", + "name": "go_info", + "type": 1, + "metric": [ + { + "gauge": { + "value": 1 + }, + "label": [ + { + "name": "version", + "value": "go1.12.7" + } + ] + } + ] + } + ] } } ``` @@ -68,6 +137,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. | +| `metrics` | `object` | Application metrics. Optional, single key object. | +| `metrics[provider_name]` | JSON (`interface{}`) | Metrics in provider-dependent format. | If the bundle download or activation failed, the status update will contain the following additional fields. diff --git a/internal/metrics/dummy.go b/internal/metrics/dummy.go new file mode 100644 index 0000000000..1e47b633f1 --- /dev/null +++ b/internal/metrics/dummy.go @@ -0,0 +1,25 @@ +// Copyright 2019 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 metrics + +import ( + "net/http" +) + +type dummyProvider struct{} + +func (dummyProvider) RegisterEndpoints(registrar func(path, method string, handler http.Handler)) {} + +func (dummyProvider) InstrumentHandler(handler http.Handler, label string) http.Handler { + return handler +} + +func (dummyProvider) Gather() (interface{}, error) { + return nil, nil +} + +func (dummyProvider) Name() string { + return "" +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go new file mode 100644 index 0000000000..b7ed48cc95 --- /dev/null +++ b/internal/metrics/metrics.go @@ -0,0 +1,26 @@ +// Copyright 2019 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 metrics + +import ( + "encoding/json" + + "github.com/pkg/errors" + + "github.com/open-policy-agent/opa/internal/metrics/prometheus" + "github.com/open-policy-agent/opa/metrics" +) + +// NewGlobalMetrics creates a metrics provider instance given its name and config +func NewGlobalMetrics(name string, config json.RawMessage) (metrics.GlobalMetrics, error) { + switch name { + case "": + return &dummyProvider{}, nil + case prometheus.ProviderName: + return prometheus.NewPrometheusProvider(), nil + default: + return nil, errors.Errorf("Invalid metrics provider %s.", name) + } +} diff --git a/internal/metrics/prometheus/prometheus.go b/internal/metrics/prometheus/prometheus.go new file mode 100644 index 0000000000..1446565dcd --- /dev/null +++ b/internal/metrics/prometheus/prometheus.go @@ -0,0 +1,107 @@ +// Copyright 2019 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 prometheus + +import ( + "bufio" + "net" + "net/http" + "strconv" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +// ProviderName is the Prometheus provider name +const ProviderName = "prometheus" + +// Provider is the prometheus +type Provider struct { + registry *prometheus.Registry + durationHistogram *prometheus.HistogramVec + cancellationCounters *prometheus.CounterVec +} + +// NewPrometheusProvider creates new instance of the prometheus provider +func NewPrometheusProvider() *Provider { + registry := prometheus.NewRegistry() + registry.MustRegister(prometheus.NewGoCollector()) + durationHistogram := prometheus.NewHistogramVec( + prometheus.HistogramOpts{ + Name: "http_request_duration_seconds", + Help: "A histogram of duration for requests.", + }, + []string{"code", "handler", "method"}, + ) + registry.MustRegister(durationHistogram) + + cancellationCounters := prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "http_request_cancellations", + Help: "A count of cancelled requests.", + }, + []string{"code", "handler", "method"}, + ) + + registry.MustRegister(cancellationCounters) + return &Provider{ + registry: registry, + durationHistogram: durationHistogram, + cancellationCounters: cancellationCounters, + } +} + +// RegisterEndpoints registers `/metrics` endpoint +func (p *Provider) RegisterEndpoints(registrar func(path, method string, handler http.Handler)) { + registrar("/metrics", http.MethodGet, promhttp.HandlerFor(p.registry, promhttp.HandlerOpts{})) +} + +// InstrumentHandler returned wrapped HTTP handler with added prometheus instrumentation +func (p *Provider) InstrumentHandler(handler http.Handler, label string) http.Handler { + durationCollector := p.durationHistogram.MustCurryWith(prometheus.Labels{"handler": label}) + cancellationsCollector := p.cancellationCounters.MustCurryWith(prometheus.Labels{"handler": label}) + return promhttp.InstrumentHandlerDuration(durationCollector, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + csrw := &captureStatusResponseWriter{ResponseWriter: w, status: http.StatusOK} + var rw http.ResponseWriter + if h, ok := w.(http.Hijacker); ok { + rw = &hijacker{ResponseWriter: csrw, hijacker: h} + } else { + rw = csrw + } + handler.ServeHTTP(rw, r) + if r.Context().Err() != nil { + cancellationsCollector.With(prometheus.Labels{"code": strconv.Itoa(csrw.status), "method": r.Method}).Inc() + } + })) +} + +// Gather collects and returns all registered metrics +func (p *Provider) Gather() (interface{}, error) { + return p.registry.Gather() +} + +// Name returns the provider name +func (p *Provider) Name() string { + return ProviderName +} + +type captureStatusResponseWriter struct { + http.ResponseWriter + status int +} + +type hijacker struct { + http.ResponseWriter + hijacker http.Hijacker +} + +func (h *hijacker) Hijack() (net.Conn, *bufio.ReadWriter, error) { + return h.hijacker.Hijack() +} + +func (c *captureStatusResponseWriter) WriteHeader(statusCode int) { + c.ResponseWriter.WriteHeader(statusCode) + c.status = statusCode +} diff --git a/metrics/global.go b/metrics/global.go new file mode 100644 index 0000000000..a4eecd58a9 --- /dev/null +++ b/metrics/global.go @@ -0,0 +1,17 @@ +// Copyright 2018 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 metrics + +import ( + "net/http" +) + +// GlobalMetrics abstracts metric providers API +type GlobalMetrics interface { + RegisterEndpoints(registrar func(path, method string, handler http.Handler)) + InstrumentHandler(handler http.Handler, label string) http.Handler + Gather() (interface{}, error) + Name() string +} diff --git a/plugins/discovery/discovery.go b/plugins/discovery/discovery.go index 07358a30d5..c622839fc6 100644 --- a/plugins/discovery/discovery.go +++ b/plugins/discovery/discovery.go @@ -9,6 +9,7 @@ import ( "context" "encoding/json" "fmt" + "github.com/open-policy-agent/opa/metrics" "github.com/open-policy-agent/opa/ast" bundleApi "github.com/open-policy-agent/opa/bundle" @@ -27,12 +28,13 @@ import ( // started it will periodically download a configuration bundle and try to // reconfigure the OPA. type Discovery struct { - manager *plugins.Manager - config *Config - factories map[string]plugins.Factory - downloader *download.Downloader // discovery bundle downloader - status *bundle.Status // discovery status - etag string // discovery bundle etag for caching purposes + manager *plugins.Manager + config *Config + factories map[string]plugins.Factory + downloader *download.Downloader // discovery bundle downloader + status *bundle.Status // discovery status + etag string // discovery bundle etag for caching purposes + globalMetrics metrics.GlobalMetrics } // Factories provides a set of factory functions to use for @@ -43,6 +45,13 @@ func Factories(fs map[string]plugins.Factory) func(*Discovery) { } } +// WithMetrics sets the GlobalMetrics instance to use for instantiations +func WithMetrics(globalMetrics metrics.GlobalMetrics) func(*Discovery) { + return func(d *Discovery) { + d.globalMetrics = globalMetrics + } +} + // New returns a new discovery plugin. func New(manager *plugins.Manager, opts ...func(*Discovery)) (*Discovery, error) { @@ -59,7 +68,7 @@ func New(manager *plugins.Manager, opts ...func(*Discovery)) (*Discovery, error) if err != nil { return nil, err } else if config == nil { - if _, err := getPluginSet(result.factories, manager, manager.Config); err != nil { + if _, err := getPluginSet(result.factories, manager, manager.Config, result.globalMetrics); err != nil { return nil, err } return result, nil @@ -144,7 +153,7 @@ func (c *Discovery) processUpdate(ctx context.Context, u download.Update) { func (c *Discovery) reconfigure(ctx context.Context, u download.Update) error { - config, ps, err := processBundle(ctx, c.manager, c.factories, u.Bundle, c.config.query) + config, ps, err := processBundle(ctx, c.manager, c.factories, u.Bundle, c.config.query, c.globalMetrics) if err != nil { return err } @@ -190,14 +199,14 @@ func (c *Discovery) logrusFields() logrus.Fields { } } -func processBundle(ctx context.Context, manager *plugins.Manager, factories map[string]plugins.Factory, b *bundleApi.Bundle, query string) (*config.Config, *pluginSet, error) { +func processBundle(ctx context.Context, manager *plugins.Manager, factories map[string]plugins.Factory, b *bundleApi.Bundle, query string, globalMetrics metrics.GlobalMetrics) (*config.Config, *pluginSet, error) { config, err := evaluateBundle(ctx, manager.ID, manager.Info, b, query) if err != nil { return nil, nil, err } - ps, err := getPluginSet(factories, manager, config) + ps, err := getPluginSet(factories, manager, config, globalMetrics) return config, ps, err } @@ -257,7 +266,7 @@ type pluginfactory struct { config interface{} } -func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager, config *config.Config) (*pluginSet, error) { +func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager, config *config.Config, globalMetrics metrics.GlobalMetrics) (*pluginSet, error) { // Parse and validate plugin configurations. pluginNames := []string{} @@ -330,7 +339,7 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager } if statusConfig != nil { - p, created := getStatusPlugin(manager, statusConfig) + p, created := getStatusPlugin(manager, statusConfig, globalMetrics) if created { starts = append(starts, p) } else if p != nil { @@ -366,12 +375,12 @@ func getDecisionLogsPlugin(m *plugins.Manager, config *logs.Config) (plugin *log return plugin, created } -func getStatusPlugin(m *plugins.Manager, config *status.Config) (plugin *status.Plugin, created bool) { +func getStatusPlugin(m *plugins.Manager, config *status.Config, globalMetrics metrics.GlobalMetrics) (plugin *status.Plugin, created bool) { plugin = status.Lookup(m) if plugin == nil { - plugin = status.New(config, m) + plugin = status.New(config, m).WithMetrics(globalMetrics) m.Register(status.Name, plugin) registerBundleStatusUpdates(m) created = true diff --git a/plugins/discovery/discovery_test.go b/plugins/discovery/discovery_test.go index db027aa3fe..afe9d57aba 100644 --- a/plugins/discovery/discovery_test.go +++ b/plugins/discovery/discovery_test.go @@ -120,7 +120,7 @@ func TestProcessBundle(t *testing.T) { } `) - _, ps, err := processBundle(ctx, manager, nil, initialBundle, "data.config") + _, ps, err := processBundle(ctx, manager, nil, initialBundle, "data.config", nil) if err != nil { t.Fatal(err) } @@ -139,7 +139,7 @@ func TestProcessBundle(t *testing.T) { } `) - _, ps, err = processBundle(ctx, manager, nil, updatedBundle, "data.config") + _, ps, err = processBundle(ctx, manager, nil, updatedBundle, "data.config", nil) if err != nil { t.Fatal(err) } @@ -156,7 +156,7 @@ func TestProcessBundle(t *testing.T) { } `) - _, _, err = processBundle(ctx, manager, nil, updatedBundle, "data.config") + _, _, err = processBundle(ctx, manager, nil, updatedBundle, "data.config", nil) if err == nil { t.Fatal("Expected error but got success") } @@ -419,7 +419,7 @@ bundle: service: s2 ` manager := getTestManager(t, conf) - _, err := getPluginSet(nil, manager, manager.Config) + _, err := getPluginSet(nil, manager, manager.Config, nil) if err != nil { t.Fatalf("Unexpected error: %s", err) } @@ -455,7 +455,7 @@ bundles: service: s1 ` manager := getTestManager(t, conf) - _, err := getPluginSet(nil, manager, manager.Config) + _, err := getPluginSet(nil, manager, manager.Config, nil) if err != nil { t.Fatalf("Unexpected error: %s", err) } diff --git a/plugins/status/plugin.go b/plugins/status/plugin.go index 4b9f52ae48..1a187ae17f 100644 --- a/plugins/status/plugin.go +++ b/plugins/status/plugin.go @@ -11,6 +11,7 @@ import ( "net/http" "reflect" + "github.com/open-policy-agent/opa/metrics" "github.com/open-policy-agent/opa/plugins" "github.com/open-policy-agent/opa/plugins/bundle" "github.com/open-policy-agent/opa/util" @@ -25,6 +26,7 @@ type UpdateRequestV1 struct { 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"` } // Plugin implements status reporting. Updates can be triggered by the caller. @@ -39,12 +41,14 @@ type Plugin struct { lastDiscoStatus *bundle.Status stop chan chan struct{} reconfig chan interface{} + globalMetrics metrics.GlobalMetrics } // Config contains configuration for the plugin. type Config struct { - Service string `json:"service"` - PartitionName string `json:"partition_name,omitempty"` + Service string `json:"service"` + PartitionName string `json:"partition_name,omitempty"` + IncludeMetrics bool `json:"include_metrics"` } func (c *Config) validateAndInjectDefaults(services []string) error { @@ -91,8 +95,7 @@ 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 { - - plugin := &Plugin{ + return &Plugin{ manager: manager, config: *parsedConfig, bundleCh: make(chan bundle.Status), @@ -101,8 +104,12 @@ func New(parsedConfig *Config, manager *plugins.Manager) *Plugin { stop: make(chan chan struct{}), reconfig: make(chan interface{}), } +} - return plugin +// WithMetrics sets the global metrics provider to be used by the plugin. +func (p *Plugin) WithMetrics(globalMetrics metrics.GlobalMetrics) *Plugin { + p.globalMetrics = globalMetrics + return p } // Name identifies the plugin on manager. @@ -195,7 +202,6 @@ func (p *Plugin) loop() { } func (p *Plugin) oneShot(ctx context.Context) error { - req := &UpdateRequestV1{ Labels: p.manager.Labels(), Discovery: p.lastDiscoStatus, @@ -203,6 +209,15 @@ func (p *Plugin) oneShot(ctx context.Context) error { Bundles: p.lastBundleStatuses, } + if p.config.IncludeMetrics && p.globalMetrics != nil { + name := p.globalMetrics.Name() + globalMetrics, err := p.globalMetrics.Gather() + if err != nil { + p.logError("Cannot gather metrics: %v.", err) + } else if name != "" { + req.Metrics = map[string]interface{}{name: globalMetrics} + } + } resp, err := p.manager.Client(p.config.Service). WithJSON(req). Do(ctx, "POST", fmt.Sprintf("/status/%v", p.config.PartitionName)) diff --git a/plugins/status/plugin_test.go b/plugins/status/plugin_test.go index 9bd09ff572..072ce10167 100644 --- a/plugins/status/plugin_test.go +++ b/plugins/status/plugin_test.go @@ -14,6 +14,7 @@ import ( "testing" "time" + "github.com/open-policy-agent/opa/metrics" "github.com/open-policy-agent/opa/plugins" "github.com/open-policy-agent/opa/plugins/bundle" "github.com/open-policy-agent/opa/storage/inmem" @@ -30,7 +31,7 @@ func TestMain(m *testing.M) { func TestPluginStart(t *testing.T) { - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) fixture.server.ch = make(chan UpdateRequestV1) defer fixture.server.stop() @@ -60,7 +61,7 @@ func TestPluginStart(t *testing.T) { func TestPluginStartBulkUpdate(t *testing.T) { - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) fixture.server.ch = make(chan UpdateRequestV1) defer fixture.server.stop() @@ -90,7 +91,7 @@ func TestPluginStartBulkUpdate(t *testing.T) { func TestPluginStartBulkUpdateMultiple(t *testing.T) { - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) fixture.server.ch = make(chan UpdateRequestV1) defer fixture.server.stop() @@ -142,7 +143,7 @@ func TestPluginStartBulkUpdateMultiple(t *testing.T) { func TestPluginStartDiscovery(t *testing.T) { - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) fixture.server.ch = make(chan UpdateRequestV1) defer fixture.server.stop() @@ -171,7 +172,7 @@ func TestPluginStartDiscovery(t *testing.T) { } func TestPluginBadAuth(t *testing.T) { - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) ctx := context.Background() fixture.server.expCode = 401 defer fixture.server.stop() @@ -183,7 +184,7 @@ func TestPluginBadAuth(t *testing.T) { } func TestPluginBadPath(t *testing.T) { - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) ctx := context.Background() fixture.server.expCode = 404 defer fixture.server.stop() @@ -195,7 +196,7 @@ func TestPluginBadPath(t *testing.T) { } func TestPluginBadStatus(t *testing.T) { - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) ctx := context.Background() fixture.server.expCode = 500 defer fixture.server.stop() @@ -208,7 +209,7 @@ func TestPluginBadStatus(t *testing.T) { func TestPluginReconfigure(t *testing.T) { ctx := context.Background() - fixture := newTestFixture(t) + fixture := newTestFixture(t, nil) defer fixture.server.stop() if err := fixture.plugin.Start(ctx); err != nil { @@ -230,13 +231,35 @@ func TestPluginReconfigure(t *testing.T) { } } +func TestMetrics(t *testing.T) { + testMetrics := []interface{}{"a", "b", "c"} + fixture := newTestFixture(t, &testMetricsProvider{data: testMetrics}) + fixture.server.ch = make(chan UpdateRequestV1) + fixture.plugin.config.IncludeMetrics = true + defer fixture.server.stop() + + ctx := context.Background() + + fixture.plugin.Start(ctx) + defer fixture.plugin.Stop(ctx) + + status := testStatus() + + fixture.plugin.BulkUpdateBundleStatus(map[string]*bundle.Status{"bundle": status}) + result := <-fixture.server.ch + + if !reflect.DeepEqual(result.Metrics, map[string]interface{}{"test": testMetrics}) { + t.Error("Test metrics were not returned") + } +} + type testFixture struct { manager *plugins.Manager plugin *Plugin server *testServer } -func newTestFixture(t *testing.T) testFixture { +func newTestFixture(t *testing.T, globalMetrics metrics.GlobalMetrics) testFixture { ts := testServer{ t: t, @@ -271,9 +294,9 @@ func newTestFixture(t *testing.T) testFixture { "service": "example", }`)) - config, _ := ParseConfig([]byte(pluginConfig), manager.Services()) + config, _ := ParseConfig(pluginConfig, manager.Services()) - p := New(config, manager) + p := New(config, manager).WithMetrics(globalMetrics) return testFixture{ manager: manager, @@ -327,3 +350,22 @@ func testStatus() *bundle.Status { return &status } + +type testMetricsProvider struct { + data interface{} +} + +func (t testMetricsProvider) RegisterEndpoints(registrar func(path, method string, handler http.Handler)) { +} + +func (t testMetricsProvider) InstrumentHandler(handler http.Handler, label string) http.Handler { + return handler +} + +func (t testMetricsProvider) Gather() (interface{}, error) { + return t.data, nil +} + +func (t testMetricsProvider) Name() string { + return "test" +} diff --git a/runtime/runtime.go b/runtime/runtime.go index bf176137d4..e2a4cd5982 100644 --- a/runtime/runtime.go +++ b/runtime/runtime.go @@ -11,6 +11,7 @@ import ( "crypto/tls" "crypto/x509" "fmt" + "github.com/open-policy-agent/opa/metrics" "io" "os" "os/signal" @@ -19,6 +20,7 @@ import ( "time" "github.com/open-policy-agent/opa/ast" + imetrics "github.com/open-policy-agent/opa/internal/metrics" "github.com/open-policy-agent/opa/internal/runtime" storedversion "github.com/open-policy-agent/opa/internal/version" "github.com/open-policy-agent/opa/loader" @@ -32,7 +34,7 @@ import ( "github.com/open-policy-agent/opa/version" "github.com/pkg/errors" "github.com/sirupsen/logrus" - fsnotify "gopkg.in/fsnotify.v1" + "gopkg.in/fsnotify.v1" ) var ( @@ -161,7 +163,8 @@ type Runtime struct { // and doesn't have to duplicated here or on the server. info *ast.Term // runtime information provided to evaluation engine - server *server.Server + server *server.Server + globalMetrics metrics.GlobalMetrics } // NewRuntime returns a new Runtime object initialized with params. @@ -221,7 +224,11 @@ func NewRuntime(ctx context.Context, params Params) (*Runtime, error) { return nil, errors.Wrapf(err, "config error") } - disco, err := discovery.New(manager, discovery.Factories(registeredPlugins)) + gm, err := imetrics.NewGlobalMetrics(manager.Config.MetricsProvider.Name, manager.Config.MetricsProvider.Config) + if err != nil { + return nil, errors.Wrapf(err, "config error") + } + disco, err := discovery.New(manager, discovery.Factories(registeredPlugins), discovery.WithMetrics(gm)) if err != nil { return nil, errors.Wrapf(err, "config error") } @@ -229,10 +236,11 @@ func NewRuntime(ctx context.Context, params Params) (*Runtime, error) { manager.Register("discovery", disco) rt := &Runtime{ - Store: store, - Params: params, - Manager: manager, - info: info, + Store: store, + Params: params, + Manager: manager, + info: info, + globalMetrics: gm, } return rt, nil @@ -280,6 +288,7 @@ func (rt *Runtime) Serve(ctx context.Context) error { WithDecisionIDFactory(rt.decisionIDFactory). WithDecisionLoggerWithErr(rt.decisionLogger). WithRuntime(rt.info). + WithMetrics(rt.globalMetrics). Init(ctx) if err != nil { diff --git a/server/server.go b/server/server.go index 8307d0cfbf..f197fa1f8e 100644 --- a/server/server.go +++ b/server/server.go @@ -5,7 +5,6 @@ package server import ( - "bufio" "bytes" "context" "crypto/tls" @@ -27,6 +26,8 @@ import ( "time" "github.com/gorilla/mux" + "github.com/pkg/errors" + "github.com/open-policy-agent/opa/ast" "github.com/open-policy-agent/opa/bundle" "github.com/open-policy-agent/opa/metrics" @@ -43,9 +44,6 @@ import ( "github.com/open-policy-agent/opa/util" "github.com/open-policy-agent/opa/version" "github.com/open-policy-agent/opa/watch" - "github.com/pkg/errors" - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promhttp" ) // AuthenticationScheme enumerates the supported authentication schemes. The @@ -81,7 +79,7 @@ const ( PromHandlerHealth = "health" ) -// map of unsafe buitins +// map of unsafe builtins var unsafeBuiltinsMap = map[string]struct{}{ast.HTTPSend.Name: struct{}{}} // Server represents an instance of OPA running in server mode. @@ -111,6 +109,7 @@ type Server struct { httpListeners []httpListener bundleStatuses map[string]*bundlePlugin.Status bundleStatusMtx *sync.RWMutex + globalMetrics metrics.GlobalMetrics } // Loop will contain all the calls from the server that we'll be listening on. @@ -124,7 +123,6 @@ func New() *Server { // Init initializes the server. This function MUST be called before Loop. func (s *Server) Init(ctx context.Context) (*Server, error) { - s.initRouter() // Add authorization handler. This must come BEFORE authentication handler @@ -263,6 +261,12 @@ func (s *Server) WithStore(store storage.Store) *Server { return s } +// WithMetrics sets the metrics provider used by the server. +func (s *Server) WithMetrics(globalMetrics metrics.GlobalMetrics) *Server { + s.globalMetrics = globalMetrics + return s +} + // WithManager sets the plugins manager used by the server. func (s *Server) WithManager(manager *plugins.Manager) *Server { s.manager = manager @@ -509,43 +513,6 @@ func (s *Server) getListenerForUNIXSocket(u *url.URL) (Loop, httpListener, error } func (s *Server) initRouter() { - - promRegistry := prometheus.NewRegistry() - duration := prometheus.NewHistogramVec( - prometheus.HistogramOpts{ - Name: "http_request_duration_seconds", - Help: "A histogram of duration for requests.", - }, - []string{"code", "handler", "method"}, - ) - v0DataDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerV0Data}) - v1DataDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Data}) - v1PoliciesDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Policies}) - v1QueryDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Query}) - v1CompileDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Compile}) - indexDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerIndex}) - catchAllDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerCatch}) - getHealthDur := duration.MustCurryWith(prometheus.Labels{"handler": PromHandlerHealth}) - promRegistry.MustRegister(duration) - promRegistry.MustRegister(prometheus.NewGoCollector()) - - cancellations := prometheus.NewCounterVec( - prometheus.CounterOpts{ - Name: "http_request_cancellations", - Help: "A count of cancelled requests.", - }, - []string{"code", "handler", "method"}, - ) - v0DataCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerV0Data}) - v1DataCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Data}) - v1PoliciesCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Policies}) - v1QueryCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Query}) - v1CompileCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerV1Compile}) - indexCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerIndex}) - catchAllCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerCatch}) - getHealthCancellations := cancellations.MustCurryWith(prometheus.Labels{"handler": PromHandlerHealth}) - promRegistry.MustRegister(cancellations) - router := s.router if router == nil { @@ -554,8 +521,12 @@ func (s *Server) initRouter() { router.UseEncodedPath() router.StrictSlash(true) - router.Handle("/metrics", promhttp.HandlerFor(promRegistry, promhttp.HandlerOpts{})).Methods(http.MethodGet) - router.Handle("/health", instrumentHandler(s.unversionedGetHealth, getHealthDur, getHealthCancellations)).Methods(http.MethodGet) + if s.globalMetrics != nil { + s.globalMetrics.RegisterEndpoints(func(path, method string, handler http.Handler) { + router.Handle(path, handler).Methods(method) + }) + } + router.Handle("/health", s.instrumentHandler(http.HandlerFunc(s.unversionedGetHealth), PromHandlerHealth)).Methods(http.MethodGet) if s.pprofEnabled { router.HandleFunc("/debug/pprof/", pprof.Index) router.Handle("/debug/pprof/allocs", pprof.Handler("allocs")) @@ -567,52 +538,59 @@ func (s *Server) initRouter() { router.HandleFunc("/debug/pprof/symbol", pprof.Symbol) router.HandleFunc("/debug/pprof/trace", pprof.Trace) } - s.registerHandler(router, 0, "/data/{path:.+}", http.MethodPost, instrumentHandler(s.v0DataPost, v0DataDur, v0DataCancellations)) - s.registerHandler(router, 0, "/data", http.MethodPost, instrumentHandler(s.v0DataPost, v0DataDur, v0DataCancellations)) - s.registerHandler(router, 1, "/data/{path:.+}", http.MethodDelete, instrumentHandler(s.v1DataDelete, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data/{path:.+}", http.MethodPut, instrumentHandler(s.v1DataPut, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data", http.MethodPut, instrumentHandler(s.v1DataPut, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data/{path:.+}", http.MethodGet, instrumentHandler(s.v1DataGet, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data", http.MethodGet, instrumentHandler(s.v1DataGet, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data/{path:.+}", http.MethodPatch, instrumentHandler(s.v1DataPatch, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data", http.MethodPatch, instrumentHandler(s.v1DataPatch, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data/{path:.+}", http.MethodPost, instrumentHandler(s.v1DataPost, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/data", http.MethodPost, instrumentHandler(s.v1DataPost, v1DataDur, v1DataCancellations)) - s.registerHandler(router, 1, "/policies", http.MethodGet, instrumentHandler(s.v1PoliciesList, v1PoliciesDur, v1PoliciesCancellations)) - s.registerHandler(router, 1, "/policies/{path:.+}", http.MethodDelete, instrumentHandler(s.v1PoliciesDelete, v1PoliciesDur, v1PoliciesCancellations)) - s.registerHandler(router, 1, "/policies/{path:.+}", http.MethodGet, instrumentHandler(s.v1PoliciesGet, v1PoliciesDur, v1PoliciesCancellations)) - s.registerHandler(router, 1, "/policies/{path:.+}", http.MethodPut, instrumentHandler(s.v1PoliciesPut, v1PoliciesDur, v1PoliciesCancellations)) - s.registerHandler(router, 1, "/query", http.MethodGet, instrumentHandler(s.v1QueryGet, v1QueryDur, v1QueryCancellations)) - s.registerHandler(router, 1, "/query", http.MethodPost, instrumentHandler(s.v1QueryPost, v1QueryDur, v1QueryCancellations)) - s.registerHandler(router, 1, "/compile", http.MethodPost, instrumentHandler(s.v1CompilePost, v1CompileDur, v1CompileCancellations)) - router.HandleFunc("/", instrumentHandler(s.unversionedPost, indexDur, indexCancellations)).Methods(http.MethodPost) - router.HandleFunc("/", instrumentHandler(s.indexGet, indexDur, indexCancellations)).Methods(http.MethodGet) + s.registerHandler(router, 0, "/data/{path:.+}", http.MethodPost, s.instrumentHandler(s.v0DataPost, PromHandlerV0Data)) + s.registerHandler(router, 0, "/data", http.MethodPost, s.instrumentHandler(s.v0DataPost, PromHandlerV0Data)) + s.registerHandler(router, 1, "/data/{path:.+}", http.MethodDelete, s.instrumentHandler(s.v1DataDelete, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data/{path:.+}", http.MethodPut, s.instrumentHandler(s.v1DataPut, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data", http.MethodPut, s.instrumentHandler(s.v1DataPut, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data/{path:.+}", http.MethodGet, s.instrumentHandler(s.v1DataGet, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data", http.MethodGet, s.instrumentHandler(s.v1DataGet, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data/{path:.+}", http.MethodPatch, s.instrumentHandler(s.v1DataPatch, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data", http.MethodPatch, s.instrumentHandler(s.v1DataPatch, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data/{path:.+}", http.MethodPost, s.instrumentHandler(s.v1DataPost, PromHandlerV1Data)) + s.registerHandler(router, 1, "/data", http.MethodPost, s.instrumentHandler(s.v1DataPost, PromHandlerV1Data)) + s.registerHandler(router, 1, "/policies", http.MethodGet, s.instrumentHandler(s.v1PoliciesList, PromHandlerV1Policies)) + s.registerHandler(router, 1, "/policies/{path:.+}", http.MethodDelete, s.instrumentHandler(s.v1PoliciesDelete, PromHandlerV1Policies)) + s.registerHandler(router, 1, "/policies/{path:.+}", http.MethodGet, s.instrumentHandler(s.v1PoliciesGet, PromHandlerV1Policies)) + s.registerHandler(router, 1, "/policies/{path:.+}", http.MethodPut, s.instrumentHandler(s.v1PoliciesPut, PromHandlerV1Policies)) + s.registerHandler(router, 1, "/query", http.MethodGet, s.instrumentHandler(s.v1QueryGet, PromHandlerV1Query)) + s.registerHandler(router, 1, "/query", http.MethodPost, s.instrumentHandler(s.v1QueryPost, PromHandlerV1Query)) + s.registerHandler(router, 1, "/compile", http.MethodPost, s.instrumentHandler(s.v1CompilePost, PromHandlerV1Compile)) + router.Handle("/", s.instrumentHandler(http.HandlerFunc(s.unversionedPost), PromHandlerIndex)).Methods(http.MethodPost) + router.Handle("/", s.instrumentHandler(http.HandlerFunc(s.indexGet), PromHandlerIndex)).Methods(http.MethodGet) // These are catch all handlers that respond 405 for resources that exist but the method is not allowed - router.HandleFunc("/v0/data/{path:.*}", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodGet, http.MethodHead, + router.Handle("/v0/data/{path:.*}", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodGet, http.MethodHead, http.MethodConnect, http.MethodDelete, http.MethodOptions, http.MethodPatch, http.MethodPut, http.MethodTrace) - router.HandleFunc("/v0/data", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodGet, http.MethodHead, + router.Handle("/v0/data", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodGet, http.MethodHead, http.MethodConnect, http.MethodDelete, http.MethodOptions, http.MethodPatch, http.MethodPut, http.MethodTrace) // v1 Data catch all - router.HandleFunc("/v1/data/{path:.*}", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodHead, + router.Handle("/v1/data/{path:.*}", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodHead, http.MethodConnect, http.MethodOptions, http.MethodTrace) - router.HandleFunc("/v1/data", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodHead, + router.Handle("/v1/data", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodHead, http.MethodConnect, http.MethodDelete, http.MethodOptions, http.MethodTrace) // Policies catch all - router.HandleFunc("/v1/policies", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodHead, + router.Handle("/v1/policies", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodHead, http.MethodConnect, http.MethodDelete, http.MethodOptions, http.MethodTrace, http.MethodPost, http.MethodPut, http.MethodPatch) // Policies (/policies/{path.+} catch all - router.HandleFunc("/v1/policies/{path:.*}", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodHead, + router.Handle("/v1/policies/{path:.*}", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodHead, http.MethodConnect, http.MethodOptions, http.MethodTrace, http.MethodPost) // Query catch all - router.HandleFunc("/v1/query/{path:.*}", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodHead, + router.Handle("/v1/query/{path:.*}", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodHead, http.MethodConnect, http.MethodDelete, http.MethodOptions, http.MethodTrace, http.MethodPost, http.MethodPut, http.MethodPatch) - router.HandleFunc("/v1/query", instrumentHandler(writer.HTTPStatus(405), catchAllDur, catchAllCancellations)).Methods(http.MethodHead, + router.Handle("/v1/query", s.instrumentHandler(writer.HTTPStatus(405), PromHandlerCatch)).Methods(http.MethodHead, http.MethodConnect, http.MethodDelete, http.MethodOptions, http.MethodTrace, http.MethodPut, http.MethodPatch) s.Handler = router } +func (s *Server) instrumentHandler(handler func(http.ResponseWriter, *http.Request), label string) http.Handler { + if s.globalMetrics != nil { + return s.globalMetrics.InstrumentHandler(http.HandlerFunc(handler), label) + } + return http.HandlerFunc(handler) +} + func (s *Server) execQuery(ctx context.Context, r *http.Request, txn storage.Transaction, decisionID string, parsedQuery ast.Body, input ast.Value, m metrics.Metrics, explainMode types.ExplainModeV1, includeMetrics, includeInstrumentation, pretty bool) (results types.QueryResponseV1, err error) { logger := s.getDecisionLogger() @@ -723,9 +701,9 @@ func (s *Server) indexGet(w http.ResponseWriter, r *http.Request) { renderQueryResult(w, results, err, t0) } -func (s *Server) registerHandler(router *mux.Router, version int, path string, method string, h func(http.ResponseWriter, *http.Request)) { +func (s *Server) registerHandler(router *mux.Router, version int, path string, method string, h http.Handler) { prefix := fmt.Sprintf("/v%d", version) - router.HandleFunc(prefix+path, h).Methods(method) + router.Handle(prefix+path, h).Methods(method) } func (s *Server) reload(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { @@ -2077,41 +2055,6 @@ func (s *Server) hasLegacyBundle() bool { return s.legacyRevision != "" || (bp != nil && !bp.Config().IsMultiBundle()) } -type captureStatusResponseWriter struct { - http.ResponseWriter - status int -} - -type hijacker struct { - http.ResponseWriter - hijacker http.Hijacker -} - -func (h *hijacker) Hijack() (net.Conn, *bufio.ReadWriter, error) { - return h.hijacker.Hijack() -} - -func (c *captureStatusResponseWriter) WriteHeader(statusCode int) { - c.ResponseWriter.WriteHeader(statusCode) - c.status = statusCode -} - -func instrumentHandler(handler func(http.ResponseWriter, *http.Request), durationCollector prometheus.ObserverVec, cancellationsCollector *prometheus.CounterVec) http.HandlerFunc { - return promhttp.InstrumentHandlerDuration(durationCollector, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - csrw := &captureStatusResponseWriter{ResponseWriter: w, status: http.StatusOK} - var rw http.ResponseWriter - if h, ok := w.(http.Hijacker); ok { - rw = &hijacker{ResponseWriter: csrw, hijacker: h} - } else { - rw = csrw - } - handler(rw, r) - if r.Context().Err() != nil { - cancellationsCollector.With(prometheus.Labels{"code": strconv.Itoa(csrw.status), "method": r.Method}).Inc() - } - })) -} - // parsePatchPathEscaped returns a new path for the given escaped str. // This is based on storage.ParsePathEscaped so will do URL unescaping of // the provided str for backwards compatibility, but also handles the diff --git a/server/server_test.go b/server/server_test.go index e83914ba13..8992705760 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -27,6 +27,7 @@ import ( "github.com/open-policy-agent/opa/ast" "github.com/open-policy-agent/opa/bundle" + imetrics "github.com/open-policy-agent/opa/internal/metrics" "github.com/open-policy-agent/opa/metrics" "github.com/open-policy-agent/opa/plugins" pluginBundle "github.com/open-policy-agent/opa/plugins/bundle" @@ -2550,8 +2551,13 @@ func TestQueryWatchMigrateInvalidate(t *testing.T) { } func TestMetricsEndpoint(t *testing.T) { - - f := newFixture(t) + f := newFixture(t, func(s *Server) { + gm, err := imetrics.NewGlobalMetrics("prometheus", nil) + if err != nil { + t.Fatal(err) + } + s.WithMetrics(gm) + }) module := `package test @@ -3350,7 +3356,7 @@ type fixture struct { t *testing.T } -func newFixture(t *testing.T) *fixture { +func newFixture(t *testing.T, opts ...func(*Server)) *fixture { ctx := context.Background() store := inmem.New() m, err := plugins.New([]byte{}, "test", store) @@ -3362,11 +3368,14 @@ func newFixture(t *testing.T) *fixture { panic(err) } - server, err := New(). + server := New(). WithAddresses([]string{":8182"}). WithStore(store). - WithManager(m). - Init(ctx) + WithManager(m) + for _, opt := range opts { + opt(server) + } + server, err = server.Init(ctx) if err != nil { panic(err) } diff --git a/server/writer/writer.go b/server/writer/writer.go index 54413b09e6..916e06b6f9 100644 --- a/server/writer/writer.go +++ b/server/writer/writer.go @@ -16,7 +16,7 @@ import ( // HTTPStatus is used to set a specific status code // Adapted from https://stackoverflow.com/questions/27711154/what-response-code-to-return-on-a-non-supported-http-method-on-rest -func HTTPStatus(code int) func(w http.ResponseWriter, req *http.Request) { +func HTTPStatus(code int) http.HandlerFunc { return func(w http.ResponseWriter, req *http.Request) { w.WriteHeader(code) }