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 }