From 346aa964e8a542dd0cc2ec030fc0617c8aaa702a Mon Sep 17 00:00:00 2001 From: Patrick East Date: Tue, 25 Jun 2019 13:30:17 -0700 Subject: [PATCH] Add support for multiple bundles This change brings in support for multiple bundles to be downloaded and activated OPA. This is enabled by using the new config option `bundles` to define the bundles, and deprecates the older `bundle` option. The new `bundles` keyword and structure is propagated through to the decision logs, status API, provenance, stored manifests, etc. Check out the doc changes for all the updated structures. That being said any existing configuration using `bundle` will *not* see the new structure, everything is intended to be backwards compatible (almost to a fault). Fixes: #721 Signed-off-by: Patrick East --- bundle/bundle.go | 8 +- bundle/manifest/manifest.go | 82 --- bundle/store.go | 166 ++++++ bundle/store_test.go | 203 ++++++++ config/config.go | 5 +- config/config_test.go | 70 +++ docs/content/bundles.md | 49 +- docs/content/configuration.md | 28 +- docs/content/decision-logs.md | 10 +- docs/content/rest-api.md | 18 +- docs/content/status.md | 24 +- download/download.go | 18 +- plugins/bundle/config.go | 143 +++++- plugins/bundle/config_test.go | 224 ++++++++ plugins/bundle/plugin.go | 363 +++++++++---- plugins/bundle/plugin_test.go | 771 +++++++++++++++++++++++----- plugins/discovery/discovery.go | 23 +- plugins/discovery/discovery_test.go | 83 +++ plugins/logs/plugin.go | 41 +- plugins/logs/plugin_test.go | 36 ++ plugins/status/plugin.go | 69 ++- plugins/status/plugin_test.go | 91 +++- server/buffer.go | 8 +- server/server.go | 158 ++++-- server/server_test.go | 391 +++++++++++++- server/types/types.go | 16 +- util/close.go | 2 +- 27 files changed, 2609 insertions(+), 491 deletions(-) delete mode 100644 bundle/manifest/manifest.go create mode 100644 bundle/store.go create mode 100644 bundle/store_test.go create mode 100644 config/config_test.go diff --git a/bundle/bundle.go b/bundle/bundle.go index 9db3b18de1..d72aa4c366 100644 --- a/bundle/bundle.go +++ b/bundle/bundle.go @@ -32,8 +32,6 @@ const ( const bundleLimitBytes = (1024 * 1024 * 1024) + 1 // limit bundle reads to 1GB to protect against gzip bombs -var manifestPath = []string{"system", "bundle", "manifest"} - // Bundle represents a loaded bundle. The bundle can contain data and policies. type Bundle struct { Manifest Manifest @@ -234,8 +232,10 @@ func (r *Reader) Read() (Bundle, error) { return bundle, errors.Wrap(err, "bundle load failed on manifest unmarshal") } - if err := bundle.insert(manifestPath, metadata); err != nil { - return bundle, errors.Wrapf(err, "bundle load failed on %v", manifestPath) + // For backwards compatibility always write to the old unnamed manifest path + // This will *not* be correct if >1 bundle is in use... + if err := bundle.insert(legacyManifestStoragePath, metadata); err != nil { + return bundle, errors.Wrapf(err, "bundle load failed on %v", legacyRevisionStoragePath) } } diff --git a/bundle/manifest/manifest.go b/bundle/manifest/manifest.go deleted file mode 100644 index 4387008f8d..0000000000 --- a/bundle/manifest/manifest.go +++ /dev/null @@ -1,82 +0,0 @@ -// Copyright 2019 The OPA Authors. All rights reserved. -// Use of this source code is governed by an Apache2 -// license that can be found in the LICENSE file. - -// Package manifest implements helper functions for the stored manifest. -package manifest - -import ( - "context" - "fmt" - - "github.com/open-policy-agent/opa/bundle" - "github.com/open-policy-agent/opa/storage" - "github.com/open-policy-agent/opa/util" -) - -var bundlePath = storage.MustParsePath("/system/bundle") -var manifestPath = storage.MustParsePath("/system/bundle/manifest") -var revisionPath = storage.MustParsePath("/system/bundle/manifest/revision") -var rootsPath = storage.MustParsePath("/system/bundle/manifest/roots") - -// Write the manifest into the storage. This function is called when -// the bundle is activated. -func Write(ctx context.Context, store storage.Store, txn storage.Transaction, m bundle.Manifest) error { - - var value interface{} = m - - if err := util.RoundTrip(&value); err != nil { - return err - } - - if err := storage.MakeDir(ctx, store, txn, bundlePath); err != nil { - return err - } - - return store.Write(ctx, txn, storage.AddOp, manifestPath, value) -} - -// ReadBundleRoots returns the roots specified in the currently -// activated bundle. If there is no activated bundle, this function -// will return storage NotFound error. -func ReadBundleRoots(ctx context.Context, store storage.Store, txn storage.Transaction) ([]string, error) { - - value, err := store.Read(ctx, txn, rootsPath) - if err != nil { - return nil, err - } - - sl, ok := value.([]interface{}) - if !ok { - return nil, fmt.Errorf("corrupt manifest roots") - } - - roots := make([]string, len(sl)) - - for i := range sl { - roots[i], ok = sl[i].(string) - if !ok { - return nil, fmt.Errorf("corrupt manifest root") - } - } - - return roots, nil -} - -// ReadBundleRevision returns the revision in the currently activated -// bundle. If there is no activated bundle, ths function will return -// storage NotFound error. -func ReadBundleRevision(ctx context.Context, store storage.Store, txn storage.Transaction) (string, error) { - - value, err := store.Read(ctx, txn, revisionPath) - if err != nil { - return "", err - } - - str, ok := value.(string) - if !ok { - return "", fmt.Errorf("corrupt manifest revision") - } - - return str, nil -} diff --git a/bundle/store.go b/bundle/store.go new file mode 100644 index 0000000000..d52664a2cf --- /dev/null +++ b/bundle/store.go @@ -0,0 +1,166 @@ +// Copyright 2019 The OPA Authors. All rights reserved. +// Use of this source code is governed by an Apache2 +// license that can be found in the LICENSE file. + +package bundle + +import ( + "context" + "fmt" + + "github.com/open-policy-agent/opa/storage" + "github.com/open-policy-agent/opa/util" +) + +var bundlesBasePath = storage.MustParsePath("/system/bundles") + +// Note: As needed these helpers could be memoized. + +// ManifestStoragePath is the storage path used for the given named bundle manifest. +func ManifestStoragePath(name string) storage.Path { + return append(bundlesBasePath, name, "manifest") +} + +func namedBundlePath(name string) storage.Path { + return append(bundlesBasePath, name) +} + +func rootsPath(name string) storage.Path { + return append(bundlesBasePath, name, "manifest", "roots") +} + +func revisionPath(name string) storage.Path { + return append(bundlesBasePath, name, "manifest", "revision") +} + +// ReadBundleNamesFromStore will return a list of bundle names which have had their metadata stored. +func ReadBundleNamesFromStore(ctx context.Context, store storage.Store, txn storage.Transaction) ([]string, error) { + value, err := store.Read(ctx, txn, bundlesBasePath) + if err != nil { + return nil, err + } + + bundleMap, ok := value.(map[string]interface{}) + if !ok { + return nil, fmt.Errorf("corrupt manifest roots") + } + + bundles := make([]string, len(bundleMap)) + idx := 0 + for name := range bundleMap { + bundles[idx] = name + idx++ + } + return bundles, nil +} + +// WriteManifestToStore will write the manifest into the storage. This function is called when +// the bundle is activated. +func WriteManifestToStore(ctx context.Context, store storage.Store, txn storage.Transaction, name string, manifest Manifest) error { + return write(ctx, store, txn, ManifestStoragePath(name), manifest) +} + +func write(ctx context.Context, store storage.Store, txn storage.Transaction, path storage.Path, manifest Manifest) error { + var value interface{} = manifest + if err := util.RoundTrip(&value); err != nil { + return err + } + + var dir []string + if len(path) > 1 { + dir = path[:len(path)-1] + } + + if err := storage.MakeDir(ctx, store, txn, dir); err != nil { + return err + } + + return store.Write(ctx, txn, storage.AddOp, path, value) +} + +// EraseManifestFromStore will remove the manifest from storage. This function is called +// when the bundle is deactivated. +func EraseManifestFromStore(ctx context.Context, store storage.Store, txn storage.Transaction, name string) error { + path := namedBundlePath(name) + err := store.Write(ctx, txn, storage.RemoveOp, path, nil) + if err != nil && !storage.IsNotFound(err) { + return err + } + return nil +} + +// ReadBundleRootsFromStore returns the roots in the specified bundle. +// If the bundle is not activated, this function will return +// storage NotFound error. +func ReadBundleRootsFromStore(ctx context.Context, store storage.Store, txn storage.Transaction, name string) ([]string, error) { + value, err := store.Read(ctx, txn, rootsPath(name)) + if err != nil { + return nil, err + } + + sl, ok := value.([]interface{}) + if !ok { + return nil, fmt.Errorf("corrupt manifest roots") + } + + roots := make([]string, len(sl)) + + for i := range sl { + roots[i], ok = sl[i].(string) + if !ok { + return nil, fmt.Errorf("corrupt manifest root") + } + } + + return roots, nil +} + +// ReadBundleRevisionFromStore returns the revision in the specified bundle. +// If the bundle is not activated, this function will return +// storage NotFound error. +func ReadBundleRevisionFromStore(ctx context.Context, store storage.Store, txn storage.Transaction, name string) (string, error) { + return readRevisionFromStore(ctx, store, txn, revisionPath(name)) +} + +func readRevisionFromStore(ctx context.Context, store storage.Store, txn storage.Transaction, path storage.Path) (string, error) { + value, err := store.Read(ctx, txn, path) + if err != nil { + return "", err + } + + str, ok := value.(string) + if !ok { + return "", fmt.Errorf("corrupt manifest revision") + } + + return str, nil +} + +// Helpers for the older single (unnamed) bundle style manifest storage. + +// LegacyManifestStoragePath is the older unnamed bundle path for manifests to be stored. +// Deprecated: Use ManifestStoragePath and named bundles instead. +var legacyManifestStoragePath = storage.MustParsePath("/system/bundle/manifest") +var legacyRevisionStoragePath = append(legacyManifestStoragePath, "revision") + +// LegacyWriteManifestToStore will write the bundle manifest to the older single (unnamed) bundle manifest location. +// Deprecated: Use WriteManifestToStore and named bundles instead. +func LegacyWriteManifestToStore(ctx context.Context, store storage.Store, txn storage.Transaction, manifest Manifest) error { + return write(ctx, store, txn, legacyManifestStoragePath, manifest) +} + +// LegacyEraseManifestFromStore will erase the bundle manifest from the older single (unnamed) bundle manifest location. +// Deprecated: Use WriteManifestToStore and named bundles instead. +func LegacyEraseManifestFromStore(ctx context.Context, store storage.Store, txn storage.Transaction) error { + err := store.Write(ctx, txn, storage.RemoveOp, legacyManifestStoragePath, nil) + if err != nil { + return err + } + return nil +} + +// LegacyReadRevisionFromStore will read the bundle manifest revision from the older single (unnamed) bundle manifest location. +// Deprecated: Use ReadBundleRevisionFromStore and named bundles instead. +func LegacyReadRevisionFromStore(ctx context.Context, store storage.Store, txn storage.Transaction) (string, error) { + return readRevisionFromStore(ctx, store, txn, legacyRevisionStoragePath) +} diff --git a/bundle/store_test.go b/bundle/store_test.go new file mode 100644 index 0000000000..1fcc4071aa --- /dev/null +++ b/bundle/store_test.go @@ -0,0 +1,203 @@ +package bundle + +import ( + "context" + "testing" + + "github.com/open-policy-agent/opa/storage" + "github.com/open-policy-agent/opa/storage/inmem" +) + +func TestManifestStoreLifecycleSingleBundle(t *testing.T) { + store := inmem.New() + ctx := context.Background() + tb := Manifest{ + Revision: "abc123", + Roots: &[]string{"/a/b", "/a/c"}, + } + name := "test_bundle" + verifyWriteManifests(ctx, t, store, map[string]Manifest{name: tb}) // write one + verifyReadBundleNames(ctx, t, store, []string{name}) // read one + verifyDeleteManifest(ctx, t, store, name) // delete it + verifyReadBundleNames(ctx, t, store, []string{}) // ensure it was removed +} + +func TestManifestStoreLifecycleMultiBundle(t *testing.T) { + store := inmem.New() + ctx := context.Background() + + bundles := map[string]Manifest{ + "bundle1": { + Revision: "abc123", + Roots: &[]string{"/a/b", "/a/c"}, + }, + "bundle2": { + Revision: "def123", + Roots: &[]string{"/x/y", "/z"}, + }, + } + + verifyWriteManifests(ctx, t, store, bundles) // write multiple + verifyReadBundleNames(ctx, t, store, []string{"bundle1", "bundle2"}) // read them + verifyDeleteManifest(ctx, t, store, "bundle1") // delete one + verifyReadBundleNames(ctx, t, store, []string{"bundle2"}) // ensure it was removed + verifyDeleteManifest(ctx, t, store, "bundle2") // delete the last one + verifyReadBundleNames(ctx, t, store, []string{}) // ensure it was removed +} + +func TestLegacyManifestStoreLifecycle(t *testing.T) { + store := inmem.New() + ctx := context.Background() + tb := Manifest{ + Revision: "abc123", + Roots: &[]string{"/a/b", "/a/c"}, + } + + // write a "legacy" manifest + err := storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { + if err := LegacyWriteManifestToStore(ctx, store, txn, tb); err != nil { + t.Fatalf("Failed to write manifest to store: %s", err) + return err + } + return nil + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + + // make sure it can be retrieved + verifyReadLegacyRevision(ctx, t, store, tb.Revision) + + // delete it + err = storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { + if err := LegacyEraseManifestFromStore(ctx, store, txn); err != nil { + t.Fatalf("Failed to erase manifest from store: %s", err) + return err + } + return nil + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + + verifyReadLegacyRevision(ctx, t, store, "") +} + +func TestMixedManifestStoreLifecycle(t *testing.T) { + store := inmem.New() + ctx := context.Background() + bundles := map[string]Manifest{ + "bundle1": { + Revision: "abc123", + Roots: &[]string{"/a/b", "/a/c"}, + }, + "bundle2": { + Revision: "def123", + Roots: &[]string{"/x/y", "/z"}, + }, + } + + // Write the legacy one first + err := storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { + if err := LegacyWriteManifestToStore(ctx, store, txn, bundles["bundle1"]); err != nil { + t.Fatalf("Failed to write manifest to store: %s", err) + return err + } + return nil + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + + verifyReadBundleNames(ctx, t, store, []string{}) + + // Write both new ones + verifyWriteManifests(ctx, t, store, bundles) + verifyReadBundleNames(ctx, t, store, []string{"bundle1", "bundle2"}) + + // Ensure the original legacy one is still there + verifyReadLegacyRevision(ctx, t, store, bundles["bundle1"].Revision) +} + +func verifyDeleteManifest(ctx context.Context, t *testing.T, store storage.Store, name string) { + err := storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { + err := EraseManifestFromStore(ctx, store, txn, name) + if err != nil { + t.Fatalf("Failed to delete manifest from store: %s", err) + } + return err + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } +} + +func verifyWriteManifests(ctx context.Context, t *testing.T, store storage.Store, bundles map[string]Manifest) { + t.Helper() + for name, manifest := range bundles { + err := storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { + err := WriteManifestToStore(ctx, store, txn, name, manifest) + if err != nil { + t.Fatalf("Failed to write manifest to store: %s", err) + } + return err + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + } +} + +func verifyReadBundleNames(ctx context.Context, t *testing.T, store storage.Store, expected []string) { + t.Helper() + var actualNames []string + err := storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { + var err error + actualNames, err = ReadBundleNamesFromStore(ctx, store, txn) + if err != nil && !storage.IsNotFound(err) { + t.Fatalf("Failed to read manifest names from store: %s", err) + return err + } + return nil + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + + if len(actualNames) != len(expected) { + t.Fatalf("Expected %d name, found %d \n\t\tActual: %v\n", len(expected), len(actualNames), actualNames) + } + + for _, actualName := range actualNames { + found := false + for _, expectedName := range expected { + if actualName == expectedName { + found = true + break + } + } + if !found { + t.Errorf("Found unexpecxted bundle name %s, expected names: %+v", actualName, expected) + } + } +} + +func verifyReadLegacyRevision(ctx context.Context, t *testing.T, store storage.Store, expected string) { + t.Helper() + var actual string + err := storage.Txn(ctx, store, storage.WriteParams, func(txn storage.Transaction) error { + var err error + if actual, err = LegacyReadRevisionFromStore(ctx, store, txn); err != nil && !storage.IsNotFound(err) { + t.Fatalf("Failed to read manifest revision from store: %s", err) + return err + } + return nil + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + + if actual != expected { + t.Fatalf("Expected revision %s, got %s", expected, actual) + } +} diff --git a/config/config.go b/config/config.go index 7d36e01ba9..8834ce7782 100644 --- a/config/config.go +++ b/config/config.go @@ -20,7 +20,8 @@ type Config struct { Services json.RawMessage `json:"services"` Labels map[string]string `json:"labels"` Discovery json.RawMessage `json:"discovery"` - Bundle json.RawMessage `json:"bundle"` + Bundle json.RawMessage `json:"bundle"` // Deprecated: Use `bundles` instead + Bundles json.RawMessage `json:"bundles"` DecisionLogs json.RawMessage `json:"decision_logs"` Status json.RawMessage `json:"status"` Plugins map[string]json.RawMessage `json:"plugins"` @@ -40,7 +41,7 @@ func ParseConfig(raw []byte, id string) (*Config, error) { // 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 + return c.Bundle != nil || c.Bundles != nil || c.DecisionLogs != nil || c.Status != nil || len(c.Plugins) > 0 } // DefaultDecisionRef returns the default decision as a reference. diff --git a/config/config_test.go b/config/config_test.go new file mode 100644 index 0000000000..10c1c6cc66 --- /dev/null +++ b/config/config_test.go @@ -0,0 +1,70 @@ +// Copyright 2019 The OPA Authors. All rights reserved. +// Use of this source code is governed by an Apache2 +// license that can be found in the LICENSE file. + +package config + +import ( + "encoding/json" + "testing" +) + +func TestConfigPluginsEnabled(t *testing.T) { + tests := []struct { + name string + conf Config + expected bool + }{ + { + name: "empty config", + conf: Config{}, + expected: false, + }, + { + name: "bundle", + conf: Config{ + Bundle: []byte(`{"bundle": {"name": "test-bundle"}}`), + }, + expected: true, + }, + { + name: "bundles", + conf: Config{ + Bundles: []byte(`{"bundles": {"test-bundle": {}}`), + }, + expected: true, + }, + { + name: "decision_logs", + conf: Config{ + DecisionLogs: []byte(`{decision_logs: {}}`), + }, + expected: true, + }, + { + name: "status", + conf: Config{ + Status: []byte(`{status: {}}`), + }, + expected: true, + }, + { + name: "plugins", + conf: Config{ + Plugins: map[string]json.RawMessage{ + "some-plugin": {}, + }, + }, + expected: true, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + actual := test.conf.PluginsEnabled() + if actual != test.expected { + t.Errorf("Expected %t but got %t", test.expected, actual) + } + }) + } +} diff --git a/docs/content/bundles.md b/docs/content/bundles.md index c39e76be2a..7dd652d837 100644 --- a/docs/content/bundles.md +++ b/docs/content/bundles.md @@ -15,10 +15,8 @@ By configuring OPA to download bundles from a remote HTTP server, you can ensure that OPA has an up-to-date copy of policies and data required for enforcement at all times. -OPA can only be configured to download one bundle at a time. You -cannot configure OPA to download multiple bundles. By default, the OPA -REST APIs will prevent you from modifying policy and data loaded via -bundles. If you need to load policy and data from multiple sources, +By default, the OPA REST APIs will prevent you from modifying policy and data +loaded via bundles. If you need to load policy and data from multiple sources, see the section below. See the [Configuration Reference](../configuration) for configuration details. @@ -26,10 +24,11 @@ See the [Configuration Reference](../configuration) for configuration details. ## Bundle Service API OPA expects the service to expose an API endpoint that serves bundles. The -bundle API should allow clients to download named bundles. +bundle API should allow clients to download bundles at an arbitrary URL. In +combination with a service's `url` path. ```http -GET // HTTP/1.1 +GET // HTTP/1.1 ``` If the bundle exists, the server should respond with an HTTP 200 OK status @@ -51,29 +50,39 @@ services: token: "bGFza2RqZmxha3NkamZsa2Fqc2Rsa2ZqYWtsc2RqZmtramRmYWxkc2tm" bundle: - name: authz/bundle.tar.gz - prefix: somedir - service: acmecorp - polling: + authz: + service: acmecorp + resource: somedir/bundle.tar.gz + polling: min_delay_seconds: 10 max_delay_seconds: 20 ``` Using this configuration, OPA will fetch bundles from -`https://example.com/service/v1/somedir/authz/bundle.tar.gz`. +`https://example.com/service/v1/somedir/bundle.tar.gz`. The URL is constructed as follows: ``` -https://example.com/service/v1/somedir/authz/bundle.tar.gz -^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^ ^^^^^^^^^^^^^^^^^^^ -services[0].url prefix name +https://example.com/service/v1/somedir/bundle.tar.gz +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^^^^^^^ +services[0].url resource ``` -If the `bundle.prefix` field is not defined, the value defaults to `bundles`. +If the `bundle.resource` field is not defined, the value defaults to `bundles/` +where the `name` is the key value in the configuration. For the example above this is `authz` +and would default to `bundles/authz`. + +Bundle names can have any valid yaml characters in them, including `/`. This can +be useful when relying on default `resource` behavior with a name like `authz/bundle.tar.gz` +which results in a `resource` of `bundles/authz/bundle.tar.gz`. See the following section for details on the bundle file format. +> Note: The `bundle` config keyword will still work with the current versions + of OPA, but has been deprecated. It is highly recommended to switch to the + `bundles` configuration. + ### Caching Services implementing the Bundle Service API should set the HTTP `Etag` header @@ -181,7 +190,9 @@ loaded under other roots is left intact. When OPA loads scoped bundles, it validates that: * The roots are not overlapping (e.g., `a/b/c` and `a/b` are - overlapped and will result in an error.) + overlapped and will result in an error.) Note: This is *not* + enforced across multiple bundles. Only within the same bundle + manifest. * The policies in the bundle are contained under the roots. This is determined by inspecting the `package` statement in each of the @@ -194,6 +205,12 @@ When OPA loads scoped bundles, it validates that: If bundle validation fails, OPA will report the validation error via the Status API. +> **Warning!** When using multiple bundles the roots are *not* checked + against other bundles. It is the responsibility of the bundle creator + to ensure the manifest claims roots that are unique to that bundle! + There are *no* ordering guarantees for which bundle loads first and + takes over some root. + ## Debugging Your Bundles When you run OPA, you can provide bundle files over the command line. This diff --git a/docs/content/configuration.md b/docs/content/configuration.md index b67b9242bc..ec85808140 100644 --- a/docs/content/configuration.md +++ b/docs/content/configuration.md @@ -34,13 +34,13 @@ labels: region: west environment: production -bundle: - name: http/example/authz - service: acmecorp - prefix: bundles - polling: - min_delay_seconds: 60 - max_delay_seconds: 120 +bundles: + authz: + service: acmecorp + resource: bundles/http/example/authz.tar.gz + polling: + min_delay_seconds: 60 + max_delay_seconds: 120 decision_logs: service: acmecorp @@ -315,6 +315,20 @@ services: ## Bundles +Bundles are defined with a key that is the `name` of the bundle. This `name` is used in the status API, decision logs, +server provenance, etc. + +| Field | Type | Required | Description | +| --- | --- | --- | --- | +| `bundles[_].resource` | `string` | No (default: `bundles/`) | Resource path to use to download bundle from configured service. | +| `bundles[_].service` | `string` | Yes | Name of service to use to contact remote server. | +| `bundles[_].polling.min_delay_seconds` | `int64` | No (default: `60`) | Minimum amount of time to wait between bundle downloads. | +| `bundles[_].polling.max_delay_seconds` | `int64` | No (default: `120`) | Maximum amount of time to wait between bundle downloads. | + +## Bundle (Deprecated) + +> Deprecated in favor of `bundles` (see above). + | Field | Type | Required | Description | | --- | --- | --- | --- | | `bundle.name` | `string` | Yes | Name of the bundle to download. | diff --git a/docs/content/decision-logs.md b/docs/content/decision-logs.md index 9a5b20634b..1d479d49c8 100644 --- a/docs/content/decision-logs.md +++ b/docs/content/decision-logs.md @@ -40,7 +40,11 @@ represents a policy decision returned by OPA. "version": "{{< current_version >}}" }, "decision_id": "4ca636c1-55e4-417a-b1d8-4aceb67960d1", - "revision": "W3sibCI6InN5cy9jYXRhbG9nIiwicyI6NDA3MX1d", + "bundles": { + "authz": { + "revision": "W3sibCI6InN5cy9jYXRhbG9nIiwicyI6NDA3MX1d" + } + }, "path": "http/example/authz/allow", "input": { "method": "GET", @@ -59,7 +63,9 @@ Decision log updates contain the following fields: | --- | --- | --- | | `[_].labels` | `object` | Set of key-value pairs that uniquely identify the OPA instance. | | `[_].decision_id` | `string` | Unique identifier generated for each decision for traceability. | -| `[_].revision` | `string` | Bundle revision that contained the policy used to produce the decision. | +| `[_].revision` | `string` | (Deprecated) Bundle revision that contained the policy used to produce the decision. Omitted when `bundles` are configured. | +| `[_].bundles` | `object` | Set of key-value pairs describing the bundles which contained policy used to produce the decision. | +| `[_].bundles[_].revision` | `string` | Revision of the bundle at the time of evaluation. | | `[_].path` | `string` | Hierarchical policy decision path, e.g., `/http/example/authz/allow`. Receivers should tolerate slash-prefixed paths. | | `[_].query` | `string` | Ad-hoc Rego query received by Query API. | | `[_].input` | `any` | Input data provided in the policy query. | diff --git a/docs/content/rest-api.md b/docs/content/rest-api.md index 0704f79402..fe47cebc9e 100644 --- a/docs/content/rest-api.md +++ b/docs/content/rest-api.md @@ -1971,7 +1971,11 @@ Content-Type: application/json "build_commit": "1955fc4d", "build_host": "foo.com", "build_timestamp": "2019-04-29T23:42:04Z", - "revision": "ID-b1298a6c-6ad8-11e9-a26f-d38b5ceadad5", + "bundles": { + "authz": { + "revision": "ID-b1298a6c-6ad8-11e9-a26f-d38b5ceadad5" + } + }, "version": "0.10.8-dev" }, "result": true @@ -1984,7 +1988,11 @@ OPA currently supports the following query provenance information: - **build_commit**: The git commit id of this OPA build. - **build_timestamp**: The timestamp when this instance was built. - **build_host**: The hostname where this instance was built. -- **revision**: The _revision_ string included in a .manifest file (if present) within a bundle. +- **revision**: (Deprecated) The _revision_ string included in a .manifest file (if present) within + a bundle. Omitted when `bundles` are configured. +- **bundles**: A set of key-value pairs describing each bundle activated on the server. Includes + the `revision` field which is the _revision_ string included in a .manifest file (if present) + within a bundle ## Watches @@ -2040,10 +2048,10 @@ that the server is operational. Optionally it can account for bundle activation `bundle` - Boolean parameter to account for bundle activation status in response. #### Status Codes -- **200** - OPA service is healthy. If `bundle=true` the configured bundle has +- **200** - OPA service is healthy. If `bundle=true` then all configured bundles have been activated. -- **500** - OPA service is not healthy. If `bundle=true` this can mean the - configured bundle has not yet been activated. +- **500** - OPA service is not healthy. If `bundle=true` this can mean any of the configured + bundles have not yet been activated. > *Note*: The bundle activation check is only for initial startup. Subsequent downloads will not affect the health check. The [Status](/docs/{{< current_version >}}/status) diff --git a/docs/content/status.md b/docs/content/status.md index 1f51730c11..b71fdae6a5 100644 --- a/docs/content/status.md +++ b/docs/content/status.md @@ -40,11 +40,12 @@ on the agent, updates will be sent to `/status`. "id": "1780d507-aea2-45cc-ae50-fa153c8e4a5a", "version": "{{< current_version >}}" }, - "bundle": { - "name": "http/example/authz", - "active_revision": "TODO", - "last_successful_download": "2018-01-01T00:00:00.000Z", - "last_successful_activation": "2018-01-01T00:00:00.000Z" + "bundles": { + "http/example/authz": { + "active_revision": "TODO", + "last_successful_download": "2018-01-01T00:00:00.000Z", + "last_successful_activation": "2018-01-01T00:00:00.000Z" + } } } ``` @@ -54,10 +55,15 @@ Status updates contain the following fields: | Field | Type | Description | | --- | --- | --- | | `labels` | `object` | Set of key-value pairs that uniquely identify the OPA instance. | -| `bundle.name` | `string` | Name of bundle that the OPA instance is configured to download. | -| `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. | +| `bundle.name` | `string` | (Deprecated) Name of bundle that the OPA instance is configured to download. Omitted when `bundles` are configured. | +| `bundle.active_revision` | `string` | (Deprecated) Opaque revision identifier of the last successful activation. Omitted when `bundles` are configured. | +| `bundle.last_successful_download` | `string` | (Deprecated) RFC3339 timestamp of last successful bundle download. Omitted when `bundles` are configured. | +| `bundle.last_successful_activation` | `string` | (Deprecated) RFC3339 timestamp of last successful bundle activation. Omitted when `bundles` are configured. | +| `bundles` | `object` | Set of objects describing the status for each bundle configured with OPA. | +| `bundles[_].name` | `string` | Name of bundle that the OPA instance is configured to download. | +| `bundles[_].active_revision` | `string` | Opaque revision identifier of the last successful activation. | +| `bundles[_].last_successful_download` | `string` | RFC3339 timestamp of last successful bundle download. | +| `bundles[_].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. | diff --git a/download/download.go b/download/download.go index 67d098a246..a2a3dbc376 100644 --- a/download/download.go +++ b/download/download.go @@ -34,7 +34,7 @@ type Update struct { // 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 +// updates 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 @@ -143,14 +143,18 @@ func (d *Downloader) download(ctx context.Context) (*bundle.Bundle, string, erro switch resp.StatusCode { case http.StatusOK: - d.logDebug("Download in progress.") - - b, err := bundle.NewReader(resp.Body).Read() - if err != nil { - return nil, "", err + if resp.Body != nil { + 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 } - return &b, resp.Header.Get("ETag"), nil + d.logDebug("Server replied with empty body.") + return nil, "", nil + case http.StatusNotModified: return nil, resp.Header.Get("ETag"), nil case http.StatusNotFound: diff --git a/plugins/bundle/config.go b/plugins/bundle/config.go index 2acaf0478a..d926b46fa6 100644 --- a/plugins/bundle/config.go +++ b/plugins/bundle/config.go @@ -6,14 +6,17 @@ package bundle import ( "fmt" - "github.com/open-policy-agent/opa/download" "github.com/open-policy-agent/opa/util" + "path" + "strings" ) -// ParseConfig validates the config and injects default values. +// ParseConfig validates the config and injects default values. This is +// for the legacy single bundle configuration. This will add the bundle +// to the `Bundles` map to provide compatibility with newer clients. +// Deprecated: Use `ParseBundlesConfig` with `bundles` OPA config option instead func ParseConfig(config []byte, services []string) (*Config, error) { - if config == nil { return nil, nil } @@ -28,20 +31,104 @@ func ParseConfig(config []byte, services []string) (*Config, error) { return nil, err } + // For forwards compatibility make a new Source as if the bundle + // was configured with `bundles` in the newer format. + parsedConfig.Bundles = map[string]*Source{ + parsedConfig.Name: { + Config: parsedConfig.Config, + Service: parsedConfig.Service, + Resource: parsedConfig.generateLegacyResourcePath(), + }, + } + return &parsedConfig, nil } +// ParseBundlesConfig validates the config and injects default values for +// the defined `bundles`. This expects a map of bundle names to resource +// configurations. +func ParseBundlesConfig(config []byte, services []string) (*Config, error) { + if config == nil { + return nil, nil + } + + var bundleConfigs map[string]*Source + + if err := util.Unmarshal(config, &bundleConfigs); err != nil { + return nil, err + } + + // Build a `Config` out of the parsed map + c := Config{Bundles: map[string]*Source{}} + for name, source := range bundleConfigs { + if source != nil { + c.Bundles[name] = source + } + } + + err := c.validateAndInjectDefaults(services) + if err != nil { + return nil, err + } + + return &c, nil +} + // Config represents the configuration of the plugin. +// The Config can define a single bundle source or a map of +// `Source` objects defining where/how to download bundles. The +// older single bundle configuration is deprecated and will be +// removed in the future in favor of the `Bundles` map. type Config struct { + download.Config // Deprecated: Use `Bundles` map instead + + Bundles map[string]*Source + + Name string `json:"name"` // Deprecated: Use `Bundles` map instead + Service string `json:"service"` // Deprecated: Use `Bundles` map instead + Prefix *string `json:"prefix"` // Deprecated: Use `Bundles` map instead +} + +// Source is a configured bundle source to download bundles from +type Source struct { download.Config - Name string `json:"name"` - Service string `json:"service"` - Prefix *string `json:"prefix"` + Service string `json:"service"` + Resource string `json:"resource"` +} + +// IsMultiBundle returns whether or not the config is the newer multi-bundle +// style config that uses `bundles` instead of top level bundle information. +// If/when we drop support for the older style config we can remove this too. +func (c *Config) IsMultiBundle() bool { + // If a `Name` was set then the config is in "legacy" single plugin mode + return c.Name == "" } func (c *Config) validateAndInjectDefaults(services []string) error { + if c.Bundles == nil { + return c.validateAndInjectDefaultsLegacy(services) + } + for name, source := range c.Bundles { + if source.Resource == "" { + source.Resource = path.Join(defaultBundlePathPrefix, name) + } + + var err error + source.Service, err = c.getServiceFromList(source.Service, services) + if err == nil { + err = source.Config.ValidateAndInjectDefaults() + } + if err != nil { + return fmt.Errorf("invalid configuration for bundle %q: %s", name, err.Error()) + } + } + + return nil +} + +func (c *Config) validateAndInjectDefaultsLegacy(services []string) error { if c.Name == "" { return fmt.Errorf("invalid bundle name %q", c.Name) } @@ -51,24 +138,36 @@ func (c *Config) validateAndInjectDefaults(services []string) error { c.Prefix = &s } - 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) - } + var err error + c.Service, err = c.getServiceFromList(c.Service, services) + if err == nil { + err = c.Config.ValidateAndInjectDefaults() } - return c.ValidateAndInjectDefaults() + if err != nil { + return fmt.Errorf("invalid configuration for bundle %q: %s", c.Name, err.Error()) + } + + return nil +} + +func (c *Config) getServiceFromList(service string, services []string) (string, error) { + if service == "" && len(services) != 0 { + return services[0], nil + } + for _, svc := range services { + if svc == service { + return service, nil + } + } + return service, fmt.Errorf("service name %q not found", service) +} + +// generateLegacyDownloadPath will return the Resource path +// from the older style prefix+name configuration. +func (c *Config) generateLegacyResourcePath() string { + joined := path.Join(*c.Prefix, c.Name) + return strings.TrimPrefix(joined, "/") } const ( diff --git a/plugins/bundle/config_test.go b/plugins/bundle/config_test.go index 8bda343493..277c53ae35 100644 --- a/plugins/bundle/config_test.go +++ b/plugins/bundle/config_test.go @@ -6,6 +6,7 @@ package bundle import ( "fmt" + "gopkg.in/yaml.v2" "testing" ) @@ -89,3 +90,226 @@ func TestConfigCorrupted(t *testing.T) { t.Fatalf("want %v got %v", "bundles", *(config.Prefix)) } } + +func TestLegacyDownloadPath(t *testing.T) { + testCases := []struct { + prefix string + name string + result string + }{ + { + prefix: "/", + name: "bundles/bundles.tar.gz", + result: "bundles/bundles.tar.gz", + }, + { + prefix: "bundles", + name: "bundles.tar.gz", + result: "bundles/bundles.tar.gz", + }, + { + prefix: "", + name: "bundles/bundles.tar.gz", + result: "bundles/bundles.tar.gz", + }, + { + prefix: "", + name: "/bundles.tar.gz", + result: "bundles.tar.gz", + }, + } + for i, test := range testCases { + t.Run(fmt.Sprintf("case_%d", i), func(t *testing.T) { + config := Config{ + Name: test.name, + Prefix: &test.prefix, + } + + bs, err := yaml.Marshal(&config) + if err != nil { + t.Fatalf("Unexpected error marshalling config: %s", err) + } + + parsed, err := ParseConfig(bs, []string{"service1"}) + if err != nil { + t.Fatalf("Unexpected error parsing config: %s", err) + } + + b, ok := parsed.Bundles[test.name] + if !ok { + t.Fatalf("Expected resource %q on bundle with name %q", test.result, test.name) + } + + if b.Resource != test.result { + t.Errorf("Expected resource %q on bundle with name %q, actual: %s", test.result, test.name, b.Resource) + } + }) + } +} + +func TestParseAndValidateBundlesConfig(t *testing.T) { + tests := []struct { + conf string + services []string + wantError bool + }{ + { + conf: "", + services: []string{}, + wantError: false, + }, + { + conf: "{{{", + services: []string{}, + wantError: true, + }, + { + conf: `{"b1":{"service": "s1"}}`, + services: []string{}, + wantError: true, + }, + { + conf: `{"b1":{"service": "s1"}}`, + services: []string{"s1"}, + wantError: false, + }, + { + conf: `{"b1":{"service": "s1"}, "b2":{"service": "s1"}}`, + services: []string{"s1"}, + wantError: false, + }, + { + conf: `{"b1":{"service": "s1"}, "b2":{"service": "s2"}}`, + services: []string{"s1"}, + wantError: true, + }, + { + conf: `{"b1":{"service": "s1"}, "b2":{"service": "s2"}}`, + services: []string{"s1", "s2"}, + wantError: false, + }, + { + conf: `{"b1":{"service": "s1", "polling": {"min_delay_seconds": 1, "max_delay_seconds": 5}}}`, + services: []string{"s1"}, + wantError: false, + }, + { + conf: `{"b1":{"service": "s1", "polling": {"min_delay_seconds": 5, "max_delay_seconds": 1}}}`, + services: []string{"s1"}, + wantError: true, + }, + } + + for i := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + _, err := ParseBundlesConfig([]byte(tests[i].conf), tests[i].services) + if err != nil && !tests[i].wantError { + t.Fatalf("Unexpected error: %s", err) + } + if err == nil && tests[i].wantError { + t.Fatalf("Expected an error but didn't get one") + } + }) + } +} + +func TestParseBundlesConfig(t *testing.T) { + conf := []byte(` +bundle.tar.gz: + service: s1 +b2: + service: s1 + resource: /b2/path/ +b3: + service: s3 + resource: /some/longer/path/bundle.tar.gz +`) + services := []string{"s1", "s3"} + parsedConfig, err := ParseBundlesConfig(conf, services) + if err != nil { + t.Fatalf("Unexpected error: %s", err) + } + + if parsedConfig.Name != "" { + t.Fatalf("Expected config `Name` to be empty, actual: %s", parsedConfig.Name) + } + + if len(parsedConfig.Bundles) != 3 { + t.Fatalf("Expected 3 bundles in parsed config, got: %+v", parsedConfig.Bundles) + } + + expectedSources := map[string]struct { + service string + resource string + }{ + "bundle.tar.gz": { + service: "s1", + resource: "bundles/bundle.tar.gz", + }, + "b2": { + service: "s1", + resource: "/b2/path/", + }, + "b3": { + service: "s3", + resource: "/some/longer/path/bundle.tar.gz", + }, + } + + for name, expected := range expectedSources { + actual, ok := parsedConfig.Bundles[name] + if !ok { + t.Fatalf("Expected to have bundle with name %s configured, actual: %+v", name, parsedConfig.Bundles) + } + if expected.resource != actual.Resource { + t.Errorf("Expected resource '%s', found '%s'", expected.resource, actual.Resource) + } + if expected.service != actual.Service { + t.Errorf("Expected service '%s', found '%s'", expected.service, actual.Service) + } + } +} + +func TestConfigIsMultiBundle(t *testing.T) { + tests := []struct { + conf Config + expected bool + }{ + { + conf: Config{}, + expected: true, + }, + { + conf: Config{Name: "bundle.tar.gz"}, + expected: false, + }, + { + conf: Config{ + Name: "bundle.tar.gz", + Bundles: map[string]*Source{ + "bundle.tar.gz": &Source{}, + }, + }, + expected: false, + }, + { + conf: Config{ + Name: "", + Bundles: map[string]*Source{ + "bundle.tar.gz": &Source{}, + }, + }, + expected: true, + }, + } + + for i := range tests { + t.Run(fmt.Sprintf("%d", i), func(t *testing.T) { + actual := tests[i].conf.IsMultiBundle() + if actual != tests[i].expected { + t.Errorf("expected %t but got %t", tests[i].expected, actual) + } + }) + } + +} diff --git a/plugins/bundle/plugin.go b/plugins/bundle/plugin.go index ddad4b76b2..4224e61880 100644 --- a/plugins/bundle/plugin.go +++ b/plugins/bundle/plugin.go @@ -7,6 +7,7 @@ package bundle import ( "context" + "errors" "fmt" "reflect" "strings" @@ -15,7 +16,6 @@ import ( "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/internal/manifest" "github.com/open-policy-agent/opa/plugins" "github.com/open-policy-agent/opa/storage" "github.com/sirupsen/logrus" @@ -23,25 +23,35 @@ import ( // Plugin implements bundle activation. type Plugin struct { - 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 + config Config + manager *plugins.Manager // plugin manager for storage and service clients + status map[string]*Status // current status for each bundle + etags map[string]string // etag on last successful activation + listeners map[interface{}]func(Status) // listeners to send status updates to + bulkListeners map[interface{}]func(map[string]*Status) // listeners to send aggregated status updates to + downloaders map[string]*download.Downloader + mtx sync.Mutex + cfgMtx sync.Mutex + legacyConfig bool } // New returns a new Plugin with the given config. func New(parsedConfig *Config, manager *plugins.Manager) *Plugin { - p := &Plugin{ - manager: manager, - config: *parsedConfig, - status: &Status{ - Name: parsedConfig.Name, - }, + initialStatus := map[string]*Status{} + for name := range parsedConfig.Bundles { + initialStatus[name] = &Status{ + Name: name, + } } - p.initDownloader() + + p := &Plugin{ + manager: manager, + config: *parsedConfig, + status: initialStatus, + downloaders: make(map[string]*download.Downloader), + etags: make(map[string]string), + } + p.initDownloaders() return p } @@ -60,40 +70,109 @@ func Lookup(manager *plugins.Manager) *Plugin { // 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 { - p.logInfo("Starting bundle downloader.") p.mtx.Lock() defer p.mtx.Unlock() - p.downloader.Start(ctx) + for name, dl := range p.downloaders { + p.logInfo(name, "Starting bundle downloader.") + dl.Start(ctx) + } return nil } // Stop stops the plugin. func (p *Plugin) Stop(ctx context.Context) { - p.logInfo("Stopping bundle downloader.") p.mtx.Lock() defer p.mtx.Unlock() - p.downloader.Stop(ctx) + for name, dl := range p.downloaders { + p.logInfo(name, "Stopping bundle downloader.") + dl.Stop(ctx) + } } // Reconfigure notifies the plugin that it's configuration has changed. +// Any bundle configs that have changed or been added/removed will take +// affect. func (p *Plugin) Reconfigure(ctx context.Context, config interface{}) { - p.mtx.Lock() - defer p.mtx.Unlock() + // Reconfiguring should not occur in parallel, lock to ensure + // nothing swaps underneath us with the current p.config and the updated one. + // Use p.cfgMtx instead of p.mtx so as to not block any bundle downloads/activations + // that are in progress. We upgrade to p.mtx locking after stopping downloaders. + p.cfgMtx.Lock() + defer p.cfgMtx.Unlock() + // Look for any bundles that have had their config changed, are new, or have been removed newConfig := config.(*Config) - if reflect.DeepEqual(p.config, *newConfig) { - p.logDebug("Bundle downloader configuration unchanged.") + newBundles, updatedBundles, deletedBundles := p.configDelta(newConfig) + p.config = *newConfig + + if len(updatedBundles) == 0 && len(newBundles) == 0 && len(deletedBundles) == 0 { + // no relevant config changes return } - p.logInfo("Bundle downloader configuration changed. Restarting bundle downloader.") - p.config = *config.(*Config) - p.downloader.Stop(ctx) - p.initDownloader() - p.downloader.Start(ctx) + // Stop the downloaders outside p.mtx to allow them to finish handling any in-progress requests. + for name, dl := range p.downloaders { + _, updated := updatedBundles[name] + _, deleted := deletedBundles[name] + if updated || deleted { + dl.Stop(ctx) + } + } + + // Only lock p.mtx once we start changing the internal maps + // and downloader configs. + p.mtx.Lock() + defer p.mtx.Unlock() + + // Cleanup existing downloaders that are deleted + for name := range p.downloaders { + if _, deleted := deletedBundles[name]; deleted { + p.logInfo(name, "Bundle downloader configuration removed. Stopping bundle downloader.") + delete(p.downloaders, name) + delete(p.status, name) + delete(p.etags, name) + } + } + + // Deactivate the bundles that were removed + params := storage.WriteParams + params.Context = storage.NewContext() + err := storage.Txn(ctx, p.manager.Store, params, func(txn storage.Transaction) error { + for name := range deletedBundles { + _, err := p.deactivate(ctx, txn, name, nil) + if err != nil { + p.logError(name, "Failed to deactivate bundle: %s", err) + return err + } + } + return nil + }) + if err != nil { + // TODO(patrick-east): This probably shouldn't panic.. But OPA shouldn't + // continue in a potentially inconsistent state. + panic(errors.New("Unable deactivate bundle: " + err.Error())) + } + + for name, source := range p.config.Bundles { + _, updated := updatedBundles[name] + _, isNew := newBundles[name] + + if isNew || updated { + if isNew { + p.status[name] = &Status{Name: name} + p.logInfo(name, "New bundle downloader configuration added. Starting bundle downloader.") + } else { + p.logInfo(name, "Bundle downloader configuration changed. Restarting bundle downloader.") + } + p.downloaders[name] = p.newDownloader(name, source) + p.downloaders[name].Start(ctx) + } + } } // Register a listener to receive status updates. The name must be comparable. +// The listener will receive a status update for each bundle configured, they are +// not going to be aggregated. For all status updates use `RegisterBulkListener`. func (p *Plugin) Register(name interface{}, listener func(Status)) { p.mtx.Lock() defer p.mtx.Unlock() @@ -110,107 +189,123 @@ func (p *Plugin) Unregister(name interface{}) { p.mtx.Lock() defer p.mtx.Unlock() - delete(p.listeners, name) + delete(p.bulkListeners, name) } -func (p *Plugin) initDownloader() { - client := p.manager.Client(p.config.Service) - path := p.generateDownloadPath(*(p.config.Prefix), p.config.Name) - p.downloader = download.New(p.config.Config, client, path).WithCallback(p.oneShot) -} - -func (p *Plugin) generateDownloadPath(prefix string, name string) string { - res := "" - trimmedPrefix := strings.Trim(prefix, "/") - if trimmedPrefix != "" { - res += trimmedPrefix + "/" - } - - res += strings.Trim(name, "/") - - return res -} - -func (p *Plugin) oneShot(ctx context.Context, u download.Update) { +// RegisterBulkListener registers a listener to receive bulk (aggregated) status updates. The name must be comparable. +func (p *Plugin) RegisterBulkListener(name interface{}, listener func(map[string]*Status)) { p.mtx.Lock() defer p.mtx.Unlock() - p.process(ctx, u) - status := *p.status + if p.bulkListeners == nil { + p.bulkListeners = map[interface{}]func(map[string]*Status){} + } - for _, listener := range p.listeners { - listener(status) + p.bulkListeners[name] = listener +} + +// UnregisterBulkListener unregisters a listener to stop receiving aggregated status updates. +func (p *Plugin) UnregisterBulkListener(name interface{}) { + p.mtx.Lock() + defer p.mtx.Unlock() + + delete(p.bulkListeners, name) +} + +// Config returns the plugins current configuration +func (p *Plugin) Config() *Config { + return &p.config +} + +func (p *Plugin) initDownloaders() { + // Initialize a downloader for each bundle configured. + for name, source := range p.config.Bundles { + p.downloaders[name] = p.newDownloader(name, source) } } -func (p *Plugin) process(ctx context.Context, u download.Update) { +func (p *Plugin) newDownloader(name string, source *Source) *download.Downloader { + conf := source.Config + client := p.manager.Client(source.Service) + path := source.Resource + return download.New(conf, client, path).WithCallback(func(ctx context.Context, u download.Update) { + // wrap the callback to include the name of the bundle that was updated + p.oneShot(ctx, name, u) + }) +} + +func (p *Plugin) oneShot(ctx context.Context, name string, u download.Update) { + p.mtx.Lock() + defer p.mtx.Unlock() + + p.process(ctx, name, u) + + for _, listener := range p.listeners { + listener(*p.status[name]) + } + + for _, listener := range p.bulkListeners { + listener(p.status) + } +} + +func (p *Plugin) process(ctx context.Context, name string, u download.Update) { if u.Error != nil { - p.logError("Bundle download failed: %v", u.Error) - p.status.SetError(u.Error) + p.logError(name, "Bundle download failed: %v", u.Error) + p.status[name].SetError(u.Error) return } if u.Bundle != nil { - p.status.SetDownloadSuccess() + p.status[name].SetDownloadSuccess() - if err := p.activate(ctx, u.Bundle); err != nil { - p.logError("Bundle activation failed: %v", err) - p.status.SetError(err) + if err := p.activate(ctx, name, u.Bundle); err != nil { + p.logError(name, "Bundle activation failed: %v", err) + p.status[name].SetError(err) return } - p.status.SetError(nil) - p.status.SetActivateSuccess(u.Bundle.Manifest.Revision) + p.status[name].SetError(nil) + p.status[name].SetActivateSuccess(u.Bundle.Manifest.Revision) if u.ETag != "" { - p.logInfo("Bundle downloaded and activated successfully. Etag updated to %v.", u.ETag) + p.logInfo(name, "Bundle downloaded and activated successfully. Etag updated to %v.", u.ETag) } else { - p.logInfo("Bundle downloaded and activated successfully.") + p.logInfo(name, "Bundle downloaded and activated successfully.") } - p.etag = u.ETag + p.etags[name] = u.ETag return } - if u.ETag == p.etag { - p.logDebug("Bundle download skipped, server replied with not modified.") - p.status.SetError(nil) + if etag, ok := p.etags[name]; ok && u.ETag == etag { + p.logDebug(name, "Bundle download skipped, server replied with not modified.") + p.status[name].SetError(nil) return } } -func (p *Plugin) activate(ctx context.Context, b *bundle.Bundle) error { - p.logDebug("Bundle activation in progress. Opening storage transaction.") +func (p *Plugin) activate(ctx context.Context, name string, b *bundle.Bundle) error { + p.logDebug(name, "Bundle activation in progress. Opening storage transaction.") params := storage.WriteParams params.Context = storage.NewContext() return storage.Txn(ctx, p.manager.Store, params, func(txn storage.Transaction) error { - p.logDebug("Opened storage transaction (%v).", txn.ID()) - defer p.logDebug("Closing storage transaction (%v).", txn.ID()) + p.logDebug(name, "Opened storage transaction (%v).", txn.ID()) + defer p.logDebug(name, "Closing storage transaction (%v).", txn.ID()) - // Build set of roots from old and new bundles. This set of - // roots should be erased. - erase := map[string]struct{}{} + // Erase data at new roots to prepare for writing the new data + newRoots := map[string]struct{}{} if b.Manifest.Roots != nil { for _, root := range *b.Manifest.Roots { - erase[root] = struct{}{} + newRoots[root] = struct{}{} } } - if roots, err := manifest.ReadBundleRoots(ctx, p.manager.Store, txn); err == nil { - for _, root := range roots { - erase[root] = struct{}{} - } - } else if !storage.IsNotFound(err) { - return err - } - - if err := p.eraseData(ctx, txn, erase); err != nil { - return err - } - - remaining, err := p.erasePolicies(ctx, txn, erase) + // Erase data and policies at new + old roots, and remove the old + // manifest before activating a new bundle. + remaining, err := p.deactivate(ctx, txn, name, newRoots) if err != nil { return err } @@ -226,9 +321,16 @@ func (p *Plugin) activate(ctx context.Context, b *bundle.Bundle) error { return err } - if err := manifest.Write(ctx, p.manager.Store, txn, b.Manifest); err != nil { + // Always write manifests to the named location. If the plugin is in the older style config + // then also write to the old legacy unnamed location. + if err := bundle.WriteManifestToStore(ctx, p.manager.Store, txn, name, b.Manifest); err != nil { return err } + if !p.config.IsMultiBundle() { + if err := bundle.LegacyWriteManifestToStore(ctx, p.manager.Store, txn, b.Manifest); err != nil { + return err + } + } plugins.SetCompilerOnContext(params.Context, compiler) @@ -236,6 +338,45 @@ func (p *Plugin) activate(ctx context.Context, b *bundle.Bundle) error { }) } +// deactivate a bundle by name. This will clear all policies and data at its roots and remove its +// manifest from storage. If additionalRoots are provided they will be deleted along with the +// roots found in storage for the bundle. +func (p *Plugin) deactivate(ctx context.Context, txn storage.Transaction, name string, additionalRoots map[string]struct{}) (map[string]*ast.Module, error) { + erase := additionalRoots + if erase == nil { + erase = map[string]struct{}{} + } + + if roots, err := bundle.ReadBundleRootsFromStore(ctx, p.manager.Store, txn, name); err == nil { + for _, root := range roots { + erase[root] = struct{}{} + } + } else if !storage.IsNotFound(err) { + return nil, err + } + + p.logDebug(name, "Erasing data and polices with roots at %+v", erase) + + if err := p.eraseData(ctx, txn, erase); err != nil { + return nil, err + } + + remaining, err := p.erasePolicies(ctx, txn, erase) + if err != nil { + return nil, err + } + + if err := bundle.EraseManifestFromStore(ctx, p.manager.Store, txn, name); err != nil && !storage.IsNotFound(err) { + return nil, err + } + + if err := bundle.LegacyEraseManifestFromStore(ctx, p.manager.Store, txn); err != nil && !storage.IsNotFound(err) { + return nil, err + } + + return remaining, nil +} + func (p *Plugin) eraseData(ctx context.Context, txn storage.Transaction, roots map[string]struct{}) error { for root := range roots { path, ok := storage.ParsePathEscaped("/" + root) @@ -334,23 +475,49 @@ func (p *Plugin) writeModules(ctx context.Context, txn storage.Transaction, file return compiler, nil } -func (p *Plugin) logError(fmt string, a ...interface{}) { - logrus.WithFields(p.logrusFields()).Errorf(fmt, a...) +func (p *Plugin) logError(bundleName string, fmt string, a ...interface{}) { + logrus.WithFields(p.logrusFields(bundleName)).Errorf(fmt, a...) } -func (p *Plugin) logInfo(fmt string, a ...interface{}) { - logrus.WithFields(p.logrusFields()).Infof(fmt, a...) +func (p *Plugin) logInfo(bundleName string, fmt string, a ...interface{}) { + logrus.WithFields(p.logrusFields(bundleName)).Infof(fmt, a...) } -func (p *Plugin) logDebug(fmt string, a ...interface{}) { - logrus.WithFields(p.logrusFields()).Debugf(fmt, a...) +func (p *Plugin) logDebug(bundleName string, fmt string, a ...interface{}) { + logrus.WithFields(p.logrusFields(bundleName)).Debugf(fmt, a...) } -func (p *Plugin) logrusFields() logrus.Fields { - return logrus.Fields{ +func (p *Plugin) logrusFields(bundleName string) logrus.Fields { + + f := logrus.Fields{ "plugin": Name, - "name": p.config.Name, + "name": bundleName, } + + return f +} + +// configDelta will return a map of new bundle sources, updated bundle sources, and a set of deleted bundle names +func (p *Plugin) configDelta(newConfig *Config) (map[string]*Source, map[string]*Source, map[string]struct{}) { + deletedBundles := map[string]struct{}{} + for name := range p.config.Bundles { + deletedBundles[name] = struct{}{} + } + newBundles := map[string]*Source{} + updatedBundles := map[string]*Source{} + for name, source := range newConfig.Bundles { + oldSource, found := p.config.Bundles[name] + if !found { + newBundles[name] = source + } else { + delete(deletedBundles, name) + if !reflect.DeepEqual(oldSource, source) { + updatedBundles[name] = source + } + } + } + + return newBundles, updatedBundles, deletedBundles } func lookup(path storage.Path, data map[string]interface{}) (interface{}, bool) { diff --git a/plugins/bundle/plugin_test.go b/plugins/bundle/plugin_test.go index 8b1004c612..c3397088cb 100644 --- a/plugins/bundle/plugin_test.go +++ b/plugins/bundle/plugin_test.go @@ -8,14 +8,17 @@ import ( "bytes" "context" "fmt" + "net/http" + "net/http/httptest" "reflect" "sort" + "strings" "testing" "github.com/open-policy-agent/opa/ast" "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/internal/manifest" "github.com/open-policy-agent/opa/plugins" "github.com/open-policy-agent/opa/storage" "github.com/open-policy-agent/opa/storage/inmem" @@ -26,7 +29,9 @@ func TestPluginOneShot(t *testing.T) { ctx := context.Background() manager := getTestManager() - plugin := Plugin{manager: manager, status: &Status{}} + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} module := "package foo\n\ncorge=1" @@ -44,7 +49,7 @@ func TestPluginOneShot(t *testing.T) { b.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: &b}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) txn := storage.NewTransactionOrDie(ctx, manager.Store) defer manager.Store.Abort(ctx, txn) @@ -65,7 +70,7 @@ func TestPluginOneShot(t *testing.T) { } data, err := manager.Store.Read(ctx, txn, storage.Path{}) - expData := util.MustUnmarshalJSON([]byte(`{"foo": {"bar": 1, "baz": "qux"}, "system": {"bundle": {"manifest": {"revision": "quickbrownfaux", "roots": [""]}}}}`)) + expData := util.MustUnmarshalJSON([]byte(`{"foo": {"bar": 1, "baz": "qux"}, "system": {"bundles": {"test-bundle": {"manifest": {"revision": "quickbrownfaux", "roots": [""]}}}}}`)) if err != nil { t.Fatal(err) } else if !reflect.DeepEqual(data, expData) { @@ -78,7 +83,9 @@ func TestPluginOneShotCompileError(t *testing.T) { ctx := context.Background() manager := getTestManager() - plugin := Plugin{manager: manager, status: &Status{}} + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} raw1 := "package foo\n\np[x] { x = 1 }" b1 := &bundle.Bundle{ @@ -93,7 +100,7 @@ func TestPluginOneShotCompileError(t *testing.T) { } b1.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: b1}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: b1}) b2 := &bundle.Bundle{ Data: map[string]interface{}{"a": "b"}, @@ -106,7 +113,7 @@ func TestPluginOneShotCompileError(t *testing.T) { } b2.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: b2}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: b2}) txn := storage.NewTransactionOrDie(ctx, manager.Store) _, err := manager.Store.GetPolicy(ctx, txn, "/example.rego") @@ -132,7 +139,7 @@ func TestPluginOneShotCompileError(t *testing.T) { } b3.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: b3}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: b3}) txn = storage.NewTransactionOrDie(ctx, manager.Store) @@ -148,11 +155,13 @@ func TestPluginOneShotCompileError(t *testing.T) { } -func TestPluginOneShotActivatationRemovesOld(t *testing.T) { +func TestPluginOneShotActivationRemovesOld(t *testing.T) { ctx := context.Background() manager := getTestManager() - plugin := Plugin{manager: manager, status: &Status{}} + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} module1 := `package example @@ -172,7 +181,7 @@ func TestPluginOneShotActivatationRemovesOld(t *testing.T) { } b1.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: &b1}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b1}) module2 := `package example @@ -192,7 +201,7 @@ func TestPluginOneShotActivatationRemovesOld(t *testing.T) { } b2.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: &b2}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b2}) err := storage.Txn(ctx, manager.Store, storage.TransactionParams{}, func(txn storage.Transaction) error { ids, err := manager.Store.ListPolicies(ctx, txn) @@ -221,7 +230,9 @@ func TestPluginListener(t *testing.T) { ctx := context.Background() manager := getTestManager() - plugin := Plugin{manager: manager, status: &Status{}} + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} ch := make(chan Status) plugin.Register("test", func(status Status) { @@ -248,7 +259,7 @@ func TestPluginListener(t *testing.T) { // 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}) + go plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) s1 := <-ch if s1.ActiveRevision != "quickbrownfaux" || s1.Code != "" { @@ -265,7 +276,7 @@ func TestPluginListener(t *testing.T) { } // Test that next update is failed. - go plugin.oneShot(ctx, download.Update{Bundle: &b}) + go plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) s2 := <-ch if s2.ActiveRevision != "quickbrownfaux" || s2.Code == "" || s2.Message == "" || len(s2.Errors) == 0 { @@ -281,7 +292,7 @@ func TestPluginListener(t *testing.T) { } // Test that new update is successful. - go plugin.oneShot(ctx, download.Update{Bundle: &b}) + go plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) s3 := <-ch if s3.ActiveRevision != "fancybluederg" || s3.Code != "" || s3.Message != "" || len(s3.Errors) != 0 { @@ -289,7 +300,7 @@ func TestPluginListener(t *testing.T) { } // Test that empty download update results in status update. - go plugin.oneShot(ctx, download.Update{}) + go plugin.oneShot(ctx, bundleName, download.Update{}) s4 := <-ch if !reflect.DeepEqual(s3, s4) { @@ -301,7 +312,9 @@ func TestPluginListener(t *testing.T) { func TestPluginListenerErrorClearedOn304(t *testing.T) { ctx := context.Background() manager := getTestManager() - plugin := Plugin{manager: manager, status: &Status{}} + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} ch := make(chan Status) plugin.Register("test", func(status Status) { @@ -318,7 +331,7 @@ func TestPluginListenerErrorClearedOn304(t *testing.T) { b.Manifest.Init() // Test that initial bundle is ok. - go plugin.oneShot(ctx, download.Update{Bundle: &b}) + go plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) s1 := <-ch if s1.ActiveRevision != "quickbrownfaux" || s1.Code != "" { @@ -326,7 +339,7 @@ func TestPluginListenerErrorClearedOn304(t *testing.T) { } // Test that service error triggers failure notification. - go plugin.oneShot(ctx, download.Update{Error: fmt.Errorf("some error")}) + go plugin.oneShot(ctx, bundleName, download.Update{Error: fmt.Errorf("some error")}) s2 := <-ch if s2.ActiveRevision != "quickbrownfaux" || s2.Code == "" { @@ -334,7 +347,7 @@ func TestPluginListenerErrorClearedOn304(t *testing.T) { } // Test that service recovery triggers healthy notification. - go plugin.oneShot(ctx, download.Update{}) + go plugin.oneShot(ctx, bundleName, download.Update{}) s3 := <-ch if s3.ActiveRevision != "quickbrownfaux" || s3.Code != "" { @@ -342,11 +355,183 @@ func TestPluginListenerErrorClearedOn304(t *testing.T) { } } +func TestPluginBulkListener(t *testing.T) { + ctx := context.Background() + manager := getTestManager() + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleNames := []string{ + "b1", + "b2", + "b3", + } + for _, name := range bundleNames { + plugin.status[name] = &Status{Name: name} + } + bulkChan := make(chan map[string]*Status) + + plugin.RegisterBulkListener("bulk test", func(status map[string]*Status) { + bulkChan <- status + }) + + 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), + }, + }, + } + + b.Manifest.Init() + + // Test that initial bundle is ok. Defer to separate goroutine so we can + // check result with channel. + go plugin.oneShot(ctx, bundleNames[0], download.Update{Bundle: &b}) + s1 := <-bulkChan + + s := s1[bundleNames[0]] + if s.ActiveRevision != "quickbrownfaux" || s.Code != "" { + t.Fatal("Unexpected status update, got:", s1) + } + + for i := 1; i < len(bundleNames); i++ { + name := bundleNames[i] + s, ok := s1[name] + if !ok { + t.Errorf("Expected to have bundle status for %q included in update, got: %+v", name, s1) + } + // they should be defaults at this point + if !reflect.DeepEqual(s, &Status{Name: name}) { + t.Errorf("Expected bundle %q to have an empty status, got: %+v", name, s1) + } + } + + module = "package gork\np[x]" + + b.Manifest.Revision = "slowgreenburd" + b.Modules[0] = bundle.ModuleFile{ + Path: "/foo.rego", + Raw: []byte(module), + Parsed: ast.MustParseModule(module), + } + + // Test that next update is failed. + go plugin.oneShot(ctx, bundleNames[0], download.Update{Bundle: &b}) + s2 := <-bulkChan + + s = s2[bundleNames[0]] + if s.ActiveRevision != "quickbrownfaux" || s.Code == "" || s.Message == "" || len(s.Errors) == 0 { + t.Fatal("Unexpected status update, got:", s2) + } + + for i := 1; i < len(bundleNames); i++ { + name := bundleNames[i] + s, ok := s2[name] + if !ok { + t.Errorf("Expected to have bundle status for %q included in update, got: %+v", name, s2) + } + // they should be still defaults + if !reflect.DeepEqual(s, &Status{Name: name}) { + t.Errorf("Expected bundle %q to have an empty status, got: %+v", name, s2) + } + } + + module = "package gork\np[1]" + b.Manifest.Revision = "fancybluederg" + b.Modules[0] = bundle.ModuleFile{ + Path: "/foo.rego", + Raw: []byte(module), + Parsed: ast.MustParseModule(module), + } + + // Test that new update is successful. + go plugin.oneShot(ctx, bundleNames[0], download.Update{Bundle: &b}) + s3 := <-bulkChan + + s = s3[bundleNames[0]] + if s.ActiveRevision != "fancybluederg" || s.Code != "" || s.Message != "" || len(s.Errors) != 0 { + t.Fatal("Unexpected status update, got:", s3) + } + + for i := 1; i < len(bundleNames); i++ { + name := bundleNames[i] + s, ok := s3[name] + if !ok { + t.Errorf("Expected to have bundle status for %q included in update, got: %+v", name, s3) + } + // they should still be defaults + if !reflect.DeepEqual(s, &Status{Name: name}) { + t.Errorf("Expected bundle %q to have an empty status, got: %+v", name, s3) + } + } + + // Test that empty download update results in status update. + go plugin.oneShot(ctx, bundleNames[0], download.Update{}) + s4 := <-bulkChan + + if !reflect.DeepEqual(s3, s4) { + t.Fatalf("Expected: %v but got: %v", s3, s4) + } + + // Test updates the other bundles + module = "package p1\np[x] { x = 1 }" + + b1 := bundle.Bundle{ + Manifest: bundle.Manifest{ + Revision: "123", + }, + Data: map[string]interface{}{}, + Modules: []bundle.ModuleFile{ + { + Path: "/foo1.rego", + Parsed: ast.MustParseModule(module), + Raw: []byte(module), + }, + }, + } + + b1.Manifest.Init() + + // Test that new update is successful. + go plugin.oneShot(ctx, bundleNames[1], download.Update{Bundle: &b1}) + s5 := <-bulkChan + + s = s5[bundleNames[1]] + if s.ActiveRevision != "123" || s.Code != "" || s.Message != "" || len(s.Errors) != 0 { + t.Fatal("Unexpected status update, got:", s5) + } + + if !reflect.DeepEqual(s5[bundleNames[0]], s4[bundleNames[0]]) { + t.Fatalf("Expected bundle %q to have the same status as before updating bundle %q, got: %+v", bundleNames[0], bundleNames[1], s5) + } + + for i := 2; i < len(bundleNames); i++ { + name := bundleNames[i] + s, ok := s5[name] + if !ok { + t.Errorf("Expected to have bundle status for %q included in update, got: %+v", name, s5) + } + // they should still be defaults + if !reflect.DeepEqual(s, &Status{Name: name}) { + t.Errorf("Expected bundle %q to have an empty status, got: %+v", name, s5) + } + } +} + func TestPluginActivateScopedBundle(t *testing.T) { ctx := context.Background() manager := getTestManager() - plugin := Plugin{manager: manager, status: &Status{}} + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} // Transact test data and policies that represent data coming from // _outside_ the bundle. The test will verify that data _outside_ @@ -398,38 +583,13 @@ func TestPluginActivateScopedBundle(t *testing.T) { b.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: &b}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) - // Ensure a/a3-6 are intact. a1-2 are overwritten by bundle. - if err := storage.Txn(ctx, manager.Store, storage.TransactionParams{}, func(txn storage.Transaction) error { - value, err := manager.Store.Read(ctx, txn, storage.Path{"a"}) - if err != nil { - return err - } - - expData := util.MustUnmarshalJSON([]byte(`{"a1": "foo", "a3": "x2", "a5": "x3"}`)) - - if !reflect.DeepEqual(value, expData) { - return fmt.Errorf("Expected %v but got %v", expData, value) - } - - ids, err := manager.Store.ListPolicies(ctx, txn) - if err != nil { - return err - } - - expIds := []string{"bundle/id1", "some/id2", "some/id3"} - sort.Strings(ids) - - if !reflect.DeepEqual(ids, expIds) { - return fmt.Errorf("Expected ids %v but got %v", expIds, ids) - } - - return nil - - }); err != nil { - t.Fatal(err) - } + // Ensure a/a3-6 are intact. a1-2 are overwritten by bundle, and + // that the manifest has been written to storage. + expData := util.MustUnmarshalJSON([]byte(`{"a1": "foo", "a3": "x2", "a5": "x3"}`)) + expIds := []string{"bundle/id1", "some/id2", "some/id3"} + validateStoreState(ctx, t, manager.Store, "/a", expData, expIds, bundleName, "quickbrownfaux") // Activate a bundle that is scoped to a/a3 ad a/a6. Include a function // inside package a.a4 that we can depend on outside of the bundle scope to @@ -437,7 +597,7 @@ func TestPluginActivateScopedBundle(t *testing.T) { module = "package a.a4\n\nbar=1\n\nfunc(x) = x" b = bundle.Bundle{ - Manifest: bundle.Manifest{Revision: "quickbrownfaux", Roots: &[]string{"a/a3", "a/a4"}}, + Manifest: bundle.Manifest{Revision: "quickbrownfaux-2", Roots: &[]string{"a/a3", "a/a4"}}, Data: map[string]interface{}{ "a": map[string]interface{}{ "a3": "foo", @@ -453,38 +613,12 @@ func TestPluginActivateScopedBundle(t *testing.T) { } b.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: &b}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) // Ensure a/a5-a6 are intact. a3 and a4 are overwritten by bundle. - if err := storage.Txn(ctx, manager.Store, storage.TransactionParams{}, func(txn storage.Transaction) error { - value, err := manager.Store.Read(ctx, txn, storage.Path{"a"}) - if err != nil { - return err - } - - expData := util.MustUnmarshalJSON([]byte(`{"a3": "foo", "a5": "x3"}`)) - - if !reflect.DeepEqual(value, expData) { - return fmt.Errorf("Expected %v but got %v", expData, value) - } - - ids, err := manager.Store.ListPolicies(ctx, txn) - if err != nil { - return err - } - - expIds := []string{"bundle/id2", "some/id3"} - sort.Strings(ids) - - if !reflect.DeepEqual(ids, expIds) { - return fmt.Errorf("Expected ids %v but got %v", expIds, ids) - } - - return nil - - }); err != nil { - t.Fatal(err) - } + expData = util.MustUnmarshalJSON([]byte(`{"a3": "foo", "a5": "x3"}`)) + expIds = []string{"bundle/id2", "some/id3"} + validateStoreState(ctx, t, manager.Store, "/a", expData, expIds, bundleName, "quickbrownfaux-2") // Upsert policy outside of bundle scope that depends on bundle. if err := storage.Txn(ctx, manager.Store, storage.WriteParams, func(txn storage.Transaction) error { @@ -494,35 +628,27 @@ func TestPluginActivateScopedBundle(t *testing.T) { } b = bundle.Bundle{ - Manifest: bundle.Manifest{Revision: "quickbrownfaux-2", Roots: &[]string{"a/a3", "a/a4"}}, + Manifest: bundle.Manifest{Revision: "quickbrownfaux-3", Roots: &[]string{"a/a3", "a/a4"}}, Data: map[string]interface{}{}, Modules: []bundle.ModuleFile{}, } b.Manifest.Init() - plugin.oneShot(ctx, download.Update{Bundle: &b}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) // Ensure bundle activation failed by checking that previous revision is // still active. - if err := storage.Txn(ctx, manager.Store, storage.TransactionParams{}, func(txn storage.Transaction) error { - revision, err := manifest.ReadBundleRevision(ctx, manager.Store, txn) - if err != nil { - return err - } - if revision != "quickbrownfaux" { - return fmt.Errorf("Expected revision to be quickbrownfaux but got: %v", revision) - } - return nil - }); err != nil { - t.Fatal(err) - } + expIds = []string{"bundle/id2", "not_scoped", "some/id3"} + validateStoreState(ctx, t, manager.Store, "/a", expData, expIds, bundleName, "quickbrownfaux-2") } func TestPluginSetCompilerOnContext(t *testing.T) { ctx := context.Background() manager := getTestManager() - plugin := Plugin{manager: manager, status: &Status{}} + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} module := ` package test @@ -557,7 +683,7 @@ func TestPluginSetCompilerOnContext(t *testing.T) { t.Fatal(err) } - plugin.oneShot(ctx, download.Update{Bundle: &b}) + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) exp := ast.MustParseModule(module) @@ -580,40 +706,441 @@ func getTestManager() *plugins.Manager { return manager } -func TestInitDownloader(t *testing.T) { - plugin := Plugin{} +func TestPluginReconfigure(t *testing.T) { + tsURLBase := "/opa-test/" + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasPrefix(r.URL.Path, tsURLBase) { + t.Fatalf("Invalid request URL path: %s, expected prefix %s", r.URL.Path, tsURLBase) + } + fmt.Fprintln(w, "") // Note: this is an invalid bundle and will fail the download + })) + defer ts.Close() - testCases := []struct { - prefix string - name string - result string + ctx := context.Background() + manager := getTestManager() + + serviceName := "test-svc" + err := manager.Reconfigure(&config.Config{ + Services: []byte(fmt.Sprintf("{\"%s\":{ \"url\": \"%s\"}}", serviceName, ts.URL+tsURLBase)), + }) + if err != nil { + t.Fatalf("Error configuring plugin manager: %s", err) + } + + plugin := New(&Config{}, manager) + + var delay int64 = 10 + baseConf := download.Config{Polling: download.PollingConfig{MinDelaySeconds: &delay, MaxDelaySeconds: &delay}} + + // Note: test stages are accumulating state with reconfigures between them, the order does matter! + // Each stage defines the new config, side effects are validated. + stages := []struct { + name string + cfg *Config }{ { - prefix: "/", - name: "bundles/bundles.tar.gz", - result: "bundles/bundles.tar.gz", + name: "start with single legacy bundle", + cfg: &Config{ + Name: "bundle.tar.gz", + Service: serviceName, + Config: baseConf, + // Note: the config validation and default injection will add an entry + // to the Bundles map for the older style configuration. + Bundles: map[string]*Source{ + "bundle.tar.gz": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle.tar.gz"}, + }, + }, }, { - prefix: "bundles", - name: "bundles.tar.gz", - result: "bundles/bundles.tar.gz", + name: "switch to mutli-bundle", + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle.tar.gz"}, + }, + }, }, { - prefix: "", - name: "bundles/bundles.tar.gz", - result: "bundles/bundles.tar.gz", + name: "add second bundle", + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle1.tar.gz"}, + "b2": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle2.tar.gz"}, + }, + }, }, { - prefix: "", - name: "/bundles.tar.gz", - result: "bundles.tar.gz", + name: "remove initial bundle", + cfg: &Config{ + Bundles: map[string]*Source{ + "b2": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle2.tar.gz"}, + }, + }, + }, + { + name: "Update single bundle", + cfg: &Config{ + Bundles: map[string]*Source{ + "b2": {Config: baseConf, Service: serviceName, Resource: "/new/path/bundles/bundle2.tar.gz"}, + }, + }, + }, + { + name: "Add multiple new bundles", + cfg: &Config{ + Bundles: map[string]*Source{ + "b3": {Config: baseConf, Service: serviceName, Resource: "/bundle3.tar.gz"}, + "b4": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle4.tar.gz"}, + "b5": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle5.tar.gz"}, + }, + }, + }, + { + name: "Remove multiple bundles", + cfg: &Config{ + Bundles: map[string]*Source{ + "b2": {Config: baseConf, Service: serviceName, Resource: "/new/path/bundles/bundle2.tar.gz"}, + "b4": {Config: baseConf, Service: serviceName, Resource: "/bundles/bundle4.tar.gz"}, + }, + }, + }, + { + name: "Update multiple bundles", + cfg: &Config{ + Bundles: map[string]*Source{ + "b2": {Config: baseConf, Service: serviceName, Resource: "/update2/bundle2.tar.gz"}, + "b4": {Config: baseConf, Service: serviceName, Resource: "/update2/bundle4.tar.gz"}, + }, + }, + }, + { + name: "Remove and add bundle", + cfg: &Config{ + Bundles: map[string]*Source{ + "b6": {Config: baseConf, Service: serviceName, Resource: "bundle6.tar.gz"}, + }, + }, + }, + { + name: "Add and update bundle", + cfg: &Config{ + Bundles: map[string]*Source{ + "b6": {Config: baseConf, Service: serviceName, Resource: "/update3/bundle6.tar.gz"}, + "b7": {Config: baseConf, Service: serviceName, Resource: "bundle7.tar.gz"}, + "b8": {Config: baseConf, Service: serviceName, Resource: "bundle8.tar.gz"}, + }, + }, + }, + { + name: "Update and remove", + cfg: &Config{ + Bundles: map[string]*Source{ + "b6": {Config: baseConf, Service: serviceName, Resource: "/update4/bundle6.tar.gz"}, + "b8": {Config: baseConf, Service: serviceName, Resource: "bundle8.tar.gz"}, + }, + }, + }, + // Add, Update, and Remove + { + name: "Add update and remove", + cfg: &Config{ + Bundles: map[string]*Source{ + "b8": {Config: baseConf, Service: serviceName, Resource: "/update5/bundle8.tar.gz"}, + "b9": {Config: baseConf, Service: serviceName, Resource: "bundle9.tar.gz"}, + }, + }, }, } - for i, test := range testCases { - t.Run(fmt.Sprintf("case_%d", i), func(t *testing.T) { - if out := plugin.generateDownloadPath(test.prefix, test.name); out != test.result { - t.Fatalf("want %v got %v", test.result, out) + + for _, stage := range stages { + t.Run(stage.name, func(t *testing.T) { + + plugin.Reconfigure(ctx, stage.cfg) + + var expectedNumBundles int + if stage.cfg.Name != "" { + expectedNumBundles = 1 + } else { + expectedNumBundles = len(stage.cfg.Bundles) + } + + if expectedNumBundles != len(plugin.downloaders) { + t.Fatalf("Expected a downloader for each configured bundle, expected %d found %d", expectedNumBundles, len(plugin.downloaders)) + } + + if expectedNumBundles != len(plugin.status) { + t.Fatalf("Expected a status entry for each configured bundle, expected %d found %d", expectedNumBundles, len(plugin.status)) + } + + for name := range stage.cfg.Bundles { + if _, found := plugin.downloaders[name]; !found { + t.Fatalf("bundle %q not found in downloaders map", name) + } + + if _, found := plugin.status[name]; !found { + t.Fatalf("bundle %q not found in status map", name) + } } }) } } + +func TestUpgradeLegacyBundleToMuiltiBundleSameBundle(t *testing.T) { + + ctx := context.Background() + manager := getTestManager() + plugin := Plugin{manager: manager, status: map[string]*Status{}, etags: map[string]string{}} + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} + + // Start with a "legacy" style config for a single bundle + plugin.config = Config{ + Bundles: map[string]*Source{ + bundleName: &Source{ + Service: "s1", + }, + }, + Name: bundleName, + Service: "s1", + Prefix: nil, + } + + module := "package a.a1\n\nbar=1" + + b := bundle.Bundle{ + Manifest: bundle.Manifest{Revision: "quickbrownfaux", Roots: &[]string{"a/a1", "a/a2"}}, + Data: map[string]interface{}{ + "a": map[string]interface{}{ + "a2": "foo", + }, + }, + Modules: []bundle.ModuleFile{ + bundle.ModuleFile{ + Path: "bundle/id1", + Parsed: ast.MustParseModule(module), + Raw: []byte(module), + }, + }, + } + + b.Manifest.Init() + + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) + + // Ensure it has been activated + expData := util.MustUnmarshalJSON([]byte(`{"a2": "foo"}`)) + expIds := []string{"bundle/id1"} + validateStoreState(ctx, t, manager.Store, "/a", expData, expIds, bundleName, "quickbrownfaux") + + if plugin.config.IsMultiBundle() { + t.Fatalf("Expected plugin to be in non-multi bundle config mode") + } + + // Update to the newer style config with the same bundle + multiBundleConf := &Config{ + Bundles: map[string]*Source{ + bundleName: &Source{ + Service: "s1", + }, + }, + } + + plugin.Reconfigure(ctx, multiBundleConf) + b.Manifest.Revision = "quickbrownfaux-2" + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) + + // None of the data should have changed, only the revision + validateStoreState(ctx, t, manager.Store, "/a", expData, expIds, bundleName, "quickbrownfaux-2") + + // Make sure the legacy path is gone now that we are in multi-bundle mode + var actual string + err := storage.Txn(ctx, plugin.manager.Store, storage.WriteParams, func(txn storage.Transaction) error { + var err error + if actual, err = bundle.LegacyReadRevisionFromStore(ctx, plugin.manager.Store, txn); err != nil && !storage.IsNotFound(err) { + t.Fatalf("Failed to read manifest revision from store: %s", err) + return err + } + return nil + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + if actual != "" { + t.Fatalf("Expected to not find manifest revision but got %s", actual) + } + + if !plugin.config.IsMultiBundle() { + t.Fatalf("Expected plugin to be in multi bundle config mode") + } +} + +func TestUpgradeLegacyBundleToMuiltiBundleNewBundles(t *testing.T) { + ctx := context.Background() + manager := getTestManager() + plugin := Plugin{ + manager: manager, + status: map[string]*Status{}, + etags: map[string]string{}, + downloaders: map[string]*download.Downloader{}, + } + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} + + tsURLBase := "/opa-test/" + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasPrefix(r.URL.Path, tsURLBase) { + t.Fatalf("Invalid request URL path: %s, expected prefix %s", r.URL.Path, tsURLBase) + } + fmt.Fprintln(w, "") // Note: this is an invalid bundle and will fail the download + })) + defer ts.Close() + + serviceName := "test-svc" + err := manager.Reconfigure(&config.Config{ + Services: []byte(fmt.Sprintf("{\"%s\":{ \"url\": \"%s\"}}", serviceName, ts.URL+tsURLBase)), + }) + if err != nil { + t.Fatalf("Error configuring plugin manager: %s", err) + } + + var delay int64 = 10 + downloadConf := download.Config{Polling: download.PollingConfig{MinDelaySeconds: &delay, MaxDelaySeconds: &delay}} + + // Start with a "legacy" style config for a single bundle + plugin.config = Config{ + Bundles: map[string]*Source{ + bundleName: &Source{ + Config: downloadConf, + Service: serviceName, + }, + }, + Name: bundleName, + Service: serviceName, + Prefix: nil, + } + + module := "package a.a1\n\nbar=1" + + b := bundle.Bundle{ + Manifest: bundle.Manifest{Revision: "quickbrownfaux", Roots: &[]string{"a/a1", "a/a2"}}, + Data: map[string]interface{}{ + "a": map[string]interface{}{ + "a2": "foo", + }, + }, + Modules: []bundle.ModuleFile{ + bundle.ModuleFile{ + Path: "bundle/id1", + Parsed: ast.MustParseModule(module), + Raw: []byte(module), + }, + }, + } + + b.Manifest.Init() + + plugin.oneShot(ctx, bundleName, download.Update{Bundle: &b}) + + // Ensure it has been activated + expData := util.MustUnmarshalJSON([]byte(`{"a2": "foo"}`)) + expIds := []string{"bundle/id1"} + validateStoreState(ctx, t, manager.Store, "/a", expData, expIds, bundleName, "quickbrownfaux") + + if plugin.config.IsMultiBundle() { + t.Fatalf("Expected plugin to be in non-multi bundle config mode") + } + + // Update to the newer style config with a new bundle + multiBundleConf := &Config{ + Bundles: map[string]*Source{ + "b2": &Source{ + Config: downloadConf, + Service: serviceName, + }, + }, + } + + plugin.Reconfigure(ctx, multiBundleConf) + + module = "package a.c\n\nbar=1" + b = bundle.Bundle{ + Manifest: bundle.Manifest{Revision: fmt.Sprintf("b2-1"), Roots: &[]string{"a/b2", "a/c"}}, + Data: map[string]interface{}{ + "a": map[string]interface{}{ + "b2": "foo", + }, + }, + Modules: []bundle.ModuleFile{ + bundle.ModuleFile{ + Path: "b2/id1", + Parsed: ast.MustParseModule(module), + Raw: []byte(module), + }, + }, + } + b.Manifest.Init() + plugin.oneShot(ctx, "b2", download.Update{Bundle: &b}) + + expData = util.MustUnmarshalJSON([]byte(`{"b2": "foo"}`)) + expIds = []string{"b2/id1"} + validateStoreState(ctx, t, manager.Store, "/a", expData, expIds, "b2", "b2-1") + + // Make sure the legacy path is gone now that we are in multi-bundle mode + var actual string + err = storage.Txn(ctx, plugin.manager.Store, storage.WriteParams, func(txn storage.Transaction) error { + var err error + if actual, err = bundle.LegacyReadRevisionFromStore(ctx, plugin.manager.Store, txn); err != nil && !storage.IsNotFound(err) { + t.Fatalf("Failed to read manifest revision from store: %s", err) + return err + } + return nil + }) + if err != nil { + t.Fatalf("Unexpected error finishing transaction: %s", err) + } + if actual != "" { + t.Fatalf("Expected to not find manifest revision but got %s", actual) + } + + if !plugin.config.IsMultiBundle() { + t.Fatalf("Expected plugin to be in multi bundle config mode") + } +} + +func validateStoreState(ctx context.Context, t *testing.T, store storage.Store, root string, expData interface{}, expIds []string, expBundleName string, expBundleRev string) { + t.Helper() + if err := storage.Txn(ctx, store, storage.TransactionParams{}, func(txn storage.Transaction) error { + value, err := store.Read(ctx, txn, storage.MustParsePath(root)) + if err != nil { + return err + } + + if !reflect.DeepEqual(value, expData) { + return fmt.Errorf("Expected %v but got %v", expData, value) + } + + ids, err := store.ListPolicies(ctx, txn) + if err != nil { + return err + } + + sort.Strings(ids) + + if !reflect.DeepEqual(ids, expIds) { + return fmt.Errorf("Expected ids %v but got %v", expIds, ids) + } + + rev, err := bundle.ReadBundleRevisionFromStore(ctx, store, txn, expBundleName) + if err != nil { + return fmt.Errorf("Unexpected error when reading bundle revision from store: %s", err) + } + + if rev != expBundleRev { + return fmt.Errorf("Unexpected revision found on bundle: %s", rev) + } + + return nil + + }); err != nil { + t.Fatal(err) + } +} diff --git a/plugins/discovery/discovery.go b/plugins/discovery/discovery.go index 1007e0dd68..07358a30d5 100644 --- a/plugins/discovery/discovery.go +++ b/plugins/discovery/discovery.go @@ -284,10 +284,18 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager } // Parse and validate bundle/logs/status configurations. + + // If `bundle` was configured use that, otherwise try the new `bundles` option bundleConfig, err := bundle.ParseConfig(config.Bundle, manager.Services()) if err != nil { return nil, err } + if bundleConfig == nil { + bundleConfig, err = bundle.ParseBundlesConfig(config.Bundles, manager.Services()) + if err != nil { + return nil, err + } + } decisionLogsConfig, err := logs.ParseConfig(config.DecisionLogs, manager.Services(), pluginNames) if err != nil { @@ -391,7 +399,16 @@ func registerBundleStatusUpdates(m *plugins.Manager) { return } type pluginlistener string - bp.Register(pluginlistener(status.Name), func(s bundle.Status) { - sp.UpdateBundleStatus(s) - }) + + // Depending on how the plugin was configured we will want to use different listeners + // for backwards compatibility. + if !bp.Config().IsMultiBundle() { + bp.Register(pluginlistener(status.Name), func(s bundle.Status) { + sp.UpdateBundleStatus(s) + }) + } else { + bp.RegisterBulkListener(pluginlistener(status.Name), func(s map[string]*bundle.Status) { + sp.BulkUpdateBundleStatus(s) + }) + } } diff --git a/plugins/discovery/discovery_test.go b/plugins/discovery/discovery_test.go index 8c0d3509bd..db027aa3fe 100644 --- a/plugins/discovery/discovery_test.go +++ b/plugins/discovery/discovery_test.go @@ -391,3 +391,86 @@ func makeDataBundle(n int, s string) *bundleApi.Bundle { Data: util.MustUnmarshalJSON([]byte(s)).(map[string]interface{}), } } + +func getTestManager(t *testing.T, conf string) *plugins.Manager { + t.Helper() + store := inmem.New() + manager, err := plugins.New([]byte(conf), "test-instance-id", store) + if err != nil { + t.Fatalf("failed to create plugin manager: %s", err) + } + return manager +} + +func TestGetPluginSetWithMixedConfig(t *testing.T) { + conf := ` +services: + s1: + url: http://test1.com + s2: + url: http://test2.com + +bundles: + bundle-new: + service: s1 + +bundle: + name: bundle-classic + service: s2 +` + manager := getTestManager(t, conf) + _, err := getPluginSet(nil, manager, manager.Config) + if err != nil { + t.Fatalf("Unexpected error: %s", err) + } + + p := manager.Plugin(bundle.Name) + if p == nil { + t.Fatal("Unable to find bundle plugin on manager") + } + bp := p.(*bundle.Plugin) + + // make sure the older style `bundle` config takes precedence + if bp.Config().Name != "bundle-classic" { + t.Fatal("Expected bundle plugin config Name to be 'bundle-classic'") + } + + if len(bp.Config().Bundles) != 1 { + t.Fatal("Expected a single bundle configured") + } + + if bp.Config().Bundles["bundle-classic"].Service != "s2" { + t.Fatalf("Expected the classic bundle to be configured as bundles[0], got: %+v", bp.Config().Bundles) + } +} + +func TestGetPluginSetWithBundlesConfig(t *testing.T) { + conf := ` +services: + s1: + url: http://test1.com + +bundles: + bundle-new: + service: s1 +` + manager := getTestManager(t, conf) + _, err := getPluginSet(nil, manager, manager.Config) + if err != nil { + t.Fatalf("Unexpected error: %s", err) + } + + p := manager.Plugin(bundle.Name) + if p == nil { + t.Fatal("Unable to find bundle plugin on manager") + } + bp := p.(*bundle.Plugin) + + if len(bp.Config().Bundles) != 1 { + t.Fatal("Expected a single bundle configured") + } + + if bp.Config().Bundles["bundle-new"].Service != "s1" { + t.Fatalf("Expected the bundle to be configured as bundles[0], got: %+v", bp.Config().Bundles) + } +} diff --git a/plugins/logs/plugin.go b/plugins/logs/plugin.go index 09a0fd6e91..aa1bf26faa 100644 --- a/plugins/logs/plugin.go +++ b/plugins/logs/plugin.go @@ -36,18 +36,24 @@ type Logger interface { // EventV1 represents a decision log event. type EventV1 struct { - Labels map[string]string `json:"labels"` - DecisionID string `json:"decision_id"` - Revision string `json:"revision,omitempty"` - Path string `json:"path,omitempty"` - Query string `json:"query,omitempty"` - Input *interface{} `json:"input,omitempty"` - Result *interface{} `json:"result,omitempty"` - Erased []string `json:"erased,omitempty"` - Error error `json:"error,omitempty"` - RequestedBy string `json:"requested_by"` - Timestamp time.Time `json:"timestamp"` - Metrics map[string]interface{} `json:"metrics,omitempty"` + Labels map[string]string `json:"labels"` + DecisionID string `json:"decision_id"` + Revision string `json:"revision,omitempty"` // Deprecated: Use Bundles instead + Bundles map[string]BundleInfoV1 `json:"bundles,omitempty"` + Path string `json:"path,omitempty"` + Query string `json:"query,omitempty"` + Input *interface{} `json:"input,omitempty"` + Result *interface{} `json:"result,omitempty"` + Erased []string `json:"erased,omitempty"` + Error error `json:"error,omitempty"` + RequestedBy string `json:"requested_by"` + Timestamp time.Time `json:"timestamp"` + Metrics map[string]interface{} `json:"metrics,omitempty"` +} + +// BundleInfoV1 describes a bundle associated with a decision log event. +type BundleInfoV1 struct { + Revision string `json:"revision,omitempty"` } const ( @@ -261,10 +267,16 @@ func (p *Plugin) Log(ctx context.Context, decision *server.Info) error { path := strings.Replace(strings.TrimPrefix(decision.Path, "data."), ".", "/", -1) + bundles := map[string]BundleInfoV1{} + for name, info := range decision.Bundles { + bundles[name] = BundleInfoV1{Revision: info.Revision} + } + event := EventV1{ Labels: p.manager.Labels(), DecisionID: decision.DecisionID, Revision: decision.Revision, + Bundles: bundles, Path: path, Query: decision.Query, Input: decision.Input, @@ -289,7 +301,10 @@ func (p *Plugin) Log(ctx context.Context, decision *server.Info) error { } if p.config.ConsoleLogs { - p.logEvent(ctx, event) + err := p.logEvent(ctx, event) + if err != nil { + p.logError("Failed to log to console: %v.", err) + } } if p.config.Plugin != nil { diff --git a/plugins/logs/plugin_test.go b/plugins/logs/plugin_test.go index 5ba0a821a0..acfd776df0 100644 --- a/plugins/logs/plugin_test.go +++ b/plugins/logs/plugin_test.go @@ -70,6 +70,42 @@ func TestPluginCustomBackend(t *testing.T) { if len(backend.events) != 2 || backend.events[0].Revision != "A" || backend.events[1].Revision != "B" { t.Fatal("Unexpected events:", backend.events) } + + // Server events with only `Revision` should not include bundles in the EventV1 struct + for _, e := range backend.events { + if len(e.Bundles) > 0 { + t.Errorf("Unexpected `bundles` in event") + } + } +} + +func TestPluginSingleBundle(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{Bundles: map[string]server.BundleInfo{"b1": {Revision: "A"}}}) + + // Server events with `Bundles` should *not* have `Revision` set + if len(backend.events) != 1 { + t.Fatalf("Unexpected number of events: %v", backend.events) + } + + if backend.events[0].Revision != "" || backend.events[0].Bundles["b1"].Revision != "A" { + t.Fatal("Unexpected events: ", backend.events) + } +} + +func TestPluginMultiBundle(t *testing.T) { + } func TestPluginErrorNoResult(t *testing.T) { diff --git a/plugins/status/plugin.go b/plugins/status/plugin.go index 8cdbfd2e32..4b9f52ae48 100644 --- a/plugins/status/plugin.go +++ b/plugins/status/plugin.go @@ -21,21 +21,24 @@ import ( // UpdateRequestV1 represents the status update message that OPA sends to // remote HTTP endpoints. type UpdateRequestV1 struct { - Labels map[string]string `json:"labels"` - Bundle *bundle.Status `json:"bundle,omitempty"` - Discovery *bundle.Status `json:"discovery,omitempty"` + Labels map[string]string `json:"labels"` + Bundle *bundle.Status `json:"bundle,omitempty"` // Deprecated: Use bulk `bundles` status updates instead + Bundles map[string]*bundle.Status `json:"bundles,omitempty"` + Discovery *bundle.Status `json:"discovery,omitempty"` } // Plugin implements status reporting. Updates can be triggered by the caller. type Plugin struct { - 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{} + manager *plugins.Manager + config Config + bundleCh chan bundle.Status // Deprecated: Use bulk bundle status updates instead + lastBundleStatus *bundle.Status // Deprecated: Use bulk bundle status updates instead + bulkBundleCh chan map[string]*bundle.Status + lastBundleStatuses map[string]*bundle.Status + discoCh chan bundle.Status + lastDiscoStatus *bundle.Status + stop chan chan struct{} + reconfig chan interface{} } // Config contains configuration for the plugin. @@ -90,12 +93,13 @@ func ParseConfig(config []byte, services []string) (*Config, error) { func New(parsedConfig *Config, manager *plugins.Manager) *Plugin { plugin := &Plugin{ - manager: manager, - config: *parsedConfig, - bundleCh: make(chan bundle.Status), - discoCh: make(chan bundle.Status), - stop: make(chan chan struct{}), - reconfig: make(chan interface{}), + manager: manager, + config: *parsedConfig, + bundleCh: make(chan bundle.Status), + bulkBundleCh: make(chan map[string]*bundle.Status), + discoCh: make(chan bundle.Status), + stop: make(chan chan struct{}), + reconfig: make(chan interface{}), } return plugin @@ -128,10 +132,16 @@ func (p *Plugin) Stop(ctx context.Context) { } // UpdateBundleStatus notifies the plugin that the policy bundle was updated. +// Deprecated: Use BulkUpdateBundleStatus instead. func (p *Plugin) UpdateBundleStatus(status bundle.Status) { p.bundleCh <- status } +// BulkUpdateBundleStatus notifies the plugin that the policy bundle was updated. +func (p *Plugin) BulkUpdateBundleStatus(status map[string]*bundle.Status) { + p.bulkBundleCh <- status +} + // UpdateDiscoveryStatus notifies the plugin that the discovery bundle was updated. func (p *Plugin) UpdateDiscoveryStatus(status bundle.Status) { p.discoCh <- status @@ -148,15 +158,25 @@ func (p *Plugin) loop() { for { select { + case statuses := <-p.bulkBundleCh: + p.lastBundleStatuses = statuses + err := p.oneShot(ctx) + if err != nil { + p.logError("%v.", err) + } else { + p.logInfo("Status update sent successfully in response to bundle update.") + } case status := <-p.bundleCh: - err := p.oneShot(ctx, false, status) + p.lastBundleStatus = &status + err := p.oneShot(ctx) if err != nil { p.logError("%v.", err) } else { p.logInfo("Status update sent successfully in response to bundle update.") } case status := <-p.discoCh: - err := p.oneShot(ctx, true, status) + p.lastDiscoStatus = &status + err := p.oneShot(ctx) if err != nil { p.logError("%v.", err) } else { @@ -174,18 +194,13 @@ func (p *Plugin) loop() { } } -func (p *Plugin) oneShot(ctx context.Context, disco bool, status bundle.Status) error { +func (p *Plugin) oneShot(ctx context.Context) error { - if disco { - p.lastDiscoStatus = &status - } else { - p.lastBundleStatus = &status - } - - req := UpdateRequestV1{ + req := &UpdateRequestV1{ Labels: p.manager.Labels(), Discovery: p.lastDiscoStatus, Bundle: p.lastBundleStatus, + Bundles: p.lastBundleStatuses, } resp, err := p.manager.Client(p.config.Service). diff --git a/plugins/status/plugin_test.go b/plugins/status/plugin_test.go index b925ce1f50..9bd09ff572 100644 --- a/plugins/status/plugin_test.go +++ b/plugins/status/plugin_test.go @@ -58,6 +58,88 @@ func TestPluginStart(t *testing.T) { } } +func TestPluginStartBulkUpdate(t *testing.T) { + + fixture := newTestFixture(t) + fixture.server.ch = make(chan UpdateRequestV1) + defer fixture.server.stop() + + ctx := context.Background() + + fixture.plugin.Start(ctx) + defer fixture.plugin.Stop(ctx) + + status := testStatus() + + fixture.plugin.BulkUpdateBundleStatus(map[string]*bundle.Status{status.Name: status}) + result := <-fixture.server.ch + + exp := UpdateRequestV1{ + Labels: map[string]string{ + "id": "test-instance-id", + "app": "example-app", + "version": version.Version, + }, + Bundles: map[string]*bundle.Status{status.Name: status}, + } + + if !reflect.DeepEqual(result, exp) { + t.Fatalf("Expected: %v but got: %v", exp, result) + } +} + +func TestPluginStartBulkUpdateMultiple(t *testing.T) { + + fixture := newTestFixture(t) + fixture.server.ch = make(chan UpdateRequestV1) + defer fixture.server.stop() + + ctx := context.Background() + + fixture.plugin.Start(ctx) + defer fixture.plugin.Stop(ctx) + + statuses := map[string]*bundle.Status{} + tDownload, _ := time.Parse("2018-01-01T00:00:00.0000000Z", time.RFC3339Nano) + tActivate, _ := time.Parse("2018-01-01T00:00:01.0000000Z", time.RFC3339Nano) + for i := 0; i < 20; i++ { + name := fmt.Sprintf("test-bundle-%d", i) + statuses[name] = &bundle.Status{ + Name: name, + ActiveRevision: fmt.Sprintf("v%d", i), + LastSuccessfulDownload: tDownload, + LastSuccessfulActivation: tActivate, + } + } + + fixture.plugin.BulkUpdateBundleStatus(statuses) + result := <-fixture.server.ch + + expLabels := map[string]string{ + "id": "test-instance-id", + "app": "example-app", + "version": version.Version, + } + + if !reflect.DeepEqual(result.Labels, expLabels) { + t.Fatalf("Unexpected status labels: %+v", result.Labels) + } + + if len(result.Bundles) != len(statuses) { + t.Fatalf("Expected %d statuses, got %d", len(statuses), len(result.Bundles)) + } + + for name, s := range statuses { + actualStatus := result.Bundles[name] + if actualStatus.Name != s.Name || + actualStatus.LastSuccessfulActivation != s.LastSuccessfulActivation || + actualStatus.LastSuccessfulDownload != s.LastSuccessfulDownload || + actualStatus.ActiveRevision != s.ActiveRevision { + t.Errorf("Bundle %s has unexpected status:\n\n %v\n\nExpected:\n%v\n\n", name, actualStatus, s) + } + } +} + func TestPluginStartDiscovery(t *testing.T) { fixture := newTestFixture(t) @@ -93,7 +175,8 @@ func TestPluginBadAuth(t *testing.T) { ctx := context.Background() fixture.server.expCode = 401 defer fixture.server.stop() - err := fixture.plugin.oneShot(ctx, false, bundle.Status{}) + fixture.plugin.lastBundleStatus = &bundle.Status{} + err := fixture.plugin.oneShot(ctx) if err == nil { t.Fatal("Expected error") } @@ -104,7 +187,8 @@ func TestPluginBadPath(t *testing.T) { ctx := context.Background() fixture.server.expCode = 404 defer fixture.server.stop() - err := fixture.plugin.oneShot(ctx, false, bundle.Status{}) + fixture.plugin.lastBundleStatus = &bundle.Status{} + err := fixture.plugin.oneShot(ctx) if err == nil { t.Fatal("Expected error") } @@ -115,7 +199,8 @@ func TestPluginBadStatus(t *testing.T) { ctx := context.Background() fixture.server.expCode = 500 defer fixture.server.stop() - err := fixture.plugin.oneShot(ctx, false, bundle.Status{}) + fixture.plugin.lastBundleStatus = &bundle.Status{} + err := fixture.plugin.oneShot(ctx) if err == nil { t.Fatal("Expected error") } diff --git a/server/buffer.go b/server/buffer.go index dbaf2a5231..64f908bca9 100644 --- a/server/buffer.go +++ b/server/buffer.go @@ -74,7 +74,8 @@ func (b *buffer) Iter(fn func(*Info)) { // Info contains information describing a policy decision. type Info struct { Txn storage.Transaction - Revision string + Revision string // Deprecated: Use `Bundles` instead + Bundles map[string]BundleInfo DecisionID string RemoteAddr string Query string @@ -86,3 +87,8 @@ type Info struct { Metrics metrics.Metrics Trace []*topdown.Event } + +// BundleInfo contains information describing a bundle +type BundleInfo struct { + Revision string +} diff --git a/server/server.go b/server/server.go index 8562576e2b..006239c3c1 100644 --- a/server/server.go +++ b/server/server.go @@ -11,6 +11,7 @@ import ( "crypto/x509" "encoding/json" "fmt" + "github.com/open-policy-agent/opa/bundle" "html/template" "io" "io/ioutil" @@ -27,10 +28,9 @@ import ( "github.com/gorilla/mux" "github.com/open-policy-agent/opa/ast" - "github.com/open-policy-agent/opa/bundle/manifest" "github.com/open-policy-agent/opa/metrics" "github.com/open-policy-agent/opa/plugins" - "github.com/open-policy-agent/opa/plugins/bundle" + bundlePlugin "github.com/open-policy-agent/opa/plugins/bundle" "github.com/open-policy-agent/opa/rego" "github.com/open-policy-agent/opa/server/authorizer" "github.com/open-policy-agent/opa/server/identifier" @@ -101,15 +101,15 @@ type Server struct { watcher *watch.Watcher decisionIDFactory func() string diagnostics Buffer - revision string + revisions map[string]string + legacyRevision string logger func(context.Context, *Info) error errLimit int pprofEnabled bool runtime *ast.Term httpListeners []httpListener - bundleStatus bundle.Status + bundleStatuses map[string]*bundlePlugin.Status bundleStatusMtx *sync.RWMutex - hasBundle bool } // Loop will contain all the calls from the server that we'll be listening on. @@ -169,13 +169,23 @@ func (s *Server) Init(ctx context.Context) (*Server, error) { s.partials = map[string]rego.PartialResult{} - bp := bundle.Lookup(s.manager) + bp := bundlePlugin.Lookup(s.manager) if bp != nil { s.bundleStatusMtx = new(sync.RWMutex) - s.hasBundle = true - bp.Register("REST API Server", func(status bundle.Status) { - s.updateBundleStatus(status) - }) + + // initialize statuses to empty defaults for server /health check + s.bundleStatuses = map[string]*bundlePlugin.Status{} + for bundleName := range bp.Config().Bundles { + s.bundleStatuses[bundleName] = &bundlePlugin.Status{Name: bundleName} + } + + bp.RegisterBulkListener("REST API Server", s.updateBundleStatus) + } + + // Check if there is a bundle revision available at the legacy storage path + rev, err := bundle.LegacyReadRevisionFromStore(ctx, s.store, txn) + if err == nil && rev != "" { + s.legacyRevision = rev } return s, s.store.Commit(ctx, txn) @@ -714,16 +724,29 @@ func (s *Server) registerHandler(router *mux.Router, version int, path string, m } func (s *Server) reload(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { + // reset some cached info + s.partials = map[string]rego.PartialResult{} + s.revisions = map[string]string{} - if revision, err := manifest.ReadBundleRevision(ctx, s.store, txn); err != nil { - if !storage.IsNotFound(err) { - panic(err) - } - } else { - s.revision = revision + // read all bundle revisions from storage (if any exist) + names, err := bundle.ReadBundleNamesFromStore(ctx, s.store, txn) + if err != nil && !storage.IsNotFound(err) { + panic(err) } - s.partials = map[string]rego.PartialResult{} + for _, name := range names { + r, err := bundle.ReadBundleRevisionFromStore(ctx, s.store, txn, name) + if err != nil && !storage.IsNotFound(err) { + panic(err) + } + s.revisions[name] = r + } + + // Check if we still have a legacy bundle manifest in the store + s.legacyRevision, err = bundle.LegacyReadRevisionFromStore(ctx, s.store, txn) + if err != nil && !storage.IsNotFound(err) { + panic(err) + } } func (s *Server) migrateWatcher(txn storage.Transaction) { @@ -865,10 +888,10 @@ func (s *Server) v1DiagnosticsGet(w http.ResponseWriter, r *http.Request) { writer.JSON(w, 200, resp, pretty) } -func (s *Server) updateBundleStatus(status bundle.Status) { +func (s *Server) updateBundleStatus(status map[string]*bundlePlugin.Status) { s.bundleStatusMtx.Lock() defer s.bundleStatusMtx.Unlock() - s.bundleStatus = status + s.bundleStatuses = status } func (s *Server) canEval(ctx context.Context) bool { @@ -891,12 +914,18 @@ func (s *Server) canEval(ctx context.Context) bool { return false } -func (s *Server) bundleActivated() bool { +func (s *Server) bundlesActivated() bool { s.bundleStatusMtx.RLock() defer s.bundleStatusMtx.RUnlock() - // Ensure that the bundle status has an activation time set on it - return s.bundleStatus.LastSuccessfulActivation != time.Time{} + for _, status := range s.bundleStatuses { + // Ensure that all of the bundle statuses have an activation time set on them + if (status.LastSuccessfulActivation == time.Time{}) { + return false + } + } + + return true } func (s *Server) unversionedGetHealth(w http.ResponseWriter, r *http.Request) { @@ -912,7 +941,7 @@ func (s *Server) unversionedGetHealth(w http.ResponseWriter, r *http.Request) { // Ensure that bundles (if configured, and requested to be included in the result) // have been activated successfully. - if includeBundleStatus && s.hasBundle && !s.bundleActivated() { + if includeBundleStatus && s.hasBundle() && !s.bundlesActivated() { writer.JSON(w, http.StatusInternalServerError, emptyObject{}, false) return } @@ -1107,7 +1136,7 @@ func (s *Server) v1DataGet(w http.ResponseWriter, r *http.Request) { } if provenance { - result.Provenance = getProvenance(s.revision) + result.Provenance = s.getProvenance() } if len(rs) == 0 { @@ -1285,7 +1314,7 @@ func (s *Server) v1DataPost(w http.ResponseWriter, r *http.Request) { } if provenance { - result.Provenance = getProvenance(s.revision) + result.Provenance = s.getProvenance() } if len(rs) == 0 { @@ -1881,7 +1910,7 @@ func (s *Server) checkPolicyPackageScope(ctx context.Context, txn storage.Transa func (s *Server) checkPathScope(ctx context.Context, txn storage.Transaction, path storage.Path) error { - roots, err := manifest.ReadBundleRoots(ctx, s.store, txn) + names, err := bundle.ReadBundleNamesFromStore(ctx, s.store, txn) if err != nil { if !storage.IsNotFound(err) { return err @@ -1889,11 +1918,22 @@ func (s *Server) checkPathScope(ctx context.Context, txn storage.Transaction, pa return nil } + bundleRoots := map[string][]string{} + for _, name := range names { + roots, err := bundle.ReadBundleRootsFromStore(ctx, s.store, txn, name) + if err != nil && !storage.IsNotFound(err) { + return err + } + bundleRoots[name] = roots + } + spath := strings.Trim(path.String(), "/") - for i := range roots { - if strings.HasPrefix(spath, roots[i]) || strings.HasPrefix(roots[i], spath) { - return types.BadRequestErr(fmt.Sprintf("path %v is owned by bundle", spath)) + for name, roots := range bundleRoots { + for _, root := range roots { + if strings.HasPrefix(spath, root) || strings.HasPrefix(root, spath) { + return types.BadRequestErr(fmt.Sprintf("path %v is owned by bundle %q", spath, name)) + } } } @@ -1907,7 +1947,12 @@ func (s *Server) evalDiagnosticPolicy(r *http.Request) (logger diagnosticsLogger // logger will make sure to call the decision logger regardless of whether a // diagnostic policy is configured. In the future, we can refactor this. defer func() { - logger.revision = s.revision + // For backwards compatibility use `revision` as needed. + if s.hasLegacyBundle() { + logger.revision = s.legacyRevision + } else { + logger.revisions = s.revisions + } logger.logger = s.logger }() @@ -2101,6 +2146,39 @@ func (s *Server) generateDecisionID() string { return "" } +func (s *Server) getProvenance() *types.ProvenanceV1 { + + p := &types.ProvenanceV1{ + Version: version.Version, + Vcs: version.Vcs, + Timestamp: version.Timestamp, + Hostname: version.Hostname, + } + + // For backwards compatibility, if the bundles are using the old + // style config we need to fill in the older `Revision` field. + // Otherwise use the newer `Bundles` keyword. + if s.hasLegacyBundle() { + p.Revision = s.legacyRevision + } else { + p.Bundles = map[string]types.ProvenanceBundleV1{} + for name, revision := range s.revisions { + p.Bundles[name] = types.ProvenanceBundleV1{Revision: revision} + } + } + + return p +} + +func (s *Server) hasBundle() bool { + return bundlePlugin.Lookup(s.manager) != nil || s.legacyRevision != "" +} + +func (s *Server) hasLegacyBundle() bool { + bp := bundlePlugin.Lookup(s.manager) + return s.legacyRevision != "" || (bp != nil && !bp.Config().IsMultiBundle()) +} + // parsePatchPathEscaped returns a new path for the given escaped str. // This is based on storage.ParsePathEscaped so will do URL unescaping of // the provided str for backwards compatibility, but also handles the @@ -2455,7 +2533,8 @@ func renderVersion(w http.ResponseWriter) { type diagnosticsLogger struct { logger func(context.Context, *Info) error - revision string + revisions map[string]string + revision string // Deprecated: Use `revisions` instead. explain bool instrument bool buffer Buffer @@ -2471,9 +2550,15 @@ func (l diagnosticsLogger) Instrument() bool { func (l diagnosticsLogger) Log(ctx context.Context, txn storage.Transaction, decisionID, remoteAddr, path string, query string, input *interface{}, results *interface{}, err error, m metrics.Metrics, tracer *topdown.BufferTracer) error { + bundles := map[string]BundleInfo{} + for name, rev := range l.revisions { + bundles[name] = BundleInfo{Revision: rev} + } + info := &Info{ Txn: txn, Revision: l.revision, + Bundles: bundles, Timestamp: time.Now().UTC(), DecisionID: decisionID, RemoteAddr: remoteAddr, @@ -2519,14 +2604,3 @@ func parseURL(s string, useHTTPSByDefault bool) (*url.URL, error) { } return url.Parse(s) } - -func getProvenance(revision string) *types.ProvenanceV1 { - - return &types.ProvenanceV1{ - Version: version.Version, - Vcs: version.Vcs, - Timestamp: version.Timestamp, - Hostname: version.Hostname, - Revision: revision, - } -} diff --git a/server/server_test.go b/server/server_test.go index 01517651ba..6902ea7c36 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -27,7 +27,6 @@ import ( "github.com/open-policy-agent/opa/ast" "github.com/open-policy-agent/opa/bundle" - "github.com/open-policy-agent/opa/bundle/manifest" "github.com/open-policy-agent/opa/metrics" "github.com/open-policy-agent/opa/plugins" pluginBundle "github.com/open-policy-agent/opa/plugins/bundle" @@ -74,30 +73,88 @@ func TestUnversionedGetHealthBundleNoBundleSet(t *testing.T) { } } -func TestUnversionedGetHealthCheckBundleActivation(t *testing.T) { +func TestUnversionedGetHealthCheckBundleActivationSingle(t *testing.T) { f := newFixture(t) + bundleName := "test-bundle" // Initialize the server as if a bundle plugin was // configured on the manager. - f.server.hasBundle = true + f.server.manager.Register(pluginBundle.Name, &pluginBundle.Plugin{}) f.server.bundleStatusMtx = new(sync.RWMutex) + f.server.bundleStatuses = map[string]*pluginBundle.Status{ + bundleName: &pluginBundle.Status{Name: bundleName}, + } - // The bundle hasnt been activated yet, expect it to be activated + // The bundle hasn't been activated yet, expect the health check to fail req := newReqUnversioned(http.MethodGet, "/health?bundle=true", "") if err := f.executeRequest(req, 500, `{}`); err != nil { - t.Fatalf("Unexpected error while health check: %v", err) + t.Fatal(err) } // Set the bundle to be activated. - status := pluginBundle.Status{} - status.SetActivateSuccess("") + status := map[string]*pluginBundle.Status{ + bundleName: &pluginBundle.Status{}, + } + status[bundleName].SetActivateSuccess("") f.server.updateBundleStatus(status) // The heath check should now respond as healthy req = newReqUnversioned(http.MethodGet, "/health?bundle=true", "") if err := f.executeRequest(req, 200, `{}`); err != nil { - t.Fatalf("Unexpected error while health check: %v", err) + t.Fatal(err) + } +} + +func TestUnversionedGetHealthCheckBundleActivationMulti(t *testing.T) { + + f := newFixture(t) + + // Initialize the server as if a bundle plugin was + // configured on the manager. + bp := pluginBundle.New(&pluginBundle.Config{Bundles: map[string]*pluginBundle.Source{ + "b1": {Service: "s1", Resource: "bundle.tar.gz"}, + "b2": {Service: "s2", Resource: "bundle.tar.gz"}, + "b3": {Service: "s3", Resource: "bundle.tar.gz"}, + }}, f.server.manager) + f.server.manager.Register(pluginBundle.Name, bp) + f.server.bundleStatusMtx = new(sync.RWMutex) + f.server.bundleStatuses = map[string]*pluginBundle.Status{ + "b1": {Name: "b1"}, + "b2": {Name: "b2"}, + "b3": {Name: "b3"}, + } + + // No bundle has been activated yet, expect the health check to fail + req := newReqUnversioned(http.MethodGet, "/health?bundle=true", "") + if err := f.executeRequest(req, 500, `{}`); err != nil { + t.Fatal(err) + } + + // Set one bundle to be activated + update := map[string]*pluginBundle.Status{ + "b1": {Name: "b1"}, + "b2": {Name: "b2"}, + "b3": {Name: "b3"}, + } + update["b2"].SetActivateSuccess("A") + f.server.updateBundleStatus(update) + + // The heath check should still respond as unhealthy + req = newReqUnversioned(http.MethodGet, "/health?bundle=true", "") + if err := f.executeRequest(req, 500, `{}`); err != nil { + t.Fatal(err) + } + + // Activate all the bundles + update["b1"].SetActivateSuccess("B") + update["b3"].SetActivateSuccess("C") + f.server.updateBundleStatus(update) + + // The heath check should succeed now + req = newReqUnversioned(http.MethodGet, "/health?bundle=true", "") + if err := f.executeRequest(req, 200, `{}`); err != nil { + t.Fatal(err) } } @@ -108,7 +165,14 @@ func TestInitWithBundlePlugin(t *testing.T) { t.Fatalf("Unexpected error creating plugin manager: %s", err.Error()) } - m.Register(pluginBundle.Name, new(pluginBundle.Plugin)) + bundleName := "test-bundle" + bundleConf := &pluginBundle.Config{ + Name: bundleName, + Service: "s1", + Bundles: map[string]*pluginBundle.Source{"b1": {}}, + } + + m.Register(pluginBundle.Name, pluginBundle.New(bundleConf, m)) server, err := New(). WithStore(store). @@ -119,7 +183,7 @@ func TestInitWithBundlePlugin(t *testing.T) { t.Fatalf("Unexpected error initializing server: %s", err.Error()) } - if server.hasBundle == false { + if !server.hasBundle() { t.Error("server.hasBundle should be true") } @@ -127,7 +191,45 @@ func TestInitWithBundlePlugin(t *testing.T) { t.Error("server.bundleStatusMtx should be initialized") } - isActivated := server.bundleActivated() + isActivated := server.bundlesActivated() + if isActivated { + t.Error("bundle should not be initialized to activated status") + } +} + +func TestInitWithBundlePluginMultiBundle(t *testing.T) { + store := inmem.New() + m, err := plugins.New([]byte{}, "test", store) + if err != nil { + t.Fatalf("Unexpected error creating plugin manager: %s", err.Error()) + } + + bundleConf := &pluginBundle.Config{Bundles: map[string]*pluginBundle.Source{ + "b1": {}, + "b2": {}, + "b3": {}, + }} + + m.Register(pluginBundle.Name, pluginBundle.New(bundleConf, m)) + + server, err := New(). + WithStore(store). + WithManager(m). + Init(context.Background()) + + if err != nil { + t.Fatalf("Unexpected error initializing server: %s", err.Error()) + } + + if !server.hasBundle() { + t.Error("server.hasBundle should be true") + } + + if server.bundleStatusMtx == nil { + t.Error("server.bundleStatusMtx should be initialized") + } + + isActivated := server.bundlesActivated() if isActivated { t.Error("bundle should not be initialized to activated") } @@ -960,7 +1062,7 @@ func TestBundleScope(t *testing.T) { txn := storage.NewTransactionOrDie(ctx, f.server.store, storage.WriteParams) - if err := manifest.Write(ctx, f.server.store, txn, bundle.Manifest{ + if err := bundle.WriteManifestToStore(ctx, f.server.store, txn, "test-bundle", bundle.Manifest{ Revision: "AAAAA", Roots: &[]string{"a/b/c", "x/y"}, }); err != nil { @@ -981,47 +1083,112 @@ func TestBundleScope(t *testing.T) { path: "/data/a/b", body: "1", code: http.StatusBadRequest, - resp: `{"code": "invalid_parameter", "message": "path a/b is owned by bundle"}`, + resp: `{"code": "invalid_parameter", "message": "path a/b is owned by bundle \"test-bundle\""}`, }, { method: "PUT", path: "/data/a/b/c", body: "1", code: http.StatusBadRequest, - resp: `{"code": "invalid_parameter", "message": "path a/b/c is owned by bundle"}`, + resp: `{"code": "invalid_parameter", "message": "path a/b/c is owned by bundle \"test-bundle\""}`, }, { method: "PUT", path: "/data/a/b/c/d", body: "1", code: http.StatusBadRequest, - resp: `{"code": "invalid_parameter", "message": "path a/b/c/d is owned by bundle"}`, + resp: `{"code": "invalid_parameter", "message": "path a/b/c/d is owned by bundle \"test-bundle\""}`, }, { method: "PATCH", path: "/data/a", body: `[{"path": "/b/c", "op": "add", "value": 1}]`, code: http.StatusBadRequest, - resp: `{"code": "invalid_parameter", "message": "path a/b/c is owned by bundle"}`, + resp: `{"code": "invalid_parameter", "message": "path a/b/c is owned by bundle \"test-bundle\""}`, }, { method: "DELETE", path: "/data/a", code: http.StatusBadRequest, - resp: `{"code": "invalid_parameter", "message": "path a is owned by bundle"}`, + resp: `{"code": "invalid_parameter", "message": "path a is owned by bundle \"test-bundle\""}`, }, { method: "PUT", path: "/policies/test1", body: `package a.b`, code: http.StatusBadRequest, - resp: `{"code": "invalid_parameter", "message": "path a/b is owned by bundle"}`, + resp: `{"code": "invalid_parameter", "message": "path a/b is owned by bundle \"test-bundle\""}`, }, { method: "DELETE", path: "/policies/someid", code: http.StatusBadRequest, - resp: `{"code": "invalid_parameter", "message": "path x/y/z is owned by bundle"}`, + resp: `{"code": "invalid_parameter", "message": "path x/y/z is owned by bundle \"test-bundle\""}`, + }, + { + method: "PUT", + path: "/data/foo/bar", + body: "1", + code: http.StatusNoContent, + }, + } + + if err := f.v1TestRequests(cases); err != nil { + t.Fatal(err) + } +} + +func TestBundleScopeMultiBundle(t *testing.T) { + + ctx := context.Background() + + f := newFixture(t) + + txn := storage.NewTransactionOrDie(ctx, f.server.store, storage.WriteParams) + + if err := bundle.WriteManifestToStore(ctx, f.server.store, txn, "test-bundle1", bundle.Manifest{ + Revision: "AAAAA", + Roots: &[]string{"a/b/c", "x/y"}, + }); err != nil { + t.Fatal(err) + } + + if err := bundle.WriteManifestToStore(ctx, f.server.store, txn, "test-bundle2", bundle.Manifest{ + Revision: "AAAAA", + Roots: &[]string{"a/b/d"}, + }); err != nil { + t.Fatal(err) + } + + if err := bundle.WriteManifestToStore(ctx, f.server.store, txn, "test-bundle3", bundle.Manifest{ + Revision: "AAAAA", + Roots: &[]string{"a/b/e", "a/b/f"}, + }); err != nil { + t.Fatal(err) + } + + if err := f.server.store.UpsertPolicy(ctx, txn, "someid", []byte(`package x.y.z`)); err != nil { + t.Fatal(err) + } + + if err := f.server.store.Commit(ctx, txn); err != nil { + t.Fatal(err) + } + + cases := []tr{ + { + method: "PUT", + path: "/data/x/y", + body: "1", + code: http.StatusBadRequest, + resp: `{"code": "invalid_parameter", "message": "path x/y is owned by bundle \"test-bundle1\""}`, + }, + { + method: "PUT", + path: "/data/a/b/d", + body: "1", + code: http.StatusBadRequest, + resp: `{"code": "invalid_parameter", "message": "path a/b/d is owned by bundle \"test-bundle2\""}`, }, { method: "PUT", @@ -1481,7 +1648,7 @@ func TestDataPostExplainNotes(t *testing.T) { } } -func TestDataProvenance(t *testing.T) { +func TestDataProvenanceSingleBundle(t *testing.T) { f := newFixture(t) @@ -1492,6 +1659,14 @@ func TestDataProvenance(t *testing.T) { version.Timestamp = "today" version.Hostname = "foo.bar.com" + // Initialize as if a bundle plugin is running + bp := pluginBundle.New(&pluginBundle.Config{Name: "b1"}, f.server.manager) + f.server.manager.Register(pluginBundle.Name, bp) + f.server.bundleStatusMtx = new(sync.RWMutex) + f.server.bundleStatuses = map[string]*pluginBundle.Status{ + "b1": {Name: "b1"}, + } + req := newReqV1(http.MethodPost, "/data?provenance", "") f.reset() f.server.Handler.ServeHTTP(f.recorder, req) @@ -1505,6 +1680,182 @@ func TestDataProvenance(t *testing.T) { if result.Provenance == nil { t.Fatalf("Expected non-nil provenance: %v", result.Provenance) } + + expectedProvenance := &types.ProvenanceV1{ + Version: version.Version, + Vcs: version.Vcs, + Timestamp: version.Timestamp, + Hostname: version.Hostname, + } + + if !reflect.DeepEqual(result.Provenance, expectedProvenance) { + t.Errorf("Unexpected provenance data: \n\n%+v\n\nExpected:\n%+v\n\n", result.Provenance, expectedProvenance) + } + + // Update bundle revision and request again + f.server.revisions["b1"] = "r1" + f.server.legacyRevision = "r1" + + req = newReqV1(http.MethodPost, "/data?provenance", "") + f.reset() + f.server.Handler.ServeHTTP(f.recorder, req) + + result = types.DataResponseV1{} + + if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil { + t.Fatalf("Unexpected JSON decode error: %v", err) + } + + if result.Provenance == nil { + t.Fatalf("Expected non-nil provenance: %v", result.Provenance) + } + + expectedProvenance.Revision = "r1" + if !reflect.DeepEqual(result.Provenance, expectedProvenance) { + t.Errorf("Unexpected provenance data: \n\n%+v\n\nExpected:\n%+v\n\n", result.Provenance, expectedProvenance) + } +} + +func TestDataProvenanceSingleFileBundle(t *testing.T) { + + f := newFixture(t) + + // Dummy up since we are not using ld... + // Note: No bundle 'revision'... + version.Version = "0.10.7" + version.Vcs = "ac23eb45" + version.Timestamp = "today" + version.Hostname = "foo.bar.com" + + // No bundle plugin initialized, just a legacy revision set + f.server.legacyRevision = "r1" + + req := newReqV1(http.MethodPost, "/data?provenance", "") + f.reset() + f.server.Handler.ServeHTTP(f.recorder, req) + + result := types.DataResponseV1{} + + if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil { + t.Fatalf("Unexpected JSON decode error: %v", err) + } + + if result.Provenance == nil { + t.Fatalf("Expected non-nil provenance: %v", result.Provenance) + } + + expectedProvenance := &types.ProvenanceV1{ + Version: version.Version, + Vcs: version.Vcs, + Timestamp: version.Timestamp, + Hostname: version.Hostname, + } + + expectedProvenance.Revision = "r1" + if !reflect.DeepEqual(result.Provenance, expectedProvenance) { + t.Errorf("Unexpected provenance data: \n\n%+v\n\nExpected:\n%+v\n\n", result.Provenance, expectedProvenance) + } +} + +func TestDataProvenanceMultiBundle(t *testing.T) { + + f := newFixture(t) + + // Dummy up since we are not using ld... + version.Version = "0.10.7" + version.Vcs = "ac23eb45" + version.Timestamp = "today" + version.Hostname = "foo.bar.com" + + // Initialize as if a bundle plugin is running with 2 bundles + bp := pluginBundle.New(&pluginBundle.Config{Bundles: map[string]*pluginBundle.Source{ + "b1": {Service: "s1", Resource: "bundle.tar.gz"}, + "b2": {Service: "s2", Resource: "bundle.tar.gz"}, + }}, f.server.manager) + f.server.manager.Register(pluginBundle.Name, bp) + + f.server.bundleStatusMtx = new(sync.RWMutex) + f.server.bundleStatuses = map[string]*pluginBundle.Status{ + "b1": {Name: "b1"}, + "b2": {Name: "b2"}, + } + + req := newReqV1(http.MethodPost, "/data?provenance", "") + f.reset() + f.server.Handler.ServeHTTP(f.recorder, req) + + var result types.DataResponseV1 + + if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil { + t.Fatalf("Unexpected JSON decode error: %v", err) + } + + if result.Provenance == nil { + t.Fatalf("Expected non-nil provenance: %v", result.Provenance) + } + + expectedProvenance := &types.ProvenanceV1{ + Version: version.Version, + Vcs: version.Vcs, + Timestamp: version.Timestamp, + Hostname: version.Hostname, + } + + if !reflect.DeepEqual(result.Provenance, expectedProvenance) { + t.Errorf("Unexpected provenance data: \n\n%+v\n\nExpected:\n%+v\n\n", result.Provenance, expectedProvenance) + } + + // Update bundle revision for a single bundle and make the request again + f.server.revisions["b1"] = "r1" + + req = newReqV1(http.MethodPost, "/data?provenance", "") + f.reset() + f.server.Handler.ServeHTTP(f.recorder, req) + + result = types.DataResponseV1{} + + if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil { + t.Fatalf("Unexpected JSON decode error: %v", err) + } + + if result.Provenance == nil { + t.Fatalf("Expected non-nil provenance: %v", result.Provenance) + } + + expectedProvenance.Bundles = map[string]types.ProvenanceBundleV1{ + "b1": {Revision: "r1"}, + } + + if !reflect.DeepEqual(result.Provenance, expectedProvenance) { + t.Errorf("Unexpected provenance data: \n\n%+v\n\nExpected:\n%+v\n\n", result.Provenance, expectedProvenance) + } + + // Update both and check again + f.server.revisions["b1"] = "r2" + f.server.revisions["b2"] = "r1" + + req = newReqV1(http.MethodPost, "/data?provenance", "") + f.reset() + f.server.Handler.ServeHTTP(f.recorder, req) + + result = types.DataResponseV1{} + + if err := util.NewJSONDecoder(f.recorder.Body).Decode(&result); err != nil { + t.Fatalf("Unexpected JSON decode error: %v", err) + } + + if result.Provenance == nil { + t.Fatalf("Expected non-nil provenance: %v", result.Provenance) + } + + expectedProvenance.Bundles = map[string]types.ProvenanceBundleV1{ + "b1": {Revision: "r2"}, + "b2": {Revision: "r1"}, + } + + if !reflect.DeepEqual(result.Provenance, expectedProvenance) { + t.Errorf("Unexpected provenance data: \n\n%+v\n\nExpected:\n%+v\n\n", result.Provenance, expectedProvenance) + } } func TestDataMetrics(t *testing.T) { diff --git a/server/types/types.go b/server/types/types.go index b0b9de57da..737104b695 100644 --- a/server/types/types.go +++ b/server/types/types.go @@ -125,11 +125,17 @@ func (p PolicyV1) Equal(other PolicyV1) bool { // ProvenanceV1 models a collection of build/version information. type ProvenanceV1 struct { - Version string `json:"version"` - Vcs string `json:"build_commit"` - Timestamp string `json:"build_timestamp"` - Hostname string `json:"build_hostname"` - Revision string `json:"revision,omitempty"` + Version string `json:"version"` + Vcs string `json:"build_commit"` + Timestamp string `json:"build_timestamp"` + Hostname string `json:"build_hostname"` + Revision string `json:"revision,omitempty"` // Deprecated: Prefer `Bundles` + Bundles map[string]ProvenanceBundleV1 `json:"bundles,omitempty"` +} + +// ProvenanceBundleV1 models a bundle at some point in time +type ProvenanceBundleV1 struct { + Revision string `json:"revision"` } // DataRequestV1 models the request message for Data API POST operations. diff --git a/util/close.go b/util/close.go index 05f287bb43..97c9177311 100644 --- a/util/close.go +++ b/util/close.go @@ -14,7 +14,7 @@ import ( // ensure that the connection is freed. If the body is not read and closed, a // leak can occur. func Close(resp *http.Response) { - if resp != nil { + if resp != nil && resp.Body != nil { if _, err := io.Copy(ioutil.Discard, resp.Body); err != nil { return }