From 8885997264a574554f2d31d488c32902f05c4f2b Mon Sep 17 00:00:00 2001 From: Varun Mathur Date: Tue, 10 Jul 2018 14:39:02 -0700 Subject: [PATCH] Added ability to dynamically load .so objects and the respective required testing. Signed-off-by: Varun Mathur --- cmd/plugins.go | 81 ++++++++++++ cmd/plugins_test.go | 305 ++++++++++++++++++++++++++++++++++++++++++++ plugins/plugins.go | 11 ++ runtime/runtime.go | 39 +++--- util/test/tempfs.go | 11 +- 5 files changed, 421 insertions(+), 26 deletions(-) create mode 100644 cmd/plugins.go create mode 100644 cmd/plugins_test.go diff --git a/cmd/plugins.go b/cmd/plugins.go new file mode 100644 index 0000000000..674d40d996 --- /dev/null +++ b/cmd/plugins.go @@ -0,0 +1,81 @@ +// 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. + +// Specifies additional cmd commands that available to systems that can load plugins +// +build linux,cgo darwin,cgo + +package cmd + +import ( + "fmt" + "github.com/spf13/cobra" + "os" + "path/filepath" + "plugin" +) + +// registerSharedObjectsFromDir recursively loads all .so files in dir into OPA. +func registerSharedObjectsFromDir(dir string) error { + return filepath.Walk(dir, lambdaWalker(registerSharedObjectFromFile, ".so")) +} + +// lambdaWalker returns a walkfunc that applies lambda to every file with extension ext. Ignores all other file types. +// Lambda should take the file path as its parameter. +func lambdaWalker(lambda func(string) error, ext string) filepath.WalkFunc { + walk := func(path string, f os.FileInfo, err error) error { + // if error occurs during traversal to path, exit and crash + if err != nil { + return err + } + // skip anything that is a directory + if f.IsDir() { + return nil + } + + if filepath.Ext(path) == ext { + return lambda(path) + } + + // ignore anything else + return nil + } + return walk +} + +// loads the builtin from a file path +func registerSharedObjectFromFile(path string) error { + mod, err := plugin.Open(path) + if err != nil { + return err + } + initSym, err := mod.Lookup("Init") + if err != nil { + return err + } + + // type assert init symbol + init, ok := initSym.(func() error) + if !ok { + return fmt.Errorf("symbol Init must be of type func() error") + } + + // execute init + return init() +} + +func init() { + var pluginDir string + + // flag is persistent (can be loaded on all children commands) + RootCommand.PersistentFlags().StringVarP(&pluginDir, "plugin-dir", "p", "", `set directory path to load built-in and plugin shared object files from`) + + // Runs before *all* children commands + RootCommand.PersistentPreRunE = func(cmd *cobra.Command, args []string) error { + // only register custom plugins if directory specified + if pluginDir != "" { + return registerSharedObjectsFromDir(pluginDir) + } + return nil + } +} diff --git a/cmd/plugins_test.go b/cmd/plugins_test.go new file mode 100644 index 0000000000..c4405bee17 --- /dev/null +++ b/cmd/plugins_test.go @@ -0,0 +1,305 @@ +// 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. + +// +build linux,cgo darwin,cgo + +package cmd + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "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" + "github.com/open-policy-agent/opa/util/test" +) + +// whenever a plugin is initialized it adds an item to this channel +var initChan = make(chan struct{}, 256) +var testDirRoot string + +// makeDirWithSharedObjects creates a new temporary directory containing files under the runtime directory +// it compiles all .go files into shared object files with extension ext in the corresponding directory +// It returns the root of the directory and a cleanup function. +func makeDirWithSharedObjects(files map[string]string, ext string) (root string, cleanup func()) { + root, cleanup, err := test.MakeTempFS("./", "plugin_test_tempdir", files) + if err != nil { + panic(err) + } + for file := range files { + if filepath.Ext(file) == ".go" { + src := filepath.Join(root, file) + so := strings.TrimSuffix(filepath.Base(src), ".go") + ext + out := filepath.Join(filepath.Dir(src), so) + // build latest version of shared object + cmd := exec.Command("go", "build", "-buildmode=plugin", "-o="+out, src) + res, err := cmd.Output() + if err != nil { + panic(fmt.Sprintf("attempted to build %v to %v\n", src, out) + string(res) + err.Error()) + } + } + } + return +} + +// emptyInitChan removes all current items in initChan +func emptyInitChan() { + for len(initChan) > 0 { + <-initChan + } +} + +// Runs all tests with the filesystem given below. The plugins add an item to initChan upon activation. +// This is a separate function in order to allow deferred calls to activate. +// TestMain does not honor deferred calls as it uses os.Exit. +func testMainInEnvironment(m *testing.M) int { + + // server sends item to channel upon request + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + initChan <- struct{}{} + })) + defer ts.Close() + + config := ` +plugins: + test: + key: secret +` + badConfig := ` +plugins: + test: + key: fake +` + + files := map[string]string{ + "/builtins/true.go": getBuiltinWithName("true"), + "/plugins/test.go": getPluginWithNameAndURL("test", ts.URL), + "/plugins/config.yaml": config, + "/plugins/bad-config.yaml": badConfig, + } + + root, cleanup := makeDirWithSharedObjects(files, ".so") + testDirRoot = root + defer cleanup() + + return m.Run() +} + +// returns a builtin that always returns true with name name +func getBuiltinWithName(name string) string { + return fmt.Sprintf(` +package main + +import ( + "github.com/open-policy-agent/opa/ast" + "github.com/open-policy-agent/opa/types" + "github.com/open-policy-agent/opa/topdown" +) + +var TruthfulBuiltin = &ast.Builtin{ + Name: "%v", + Decl: types.NewFunction( + types.Args(types.N, types.N), + types.B, + ), +} + +func Truthful(a, b ast.Value) (ast.Value, error) { + return ast.Boolean(true), nil +} + +func Init() error { + ast.RegisterBuiltin(TruthfulBuiltin) + topdown.RegisterFunctionalBuiltin2(TruthfulBuiltin.Name, Truthful) + return nil +} +`, name) +} + +// returns go code for a plugin named name that makes a single get request to URL upon start and requires that +// the key "secret" is provided to start. +func getPluginWithNameAndURL(name, url string) string { + return fmt.Sprintf(` +package main + +import ( + "context" + "fmt" + "net/http" + + "github.com/open-policy-agent/opa/plugins" + "github.com/open-policy-agent/opa/util" + "github.com/open-policy-agent/opa/runtime" +) + +var Name = "%v" + +type Tester struct {} + +func (t *Tester) Start(ctx context.Context) error { + _, err := http.Get("%v") + return err +} + +func (t *Tester) Stop(ctx context.Context) { + return +} + +var Initializer plugins.PluginInitFunc = func(m *plugins.Manager, config []byte) (plugins.Plugin, error) { + var test struct { + Key string + } + + if err := util.Unmarshal(config, &test); err != nil { + return nil, err + } + + if test.Key != "secret" { + return nil, fmt.Errorf("got " + test.Key + ", expected secret") + } + + return &Tester{}, nil +} + +func Init() error { + runtime.RegisterPlugin(Name, Initializer) + return nil +} +`, name, url) +} + +func TestMain(m *testing.M) { + os.Exit(testMainInEnvironment(m)) +} + +// Tests that a single builtin is loaded correctly +func TestRegisterBuiltin(t *testing.T) { + + name := "true" + builtinDir := filepath.Join(testDirRoot, "/builtins") + err := registerSharedObjectsFromDir(builtinDir) + if err != nil { + t.Fatalf(err.Error()) + } + + expected := &ast.Builtin{ + Name: name, + Decl: types.NewFunction( + types.Args(types.N, types.N), + types.B, + ), + } + + // check that builtin function was loaded correctly + actual := ast.BuiltinMap[name] + if !reflect.DeepEqual(*expected, *actual) { + t.Fatalf("Expected builtin %v but got: %v", *expected, *actual) + } +} + +// Tests that a single plugin is loaded correctly +func TestRegisterPlugin(t *testing.T) { + + // load the plugins + pluginDir := filepath.Join(testDirRoot, "/plugins") + if err := registerSharedObjectsFromDir(pluginDir); err != nil { + t.Fatalf(err.Error()) + } + + params := runtime.NewParams() + params.ConfigFile = filepath.Join(testDirRoot, "/plugins/config.yaml") + rt, err := runtime.NewRuntime(context.Background(), params) + if err != nil { + t.Fatalf(err.Error()) + } + + // make sure starting the manager kicks the plugin in + emptyInitChan() + if err := rt.Manager.Start(context.Background()); err != nil { + t.Fatalf("Unable to initialize plugins: %v", err.Error()) + } + if len(initChan) != 1 { + t.Fatalf("Plugin was started %v times", len(initChan)) + } + + return +} + +// Tests that a plugin does not start without a config file +func TestPluginDoesNotStartWithoutConfig(t *testing.T) { + // load the plugins + pluginDir := filepath.Join(testDirRoot, "/plugins") + if err := registerSharedObjectsFromDir(pluginDir); err != nil { + t.Fatalf(err.Error()) + } + + params := runtime.NewParams() + rt, err := runtime.NewRuntime(context.Background(), params) + if err != nil { + t.Fatalf(err.Error()) + } + + // make sure starting the manager kicks the plugin in + emptyInitChan() + if err := rt.Manager.Start(context.Background()); err != nil { + t.Fatalf("Unable to initialize plugins: %v", err.Error()) + } + if len(initChan) != 0 { + t.Fatalf("Plugin was started %v times", len(initChan)) + } + + return +} + +// Tests that a plugin correctly runs its registration +func TestPluginNoRegistrationWithWrongKey(t *testing.T) { + // load the plugins + pluginDir := filepath.Join(testDirRoot, "/plugins") + if err := registerSharedObjectsFromDir(pluginDir); err != nil { + t.Fatalf(err.Error()) + } + + params := runtime.NewParams() + params.ConfigFile = filepath.Join(testDirRoot, "/plugins/bad-config.yaml") + _, err := runtime.NewRuntime(context.Background(), params) + if err == nil || !strings.Contains(err.Error(), "expected secret") { + t.Fatalf("Runtime exited incorrectly with error %v", err) + } +} + +// Tests that the recursive file walker works as expected +func TestLambdaFileWalker(t *testing.T) { + files := map[string]string{ + "one.go": "", + "two.go": "", + "fake.html": "", + "deep/three.go": "", + "deep/deeper/four.go": "", + "deep/fake/fake.html": "", + } + + test.WithTempFS(files, func(root string) { + count := 0 + err := filepath.Walk(root, lambdaWalker(func(s string) error { + count++ + return nil + }, ".go")) + if err != nil { + t.Fatalf(err.Error()) + } + if count != 4 { + t.Fatalf("Expected 4, got %v", count) + } + }) +} diff --git a/plugins/plugins.go b/plugins/plugins.go index 501171739e..538ef9d3ee 100644 --- a/plugins/plugins.go +++ b/plugins/plugins.go @@ -22,6 +22,10 @@ type Plugin interface { Stop(ctx context.Context) } +// 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 { @@ -132,6 +136,13 @@ 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) { + for _, p := range m.plugins { + p.Stop(ctx) + } +} + func (m *Manager) onCommit(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { if event.PolicyChanged() { compiler, _ := loadCompilerFromStore(ctx, m.Store, txn) diff --git a/runtime/runtime.go b/runtime/runtime.go index f505ac88a8..4da7d6900b 100644 --- a/runtime/runtime.go +++ b/runtime/runtime.go @@ -11,13 +11,14 @@ import ( "crypto/tls" "encoding/json" "fmt" + "github.com/sirupsen/logrus" + "gopkg.in/fsnotify.v1" "io" "io/ioutil" "os" - "sync" "time" - fsnotify "gopkg.in/fsnotify.v1" + "github.com/pkg/errors" "github.com/open-policy-agent/opa/ast" "github.com/open-policy-agent/opa/loader" @@ -31,30 +32,20 @@ import ( "github.com/open-policy-agent/opa/storage/inmem" "github.com/open-policy-agent/opa/util" "github.com/open-policy-agent/opa/version" - "github.com/pkg/errors" - "github.com/sirupsen/logrus" + "sync" ) var ( - registeredPlugins []pluginFactory + registeredPlugins map[string]plugins.PluginInitFunc registeredPluginsMux sync.Mutex ) -// RegisterPlugin registers a plugin with the runtime package. When a Runtime -// is created, the factory functions will be called. -func RegisterPlugin(name string, factory func(m *plugins.Manager, config []byte) (plugins.Plugin, error)) { +// RegisterPlugin registers a plugin with the plugins package. When a Runtime +// is created, the factory functions will be called. This function is idempotent. +func RegisterPlugin(name string, factory plugins.PluginInitFunc) { registeredPluginsMux.Lock() defer registeredPluginsMux.Unlock() - - registeredPlugins = append(registeredPlugins, pluginFactory{ - name: name, - factory: factory, - }) -} - -type pluginFactory struct { - name string - factory func(m *plugins.Manager, config []byte) (plugins.Plugin, error) + registeredPlugins[name] = factory } // Params stores the configuration for an OPA instance. @@ -222,6 +213,7 @@ func (rt *Runtime) StartServer(ctx context.Context) { if err := rt.Manager.Start(ctx); err != nil { logrus.WithField("err", err).Fatalf("Unable to initialize plugins.") } + defer rt.Manager.Stop(ctx) s, err := server.New(). WithStore(rt.Store). @@ -275,6 +267,7 @@ func (rt *Runtime) StartREPL(ctx context.Context) { fmt.Fprintln(rt.Params.Output, "error starting plugins:", err) os.Exit(1) } + 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) @@ -587,12 +580,12 @@ func initRegisteredPlugins(m *plugins.Manager, bs []byte) error { return err } - for _, reg := range registeredPlugins { - pc, ok := config.Plugins[reg.name] + for name, factory := range registeredPlugins { + pc, ok := config.Plugins[name] if !ok { continue } - plugin, err := reg.factory(m, pc) + plugin, err := factory(m, pc) if err != nil { return err } @@ -655,3 +648,7 @@ func uuid4() (string, error) { } type bundlePluginListener string + +func init() { + registeredPlugins = make(map[string]plugins.PluginInitFunc) +} diff --git a/util/test/tempfs.go b/util/test/tempfs.go index 6f648901d1..06f190c1e4 100644 --- a/util/test/tempfs.go +++ b/util/test/tempfs.go @@ -13,7 +13,7 @@ import ( // WithTempFS creates a temporary directory structure and invokes f with the // root directory path. func WithTempFS(files map[string]string, f func(string)) { - rootDir, cleanup, err := MakeTempFS(files) + rootDir, cleanup, err := MakeTempFS("", "loader_test", files) if err != nil { panic(err) } @@ -21,12 +21,13 @@ func WithTempFS(files map[string]string, f func(string)) { f(rootDir) } -// MakeTempFS creates a temporary directory structure for test purposes. If the -// creation fails, cleanup is nil and the caller does not have to invoke it. If +// MakeTempFS creates a temporary directory structure for test purposes rooted at root. +// If root is empty, the dir is created in the default system temp location. +// If the creation fails, cleanup is nil and the caller does not have to invoke it. If // creation succeeds, the caller should invoke cleanup when they are done. -func MakeTempFS(files map[string]string) (rootDir string, cleanup func(), err error) { +func MakeTempFS(root, prefix string, files map[string]string) (rootDir string, cleanup func(), err error) { - rootDir, err = ioutil.TempDir("", "loader_test") + rootDir, err = ioutil.TempDir(root, prefix) if err != nil { return "", nil, err