diff --git a/docs/book/plugins.md b/docs/book/plugins.md index 949320bbdf..1b637cd5b1 100644 --- a/docs/book/plugins.md +++ b/docs/book/plugins.md @@ -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 } diff --git a/plugins/discovery/discovery.go b/plugins/discovery/discovery.go index 3bca86879c..1007e0dd68 100644 --- a/plugins/discovery/discovery.go +++ b/plugins/discovery/discovery.go @@ -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 } diff --git a/plugins/logs/plugin.go b/plugins/logs/plugin.go index 458f379904..2688233cdc 100644 --- a/plugins/logs/plugin.go +++ b/plugins/logs/plugin.go @@ -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 { diff --git a/plugins/logs/plugin_test.go b/plugins/logs/plugin_test.go index 7b158f1cbe..84c3740ee3 100644 --- a/plugins/logs/plugin_test.go +++ b/plugins/logs/plugin_test.go @@ -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)