ewilliams28/releases
1
0
Fork 0
mirror of https://github.com/open-policy-agent/opa.git synced 2026-08-12 19:32:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

Refactor discovery implementation

Browse Source 
These changes refactor the discovery implementation a bit to improve
test coverage and remove duplication of common logic shared with the
bundle plugin.

Specifically, the downloading logic has been moved into a separate
package that is shared by bundle and discovery. Second, test coverage in
the discovery implementation is increased from ~15% to ~85%.

These changes also include a few functional improvements:

- The default decision paths can be updated dynamically
- The decision logger can be enabled dynamically
- Discovery downloading errors are reported in status updates
- Discovery bundle is evaluated with all runtime params
- Custom plugins can be created dynamically
- Status updates include both discovery and bundle status

Signed-off-by: Torin Sandall <torinsandall@gmail.com>
This commit is contained in:
Torin Sandall
2018-11-29 12:42:41 -08:00
parent a73e3bacc8
commit 2d425494aa
30 changed files with 2247 additions and 2527 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/plugins_test.go
+4 -5
View File
@@ -11,14 +11,13 @@ import (
"fmt"
"net/http"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
"reflect"
"strings"
"testing"
"os"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/runtime"
"github.com/open-policy-agent/opa/types"
@@ -156,7 +155,7 @@ func (t *Tester) Stop(ctx context.Context) {
return
}
func (t *Tester) Reconfigure(config interface{}) {
func (t *Tester) Reconfigure(ctx context.Context, config interface{}) {
return
}
@@ -230,7 +229,7 @@ func TestRegisterPlugin(t *testing.T) {
// make sure starting the manager kicks the plugin in
emptyInitChan()
if err := rt.Discovery.Start(context.Background()); err != nil {
if err := rt.Manager.Start(context.Background()); err != nil {
t.Fatalf("Unable to initialize plugins: %v", err.Error())
}
@@ -257,7 +256,7 @@ func TestPluginDoesNotStartWithoutConfig(t *testing.T) {
// make sure starting the manager kicks the plugin in
emptyInitChan()
if err := rt.Discovery.Start(context.Background()); err != nil {
if err := rt.Manager.Start(context.Background()); err != nil {
t.Fatalf("Unable to initialize plugins: %v", err.Error())
}
if len(initChan) != 0 {
config/config.go
+96
View File
@@ -0,0 +1,96 @@
// 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 config implements OPA configuration file parsing and validation.
package config
import (
"encoding/json"
"strings"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/util"
)
// Config represents the configuration file that OPA can be started with.
type Config struct {
Services json.RawMessage `json:"services"`
Labels map[string]string `json:"labels"`
Discovery json.RawMessage `json:"discovery"`
Bundle json.RawMessage `json:"bundle"`
DecisionLogs json.RawMessage `json:"decision_logs"`
Status json.RawMessage `json:"status"`
Plugins map[string]json.RawMessage `json:"plugins"`
DefaultDecision *string `json:"default_decision"`
DefaultAuthorizationDecision *string `json:"default_authorization_decision"`
}
// ParseConfig returns a valid Config object with defaults injected. The id
// parameter will be set in the labels map.
func ParseConfig(raw []byte, id string) (*Config, error) {
var result Config
if err := util.Unmarshal(raw, &result); err != nil {
return nil, err
}
return &result, result.validateAndInjectDefaults(id)
}
// PluginsEnabled returns true if one or more plugin features are enabled.
func (c Config) PluginsEnabled() bool {
return c.Bundle != nil || c.DecisionLogs != nil || c.Status != nil || len(c.Plugins) > 0
}
// DefaultDecisionRef returns the default decision as a reference.
func (c Config) DefaultDecisionRef() ast.Ref {
ref, _ := parsePathToRef(*c.DefaultDecision)
return ref
}
// DefaultAuthorizationDecisionRef returns the default authorization decision
// as a reference.
func (c Config) DefaultAuthorizationDecisionRef() ast.Ref {
ref, _ := parsePathToRef(*c.DefaultAuthorizationDecision)
return ref
}
func (c *Config) validateAndInjectDefaults(id string) error {
if c.DefaultDecision == nil {
s := defaultDecisionPath
c.DefaultDecision = &s
}
_, err := parsePathToRef(*c.DefaultDecision)
if err != nil {
return err
}
if c.DefaultAuthorizationDecision == nil {
s := defaultAuthorizationDecisionPath
c.DefaultAuthorizationDecision = &s
}
_, err = parsePathToRef(*c.DefaultAuthorizationDecision)
if err != nil {
return err
}
if c.Labels == nil {
c.Labels = map[string]string{}
}
c.Labels["id"] = id
return nil
}
func parsePathToRef(s string) (ast.Ref, error) {
s = strings.Replace(strings.Trim(s, "/"), "/", ".", -1)
return ast.ParseRef("data." + s)
}
const (
defaultDecisionPath = "/system/main"
defaultAuthorizationDecisionPath = "/system/authz/allow"
)
discovery/discovery.go
-867
View File
@@ -1,867 +0,0 @@
// 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 discovery
import (
"context"
"encoding/json"
"fmt"
"io/ioutil"
"math/rand"
"net/http"
"strings"
"sync"
"time"
"github.com/open-policy-agent/opa/ast"
bundleApi "github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/internal/runtime"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/bundle"
"github.com/open-policy-agent/opa/plugins/logs"
"github.com/open-policy-agent/opa/plugins/status"
"github.com/open-policy-agent/opa/rego"
"github.com/open-policy-agent/opa/server/types"
"github.com/open-policy-agent/opa/storage"
"github.com/open-policy-agent/opa/storage/inmem"
"github.com/open-policy-agent/opa/util"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
type bundlePluginListener string
// Discovery performs periodic configuration discovery and configures plugins.
type Discovery struct {
Manager *plugins.Manager
Plugins map[string]plugins.Plugin
DefaultDecision ast.Ref
DefaultAuthorizationDecision ast.Ref
dynamic bool
config *discoveryPathConfig
stop chan chan struct{}
pluginsMux sync.Mutex
status *bundle.Status
}
// Params contains the input needed for creating an instance of
// the discovery plugin.
type Params struct {
ID string
ConfigFile string
Store storage.Store
RegisteredPlugins map[string]plugins.PluginInitFunc
}
// PollingConfig represents configuration for discovery's polling behaviour.
type PollingConfig struct {
MinDelaySeconds *int64 `json:"min_delay_seconds,omitempty"` // min amount of time to wait between successful poll attempts
MaxDelaySeconds *int64 `json:"max_delay_seconds,omitempty"` // max amount of time to wait between poll attempts
}
type rawConfig struct {
DefaultDecision string `json:"default_decision"`
DefaultAuthorizationDecision string `json:"default_authorization_decision"`
}
type discoveryConfig struct {
Discovery json.RawMessage `json:"discovery"`
}
type discoveryPathConfig struct {
Name *string `json:"name"`
Prefix *string `json:"prefix"`
Polling PollingConfig `json:"polling"`
urlPath string
service string
}
func parsePathToRef(s string) (ast.Ref, error) {
s = strings.Replace(strings.Trim(s, "/"), "/", ".", -1)
return ast.ParseRef("data." + s)
}
const (
defaultDecisionPath = "/system/main"
defaultAuthorizationDecisionPath = "/system/authz/allow"
defaultDiscoveryPathPrefix = "bundles"
defaultDiscoveryQueryPrefix = "data"
// min amount of time to wait following a failure
minRetryDelay = time.Millisecond * 100
defaultMinDelaySeconds = int64(60)
defaultMaxDelaySeconds = int64(120)
errCode = "discovery_error"
)
// New takes the provided configuration and instantiates the plugins. If
// discovery is enabled, it validates the discovery configuration.
func New(ctx context.Context, params Params) (*Discovery, error) {
var bs []byte
var err error
if params.ConfigFile != "" {
bs, err = ioutil.ReadFile(params.ConfigFile)
if err != nil {
return nil, err
}
}
m, err := plugins.New(bs, params.ID, params.Store)
if err != nil {
return nil, err
}
var config *discoveryConfig
var discoveryEnabled bool
var discPathConfig *discoveryPathConfig
config, discoveryEnabled = isDiscoveryEnabled(bs)
if discoveryEnabled {
defined, err := pluginsDefinedOnBoot(bs)
if err != nil {
return nil, err
}
if defined {
return nil, fmt.Errorf("plugins cannot be specified in the bootstrap configuration when discovery enabled")
}
discPathConfig, err = validateAndInjectDefaults(config, m.Services())
if err != nil {
return nil, err
}
}
p, err := initPlugins(m, bs, params.RegisteredPlugins)
if err != nil {
return nil, err
}
defaultDecision, defaultAuthorizationDecision, err := getDefaultDecisionRefs(bs)
if err != nil {
return nil, err
}
c := &Discovery{
config: discPathConfig,
dynamic: discoveryEnabled,
Manager: m,
Plugins: p,
DefaultDecision: defaultDecision,
DefaultAuthorizationDecision: defaultAuthorizationDecision,
}
if discoveryEnabled {
c.status = &bundle.Status{
Name: *discPathConfig.Name,
}
}
return c, nil
}
// Start starts the plugin manager and periodic configuration discovery.
func (c *Discovery) Start(ctx context.Context) error {
if err := c.Manager.Start(ctx); err != nil {
return err
}
if c.dynamic {
go c.loop()
}
return nil
}
// Stop stops the plugin manager and periodic configuration discovery.
func (c *Discovery) Stop(ctx context.Context) {
c.Manager.Stop(ctx)
if c.dynamic {
done := make(chan struct{})
c.stop <- done
_ = <-done
}
}
func (c *Discovery) loop() {
ctx, cancel := context.WithCancel(context.Background())
var retry int
for {
bs, updated, err := discoveryHandler(ctx, c.config, c.Manager)
if err != nil {
c.logError("%v.", err)
} else if !updated {
c.logDebug("Configuration download skipped, server replied with not modified.")
} else {
c.logInfo("New configuration successfully downloaded. Now updating plugins.")
if bs != nil {
c.Manager.Update(bs)
}
err = c.validateAndConfigurePlugins(ctx, bs)
if err != nil {
c.logError("%v.", err)
}
}
c.status.LastSuccessfulDownload = time.Now().UTC()
var delay time.Duration
if err == nil {
min := float64(*c.config.Polling.MinDelaySeconds)
max := float64(*c.config.Polling.MaxDelaySeconds)
delay = time.Duration(((max - min) * rand.Float64()) + min)
} else {
delay = util.DefaultBackoff(float64(minRetryDelay), float64(*c.config.Polling.MaxDelaySeconds), retry)
}
c.logDebug("Waiting %v before next download/retry.", delay)
timer := time.NewTimer(delay)
select {
case <-timer.C:
if err != nil {
retry++
} else {
retry = 0
}
case done := <-c.stop:
cancel()
done <- struct{}{}
return
}
}
}
func validateBundlePluginConfig(m *plugins.Manager, bs []byte) (*bundle.Config, error) {
var config struct {
Bundle json.RawMessage `json:"bundle"`
}
if err := util.Unmarshal(bs, &config); err != nil {
return nil, err
}
if config.Bundle == nil {
return nil, nil
}
return bundle.ParseConfig(config.Bundle, m.Services())
}
func validateDecisionLogsPluginConfig(m *plugins.Manager, bs []byte) (*logs.Config, error) {
var config struct {
DecisionLogs json.RawMessage `json:"decision_logs"`
}
if err := util.Unmarshal(bs, &config); err != nil {
return nil, err
}
if config.DecisionLogs == nil {
return nil, nil
}
return logs.ParseConfig(config.DecisionLogs, m.Services())
}
func validateStatusPluginConfig(m *plugins.Manager, bs []byte) (*status.Config, error) {
var config struct {
Status json.RawMessage `json:"status"`
}
if err := util.Unmarshal(bs, &config); err != nil {
return nil, err
}
if config.Status == nil {
return nil, nil
}
return status.ParseConfig(config.Status, m.Services())
}
func (c *Discovery) validateAndConfigurePlugins(ctx context.Context, bs []byte) (err error) {
defer func() {
c.setErrorStatus(err)
if plugin, ok := c.Plugins["status"]; ok {
plugin.(*status.Plugin).UpdateDiscoveryStatus(*c.status)
}
}()
bundleConfig, err := validateBundlePluginConfig(c.Manager, bs)
if err != nil {
return err
}
decisionLogsConfig, err := validateDecisionLogsPluginConfig(c.Manager, bs)
if err != nil {
return err
}
statusConfig, err := validateStatusPluginConfig(c.Manager, bs)
if err != nil {
return err
}
if bundleConfig != nil {
err := c.configureBundlePlugin(ctx, bundleConfig)
if err != nil {
return err
}
}
if decisionLogsConfig != nil {
err = c.configureDecisionLogsPlugin(ctx, decisionLogsConfig)
if err != nil {
return err
}
}
if statusConfig != nil {
err = c.configureStatusPlugin(ctx, statusConfig)
if err != nil {
return err
}
}
err = c.configureRegisteredPlugins(bs)
if err != nil {
return err
}
return nil
}
func (c *Discovery) configureBundlePlugin(ctx context.Context, config *bundle.Config) error {
plugin, ok := c.Plugins["bundle"]
if !ok {
bundlePlugin, err := createBundlePlugin(c.Manager, config)
if err != nil {
return err
} else if bundlePlugin != nil {
c.pluginsMux.Lock()
c.Plugins["bundle"] = bundlePlugin
c.pluginsMux.Unlock()
err = bundlePlugin.Start(ctx)
if err != nil {
return err
}
c.logInfo("Bundle plugin configured successfully.")
}
} else {
if plugin.(*bundle.Plugin).Equal(config) {
c.logDebug("No updated configuration for bundle plugin.")
} else {
plugin.(*bundle.Plugin).Reconfigure(config)
c.logInfo("Bundle plugin reconfigured successfully.")
}
}
return nil
}
func (c *Discovery) configureDecisionLogsPlugin(ctx context.Context, config *logs.Config) error {
plugin, ok := c.Plugins["decision_logs"]
if !ok {
decisionLogsPlugin, err := createDecisionLogsPlugin(c.Manager, config)
if err != nil {
return err
} else if decisionLogsPlugin != nil {
c.pluginsMux.Lock()
c.Plugins["decision_logs"] = decisionLogsPlugin
c.pluginsMux.Unlock()
err = decisionLogsPlugin.Start(ctx)
if err != nil {
return err
}
c.logInfo("Decision logs plugin configured successfully.")
}
} else {
if plugin.(*logs.Plugin).Equal(config) {
c.logDebug("No updated configuration for decision logs plugin.")
} else {
plugin.(*logs.Plugin).Reconfigure(config)
c.logInfo("Decision logs plugin reconfigured successfully.")
}
}
return nil
}
func (c *Discovery) configureStatusPlugin(ctx context.Context, config *status.Config) error {
plugin, ok := c.Plugins["status"]
if !ok {
bundlePlugin := c.Plugins["bundle"]
if bundlePlugin != nil {
statusPlugin, err := createStatusPlugin(c.Manager, config, bundlePlugin.(*bundle.Plugin))
if err != nil {
return err
} else if statusPlugin != nil {
c.pluginsMux.Lock()
c.Plugins["status"] = statusPlugin
c.pluginsMux.Unlock()
err = statusPlugin.Start(ctx)
if err != nil {
return err
}
c.logInfo("Status plugin configured successfully.")
}
}
} else {
if plugin.(*status.Plugin).Equal(config) {
c.logDebug("No updated configuration for status plugin.")
} else {
plugin.(*status.Plugin).Reconfigure(config)
c.logInfo("Status plugin reconfigured successfully.")
}
}
return nil
}
func (c *Discovery) configureRegisteredPlugins(bs []byte) error {
var config struct {
Plugins map[string]json.RawMessage `json:"plugins"`
}
if err := util.Unmarshal(bs, &config); err != nil {
return err
}
for name, plugin := range c.Plugins {
pc, ok := config.Plugins[name]
if !ok {
continue
}
plugin.Reconfigure(pc)
}
return nil
}
func initPlugins(m *plugins.Manager, bs []byte, registeredPlugins map[string]plugins.PluginInitFunc) (map[string]plugins.Plugin, error) {
plugins := map[string]plugins.Plugin{}
bundlePlugin, err := initBundlePlugin(m, bs)
if err != nil {
return nil, err
} else if bundlePlugin != nil {
plugins["bundle"] = bundlePlugin
}
if bundlePlugin != nil {
statusPlugin, err := initStatusPlugin(m, bs, bundlePlugin)
if err != nil {
return nil, err
} else if statusPlugin != nil {
plugins["status"] = statusPlugin
}
}
decisionLogsPlugin, err := initDecisionLogsPlugin(m, bs)
if err != nil {
return nil, err
} else if decisionLogsPlugin != nil {
plugins["decision_logs"] = decisionLogsPlugin
}
err = initRegisteredPlugins(m, bs, registeredPlugins, plugins)
if err != nil {
return nil, err
}
return plugins, nil
}
func initBundlePlugin(m *plugins.Manager, bs []byte) (*bundle.Plugin, error) {
bundleConfig, err := validateBundlePluginConfig(m, bs)
if err != nil {
return nil, err
}
if bundleConfig == nil {
return nil, nil
}
return createBundlePlugin(m, bundleConfig)
}
func initDecisionLogsPlugin(m *plugins.Manager, bs []byte) (*logs.Plugin, error) {
decisionLogsConfig, err := validateDecisionLogsPluginConfig(m, bs)
if err != nil {
return nil, err
}
if decisionLogsConfig == nil {
return nil, nil
}
return createDecisionLogsPlugin(m, decisionLogsConfig)
}
func initStatusPlugin(m *plugins.Manager, bs []byte, bundlePlugin *bundle.Plugin) (*status.Plugin, error) {
statusConfig, err := validateStatusPluginConfig(m, bs)
if err != nil {
return nil, err
}
if statusConfig == nil {
return nil, nil
}
return createStatusPlugin(m, statusConfig, bundlePlugin)
}
func initRegisteredPlugins(m *plugins.Manager, bs []byte, registeredPlugins map[string]plugins.PluginInitFunc, allPlugins map[string]plugins.Plugin) error {
var config struct {
Plugins map[string]json.RawMessage `json:"plugins"`
}
if err := util.Unmarshal(bs, &config); err != nil {
return err
}
for name, factory := range registeredPlugins {
pc, ok := config.Plugins[name]
if !ok {
continue
}
plugin, err := factory(m, pc)
if err != nil {
return err
}
m.Register(plugin)
allPlugins[name] = plugin
}
return nil
}
func createBundlePlugin(m *plugins.Manager, config *bundle.Config) (*bundle.Plugin, error) {
p, err := bundle.New(config, m)
if err != nil {
return nil, err
}
m.Register(p)
return p, nil
}
func createDecisionLogsPlugin(m *plugins.Manager, config *logs.Config) (*logs.Plugin, error) {
p, err := logs.New(config, m)
if err != nil {
return nil, err
}
m.Register(p)
return p, nil
}
func createStatusPlugin(m *plugins.Manager, config *status.Config, bundlePlugin *bundle.Plugin) (*status.Plugin, error) {
p, err := status.New(config, m)
if err != nil {
return nil, err
}
m.Register(p)
bundlePlugin.Register(bundlePluginListener("status-plugin"), func(s bundle.Status) {
p.UpdateBundleStatus(s)
})
return p, nil
}
func validateAndInjectDefaults(config *discoveryConfig, services []string) (*discoveryPathConfig, error) {
var discoveryConfig discoveryPathConfig
if err := util.Unmarshal(config.Discovery, &discoveryConfig); err != nil {
return nil, err
}
if len(services) == 0 {
return nil, fmt.Errorf("endpoint to fetch discovery configuration from not provided")
}
if discoveryConfig.Name == nil {
return nil, fmt.Errorf("discovery plugin name not provided")
}
discoveryConfig.service = services[0]
discoveryConfig.urlPath = getDiscoveryServicePath(discoveryConfig)
min := defaultMinDelaySeconds
max := defaultMaxDelaySeconds
// reject bad min/max values
if discoveryConfig.Polling.MaxDelaySeconds != nil && discoveryConfig.Polling.MinDelaySeconds != nil {
if *discoveryConfig.Polling.MaxDelaySeconds < *discoveryConfig.Polling.MinDelaySeconds {
return nil, fmt.Errorf("max polling delay must be >= min polling delay in discovery configuration")
}
min = *discoveryConfig.Polling.MinDelaySeconds
max = *discoveryConfig.Polling.MaxDelaySeconds
} else if discoveryConfig.Polling.MaxDelaySeconds == nil && discoveryConfig.Polling.MinDelaySeconds != nil {
return nil, fmt.Errorf("polling configuration missing 'max_delay_seconds' in discovery configuration")
} else if discoveryConfig.Polling.MinDelaySeconds == nil && discoveryConfig.Polling.MaxDelaySeconds != nil {
return nil, fmt.Errorf("polling configuration missing 'min_delay_seconds' in discovery configuration")
}
// scale to seconds
minSeconds := int64(time.Duration(min) * time.Second)
discoveryConfig.Polling.MinDelaySeconds = &minSeconds
maxSeconds := int64(time.Duration(max) * time.Second)
discoveryConfig.Polling.MaxDelaySeconds = &maxSeconds
return &discoveryConfig, nil
}
func getDefaultDecisionRefs(bs []byte) (ast.Ref, ast.Ref, error) {
var raw rawConfig
if err := util.Unmarshal(bs, &raw); err != nil {
return nil, nil, err
}
if raw.DefaultDecision == "" {
raw.DefaultDecision = defaultDecisionPath
}
if raw.DefaultAuthorizationDecision == "" {
raw.DefaultAuthorizationDecision = defaultAuthorizationDecisionPath
}
defaultDecision, err := parsePathToRef(raw.DefaultDecision)
if err != nil {
return nil, nil, err
}
defaultAuthorizationDecision, err := parsePathToRef(raw.DefaultAuthorizationDecision)
if err != nil {
return nil, nil, err
}
return defaultDecision, defaultAuthorizationDecision, nil
}
func getDiscoveryConfig(bs []byte) (*discoveryConfig, error) {
var config discoveryConfig
if err := util.Unmarshal(bs, &config); err != nil {
return nil, err
}
return &config, nil
}
func isDiscoveryEnabled(bs []byte) (*discoveryConfig, bool) {
config, err := getDiscoveryConfig(bs)
if err != nil {
return nil, false
}
if config.Discovery == nil {
return nil, false
}
return config, true
}
func pluginsDefinedOnBoot(bs []byte) (bool, error) {
var config struct {
Bundle json.RawMessage `json:"bundle"`
DecisionLogs json.RawMessage `json:"decision_logs"`
Status json.RawMessage `json:"status"`
Plugins map[string]json.RawMessage `json:"plugins"`
}
if err := util.Unmarshal(bs, &config); err != nil {
return false, err
}
if config.Bundle != nil || config.DecisionLogs != nil || config.Status != nil || config.Plugins != nil {
return true, nil
}
return false, nil
}
func discoveryHandler(ctx context.Context, discoveryConfig *discoveryPathConfig, manager *plugins.Manager) ([]byte, bool, error) {
resp, err := manager.Client(discoveryConfig.service).
Do(ctx, "GET", discoveryConfig.urlPath)
if err != nil {
return nil, false, errors.Wrap(err, "Download request failed")
}
defer util.Close(resp)
switch resp.StatusCode {
case http.StatusOK:
return process(ctx, resp, discoveryConfig.Name)
case http.StatusNotModified:
return nil, false, nil
case http.StatusNotFound:
return nil, false, fmt.Errorf("Discovery configuration download failed, server replied with not found")
case http.StatusUnauthorized:
return nil, false, fmt.Errorf("Discovery configuration download failed, server replied with not authorized")
default:
return nil, false, fmt.Errorf("Discovery configuration download failed, server replied with HTTP %v", resp.StatusCode)
}
}
func getDiscoveryServicePath(config discoveryPathConfig) string {
prefix := defaultDiscoveryPathPrefix
path := ""
if config.Prefix != nil {
prefix = *config.Prefix
}
if config.Name != nil {
path = *config.Name
}
return fmt.Sprintf("%v/%v", strings.Trim(prefix, "/"), strings.Trim(path, "/"))
}
func process(ctx context.Context, resp *http.Response, path *string) ([]byte, bool, error) {
br := bundleApi.NewReader(resp.Body)
b, err := br.Read()
if err != nil {
return nil, false, err
}
query := defaultDiscoveryQueryPrefix
if path != nil && *path != "" {
*path = strings.Trim(*path, "/")
query = fmt.Sprintf("%v.%v", query, strings.Replace(*path, "/", ".", -1))
}
return processBundle(ctx, b, query)
}
func processBundle(ctx context.Context, b bundleApi.Bundle, query string) ([]byte, bool, error) {
modules := map[string]*ast.Module{}
for _, file := range b.Modules {
modules[file.Path] = file.Parsed
}
compiler := ast.NewCompiler()
if compiler.Compile(modules); compiler.Failed() {
return nil, false, compiler.Errors
}
store := inmem.NewFromObject(b.Data)
info, err := runtime.Term(runtime.Params{})
if err != nil {
return nil, false, err
}
rego := rego.New(
rego.Query(query),
rego.Compiler(compiler),
rego.Store(store),
rego.Runtime(info),
)
rs, err := rego.Eval(ctx)
if err != nil {
return nil, false, err
}
if len(rs) == 0 {
return nil, false, fmt.Errorf("undefined configuration")
}
result := rs[0].Expressions[0].Value
switch result.(type) {
case map[string]interface{}:
newConfig, err := json.Marshal(result)
if err != nil {
return nil, false, err
}
return newConfig, true, nil
default:
return nil, false, fmt.Errorf("expected discovery rule to generate an object but got %T", result)
}
}
func (c *Discovery) logError(fmt string, a ...interface{}) {
logrus.WithFields(c.logrusFields()).Errorf(fmt, a...)
}
func (c *Discovery) logInfo(fmt string, a ...interface{}) {
logrus.WithFields(c.logrusFields()).Infof(fmt, a...)
}
func (c *Discovery) logDebug(fmt string, a ...interface{}) {
logrus.WithFields(c.logrusFields()).Debugf(fmt, a...)
}
func (c *Discovery) logrusFields() logrus.Fields {
return logrus.Fields{
"plugin": "discovery",
"name": *c.config.Name,
}
}
func (c *Discovery) setErrorStatus(err error) {
if err == nil {
c.status.Code = ""
c.status.Message = ""
c.status.Errors = nil
return
}
cause := errors.Cause(err)
if astErr, ok := cause.(ast.Errors); ok {
c.status.Code = errCode
c.status.Message = types.MsgPluginConfigError
c.status.Errors = make([]error, len(astErr))
for i := range astErr {
c.status.Errors[i] = astErr[i]
}
} else {
c.status.Code = errCode
c.status.Message = err.Error()
c.status.Errors = nil
}
}
discovery/discovery_test.go
-483
View File
@@ -1,483 +0,0 @@
// 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 discovery
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"github.com/open-policy-agent/opa/ast"
bundleApi "github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/bundle"
"github.com/open-policy-agent/opa/storage"
"github.com/open-policy-agent/opa/storage/inmem"
"github.com/open-policy-agent/opa/util"
)
func TestConfigDiscoveryEnabled(t *testing.T) {
config := []byte(`{
"discovery": {
"path": "/foo/bar"
}}`)
_, result := isDiscoveryEnabled(config)
if !result {
t.Fatal("Expected discovery to be enabled")
}
}
func TestConfigDiscoveryDisabled(t *testing.T) {
_, result := isDiscoveryEnabled([]byte{})
if result {
t.Fatal("Expected discovery not to be enabled")
}
}
func TestGetDiscoveryServicePath(t *testing.T) {
prefix := "/bundles"
path := "/foo/bar/"
config := discoveryPathConfig{
Prefix: &prefix,
Name: &path,
}
expected := "bundles/foo/bar"
result := getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
prefix = ""
path = "/foo/bar/"
config = discoveryPathConfig{
Prefix: &prefix,
Name: &path,
}
expected = "/foo/bar"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
prefix = "bundles/v1"
path = ""
config = discoveryPathConfig{
Prefix: &prefix,
Name: &path,
}
expected = "bundles/v1/"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
prefix = ""
path = ""
config = discoveryPathConfig{
Prefix: &prefix,
Name: &path,
}
expected = "/"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
prefix = "/bundles"
config = discoveryPathConfig{
Prefix: &prefix,
}
expected = "bundles/"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
prefix = ""
config = discoveryPathConfig{
Prefix: &prefix,
}
expected = "/"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
path = "/foo/bar/"
config = discoveryPathConfig{
Name: &path,
}
expected = "bundles/foo/bar"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
path = ""
config = discoveryPathConfig{
Name: &path,
}
expected = "bundles/"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
config = discoveryPathConfig{}
expected = "bundles/"
result = getDiscoveryServicePath(config)
if result != expected {
t.Fatalf("Expected discovery service path %v, but got %v", expected, result)
}
}
func TestConfigDiscoveryHandlerWithModule(t *testing.T) {
fixture := newTestFixtureWithModule(t)
defer fixture.server.stop()
testConfigDiscoveryHandler(t, fixture)
}
func TestConfigDiscoveryHandlerWithData(t *testing.T) {
fixture := newTestFixtureWithData(t)
defer fixture.server.stop()
testConfigDiscoveryHandler(t, fixture)
}
func TestConfigDiscoveryHandlerWithBadConfig(t *testing.T) {
fixture := newTestFixtureWithBadConfig(t)
defer fixture.server.stop()
discConfig, err := getDiscoveryConfig(fixture.managerConfig)
if err != nil {
t.Fatal("Unexpected error:", err)
}
discPathConfig, err := validateAndInjectDefaults(discConfig, fixture.manager.Services())
if err != nil {
t.Fatal("Unexpected error:", err)
}
_, _, err = discoveryHandler(context.Background(), discPathConfig, fixture.manager)
if err == nil {
t.Fatal("Expected error but got nil")
}
}
func TestConfigDiscoveryHandler404Status(t *testing.T) {
fixture := newTestFixtureWithData(t)
defer fixture.server.stop()
fixture.managerConfig = []byte(fmt.Sprintf(`{
"services": [
{
"name": "example",
"url": %q,
"credentials": {
"bearer": {
"scheme": "Bearer",
"token": "secret"
}
}
}
],
"discovery": {
"name": "/foo"
}}`, fixture.server.server.URL))
discConfig, err := getDiscoveryConfig(fixture.managerConfig)
if err != nil {
t.Fatal("Unexpected error:", err)
}
discPathConfig, err := validateAndInjectDefaults(discConfig, fixture.manager.Services())
if err != nil {
t.Fatal("Unexpected error:", err)
}
_, _, err = discoveryHandler(context.Background(), discPathConfig, fixture.manager)
if err == nil {
t.Fatal("Expected error but got nil")
}
expected := "Discovery configuration download failed, server replied with not found"
if err.Error() != expected {
t.Fatalf("Expected error: %v but got %v", expected, err.Error())
}
}
func testConfigDiscoveryHandler(t *testing.T, fixture testFixture) {
ctx := context.Background()
discConfig, err := getDiscoveryConfig(fixture.managerConfig)
if err != nil {
t.Fatal("Unexpected error:", err)
}
discPathConfig, err := validateAndInjectDefaults(discConfig, fixture.manager.Services())
if err != nil {
t.Fatal("Unexpected error:", err)
}
newConfig, _, err := discoveryHandler(ctx, discPathConfig, fixture.manager)
if err != nil {
t.Fatal("Unexpected error:", err)
}
var config struct {
Bundle json.RawMessage `json:"bundle"`
}
if err := util.Unmarshal(newConfig, &config); err != nil {
t.Fatal("Unexpected error:", err)
}
if config.Bundle == nil {
t.Fatal("Expected a bundle configuration")
}
var parsedConfig bundle.Config
if err := util.Unmarshal(config.Bundle, &parsedConfig); err != nil {
t.Fatal("Unexpected error:", err)
}
expectedBundleConfig := bundle.Config{
Name: "test/bundle1",
Service: "example",
}
if !reflect.DeepEqual(expectedBundleConfig, parsedConfig) {
t.Fatalf("Expected bundle config %v, but got %v", expectedBundleConfig, parsedConfig)
}
}
type testFixture struct {
store storage.Store
manager *plugins.Manager
server *testServer
managerConfig []byte
}
func newTestFixtureWithData(t *testing.T) testFixture {
ts := testServer{
t: t,
expAuth: "Bearer secret",
bundles: map[string]bundleApi.Bundle{
"bundles/foo/bar": {
Manifest: bundleApi.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{
"foo": map[string]interface{}{
"bar": map[string]interface{}{"bundle": map[string]interface{}{"name": "test/bundle1", "service": "example"}},
"baz": "qux",
},
},
},
},
}
ts.start()
return getFixture(t, ts)
}
func newTestFixtureWithModule(t *testing.T) testFixture {
sampleModule := `
package foo
bar = {
"bundle": {
"name": "test/bundle1",
"service": "example"
}
}
`
ts := testServer{
t: t,
expAuth: "Bearer secret",
bundles: map[string]bundleApi.Bundle{
"bundles/foo/bar": {
Manifest: bundleApi.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{
"baz": "qux",
},
Modules: []bundleApi.ModuleFile{
{
Path: `/example.rego`,
Raw: []byte(sampleModule),
Parsed: ast.MustParseModule(sampleModule),
},
},
},
},
}
ts.start()
return getFixture(t, ts)
}
func newTestFixtureWithBadConfig(t *testing.T) testFixture {
sampleModule := `
package foo
bar = [{
"bundle": {
"name": "test/bundle1",
"service": "example"
}
}]
`
ts := testServer{
t: t,
expAuth: "Bearer secret",
bundles: map[string]bundleApi.Bundle{
"bundles/foo/bar": {
Manifest: bundleApi.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{
"baz": "qux",
},
Modules: []bundleApi.ModuleFile{
{
Path: `/example.rego`,
Raw: []byte(sampleModule),
Parsed: ast.MustParseModule(sampleModule),
},
},
},
},
}
ts.start()
return getFixture(t, ts)
}
func getFixture(t *testing.T, ts testServer) testFixture {
managerConfig := []byte(fmt.Sprintf(`{
"services": [
{
"name": "example",
"url": %q,
"credentials": {
"bearer": {
"scheme": "Bearer",
"token": "secret"
}
}
}
],
"discovery": {
"name": "/foo/bar"
}}`, ts.server.URL))
store := inmem.New()
manager, err := plugins.New(managerConfig, "test-instance-id", store)
if err != nil {
t.Fatal(err)
}
return testFixture{
store: store,
manager: manager,
server: &ts,
managerConfig: managerConfig,
}
}
type testServer struct {
t *testing.T
expCode int
expEtag string
expAuth string
bundles map[string]bundleApi.Bundle
server *httptest.Server
}
func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
if t.expCode != 0 {
w.WriteHeader(t.expCode)
return
}
if t.expAuth != "" {
if r.Header.Get("Authorization") != t.expAuth {
w.WriteHeader(401)
return
}
}
name := strings.TrimPrefix(r.URL.Path, "/")
b, ok := t.bundles[name]
if !ok {
w.WriteHeader(404)
return
}
if t.expEtag != "" {
etag := r.Header.Get("If-None-Match")
if etag == t.expEtag {
w.WriteHeader(304)
return
}
}
w.Header().Add("Content-Type", "application/gzip")
if t.expEtag != "" {
w.Header().Add("Etag", t.expEtag)
}
w.WriteHeader(200)
var buf bytes.Buffer
if err := bundleApi.Write(&buf, b); err != nil {
w.WriteHeader(500)
}
if _, err := w.Write(buf.Bytes()); err != nil {
panic(err)
}
}
func (t *testServer) start() {
t.server = httptest.NewServer(http.HandlerFunc(t.handle))
}
func (t *testServer) stop() {
t.server.Close()
}
docs/book/discovery.md
+34 -8
View File
@@ -1,17 +1,34 @@
# Discovery
OPA can be configured to download bundles of policy and data, report status, and upload decision logs to remote endpoints. The discovery feature helps you centrally manage the OPA configuration for these features. You should use the discovery feature if you want to avoid managing OPA configuration updates in number of different locations.
OPA can be configured to download bundles of policy and data, report status, and
upload decision logs to remote endpoints. The discovery feature helps you
centrally manage the OPA configuration for these features. You should use the
discovery feature if you want to avoid managing OPA configuration updates in
number of different locations.
When the discovery feature is enabled, OPA will periodically download a *discovery bundle*. Like regular bundles, the discovery bundle may contain JSON and Rego files. OPA will evaluate the data and policies contained in the discovery bundle to generate the rest of the configuration. There are two main ways to structure the discovery bundle:
When the discovery feature is enabled, OPA will periodically download a
*discovery bundle*. Like regular bundles, the discovery bundle may contain JSON
and Rego files. OPA will evaluate the data and policies contained in the
discovery bundle to generate the rest of the configuration. There are two main
ways to structure the discovery bundle:
1. Include static JSON configuration files that define the OPA configuration.
2. Include Rego files that can be evaluated to produce the OPA configuration.
> If you need OPA to select which policy to download dynamically (e.g., based on environment variables like the region where OPA is running), use the second option.
> If you need OPA to select which policy to download dynamically (e.g., based on
> environment variables like the region where OPA is running), use the second
> option.
If no discovery path is specified OPA will query the *data* document to produce the configuration. If a discovery path is specified, OPA will translate the path to a reference and evaluate it relative to the *data* document. For example, if the discovery path is */example/discovery* OPA will evaluate *data.example.discovery* to produce the configuration.
If no discovery path is specified OPA will query the *data* document to produce
the configuration. If a discovery path is specified, OPA will translate the path
to a reference and evaluate it relative to the *data* document. For example, if
the discovery path is */example/discovery* OPA will evaluate
*data.example.discovery* to produce the configuration.
If discovery is enabled, other features like bundle downloading and status reporting **cannot** be configured manually.
If discovery is enabled, other features like bundle downloading and status
reporting **cannot** be configured manually. Similarly, discovered configuration
cannot override the original discovery settings in the configuration file that
OPA was booted with.
See the [Configuration Reference](configuration.md) for configuration details.
@@ -46,9 +63,15 @@ discovery:
prefix: configuration
```
OPA will fetch it's configuration from `https://example.com/control-plane-api/v1/configuration/example/discovery` and use that to initialize the other plugins like `bundles`, `status`, `decision logs`. The `prefix` field is optional and by default set to `bundles`. Hence if `prefix` is not provided, OPA will fetch it's configuration from `https://example.com/control-plane-api/v1/bundles/example/discovery`.
OPA will fetch it's configuration from
`https://example.com/control-plane-api/v1/configuration/example/discovery` and
use that to initialize the other plugins like `bundles`, `status`, `decision
logs`. The `prefix` field is optional and by default set to `bundles`. Hence if
`prefix` is not provided, OPA will fetch it's configuration from
`https://example.com/control-plane-api/v1/bundles/example/discovery`.
Below is an example of how configuration for `decision logs` can be included inside a policy file.
Below is an example of how configuration for `decision logs` can be included
inside a policy file.
**example.rego**
@@ -76,4 +99,7 @@ The same configuration can also be provided as data.
}
```
In both cases, OPA's configuration is hierarchically organized under the `discovery.name` value. If discovery is enabled, the `service` field in the `bundles`, `status`, `decision logs` plugins is optional and will default to one of the services from the discovery configuration.
In both cases, OPA's configuration is hierarchically organized under the
`discovery.name` value. If discovery is enabled, the `service` field in the
`bundles`, `status`, `decision logs` plugins is optional and will default to one
of the services from the discovery configuration.
docs/book/status.md
+14 -1
View File
@@ -52,6 +52,10 @@ Status updates contain the following fields:
| `bundle.active_revision` | `string` | Opaque revision identifier of the last successful activation. |
| `bundle.last_successful_download` | `string` | RFC3339 timestamp of last successful bundle download. |
| `bundle.last_successful_activation` | `string` | RFC3339 timestamp of last successful bundle activation. |
| `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_successful_download` | `string` | RFC3339 timestamp of last successful discovery bundle download. |
| `discovery.last_successful_activation` | `string` | RFC3339 timestamp of last successful discovery bundle activation. |
If the bundle download or activation failed, the status update will contain
the following additional fields.
@@ -62,5 +66,14 @@ the following additional fields.
| `bundle.message` | `string` | Human readable messages describing the error(s). |
| `bundle.errors` | `array` | Collection of detailed parse or compile errors that occurred during activation. |
If the bundle download or activation failed, the status update will contain
the following additional fields.
| Field | Type | Description |
| --- | --- | --- |
| `discovery.code` | `string` | If present, indicates error(s) occurred. |
| `discovery.message` | `string` | Human readable messages describing the error(s). |
| `discovery.errors` | `array` | Collection of detailed parse or compile errors that occurred during activation. |
Services should reply with HTTP status `200 OK` if the status update is
processed successfully.
processed successfully.
download/config.go
+57
View File
@@ -0,0 +1,57 @@
// 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 download
import (
"fmt"
"time"
)
const (
defaultMinDelaySeconds = int64(60)
defaultMaxDelaySeconds = int64(120)
)
// PollingConfig represents polling configuration for the downloader.
type PollingConfig struct {
MinDelaySeconds *int64 `json:"min_delay_seconds,omitempty"` // min amount of time to wait between successful poll attempts
MaxDelaySeconds *int64 `json:"max_delay_seconds,omitempty"` // max amount of time to wait between poll attempts
}
// Config represents the configuration for the downloader.
type Config struct {
Polling PollingConfig `json:"polling"`
}
// ValidateAndInjectDefaults checks for configuration errors and ensures all
// values are set on the Config object.
func (c *Config) ValidateAndInjectDefaults() error {
min := defaultMinDelaySeconds
max := defaultMaxDelaySeconds
// reject bad min/max values
if c.Polling.MaxDelaySeconds != nil && c.Polling.MinDelaySeconds != nil {
if *c.Polling.MaxDelaySeconds < *c.Polling.MinDelaySeconds {
return fmt.Errorf("max polling delay must be >= min polling delay")
}
min = *c.Polling.MinDelaySeconds
max = *c.Polling.MaxDelaySeconds
} else if c.Polling.MaxDelaySeconds == nil && c.Polling.MinDelaySeconds != nil {
return fmt.Errorf("polling configuration missing 'max_delay_seconds'")
} else if c.Polling.MinDelaySeconds == nil && c.Polling.MaxDelaySeconds != nil {
return fmt.Errorf("polling configuration missing 'min_delay_seconds'")
}
// scale to seconds
minSeconds := int64(time.Duration(min) * time.Second)
c.Polling.MinDelaySeconds = &minSeconds
maxSeconds := int64(time.Duration(max) * time.Second)
c.Polling.MaxDelaySeconds = &maxSeconds
return nil
}
download/config_test.go
+91
View File
@@ -0,0 +1,91 @@
// 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 download
import (
"encoding/json"
"testing"
"time"
)
func TestConfigValidation(t *testing.T) {
tests := []struct {
note string
input string
wantErr bool
expMin time.Duration
expMax time.Duration
}{
{
note: "min > max",
input: `{
"polling": {
"min_delay_seconds": 10,
"max_delay_seconds": 1
}
}`,
wantErr: true,
},
{
note: "empty",
input: `{}`,
expMin: time.Second * time.Duration(defaultMinDelaySeconds),
expMax: time.Second * time.Duration(defaultMaxDelaySeconds),
},
{
note: "min missing",
input: `{
"polling": {
"max_delay_seconds": 10
}
}`,
wantErr: true,
},
{
note: "max missing",
input: `{
"polling": {
"min_delay_seconds": 1
}
}`,
wantErr: true,
},
{
note: "user supplied",
input: `{
"polling": {
"min_delay_seconds": 10,
"max_delay_seconds": 30
}
}`,
expMin: time.Second * 10,
expMax: time.Second * 30,
},
}
for _, test := range tests {
var config Config
if err := json.Unmarshal([]byte(test.input), &config); err != nil {
t.Fatal(err)
}
err := config.ValidateAndInjectDefaults()
if err != nil && !test.wantErr {
t.Errorf("Unexpected error on: %v, err: %v", test.input, err)
}
if err == nil {
if time.Duration(*config.Polling.MinDelaySeconds) != test.expMin {
t.Errorf("For %q expected min %v but got %v", test.note, test.expMin, time.Duration(*config.Polling.MinDelaySeconds))
}
if time.Duration(*config.Polling.MaxDelaySeconds) != test.expMax {
t.Errorf("For %q expected min %v but got %v", test.note, test.expMax, time.Duration(*config.Polling.MaxDelaySeconds))
}
}
}
}
download/download.go
+183
View File
@@ -0,0 +1,183 @@
// 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 download implements low-level OPA bundle downloading.
package download
import (
"context"
"fmt"
"math/rand"
"net/http"
"time"
"github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/plugins/rest"
"github.com/open-policy-agent/opa/util"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
const (
minRetryDelay = time.Millisecond * 100
)
// Update contains the result of a download. If an error occurred, the Error
// field will be non-nil. If a new bundle is available, the Bundle field will
// be non-nil.
type Update struct {
ETag string
Bundle *bundle.Bundle
Error error
}
// Downloader implements low-level OPA bundle downloading. Downloader can be
// started and stopped. After starting, the downloader will request bundle
// updatest from the remote HTTP endpoint that the client is configured to
// connect to.
type Downloader struct {
config Config // downloader configuration for tuning polling and other downloader behaviour
client rest.Client // HTTP client to use for bundle downloading
path string // path to use in bundle download request
stop chan chan struct{} // used to signal plugin to stop running
f func(context.Context, Update) // callback function invoked when download updates occur
logAttrs [][2]string // optional attributes to include in log messages
etag string // HTTP Etag for caching purposes
}
// New returns a new Downloader that can be started.
func New(config Config, client rest.Client, path string) *Downloader {
return &Downloader{
config: config,
client: client,
path: path,
stop: make(chan chan struct{}),
}
}
// WithCallback registers a function f to be called when download updates occur.
func (d *Downloader) WithCallback(f func(context.Context, Update)) *Downloader {
d.f = f
return d
}
// WithLogAttrs sets an optional set of key/value pair attributes to include in
// log messages emitted by the downloader.
func (d *Downloader) WithLogAttrs(attrs [][2]string) *Downloader {
d.logAttrs = attrs
return d
}
// Start tells the Downloader to begin downloading bundles.
func (d *Downloader) Start(ctx context.Context) {
go d.loop()
}
// Stop tells the Downloader to stop begin downloading bundles.
func (d *Downloader) Stop(ctx context.Context) {
done := make(chan struct{})
d.stop <- done
_ = <-done
}
func (d *Downloader) loop() {
ctx, cancel := context.WithCancel(context.Background())
var retry int
for {
err := d.oneShot(ctx)
var delay time.Duration
if err == nil {
min := float64(*d.config.Polling.MinDelaySeconds)
max := float64(*d.config.Polling.MaxDelaySeconds)
delay = time.Duration(((max - min) * rand.Float64()) + min)
} else {
delay = util.DefaultBackoff(float64(minRetryDelay), float64(*d.config.Polling.MaxDelaySeconds), retry)
}
d.logDebug("Waiting %v before next download/retry.", delay)
timer := time.NewTimer(delay)
select {
case <-timer.C:
if err != nil {
retry++
} else {
retry = 0
}
case done := <-d.stop:
cancel()
done <- struct{}{}
return
}
}
}
func (d *Downloader) oneShot(ctx context.Context) error {
b, etag, err := d.download(ctx)
if d.f != nil {
d.f(ctx, Update{ETag: etag, Bundle: b, Error: err})
}
d.etag = etag
return err
}
func (d *Downloader) download(ctx context.Context) (*bundle.Bundle, string, error) {
d.logDebug("Download starting.")
resp, err := d.client.WithHeader("If-None-Match", d.etag).Do(ctx, "GET", d.path)
if err != nil {
return nil, "", errors.Wrap(err, "request failed")
}
defer util.Close(resp)
switch resp.StatusCode {
case http.StatusOK:
d.logDebug("Download in progress.")
b, err := bundle.NewReader(resp.Body).Read()
if err != nil {
return nil, "", err
}
return &b, resp.Header.Get("ETag"), nil
case http.StatusNotModified:
return nil, resp.Header.Get("ETag"), nil
case http.StatusNotFound:
return nil, "", fmt.Errorf("server replied with not found")
case http.StatusUnauthorized:
return nil, "", fmt.Errorf("server replied with not authorized")
default:
return nil, "", fmt.Errorf("server replied with HTTP %v", resp.StatusCode)
}
}
func (d *Downloader) logError(fmt string, a ...interface{}) {
logrus.WithFields(d.logrusFields()).Errorf(fmt, a...)
}
func (d *Downloader) logInfo(fmt string, a ...interface{}) {
logrus.WithFields(d.logrusFields()).Infof(fmt, a...)
}
func (d *Downloader) logDebug(fmt string, a ...interface{}) {
logrus.WithFields(d.logrusFields()).Debugf(fmt, a...)
}
func (d *Downloader) logrusFields() logrus.Fields {
flds := logrus.Fields{}
for i := range d.logAttrs {
flds[d.logAttrs[i][0]] = flds[d.logAttrs[i][1]]
}
return flds
}
download/download_test.go
+234
View File
@@ -0,0 +1,234 @@
// 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 download
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/plugins/rest"
)
func TestStartStop(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
called := make(chan struct{})
config := Config{}
if err := config.ValidateAndInjectDefaults(); err != nil {
t.Fatal(err)
}
d := New(config, fixture.client, "/bundles/test/bundle1").WithCallback(func(context.Context, Update) {
called <- struct{}{}
})
d.Start(ctx)
_ = <-called
d.Stop(ctx)
}
func TestEtagCaching(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
fixture.server.expEtag = "some etag value"
defer fixture.server.stop()
updates := []Update{}
d := New(Config{}, fixture.client, "/bundles/test/bundle1").WithCallback(func(ctx context.Context, u Update) {
updates = append(updates, u)
})
err := d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(updates) != 1 || updates[0].ETag != "some etag value" {
t.Fatal("expected update")
}
err = d.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if len(updates) != 2 || updates[1].Bundle != nil {
t.Fatal("expected no change")
}
}
func TestFailureAuthn(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
fixture.server.expAuth = "Bearer anothersecret"
defer fixture.server.stop()
d := New(Config{}, fixture.client, "/bundles/test/bundle1")
err := d.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
}
func TestFailureNotFound(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
delete(fixture.server.bundles, "test/bundle1")
defer fixture.server.stop()
d := New(Config{}, fixture.client, "/bundles/test/non-existent")
err := d.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
}
func TestFailureUnexpected(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
fixture.server.expCode = 500
defer fixture.server.stop()
d := New(Config{}, fixture.client, "/bundles/test/bundle1")
err := d.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
}
type testFixture struct {
d *Downloader
client rest.Client
server *testServer
}
func newTestFixture(t *testing.T) testFixture {
ts := testServer{
t: t,
expAuth: "Bearer secret",
bundles: map[string]bundle.Bundle{
"test/bundle1": {
Manifest: bundle.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{
"foo": map[string]interface{}{
"bar": json.Number("1"),
"baz": "qux",
},
},
Modules: []bundle.ModuleFile{
{
Path: `/example.rego`,
Raw: []byte("package foo\n\ncorge=1"),
},
},
},
},
}
ts.start()
restConfig := []byte(fmt.Sprintf(`{
"url": %q,
"credentials": {
"bearer": {
"scheme": "Bearer",
"token": "secret"
}
}
}`, ts.server.URL))
tc, err := rest.New(restConfig)
if err != nil {
t.Fatal(err)
}
return testFixture{
client: tc,
server: &ts,
}
}
type testServer struct {
t *testing.T
expCode int
expEtag string
expAuth string
bundles map[string]bundle.Bundle
server *httptest.Server
}
func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
if t.expCode != 0 {
w.WriteHeader(t.expCode)
return
}
if t.expAuth != "" {
if r.Header.Get("Authorization") != t.expAuth {
w.WriteHeader(401)
return
}
}
name := strings.TrimPrefix(r.URL.Path, "/bundles/")
b, ok := t.bundles[name]
if !ok {
w.WriteHeader(404)
return
}
if t.expEtag != "" {
etag := r.Header.Get("If-None-Match")
if etag == t.expEtag {
w.WriteHeader(304)
return
}
}
w.Header().Add("Content-Type", "application/gzip")
if t.expEtag != "" {
w.Header().Add("Etag", t.expEtag)
}
w.WriteHeader(200)
var buf bytes.Buffer
if err := bundle.Write(&buf, b); err != nil {
w.WriteHeader(500)
}
if _, err := w.Write(buf.Bytes()); err != nil {
panic(err)
}
}
func (t *testServer) start() {
t.server = httptest.NewServer(http.HandlerFunc(t.handle))
}
func (t *testServer) stop() {
t.server.Close()
}
internal/runtime/runtime.go
+3 -9
View File
@@ -6,7 +6,6 @@
package runtime
import (
"io/ioutil"
"os"
"strings"
@@ -16,7 +15,7 @@ import (
// Params controls the types of runtime information to return.
type Params struct {
ConfigFile string
Config []byte
}
// Term returns the runtime information as an ast.Term object.
@@ -24,15 +23,10 @@ func Term(params Params) (*ast.Term, error) {
obj := ast.NewObject()
if params.ConfigFile != "" {
bs, err := ioutil.ReadFile(params.ConfigFile)
if err != nil {
return nil, err
}
if params.Config != nil {
var x interface{}
if err := util.Unmarshal(bs, &x); err != nil {
if err := util.Unmarshal(params.Config, &x); err != nil {
return nil, err
}
plugins/bundle/config.go
+66
View File
@@ -0,0 +1,66 @@
// 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 bundle
import (
"fmt"
"github.com/open-policy-agent/opa/download"
"github.com/open-policy-agent/opa/util"
)
// ParseConfig validates the config and injects default values.
func ParseConfig(config []byte, services []string) (*Config, error) {
if config == nil {
return nil, nil
}
var parsedConfig Config
if err := util.Unmarshal(config, &parsedConfig); err != nil {
return nil, err
}
if err := parsedConfig.validateAndInjectDefaults(services); err != nil {
return nil, err
}
return &parsedConfig, nil
}
// Config represents configuration the plguin.
type Config struct {
download.Config
Name string `json:"name"`
Service string `json:"service"`
}
func (c *Config) validateAndInjectDefaults(services []string) error {
if c.Name == "" {
return fmt.Errorf("invalid bundle name %q", c.Name)
}
if c.Service == "" && len(services) != 0 {
c.Service = services[0]
} else {
found := false
for _, svc := range services {
if svc == c.Service {
found = true
break
}
}
if !found {
return fmt.Errorf("invalid service name %q in bundle %q", c.Service, c.Name)
}
}
return c.ValidateAndInjectDefaults()
}
plugins/bundle/config_test.go
+35
View File
@@ -0,0 +1,35 @@
// 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 bundle
import "testing"
func TestConfigValidation(t *testing.T) {
tests := []struct {
input string
wantErr bool
}{
{
input: `{}`,
wantErr: true,
},
{
input: `{"name": "a/b/c", "service": "invalid"}`,
wantErr: true,
},
{
input: `{"name": a/b/c", "service": "service2"}`,
wantErr: false,
},
}
for _, test := range tests {
config, _ := ParseConfig([]byte(test.input), []string{"service1", "service2"})
if config == nil {
continue
}
}
}
plugins/bundle/plugin.go
+114 -312
View File
@@ -8,196 +8,99 @@ package bundle
import (
"context"
"fmt"
"math/rand"
"net/http"
"reflect"
"sync"
"time"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/download"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/server/types"
"github.com/open-policy-agent/opa/storage"
"github.com/open-policy-agent/opa/util"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
const (
// min amount of time to wait following a failure
minRetryDelay = time.Millisecond * 100
defaultMinDelaySeconds = int64(60)
defaultMaxDelaySeconds = int64(120)
)
// PollingConfig represents configuration for the plugin's polling behaviour.
type PollingConfig struct {
MinDelaySeconds *int64 `json:"min_delay_seconds,omitempty"` // min amount of time to wait between successful poll attempts
MaxDelaySeconds *int64 `json:"max_delay_seconds,omitempty"` // max amount of time to wait between poll attempts
}
// Config represents configuration the plguin.
type Config struct {
Name string `json:"name"`
Service string `json:"service"`
Polling PollingConfig `json:"polling"`
}
func (c *Config) validateAndInjectDefaults(services []string) error {
if c.Name == "" {
return fmt.Errorf("invalid bundle name %q", c.Name)
}
if c.Service == "" && len(services) != 0 {
c.Service = services[0]
} else {
found := false
for _, svc := range services {
if svc == c.Service {
found = true
break
}
}
if !found {
return fmt.Errorf("invalid service name %q in bundle %q", c.Service, c.Name)
}
}
min := defaultMinDelaySeconds
max := defaultMaxDelaySeconds
// reject bad min/max values
if c.Polling.MaxDelaySeconds != nil && c.Polling.MinDelaySeconds != nil {
if *c.Polling.MaxDelaySeconds < *c.Polling.MinDelaySeconds {
return fmt.Errorf("max polling delay must be >= min polling delay in bundle %q", c.Name)
}
min = *c.Polling.MinDelaySeconds
max = *c.Polling.MaxDelaySeconds
} else if c.Polling.MaxDelaySeconds == nil && c.Polling.MinDelaySeconds != nil {
return fmt.Errorf("polling configuration missing 'max_delay_seconds' in bundle %q", c.Name)
} else if c.Polling.MinDelaySeconds == nil && c.Polling.MaxDelaySeconds != nil {
return fmt.Errorf("polling configuration missing 'min_delay_seconds' in bundle %q", c.Name)
}
// scale to seconds
minSeconds := int64(time.Duration(min) * time.Second)
c.Polling.MinDelaySeconds = &minSeconds
maxSeconds := int64(time.Duration(max) * time.Second)
c.Polling.MaxDelaySeconds = &maxSeconds
return nil
}
func (c *Config) equal(other Config) bool {
if c.Name != other.Name {
return false
}
if c.Service != other.Service {
return false
}
if *c.Polling.MaxDelaySeconds != *other.Polling.MaxDelaySeconds {
return false
}
if *c.Polling.MinDelaySeconds != *other.Polling.MinDelaySeconds {
return false
}
return true
}
const (
errCode = "bundle_error"
)
// Status represents the status of the plugin.
type Status struct {
Name string `json:"name"`
ActiveRevision string `json:"active_revision,omitempty"`
LastSuccessfulActivation time.Time `json:"last_successful_activation,omitempty"`
LastSuccessfulDownload time.Time `json:"last_successful_download,omitempty"`
DiscoveryStatus bool `json:"discovery_status"`
Code string `json:"code,omitempty"`
Message string `json:"message,omitempty"`
Errors []error `json:"errors,omitempty"`
}
// Plugin implements bundle downloading and activation.
// Plugin implements bundle activation.
type Plugin struct {
manager *plugins.Manager // plugin manager for storage and service clients
config Config // plugin config
stop chan chan struct{} // used to signal plugin to stop running
reconfig chan interface{} // used to receive updated configuration from periodic discovery
etag string // last ETag header for caching purposes
status *Status // current plugin status
listeners map[interface{}]func(
Status) // listeners to send status updates to
mtx sync.Mutex
}
// ParseConfig validates the config and injects default values.
func ParseConfig(config []byte, services []string) (*Config, error) {
var parsedConfig Config
if err := util.Unmarshal(config, &parsedConfig); err != nil {
return nil, err
}
if err := parsedConfig.validateAndInjectDefaults(services); err != nil {
return nil, err
}
return &parsedConfig, nil
config Config
manager *plugins.Manager // plugin manager for storage and service clients
status *Status // current plugin status
etag string // etag on last successful activation
listeners map[interface{}]func(Status) // listeners to send status updates to
downloader *download.Downloader
mtx sync.Mutex
}
// New returns a new Plugin with the given config.
func New(parsedConfig *Config, manager *plugins.Manager) (*Plugin, error) {
plugin := &Plugin{
func New(parsedConfig *Config, manager *plugins.Manager) *Plugin {
p := &Plugin{
manager: manager,
config: *parsedConfig,
stop: make(chan chan struct{}),
status: &Status{
Name: parsedConfig.Name,
},
listeners: map[interface{}]func(Status){},
reconfig: make(chan interface{}),
}
p.initDownloader()
return p
}
return plugin, nil
// Name identifies the plugin on manager.
const Name = "bundle"
// Lookup returns the bundle plugin registered with the manager.
func Lookup(manager *plugins.Manager) *Plugin {
if p := manager.Plugin(Name); p != nil {
return p.(*Plugin)
}
return nil
}
// Start runs the plugin. The plugin will periodically try to download bundles
// from the configured service. When a new bundle is downloaded, the data and
// policies are extracted and inserted into storage.
func (p *Plugin) Start(ctx context.Context) error {
go p.loop()
p.logInfo("Starting bundle downloader.")
p.mtx.Lock()
defer p.mtx.Unlock()
p.downloader.Start(ctx)
return nil
}
// Stop stops the plugin.
func (p *Plugin) Stop(ctx context.Context) {
done := make(chan struct{})
p.stop <- done
_ = <-done
p.logInfo("Stopping bundle downloader.")
p.mtx.Lock()
defer p.mtx.Unlock()
p.downloader.Stop(ctx)
}
// Reconfigure notifies the plugin with a new configuration.
func (p *Plugin) Reconfigure(config interface{}) {
p.reconfig <- config
// Reconfigure notifies the plugin that it's configuration has changed.
func (p *Plugin) Reconfigure(ctx context.Context, config interface{}) {
p.mtx.Lock()
defer p.mtx.Unlock()
newConfig := config.(*Config)
if reflect.DeepEqual(p.config, *newConfig) {
p.logDebug("Bundle downloader configuration unchanged.")
return
}
p.logInfo("Bundle downloader configuration changed. Restarting bundle downloader.")
p.config = *config.(*Config)
p.downloader.Stop(ctx)
p.initDownloader()
p.downloader.Start(ctx)
}
// Register a lisetner to receive status updates. The name must be comparable.
// Register a listener to receive status updates. The name must be comparable.
func (p *Plugin) Register(name interface{}, listener func(Status)) {
p.mtx.Lock()
defer p.mtx.Unlock()
if p.listeners == nil {
p.listeners = map[interface{}]func(Status){}
}
p.listeners[name] = listener
}
@@ -205,136 +108,64 @@ func (p *Plugin) Register(name interface{}, listener func(Status)) {
func (p *Plugin) Unregister(name interface{}) {
p.mtx.Lock()
defer p.mtx.Unlock()
delete(p.listeners, name)
}
// Equal checks if the current and provided input config are equal.
func (p *Plugin) Equal(other *Config) bool {
return p.config.equal(*other)
func (p *Plugin) initDownloader() {
client := p.manager.Client(p.config.Service)
path := fmt.Sprintf("/bundles/%v", p.config.Name)
p.downloader = download.New(p.config.Config, client, path).WithCallback(p.oneShot)
}
func (p *Plugin) loop() {
func (p *Plugin) oneShot(ctx context.Context, u download.Update) {
p.mtx.Lock()
defer p.mtx.Unlock()
ctx, cancel := context.WithCancel(context.Background())
p.process(ctx, u)
status := *p.status
var retry int
for _, listener := range p.listeners {
listener(status)
}
}
for {
updated, err := p.oneShot(ctx)
func (p *Plugin) process(ctx context.Context, u download.Update) {
if err != nil {
p.logError("%v.", err)
} else if !updated {
p.logDebug("Bundle download skipped, server replied with not modified.")
} else if p.etag != "" {
p.logInfo("Bundle downloaded and activated successfully. Etag updated to %v.", p.etag)
if u.Error != nil {
p.logError("Bundle download failed: %v", u.Error)
p.status.SetError(u.Error)
return
}
if u.Bundle != nil {
p.status.SetDownloadSuccess()
if err := p.activate(ctx, u.Bundle); err != nil {
p.logError("Bundle activation failed: %v", err)
p.status.SetError(err)
return
}
p.status.SetError(nil)
p.status.SetActivateSuccess(u.Bundle.Manifest.Revision)
if u.ETag != "" {
p.logInfo("Bundle downloaded and activated successfully. Etag updated to %v.", u.ETag)
} else {
p.logInfo("Bundle downloaded and activated successfully.")
}
var delay time.Duration
if err == nil {
min := float64(*p.config.Polling.MinDelaySeconds)
max := float64(*p.config.Polling.MaxDelaySeconds)
delay = time.Duration(((max - min) * rand.Float64()) + min)
} else {
delay = util.DefaultBackoff(float64(minRetryDelay), float64(*p.config.Polling.MaxDelaySeconds), retry)
}
p.logDebug("Waiting %v before next download/retry.", delay)
timer := time.NewTimer(delay)
select {
case <-timer.C:
if err != nil {
retry++
} else {
retry = 0
}
case newConfig := <-p.reconfig:
p.reconfigure(newConfig)
case done := <-p.stop:
cancel()
done <- struct{}{}
return
}
p.etag = u.ETag
return
}
}
func (p *Plugin) reconfigure(newConfig interface{}) {
p.config = *newConfig.(*Config)
}
func (p *Plugin) oneShot(ctx context.Context) (updated bool, err error) {
defer func() {
p.setErrorStatus(err)
status := *p.status
for _, listener := range p.listeners {
listener(status)
}
}()
p.logDebug("Download starting.")
resp, err := p.manager.Client(p.config.Service).
WithHeader("If-None-Match", p.etag).
Do(ctx, "GET", fmt.Sprintf("/bundles/%v", p.config.Name))
if err != nil {
return false, errors.Wrap(err, "Download request failed")
}
defer util.Close(resp)
switch resp.StatusCode {
case http.StatusOK:
if err := p.process(ctx, resp); err != nil {
return false, err
}
return true, nil
case http.StatusNotModified:
return false, nil
case http.StatusNotFound:
return false, fmt.Errorf("Bundle download failed, server replied with not found")
case http.StatusUnauthorized:
return false, fmt.Errorf("Bundle download failed, server replied with not authorized")
default:
return false, fmt.Errorf("Bundle download failed, server replied with HTTP %v", resp.StatusCode)
if u.ETag == p.etag {
p.logDebug("Bundle download skipped, server replied with not modified.")
p.status.SetError(nil)
return
}
}
func (p *Plugin) process(ctx context.Context, resp *http.Response) error {
b, err := p.download(ctx, resp)
if err != nil {
return errors.Wrap(err, "Bundle download failed")
}
if err := p.activate(ctx, resp.Header.Get("ETag"), *b); err != nil {
return errors.Wrap(err, "Bundle activation failed")
}
return nil
}
func (p *Plugin) download(ctx context.Context, resp *http.Response) (*bundle.Bundle, error) {
p.logDebug("Bundle download in progress.")
b, err := bundle.NewReader(resp.Body).Read()
if err != nil {
return nil, err
}
p.status.LastSuccessfulDownload = time.Now().UTC()
return &b, nil
}
func (p *Plugin) activate(ctx context.Context, etag string, b bundle.Bundle) error {
func (p *Plugin) activate(ctx context.Context, b *bundle.Bundle) error {
p.logDebug("Bundle activation in progress. Opening storage transaction.")
return storage.Txn(ctx, p.manager.Store, storage.WriteParams, func(txn storage.Transaction) error {
@@ -381,58 +212,10 @@ func (p *Plugin) activate(ctx context.Context, etag string, b bundle.Bundle) err
}
}
p.status.LastSuccessfulActivation = time.Now().UTC()
p.status.ActiveRevision = b.Manifest.Revision
p.etag = etag
return nil
})
}
func (p *Plugin) logError(fmt string, a ...interface{}) {
logrus.WithFields(p.logrusFields()).Errorf(fmt, a...)
}
func (p *Plugin) logInfo(fmt string, a ...interface{}) {
logrus.WithFields(p.logrusFields()).Infof(fmt, a...)
}
func (p *Plugin) logDebug(fmt string, a ...interface{}) {
logrus.WithFields(p.logrusFields()).Debugf(fmt, a...)
}
func (p *Plugin) logrusFields() logrus.Fields {
return logrus.Fields{
"plugin": "bundle",
"name": p.config.Name,
}
}
func (p *Plugin) setErrorStatus(err error) {
if err == nil {
p.status.Code = ""
p.status.Message = ""
p.status.Errors = nil
return
}
cause := errors.Cause(err)
if astErr, ok := cause.(ast.Errors); ok {
p.status.Code = errCode
p.status.Message = types.MsgCompileModuleError
p.status.Errors = make([]error, len(astErr))
for i := range astErr {
p.status.Errors[i] = astErr[i]
}
} else {
p.status.Code = errCode
p.status.Message = err.Error()
p.status.Errors = nil
}
}
func (p *Plugin) writeManifest(ctx context.Context, txn storage.Transaction, m bundle.Manifest) error {
var value interface{} = m
@@ -448,6 +231,25 @@ func (p *Plugin) writeManifest(ctx context.Context, txn storage.Transaction, m b
return p.manager.Store.Write(ctx, txn, storage.AddOp, manifestPath, value)
}
func (p *Plugin) logError(fmt string, a ...interface{}) {
logrus.WithFields(p.logrusFields()).Errorf(fmt, a...)
}
func (p *Plugin) logInfo(fmt string, a ...interface{}) {
logrus.WithFields(p.logrusFields()).Infof(fmt, a...)
}
func (p *Plugin) logDebug(fmt string, a ...interface{}) {
logrus.WithFields(p.logrusFields()).Debugf(fmt, a...)
}
func (p *Plugin) logrusFields() logrus.Fields {
return logrus.Fields{
"plugin": Name,
"name": p.config.Name,
}
}
var (
bundlePath = storage.MustParsePath("/system/bundle")
manifestPath = storage.MustParsePath("/system/bundle/manifest")
plugins/bundle/plugin_test.go
+138 -488
View File
@@ -7,351 +7,122 @@ package bundle
import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"strings"
"testing"
"time"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/download"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/storage"
"github.com/open-policy-agent/opa/storage/inmem"
"github.com/open-policy-agent/opa/util"
)
func TestNew(t *testing.T) {
store := inmem.New()
manager, err := plugins.New([]byte(`{
"services": [
{
"name": "foo"
}
]
}`), "test-instance-id", store)
if err != nil {
t.Fatal(err)
}
tests := []struct {
input string
wantErr bool
expMin time.Duration
expMax time.Duration
}{
{
input: `{
}`,
wantErr: true,
},
{
input: `{
"name": "a/b/c",
"service": "missing service"
}`,
wantErr: true,
},
{
input: `{
"name": "bad/delays",
"service": "foo",
"polling": {
"min_delay_seconds": 10,
"max_delay_seconds": 1
}
}`,
wantErr: true,
},
{
input: `{
"name": "defaults",
"service": "foo"
}`,
expMin: time.Second * time.Duration(defaultMinDelaySeconds),
expMax: time.Second * time.Duration(defaultMaxDelaySeconds),
},
{
input: `{
"name": "missing/min",
"service": "foo",
"polling": {
"max_delay_seconds": 10
}
}`,
wantErr: true,
},
{
input: `{
"name": "missing/max",
"service": "foo",
"polling": {
"min_delay_seconds": 1
}
}`,
wantErr: true,
},
{
input: `{
"name": "user/min/max",
"service": "foo",
"polling": {
"min_delay_seconds": 10,
"max_delay_seconds": 30
}
}`,
expMin: time.Second * 10,
expMax: time.Second * 30,
},
}
for _, test := range tests {
config, _ := ParseConfig([]byte(test.input), manager.Services())
if config == nil {
continue
}
p, err := New(config, manager)
if err != nil && !test.wantErr {
t.Errorf("Unexpected error on: %v, err: %v", test.input, err)
}
if err == nil {
if time.Duration(*p.config.Polling.MinDelaySeconds) != test.expMin {
t.Errorf("For %q expected min %v but got %v", p.config.Name, test.expMin, time.Duration(*p.config.Polling.MinDelaySeconds))
}
if time.Duration(*p.config.Polling.MaxDelaySeconds) != test.expMax {
t.Errorf("For %q expected min %v but got %v", p.config.Name, test.expMax, time.Duration(*p.config.Polling.MaxDelaySeconds))
}
}
}
}
func TestPluginStart(t *testing.T) {
func TestPluginOneShot(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
defer fixture.server.stop()
manager := getTestManager()
plugin := Plugin{manager: manager, status: &Status{}}
if err := fixture.plugin.Start(ctx); err != nil {
t.Fatal(err)
}
module := "package foo\n\ncorge=1"
txn := storage.NewTransactionOrDie(ctx, fixture.store, storage.WriteParams)
done := make(chan struct{})
fixture.store.Register(ctx, txn, storage.TriggerConfig{
OnCommit: func(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) {
if !event.DataChanged() || !event.PolicyChanged() {
return
}
ids, err := fixture.store.ListPolicies(ctx, txn)
if err != nil {
t.Fatal(err)
} else if len(ids) != 1 {
t.Fatal("Expected 1 policy")
}
bs, err := fixture.store.GetPolicy(ctx, txn, ids[0])
exp := []byte("package foo\n\ncorge=1")
if err != nil {
t.Fatal(err)
} else if !bytes.Equal(bs, exp) {
t.Fatalf("Bad policy content. Exp:\n%v\n\nGot:\n\n%v", string(exp), string(bs))
}
data, err := fixture.store.Read(ctx, txn, storage.Path{})
expData := util.MustUnmarshalJSON([]byte(`{"foo": {"bar": 1, "baz": "qux"}, "system": {"bundle": {"manifest": {"revision": "quickbrownfaux"}}}}`))
if err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(data, expData) {
t.Fatalf("Bad data content. Exp:\n%v\n\nGot:\n\n%v", expData, data)
}
done <- struct{}{}
},
})
if err := fixture.store.Commit(ctx, txn); err != nil {
t.Fatal(err)
}
<-done
fixture.plugin.Stop(ctx)
}
func TestPluginEtagCaching(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
fixture.server.expEtag = "some etag value"
defer fixture.server.stop()
updated, err := fixture.plugin.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if !updated {
t.Fatal("expected update")
}
updated, err = fixture.plugin.oneShot(ctx)
if err != nil {
t.Fatal("Unexpected:", err)
} else if updated {
t.Fatal("expected not update")
}
}
func TestPluginFailureAuthn(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
fixture.server.expAuth = "Bearer anothersecret"
defer fixture.server.stop()
_, err := fixture.plugin.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
}
func TestPluginFailureNotFound(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
delete(fixture.server.bundles, "test/bundle1")
defer fixture.server.stop()
_, err := fixture.plugin.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
}
func TestPluginFailureUnexpected(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
fixture.server.expCode = 500
defer fixture.server.stop()
_, err := fixture.plugin.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
}
}
func TestPluginFailureCompile(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
defer fixture.server.stop()
_, err := fixture.plugin.oneShot(ctx)
if err != nil {
t.Fatal("expected error")
}
fixture.server.bundles["test/bundle1"] = bundle.Bundle{
Data: map[string]interface{}{},
b := bundle.Bundle{
Manifest: bundle.Manifest{Revision: "quickbrownfaux"},
Data: util.MustUnmarshalJSON([]byte(`{"foo": {"bar": 1, "baz": "qux"}}`)).(map[string]interface{}),
Modules: []bundle.ModuleFile{
{
Path: `/example.rego`,
Raw: []byte("package foo\n\np[x]"),
bundle.ModuleFile{
Path: "/foo/bar",
Parsed: ast.MustParseModule(module),
Raw: []byte(module),
},
},
}
_, err = fixture.plugin.oneShot(ctx)
if err == nil {
t.Fatal("expected error")
plugin.oneShot(ctx, download.Update{Bundle: &b})
txn := storage.NewTransactionOrDie(ctx, manager.Store)
defer manager.Store.Abort(ctx, txn)
ids, err := manager.Store.ListPolicies(ctx, txn)
if err != nil {
t.Fatal(err)
} else if len(ids) != 1 {
t.Fatal("Expected 1 policy")
}
// ensure data/policy is intact
txn := storage.NewTransactionOrDie(ctx, fixture.store, storage.TransactionParams{})
bs, err := manager.Store.GetPolicy(ctx, txn, ids[0])
exp := []byte("package foo\n\ncorge=1")
if err != nil {
t.Fatal(err)
} else if !bytes.Equal(bs, exp) {
t.Fatalf("Bad policy content. Exp:\n%v\n\nGot:\n\n%v", string(exp), string(bs))
}
_, err = fixture.store.GetPolicy(ctx, txn, "/example.rego")
data, err := manager.Store.Read(ctx, txn, storage.Path{})
expData := util.MustUnmarshalJSON([]byte(`{"foo": {"bar": 1, "baz": "qux"}, "system": {"bundle": {"manifest": {"revision": "quickbrownfaux"}}}}`))
if err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(data, expData) {
t.Fatalf("Bad data content. Exp:\n%v\n\nGot:\n\n%v", expData, data)
}
}
func TestPluginOneShotCompileError(t *testing.T) {
ctx := context.Background()
manager := getTestManager()
plugin := Plugin{manager: manager, status: &Status{}}
b1 := &bundle.Bundle{
Data: map[string]interface{}{"a": "b"},
Modules: []bundle.ModuleFile{
{
Path: "/example.rego",
Parsed: ast.MustParseModule("package foo\n\np[x] { x = 1 }"),
},
},
}
plugin.oneShot(ctx, download.Update{Bundle: b1})
b2 := &bundle.Bundle{
Modules: []bundle.ModuleFile{
{
Path: "/example.rego",
Parsed: ast.MustParseModule("package foo\n\np[x]"),
},
},
}
plugin.oneShot(ctx, download.Update{Bundle: b2})
txn := storage.NewTransactionOrDie(ctx, manager.Store)
_, err := manager.Store.GetPolicy(ctx, txn, "/example.rego")
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
data, err := fixture.store.Read(ctx, txn, storage.Path{})
data, err := manager.Store.Read(ctx, txn, storage.Path{})
if err != nil || data == nil {
t.Fatalf("Expected data to be intact but got: %v, err: %v", data, err)
}
}
func TestPluginReconfigure(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
defer fixture.server.stop()
if err := fixture.plugin.Start(ctx); err != nil {
t.Fatal(err)
}
minDelay := 2
maxDelay := 3
pluginConfig := []byte(fmt.Sprintf(`{
"name": "test/bundle1",
"service": "example",
"polling": {
"min_delay_seconds": %v,
"max_delay_seconds": %v
}
}`, minDelay, maxDelay))
config, _ := ParseConfig(pluginConfig, fixture.manager.Services())
fixture.plugin.Reconfigure(config)
fixture.plugin.Stop(ctx)
actualMin := time.Duration(*fixture.plugin.config.Polling.MinDelaySeconds) / time.Nanosecond
expectedMin := time.Duration(minDelay) * time.Second
if actualMin != expectedMin {
t.Fatalf("Expected minimum polling interval: %v but got %v", expectedMin, actualMin)
}
actualMax := time.Duration(*fixture.plugin.config.Polling.MaxDelaySeconds) / time.Nanosecond
expectedMax := time.Duration(maxDelay) * time.Second
if actualMax != expectedMax {
t.Fatalf("Expected maximum polling interval: %v but got %v", expectedMax, actualMax)
}
}
func TestPluginActivatationRemovesOld(t *testing.T) {
managerConfig := []byte(`{
"services": [
{
"name": "example",
"url": "http://localhost"
}
]
}`)
store := inmem.New()
manager, err := plugins.New(managerConfig, "test-instance-id", store)
if err != nil {
t.Fatal(err)
}
config, _ := ParseConfig([]byte(`{"name": "test", "service": "example"}`), manager.Services())
p, err := New(config, manager)
if err != nil {
t.Fatal(err)
}
func TestPluginOneShotActivatationRemovesOld(t *testing.T) {
ctx := context.Background()
manager := getTestManager()
plugin := Plugin{manager: manager, status: &Status{}}
module1 := `package example
p = 1`
p = 1`
b := bundle.Bundle{
b1 := bundle.Bundle{
Data: map[string]interface{}{
"foo": "bar",
},
@@ -364,13 +135,11 @@ func TestPluginActivatationRemovesOld(t *testing.T) {
},
}
if err := p.activate(ctx, "firstetag", b); err != nil {
t.Fatal("Unexpected:", err)
}
plugin.oneShot(ctx, download.Update{Bundle: &b1})
module2 := `package example
p = 2`
p = 2`
b2 := bundle.Bundle{
Data: map[string]interface{}{
@@ -385,18 +154,16 @@ func TestPluginActivatationRemovesOld(t *testing.T) {
},
}
if err := p.activate(ctx, "secondetag", b2); err != nil {
t.Fatal("Unexpected:", err)
}
plugin.oneShot(ctx, download.Update{Bundle: &b2})
err = storage.Txn(ctx, store, storage.TransactionParams{}, func(txn storage.Transaction) error {
ids, err := store.ListPolicies(ctx, txn)
err := storage.Txn(ctx, manager.Store, storage.TransactionParams{}, func(txn storage.Transaction) error {
ids, err := manager.Store.ListPolicies(ctx, txn)
if err != nil {
return err
} else if !reflect.DeepEqual([]string{"/example2.rego"}, ids) {
return fmt.Errorf("expected updated policy ids")
}
data, err := store.Read(ctx, txn, storage.Path{})
data, err := manager.Store.Read(ctx, txn, storage.Path{})
// remove system key to make comparison simpler
delete(data.(map[string]interface{}), "system")
if err != nil {
@@ -415,57 +182,74 @@ func TestPluginActivatationRemovesOld(t *testing.T) {
func TestPluginListener(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
defer fixture.server.stop()
manager := getTestManager()
plugin := Plugin{manager: manager, status: &Status{}}
ch := make(chan Status)
b := fixture.server.bundles["test/bundle1"]
ch := make(chan Status, 1)
fixture.plugin.Register("test", func(status Status) {
plugin.Register("test", func(status Status) {
ch <- status
})
// Test that initial bundle is ok.
fixture.plugin.oneShot(ctx)
module := "package gork\np[x] { x = 1 }"
b := bundle.Bundle{
Manifest: bundle.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{},
Modules: []bundle.ModuleFile{
{
Path: "/foo.rego",
Parsed: ast.MustParseModule(module),
Raw: []byte(module),
},
},
}
// Test that initial bundle is ok. Defer to separate goroutine so we can
// check result with channel.
go plugin.oneShot(ctx, download.Update{Bundle: &b})
s1 := <-ch
if s1.ActiveRevision != "quickbrownfaux" || s1.Code != "" {
t.Fatal("Unexpected status update, got:", s1)
}
// Test that next update is failed.
module = "package gork\np[x]"
b.Manifest.Revision = "slowgreenburd"
b.Modules[0] = bundle.ModuleFile{
Path: "/foo.rego",
Raw: []byte("package gork\np[x]"),
Path: "/foo.rego",
Raw: []byte(module),
Parsed: ast.MustParseModule(module),
}
fixture.server.bundles["test/bundle1"] = b
fixture.plugin.oneShot(ctx)
// Test that next update is failed.
go plugin.oneShot(ctx, download.Update{Bundle: &b})
s2 := <-ch
if s2.ActiveRevision != "quickbrownfaux" || s2.Code == "" || s2.Message == "" || len(s2.Errors) == 0 {
t.Fatal("Unexpected status update, got:", s2)
}
// Test that new update is successful.
module = "package gork\np[1]"
b.Manifest.Revision = "fancybluederg"
b.Modules[0] = bundle.ModuleFile{
Path: "/foo.rego",
Raw: []byte("package gork\np[1]"),
Path: "/foo.rego",
Raw: []byte(module),
Parsed: ast.MustParseModule(module),
}
fixture.server.bundles["test/bundle1"] = b
fixture.server.expEtag = "etagvalue"
fixture.plugin.oneShot(ctx)
// Test that new update is successful.
go plugin.oneShot(ctx, download.Update{Bundle: &b})
s3 := <-ch
if s3.ActiveRevision != "fancybluederg" || s3.Code != "" || s3.Message != "" || len(s3.Errors) != 0 {
t.Fatal("Unexpected status update, got:", s3)
}
// Test that 304 results in status update.
fixture.plugin.oneShot(ctx)
// Test that empty download update results in status update.
go plugin.oneShot(ctx, download.Update{})
s4 := <-ch
if !reflect.DeepEqual(s3, s4) {
@@ -476,19 +260,23 @@ func TestPluginListener(t *testing.T) {
func TestPluginListenerErrorClearedOn304(t *testing.T) {
ctx := context.Background()
fixture := newTestFixture(t)
defer fixture.server.stop()
manager := getTestManager()
plugin := Plugin{manager: manager, status: &Status{}}
ch := make(chan Status)
// b := fixture.server.bundles["test/bundle1"]
ch := make(chan Status, 1)
fixture.plugin.Register("test", func(status Status) {
plugin.Register("test", func(status Status) {
ch <- status
})
b := bundle.Bundle{
Manifest: bundle.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{"foo": "bar"},
}
// Test that initial bundle is ok.
fixture.server.expEtag = "etagvalue"
fixture.plugin.oneShot(ctx)
go plugin.oneShot(ctx, download.Update{Bundle: &b})
s1 := <-ch
if s1.ActiveRevision != "quickbrownfaux" || s1.Code != "" {
@@ -496,8 +284,7 @@ func TestPluginListenerErrorClearedOn304(t *testing.T) {
}
// Test that service error triggers failure notification.
fixture.server.expCode = 500
fixture.plugin.oneShot(ctx)
go plugin.oneShot(ctx, download.Update{Error: fmt.Errorf("some error")})
s2 := <-ch
if s2.ActiveRevision != "quickbrownfaux" || s2.Code == "" {
@@ -505,8 +292,7 @@ func TestPluginListenerErrorClearedOn304(t *testing.T) {
}
// Test that service recovery triggers healthy notification.
fixture.server.expCode = 304
fixture.plugin.oneShot(ctx)
go plugin.oneShot(ctx, download.Update{})
s3 := <-ch
if s3.ActiveRevision != "quickbrownfaux" || s3.Code != "" {
@@ -514,147 +300,11 @@ func TestPluginListenerErrorClearedOn304(t *testing.T) {
}
}
type testFixture struct {
store storage.Store
manager *plugins.Manager
plugin *Plugin
server *testServer
}
func newTestFixture(t *testing.T) testFixture {
ts := testServer{
t: t,
expAuth: "Bearer secret",
bundles: map[string]bundle.Bundle{
"test/bundle1": {
Manifest: bundle.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{
"foo": map[string]interface{}{
"bar": json.Number("1"),
"baz": "qux",
},
},
Modules: []bundle.ModuleFile{
{
Path: `/example.rego`,
Raw: []byte("package foo\n\ncorge=1"),
},
},
},
},
}
ts.start()
managerConfig := []byte(fmt.Sprintf(`{
"services": [
{
"name": "example",
"url": %q,
"credentials": {
"bearer": {
"scheme": "Bearer",
"token": "secret"
}
}
}
]}`, ts.server.URL))
func getTestManager() *plugins.Manager {
store := inmem.New()
manager, err := plugins.New(managerConfig, "test-instance-id", store)
manager, err := plugins.New(nil, "test-instance-id", store)
if err != nil {
t.Fatal(err)
}
pluginConfig := []byte(fmt.Sprintf(`{
"name": "test/bundle1",
"service": "example",
"polling": {
"min_delay_seconds": 1,
"max_delay_seconds": 1
}
}`))
config, _ := ParseConfig(pluginConfig, manager.Services())
p, err := New(config, manager)
if err != nil {
t.Fatal(err)
}
return testFixture{
store: store,
manager: manager,
plugin: p,
server: &ts,
}
}
type testServer struct {
t *testing.T
expCode int
expEtag string
expAuth string
bundles map[string]bundle.Bundle
server *httptest.Server
}
func (t *testServer) handle(w http.ResponseWriter, r *http.Request) {
if t.expCode != 0 {
w.WriteHeader(t.expCode)
return
}
if t.expAuth != "" {
if r.Header.Get("Authorization") != t.expAuth {
w.WriteHeader(401)
return
}
}
name := strings.TrimPrefix(r.URL.Path, "/bundles/")
b, ok := t.bundles[name]
if !ok {
w.WriteHeader(404)
return
}
if t.expEtag != "" {
etag := r.Header.Get("If-None-Match")
if etag == t.expEtag {
w.WriteHeader(304)
return
}
}
w.Header().Add("Content-Type", "application/gzip")
if t.expEtag != "" {
w.Header().Add("Etag", t.expEtag)
}
w.WriteHeader(200)
var buf bytes.Buffer
if err := bundle.Write(&buf, b); err != nil {
w.WriteHeader(500)
}
if _, err := w.Write(buf.Bytes()); err != nil {
panic(err)
}
}
func (t *testServer) start() {
t.server = httptest.NewServer(http.HandlerFunc(t.handle))
}
func (t *testServer) stop() {
t.server.Close()
return manager
}
plugins/bundle/status.go
+68
View File
@@ -0,0 +1,68 @@
// 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 bundle
import (
"time"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/server/types"
"github.com/pkg/errors"
)
const (
errCode = "bundle_error"
)
// Status represents the status of processing a bundle.
type Status struct {
Name string `json:"name"`
ActiveRevision string `json:"active_revision,omitempty"`
LastSuccessfulActivation time.Time `json:"last_successful_activation,omitempty"`
LastSuccessfulDownload time.Time `json:"last_successful_download,omitempty"`
Code string `json:"code,omitempty"`
Message string `json:"message,omitempty"`
Errors []error `json:"errors,omitempty"`
}
// SetActivateSuccess updates the status object to reflect a successful
// activation.
func (s *Status) SetActivateSuccess(revision string) {
s.LastSuccessfulActivation = time.Now().UTC()
s.ActiveRevision = revision
}
// SetDownloadSuccess updates the status object to reflect a successful
// download.
func (s *Status) SetDownloadSuccess() {
s.LastSuccessfulDownload = time.Now().UTC()
}
// SetError updates the status object to reflect a failure to download or
// activate. If err is nil, the error status is cleared.
func (s *Status) SetError(err error) {
if err == nil {
s.Code = ""
s.Message = ""
s.Errors = nil
return
}
cause := errors.Cause(err)
if astErr, ok := cause.(ast.Errors); ok {
s.Code = errCode
s.Message = types.MsgCompileModuleError
s.Errors = make([]error, len(astErr))
for i := range astErr {
s.Errors[i] = astErr[i]
}
} else {
s.Code = errCode
s.Message = err.Error()
s.Errors = nil
}
}
plugins/discovery/config.go
+68
View File
@@ -0,0 +1,68 @@
// 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 discovery
import (
"fmt"
"strings"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/download"
"github.com/open-policy-agent/opa/util"
)
// Config represents the configuration for the discovery feature.
type Config struct {
download.Config // bundle downloader configuration
Name *string `json:"name"` // name of the discovery bundle
Prefix *string `json:"prefix"` // path prefix for downloader
service string
path string
query string
}
// ParseConfig returns a valid Config object with defaults injected.
func ParseConfig(bs []byte, services []string) (*Config, error) {
if bs == nil {
return nil, nil
}
var result Config
if err := util.Unmarshal(bs, &result); err != nil {
return nil, err
}
return &result, result.validateAndInjectDefaults(services)
}
func (c *Config) validateAndInjectDefaults(services []string) error {
if c.Name == nil {
return fmt.Errorf("missing required discovery.name field")
}
if c.Prefix == nil {
s := defaultDiscoveryPathPrefix
c.Prefix = &s
}
if len(services) == 0 {
return fmt.Errorf("discovery requires exactly one service")
}
c.service = services[0]
c.path = fmt.Sprintf("%v/%v", strings.Trim(*c.Prefix, "/"), strings.Trim(*c.Name, "/"))
c.query = fmt.Sprintf("%v.%v", ast.DefaultRootDocument, strings.Replace(strings.Trim(*c.Name, "/"), "/", ".", -1))
return c.Config.ValidateAndInjectDefaults()
}
const (
defaultDiscoveryPathPrefix = "bundles"
defaultDiscoveryQueryPrefix = "data"
)
plugins/discovery/discovery.go
+375
View File
@@ -0,0 +1,375 @@
// 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 discovery implements configuration discovery.
package discovery
import (
"context"
"encoding/json"
"fmt"
"github.com/open-policy-agent/opa/ast"
bundleApi "github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/config"
"github.com/open-policy-agent/opa/download"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/bundle"
"github.com/open-policy-agent/opa/plugins/logs"
"github.com/open-policy-agent/opa/plugins/status"
"github.com/open-policy-agent/opa/rego"
"github.com/open-policy-agent/opa/storage/inmem"
"github.com/sirupsen/logrus"
)
// Discovery implements configuration discovery for OPA. When discovery is
// started it will periodically download a configuration bundle and try to
// reconfigure the OPA.
type Discovery struct {
manager *plugins.Manager
config *Config
initfuncs map[string]plugins.PluginInitFunc // factory functions for custom plugins
downloader *download.Downloader // discovery bundle downloader
status *bundle.Status // discovery status
etag string // discovery bundle etag for caching purposes
}
// CustomPlugins provides a set of factory functions to use for instantiating
// custom plugins.
func CustomPlugins(funcs map[string]plugins.PluginInitFunc) func(*Discovery) {
return func(d *Discovery) {
d.initfuncs = funcs
}
}
// New returns a new discovery plugin.
func New(manager *plugins.Manager, opts ...func(*Discovery)) (*Discovery, error) {
result := &Discovery{
manager: manager,
}
for _, f := range opts {
f(result)
}
config, err := ParseConfig(manager.Config.Discovery, manager.Services())
if err != nil {
return nil, err
} else if config == nil {
if _, err := getPluginSet(result.initfuncs, manager, manager.Config); err != nil {
return nil, err
}
return result, nil
}
if manager.Config.PluginsEnabled() {
return nil, fmt.Errorf("plugins cannot be specified in the bootstrap configuration when discovery enabled")
}
result.config = config
result.downloader = download.New(config.Config, manager.Client(config.service), config.path).WithCallback(result.oneShot)
result.status = &bundle.Status{
Name: *config.Name,
}
return result, nil
}
// Start starts the dynamic discovery process if configured.
func (c *Discovery) Start(ctx context.Context) error {
if c.downloader != nil {
c.downloader.Start(ctx)
}
return nil
}
// Stop stops the dynamic discovery process if configured.
func (c *Discovery) Stop(ctx context.Context) {
if c.downloader != nil {
c.downloader.Stop(ctx)
}
}
// Reconfigure is a no-op on discovery.
func (c *Discovery) Reconfigure(_ context.Context, _ interface{}) {
}
func (c *Discovery) oneShot(ctx context.Context, u download.Update) {
c.processUpdate(ctx, u)
if p := status.Lookup(c.manager); p != nil {
p.UpdateDiscoveryStatus(*c.status)
}
}
func (c *Discovery) processUpdate(ctx context.Context, u download.Update) {
if u.Error != nil {
c.logError("Discovery download failed: %v", u.Error)
c.status.SetError(u.Error)
return
}
if u.Bundle != nil {
c.status.SetDownloadSuccess()
if err := c.reconfigure(ctx, u); err != nil {
c.logError("Discovery reconfiguration error occurred: %v", err)
c.status.SetError(err)
return
}
c.status.SetError(nil)
c.status.SetActivateSuccess(u.Bundle.Manifest.Revision)
if u.ETag != "" {
c.logInfo("Discovery update processed successfully. Etag updated to %v.", u.ETag)
} else {
c.logInfo("Discovery update processed successfully.")
}
c.etag = u.ETag
return
}
if u.ETag == c.etag {
c.logError("Discovery update skipped, server replied with not modified.")
c.status.SetError(nil)
return
}
}
func (c *Discovery) reconfigure(ctx context.Context, u download.Update) error {
config, ps, err := processBundle(ctx, c.manager, c.initfuncs, u.Bundle, c.config.query)
if err != nil {
return err
}
if err := c.manager.Reconfigure(config); err != nil {
return err
}
// TODO(tsandall): we don't currently support changes to discovery
// configuration. These changes are risky because errors would be
// unrecoverable (without keeping track of changes and rolling back...)
// TODO(tsandall): add protection against discovery -service- changing.
for _, p := range ps.Start {
if err := p.Start(ctx); err != nil {
return err
}
}
for _, p := range ps.Reconfig {
p.Plugin.Reconfigure(ctx, p.Config)
}
return nil
}
func (c *Discovery) logError(fmt string, a ...interface{}) {
logrus.WithFields(c.logrusFields()).Errorf(fmt, a...)
}
func (c *Discovery) logInfo(fmt string, a ...interface{}) {
logrus.WithFields(c.logrusFields()).Infof(fmt, a...)
}
func (c *Discovery) logDebug(fmt string, a ...interface{}) {
logrus.WithFields(c.logrusFields()).Debugf(fmt, a...)
}
func (c *Discovery) logrusFields() logrus.Fields {
return logrus.Fields{
"name": *c.config.Name,
"plugin": "discovery",
}
}
func processBundle(ctx context.Context, manager *plugins.Manager, initfuncs map[string]plugins.PluginInitFunc, b *bundleApi.Bundle, query string) (*config.Config, *pluginSet, error) {
config, err := evaluateBundle(ctx, manager.ID, manager.Info, b, query)
if err != nil {
return nil, nil, err
}
ps, err := getPluginSet(initfuncs, manager, config)
return config, ps, err
}
func evaluateBundle(ctx context.Context, id string, info *ast.Term, b *bundleApi.Bundle, query string) (*config.Config, error) {
modules := map[string]*ast.Module{}
for _, file := range b.Modules {
modules[file.Path] = file.Parsed
}
compiler := ast.NewCompiler()
if compiler.Compile(modules); compiler.Failed() {
return nil, compiler.Errors
}
store := inmem.NewFromObject(b.Data)
rego := rego.New(
rego.Query(query),
rego.Compiler(compiler),
rego.Store(store),
rego.Runtime(info),
)
rs, err := rego.Eval(ctx)
if err != nil {
return nil, err
}
if len(rs) == 0 {
return nil, fmt.Errorf("undefined configuration")
}
bs, err := json.Marshal(rs[0].Expressions[0].Value)
if err != nil {
return nil, err
}
return config.ParseConfig(bs, id)
}
type pluginSet struct {
Start []plugins.Plugin
Reconfig []pluginreconfig
}
type pluginreconfig struct {
Config interface{}
Plugin plugins.Plugin
}
func getPluginSet(initfuncs map[string]plugins.PluginInitFunc, manager *plugins.Manager, config *config.Config) (*pluginSet, error) {
bundleConfig, err := bundle.ParseConfig(config.Bundle, manager.Services())
if err != nil {
return nil, err
}
decisionLogsConfig, err := logs.ParseConfig(config.DecisionLogs, manager.Services())
if err != nil {
return nil, err
}
statusConfig, err := status.ParseConfig(config.Status, manager.Services())
if err != nil {
return nil, err
}
starts := []plugins.Plugin{}
reconfigs := []pluginreconfig{}
if bundleConfig != nil {
p, created := getBundlePlugin(manager, bundleConfig)
if created {
starts = append(starts, p)
} else if p != nil {
reconfigs = append(reconfigs, pluginreconfig{bundleConfig, p})
}
}
if decisionLogsConfig != nil {
p, created := getDecisionLogsPlugin(manager, decisionLogsConfig)
if created {
starts = append(starts, p)
} else if p != nil {
reconfigs = append(reconfigs, pluginreconfig{decisionLogsConfig, p})
}
}
if statusConfig != nil {
p, created := getStatusPlugin(manager, statusConfig)
if created {
starts = append(starts, p)
} else if p != nil {
reconfigs = append(reconfigs, pluginreconfig{statusConfig, p})
}
}
result := &pluginSet{starts, reconfigs}
return result, getCustomPlugins(initfuncs, manager, config.Plugins, result)
}
func getBundlePlugin(m *plugins.Manager, config *bundle.Config) (plugin *bundle.Plugin, created bool) {
plugin = bundle.Lookup(m)
if plugin == nil {
plugin = bundle.New(config, m)
m.Register(bundle.Name, plugin)
registerBundleStatusUpdates(m)
created = true
}
return plugin, created
}
func getDecisionLogsPlugin(m *plugins.Manager, config *logs.Config) (plugin *logs.Plugin, created bool) {
plugin = logs.Lookup(m)
if plugin == nil {
plugin = logs.New(config, m)
m.Register(logs.Name, plugin)
created = true
}
return plugin, created
}
func getStatusPlugin(m *plugins.Manager, config *status.Config) (plugin *status.Plugin, created bool) {
plugin = status.Lookup(m)
if plugin == nil {
plugin = status.New(config, m)
m.Register(status.Name, plugin)
registerBundleStatusUpdates(m)
created = true
}
return plugin, created
}
func getCustomPlugins(initfuncs map[string]plugins.PluginInitFunc, manager *plugins.Manager, configs map[string]json.RawMessage, result *pluginSet) error {
for name, config := range configs {
plugin := manager.Plugin(name)
if plugin == nil {
// TODO: report missing plugin initfunc to controller...
if f, ok := initfuncs[name]; ok {
plugin, err := f(manager, config)
if err != nil {
return err
} else if plugin == nil {
return fmt.Errorf("plugin %q returned nil object", name)
}
manager.Register(name, plugin)
result.Start = append(result.Start, plugin)
}
} else {
result.Reconfig = append(result.Reconfig, pluginreconfig{config, plugin})
}
}
return nil
}
func registerBundleStatusUpdates(m *plugins.Manager) {
bp := bundle.Lookup(m)
sp := status.Lookup(m)
if bp == nil || sp == nil {
return
}
type pluginlistener string
bp.Register(pluginlistener(status.Name), func(s bundle.Status) {
sp.UpdateBundleStatus(s)
})
}
plugins/discovery/discovery_test.go
+361
View File
@@ -0,0 +1,361 @@
// 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 discovery
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"testing"
"time"
"github.com/open-policy-agent/opa/ast"
bundleApi "github.com/open-policy-agent/opa/bundle"
"github.com/open-policy-agent/opa/download"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/bundle"
"github.com/open-policy-agent/opa/plugins/status"
"github.com/open-policy-agent/opa/storage/inmem"
"github.com/open-policy-agent/opa/util"
)
func TestEvaluateBundle(t *testing.T) {
sampleModule := `
package foo.bar
bundle = {
"name": rt.name,
"service": "example"
} {
rt := opa.runtime()
}
`
b := &bundleApi.Bundle{
Manifest: bundleApi.Manifest{
Revision: "quickbrownfaux",
},
Data: map[string]interface{}{
"foo": map[string]interface{}{
"bar": map[string]interface{}{
"status": map[string]interface{}{},
},
},
},
Modules: []bundleApi.ModuleFile{
{
Path: `/example.rego`,
Raw: []byte(sampleModule),
Parsed: ast.MustParseModule(sampleModule),
},
},
}
info := ast.MustParseTerm(`{"name": "test/bundle1"}`)
config, err := evaluateBundle(context.Background(), "test-id", info, b, "data.foo.bar")
if err != nil {
t.Fatal(err)
}
if config.Bundle == nil {
t.Fatal("Expected a bundle configuration")
}
var parsedConfig bundle.Config
if err := util.Unmarshal(config.Bundle, &parsedConfig); err != nil {
t.Fatal("Unexpected error:", err)
}
expectedBundleConfig := bundle.Config{
Name: "test/bundle1",
Service: "example",
}
if !reflect.DeepEqual(expectedBundleConfig, parsedConfig) {
t.Fatalf("Expected bundle config %v, but got %v", expectedBundleConfig, parsedConfig)
}
}
func TestProcessBundle(t *testing.T) {
ctx := context.Background()
manager, err := plugins.New([]byte(`{
"services": {
"default": {
"url": "http://localhost:8181"
}
}
}`), "test-id", inmem.New())
if err != nil {
t.Fatal(err)
}
initialBundle := makeDataBundle(1, `
{
"config": {
"bundle": {"name": "test1"},
"status": {},
"decision_logs": {}
}
}
`)
_, ps, err := processBundle(ctx, manager, nil, initialBundle, "data.config")
if err != nil {
t.Fatal(err)
}
if len(ps.Start) != 3 || len(ps.Reconfig) != 0 {
t.Fatalf("Expected exactly three start events but got %v", ps)
}
updatedBundle := makeDataBundle(1, `
{
"config": {
"bundle": {"name": "test2"},
"status": {"partition_name": "foo"},
"decision_logs": {"partition_name": "bar"}
}
}
`)
_, ps, err = processBundle(ctx, manager, nil, updatedBundle, "data.config")
if err != nil {
t.Fatal(err)
}
if len(ps.Start) != 0 || len(ps.Reconfig) != 3 {
t.Fatalf("Expected exactly three start events but got %v", ps)
}
updatedBundle = makeDataBundle(2, `
{
"config": {
"bundle": {"service": "missing service name", "name": "test2"}
}
}
`)
_, _, err = processBundle(ctx, manager, nil, updatedBundle, "data.config")
if err == nil {
t.Fatal("Expected error but got success")
}
}
type reconfigureTestPlugin struct {
counts map[string]int
}
func (r *reconfigureTestPlugin) Start(context.Context) error {
r.counts["start"]++
return nil
}
func (r *reconfigureTestPlugin) Stop(context.Context) {
}
func (r *reconfigureTestPlugin) Reconfigure(_ context.Context, config interface{}) {
r.counts["reconfig"]++
}
func TestReconfigure(t *testing.T) {
manager, err := plugins.New([]byte(`{
"labels": {"x": "y"},
"services": {
"localhost": {
"url": "http://localhost:9999"
}
},
"discovery": {"name": "config"},
}`), "test-id", inmem.New())
if err != nil {
t.Fatal(err)
}
testPlugin := &reconfigureTestPlugin{counts: map[string]int{}}
disco, err := New(manager, CustomPlugins(map[string]plugins.PluginInitFunc{
"test_plugin": func(*plugins.Manager, []byte) (plugins.Plugin, error) {
return testPlugin, nil
},
}))
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
initialBundle := makeDataBundle(1, `
{
"config": {
"labels": {"x": "label value changed"},
"default_decision": "bar/baz",
"default_authorization_decision": "baz/qux",
"plugins": {
"test_plugin": {"a": "b"}
}
}
}
`)
disco.oneShot(ctx, download.Update{Bundle: initialBundle})
// Verify labels are unchanged
exp := map[string]string{"x": "y", "id": "test-id"}
if !reflect.DeepEqual(manager.Labels(), exp) {
t.Errorf("Expected labels to be unchanged (%v) but got %v", exp, manager.Labels())
}
// Verify decision ids set
expDecision := ast.MustParseTerm("data.bar.baz")
expAuthzDecision := ast.MustParseTerm("data.baz.qux")
if !manager.Config.DefaultDecisionRef().Equal(expDecision.Value) {
t.Errorf("Expected default decision to be %v but got %v", expDecision, manager.Config.DefaultDecisionRef())
}
if !manager.Config.DefaultAuthorizationDecisionRef().Equal(expAuthzDecision.Value) {
t.Errorf("Expected default authz decision to be %v but got %v", expAuthzDecision, manager.Config.DefaultAuthorizationDecisionRef())
}
// Verify plugins started
if !reflect.DeepEqual(testPlugin.counts, map[string]int{"start": 1}) {
t.Errorf("Expected exactly one plugin start but got %v", testPlugin)
}
// Verify plugins reconfigured
updatedBundle := makeDataBundle(2, `
{
"config": {
"labels": {"x": "label value changed"},
"default_decision": "bar/baz",
"default_authorization_decision": "baz/qux",
"plugins": {
"test_plugin": {"a": "plugin parameter value changed"}
}
}
}
`)
disco.oneShot(ctx, download.Update{Bundle: updatedBundle})
if !reflect.DeepEqual(testPlugin.counts, map[string]int{"start": 1, "reconfig": 1}) {
t.Errorf("Expected one plugin start and one reconfig but got %v", testPlugin)
}
}
type testServer struct {
t *testing.T
server *httptest.Server
updates []status.UpdateRequestV1
}
func (ts *testServer) Start() {
ts.server = httptest.NewServer(http.HandlerFunc(ts.handle))
}
func (ts *testServer) Stop() {
ts.server.Close()
}
func (ts *testServer) handle(w http.ResponseWriter, r *http.Request) {
var update status.UpdateRequestV1
if err := json.NewDecoder(r.Body).Decode(&update); err != nil {
ts.t.Fatal(err)
}
ts.updates = append(ts.updates, update)
w.WriteHeader(200)
}
func TestStatusUpdates(t *testing.T) {
ts := testServer{t: t}
ts.Start()
defer ts.Stop()
manager, err := plugins.New([]byte(fmt.Sprintf(`{
"labels": {"x": "y"},
"services": {
"localhost": {
"url": %q
}
},
"discovery": {"name": "config"},
}`, ts.server.URL)), "test-id", inmem.New())
if err != nil {
t.Fatal(err)
}
disco, err := New(manager)
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
// Enable status plugin which sends initial update.
disco.oneShot(ctx, download.Update{ETag: "etag-1", Bundle: makeDataBundle(1, `{
"config": {
"status": {}
}
}`)})
// Downloader error.
disco.oneShot(ctx, download.Update{Error: fmt.Errorf("unknown error")})
// Clear error.
disco.oneShot(ctx, download.Update{ETag: "etag-2", Bundle: makeDataBundle(2, `{
"config": {
"status": {}
}
}`)})
// Configuration error.
disco.oneShot(ctx, download.Update{ETag: "etag-3", Bundle: makeDataBundle(3, `{
"config": {
"status": {"service": "missing service"}
}
}`)})
// Clear error (last successful reconfigure).
disco.oneShot(ctx, download.Update{ETag: "etag-2"})
// Check that all updates were received and active revisions are expected.
var ok bool
t0 := time.Now()
for !ok && time.Since(t0) < time.Second {
ok = len(ts.updates) == 5 &&
ts.updates[0].Discovery.ActiveRevision == "test-revision-1" && ts.updates[0].Discovery.Code == "" &&
ts.updates[1].Discovery.ActiveRevision == "test-revision-1" && ts.updates[1].Discovery.Code == "bundle_error" &&
ts.updates[2].Discovery.ActiveRevision == "test-revision-2" && ts.updates[2].Discovery.Code == "" &&
ts.updates[3].Discovery.ActiveRevision == "test-revision-2" && ts.updates[3].Discovery.Code == "bundle_error" &&
ts.updates[4].Discovery.ActiveRevision == "test-revision-2" && ts.updates[4].Discovery.Code == ""
}
if !ok {
t.Fatalf("Did not receive expected updates before timeout expired. Received: %+v", ts.updates)
}
}
func makeDataBundle(n int, s string) *bundleApi.Bundle {
return &bundleApi.Bundle{
Manifest: bundleApi.Manifest{Revision: fmt.Sprintf("test-revision-%v", n)},
Data: util.MustUnmarshalJSON([]byte(s)).(map[string]interface{}),
}
}
plugins/logs/plugin.go
+35 -41
View File
@@ -10,6 +10,7 @@ import (
"fmt"
"math/rand"
"net/http"
"reflect"
"strings"
"sync"
"time"
@@ -119,35 +120,6 @@ func (c *Config) validateAndInjectDefaults(services []string) error {
return nil
}
func (c *Config) equal(other Config) bool {
if c.Service != other.Service {
return false
}
if c.PartitionName != other.PartitionName {
return false
}
if *c.Reporting.MaxDelaySeconds != *other.Reporting.MaxDelaySeconds {
return false
}
if *c.Reporting.MinDelaySeconds != *other.Reporting.MinDelaySeconds {
return false
}
if *c.Reporting.UploadSizeLimitBytes != *other.Reporting.UploadSizeLimitBytes {
return false
}
if *c.Reporting.BufferSizeLimitBytes != *other.Reporting.BufferSizeLimitBytes {
return false
}
return true
}
// Plugin implements decision log buffering and uploading.
type Plugin struct {
manager *plugins.Manager
@@ -161,6 +133,11 @@ type Plugin struct {
// ParseConfig validates the config and injects default values.
func ParseConfig(config []byte, services []string) (*Config, error) {
if config == nil {
return nil, nil
}
var parsedConfig Config
if err := util.Unmarshal(config, &parsedConfig); err != nil {
@@ -175,7 +152,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, error) {
func New(parsedConfig *Config, manager *plugins.Manager) *Plugin {
plugin := &Plugin{
manager: manager,
@@ -186,17 +163,30 @@ func New(parsedConfig *Config, manager *plugins.Manager) (*Plugin, error) {
reconfig: make(chan interface{}),
}
return plugin, nil
return plugin
}
// Name identifies the plugin on manager.
const Name = "decision_logs"
// Lookup returns the decision logs plugin registered with the manager.
func Lookup(manager *plugins.Manager) *Plugin {
if p := manager.Plugin(Name); p != nil {
return p.(*Plugin)
}
return nil
}
// Start starts the plugin.
func (p *Plugin) Start(ctx context.Context) error {
p.logInfo("Starting decision log uploader.")
go p.loop()
return nil
}
// Stop stops the plugin.
func (p *Plugin) Stop(ctx context.Context) {
p.logInfo("Stopping decision log uploader.")
done := make(chan struct{})
p.stop <- done
_ = <-done
@@ -207,7 +197,7 @@ func (p *Plugin) Log(ctx context.Context, decision *server.Info) {
path := strings.Replace(strings.TrimPrefix(decision.Query, "data."), ".", "/", -1)
event := EventV1{
Labels: p.manager.Labels,
Labels: p.manager.Labels(),
DecisionID: decision.DecisionID,
Revision: decision.Revision,
Path: path,
@@ -232,15 +222,10 @@ func (p *Plugin) Log(ctx context.Context, decision *server.Info) {
}
// Reconfigure notifies the plugin with a new configuration.
func (p *Plugin) Reconfigure(config interface{}) {
func (p *Plugin) Reconfigure(_ context.Context, config interface{}) {
p.reconfig <- config
}
// Equal checks if the current and provided input config are equal.
func (p *Plugin) Equal(other *Config) bool {
return p.config.equal(*other)
}
func (p *Plugin) loop() {
ctx, cancel := context.WithCancel(context.Background())
@@ -328,8 +313,17 @@ func (p *Plugin) oneShot(ctx context.Context) (ok bool, err error) {
return true, nil
}
func (p *Plugin) reconfigure(newConfig interface{}) {
p.config = *newConfig.(*Config)
func (p *Plugin) reconfigure(config interface{}) {
newConfig := config.(*Config)
if reflect.DeepEqual(p.config, *newConfig) {
p.logDebug("Decision log uploader configuration unchanged.")
return
}
p.logInfo("Decision log uploader configuration changed.")
p.config = *newConfig
}
func (p *Plugin) bufferChunk(buffer *logBuffer, bs []byte) {
@@ -379,6 +373,6 @@ func (p *Plugin) logDebug(fmt string, a ...interface{}) {
func (p *Plugin) logrusFields() logrus.Fields {
return logrus.Fields{
"plugin": "decision_logs",
"plugin": Name,
}
}
plugins/logs/plugin_test.go
+2 -5
View File
@@ -281,7 +281,7 @@ func TestPluginReconfigure(t *testing.T) {
config, _ := ParseConfig(pluginConfig, fixture.manager.Services())
fixture.plugin.Reconfigure(config)
fixture.plugin.Reconfigure(ctx, config)
fixture.plugin.Stop(ctx)
actualMin := time.Duration(*fixture.plugin.config.Reporting.MinDelaySeconds) / time.Nanosecond
@@ -342,10 +342,7 @@ func newTestFixture(t *testing.T) testFixture {
config, _ := ParseConfig([]byte(pluginConfig), manager.Services())
p, err := New(config, manager)
if err != nil {
t.Fatal(err)
}
p := New(config, manager)
return testFixture{
manager: manager,
plugins/plugins.go
+109 -66
View File
@@ -11,6 +11,7 @@ import (
"sync"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/config"
"github.com/open-policy-agent/opa/plugins/rest"
"github.com/open-policy-agent/opa/storage"
"github.com/open-policy-agent/opa/util"
@@ -20,63 +21,110 @@ import (
type Plugin interface {
Start(ctx context.Context) error
Stop(ctx context.Context)
Reconfigure(config interface{})
Reconfigure(ctx context.Context, config interface{})
}
// PluginInitFunc defines the interface for the constructing plugins from configuration.
// The function will be called with the plugin manager (which provides access to OPA's storage layer, compiler, and service clients) and the configuration for the plugin itself.
// PluginInitFunc defines the interface for the constructing plugins from
// configuration. The function will be called with the plugin manager (which
// provides access to OPA's storage layer, compiler, and service clients) and
// the configuration for the plugin itself.
type PluginInitFunc func(m *Manager, config []byte) (Plugin, error)
// Manager implements lifecycle management of plugins and gives plugins access
// to engine-wide components like storage.
type Manager struct {
Labels map[string]string
Store storage.Store
compiler *ast.Compiler
services map[string]rest.Client
plugins []Plugin
registeredTriggers []func(txn storage.Transaction)
registeredTriggersMux sync.Mutex
compilerMux sync.RWMutex
labelSvcMux sync.Mutex
Store storage.Store
Config *config.Config
Info *ast.Term
ID string
compiler *ast.Compiler
compilerMux sync.RWMutex
services map[string]rest.Client
plugins []namedplugin
registeredTriggers []func(txn storage.Transaction)
mtx sync.Mutex
}
type namedplugin struct {
name string
plugin Plugin
}
// Info sets the runtime information on the manager. The runtime information is
// propagated to opa.runtime() built-in function calls.
func Info(term *ast.Term) func(*Manager) {
return func(m *Manager) {
m.Info = term
}
}
// New creates a new Manager using config.
func New(config []byte, id string, store storage.Store) (*Manager, error) {
func New(raw []byte, id string, store storage.Store, opts ...func(*Manager)) (*Manager, error) {
var parsedConfig struct {
Services json.RawMessage `json:"services"`
Labels map[string]string `json:"labels"`
}
if err := util.Unmarshal(config, &parsedConfig); err != nil {
parsedConfig, err := config.ParseConfig(raw, id)
if err != nil {
return nil, err
}
if parsedConfig.Labels == nil {
parsedConfig.Labels = map[string]string{}
}
services, err := parseServicesConfig(parsedConfig.Services)
if err != nil {
return nil, err
}
parsedConfig.Labels["id"] = id
m := &Manager{
Labels: parsedConfig.Labels,
Store: store,
Config: parsedConfig,
ID: id,
services: services,
}
for _, f := range opts {
f(m)
}
return m, nil
}
// Labels returns the set of labels from the configuration.
func (m *Manager) Labels() map[string]string {
m.mtx.Lock()
defer m.mtx.Unlock()
return m.Config.Labels
}
// Register adds a plugin to the manager. When the manager is started, all of
// the plugins will be started.
func (m *Manager) Register(plugin Plugin) {
m.plugins = append(m.plugins, plugin)
func (m *Manager) Register(name string, plugin Plugin) {
m.mtx.Lock()
defer m.mtx.Unlock()
m.plugins = append(m.plugins, namedplugin{
name: name,
plugin: plugin,
})
}
// Plugins returns the list of plugins registered with the manager.
func (m *Manager) Plugins() []string {
m.mtx.Lock()
defer m.mtx.Unlock()
result := make([]string, len(m.plugins))
for i := range m.plugins {
result[i] = m.plugins[i].name
}
return result
}
// Plugin returns the plugin registered with name or nil if name is not found.
func (m *Manager) Plugin(name string) Plugin {
m.mtx.Lock()
defer m.mtx.Unlock()
for i := range m.plugins {
if m.plugins[i].name == name {
return m.plugins[i].plugin
}
}
return nil
}
// GetCompiler returns the manager's compiler.
@@ -86,38 +134,6 @@ func (m *Manager) GetCompiler() *ast.Compiler {
return m.compiler
}
// Update updates the manager's services and labels.
func (m *Manager) Update(config []byte) error {
m.labelSvcMux.Lock()
defer m.labelSvcMux.Unlock()
var parsedConfig struct {
Services json.RawMessage `json:"services"`
Labels map[string]string `json:"labels"`
}
if err := util.Unmarshal(config, &parsedConfig); err != nil {
return err
}
if parsedConfig.Labels != nil {
for k, v := range parsedConfig.Labels {
m.Labels[k] = v
}
}
services, err := parseServicesConfig(parsedConfig.Services)
if err != nil {
return err
}
for k, v := range services {
m.services[k] = v
}
return nil
}
func (m *Manager) setCompiler(compiler *ast.Compiler) {
m.compilerMux.Lock()
defer m.compilerMux.Unlock()
@@ -127,8 +143,8 @@ func (m *Manager) setCompiler(compiler *ast.Compiler) {
// RegisterCompilerTrigger registers for change notifications when the compiler
// is changed.
func (m *Manager) RegisterCompilerTrigger(f func(txn storage.Transaction)) {
m.registeredTriggersMux.Lock()
defer m.registeredTriggersMux.Unlock()
m.mtx.Lock()
defer m.mtx.Unlock()
m.registeredTriggers = append(m.registeredTriggers, f)
}
@@ -151,10 +167,19 @@ func (m *Manager) Start(ctx context.Context) error {
return err
}
for _, p := range m.plugins {
if err := p.Start(ctx); err != nil {
return err
if err := func() error {
m.mtx.Lock()
defer m.mtx.Unlock()
for _, p := range m.plugins {
if err := p.plugin.Start(ctx); err != nil {
return err
}
}
return nil
}(); err != nil {
return err
}
config := storage.TriggerConfig{OnCommit: m.onCommit}
@@ -167,11 +192,29 @@ func (m *Manager) Start(ctx context.Context) error {
// Stop stops the manager, stopping all the plugins registered with it
func (m *Manager) Stop(ctx context.Context) {
m.mtx.Lock()
defer m.mtx.Unlock()
for _, p := range m.plugins {
p.Stop(ctx)
p.plugin.Stop(ctx)
}
}
// Reconfigure updates the configuration on the manager.
func (m *Manager) Reconfigure(config *config.Config) error {
services, err := parseServicesConfig(config.Services)
if err != nil {
return err
}
m.mtx.Lock()
defer m.mtx.Unlock()
config.Labels = m.Config.Labels // don't overwrite labels
m.Config = config
for name, client := range services {
m.services[name] = client
}
return nil
}
func (m *Manager) onCommit(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) {
if event.PolicyChanged() {
compiler, _ := loadCompilerFromStore(ctx, m.Store, txn)
plugins/plugins_test.go
-85
View File
@@ -1,85 +0,0 @@
// 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 plugins
import (
"fmt"
"reflect"
"testing"
"github.com/open-policy-agent/opa/storage/inmem"
)
func TestMangerUpdate(t *testing.T) {
managerConfig := []byte(fmt.Sprintf(`{
"services": [
{
"name": "example",
"url": "example.com",
"credentials": {
"bearer": {
"scheme": "Bearer",
"token": "secret"
}
}
}
],
"labels": {
"app": "myapp",
"environment": "production"
}}`))
store := inmem.New()
manager, err := New(managerConfig, "test-instance-id", store)
if err != nil {
t.Fatal(err)
}
updatedConfig := []byte(fmt.Sprintf(`{
"services": [
{
"name": "example",
"url": "example.io",
"credentials": {
"bearer": {
"scheme": "Bearer",
"token": "secret"
}
}
},
{
"name": "blah",
"url": "blah.com"
}
],
"labels": {
"app": "myapp",
"region": "west",
"environment": "dev"
}}`))
err = manager.Update(updatedConfig)
if err != nil {
t.Fatal(err)
}
expectedLabels := map[string]string{}
expectedLabels["id"] = "test-instance-id"
expectedLabels["app"] = "myapp"
expectedLabels["region"] = "west"
expectedLabels["environment"] = "dev"
expectedServices := []string{"example", "blah"}
if !reflect.DeepEqual(expectedLabels, manager.Labels) {
t.Fatalf("Expected labels %v, but got %v", expectedLabels, manager.Labels)
}
if len(expectedServices) != len(manager.Services()) {
t.Fatalf("Expected services %v, but got %v", expectedServices, manager.Services())
}
}
plugins/status/plugin.go
+68 -47
View File
@@ -9,6 +9,7 @@ import (
"context"
"fmt"
"net/http"
"reflect"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/bundle"
@@ -21,17 +22,20 @@ import (
// remote HTTP endpoints.
type UpdateRequestV1 struct {
Labels map[string]string `json:"labels"`
Bundle bundle.Status `json:"bundle"`
Discovery bundle.Status `json:"discovery"`
Bundle *bundle.Status `json:"bundle,omitempty"`
Discovery *bundle.Status `json:"discovery,omitempty"`
}
// Plugin implements status reporting. Updates can be triggered by the caller.
type Plugin struct {
manager *plugins.Manager
config Config
update chan bundle.Status
stop chan chan struct{}
reconfig chan interface{}
manager *plugins.Manager
config Config
bundleCh chan bundle.Status
lastBundleStatus *bundle.Status
discoCh chan bundle.Status
lastDiscoStatus *bundle.Status
stop chan chan struct{}
reconfig chan interface{}
}
// Config contains configuration for the plugin.
@@ -62,21 +66,13 @@ func (c *Config) validateAndInjectDefaults(services []string) error {
return nil
}
func (c *Config) equal(other Config) bool {
if c.Service != other.Service {
return false
}
if c.PartitionName != other.PartitionName {
return false
}
return true
}
// ParseConfig validates the config and injects default values.
func ParseConfig(config []byte, services []string) (*Config, error) {
if config == nil {
return nil, nil
}
var parsedConfig Config
if err := util.Unmarshal(config, &parsedConfig); err != nil {
@@ -91,65 +87,80 @@ 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, error) {
func New(parsedConfig *Config, manager *plugins.Manager) *Plugin {
plugin := &Plugin{
manager: manager,
config: *parsedConfig,
update: make(chan bundle.Status),
bundleCh: make(chan bundle.Status),
discoCh: make(chan bundle.Status),
stop: make(chan chan struct{}),
reconfig: make(chan interface{}),
}
return plugin, nil
return plugin
}
// Name identifies the plugin on manager.
const Name = "status"
// Lookup returns the status plugin registered with the manager.
func Lookup(manager *plugins.Manager) *Plugin {
if p := manager.Plugin(Name); p != nil {
return p.(*Plugin)
}
return nil
}
// Start starts the plugin.
func (p *Plugin) Start(ctx context.Context) error {
p.logInfo("Starting status reporter.")
go p.loop()
return nil
}
// Stop stops the plugin.
func (p *Plugin) Stop(ctx context.Context) {
p.logInfo("Stopping status reporter.")
done := make(chan struct{})
p.stop <- done
_ = <-done
}
// UpdateBundleStatus notifies the plugin with a new bundle plugin status.
// UpdateBundleStatus notifies the plugin that the policy bundle was updated.
func (p *Plugin) UpdateBundleStatus(status bundle.Status) {
p.update <- status
p.bundleCh <- status
}
// UpdateDiscoveryStatus notifies the plugin with a new discovery plugin status.
// UpdateDiscoveryStatus notifies the plugin that the discovery bundle was updated.
func (p *Plugin) UpdateDiscoveryStatus(status bundle.Status) {
status.DiscoveryStatus = true
p.update <- status
p.discoCh <- status
}
// Reconfigure notifies the plugin with a new configuration.
func (p *Plugin) Reconfigure(config interface{}) {
func (p *Plugin) Reconfigure(_ context.Context, config interface{}) {
p.reconfig <- config
}
// Equal checks if the current and provided input config are equal.
func (p *Plugin) Equal(other *Config) bool {
return p.config.equal(*other)
}
func (p *Plugin) loop() {
ctx, cancel := context.WithCancel(context.Background())
for {
select {
case status := <-p.update:
err := p.oneShot(ctx, status)
case status := <-p.bundleCh:
err := p.oneShot(ctx, false, status)
if err != nil {
p.logError("%v.", err)
} else {
p.logInfo("Status update sent successfully.")
p.logInfo("Status update sent successfully in response to bundle update.")
}
case status := <-p.discoCh:
err := p.oneShot(ctx, true, status)
if err != nil {
p.logError("%v.", err)
} else {
p.logInfo("Status update sent successfully in response to discovery update.")
}
case newConfig := <-p.reconfig:
@@ -163,16 +174,18 @@ func (p *Plugin) loop() {
}
}
func (p *Plugin) oneShot(ctx context.Context, status bundle.Status) error {
func (p *Plugin) oneShot(ctx context.Context, disco bool, status bundle.Status) error {
req := UpdateRequestV1{
Labels: p.manager.Labels,
if disco {
p.lastDiscoStatus = &status
} else {
p.lastBundleStatus = &status
}
if status.DiscoveryStatus {
req.Discovery = status
} else {
req.Bundle = status
req := UpdateRequestV1{
Labels: p.manager.Labels(),
Discovery: p.lastDiscoStatus,
Bundle: p.lastBundleStatus,
}
resp, err := p.manager.Client(p.config.Service).
@@ -197,8 +210,16 @@ func (p *Plugin) oneShot(ctx context.Context, status bundle.Status) error {
}
}
func (p *Plugin) reconfigure(newConfig interface{}) {
p.config = *newConfig.(*Config)
func (p *Plugin) reconfigure(config interface{}) {
newConfig := config.(*Config)
if reflect.DeepEqual(p.config, *newConfig) {
p.logDebug("Status reporter configuration unchanged.")
return
}
p.logInfo("Status reporter configuration changed.")
p.config = *newConfig
}
func (p *Plugin) logError(fmt string, a ...interface{}) {
@@ -215,6 +236,6 @@ func (p *Plugin) logDebug(fmt string, a ...interface{}) {
func (p *Plugin) logrusFields() logrus.Fields {
return logrus.Fields{
"plugin": "status",
"plugin": Name,
}
}
plugins/status/plugin_test.go
+9 -13
View File
@@ -32,7 +32,7 @@ func TestPluginStart(t *testing.T) {
status := testStatus()
fixture.plugin.UpdateBundleStatus(status)
fixture.plugin.UpdateBundleStatus(*status)
result := <-fixture.server.ch
exp := UpdateRequestV1{
@@ -61,10 +61,9 @@ func TestPluginStartDiscovery(t *testing.T) {
status := testStatus()
fixture.plugin.UpdateDiscoveryStatus(status)
fixture.plugin.UpdateDiscoveryStatus(*status)
result := <-fixture.server.ch
status.DiscoveryStatus = true
exp := UpdateRequestV1{
Labels: map[string]string{
"id": "test-instance-id",
@@ -83,7 +82,7 @@ func TestPluginBadAuth(t *testing.T) {
ctx := context.Background()
fixture.server.expCode = 401
defer fixture.server.stop()
err := fixture.plugin.oneShot(ctx, bundle.Status{})
err := fixture.plugin.oneShot(ctx, false, bundle.Status{})
if err == nil {
t.Fatal("Expected error")
}
@@ -94,7 +93,7 @@ func TestPluginBadPath(t *testing.T) {
ctx := context.Background()
fixture.server.expCode = 404
defer fixture.server.stop()
err := fixture.plugin.oneShot(ctx, bundle.Status{})
err := fixture.plugin.oneShot(ctx, false, bundle.Status{})
if err == nil {
t.Fatal("Expected error")
}
@@ -105,7 +104,7 @@ func TestPluginBadStatus(t *testing.T) {
ctx := context.Background()
fixture.server.expCode = 500
defer fixture.server.stop()
err := fixture.plugin.oneShot(ctx, bundle.Status{})
err := fixture.plugin.oneShot(ctx, false, bundle.Status{})
if err == nil {
t.Fatal("Expected error")
}
@@ -127,7 +126,7 @@ func TestPluginReconfigure(t *testing.T) {
config, _ := ParseConfig(pluginConfig, fixture.manager.Services())
fixture.plugin.Reconfigure(config)
fixture.plugin.Reconfigure(ctx, config)
fixture.plugin.Stop(ctx)
if fixture.plugin.config.PartitionName != "test" {
@@ -178,10 +177,7 @@ func newTestFixture(t *testing.T) testFixture {
config, _ := ParseConfig([]byte(pluginConfig), manager.Services())
p, err := New(config, manager)
if err != nil {
t.Fatal(err)
}
p := New(config, manager)
return testFixture{
manager: manager,
@@ -221,7 +217,7 @@ func (t *testServer) stop() {
t.server.Close()
}
func testStatus() bundle.Status {
func testStatus() *bundle.Status {
tDownload, _ := time.Parse("2018-01-01T00:00:00.0000000Z", time.RFC3339Nano)
tActivate, _ := time.Parse("2018-01-01T00:00:01.0000000Z", time.RFC3339Nano)
@@ -233,5 +229,5 @@ func testStatus() bundle.Status {
LastSuccessfulActivation: tActivate,
}
return status
return &status
}
runtime/runtime.go
+57 -51
View File
@@ -11,15 +11,16 @@ import (
"crypto/tls"
"fmt"
"io"
"io/ioutil"
"os"
"sync"
"time"
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/discovery"
"github.com/open-policy-agent/opa/internal/runtime"
"github.com/open-policy-agent/opa/loader"
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/plugins/discovery"
"github.com/open-policy-agent/opa/plugins/logs"
"github.com/open-policy-agent/opa/repl"
"github.com/open-policy-agent/opa/server"
@@ -127,14 +128,13 @@ func NewParams() Params {
// Runtime represents a single OPA instance.
type Runtime struct {
Params Params
Store storage.Store
Discovery *discovery.Discovery
Params Params
Store storage.Store
Manager *plugins.Manager
info *ast.Term // runtime information provided to evaluation engine
defaultDecision ast.Ref
defaultAuthorizationDecision ast.Ref
decisionLogger func(context.Context, *server.Info)
// TODO(tsandall): remove this field since it's available on the manager
// and doesn't have to duplicated here or on the server.
info *ast.Term // runtime information provided to evaluation engine
}
// NewRuntime returns a new Runtime object initialized with params.
@@ -174,34 +174,37 @@ func NewRuntime(ctx context.Context, params Params) (*Runtime, error) {
return nil, errors.Wrapf(err, "storage error")
}
cfg, err := loadConfig(ctx, params.ID, store, params.ConfigFile)
if err != nil {
return nil, err
}
var bs []byte
var decisionLogger func(context.Context, *server.Info)
if p, ok := cfg.Plugins["decision_logs"]; ok {
decisionLogger = p.(*logs.Plugin).Log
if params.DecisionIDFactory == nil {
params.DecisionIDFactory = generateDecisionID
if params.ConfigFile != "" {
bs, err = ioutil.ReadFile(params.ConfigFile)
if err != nil {
return nil, errors.Wrapf(err, "config error")
}
}
info, err := runtime.Term(runtime.Params{ConfigFile: params.ConfigFile})
info, err := runtime.Term(runtime.Params{Config: bs})
if err != nil {
return nil, err
}
manager, err := plugins.New(bs, params.ID, store, plugins.Info(info))
if err != nil {
return nil, errors.Wrapf(err, "config error")
}
disco, err := discovery.New(manager, discovery.CustomPlugins(registeredPlugins))
if err != nil {
return nil, errors.Wrapf(err, "config error")
}
manager.Register("discovery", disco)
rt := &Runtime{
Store: store,
Params: params,
info: info,
defaultDecision: cfg.DefaultDecision,
defaultAuthorizationDecision: cfg.DefaultAuthorizationDecision,
decisionLogger: decisionLogger,
Discovery: cfg,
Store: store,
Params: params,
Manager: manager,
info: info,
}
return rt, nil
@@ -217,14 +220,15 @@ func (rt *Runtime) StartServer(ctx context.Context) {
"insecure_addr": rt.Params.InsecureAddr,
}).Infof("First line of log stream.")
if err := rt.Discovery.Start(ctx); err != nil {
logrus.WithField("err", err).Fatalf("Unable to initialize plugins.")
if err := rt.Manager.Start(ctx); err != nil {
logrus.WithField("err", err).Fatalf("Failed to start plugins.")
}
defer rt.Discovery.Stop(ctx)
defer rt.Manager.Stop(ctx)
s, err := server.New().
WithStore(rt.Store).
WithManager(rt.Discovery.Manager).
WithManager(rt.Manager).
WithCompilerErrorLimit(rt.Params.ErrorLimit).
WithAddresses(*rt.Params.Addrs).
WithInsecureAddress(rt.Params.InsecureAddr).
@@ -232,11 +236,9 @@ func (rt *Runtime) StartServer(ctx context.Context) {
WithAuthentication(rt.Params.Authentication).
WithAuthorization(rt.Params.Authorization).
WithDiagnosticsBuffer(rt.Params.DiagnosticsBuffer).
WithDecisionIDFactory(rt.Params.DecisionIDFactory).
WithDecisionIDFactory(rt.decisionIDFactory).
WithDecisionLogger(rt.decisionLogger).
WithRuntime(rt.info).
WithDefaultDecision(rt.defaultDecision).
WithDefaultAuthorizationDecision(rt.defaultAuthorizationDecision).
Init(ctx)
if err != nil {
@@ -273,11 +275,12 @@ func (rt *Runtime) StartServer(ctx context.Context) {
// StartREPL starts the runtime in REPL mode. This function will block the calling goroutine.
func (rt *Runtime) StartREPL(ctx context.Context) {
if err := rt.Discovery.Start(ctx); err != nil {
if err := rt.Manager.Start(ctx); err != nil {
fmt.Fprintln(rt.Params.Output, "error starting plugins:", err)
os.Exit(1)
}
defer rt.Discovery.Stop(ctx)
defer rt.Manager.Stop(ctx)
banner := rt.getBanner()
repl := repl.New(rt.Store, rt.Params.HistoryPath, rt.Params.Output, rt.Params.OutputFormat, rt.Params.ErrorLimit, banner).WithRuntime(rt.info)
@@ -292,6 +295,24 @@ func (rt *Runtime) StartREPL(ctx context.Context) {
repl.Loop(ctx)
}
func (rt *Runtime) decisionIDFactory() string {
if rt.Params.DecisionIDFactory != nil {
return rt.Params.DecisionIDFactory()
}
if logs.Lookup(rt.Manager) != nil {
return generateDecisionID()
}
return ""
}
func (rt *Runtime) decisionLogger(ctx context.Context, event *server.Info) {
plugin := logs.Lookup(rt.Manager)
if plugin == nil {
return
}
plugin.Log(ctx, event)
}
func (rt *Runtime) startWatcher(ctx context.Context, paths []string, onReload func(time.Duration, error)) error {
watcher, err := getWatcher(paths)
if err != nil {
@@ -487,21 +508,6 @@ func setupLogging(config LoggingConfig) {
logrus.SetLevel(lvl)
}
func loadConfig(ctx context.Context, id string, store storage.Store, configFile string) (*discovery.Discovery, error) {
params := discovery.Params{
ID: id,
ConfigFile: configFile,
Store: store,
RegisteredPlugins: registeredPlugins,
}
discoveredConfig, err := discovery.New(ctx, params)
if err != nil {
return nil, err
}
return discoveredConfig, nil
}
func generateInstanceID() (string, error) {
return uuid4()
}
server/authorizer/authorizer.go
+4 -4
View File
@@ -24,7 +24,7 @@ type Basic struct {
compiler func() *ast.Compiler
store storage.Store
runtime *ast.Term
decision string
decision func() ast.Ref
}
// Runtime returns an argument that sets the runtime on the authorizer.
@@ -36,9 +36,9 @@ func Runtime(term *ast.Term) func(*Basic) {
// Decision returns an argument that sets the path of the authorization decision
// to query.
func Decision(ref ast.Ref) func(*Basic) {
func Decision(ref func() ast.Ref) func(*Basic) {
return func(b *Basic) {
b.decision = ref.String()
b.decision = ref
}
}
@@ -66,7 +66,7 @@ func (h *Basic) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
rego := rego.New(
rego.Query(h.decision),
rego.Query(h.decision().String()),
rego.Compiler(h.compiler()),
rego.Store(h.store),
rego.Input(input),
server/authorizer/authorizer_test.go
+3 -1
View File
@@ -163,7 +163,9 @@ func TestBasic(t *testing.T) {
req = identifier.SetIdentity(req, tc.identity)
}
NewBasic(&mockHandler{}, compiler, store, Decision(ast.MustParseRef("data.system.authz.allow"))).ServeHTTP(recorder, req)
NewBasic(&mockHandler{}, compiler, store, Decision(func() ast.Ref {
return ast.MustParseRef("data.system.authz.allow")
})).ServeHTTP(recorder, req)
if recorder.Code != tc.expectedStatus {
t.Fatalf("Expected status code %v but got: %v", tc.expectedStatus, recorder)
server/server.go
+19 -35
View File
@@ -80,25 +80,23 @@ var unsafeBuiltinsMap = map[string]bool{ast.HTTPSend.Name: true}
type Server struct {
Handler http.Handler
router *mux.Router
addrs []string
insecureAddr string
authentication AuthenticationScheme
authorization AuthorizationScheme
cert *tls.Certificate
mtx sync.RWMutex
partials map[string]rego.PartialResult
store storage.Store
manager *plugins.Manager
watcher *watch.Watcher
decisionIDFactory func() string
diagnostics Buffer
revision string
logger func(context.Context, *Info)
errLimit int
runtime *ast.Term
defaultDecision ast.Ref
defaultAuthorizationDecision ast.Ref
router *mux.Router
addrs []string
insecureAddr string
authentication AuthenticationScheme
authorization AuthorizationScheme
cert *tls.Certificate
mtx sync.RWMutex
partials map[string]rego.PartialResult
store storage.Store
manager *plugins.Manager
watcher *watch.Watcher
decisionIDFactory func() string
diagnostics Buffer
revision string
logger func(context.Context, *Info)
errLimit int
runtime *ast.Term
}
// Loop will contain all the calls from the server that we'll be listening on.
@@ -124,7 +122,7 @@ func (s *Server) Init(ctx context.Context) (*Server, error) {
s.getCompiler,
s.store,
authorizer.Runtime(s.runtime),
authorizer.Decision(s.defaultAuthorizationDecision))
authorizer.Decision(s.manager.Config.DefaultAuthorizationDecisionRef))
}
switch s.authentication {
@@ -232,20 +230,6 @@ func (s *Server) WithRuntime(term *ast.Term) *Server {
return s
}
// WithDefaultDecision sets path of the policy decision to query to serve
// requests with an empty URL path.
func (s *Server) WithDefaultDecision(ref ast.Ref) *Server {
s.defaultDecision = ref
return s
}
// WithDefaultAuthorizationDecision sets path of the policy decision to query to
// authorize requests to OPA itself.
func (s *Server) WithDefaultAuthorizationDecision(ref ast.Ref) *Server {
s.defaultAuthorizationDecision = ref
return s
}
// WithRouter sets the mux.Router to attach OPA's HTTP API routes onto. If a
// router is not supplied, the server will create it's own.
func (s *Server) WithRouter(router *mux.Router) *Server {
@@ -546,7 +530,7 @@ func (s *Server) migrateWatcher(txn storage.Transaction) {
}
func (s *Server) unversionedPost(w http.ResponseWriter, r *http.Request) {
s.v0QueryPath(w, r, s.defaultDecision)
s.v0QueryPath(w, r, s.manager.Config.DefaultDecisionRef())
}
func (s *Server) v0DataPost(w http.ResponseWriter, r *http.Request) {
server/server_test.go
-6
View File
@@ -1814,8 +1814,6 @@ func TestDiagnostics(t *testing.T) {
WithStore(f.server.store).
WithManager(f.server.manager).
WithDiagnosticsBuffer(NewBoundedBuffer(8)).
WithDefaultDecision(ast.MustParseRef("data.system.main")).
WithDefaultAuthorizationDecision(ast.MustParseRef("data.system.authz.allow")).
Init(context.Background())
queriesOnly := `package system.diagnostics
@@ -2456,8 +2454,6 @@ func TestAuthorization(t *testing.T) {
WithStore(store).
WithManager(m).
WithAuthorization(AuthorizationBasic).
WithDefaultDecision(ast.MustParseRef("data.system.main")).
WithDefaultAuthorizationDecision(ast.MustParseRef("data.system.authz.allow")).
Init(ctx)
if err != nil {
@@ -2648,8 +2644,6 @@ func newFixture(t *testing.T) *fixture {
WithAddresses([]string{":8182"}).
WithStore(store).
WithManager(m).
WithDefaultDecision(ast.MustParseRef("data.system.main")).
WithDefaultAuthorizationDecision(ast.MustParseRef("data.system.authz.allow")).
Init(ctx)
if err != nil {
panic(err)