ewilliams28/releases
1
0
Fork 0
mirror of https://github.com/open-policy-agent/opa.git synced 2026-08-13 03:42:35 -06:00
Code Issues Packages Projects Releases Wiki Activity

Add support for custom backends in decision logger

Browse Source 
These changes modify the decision logger to support custom
backends. The docs now include an example of how to implement a custom
backend that writes out to stdout.

Signed-off-by: Torin Sandall <torinsandall@gmail.com>
This commit is contained in:
Torin Sandall
2019-01-09 10:55:57 -08:00
parent 56484c9b14
commit cffc3d996d
4 changed files with 95 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/book/plugins.md
+2 -1
View File
@@ -68,6 +68,7 @@ import (
"github.com/open-policy-agent/opa/ast"
"github.com/open-policy-agent/opa/types"
"github.com/open-policy-agent/opa/topdown"
"github.com/open-policy-agent/opa/topdown/builtins"
)
var HelloBuiltin = &ast.Builtin{
@@ -79,7 +80,7 @@ var HelloBuiltin = &ast.Builtin{
}
func HelloImpl(a ast.Value) (ast.Value, error) {
s, err := builtins.StringOperand(1, a)
s, err := builtins.StringOperand(a, 1)
if err != nil {
return nil, err
}
plugins/discovery/discovery.go
+1 -1
View File
@@ -289,7 +289,7 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager
return nil, err
}
decisionLogsConfig, err := logs.ParseConfig(config.DecisionLogs, manager.Services())
decisionLogsConfig, err := logs.ParseConfig(config.DecisionLogs, manager.Services(), pluginNames)
if err != nil {
return nil, err
}
plugins/logs/plugin.go
+50 -14
View File
@@ -8,7 +8,6 @@ package logs
import (
"context"
"fmt"
"github.com/open-policy-agent/opa/version"
"math/rand"
"net/http"
"reflect"
@@ -20,10 +19,17 @@ import (
"github.com/open-policy-agent/opa/plugins/rest"
"github.com/open-policy-agent/opa/server"
"github.com/open-policy-agent/opa/util"
"github.com/open-policy-agent/opa/version"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
)
// Logger defines the interface for decision logging plugins.
type Logger interface {
plugins.Plugin
Log(context.Context, EventV1)
}
// EventV1 represents a decision log event.
type EventV1 struct {
Labels map[string]string `json:"labels"`
@@ -56,14 +62,26 @@ type ReportingConfig struct {
// Config represents the plugin configuration.
type Config struct {
Plugin *string `json:"plugin"`
Service string `json:"service"`
PartitionName string `json:"partition_name,omitempty"`
Reporting ReportingConfig `json:"reporting"`
}
func (c *Config) validateAndInjectDefaults(services []string) error {
func (c *Config) validateAndInjectDefaults(services []string, plugins []string) error {
if c.Service == "" && len(services) != 0 {
if c.Plugin != nil {
var found bool
for _, other := range plugins {
if other == *c.Plugin {
found = true
break
}
}
if !found {
return fmt.Errorf("invalid plugin name %q in decision_logs", *c.Plugin)
}
} else if c.Service == "" && len(services) != 0 {
c.Service = services[0]
} else {
found := false
@@ -134,8 +152,7 @@ type Plugin struct {
}
// ParseConfig validates the config and injects default values.
func ParseConfig(config []byte, services []string) (*Config, error) {
func ParseConfig(config []byte, services []string, plugins []string) (*Config, error) {
if config == nil {
return nil, nil
}
@@ -146,7 +163,7 @@ func ParseConfig(config []byte, services []string) (*Config, error) {
return nil, err
}
if err := parsedConfig.validateAndInjectDefaults(services); err != nil {
if err := parsedConfig.validateAndInjectDefaults(services, plugins); err != nil {
return nil, err
}
@@ -196,6 +213,7 @@ func (p *Plugin) Stop(ctx context.Context) {
// Log appends a decision log event to the buffer for uploading.
func (p *Plugin) Log(ctx context.Context, decision *server.Info) {
path := strings.Replace(strings.TrimPrefix(decision.Query, "data."), ".", "/", -1)
event := EventV1{
@@ -210,6 +228,16 @@ func (p *Plugin) Log(ctx context.Context, decision *server.Info) {
Version: version.Version,
}
if p.config.Plugin != nil {
proxy, ok := p.manager.Plugin(*p.config.Plugin).(Logger)
if !ok {
p.logError("Plugin does not implement Logger interface. Dropping event.")
return
}
proxy.Log(ctx, event)
return
}
p.mtx.Lock()
defer p.mtx.Unlock()
@@ -236,14 +264,19 @@ func (p *Plugin) loop() {
var retry int
for {
uploaded, err := p.oneShot(ctx)
var err error
if err != nil {
p.logError("%v.", err)
} else if uploaded {
p.logInfo("Logs uploaded successfully.")
} else {
p.logInfo("Log upload skipped.")
if p.config.Plugin == nil {
var uploaded bool
uploaded, err = p.oneShot(ctx)
if err != nil {
p.logError("%v.", err)
} else if uploaded {
p.logInfo("Logs uploaded successfully.")
} else {
p.logInfo("Log upload skipped.")
}
}
var delay time.Duration
@@ -256,7 +289,10 @@ func (p *Plugin) loop() {
delay = util.DefaultBackoff(float64(minRetryDelay), float64(*p.config.Reporting.MaxDelaySeconds), retry)
}
p.logDebug("Waiting %v before next upload/retry.", delay)
if p.config.Plugin == nil {
p.logDebug("Waiting %v before next upload/retry.", delay)
}
timer := time.NewTimer(delay)
select {
plugins/logs/plugin_test.go
+42 -3
View File
@@ -9,7 +9,6 @@ import (
"context"
"encoding/json"
"fmt"
"github.com/open-policy-agent/opa/version"
"net/http"
"net/http/httptest"
"os"
@@ -20,6 +19,7 @@ import (
"github.com/open-policy-agent/opa/plugins"
"github.com/open-policy-agent/opa/server"
"github.com/open-policy-agent/opa/storage/inmem"
"github.com/open-policy-agent/opa/version"
)
func TestMain(m *testing.M) {
@@ -28,6 +28,45 @@ func TestMain(m *testing.M) {
os.Exit(m.Run())
}
type testPlugin struct {
events []EventV1
}
func (p *testPlugin) Start(context.Context) error {
return nil
}
func (p *testPlugin) Stop(context.Context) {
}
func (p *testPlugin) Reconfigure(context.Context, interface{}) {
}
func (p *testPlugin) Log(_ context.Context, event EventV1) {
p.events = append(p.events, event)
}
func TestPluginCustomBackend(t *testing.T) {
ctx := context.Background()
manager, _ := plugins.New(nil, "test-instance-id", inmem.New())
backend := &testPlugin{}
manager.Register("test_plugin", backend)
config, err := ParseConfig([]byte(`{"plugin": "test_plugin"}`), nil, []string{"test_plugin"})
if err != nil {
t.Fatal(err)
}
plugin := New(config, manager)
plugin.Log(ctx, &server.Info{Revision: "A"})
plugin.Log(ctx, &server.Info{Revision: "B"})
if len(backend.events) != 2 || backend.events[0].Revision != "A" || backend.events[1].Revision != "B" {
t.Fatal("Unexpected events:", backend.events)
}
}
func TestPluginStartSameInput(t *testing.T) {
ctx := context.Background()
@@ -295,7 +334,7 @@ func TestPluginReconfigure(t *testing.T) {
}
}`, minDelay, maxDelay))
config, _ := ParseConfig(pluginConfig, fixture.manager.Services())
config, _ := ParseConfig(pluginConfig, fixture.manager.Services(), nil)
fixture.plugin.Reconfigure(ctx, config)
fixture.plugin.Stop(ctx)
@@ -356,7 +395,7 @@ func newTestFixture(t *testing.T) testFixture {
"service": "example",
}`))
config, _ := ParseConfig([]byte(pluginConfig), manager.Services())
config, _ := ParseConfig([]byte(pluginConfig), manager.Services(), nil)
p := New(config, manager)