diff --git a/pkg/adaptation/adaptation_suite_test.go b/pkg/adaptation/adaptation_suite_test.go index b8ebde7c..b1403c34 100644 --- a/pkg/adaptation/adaptation_suite_test.go +++ b/pkg/adaptation/adaptation_suite_test.go @@ -39,6 +39,7 @@ import ( nri "github.com/containerd/nri/pkg/adaptation" "github.com/containerd/nri/pkg/api" "github.com/containerd/nri/pkg/plugin" + "github.com/containerd/nri/pkg/stub" validator "github.com/containerd/nri/plugins/default-validator/builtin" ) @@ -95,6 +96,101 @@ var _ = Describe("Configuration", func() { Expect(plugin.Start(s.dir)).ToNot(Succeed()) }) }) + + When("the connection is lost while Start() is waiting for Configure()", func() { + var ( + inConfigure chan struct{} + releaseConfigure chan struct{} + ) + + BeforeEach(func() { + inConfigure = make(chan struct{}) + releaseConfigure = make(chan struct{}) + // Short request timeout so the runtime gives up on the blocked + // Configure RPC quickly and closes the connection. + nri.SetPluginRequestTimeout(200 * time.Millisecond) + s.Prepare( + &mockRuntime{}, + &mockPlugin{ + idx: "00", + name: "test", + configure: func(_ *mockPlugin, _ context.Context, _, _, _ string) (stub.EventMask, error) { + close(inConfigure) + <-releaseConfigure + return 0, fmt.Errorf("test: configure released") + }, + }, + ) + }) + + AfterEach(func() { + close(releaseConfigure) + nri.SetPluginRequestTimeout(nri.DefaultPluginRequestTimeout) + }) + + It("should cause plugin Start() to fail instead of deadlocking", func() { + var ( + runtime = s.runtime + plugin = s.plugins[0] + errCh = make(chan error, 1) + ) + + Expect(runtime.Start(s.dir)).To(Succeed()) + + go func() { + errCh <- plugin.Start(s.dir) + }() + + // Once the plugin's Configure handler is running we know register() + // has succeeded and stub.Start() is blocked on cfgErrC holding the + // stub lock. + Eventually(inConfigure, 2*time.Second).Should(BeClosed()) + + // The runtime's Configure RPC now times out and closes the + // connection, firing the plugin's ttrpc OnClose -> connClosed(). + select { + case <-time.After(3 * time.Second): + Fail("plugin Start() did not return: stub deadlocked waiting for Configure()") + case err := <-errCh: + Expect(err).To(HaveOccurred()) + } + }) + }) + + When("the connection is lost during plugin registration", func() { + BeforeEach(func() { + // Make the runtime give up on registration essentially at accept + // time so the connection is torn down while the plugin's Start() + // is still setting up. + nri.SetPluginRegistrationTimeout(1 * time.Nanosecond) + s.Prepare(&mockRuntime{}, &mockPlugin{idx: "00", name: "test"}) + }) + + AfterEach(func() { + nri.SetPluginRegistrationTimeout(nri.DefaultPluginRegistrationTimeout) + }) + + It("should cause plugin Start() to fail instead of hanging", func() { + var ( + runtime = s.runtime + plugin = s.plugins[0] + errCh = make(chan error, 1) + ) + + Expect(runtime.Start(s.dir)).To(Succeed()) + + go func() { + errCh <- plugin.Start(s.dir) + }() + + select { + case <-time.After(3 * time.Second): + Fail("plugin Start() did not return within 3s") + case err := <-errCh: + Expect(err).To(HaveOccurred()) + } + }) + }) }) var _ = Describe("Adaptation", func() { diff --git a/pkg/adaptation/suite_test.go b/pkg/adaptation/suite_test.go index 36c0830e..736cc0f8 100644 --- a/pkg/adaptation/suite_test.go +++ b/pkg/adaptation/suite_test.go @@ -385,6 +385,7 @@ type mockPlugin struct { pods map[string]*api.PodSandbox ctrs map[string]*api.Container + configure func(*mockPlugin, context.Context, string, string, string) (stub.EventMask, error) runPodSandbox func(*mockPlugin, *api.PodSandbox, *api.Container) error updatePodSandbox func(*mockPlugin, *api.PodSandbox, *api.LinuxResources, *api.LinuxResources) error postUpdatePodSandbox func(*mockPlugin, *api.PodSandbox, *api.Container) error @@ -575,7 +576,10 @@ func (m *mockPlugin) onClose() { } } -func (m *mockPlugin) Configure(_ context.Context, _, runtime, version string) (stub.EventMask, error) { +func (m *mockPlugin) Configure(ctx context.Context, cfg, runtime, version string) (stub.EventMask, error) { + if m.configure != nil { + return m.configure(m, ctx, cfg, runtime, version) + } m.q.Add(PluginConfigured) m.runtime = runtime diff --git a/pkg/stub/stub.go b/pkg/stub/stub.go index b0288bd6..62a15358 100644 --- a/pkg/stub/stub.go +++ b/pkg/stub/stub.go @@ -380,6 +380,9 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { return fmt.Errorf("stub already started") } stub.doneC = make(chan struct{}) + // Snapshot before any goroutines are running: Configure() rewrites this + // field without the stub lock once the ttrpc server is up. + cfgTimeout := stub.registrationTimeout err := stub.connect() if err != nil { @@ -423,6 +426,12 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { return fmt.Errorf("failed to multiplex ttrpc client connection: %w", err) } + // Create the error channels before ttrpc.NewClient spawns the goroutine + // whose OnClose callback (connClosed) may send on cfgErrC without holding + // stub.Lock(). + stub.srvErrC = make(chan error, 1) + stub.cfgErrC = make(chan error, 1) + clientOpts := []ttrpc.ClientOpts{ ttrpc.WithOnClose(func() { stub.connClosed() @@ -436,9 +445,6 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { } }() - stub.srvErrC = make(chan error, 1) - stub.cfgErrC = make(chan error, 1) - go func(l stdnet.Listener, doneC chan struct{}, srvErrC chan error) { srvErrC <- rpcs.Serve(ctx, l) close(doneC) @@ -456,8 +462,22 @@ func (stub *stub) Start(ctx context.Context) (retErr error) { return err } - if err = <-stub.cfgErrC; err != nil { - return err + // The runtime should Configure() us immediately after a successful register(). + // Bound the wait so Start() cannot block forever holding stub.Lock() if the + // runtime accepts registration but never issues Configure(), and let context + // cancellation and connection loss (via connClosed -> cfgErrC) break the wait. + cfgTimer := time.NewTimer(cfgTimeout) + defer cfgTimer.Stop() + + select { + case err = <-stub.cfgErrC: + if err != nil { + return err + } + case <-cfgTimer.C: + return fmt.Errorf("timed out waiting for Configure() from runtime after %s", cfgTimeout) + case <-ctx.Done(): + return fmt.Errorf("context cancelled while waiting for Configure() from runtime: %w", ctx.Err()) } stub.logger.Infof(ctx, "Started plugin %s...", stub.Name()) @@ -624,6 +644,14 @@ func (stub *stub) register(ctx context.Context) error { // Handle a lost connection. func (stub *stub) connClosed() { + // Start() may be blocked on cfgErrC while holding stub.Lock(). Signal it + // first (non-blocking; cfgErrC has cap 1) so it can return and release the + // lock before we take it below. + select { + case stub.cfgErrC <- fmt.Errorf("connection closed before Configure(): %w", ttrpc.ErrClosed): + default: + } + stub.Lock() stub.close() stub.Unlock() @@ -680,7 +708,13 @@ func (stub *stub) Configure(ctx context.Context, req *api.ConfigureRequest) (rpl stub.runtimeNRIVersion = req.NRIVersion defer func() { - stub.cfgErrC <- retErr + // Non-blocking: Start() may have already given up (timeout / ctx / conn + // loss) and is no longer receiving; the buffered slot may also already + // hold connClosed()'s error. + select { + case stub.cfgErrC <- retErr: + default: + } }() if handler := stub.handlers.Configure; handler == nil {