diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go index 2b76d3bd348..5305546ccf9 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go @@ -62,6 +62,9 @@ type DRAPluginManager struct { wipingDelay time.Duration streamHandler StreamHandler + // withIdleTimeout is only for unit testing, ignore if <= 0. + withIdleTimeout time.Duration + wg sync.WaitGroup mutex sync.RWMutex @@ -115,7 +118,13 @@ func (m *monitoredPlugin) HandleConn(_ context.Context, stats grpcstats.ConnStat case *grpcstats.ConnEnd: // We have to ask for a reconnect, otherwise gRPC wouldn't try and // thus we wouldn't be notified about a restart of the plugin. - m.conn.Connect() + // + // This must be done in a goroutine because gRPC deadlocks + // when called directly from inside HandleConn when a connection + // goes idle (and only then). It looks like cc.idlenessMgr.ExitIdleMode + // in Connect tries to lock a mutex that is already locked by + // the caller of HandleConn. + go m.conn.Connect() default: return } @@ -361,12 +370,15 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic // The gRPC connection gets created once. gRPC then connects to the gRPC server on demand. target := "unix:" + endpoint logger.V(4).Info("Creating new gRPC connection", "target", target) - conn, err := grpc.NewClient( - target, + options := []grpc.DialOption{ grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithChainUnaryInterceptor(newMetricsInterceptor(driverName)), grpc.WithStatsHandler(mp), - ) + } + if pm.withIdleTimeout > 0 { + options = append(options, grpc.WithIdleTimeout(pm.withIdleTimeout)) + } + conn, err := grpc.NewClient(target, options...) if err != nil { return fmt.Errorf("create gRPC connection to DRA driver %s plugin at endpoint %s: %w", driverName, endpoint, err) } diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go index b3af8c75b32..c8eb975b9bb 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go @@ -185,6 +185,44 @@ func TestGRPCConnIsReused(t *testing.T) { require.Equal(t, 2, reusedConns[conn], "expected counter to be 2 but got %d", reusedConns[conn]) } +func TestGRPCConnUsableAfterIdle(t *testing.T) { + tCtx := ktesting.Init(t) + service := drapbv1.DRAPluginService + addr := path.Join(t.TempDir(), "dra.sock") + teardown, err := setupFakeGRPCServer(service, addr) + require.NoError(t, err) + defer teardown() + + driverName := "dummy-driver" + + // ensure the plugin we are using is registered + draPlugins := NewDRAPluginManager(tCtx, nil, nil, &mockStreamHandler{}, 0) + draPlugins.withIdleTimeout = 5 * time.Second + tCtx.ExpectNoError(draPlugins.add(driverName, addr, service, defaultClientCallTimeout), "add plugin") + plugin, err := draPlugins.GetPlugin(driverName) + tCtx.ExpectNoError(err, "get plugin") + + // The connection doesn't really become idle because HandleConn + // kicks it back to ready by calling Connect. Just sleep long + // enough here, the code should be reached... + tCtx.Log("Waiting for idle timeout...") + time.Sleep(2 * draPlugins.withIdleTimeout) + + req := &drapbv1.NodePrepareResourcesRequest{ + Claims: []*drapbv1.Claim{ + { + Namespace: "dummy-namespace", + UID: "dummy-uid", + Name: "dummy-claim", + }, + }, + } + + callCtx := ktesting.WithTimeout(tCtx, 10*time.Second, "call timed out") + _, err = plugin.NodePrepareResources(callCtx, req) + tCtx.ExpectNoError(err, "NodePrepareResources") +} + func TestGetDRAPlugin(t *testing.T) { for _, test := range []struct { description string