From b7078b2e19839ef6ca10505bf42c97af4a143cb5 Mon Sep 17 00:00:00 2001 From: Ashutosh Narkar Date: Wed, 19 May 2021 16:52:36 -0700 Subject: [PATCH] Fix OPA deadlock while stopping bundle plugin MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit fixes couple of issues that could result in blocking OPA: 1) When the bundle plugin attempts to stop the bundle downloader, it first grabs the lock on the plugin and then stops the downloader. The downloader in-turn calls the plugin’s callback function which now waits for the lock to be released by the plugin's stop function. This results in a deadlock. This commit fixes this issue by making sure the plugin's stop function releases the lock before stopping the downloader. 2) Another issue that could block OPA is when the stop function on the same downloader gets called multiple times. Fixes: #3363 Signed-off-by: Ashutosh Narkar --- download/download.go | 10 ++++++ download/download_test.go | 40 ++++++++++++++++++++++ plugins/bundle/plugin.go | 23 ++++++++++--- plugins/bundle/plugin_test.go | 64 +++++++++++++++++++++++++++++++++++ 4 files changed, 132 insertions(+), 5 deletions(-) diff --git a/download/download.go b/download/download.go index 5cafa70de0..23e635e459 100644 --- a/download/download.go +++ b/download/download.go @@ -54,6 +54,8 @@ type Downloader struct { respHdrTimeoutSec int64 wg sync.WaitGroup logger logging.Logger + mtx sync.Mutex + stopped bool } // New returns a new Downloader that can be started. @@ -112,11 +114,19 @@ func (d *Downloader) doStart(context.Context) { done := <-d.stop // blocks until there's something to read cancel() d.wg.Wait() + d.stopped = true close(done) } // Stop tells the Downloader to stop downloading bundles. func (d *Downloader) Stop(context.Context) { + d.mtx.Lock() + defer d.mtx.Unlock() + + if d.stopped { + return + } + done := make(chan struct{}) d.stop <- done <-done diff --git a/download/download_test.go b/download/download_test.go index cb00fd8ee4..f5c3d44401 100644 --- a/download/download_test.go +++ b/download/download_test.go @@ -56,6 +56,46 @@ func TestStartStop(t *testing.T) { d.Stop(ctx) } +func TestStopWithMultipleCalls(t *testing.T) { + ctx := context.Background() + fixture := newTestFixture(t) + + updates := make(chan *Update) + + config := Config{} + if err := config.ValidateAndInjectDefaults(); err != nil { + t.Fatal(err) + } + + d := New(config, fixture.client, "/bundles/test/bundle1").WithCallback(func(_ context.Context, u Update) { + updates <- &u + }) + + d.Start(ctx) + + // Give time for some download events to occur + time.Sleep(1 * time.Second) + + u1 := <-updates + + if u1.Bundle == nil || len(u1.Bundle.Modules) == 0 { + t.Fatal("expected bundle with at least one module but got:", u1) + } + + done := make(chan struct{}) + go func() { + d.Stop(ctx) + close(done) + }() + + d.Stop(ctx) + <-done + + if !d.stopped { + t.Fatal("expected downloader to be stopped") + } +} + func TestStartStopWithLongPollNotSupported(t *testing.T) { ctx := context.Background() diff --git a/plugins/bundle/plugin.go b/plugins/bundle/plugin.go index 8671c42f4b..d2da3395dc 100644 --- a/plugins/bundle/plugin.go +++ b/plugins/bundle/plugin.go @@ -42,6 +42,7 @@ type Plugin struct { cfgMtx sync.Mutex ready bool bundlePersistPath string + stopped bool } // New returns a new Plugin with the given config. @@ -99,7 +100,6 @@ func (p *Plugin) Start(ctx context.Context) error { p.initDownloaders() for name, dl := range p.downloaders { - p.log(name).Info("Starting bundle loader.") dl.Start(ctx) } @@ -109,8 +109,15 @@ func (p *Plugin) Start(ctx context.Context) error { // Stop stops the plugin. func (p *Plugin) Stop(ctx context.Context) { p.mtx.Lock() - defer p.mtx.Unlock() + stopDownloaders := map[string]bundleLoader{} for name, dl := range p.downloaders { + stopDownloaders[name] = dl + } + p.downloaders = nil + p.stopped = true + p.mtx.Unlock() + + for name, dl := range stopDownloaders { p.log(name).Info("Stopping bundle loader.") dl.Stop(ctx) } @@ -361,7 +368,9 @@ func (p *Plugin) process(ctx context.Context, name string, u download.Update) { if u.Error != nil { p.log(name).Error("Bundle load failed: %v", u.Error) p.status[name].SetError(u.Error) - p.downloaders[name].ClearCache() + if !p.stopped { + p.downloaders[name].ClearCache() + } return } @@ -376,7 +385,9 @@ func (p *Plugin) process(ctx context.Context, name string, u download.Update) { if err := p.activate(ctx, name, u.Bundle); err != nil { p.log(name).Error("Bundle activation failed: %v", err) p.status[name].SetError(err) - p.downloaders[name].ClearCache() + if !p.stopped { + p.downloaders[name].ClearCache() + } return } @@ -387,7 +398,9 @@ func (p *Plugin) process(ctx context.Context, name string, u download.Update) { if err != nil { p.log(name).Error("Persisting bundle to disk failed: %v", err) p.status[name].SetError(err) - p.downloaders[name].ClearCache() + if !p.stopped { + p.downloaders[name].ClearCache() + } return } p.log(name).Debug("Bundle persisted to disk successfully at path %v.", filepath.Join(p.bundlePersistPath, name)) diff --git a/plugins/bundle/plugin_test.go b/plugins/bundle/plugin_test.go index b70b5b5646..461514b8ad 100644 --- a/plugins/bundle/plugin_test.go +++ b/plugins/bundle/plugin_test.go @@ -107,6 +107,70 @@ func TestPluginStart(t *testing.T) { } } +func TestStop(t *testing.T) { + var longPollTimeout int64 = 3 + done := make(chan struct{}) + 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) + } + + close(done) + + // simulate long operation + time.Sleep(time.Duration(longPollTimeout) * time.Second) + fmt.Fprintln(w) // Note: this is an invalid bundle and will fail the download + })) + defer ts.Close() + + 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) + } + + baseConf := download.Config{Polling: download.PollingConfig{LongPollingTimeoutSeconds: &longPollTimeout}} + + plugin := Plugin{ + manager: manager, + status: map[string]*Status{}, + etags: map[string]string{}, + downloaders: map[string]bundleLoader{}, + } + bundleName := "test-bundle" + plugin.status[bundleName] = &Status{Name: bundleName} + + callback := func(ctx context.Context, u download.Update) { + plugin.oneShot(ctx, bundleName, u) + } + plugin.downloaders[bundleName] = download.New(baseConf, plugin.manager.Client(serviceName), bundleName).WithCallback(callback) + + err = plugin.Start(ctx) + if err != nil { + t.Fatal("unexpected error:", err) + } + + // Give time for a long poll request to be initiated + <-done + + plugin.Stop(ctx) + + if plugin.status[bundleName].Code != errCode { + t.Fatalf("expected error code %v but got %v", errCode, plugin.status[bundleName].Code) + } + + if !strings.Contains(plugin.status[bundleName].Message, "context canceled") { + t.Fatalf("unexpected error message %v", plugin.status[bundleName].Message) + } +} + func TestPluginOneShotBundlePersistence(t *testing.T) { ctx := context.Background()