From f927cd0108858cc1688cf894d219b7058de95099 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Tue, 3 Jun 2025 12:23:57 +0200 Subject: [PATCH 1/9] DRA kubelet: simplify plugin creation and connection handling Instead of creating the gRPC connection on demand and forcing gRPC to connect, we establish it immediately and rely on gRPC to handle the underlying connection automatically like it usually does. It's not clear what benefit the one second connection timeout had. The way it is now, gRPC calls still fail when the underlying connection cannot be established. Having to have a separate context for establishing that connection just made the code more complex. The DRAPluginManager is the central component which manages plugins. Making it responsible for creating them reduces the number of places where a DRAPlugin struct needs to be initialized. Doing this in the DRAPluginManager instead of a stand-alone function simplifies the implementation of connection monitoring, because that will be something that is tied to the DRAPluginManager state. --- pkg/kubelet/cm/dra/plugin/dra_plugin.go | 76 ++----------- .../cm/dra/plugin/dra_plugin_manager.go | 103 +++++++++--------- .../cm/dra/plugin/dra_plugin_manager_test.go | 56 +++------- pkg/kubelet/cm/dra/plugin/dra_plugin_test.go | 50 ++------- 4 files changed, 88 insertions(+), 197 deletions(-) diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin.go b/pkg/kubelet/cm/dra/plugin/dra_plugin.go index ef1f52944f1..4b4b3092350 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin.go @@ -18,15 +18,10 @@ package plugin import ( "context" - "errors" "fmt" - "net" - "sync" "time" "google.golang.org/grpc" - "google.golang.org/grpc/connectivity" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" "k8s.io/klog/v2" @@ -48,58 +43,13 @@ const defaultClientCallTimeout = 45 * time.Second // It implements the kubelet operations for preparing/unpreparing by calling // a gRPC interface that is implemented by the plugin. type DRAPlugin struct { - driverName string - backgroundCtx context.Context - cancel func(cause error) - - mutex sync.Mutex + driverName string conn *grpc.ClientConn endpoint string chosenService string // e.g. drapbv1beta1.DRAPluginService clientCallTimeout time.Duration } -func (p *DRAPlugin) getOrCreateGRPCConn() (*grpc.ClientConn, error) { - p.mutex.Lock() - defer p.mutex.Unlock() - - if p.conn != nil { - return p.conn, nil - } - - ctx := p.backgroundCtx - logger := klog.FromContext(ctx) - - network := "unix" - logger.V(4).Info("Creating new gRPC connection", "protocol", network, "endpoint", p.endpoint) - // grpc.Dial is deprecated. grpc.NewClient should be used instead. - // For now this gets ignored because this function is meant to establish - // the connection, with the one second timeout below. Perhaps that - // approach should be reconsidered? - //nolint:staticcheck - conn, err := grpc.Dial( - p.endpoint, - grpc.WithTransportCredentials(insecure.NewCredentials()), - grpc.WithContextDialer(func(ctx context.Context, target string) (net.Conn, error) { - return (&net.Dialer{}).DialContext(ctx, network, target) - }), - grpc.WithChainUnaryInterceptor(newMetricsInterceptor(p.driverName)), - ) - if err != nil { - return nil, err - } - - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - - if ok := conn.WaitForStateChange(ctx, connectivity.Connecting); !ok { - return nil, errors.New("timed out waiting for gRPC connection to be ready") - } - - p.conn = conn - return p.conn, nil -} - func (p *DRAPlugin) DriverName() string { return p.driverName } @@ -110,23 +60,21 @@ func (p *DRAPlugin) NodePrepareResources( opts ...grpc.CallOption, ) (*drapbv1beta1.NodePrepareResourcesResponse, error) { logger := klog.FromContext(ctx) + logger = klog.LoggerWithValues(logger, "driverName", p.driverName, "endpoint", p.endpoint) + ctx = klog.NewContext(ctx, logger) logger.V(4).Info("Calling NodePrepareResources rpc", "request", req) - conn, err := p.getOrCreateGRPCConn() - if err != nil { - return nil, err - } - ctx, cancel := context.WithTimeout(ctx, p.clientCallTimeout) defer cancel() + var err error var response *drapbv1beta1.NodePrepareResourcesResponse switch p.chosenService { case drapbv1beta1.DRAPluginService: - nodeClient := drapbv1beta1.NewDRAPluginClient(conn) + nodeClient := drapbv1beta1.NewDRAPluginClient(p.conn) response, err = nodeClient.NodePrepareResources(ctx, req) case drapbv1alpha4.NodeService: - nodeClient := drapbv1alpha4.NewNodeClient(conn) + nodeClient := drapbv1alpha4.NewNodeClient(p.conn) response, err = drapbv1alpha4.V1Alpha4ClientWrapper{NodeClient: nodeClient}.NodePrepareResources(ctx, req) default: // Shouldn't happen, validateSupportedServices should only @@ -144,22 +92,20 @@ func (p *DRAPlugin) NodeUnprepareResources( ) (*drapbv1beta1.NodeUnprepareResourcesResponse, error) { logger := klog.FromContext(ctx) logger.V(4).Info("Calling NodeUnprepareResource rpc", "request", req) - - conn, err := p.getOrCreateGRPCConn() - if err != nil { - return nil, err - } + logger = klog.LoggerWithValues(logger, "driverName", p.driverName, "endpoint", p.endpoint) + ctx = klog.NewContext(ctx, logger) ctx, cancel := context.WithTimeout(ctx, p.clientCallTimeout) defer cancel() + var err error var response *drapbv1beta1.NodeUnprepareResourcesResponse switch p.chosenService { case drapbv1beta1.DRAPluginService: - nodeClient := drapbv1beta1.NewDRAPluginClient(conn) + nodeClient := drapbv1beta1.NewDRAPluginClient(p.conn) response, err = nodeClient.NodeUnprepareResources(ctx, req) case drapbv1alpha4.NodeService: - nodeClient := drapbv1alpha4.NewNodeClient(conn) + nodeClient := drapbv1alpha4.NewNodeClient(p.conn) response, err = drapbv1alpha4.V1Alpha4ClientWrapper{NodeClient: nodeClient}.NodeUnprepareResources(ctx, req) default: // Shouldn't happen, validateSupportedServices should only diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go index bbc35597a31..307edebb35f 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go @@ -24,6 +24,8 @@ import ( "sync" "time" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" v1 "k8s.io/api/core/v1" resourceapi "k8s.io/api/resource/v1beta1" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -35,6 +37,7 @@ import ( drapbv1alpha4 "k8s.io/kubelet/pkg/apis/dra/v1alpha4" drapbv1beta1 "k8s.io/kubelet/pkg/apis/dra/v1beta1" "k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache" + "k8s.io/utils/ptr" ) // DRAPluginManager keeps track of how to reach plugins registered for DRA drivers. @@ -44,7 +47,7 @@ import ( // [cache.PluginHandler] interface and needs to be added to the // plugin manager. // -// The null DRAPluginManager is not usable, use NewPluginManager. +// The null DRAPluginManager is not usable, use NewDRAPluginManager. type DRAPluginManager struct { // backgroundCtx is used for all future activities of the DRAPluginManager. // This is necessary because it implements APIs which don't @@ -58,7 +61,7 @@ type DRAPluginManager struct { wg sync.WaitGroup mutex sync.RWMutex - // driver name -> Plugin in the order in which they got added + // driver name -> DRAPlugin in the order in which they got added store map[string][]*DRAPlugin // pendingWipes maps a driver name to a cancel function for @@ -112,8 +115,20 @@ func NewDRAPluginManager(ctx context.Context, kubeClient kubernetes.Interface, g // Stop cancels any remaining background activities and blocks until all goroutines have stopped. func (pm *DRAPluginManager) Stop() { + defer pm.wg.Wait() // Must run after unlocking our mutex. + pm.mutex.Lock() + defer pm.mutex.Unlock() + pm.cancel(errors.New("Stop was called")) - pm.wg.Wait() + + // Close all connections, otherwise gRPC keeps doing things in the background. + for _, plugins := range pm.store { + for _, plugin := range plugins { + if err := plugin.conn.Close(); err != nil { + klog.FromContext(pm.backgroundCtx).Error(err, "Closing gRPC connection", "driverName", plugin.driverName, "endpoint", plugin.endpoint) + } + } + } } // wipeResourceSlices deletes ResourceSlices of the node, optionally just for a specific driver. @@ -196,7 +211,7 @@ func (pm *DRAPluginManager) GetPlugin(driverName string) (*DRAPlugin, error) { return plugin, nil } -// get lets you retrieve a DRA Plugin by name. +// get lets you retrieve a DRA DRAPlugin by name. func (pm *DRAPluginManager) get(driverName string) *DRAPlugin { pm.mutex.RLock() defer pm.mutex.RUnlock() @@ -223,46 +238,16 @@ func (pm *DRAPluginManager) get(driverName string) *DRAPlugin { // in advance which version to use resp. which optional services the plugin // supports. func (pm *DRAPluginManager) RegisterPlugin(driverName string, endpoint string, supportedServices []string, pluginClientTimeout *time.Duration) error { - // Prepare a context with its own logger for the plugin. - // - // The lifecycle of the plugin's background activities is tied to our - // root context, so canceling that will also cancel the plugin. - // - // The logger injects the driver name and endpoint as additional values - // into all log output related to the plugin. - ctx := pm.backgroundCtx - logger := klog.FromContext(ctx) - logger = klog.LoggerWithValues(logger, "driverName", driverName, "endpoint", endpoint) - ctx = klog.NewContext(ctx, logger) - chosenService, err := pm.validateSupportedServices(driverName, supportedServices) if err != nil { return fmt.Errorf("invalid supported gRPC versions of DRA driver plugin %s at endpoint %s: %w", driverName, endpoint, err) } - var timeout time.Duration - if pluginClientTimeout == nil { - timeout = defaultClientCallTimeout - } else { - timeout = *pluginClientTimeout - } + timeout := ptr.Deref(pluginClientTimeout, defaultClientCallTimeout) - ctx, cancel := context.WithCancelCause(ctx) - - plugin := &DRAPlugin{ - driverName: driverName, - backgroundCtx: ctx, - cancel: cancel, - conn: nil, - endpoint: endpoint, - chosenService: chosenService, - clientCallTimeout: timeout, - } - - // Storing endpoint of newly registered DRA Plugin into the map, where the DRA driver name will be the key + // Storing endpoint of newly registered DRA DRAPlugin into the map, where the DRA driver name will be the key // under which the manager will be able to get a plugin when it needs to call it. - if err := pm.add(plugin); err != nil { - cancel(err) + if err := pm.add(driverName, endpoint, chosenService, timeout); err != nil { // No wrapping, the error already contains details. return err } @@ -270,23 +255,44 @@ func (pm *DRAPluginManager) RegisterPlugin(driverName string, endpoint string, s return nil } -func (pm *DRAPluginManager) add(p *DRAPlugin) error { +func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenService string, clientCallTimeout time.Duration) error { pm.mutex.Lock() defer pm.mutex.Unlock() if pm.store == nil { pm.store = make(map[string][]*DRAPlugin) } - for _, oldP := range pm.store[p.driverName] { - if oldP.endpoint == p.endpoint { + for _, oldP := range pm.store[driverName] { + if oldP.endpoint == endpoint { // One plugin instance cannot hijack the endpoint of another instance. - return fmt.Errorf("endpoint %s already registered for plugin %s", p.endpoint, p.driverName) + return fmt.Errorf("endpoint %s already registered for DRA driver plugin %s", endpoint, driverName) } } - logger := klog.FromContext(p.backgroundCtx) + logger := klog.FromContext(pm.backgroundCtx) + + // 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, + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithChainUnaryInterceptor(newMetricsInterceptor(driverName)), + ) + if err != nil { + return fmt.Errorf("create gRPC connection to DRA driver %s plugin at endpoint %s: %w", driverName, endpoint, err) + } + + p := &DRAPlugin{ + driverName: driverName, + endpoint: endpoint, + conn: conn, + chosenService: chosenService, + clientCallTimeout: clientCallTimeout, + } + pm.store[p.driverName] = append(pm.store[p.driverName], p) - logger.V(3).Info("Registered DRA plugin", "numInstances", len(pm.store[p.driverName])) + logger.V(3).Info("Registered DRA plugin", "driverName", p.driverName, "endpoint", p.endpoint, "chosenService", p.chosenService, "numPlugins", len(pm.store[p.driverName])) pm.sync(p.driverName) return nil } @@ -317,13 +323,8 @@ func (pm *DRAPluginManager) remove(driverName, endpoint string) { } else { pm.store[driverName] = slices.Delete(plugins, i, i+1) } - if p.cancel != nil { - // This cancels background attempts to establish a connection to the plugin. - // TODO: remove this in favor of non-blocking connection management. - p.cancel(errors.New("plugin got removed")) - } - logger := klog.FromContext(p.backgroundCtx) - logger.V(3).Info("Unregistered DRA plugin", "numInstances", len(pm.store[driverName])) + logger := klog.FromContext(pm.backgroundCtx) + logger.V(3).Info("Unregistered DRA plugin", "driverName", p.driverName, "endpoint", p.endpoint, "numPlugins", len(pm.store[driverName])) pm.sync(driverName) } @@ -350,7 +351,7 @@ func (pm *DRAPluginManager) sync(driverName string) { ctx, cancel := context.WithCancelCause(pm.backgroundCtx) ctx = klog.NewContext(ctx, logger) - // Clean up the ResourceSlices for the deleted Plugin since it + // Clean up the ResourceSlices for the deleted DRAPlugin since it // may have died without doing so itself and might never come // back. // diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager_test.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager_test.go index 2abc9869af9..c104c9a88fb 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager_test.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager_test.go @@ -21,7 +21,6 @@ import ( "math/rand/v2" "testing" - "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "k8s.io/kubernetes/test/utils/ktesting" ) @@ -31,58 +30,39 @@ func TestAddSameName(t *testing.T) { // name will have a random value to avoid conflicts driverName := fmt.Sprintf("dummy-driver-%d", rand.IntN(10000)) - firstWasCancelled := false - p := &DRAPlugin{ - driverName: driverName, - backgroundCtx: tCtx, - endpoint: "old", - cancel: func(err error) { firstWasCancelled = true }, - } - // ensure the plugin we are using is registered draPlugins := NewDRAPluginManager(tCtx, nil, nil, 0) - require.NoError(t, draPlugins.add(p)) - - assert.False(t, firstWasCancelled, "should not cancel context after the first call") + tCtx.ExpectNoError(draPlugins.add(driverName, "old.sock", "", defaultClientCallTimeout), "add first plugin") + p, err := draPlugins.GetPlugin(driverName) + tCtx.ExpectNoError(err, "get first plugin") // Same name, same endpoint -> error. - require.Error(t, draPlugins.add(p)) + require.Error(tCtx, draPlugins.add(driverName, "old.sock", "", defaultClientCallTimeout)) - secondWasCancelled := false - p2 := &DRAPlugin{ - driverName: driverName, - backgroundCtx: tCtx, - endpoint: "new", - cancel: func(err error) { secondWasCancelled = true }, + tCtx.ExpectNoError(draPlugins.add(driverName, "new.sock", "", defaultClientCallTimeout), "add second plugin") + p2, err := draPlugins.GetPlugin(driverName) + tCtx.ExpectNoError(err, "get second plugin") + if p == p2 { + tCtx.Fatal("expected to get second plugin, got first one again") } - require.NoError(t, draPlugins.add(p2)) - defer draPlugins.remove(p2.driverName, p2.endpoint) - - assert.False(t, firstWasCancelled, "should not cancel context after registering the second instance") - assert.False(t, secondWasCancelled, "should not cancel context of a new plugin") // Remove old plugin. draPlugins.remove(p.driverName, p.endpoint) - assert.True(t, firstWasCancelled, "should have canceled context after the explicit removal") - assert.False(t, secondWasCancelled, "should not cancel context of a new plugin") + plugin, err := draPlugins.GetPlugin(driverName) + + // Now we should have p2 left. + tCtx.ExpectNoError(err, "get plugin") + if p2 != plugin { + tCtx.Fatal("expected to get second plugin again, got something else") + } } func TestDelete(t *testing.T) { tCtx := ktesting.Init(t) driverName := fmt.Sprintf("dummy-driver-%d", rand.IntN(10000)) - wasCancelled := false - p := &DRAPlugin{ - driverName: driverName, - backgroundCtx: tCtx, - cancel: func(err error) { wasCancelled = true }, - } - // ensure the plugin we are using is registered draPlugins := NewDRAPluginManager(tCtx, nil, nil, 0) - draPlugins.add(p) - - draPlugins.remove(p.driverName, "") - - assert.True(t, wasCancelled, "should cancel context after the second call") + tCtx.ExpectNoError(draPlugins.add(driverName, "dra.sock", "", defaultClientCallTimeout), "add plugin") + draPlugins.remove(driverName, "") } diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go index 9627e45e8b4..2644661cd8a 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go @@ -117,28 +117,13 @@ func TestGRPCConnIsReused(t *testing.T) { m := sync.Mutex{} driverName := "dummy-driver" - p := &DRAPlugin{ - driverName: driverName, - backgroundCtx: tCtx, - endpoint: addr, - chosenService: service, - clientCallTimeout: defaultClientCallTimeout, - } - - conn, err := p.getOrCreateGRPCConn() - defer func() { - err := conn.Close() - if err != nil { - t.Error(err) - } - }() - if err != nil { - t.Fatal(err) - } // ensure the plugin we are using is registered draPlugins := NewDRAPluginManager(tCtx, nil, nil, 0) - tCtx.ExpectNoError(draPlugins.add(p), "add plugin") + tCtx.ExpectNoError(draPlugins.add(driverName, addr, service, defaultClientCallTimeout), "add plugin") + plugin, err := draPlugins.GetPlugin(driverName) + tCtx.ExpectNoError(err, "get plugin") + conn := plugin.conn // we call `NodePrepareResource` 2 times and check whether a new connection is created or the same is reused for i := 0; i < 2; i++ { @@ -164,9 +149,7 @@ func TestGRPCConnIsReused(t *testing.T) { _, err = plugin.NodePrepareResources(tCtx, req) assert.NoError(t, err) - plugin.mutex.Lock() conn := plugin.conn - plugin.mutex.Unlock() m.Lock() defer m.Unlock() @@ -196,14 +179,14 @@ func TestGetDRAPlugin(t *testing.T) { shouldError: true, }, { - description: "driver-name not found in the list", + description: "driver name not found in the list", driverName: "driver-name-not-found-in-the-list", shouldError: true, }, { description: "plugin exists", setup: func(draPlugins *DRAPluginManager) error { - return draPlugins.add(&DRAPlugin{backgroundCtx: draPlugins.backgroundCtx, driverName: "dummy-driver"}) + return draPlugins.add("dummy-driver", "/tmp/dra.sock", "", defaultClientCallTimeout) }, driverName: "dummy-driver", }, @@ -272,27 +255,8 @@ func TestGRPCMethods(t *testing.T) { defer teardown() driverName := "dummy-driver" - p := &DRAPlugin{ - driverName: driverName, - backgroundCtx: tCtx, - endpoint: addr, - chosenService: test.chosenService, - clientCallTimeout: defaultClientCallTimeout, - } - - conn, err := p.getOrCreateGRPCConn() - defer func() { - err := conn.Close() - if err != nil { - t.Error(err) - } - }() - if err != nil { - t.Fatal(err) - } - draPlugins := NewDRAPluginManager(tCtx, nil, nil, 0) - draPlugins.add(p) + tCtx.ExpectNoError(draPlugins.add(driverName, addr, test.chosenService, defaultClientCallTimeout)) plugin, err := draPlugins.GetPlugin(driverName) if err != nil { t.Fatal(err) From 4ee7374b24e6cd788e96fc0362ca5f08b686dc04 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Wed, 26 Mar 2025 20:05:48 +0200 Subject: [PATCH 2/9] DRA kubelet: add connection monitoring This ensures that ResourceSlices get removed also when a plugin becomes unresponsive without removing the registration socket. Tests are from https://github.com/kubernetes/kubernetes/pull/131073 by Ed with some modifications, the implementation is new. --- .../cm/dra/plugin/dra_plugin_manager.go | 117 +++++++- pkg/kubelet/cm/dra/plugin/dra_plugin_test.go | 34 +-- .../cm/dra/plugin/registration_test.go | 242 ++++++++++++----- .../pluginmanager/pluginwatcher/README.md | 52 ++++ .../kubeletplugin/draplugin.go | 73 +++-- test/e2e_node/dra_test.go | 252 ++++++++++++++++-- 6 files changed, 636 insertions(+), 134 deletions(-) diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go index 307edebb35f..50eedce9b1a 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go @@ -26,6 +26,8 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + grpcstats "google.golang.org/grpc/stats" + v1 "k8s.io/api/core/v1" resourceapi "k8s.io/api/resource/v1beta1" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -58,11 +60,21 @@ type DRAPluginManager struct { getNode func() (*v1.Node, error) wipingDelay time.Duration + // TODO: replace pendingWipes with some kind of workqueue. + // As it stands, WaitGroup suffers from a data race: + // - Queueing a new wiping creates a goroutine and adds to + // to wg. + // - Concurrently, wg.Wait reads from it. + // + // This is not allowed, all wg.Adds must come before wg.Wait. + // + // This race can be triggered with + // go test -count=10 -race ./... wg sync.WaitGroup mutex sync.RWMutex // driver name -> DRAPlugin in the order in which they got added - store map[string][]*DRAPlugin + store map[string][]*monitoredPlugin // pendingWipes maps a driver name to a cancel function for // wiping of that plugin's ResourceSlices. Entries get added @@ -76,6 +88,54 @@ type DRAPluginManager struct { var _ cache.PluginHandler = &DRAPluginManager{} +// monitoredPlugin tracks whether the gRPC connection of a plugin is +// currently connected. Fot that it implements the [grpcstats.Handler] +// interface. +// +// The tagging functions might be useful for contextual logging. But +// for now all that matters is HandleConn. +type monitoredPlugin struct { + *DRAPlugin + pm *DRAPluginManager + + // connected is protected by store.mutex. + connected bool +} + +var _ grpcstats.Handler = &monitoredPlugin{} + +func (m *monitoredPlugin) TagRPC(ctx context.Context, info *grpcstats.RPCTagInfo) context.Context { + return ctx +} + +func (m *monitoredPlugin) HandleRPC(context.Context, grpcstats.RPCStats) { +} + +func (m *monitoredPlugin) TagConn(ctx context.Context, info *grpcstats.ConnTagInfo) context.Context { + return ctx +} + +func (m *monitoredPlugin) HandleConn(_ context.Context, stats grpcstats.ConnStats) { + connected := false + switch stats.(type) { + case *grpcstats.ConnBegin: + connected = true + 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() + default: + return + } + + logger := klog.FromContext(m.pm.backgroundCtx) + m.pm.mutex.Lock() + defer m.pm.mutex.Unlock() + logger.V(2).Info("Connection changed", "driverName", m.driverName, "endpoint", m.endpoint, "connected", connected) + m.connected = connected + m.pm.sync(m.driverName) +} + // NewDRAPluginManager creates a new DRAPluginManager, with support for wiping ResourceSlices // when the plugin(s) for a DRA driver are not available too long. // @@ -216,14 +276,29 @@ func (pm *DRAPluginManager) get(driverName string) *DRAPlugin { pm.mutex.RLock() defer pm.mutex.RUnlock() + logger := klog.FromContext(pm.backgroundCtx) + plugins := pm.store[driverName] if len(plugins) == 0 { + logger.V(5).Info("No plugin registered", "driverName", driverName) return nil } + // Heuristic: pick the most recent one. It's most likely // the newest, except when kubelet got restarted and registered // all running plugins in random order. - return plugins[len(plugins)-1] + // + // Prefer plugins which are connected, otherwise also + // disconnected ones. + for i := len(plugins) - 1; i >= 0; i-- { + if plugin := plugins[i]; plugin.connected { + logger.V(5).Info("Preferring connected plugin", "driverName", driverName, "endpoint", plugin.endpoint) + return plugin.DRAPlugin + } + } + plugin := plugins[len(plugins)-1] + logger.V(5).Info("No plugin connected, using latest one", "driverName", driverName, "endpoint", plugin.endpoint) + return plugin.DRAPlugin } // RegisterPlugin implements [cache.PluginHandler]. @@ -259,8 +334,14 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic pm.mutex.Lock() defer pm.mutex.Unlock() + p := &DRAPlugin{ + driverName: driverName, + endpoint: endpoint, + chosenService: chosenService, + clientCallTimeout: clientCallTimeout, + } if pm.store == nil { - pm.store = make(map[string][]*DRAPlugin) + pm.store = make(map[string][]*monitoredPlugin) } for _, oldP := range pm.store[driverName] { if oldP.endpoint == endpoint { @@ -271,6 +352,11 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic logger := klog.FromContext(pm.backgroundCtx) + mp := &monitoredPlugin{ + DRAPlugin: p, + pm: pm, + } + // 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) @@ -278,20 +364,20 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic target, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithChainUnaryInterceptor(newMetricsInterceptor(driverName)), + grpc.WithStatsHandler(mp), ) if err != nil { return fmt.Errorf("create gRPC connection to DRA driver %s plugin at endpoint %s: %w", driverName, endpoint, err) } + p.conn = conn - p := &DRAPlugin{ - driverName: driverName, - endpoint: endpoint, - conn: conn, - chosenService: chosenService, - clientCallTimeout: clientCallTimeout, - } + // Ensure that gRPC tries to connect even if we don't call any gRPC method. + // This is necessary to detect early whether a plugin is really available. + // This is currently an experimental gRPC method. Should it be removed we + // would need to do something else, like sending a fake gRPC method call. + conn.Connect() - pm.store[p.driverName] = append(pm.store[p.driverName], p) + pm.store[p.driverName] = append(pm.store[p.driverName], mp) logger.V(3).Info("Registered DRA plugin", "driverName", p.driverName, "endpoint", p.endpoint, "chosenService", p.chosenService, "numPlugins", len(pm.store[p.driverName])) pm.sync(p.driverName) return nil @@ -312,7 +398,7 @@ func (pm *DRAPluginManager) remove(driverName, endpoint string) { defer pm.mutex.Unlock() plugins := pm.store[driverName] - i := slices.IndexFunc(plugins, func(p *DRAPlugin) bool { return p.driverName == driverName && p.endpoint == endpoint }) + i := slices.IndexFunc(plugins, func(mp *monitoredPlugin) bool { return mp.driverName == driverName && mp.endpoint == endpoint }) if i == -1 { return } @@ -382,7 +468,12 @@ func (pm *DRAPluginManager) sync(driverName string) { // usable returns true if at least one endpoint is ready to handle gRPC calls for the DRA driver. // Must be called while holding the mutex. func (pm *DRAPluginManager) usable(driverName string) bool { - return len(pm.store[driverName]) > 0 + for _, mp := range pm.store[driverName] { + if mp.connected { + return true + } + } + return false } // ValidatePlugin implements [cache.PluginHandler]. diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go index 2644661cd8a..9d506577032 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go @@ -21,7 +21,7 @@ import ( "fmt" "net" "os" - "path/filepath" + "path" "strings" "sync" "testing" @@ -56,18 +56,13 @@ func (f *fakeGRPCServer) NodeUnprepareResources(ctx context.Context, in *drapbv1 return &drapbv1beta1.NodeUnprepareResourcesResponse{}, nil } +// tearDown is an idempotent cleanup function. type tearDown func() -func setupFakeGRPCServer(service string) (string, tearDown, error) { - p, err := os.MkdirTemp("", "dra_plugin") - if err != nil { - return "", nil, err - } - - closeCh := make(chan struct{}) - addr := filepath.Join(p, "server.sock") +func setupFakeGRPCServer(service, addr string) (tearDown, error) { + ctx, cancel := context.WithCancel(context.Background()) teardown := func() { - close(closeCh) + cancel() if err := os.RemoveAll(addr); err != nil { panic(err) } @@ -76,7 +71,7 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) { listener, err := net.Listen("unix", addr) if err != nil { teardown() - return "", nil, err + return nil, err } s := grpc.NewServer() @@ -87,7 +82,7 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) { case drapbv1alpha4.NodeService: drapbv1alpha4.RegisterNodeServer(s, drapbv1alpha4.V1Beta1ServerWrapper{DRAPluginServer: fakeGRPCServer}) default: - return "", nil, fmt.Errorf("unsupported gRPC service: %s", service) + return nil, fmt.Errorf("unsupported gRPC service: %s", service) } go func() { @@ -96,17 +91,18 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) { panic(err) } }() - <-closeCh + <-ctx.Done() s.GracefulStop() }() - return addr, teardown, nil + return teardown, nil } func TestGRPCConnIsReused(t *testing.T) { tCtx := ktesting.Init(t) service := drapbv1beta1.DRAPluginService - addr, teardown, err := setupFakeGRPCServer(service) + addr := path.Join(t.TempDir(), "dra.sock") + teardown, err := setupFakeGRPCServer(service, addr) if err != nil { t.Fatal(err) } @@ -212,27 +208,23 @@ func TestGetDRAPlugin(t *testing.T) { func TestGRPCMethods(t *testing.T) { for _, test := range []struct { description string - serverSetup func(string) (string, tearDown, error) service string chosenService string expectError string }{ { description: "v1alpha4", - serverSetup: setupFakeGRPCServer, service: drapbv1alpha4.NodeService, chosenService: drapbv1alpha4.NodeService, }, { description: "v1beta1", - serverSetup: setupFakeGRPCServer, service: drapbv1beta1.DRAPluginService, chosenService: drapbv1beta1.DRAPluginService, }, { // In practice, such a mismatch between plugin and kubelet should not happen. description: "mismatch", - serverSetup: setupFakeGRPCServer, service: drapbv1beta1.DRAPluginService, chosenService: drapbv1alpha4.NodeService, expectError: "unknown service v1alpha3.Node", @@ -240,7 +232,6 @@ func TestGRPCMethods(t *testing.T) { { // In practice, kubelet wouldn't choose an invalid service. description: "internal-error", - serverSetup: setupFakeGRPCServer, service: drapbv1beta1.DRAPluginService, chosenService: "some-other-service", expectError: "unsupported chosen service", @@ -248,7 +239,8 @@ func TestGRPCMethods(t *testing.T) { } { t.Run(test.description, func(t *testing.T) { tCtx := ktesting.Init(t) - addr, teardown, err := setupFakeGRPCServer(test.service) + addr := path.Join(t.TempDir(), "dra.sock") + teardown, err := setupFakeGRPCServer(test.service, addr) if err != nil { t.Fatal(err) } diff --git a/pkg/kubelet/cm/dra/plugin/registration_test.go b/pkg/kubelet/cm/dra/plugin/registration_test.go index 8012a060799..6cd505df934 100644 --- a/pkg/kubelet/cm/dra/plugin/registration_test.go +++ b/pkg/kubelet/cm/dra/plugin/registration_test.go @@ -17,11 +17,13 @@ limitations under the License. package plugin import ( + "path" "sort" "strings" "testing" "time" + "github.com/onsi/gomega" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -39,17 +41,78 @@ import ( ) const ( - nodeName = "worker" - pluginA = "pluginA" - endpointA = "endpointA" - pluginB = "pluginB" - endpointB = "endpointB" + nodeName = "worker" + pluginA = "pluginA" + pluginB = "pluginB" ) func getFakeNode() (*v1.Node, error) { return &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}, nil } +func getSlice(name string) *resourceapi.ResourceSlice { + return &resourceapi.ResourceSlice{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: resourceapi.ResourceSliceSpec{ + NodeName: nodeName, + }, + } +} + +func getFakeClient(t *testing.T, nodeName, driverName string, slice *resourceapi.ResourceSlice) kubernetes.Interface { + expectedSliceFields := fields.Set{"spec.nodeName": nodeName} + fakeClient := fake.NewClientset(slice) + fakeClient.AddReactor("delete-collection", "resourceslices", func(action cgotesting.Action) (bool, runtime.Object, error) { + deleteAction := action.(cgotesting.DeleteCollectionAction) + restrictions := deleteAction.GetListRestrictions() + fieldsSelector := fields.SelectorFromSet(expectedSliceFields) + // The order of field requirements is random because it comes + // from a map. We need to sort. + normalize := func(selector string) string { + requirements := strings.Split(selector, ",") + sort.Strings(requirements) + return strings.Join(requirements, ",") + } + assert.Empty(t, restrictions.Labels.String(), "label selector in DeleteCollection") + assert.Equal(t, normalize(fieldsSelector.String()), normalize(restrictions.Fields.String()), "field selector in DeleteCollection") + + // There's only one object that could get matched, so delete it. + // Delete doesn't return an error if already deleted, which is what + // we need here (no error when nothing to delete). + err := fakeClient.Tracker().Delete(resourceapi.SchemeGroupVersion.WithResource("resourceslices"), "", slice.Name) + + // Set expected slice fields for the next call of this reactor. + // The reactor will be called next time when resourceslices object is deleted + // by the kubelet after plugin deregistration. + switch len(expectedSliceFields) { + case 1: + // Startup cleanup done, now expect cleanup for test plugin. + expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": driverName} + case 2: + // Test plugin cleanup done, now expect cleanup for the other plugin. + otherPlugin := pluginA + if otherPlugin == driverName { + otherPlugin = pluginB + } + expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": otherPlugin} + } + return true, nil, err + }) + return fakeClient +} + +func requireNoSlices(tCtx ktesting.TContext) { + tCtx.Helper() + ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) error { + slices, err := tCtx.Client().ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{}) + if err != nil { + return err + } + assert.Empty(tCtx, slices.Items, "slices") + return nil + }).Should(gomega.Succeed(), "there should be no slices") +} + func TestRegistrationHandler(t *testing.T) { slice := &resourceapi.ResourceSlice{ ObjectMeta: metav1.ObjectMeta{Name: "test-slice"}, @@ -58,6 +121,10 @@ func TestRegistrationHandler(t *testing.T) { }, } + tmp := t.TempDir() + endpointA := path.Join(tmp, "dra-plugin-a.sock") + endpointB := path.Join(tmp, "dra-plugin-b.sock") + for _, test := range []struct { description string driverName string @@ -121,71 +188,43 @@ func TestRegistrationHandler(t *testing.T) { t.Run(test.description, func(t *testing.T) { tCtx := ktesting.Init(t) + // Run GRPC services for both plugins. + // + // This is necessary because otherwise connection + // monitoring will start wiping slices, regardless + // of whether the plugin is registered or not. + // + // Here we are only interested in registration. + // In TestConnectionHandling we check detection + // of the connection state. + + service := drapb.DRAPluginService + teardownA, err := setupFakeGRPCServer(service, endpointA) + require.NoError(t, err) + tCtx.Cleanup(teardownA) + + teardown, err := setupFakeGRPCServer(service, test.endpoint) + require.NoError(t, err) + tCtx.Cleanup(teardown) + // Stand-alone kubelet has no connection to an // apiserver, so faking one is optional. var client kubernetes.Interface if test.withClient { - expectedSliceFields := fields.Set{"spec.nodeName": nodeName} - fakeClient := fake.NewClientset(slice) - fakeClient.AddReactor("delete-collection", "resourceslices", func(action cgotesting.Action) (bool, runtime.Object, error) { - deleteAction := action.(cgotesting.DeleteCollectionAction) - restrictions := deleteAction.GetListRestrictions() - fieldsSelector := fields.SelectorFromSet(expectedSliceFields) - // The order of field requirements is random because it comes - // from a map. We need to sort. - normalize := func(selector string) string { - requirements := strings.Split(selector, ",") - sort.Strings(requirements) - return strings.Join(requirements, ",") - } - assert.Equal(t, "", restrictions.Labels.String(), "label selector in DeleteCollection") - assert.Equal(t, normalize(fieldsSelector.String()), normalize(restrictions.Fields.String()), "field selector in DeleteCollection") - - // There's only one object that could get matched, so delete it. - // Delete doesn't return an error if already deleted, which is what - // we need here (no error when nothing to delete). - err := fakeClient.Tracker().Delete(resourceapi.SchemeGroupVersion.WithResource("resourceslices"), "", slice.Name) - - // Set expected slice fields for the next call of this reactor. - // The reactor will be called next time when resourceslices object is deleted - // by the kubelet after plugin deregistration. - switch len(expectedSliceFields) { - case 1: - // Startup cleanup done, now expect cleanup for test plugin. - expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": test.driverName} - case 2: - // Test plugin cleanup done, now expect cleanup for the other plugin. - otherPlugin := pluginA - if otherPlugin == test.driverName { - otherPlugin = pluginB - } - expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": otherPlugin} - } - return true, nil, err - }) + fakeClient := getFakeClient(t, nodeName, test.driverName, getSlice("test-slice")) client = fakeClient + tCtx = ktesting.WithClients(tCtx, nil, nil, client, nil, nil) } // The DRAPluginManager wipes all slices at startup. draPlugins := NewDRAPluginManager(tCtx, client, getFakeNode, time.Second /* very short wiping delay for testing */) tCtx.Cleanup(draPlugins.Stop) - requireNoSlices := func() { - t.Helper() - if client == nil { - return - } - require.EventuallyWithT(t, func(t *assert.CollectT) { - slices, err := client.ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{}) - if !assert.NoError(t, err, "list slices") { - return - } - assert.Empty(t, slices.Items, "slices") - }, time.Minute, time.Second) + if test.withClient { + requireNoSlices(tCtx) } - requireNoSlices() // Simulate one existing plugin A. - err := draPlugins.RegisterPlugin(pluginA, endpointA, []string{drapb.DRAPluginService}, nil) + err = draPlugins.RegisterPlugin(pluginA, endpointA, []string{drapb.DRAPluginService}, nil) require.NoError(t, err) t.Cleanup(func() { tCtx.Logf("Removing plugin %s", pluginA) @@ -225,11 +264,94 @@ func TestRegistrationHandler(t *testing.T) { draPlugins.DeRegisterPlugin(test.driverName, test.endpoint) // Nop. draPlugins.DeRegisterPlugin(test.driverName, test.endpoint) - - requireNoSlices() + if test.withClient { + requireNoSlices(tCtx) + } }) - assert.Equal(t, test.endpoint, plugin.endpoint, "plugin endpoint") + // Which plugin was chosen is random in this test: it depends on which plugin was detected as connected, + // which can be both, one, or none at this point. Some attributes are common to both. + assert.Equal(t, test.driverName, plugin.driverName, "DRA driver driver name") assert.Equal(t, test.chosenService, plugin.chosenService, "chosen service") }) } } + +// TestConnectionHandling checks the reaction to state changes of the service connection. +func TestConnectionHandling(t *testing.T) { + t.Parallel() + for description, test := range map[string]struct { + delay time.Duration + requireSliceRemoval bool + }{ + "wipe-slices-on-disconnect": { + delay: time.Second, // very short wiping delay for testing + requireSliceRemoval: true, + }, + "no-wipe-slices-on-reconnect": { + delay: time.Hour, // long delay to avoid wiping while the test runs + requireSliceRemoval: false, + }, + } { + t.Run(description, func(t *testing.T) { + t.Parallel() + tCtx := ktesting.Init(t) + + service := drapb.DRAPluginService + driverName := "test-plugin" + sliceName := "test-slice" + + slice := getSlice(sliceName) + client := getFakeClient(t, nodeName, driverName, slice) + tCtx = ktesting.WithClients(tCtx, nil, nil, client, nil, nil) + + // The handler wipes all slices at startup. + draPlugins := NewDRAPluginManager(tCtx, client, getFakeNode, test.delay) + tCtx.Cleanup(draPlugins.Stop) + requireNoSlices(tCtx) + + // Run GRPC service. + endpoint := path.Join(t.TempDir(), "dra-plugin-test.sock") + teardown, err := setupFakeGRPCServer(service, endpoint) + require.NoError(t, err) + defer teardown() + + err = draPlugins.RegisterPlugin(driverName, endpoint, []string{service}, nil) + require.NoError(t, err) + + plugin := draPlugins.get(driverName) + assert.NotNil(t, plugin, "plugin should be present in the plugin store") + + // Create the slice as if the plugin had done that while it runs. + _, err = client.ResourceV1beta1().ResourceSlices().Create(tCtx, slice, metav1.CreateOptions{}) + require.NoError(t, err, "recreate slice") + + // Stop gRPC server. + tCtx.Log("Stopping plugin gRPC server") + teardown() + + if test.requireSliceRemoval { + // Slice should get removed. + requireNoSlices(tCtx) + } else { + // Start up gRPC server again. + tCtx.Log("Restarting plugin gRPC server") + teardown, err = setupFakeGRPCServer(service, endpoint) + require.NoError(t, err) + defer teardown() + + // There shouldn't be any pending wipes for the plugin. + require.Eventuallyf(t, func() bool { + draPlugins.mutex.Lock() + defer draPlugins.mutex.Unlock() + _, ok := draPlugins.pendingWipes[driverName] + return ok + }, time.Minute, time.Second, "wiping should be stopped for plugin %s", driverName) + + // Slice should still be there + slices, err := client.ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{}) + require.NoError(t, err, "list slices") + assert.Len(t, slices.Items, 1, "slices") + } + }) + } +} diff --git a/pkg/kubelet/pluginmanager/pluginwatcher/README.md b/pkg/kubelet/pluginmanager/pluginwatcher/README.md index 9403829a2fb..9fe366923a1 100644 --- a/pkg/kubelet/pluginmanager/pluginwatcher/README.md +++ b/pkg/kubelet/pluginmanager/pluginwatcher/README.md @@ -22,6 +22,58 @@ should end with a DNS domain that is unique for the plugin. Each time a plugin starts, it has to delete old sockets if they exist and listen anew under the same filename. +## Monitoring Plugin Connection + +**Warning**: Monitoring the plugin connection is only supported +for DRA at the moment. + +The Kubelet monitors the gRPC connection to a plugin's **service socket** using +a [gRPC stats handler](https://github.com/grpc/grpc-go/blob/master/examples/features/stats_monitoring/README.md). + +This enables the Kubelet to: +- Detect when the plugin process has crashed, exited, or restarted +- Trigger cleanup of the plugin’s resources on connection drop +- Cancel pending cleanup if the connection is restored + + +The **registration socket** is used by the plugin manager. It registers the +plugin when registration socket is created by the plugin and the GetInfo gRPC +call succeeds. It deregisters the plugin when the socket is removed. +The plugin should be ready to handle gRPC requests over the **service socket** +that it returned in response to the GetInfo call because the kubelet might try +to use the service immediately. Monitoring this service socket is therefore +more accurate for detecting the real availability of the plugin. + +### How It Works + +Internally, the plugin client configures a gRPC stats handler to observe +`ConnBegin` and `ConnEnd` events on the service socket connection. The +connection is established over a Unix Domain Socket, which provides reliable +semantics - if the connection drops, it definitively indicates that the plugin +closed its end (e.g., due to crash or shutdown). + +1. During plugin registration, the Kubelet connects to plugin's + **service socket** and attaches the stats handler. +2. A long-lived gRPC connection is established and actively monitored. +3. If the plugin process exits and the connection drops, the stats + handler observes a `ConnEnd` event. +4. This triggers a check whether cleanup is necessary: as long as at + least one gRPC connection is connected, the plugin is usable and + no cleanup is required. +5. A gRPC reconnect is initiated immediately after connection loss. +6. When the plugin resumes serving on the same service socket, the connection + is re-established and a `ConnBegin` event is observed. +7. This cancels any in-progress resource cleanup and restores communication. + +### Key Properties + +- The plugin is **not** deregistered when the connection drops. +- This model supports multi-container plugin deployments (e.g., CSI-style + sidecar setups) where the service container may restart independently of + the registrar container. +- Cleanup is only executed if the connection is not restored before a + grace period expires. + ## Seamless Upgrade To avoid downtime of a plugin on a node, it would be nice to support running an diff --git a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go index 97ec6842105..4f3ae832c28 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go @@ -397,6 +397,26 @@ func FlockDirectoryPath(path string) Option { } } +// RegistrationService controls whether the kubelet plugin gRPC service +// is started. It's on by default. This is meant for testing, normal +// DRA drivers should use the default. +func RegistrationService(enabled bool) Option { + return func(o *options) error { + o.registrationService = enabled + return nil + } +} + +// DRAService controls whether the DRA gRPC service +// is started. It's on by default. This is meant for testing, normal +// DRA drivers should use the default. +func DRAService(enabled bool) Option { + return func(o *options) error { + o.draService = enabled + return nil + } +} + type options struct { logger klog.Logger grpcVerbosity int @@ -413,6 +433,8 @@ type options struct { serialize bool flockDirectoryPath string nodeV1beta1 bool + registrationService bool + draService bool } // Helper combines the kubelet registration service and the DRA node plugin @@ -442,8 +464,9 @@ type Helper struct { resourceSliceController *resourceslice.Controller } -// Start sets up two gRPC servers (one for registration, one for the DRA node -// client) and implements them by calling a [DRAPlugin] implementation. +// Start sets up all enabled gRPC servers (by default, one for registration, +// one for the DRA node client) and implements them by calling a [DRAPlugin] +// implementation. // // The context and/or DRAPlugin.Stop can be used to stop all background activity. // Stop also blocks. A logger can be stored in the context to add values or @@ -462,6 +485,8 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe pluginRegistrationEndpoint: endpoint{ dir: KubeletRegistryDir, }, + draService: true, + registrationService: true, } for _, option := range opts { if err := option(&o); err != nil { @@ -530,34 +555,42 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe } }() - // Run the node plugin gRPC server first to ensure that it is ready. var supportedServices []string + if o.nodeV1beta1 { + logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service") + supportedServices = append(supportedServices, drapb.DRAPluginService) + } + if len(supportedServices) == 0 { + return nil, errors.New("no supported DRA gRPC API is implemented and enabled") + } draEndpoint := endpoint{ dir: o.pluginDataDirectoryPath, file: "dra" + uidPart + ".sock", // "dra" is hard-coded. The directory is unique, so we get a unique full path also without the UID. listenFunc: o.draEndpointListen, } - pluginServer, err := startGRPCServer(klog.LoggerWithName(logger, "dra"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, draEndpoint, func(grpcServer *grpc.Server) { - if o.nodeV1beta1 { - logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service") - drapb.RegisterDRAPluginServer(grpcServer, &nodePluginImplementation{Helper: d}) - supportedServices = append(supportedServices, drapb.DRAPluginService) + + if o.draService { + // Run the node plugin gRPC server first to ensure that it is ready. + pluginServer, err := startGRPCServer(klog.LoggerWithName(logger, "dra"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, draEndpoint, func(grpcServer *grpc.Server) { + if o.nodeV1beta1 { + logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service") + drapb.RegisterDRAPluginServer(grpcServer, &nodePluginImplementation{Helper: d}) + } + }) + if err != nil { + return nil, fmt.Errorf("start DRA service: %w", err) } - }) - if err != nil { - return nil, fmt.Errorf("start node client: %v", err) - } - d.pluginServer = pluginServer - if len(supportedServices) == 0 { - return nil, errors.New("no supported DRA gRPC API is implemented and enabled") + d.pluginServer = pluginServer } - // Now make it available to kubelet. - registrar, err := startRegistrar(klog.LoggerWithName(logger, "registrar"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, draEndpoint.path(), o.pluginRegistrationEndpoint) - if err != nil { - return nil, fmt.Errorf("start registrar: %v", err) + if o.registrationService { + // Now make it available to kubelet. + registrar, err := startRegistrar(klog.LoggerWithName(logger, "registrar"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, draEndpoint.path(), o.pluginRegistrationEndpoint) + if err != nil { + return nil, fmt.Errorf("start registrar: %w", err) + } + d.registrar = registrar } - d.registrar = registrar // startGRPCServer and startRegistrar don't implement cancellation // themselves, we add that for both here. diff --git a/test/e2e_node/dra_test.go b/test/e2e_node/dra_test.go index 09ade51d071..57a59dc1c6b 100644 --- a/test/e2e_node/dra_test.go +++ b/test/e2e_node/dra_test.go @@ -378,6 +378,62 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami return kubeletPlugin.CountCalls("/NodePrepareResources") }).WithTimeout(retryTestTimeout).Should(gomega.Equal(calls)) }) + + ginkgo.It("must be functional when plugin starts to listen on a service socket after registration", func(ctx context.Context) { + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, getNodeName(ctx, f), driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + draService := newDRAService(ctx, f.ClientSet, getNodeName(ctx, f), driverName) + + pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drapod", false, []string{driverName}) + + ginkgo.By("wait for NodePrepareResources call to succeed") + gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded) + + ginkgo.By("wait for pod to succeed") + err := e2epod.WaitForPodSuccessInNamespace(ctx, f.ClientSet, pod.Name, f.Namespace.Name) + framework.ExpectNoError(err) + }) + + ginkgo.It("must be functional after reconnect", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + draService := newDRAService(ctx, f.ClientSet, nodeName, driverName) + + pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drasleeppod" /* enables sleeping */, false /* pod is deleted below */, []string{driverName}) + + ginkgo.By("wait for NodePrepareResources call to succeed") + gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded) + + ginkgo.By("stop plugin") + draService.Stop() + + ginkgo.By("waiting for pod to run") + err := e2epod.WaitForPodRunningInNamespace(ctx, f.ClientSet, pod) + framework.ExpectNoError(err) + + ginkgo.By("wait for ResourceSlice removal, indicating detection of disconnect") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin") + + ginkgo.By("restarting plugin") + draService = newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("stopping pod") + err = f.ClientSet.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{}) + framework.ExpectNoError(err) + gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodeUnprepareResourcesSucceeded) + }) }) f.Context("Two resource Kubelet Plugins", f.WithSerial(), func() { @@ -513,18 +569,6 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami }) f.Context("ResourceSlice", f.WithSerial(), func() { - listResources := func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { - slices, err := f.ClientSet.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{}) - if err != nil { - return nil, err - } - return slices.Items, nil - } - - matchResourcesByNodeName := func(nodeName string) types.GomegaMatcher { - return gomega.HaveField("Spec.NodeName", gomega.Equal(nodeName)) - } - f.It("must be removed on kubelet startup", f.WithDisruptive(), func(ctx context.Context) { ginkgo.By("stop kubelet") restartKubelet := mustStopKubelet(ctx, f) @@ -543,15 +587,15 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami matchAll := gomega.ConsistOf(matchResourcesByNodeName(nodeName), matchResourcesByNodeName(otherNodeName)) matchOtherNode := gomega.ConsistOf(matchResourcesByNodeName(otherNodeName)) - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchAll, "ResourceSlices without kubelet") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchAll, "ResourceSlices without kubelet") ginkgo.By("restart kubelet") restartKubelet(ctx) restartKubelet = nil ginkgo.By("wait for exactly the node's ResourceSlice to get deleted") - gomega.Eventually(ctx, listResources).Should(matchOtherNode, "ResourceSlices with kubelet") - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchOtherNode, "ResourceSlices with kubelet") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchOtherNode, "ResourceSlices with kubelet") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchOtherNode, "ResourceSlices with kubelet") }) f.It("must be removed after plugin unregistration", func(ctx context.Context) { @@ -560,13 +604,89 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami ginkgo.By("start plugin and wait for ResourceSlice") kubeletPlugin := newKubeletPlugin(ctx, f.ClientSet, getNodeName(ctx, f), driverName) - gomega.Eventually(ctx, listResources).Should(matchNode, "ResourceSlice from kubelet plugin") - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlice from kubelet plugin") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlice from kubelet plugin") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlice from kubelet plugin") ginkgo.By("stop plugin and wait for ResourceSlice removal") kubeletPlugin.Stop() - gomega.Eventually(ctx, listResources).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") + }) + + f.It("must be removed if plugin stops after registration", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for ResourceSlice to be created by plugin") + matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName)) + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlices") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlices") + + ginkgo.By("stop plugin") + kubeletPlugin.Stop() + + ginkgo.By("wait for ResourceSlice removal") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices") + }) + + f.It("must be removed if plugin is unresponsive after registration", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("create a ResourceSlice") + createTestResourceSlice(ctx, f.ClientSet, nodeName, driverName) + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.ConsistOf(matchResourcesByNodeName(nodeName)), "ResourceSlices without plugin") + + ginkgo.By("wait for ResourceSlice removal") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices without plugin") + }) + + f.It("must not be removed if plugin restarts quickly enough", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for ResourceSlice to be created by plugin") + matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName)) + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlices") + var slices []resourceapi.ResourceSlice + gomega.Consistently(ctx, listAndStoreResources(f.ClientSet, &slices)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlices") + + ginkgo.By("stop plugin") + kubeletPlugin.Stop() + + // We know from the "must be removed if plugin is unresponsive after registration" that the kubelet + // eventually notices the dropped connection. We cannot observe when that happens, we would need + // a new metric for that ("registered DRA plugins"). Let's give it a few seconds, which is significantly + // less than the wiping delay. + time.Sleep(5 * time.Second) + + ginkgo.By("restarting plugin") + kubeletPlugin = newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("ensuring unchanged ResourceSlices") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(time.Minute).Should(gomega.Equal(slices), "ResourceSlices") }) }) }) @@ -622,10 +742,72 @@ func newKubeletPlugin(ctx context.Context, clientSet kubernetes.Interface, nodeN return plugin } +// newRegistrar starts a registrar for the specified DRA driver, without the DRA gRPC service. +func newRegistrar(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin { + ginkgo.By("start only Kubelet plugin registrar") + logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin registrar "+driverName)) + ctx = klog.NewContext(ctx, logger) + registrar, err := testdriver.StartPlugin(ctx, cdiDir, driverName, clientSet, nodeName, testdriver.FileOperations{}, kubeletplugin.DRAService(false)) + framework.ExpectNoError(err, "start only Kubelet plugin registrar") + return registrar +} + +// newDRAService starts the DRA gRPC service for the specified DRA driver, without the registrar. +func newDRAService(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin { + ginkgo.By("start only Kubelet plugin") + logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin "+driverName), "node", nodeName) + ctx = klog.NewContext(ctx, logger) + + // Ensure that directories exist, creating them if necessary. We want + // to know early if there is a setup problem that would prevent + // creating those directories. + err := os.MkdirAll(cdiDir, os.FileMode(0750)) + framework.ExpectNoError(err, "create CDI directory") + datadir := path.Join(kubeletplugin.KubeletPluginsDir, driverName) // The default, not set below. + err = os.MkdirAll(datadir, 0750) + framework.ExpectNoError(err, "create DRA socket directory") + + plugin, err := testdriver.StartPlugin( + ctx, + cdiDir, + driverName, + clientSet, + nodeName, + testdriver.FileOperations{ + DriverResources: &resourceslice.DriverResources{ + Pools: map[string]resourceslice.Pool{ + nodeName: { + Slices: []resourceslice.Slice{{ + Devices: []resourceapiv1beta2.Device{ + { + Name: "device-00", + }, + }, + }}, + }, + }, + }, + }, + kubeletplugin.RegistrationService(false), + ) + framework.ExpectNoError(err) + + ginkgo.DeferCleanup(func(ctx context.Context) { + // kubelet should do this eventually, but better make sure. + // A separate test checks this explicitly. + framework.ExpectNoError(clientSet.ResourceV1beta1().ResourceSlices().DeleteCollection(ctx, metav1.DeleteOptions{}, metav1.ListOptions{FieldSelector: resourceapi.ResourceSliceSelectorDriver + "=" + driverName})) + }) + ginkgo.DeferCleanup(plugin.Stop) + + return plugin +} + // createTestObjects creates objects required by the test // NOTE: as scheduler and controller manager are not running by the Node e2e, // the objects must contain all required data to be processed correctly by the API server -// and placed on the node without involving the scheduler and the DRA controller +// and placed on the node without involving the scheduler and the DRA controller. +// +// Instead adding more parameters, the podName determines what the pod does. func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, nodename, namespace, className, claimName, podName string, deferPodDeletion bool, driverNames []string) *v1.Pod { // DeviceClass class := &resourceapi.DeviceClass{ @@ -699,6 +881,11 @@ func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, node RestartPolicy: v1.RestartPolicyNever, }, } + switch podName { + case "drasleeppod": + // As above, plus infinite sleep. + pod.Spec.Containers[0].Command[2] += "&& sleep 100000" + } createdPod, err := clientSet.CoreV1().Pods(namespace).Create(ctx, pod, metav1.CreateOptions{}) framework.ExpectNoError(err) @@ -772,3 +959,28 @@ func createTestResourceSlice(ctx context.Context, clientSet kubernetes.Interface } }) } + +func listResources(client kubernetes.Interface) func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + return func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + slices, err := client.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, err + } + return slices.Items, nil + } +} + +func listAndStoreResources(client kubernetes.Interface, lastSlices *[]resourceapi.ResourceSlice) func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + return func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + slices, err := client.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, err + } + *lastSlices = slices.Items + return *lastSlices, nil + } +} + +func matchResourcesByNodeName(nodeName string) types.GomegaMatcher { + return gomega.HaveField("Spec.NodeName", gomega.Equal(nodeName)) +} From 165bb1da58e75854345b4d60d433369030230bc7 Mon Sep 17 00:00:00 2001 From: Patrick Ohly Date: Wed, 4 Jun 2025 10:26:42 +0200 Subject: [PATCH 3/9] DRA kubelet: use TimedWorkersQueue Conceptually TimedWorkersQueue is similar to the current code: it spawns goroutines and cancels them. Using it makes the code a bit shorter, even though the TimedWorkersQueue API could be a bit nicer and more consistent (key string vs. WorkerArgs as parameters). Depending on the tainteviction package is a bit odd, which is the reason why TimedWorkersQueue wasn't already used earlier. But there don't seem to be other implementations of this common problem. https://pkg.go.dev/k8s.io/client-go/util/workqueue#TypedDelayingInterface doesn't work because queue entries cannot be removed. This doesn't really solve the problem of tracking goroutines for wiping because TimedWorkersQueue doesn't support that. But not tracking is arguably better than doing it wrong and this only affects unit tests, so it should be okay. --- .../cm/dra/plugin/dra_plugin_manager.go | 128 ++++++++---------- .../cm/dra/plugin/registration_test.go | 11 +- 2 files changed, 62 insertions(+), 77 deletions(-) diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go index 50eedce9b1a..ae5cc972bc9 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go @@ -38,6 +38,7 @@ import ( "k8s.io/klog/v2" drapbv1alpha4 "k8s.io/kubelet/pkg/apis/dra/v1alpha4" drapbv1beta1 "k8s.io/kubelet/pkg/apis/dra/v1beta1" + timedworkers "k8s.io/kubernetes/pkg/controller/tainteviction" // TODO (?): move this common helper somewhere else? "k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache" "k8s.io/utils/ptr" ) @@ -60,30 +61,20 @@ type DRAPluginManager struct { getNode func() (*v1.Node, error) wipingDelay time.Duration - // TODO: replace pendingWipes with some kind of workqueue. - // As it stands, WaitGroup suffers from a data race: - // - Queueing a new wiping creates a goroutine and adds to - // to wg. - // - Concurrently, wg.Wait reads from it. - // - // This is not allowed, all wg.Adds must come before wg.Wait. - // - // This race can be triggered with - // go test -count=10 -race ./... wg sync.WaitGroup mutex sync.RWMutex // driver name -> DRAPlugin in the order in which they got added store map[string][]*monitoredPlugin - // pendingWipes maps a driver name to a cancel function for - // wiping of that plugin's ResourceSlices. Entries get added - // in DeRegisterPlugin and check in RegisterPlugin. If - // wiping is pending during RegisterPlugin, it gets canceled. + // pendingWipes tracks at which time ResourceSlices for a + // DRA driver should be removed. The removal then happens in + // the background in a callback function that is invoked + // by the TimedWorkerQueue. // - // Must use pointers to functions because the entries have to - // be comparable. - pendingWipes map[string]*context.CancelCauseFunc + // TimedWorkerQueue uses namespace/name as key. We use + // the driver name as name with no namespace. + pendingWipes *timedworkers.TimedWorkerQueue } var _ cache.PluginHandler = &DRAPluginManager{} @@ -127,7 +118,10 @@ func (m *monitoredPlugin) HandleConn(_ context.Context, stats grpcstats.ConnStat default: return } - + if m.pm.backgroundCtx.Err() != nil { + // Shutting down, no longer interested in connection changes... + return + } logger := klog.FromContext(m.pm.backgroundCtx) m.pm.mutex.Lock() defer m.pm.mutex.Unlock() @@ -150,8 +144,11 @@ func NewDRAPluginManager(ctx context.Context, kubeClient kubernetes.Interface, g kubeClient: kubeClient, getNode: getNode, wipingDelay: wipingDelay, - pendingWipes: make(map[string]*context.CancelCauseFunc), } + pm.pendingWipes = timedworkers.CreateWorkerQueue(func(ctx context.Context, fireAt time.Time, args *timedworkers.WorkArgs) error { + pm.wipeResourceSlices(ctx, args.Object.Name) + return nil + }) // When kubelet starts up, no DRA driver has registered yet. None of // the drivers are usable until they come back, which might not happen @@ -167,51 +164,49 @@ func NewDRAPluginManager(ctx context.Context, kubeClient kubernetes.Interface, g ctx := pm.backgroundCtx logger := klog.LoggerWithName(klog.FromContext(ctx), "startup") ctx = klog.NewContext(ctx, logger) - pm.wipeResourceSlices(ctx, 0 /* no delay */, "" /* all drivers */) + pm.wipeResourceSlices(ctx, "" /* all drivers */) }() return pm } -// Stop cancels any remaining background activities and blocks until all goroutines have stopped. +// Stop cancels any remaining background activities and blocks until all goroutines have stopped, +// with one caveat: goroutines created dynamically for wiping ResourceSlices are not tracked. +// They won't do anything because of the context cancellation. func (pm *DRAPluginManager) Stop() { defer pm.wg.Wait() // Must run after unlocking our mutex. pm.mutex.Lock() defer pm.mutex.Unlock() + logger := klog.FromContext(pm.backgroundCtx) pm.cancel(errors.New("Stop was called")) // Close all connections, otherwise gRPC keeps doing things in the background. - for _, plugins := range pm.store { + // Also cancel all pending wiping. + for driverName, plugins := range pm.store { + workerArg := timedworkers.NewWorkArgs(driverName, "") + pm.pendingWipes.CancelWork(logger, workerArg.KeyFromWorkArgs()) for _, plugin := range plugins { if err := plugin.conn.Close(); err != nil { - klog.FromContext(pm.backgroundCtx).Error(err, "Closing gRPC connection", "driverName", plugin.driverName, "endpoint", plugin.endpoint) + logger.Error(err, "Closing gRPC connection", "driverName", plugin.driverName, "endpoint", plugin.endpoint) } } } } // wipeResourceSlices deletes ResourceSlices of the node, optionally just for a specific driver. -// Wiping will delay for a while and can be canceled by canceling the context. -func (pm *DRAPluginManager) wipeResourceSlices(ctx context.Context, delay time.Duration, driver string) { +// +// It gets called in a stand-alone goroutine at kubelet startup and as callback +// of a TimedWorkersQueue. In both cases the caller has no way of handling errors, +// so wipeResourceSlices must implement it's own retry mechanism. +// +// Can be canceled by canceling the context. +func (pm *DRAPluginManager) wipeResourceSlices(ctx context.Context, driver string) { if pm.kubeClient == nil { return } logger := klog.FromContext(ctx) - if delay != 0 { - // Before we start deleting, give the driver time to bounce back. - // Perhaps it got removed as part of a DaemonSet update and the - // replacement pod is about to start. - logger.V(4).Info("Starting to wait before wiping ResourceSlices", "delay", delay) - select { - case <-ctx.Done(): - logger.V(4).Info("Aborting wiping of ResourceSlices", "reason", context.Cause(ctx)) - case <-time.After(delay): - logger.V(4).Info("Starting to wipe ResourceSlices after waiting", "delay", delay) - } - } - backoff := wait.Backoff{ Duration: time.Second, Factor: 2, @@ -417,52 +412,39 @@ func (pm *DRAPluginManager) remove(driverName, endpoint string) { // sync must be called each time the information about a plugin changes. // The mutex must be locked for writing. func (pm *DRAPluginManager) sync(driverName string) { + if pm.kubeClient == nil { + // Cannot wipe. + return + } ctx := pm.backgroundCtx - logger := klog.FromContext(ctx) + logger := klog.FromContext(pm.backgroundCtx) + workerArgs := timedworkers.NewWorkArgs(driverName, "") // Is the DRA driver usable again? if pm.usable(driverName) { // Yes: cancel any pending ResourceSlice wiping for the DRA driver. - if cancel := pm.pendingWipes[driverName]; cancel != nil { - (*cancel)(errors.New("new plugin instance registered")) - delete(pm.pendingWipes, driverName) - } + pm.pendingWipes.CancelWork(logger, workerArgs.KeyFromWorkArgs()) return } - // No: prepare for canceling the background wiping. This needs to run - // in the context of the DRAPluginManager. + // No: ensure that we wipe ResourceSlices of the driver. + // If this was already queued earlier, the original timeout + // continues to apply because nothing changed. + if pm.pendingWipes.GetWorkerUnsafe(workerArgs.KeyFromWorkArgs()) != nil { + // Already queued or potentially already running. + // + // There's a small time-of-check-time-of-use race here, + // but that's fine: if wiping starts after we retrieve + // the pointer and before checking it, the work gets + // done, which is what we want. + return + } + now := time.Now() + fireAt := now.Add(pm.wipingDelay) logger = klog.LoggerWithName(logger, "driver-cleanup") logger = klog.LoggerWithValues(logger, "driverName", driverName) - ctx, cancel := context.WithCancelCause(pm.backgroundCtx) ctx = klog.NewContext(ctx, logger) - - // Clean up the ResourceSlices for the deleted DRAPlugin since it - // may have died without doing so itself and might never come - // back. - // - // May get canceled if the plugin comes back quickly enough. - if cancel := pm.pendingWipes[driverName]; cancel != nil { - (*cancel)(errors.New("plugin deregistered a second time")) - } - pm.pendingWipes[driverName] = &cancel - - pm.wg.Add(1) - go func() { - defer pm.wg.Done() - defer func() { - pm.mutex.Lock() - defer pm.mutex.Unlock() - - // Cancel our own context, but remove it from the map only if it - // is the current entry. Perhaps it already got replaced. - cancel(errors.New("wiping done")) - if pm.pendingWipes[driverName] == &cancel { - delete(pm.pendingWipes, driverName) - } - }() - pm.wipeResourceSlices(ctx, pm.wipingDelay, driverName) - }() + pm.pendingWipes.AddWork(ctx, timedworkers.NewWorkArgs(driverName, ""), now, fireAt) } // usable returns true if at least one endpoint is ready to handle gRPC calls for the DRA driver. diff --git a/pkg/kubelet/cm/dra/plugin/registration_test.go b/pkg/kubelet/cm/dra/plugin/registration_test.go index 6cd505df934..6c9e06dfd03 100644 --- a/pkg/kubelet/cm/dra/plugin/registration_test.go +++ b/pkg/kubelet/cm/dra/plugin/registration_test.go @@ -37,6 +37,7 @@ import ( cgotesting "k8s.io/client-go/testing" drapbv1alpha4 "k8s.io/kubelet/pkg/apis/dra/v1alpha4" drapb "k8s.io/kubelet/pkg/apis/dra/v1beta1" + timedworkers "k8s.io/kubernetes/pkg/controller/tainteviction" "k8s.io/kubernetes/test/utils/ktesting" ) @@ -333,6 +334,11 @@ func TestConnectionHandling(t *testing.T) { // Slice should get removed. requireNoSlices(tCtx) } else { + wipingIsPending := func() bool { + return draPlugins.pendingWipes.GetWorkerUnsafe(timedworkers.NewWorkArgs(driverName, "").KeyFromWorkArgs()) != nil + } + require.Eventuallyf(t, wipingIsPending, time.Minute, time.Second, "wiping should be queued for plugin %s", driverName) + // Start up gRPC server again. tCtx.Log("Restarting plugin gRPC server") teardown, err = setupFakeGRPCServer(service, endpoint) @@ -341,10 +347,7 @@ func TestConnectionHandling(t *testing.T) { // There shouldn't be any pending wipes for the plugin. require.Eventuallyf(t, func() bool { - draPlugins.mutex.Lock() - defer draPlugins.mutex.Unlock() - _, ok := draPlugins.pendingWipes[driverName] - return ok + return !wipingIsPending() }, time.Minute, time.Second, "wiping should be stopped for plugin %s", driverName) // Slice should still be there From c90c2e0d402d7522c563fc980750fceb551576f2 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Mon, 16 Jun 2025 14:41:54 +0300 Subject: [PATCH 4/9] kubelet: DRA: fix linter warnings Fixed the following warnings: dra_test.go:884:2: singleCaseSwitch: should rewrite switch statement to if statement (gocritic) switch podName { ^ dra_test.go:686:4: SA4006: this value of kubeletPlugin is never used (staticcheck) kubeletPlugin = newDRAService(ctx, f.ClientSet, nodeName, driverName) ^ --- test/e2e_node/dra_test.go | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/test/e2e_node/dra_test.go b/test/e2e_node/dra_test.go index 57a59dc1c6b..4b258a9cef7 100644 --- a/test/e2e_node/dra_test.go +++ b/test/e2e_node/dra_test.go @@ -683,7 +683,7 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami time.Sleep(5 * time.Second) ginkgo.By("restarting plugin") - kubeletPlugin = newDRAService(ctx, f.ClientSet, nodeName, driverName) + newDRAService(ctx, f.ClientSet, nodeName, driverName) ginkgo.By("ensuring unchanged ResourceSlices") gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(time.Minute).Should(gomega.Equal(slices), "ResourceSlices") @@ -881,8 +881,7 @@ func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, node RestartPolicy: v1.RestartPolicyNever, }, } - switch podName { - case "drasleeppod": + if podName == "drasleeppod" { // As above, plus infinite sleep. pod.Spec.Containers[0].Command[2] += "&& sleep 100000" } From 3ae99f2547190bc913ab164b062fcc94fd6c9d83 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Sat, 21 Jun 2025 11:50:31 +0300 Subject: [PATCH 5/9] kubelet: DRA: fix test failure on Windows Unix socket path has to be <= 108 characters in length on Windows. Shortened DRA socket path for TestConnectionHandling and TestRegistrationHandler tests should fix the test run on Windows. --- pkg/kubelet/cm/dra/plugin/registration_test.go | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/pkg/kubelet/cm/dra/plugin/registration_test.go b/pkg/kubelet/cm/dra/plugin/registration_test.go index 6c9e06dfd03..4c8919937b2 100644 --- a/pkg/kubelet/cm/dra/plugin/registration_test.go +++ b/pkg/kubelet/cm/dra/plugin/registration_test.go @@ -123,8 +123,8 @@ func TestRegistrationHandler(t *testing.T) { } tmp := t.TempDir() - endpointA := path.Join(tmp, "dra-plugin-a.sock") - endpointB := path.Join(tmp, "dra-plugin-b.sock") + endpointA := path.Join(tmp, "a.sock") + endpointB := path.Join(tmp, "b.sock") for _, test := range []struct { description string @@ -136,7 +136,7 @@ func TestRegistrationHandler(t *testing.T) { chosenService string }{ { - description: "no-services-provided", + description: "no-services", driverName: pluginB, endpoint: endpointB, shouldError: true, @@ -284,11 +284,11 @@ func TestConnectionHandling(t *testing.T) { delay time.Duration requireSliceRemoval bool }{ - "wipe-slices-on-disconnect": { + "wipe-on-disconnect": { delay: time.Second, // very short wiping delay for testing requireSliceRemoval: true, }, - "no-wipe-slices-on-reconnect": { + "no-wipe-on-reconnect": { delay: time.Hour, // long delay to avoid wiping while the test runs requireSliceRemoval: false, }, @@ -311,7 +311,7 @@ func TestConnectionHandling(t *testing.T) { requireNoSlices(tCtx) // Run GRPC service. - endpoint := path.Join(t.TempDir(), "dra-plugin-test.sock") + endpoint := path.Join(t.TempDir(), "dra.sock") teardown, err := setupFakeGRPCServer(service, endpoint) require.NoError(t, err) defer teardown() From 7f6389e7709decee25a6b82f886792f7b9ca6ce3 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Wed, 18 Jun 2025 12:06:43 +0300 Subject: [PATCH 6/9] e2e_node: DRA: pass socket path as a parameter Added an ability to specify the socket path for the DRA gRPC service in the e2e node tests. The PluginSocket option is added to allow setting the name of the socket inside the directory where the DRA driver creates the socket for the DRA gRPC calls. This is used by the kubelet to connect to the DRA plugin. The newDRAService and newRegistrar functions are updated to accept a socketPath parameter, which is used to configure the PluginDataDirectoryPath and PluginSocket options for the DRA plugin. This change enables more flexible configuration of the DRA plugin in e2e tests, allowing for testing with different socket paths. --- .../kubeletplugin/draplugin.go | 21 ++++++- test/e2e_node/dra_test.go | 62 ++++++++++++++----- 2 files changed, 65 insertions(+), 18 deletions(-) diff --git a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go index 4f3ae832c28..06bde0ae3fa 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go @@ -261,6 +261,19 @@ func PluginDataDirectoryPath(path string) Option { } } +// PluginSocket sets the name of the socket inside the directory where +// the DRA driver creates the socket for the DRA gRPC calls. This is used +// by the kubelet to connect to the DRA plugin. +// +// This is meant for testing. Normal DRA drivers should not use this and +// instead rely on the automatic handling of the name. +func PluginSocket(name string) Option { + return func(o *options) error { + o.pluginSocket = name + return nil + } +} + // PluginListener configures how to create the registrar socket. // The default is to remove the file if it exists and to then // create a socket. @@ -424,7 +437,8 @@ type options struct { nodeName string nodeUID types.UID pluginRegistrationEndpoint endpoint - pluginDataDirectoryPath string + pluginDataDirectoryPath string // The directory where the plugin socket is created. + pluginSocket string // The socket name for the DRA gRPC service. rollingUpdateUID types.UID draEndpointListen func(ctx context.Context, path string) (net.Listener, error) unaryInterceptors []grpc.UnaryServerInterceptor @@ -513,6 +527,9 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe if o.pluginDataDirectoryPath == "" { o.pluginDataDirectoryPath = path.Join(KubeletPluginsDir, o.driverName) } + if o.pluginSocket == "" { + o.pluginSocket = "dra" + uidPart + ".sock" // "dra" is hard-coded. The directory is unique, so we get a unique full path also without the UID. + } d := &Helper{ driverName: o.driverName, @@ -565,7 +582,7 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe } draEndpoint := endpoint{ dir: o.pluginDataDirectoryPath, - file: "dra" + uidPart + ".sock", // "dra" is hard-coded. The directory is unique, so we get a unique full path also without the UID. + file: o.pluginSocket, listenFunc: o.draEndpointListen, } diff --git a/test/e2e_node/dra_test.go b/test/e2e_node/dra_test.go index 4b258a9cef7..f754511218d 100644 --- a/test/e2e_node/dra_test.go +++ b/test/e2e_node/dra_test.go @@ -381,13 +381,13 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami ginkgo.It("must be functional when plugin starts to listen on a service socket after registration", func(ctx context.Context) { ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, getNodeName(ctx, f), driverName) + registrar := newRegistrar(ctx, f.ClientSet, getNodeName(ctx, f), driverName, "") ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) ginkgo.By("start DRA plugin service") - draService := newDRAService(ctx, f.ClientSet, getNodeName(ctx, f), driverName) + draService := newDRAService(ctx, f.ClientSet, getNodeName(ctx, f), driverName, "") pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drapod", false, []string{driverName}) @@ -403,13 +403,13 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami nodeName := getNodeName(ctx, f) ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) ginkgo.By("start DRA plugin service") - draService := newDRAService(ctx, f.ClientSet, nodeName, driverName) + draService := newDRAService(ctx, f.ClientSet, nodeName, driverName, "") pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drasleeppod" /* enables sleeping */, false /* pod is deleted below */, []string{driverName}) @@ -427,7 +427,7 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin") ginkgo.By("restarting plugin") - draService = newDRAService(ctx, f.ClientSet, nodeName, driverName) + draService = newDRAService(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("stopping pod") err = f.ClientSet.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{}) @@ -617,13 +617,13 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami nodeName := getNodeName(ctx, f) ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) ginkgo.By("start DRA plugin service") - kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName) + kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("wait for ResourceSlice to be created by plugin") matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName)) @@ -642,7 +642,7 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami nodeName := getNodeName(ctx, f) ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) @@ -659,13 +659,13 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami nodeName := getNodeName(ctx, f) ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) ginkgo.By("start DRA plugin service") - kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName) + kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("wait for ResourceSlice to be created by plugin") matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName)) @@ -683,7 +683,7 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami time.Sleep(5 * time.Second) ginkgo.By("restarting plugin") - newDRAService(ctx, f.ClientSet, nodeName, driverName) + newDRAService(ctx, f.ClientSet, nodeName, driverName, "") ginkgo.By("ensuring unchanged ResourceSlices") gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(time.Minute).Should(gomega.Equal(slices), "ResourceSlices") @@ -743,17 +743,33 @@ func newKubeletPlugin(ctx context.Context, clientSet kubernetes.Interface, nodeN } // newRegistrar starts a registrar for the specified DRA driver, without the DRA gRPC service. -func newRegistrar(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin { +func newRegistrar(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName, serviceSocketPath string) *testdriver.ExamplePlugin { ginkgo.By("start only Kubelet plugin registrar") logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin registrar "+driverName)) ctx = klog.NewContext(ctx, logger) - registrar, err := testdriver.StartPlugin(ctx, cdiDir, driverName, clientSet, nodeName, testdriver.FileOperations{}, kubeletplugin.DRAService(false)) + opts := []kubeletplugin.Option{ + kubeletplugin.DRAService(false), + } + if serviceSocketPath != "" { + dir, file := path.Split(serviceSocketPath) + opts = append(opts, kubeletplugin.PluginDataDirectoryPath(dir)) + opts = append(opts, kubeletplugin.PluginSocket(file)) + } + registrar, err := testdriver.StartPlugin( + ctx, + cdiDir, + driverName, + clientSet, + nodeName, + testdriver.FileOperations{}, + opts..., + ) framework.ExpectNoError(err, "start only Kubelet plugin registrar") return registrar } // newDRAService starts the DRA gRPC service for the specified DRA driver, without the registrar. -func newDRAService(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin { +func newDRAService(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName, socketPath string) *testdriver.ExamplePlugin { ginkgo.By("start only Kubelet plugin") logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin "+driverName), "node", nodeName) ctx = klog.NewContext(ctx, logger) @@ -763,7 +779,21 @@ func newDRAService(ctx context.Context, clientSet kubernetes.Interface, nodeName // creating those directories. err := os.MkdirAll(cdiDir, os.FileMode(0750)) framework.ExpectNoError(err, "create CDI directory") - datadir := path.Join(kubeletplugin.KubeletPluginsDir, driverName) // The default, not set below. + opts := []kubeletplugin.Option{ + kubeletplugin.RegistrationService(false), + } + var datadir string + if socketPath == "" { + // The default, not set as option. + datadir = path.Join(kubeletplugin.KubeletPluginsDir, driverName) + } else { + dir, file := path.Split(socketPath) + opts = append(opts, + kubeletplugin.PluginDataDirectoryPath(dir), + kubeletplugin.PluginSocket(file), + ) + datadir = dir + } err = os.MkdirAll(datadir, 0750) framework.ExpectNoError(err, "create DRA socket directory") @@ -788,7 +818,7 @@ func newDRAService(ctx context.Context, clientSet kubernetes.Interface, nodeName }, }, }, - kubeletplugin.RegistrationService(false), + opts..., ) framework.ExpectNoError(err) From cf544da6f7a1b3c160a7d34e9a23d35cff6a91a7 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Wed, 18 Jun 2025 12:10:03 +0300 Subject: [PATCH 7/9] e2e_node: DRA: add tests for different socket setups Added tests to verify DRA functionality with 2 different socket configurations: - the same socket is used for the registration and the DRA service - 2 separate sockets are used for the registration and the DRA service Used table-driven ginkgo to avoid code duplication: specs https://onsi.github.io/ginkgo/#table-driven-tests This change enhances the robustness of the DRA e2e tests by validating its behavior with different socket setups. --- test/e2e_node/dra_test.go | 54 +++++++++++++++++++++++++++------------ 1 file changed, 38 insertions(+), 16 deletions(-) diff --git a/test/e2e_node/dra_test.go b/test/e2e_node/dra_test.go index f754511218d..74177a9a028 100644 --- a/test/e2e_node/dra_test.go +++ b/test/e2e_node/dra_test.go @@ -379,17 +379,19 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami }).WithTimeout(retryTestTimeout).Should(gomega.Equal(calls)) }) - ginkgo.It("must be functional when plugin starts to listen on a service socket after registration", func(ctx context.Context) { + functionalListenAfterRegistration := func(ctx context.Context, socketPath string) { + nodeName := getNodeName(ctx, f) + ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, getNodeName(ctx, f), driverName, "") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, socketPath) ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) ginkgo.By("start DRA plugin service") - draService := newDRAService(ctx, f.ClientSet, getNodeName(ctx, f), driverName, "") + draService := newDRAService(ctx, f.ClientSet, nodeName, driverName, socketPath) - pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drapod", false, []string{driverName}) + pod := createTestObjects(ctx, f.ClientSet, nodeName, f.Namespace.Name, "draclass", "external-claim", "drapod", false, []string{driverName}) ginkgo.By("wait for NodePrepareResources call to succeed") gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded) @@ -397,19 +399,24 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami ginkgo.By("wait for pod to succeed") err := e2epod.WaitForPodSuccessInNamespace(ctx, f.ClientSet, pod.Name, f.Namespace.Name) framework.ExpectNoError(err) - }) + } + ginkgo.DescribeTable("must be functional when plugin starts to listen on a service socket after registration", + functionalListenAfterRegistration, + ginkgo.Entry("2 sockets", ""), + ginkgo.Entry("1 common socket", path.Join(kubeletplugin.KubeletRegistryDir, driverName+"-common.sock")), + ) - ginkgo.It("must be functional after reconnect", func(ctx context.Context) { + functionalAfterServiceReconnect := func(ctx context.Context, socketPath string) { nodeName := getNodeName(ctx, f) ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, "") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, socketPath) ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) ginkgo.By("start DRA plugin service") - draService := newDRAService(ctx, f.ClientSet, nodeName, driverName, "") + draService := newDRAService(ctx, f.ClientSet, nodeName, driverName, socketPath) pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drasleeppod" /* enables sleeping */, false /* pod is deleted below */, []string{driverName}) @@ -427,13 +434,18 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin") ginkgo.By("restarting plugin") - draService = newDRAService(ctx, f.ClientSet, nodeName, driverName, "") + draService = newDRAService(ctx, f.ClientSet, nodeName, driverName, socketPath) ginkgo.By("stopping pod") err = f.ClientSet.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{}) framework.ExpectNoError(err) gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodeUnprepareResourcesSucceeded) - }) + } + ginkgo.DescribeTable("must be functional after service reconnect", + functionalAfterServiceReconnect, + ginkgo.Entry("2 sockets", ""), + ginkgo.Entry("1 common socket", path.Join(kubeletplugin.KubeletRegistryDir, driverName+"-common.sock")), + ) }) f.Context("Two resource Kubelet Plugins", f.WithSerial(), func() { @@ -613,17 +625,17 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") }) - f.It("must be removed if plugin stops after registration", func(ctx context.Context) { + removedIfPluginStopsAfterRegistration := func(ctx context.Context, socketPath string) { nodeName := getNodeName(ctx, f) ginkgo.By("start DRA registrar") - registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, "") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName, socketPath) ginkgo.By("wait for registration to complete") gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) ginkgo.By("start DRA plugin service") - kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName, "") + kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName, socketPath) ginkgo.By("wait for ResourceSlice to be created by plugin") matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName)) @@ -636,7 +648,12 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami ginkgo.By("wait for ResourceSlice removal") gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices") gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices") - }) + } + ginkgo.DescribeTable("must be removed if plugin stops after registration", + removedIfPluginStopsAfterRegistration, + ginkgo.Entry("2 sockets", ""), + ginkgo.Entry("1 common socket", path.Join(kubeletplugin.KubeletRegistryDir, driverName+"-common.sock")), + ) f.It("must be removed if plugin is unresponsive after registration", func(ctx context.Context) { nodeName := getNodeName(ctx, f) @@ -655,7 +672,7 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices without plugin") }) - f.It("must not be removed if plugin restarts quickly enough", func(ctx context.Context) { + testRemoveIfRestartsQuickly := func(ctx context.Context, socketPath string) { nodeName := getNodeName(ctx, f) ginkgo.By("start DRA registrar") @@ -687,7 +704,12 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami ginkgo.By("ensuring unchanged ResourceSlices") gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(time.Minute).Should(gomega.Equal(slices), "ResourceSlices") - }) + } + ginkgo.DescribeTable("must not be removed if plugin restarts quickly enough", + testRemoveIfRestartsQuickly, + ginkgo.Entry("2 sockets", ""), + ginkgo.Entry("1 common socket", path.Join(kubeletplugin.KubeletRegistryDir, driverName+"-common.sock")), + ) }) }) From cc7893a42c6f42635506b77d78781a8b7aaf00ab Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Mon, 23 Jun 2025 09:44:43 +0300 Subject: [PATCH 8/9] kubelet: DRA: fix unit test failure Fixed race condition caused by removing already removed socket directory. --- pkg/kubelet/cm/dra/plugin/dra_plugin_test.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go index 9d506577032..314a8fd923d 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go @@ -20,7 +20,6 @@ import ( "context" "fmt" "net" - "os" "path" "strings" "sync" @@ -63,9 +62,6 @@ func setupFakeGRPCServer(service, addr string) (tearDown, error) { ctx, cancel := context.WithCancel(context.Background()) teardown := func() { cancel() - if err := os.RemoveAll(addr); err != nil { - panic(err) - } } listener, err := net.Listen("unix", addr) From 6040344a65c9fc3eb1a9574b1ac36510de7da510 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Mon, 23 Jun 2025 11:00:32 +0300 Subject: [PATCH 9/9] kubelet: DRA: fix TestRegistrationHandler Using the same socket path for different test cases caused test failure on windows: listen unix C:\Users\azureuser\AppData\Local\Temp\TestRegistrationHandler3881105518\001/dra-plugin-a.sock: bind: Only one usage of each socket address (protocol/network address/port) is normally permitted. Creating unique socket path for every test case should fix it. --- .../cm/dra/plugin/registration_test.go | 34 ++++++++++--------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/pkg/kubelet/cm/dra/plugin/registration_test.go b/pkg/kubelet/cm/dra/plugin/registration_test.go index 4c8919937b2..004409a25a2 100644 --- a/pkg/kubelet/cm/dra/plugin/registration_test.go +++ b/pkg/kubelet/cm/dra/plugin/registration_test.go @@ -122,14 +122,13 @@ func TestRegistrationHandler(t *testing.T) { }, } - tmp := t.TempDir() - endpointA := path.Join(tmp, "a.sock") - endpointB := path.Join(tmp, "b.sock") + socketFileA := "a.sock" + socketFileB := "b.sock" for _, test := range []struct { description string driverName string - endpoint string + socketFile string withClient bool supportedServices []string shouldError bool @@ -138,27 +137,27 @@ func TestRegistrationHandler(t *testing.T) { { description: "no-services", driverName: pluginB, - endpoint: endpointB, + socketFile: socketFileB, shouldError: true, }, { description: "current-service", driverName: pluginB, - endpoint: endpointB, + socketFile: socketFileB, supportedServices: []string{drapb.DRAPluginService}, chosenService: drapb.DRAPluginService, }, { description: "two-services", driverName: pluginB, - endpoint: endpointB, + socketFile: socketFileB, supportedServices: []string{drapbv1alpha4.NodeService, drapb.DRAPluginService}, chosenService: drapb.DRAPluginService, }, { description: "old-service", driverName: pluginB, - endpoint: endpointB, + socketFile: socketFileB, supportedServices: []string{drapbv1alpha4.NodeService}, chosenService: drapbv1alpha4.NodeService, }, @@ -166,14 +165,14 @@ func TestRegistrationHandler(t *testing.T) { // Legacy behavior. description: "version", driverName: pluginB, - endpoint: endpointB, + socketFile: socketFileB, supportedServices: []string{"1.0.0"}, chosenService: drapbv1alpha4.NodeService, }, { description: "replace", driverName: pluginA, - endpoint: endpointB, + socketFile: socketFileB, supportedServices: []string{drapb.DRAPluginService}, chosenService: drapb.DRAPluginService, }, @@ -181,7 +180,7 @@ func TestRegistrationHandler(t *testing.T) { description: "manage-resource-slices", withClient: true, driverName: pluginB, - endpoint: endpointB, + socketFile: socketFileB, supportedServices: []string{drapb.DRAPluginService}, chosenService: drapb.DRAPluginService, }, @@ -200,11 +199,14 @@ func TestRegistrationHandler(t *testing.T) { // of the connection state. service := drapb.DRAPluginService + tmp := t.TempDir() + endpointA := path.Join(tmp, socketFileA) teardownA, err := setupFakeGRPCServer(service, endpointA) require.NoError(t, err) tCtx.Cleanup(teardownA) - teardown, err := setupFakeGRPCServer(service, test.endpoint) + endpoint := path.Join(tmp, test.socketFile) + teardown, err := setupFakeGRPCServer(service, endpoint) require.NoError(t, err) tCtx.Cleanup(teardown) @@ -232,7 +234,7 @@ func TestRegistrationHandler(t *testing.T) { draPlugins.DeRegisterPlugin(pluginA, endpointA) }) - err = draPlugins.ValidatePlugin(test.driverName, test.endpoint, test.supportedServices) + err = draPlugins.ValidatePlugin(test.driverName, endpoint, test.supportedServices) if test.shouldError { require.Error(t, err) } else { @@ -246,7 +248,7 @@ func TestRegistrationHandler(t *testing.T) { } // Add plugin for the first time. - err = draPlugins.RegisterPlugin(test.driverName, test.endpoint, test.supportedServices, nil) + err = draPlugins.RegisterPlugin(test.driverName, endpoint, test.supportedServices, nil) if test.shouldError { require.Error(t, err) } else { @@ -262,9 +264,9 @@ func TestRegistrationHandler(t *testing.T) { } tCtx.Logf("Removing plugin %s", test.driverName) - draPlugins.DeRegisterPlugin(test.driverName, test.endpoint) + draPlugins.DeRegisterPlugin(test.driverName, endpoint) // Nop. - draPlugins.DeRegisterPlugin(test.driverName, test.endpoint) + draPlugins.DeRegisterPlugin(test.driverName, endpoint) if test.withClient { requireNoSlices(tCtx) }