plugins/bundle: Update persisted bundle activation mechanism

Earlier errors encountered during loading and activating persisted bundles would cause the OPA runtime to exit. This behavior is different from when OPA downloads a bundle and activation errors if any would possibly get resolved in successive download attempts. This fix adds a retry mechanism to activate persisted bundles in an attempt to mimic the behavior seen during bundle downloads. Errors if any encountered during the process will be surfaced in the bundle's status update and not result in an abrupt exit. Fixes: #3840 Signed-off-by: Ashutosh Narkar <anarkar4387@gmail.com>
2026-08-12 19:32:48 -06:00 · 2022-01-13 12:47:50 -08:00
parent d98015270e
commit 4700768448
3 changed files with 225 additions and 20 deletions
@@ -86,8 +86,9 @@ OPA can optionally persist activated bundles to disk for recovery purposes. To e
 persistence, set the `bundles[_].persist` field to `true`. When bundle
 persistence is enabled, OPA will attempt to read the bundle from disk on startup. This
 allows OPA to start with the most recently activated bundle in case OPA cannot communicate
-with the bundle server. When communication between OPA and the bundle server is restored,
-the latest bundle is downloaded, activated, and persisted.
+with the bundle server. OPA will try to load and activate persisted bundles on a best-effort basis. Any errors
+encountered during the process will be surfaced in the bundle's status update. When communication between OPA and
+the bundle server is restored, the latest bundle is downloaded, activated, and persisted.

 > By default, bundles are persisted under the current working directory of the OPA process (e.g., `./.opa/bundles/<bundle-name>/bundle.tar.gz`).

@@ -27,6 +27,18 @@ import (
 	"github.com/open-policy-agent/opa/storage"
 )

+// maxActivationRetry represents the maximum number of attempts
+// to activate persisted bundles. Activation retries are useful
+// in scenarios where a persisted bundle may have a dependency on some
+// other persisted bundle. As there are no ordering guarantees for which
+// bundle loads first, retries could help in the bundle activation process.
+// Typically, multiple bundles are not encouraged. The value chosen for
+// maxActivationRetry allows upto 10 bundles to successfully activate
+// in the worst case that they depend on each other. At the same time, it also
+// ensures that too much time is not spent to activate bundles that will never
+// successfully activate.
+const maxActivationRetry = 10
+
 // Loader defines the interface that the bundle plugin uses to control bundle
 // loading via HTTP, disk, etc.
 type Loader interface {
@@ -102,10 +114,7 @@ func (p *Plugin) Start(ctx context.Context) error {
 		return err
 	}

-	err = p.loadAndActivateBundlesFromDisk(ctx)
-	if err != nil {
-		return err
-	}
+	p.loadAndActivateBundlesFromDisk(ctx)

 	p.initDownloaders()
 	for name, dl := range p.downloaders {
@@ -301,25 +310,42 @@ func (p *Plugin) initDownloaders() {
 	}
 }

-func (p *Plugin) loadAndActivateBundlesFromDisk(ctx context.Context) error {
+func (p *Plugin) loadAndActivateBundlesFromDisk(ctx context.Context) {
+
+	persistedBundles := map[string]*bundle.Bundle{}
+
 	for name, src := range p.config.Bundles {
 		if p.persistBundle(name) {
 			b, err := loadBundleFromDisk(p.bundlePersistPath, name, src)
 			if err != nil {
 				p.log(name).Error("Failed to load bundle from disk: %v", err)
-				return err
+				p.status[name].SetError(err)
+				continue
 			}

 			if b == nil {
-				return nil
+				continue
 			}

+			persistedBundles[name] = b
+		}
+	}
+
+	if len(persistedBundles) == 0 {
+		return
+	}
+
+	for retry := 0; retry < maxActivationRetry; retry++ {
+
+		numActivatedBundles := 0
+		for name, b := range persistedBundles {
 			p.status[name].Metrics = metrics.New()

-			err = p.activate(ctx, name, b)
+			err := p.activate(ctx, name, b)
 			if err != nil {
 				p.log(name).Error("Bundle activation failed: %v", err)
-				return err
+				p.status[name].SetError(err)
+				continue
 			}

 			p.status[name].SetError(nil)
@@ -328,9 +354,13 @@ func (p *Plugin) loadAndActivateBundlesFromDisk(ctx context.Context) error {
 			p.checkPluginReadiness()

 			p.log(name).Debug("Bundle loaded from disk and activated successfully.")
+			numActivatedBundles++
+		}
+
+		if numActivatedBundles == len(persistedBundles) {
+			return
 		}
 	}
-	return nil
 }

 func (p *Plugin) newDownloader(name string, source *Source) Loader {
@@ -406,10 +406,7 @@ func TestLoadAndActivateBundlesFromDisk(t *testing.T) {
 	plugin := New(&Config{Bundles: bundles}, manager)
 	plugin.bundlePersistPath = filepath.Join(dir, ".opa")

-	err = plugin.loadAndActivateBundlesFromDisk(ctx)
-	if err != nil {
-		t.Fatal("unexpected error:", err)
-	}
+	plugin.loadAndActivateBundlesFromDisk(ctx)

 	// persist a bundle to disk and then load it
 	module := "package foo\n\ncorge=1"
@@ -439,10 +436,7 @@ func TestLoadAndActivateBundlesFromDisk(t *testing.T) {
 		t.Fatalf("unexpected error %v", err)
 	}

-	err = plugin.loadAndActivateBundlesFromDisk(ctx)
-	if err != nil {
-		t.Fatal("unexpected error:", err)
-	}
+	plugin.loadAndActivateBundlesFromDisk(ctx)

 	txn := storage.NewTransactionOrDie(ctx, manager.Store)
 	defer manager.Store.Abort(ctx, txn)
@@ -471,6 +465,186 @@ func TestLoadAndActivateBundlesFromDisk(t *testing.T) {
 	}
 }

+func TestLoadAndActivateDepBundlesFromDisk(t *testing.T) {
+	ctx := context.Background()
+	manager := getTestManager()
+
+	dir, err := ioutil.TempDir("", "")
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	defer os.RemoveAll(dir)
+
+	bundleName := "test-bundle-main"
+	bundleSource := Source{
+		Persist: true,
+	}
+
+	bundleNameOther := "test-bundle-lib"
+	bundleSourceOther := Source{
+		Persist: true,
+	}
+
+	bundles := map[string]*Source{}
+	bundles[bundleName] = &bundleSource
+	bundles[bundleNameOther] = &bundleSourceOther
+
+	plugin := New(&Config{Bundles: bundles}, manager)
+	plugin.bundlePersistPath = filepath.Join(dir, ".opa")
+
+	module1 := `
+package bar
+
+import data.foo
+
+default allow = false
+
+allow {
+	foo.is_one(1)
+}`
+
+	module2 := `
+package foo
+
+is_one(x) {
+	x == 1
+}`
+
+	b1 := bundle.Bundle{
+		Manifest: bundle.Manifest{Revision: "quickbrownfauxbar", Roots: &[]string{"bar"}},
+		Data:     map[string]interface{}{},
+		Modules: []bundle.ModuleFile{
+			{
+				URL:    "/bar/policy.rego",
+				Path:   "/bar/policy.rego",
+				Parsed: ast.MustParseModule(module1),
+				Raw:    []byte(module1),
+			},
+		},
+	}
+
+	b1.Manifest.Init()
+
+	b2 := bundle.Bundle{
+		Manifest: bundle.Manifest{Revision: "quickbrownfauxfoo", Roots: &[]string{"foo"}},
+		Data:     map[string]interface{}{},
+		Modules: []bundle.ModuleFile{
+			{
+				URL:    "/foo/policy.rego",
+				Path:   "/foo/policy.rego",
+				Parsed: ast.MustParseModule(module2),
+				Raw:    []byte(module2),
+			},
+		},
+	}
+
+	b2.Manifest.Init()
+
+	var buf1 bytes.Buffer
+	if err := bundle.NewWriter(&buf1).UseModulePath(true).Write(b1); err != nil {
+		t.Fatal("unexpected error:", err)
+	}
+
+	err = plugin.saveBundleToDisk(bundleName, &buf1)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	var buf2 bytes.Buffer
+	if err := bundle.NewWriter(&buf2).UseModulePath(true).Write(b2); err != nil {
+		t.Fatal("unexpected error:", err)
+	}
+
+	err = plugin.saveBundleToDisk(bundleNameOther, &buf2)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	plugin.loadAndActivateBundlesFromDisk(ctx)
+
+	txn := storage.NewTransactionOrDie(ctx, manager.Store)
+	defer manager.Store.Abort(ctx, txn)
+
+	ids, err := manager.Store.ListPolicies(ctx, txn)
+	if err != nil {
+		t.Fatal(err)
+	} else if len(ids) != 2 {
+		t.Fatal("Expected 2 policies")
+	}
+}
+
+func TestLoadAndActivateDepBundlesFromDiskMaxAttempts(t *testing.T) {
+	ctx := context.Background()
+	manager := getTestManager()
+
+	dir, err := ioutil.TempDir("", "")
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	defer os.RemoveAll(dir)
+
+	bundleName := "test-bundle-main"
+	bundleSource := Source{
+		Persist: true,
+	}
+
+	bundles := map[string]*Source{}
+	bundles[bundleName] = &bundleSource
+
+	plugin := New(&Config{Bundles: bundles}, manager)
+	plugin.bundlePersistPath = filepath.Join(dir, ".opa")
+
+	module := `
+package bar
+
+import data.foo
+
+default allow = false
+
+allow {
+	foo.is_one(1)
+}`
+
+	b := bundle.Bundle{
+		Manifest: bundle.Manifest{Revision: "quickbrownfaux", Roots: &[]string{"bar"}},
+		Data:     map[string]interface{}{},
+		Modules: []bundle.ModuleFile{
+			{
+				URL:    "/bar/policy.rego",
+				Path:   "/bar/policy.rego",
+				Parsed: ast.MustParseModule(module),
+				Raw:    []byte(module),
+			},
+		},
+	}
+
+	b.Manifest.Init()
+
+	var buf bytes.Buffer
+	if err := bundle.NewWriter(&buf).UseModulePath(true).Write(b); err != nil {
+		t.Fatal("unexpected error:", err)
+	}
+
+	err = plugin.saveBundleToDisk(bundleName, &buf)
+	if err != nil {
+		t.Fatalf("unexpected error %v", err)
+	}
+
+	plugin.loadAndActivateBundlesFromDisk(ctx)
+
+	txn := storage.NewTransactionOrDie(ctx, manager.Store)
+	defer manager.Store.Abort(ctx, txn)
+
+	ids, err := manager.Store.ListPolicies(ctx, txn)
+	if err != nil {
+		t.Fatal(err)
+	} else if len(ids) != 0 {
+		t.Fatal("Expected 0 policies")
+	}
+}
+
 func TestPluginOneShotCompileError(t *testing.T) {

 	ctx := context.Background()