From 86721eaef9a927aadfb28b4c368cc4ab9636a4ed Mon Sep 17 00:00:00 2001 From: Asad Khan Date: Sun, 15 Oct 2023 19:20:12 +0200 Subject: [PATCH] fix for plugin state reconciliation Signed-off-by: Asad Khan --- plugins/bundle/plugin.go | 5 +- plugins/bundle/plugin_test.go | 199 +++++++++++++++++++++++++++++++++- 2 files changed, 196 insertions(+), 8 deletions(-) diff --git a/plugins/bundle/plugin.go b/plugins/bundle/plugin.go index f0047d7d67..6694cf07ec 100644 --- a/plugins/bundle/plugin.go +++ b/plugins/bundle/plugin.go @@ -540,8 +540,9 @@ func (p *Plugin) process(ctx context.Context, name string, u download.Update) { if etag, ok := p.etags[name]; ok && u.ETag == etag { p.log(name).Debug("Bundle load skipped, server replied with not modified.") p.status[name].SetError(nil) - // proposed fix to check plugin readiness on 304 - // p.checkPluginReadiness() + + // The downloader received a 304 (same etag as saved in local state), update plugin readiness + p.checkPluginReadiness() return } } diff --git a/plugins/bundle/plugin_test.go b/plugins/bundle/plugin_test.go index 9819f4ca9c..99609e2257 100644 --- a/plugins/bundle/plugin_test.go +++ b/plugins/bundle/plugin_test.go @@ -3107,9 +3107,6 @@ func TestPluginReadBundleEtagFromDiskStore(t *testing.T) { plugin.Reconfigure(ctx, cfg) - // Reconfigure should mark the state as not ready - ensurePluginState(t, plugin, plugins.StateNotReady) - // manually trigger bundle download go func() { _ = plugin.Loaders()["test"].Trigger(ctx) @@ -3117,9 +3114,6 @@ func TestPluginReadBundleEtagFromDiskStore(t *testing.T) { <-statusCh - // on bundle download with 304 the state should be OK - ensurePluginState(t, plugin, plugins.StateOK) - if notModifiedCount != 2 { t.Fatalf("Expected two bundle responses with HTTP status 304 but got %v", notModifiedCount) } @@ -3135,6 +3129,199 @@ func TestPluginReadBundleEtagFromDiskStore(t *testing.T) { }) } +func TestPluginStateReconciliationOnReconfigure(t *testing.T) { + // setup fake http server with mock bundle + mockBundles := map[string]bundle.Bundle{ + "b1": { + Data: map[string]interface{}{"b1": "x1"}, + Modules: []bundle.ModuleFile{}, + Manifest: bundle.Manifest{ + Roots: &[]string{"b1"}, + }, + }, + "b2": { + Data: map[string]interface{}{"b2": "x1"}, + Modules: []bundle.ModuleFile{}, + Manifest: bundle.Manifest{ + Roots: &[]string{"b2"}, + }, + }, + "b3_frequently_changing": { + Data: map[string]interface{}{"b3": "x1"}, + Modules: []bundle.ModuleFile{}, + Manifest: bundle.Manifest{ + Roots: &[]string{"b3"}, + }, + }, + } + + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + name := strings.TrimPrefix(r.URL.Path, "/") + etag := r.Header.Get("If-None-Match") + if etag == name && name != "b3_frequently_changing" { + w.WriteHeader(304) + return + } + + if name != "b3_frequently_changing" { + w.Header().Add("Etag", name) + } + w.WriteHeader(200) + + err := bundle.NewWriter(w).Write(mockBundles[name]) + if err != nil { + t.Fatal(err) + } + })) + + // setup plugin pointing at fake server + manager := getTestManagerWithOpts([]byte(fmt.Sprintf(`{ + "services": { + "default": { + "url": %q + } + } + }`, s.URL))) + + // setup manual trigger mode to simulate the downloader + var mode plugins.TriggerMode = "manual" + var delay int64 = 10 + polling := download.PollingConfig{MinDelaySeconds: &delay, MaxDelaySeconds: &delay} + serviceName := "default" + plugin := New(&Config{ + Bundles: map[string]*Source{ + "b1": { + Service: serviceName, + Config: download.Config{Trigger: &mode}, + Resource: "/b1", + SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes), + }, + }, + }, manager) + + statusCh := make(chan map[string]*Status) + + // register for bundle updates to observe changes + plugin.RegisterBulkListener("test-case", func(st map[string]*Status) { + statusCh <- st + }) + + ctx := context.Background() + err := plugin.Start(ctx) + if err != nil { + t.Fatal(err) + } + + // manually trigger bundle download + go func() { _ = plugin.Loaders()["b1"].Trigger(ctx) }() + <-statusCh + + // validate plugin started as expected + ensurePluginState(t, plugin, plugins.StateOK) + + // change the plugin state with multiple stages + stages := []struct { + name string + cfg *Config + noChangeDetected bool + }{ + { + name: "Add a bundle", // b1 is NOT Modified + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode}, Resource: "/b1", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + "b2": {Service: serviceName, Config: download.Config{Trigger: &mode}, Resource: "/b2", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + }, + { + name: "change download config", // both bundles are Not Modified + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode, Polling: polling}, Resource: "/b1", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + "b2": {Service: serviceName, Config: download.Config{Trigger: &mode, Polling: polling}, Resource: "/b2", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + }, + { + name: "pass the same config", // should be no change detected + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode, Polling: polling}, Resource: "/b1", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + "b2": {Service: serviceName, Config: download.Config{Trigger: &mode, Polling: polling}, Resource: "/b2", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + noChangeDetected: true, + }, + { + name: "revert download config for one bundle", // both bundles are Not Modified + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode}, Resource: "/b1", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + "b2": {Service: serviceName, Config: download.Config{Trigger: &mode, Polling: polling}, Resource: "/b2", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + }, + { + name: "remove a bundle", // b1 is Not Modified + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode}, Resource: "/b1", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + noChangeDetected: true, + }, + { + name: "change download config again", // b1 is Not Modified + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode, Polling: polling}, Resource: "/b1", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + }, + { + name: "add frequently changing bundle", + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode, Polling: polling}, Resource: "/b1", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + "b3_frequently_changing": {Service: serviceName, Config: download.Config{Trigger: &mode}, Resource: "/b3_frequently_changing", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + }, + { + name: "revert download config for Not Modified bundle", // b1 is Not Modified while b3_frequently_changing is modified + cfg: &Config{ + Bundles: map[string]*Source{ + "b1": {Service: serviceName, Config: download.Config{Trigger: &mode}, Resource: "/b2", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + "b3_frequently_changing": {Service: serviceName, Config: download.Config{Trigger: &mode}, Resource: "/b3_frequently_changing", SizeLimitBytes: int64(bundle.DefaultSizeLimitBytes)}, + }, + }, + }, + } + + for _, stage := range stages { + t.Run(stage.name, func(t *testing.T) { + plugin.Reconfigure(ctx, stage.cfg) + + if stage.noChangeDetected { + ensurePluginState(t, plugin, plugins.StateOK) + return + } + + ensurePluginState(t, plugin, plugins.StateNotReady) + + for name := range stage.cfg.Bundles { + go func() { + _ = plugin.Loaders()[name].Trigger(ctx) + }() + <-statusCh + } + + ensurePluginState(t, plugin, plugins.StateOK) + }) + } +} + func TestPluginManualTrigger(t *testing.T) { ctx := context.Background()