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..ae5cc972bc9 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go @@ -24,6 +24,10 @@ import ( "sync" "time" + "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" @@ -34,7 +38,9 @@ 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" ) // DRAPluginManager keeps track of how to reach plugins registered for DRA drivers. @@ -44,7 +50,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,21 +64,72 @@ type DRAPluginManager struct { wg sync.WaitGroup mutex sync.RWMutex - // driver name -> Plugin in the order in which they got added - store map[string][]*DRAPlugin + // 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{} +// 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 + } + 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() + 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. // @@ -87,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 @@ -104,39 +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")) - pm.wg.Wait() + + // Close all connections, otherwise gRPC keeps doing things in the background. + // 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 { + 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, @@ -196,19 +266,34 @@ 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() + 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]. @@ -223,46 +308,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 +325,55 @@ 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) + p := &DRAPlugin{ + driverName: driverName, + endpoint: endpoint, + chosenService: chosenService, + clientCallTimeout: clientCallTimeout, } - for _, oldP := range pm.store[p.driverName] { - if oldP.endpoint == p.endpoint { + if pm.store == nil { + pm.store = make(map[string][]*monitoredPlugin) + } + 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) - pm.store[p.driverName] = append(pm.store[p.driverName], p) - logger.V(3).Info("Registered DRA plugin", "numInstances", len(pm.store[p.driverName])) + 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) + conn, err := grpc.NewClient( + 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 + + // 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], 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 } @@ -306,7 +393,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 } @@ -317,71 +404,58 @@ 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) } // 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 Plugin 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. // 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_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..314a8fd923d 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go @@ -20,8 +20,7 @@ import ( "context" "fmt" "net" - "os" - "path/filepath" + "path" "strings" "sync" "testing" @@ -56,27 +55,19 @@ 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) - if err := os.RemoveAll(addr); err != nil { - panic(err) - } + cancel() } listener, err := net.Listen("unix", addr) if err != nil { teardown() - return "", nil, err + return nil, err } s := grpc.NewServer() @@ -87,7 +78,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 +87,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) } @@ -117,28 +109,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 +141,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 +171,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", }, @@ -229,27 +204,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", @@ -257,7 +228,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", @@ -265,34 +235,16 @@ 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) } 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) diff --git a/pkg/kubelet/cm/dra/plugin/registration_test.go b/pkg/kubelet/cm/dra/plugin/registration_test.go index 8012a060799..004409a25a2 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" @@ -35,21 +37,83 @@ 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" ) 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,39 +122,42 @@ func TestRegistrationHandler(t *testing.T) { }, } + socketFileA := "a.sock" + socketFileB := "b.sock" + for _, test := range []struct { description string driverName string - endpoint string + socketFile string withClient bool supportedServices []string shouldError bool chosenService string }{ { - description: "no-services-provided", + 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, }, @@ -98,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, }, @@ -113,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, }, @@ -121,78 +188,53 @@ 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 + tmp := t.TempDir() + endpointA := path.Join(tmp, socketFileA) + teardownA, err := setupFakeGRPCServer(service, endpointA) + require.NoError(t, err) + tCtx.Cleanup(teardownA) + + endpoint := path.Join(tmp, test.socketFile) + teardown, err := setupFakeGRPCServer(service, 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) 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 { @@ -206,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 { @@ -222,14 +264,99 @@ 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) - - requireNoSlices() + draPlugins.DeRegisterPlugin(test.driverName, endpoint) + 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-on-disconnect": { + delay: time.Second, // very short wiping delay for testing + requireSliceRemoval: true, + }, + "no-wipe-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.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 { + 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) + require.NoError(t, err) + defer teardown() + + // There shouldn't be any pending wipes for the plugin. + require.Eventuallyf(t, func() bool { + return !wipingIsPending() + }, 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..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. @@ -397,6 +410,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 @@ -404,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 @@ -413,6 +447,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 +478,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 +499,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 { @@ -488,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, @@ -530,34 +572,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 - 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, + if o.nodeV1beta1 { + logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service") + supportedServices = append(supportedServices, drapb.DRAPluginService) } - 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 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") } - - // 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) + draEndpoint := endpoint{ + dir: o.pluginDataDirectoryPath, + file: o.pluginSocket, + listenFunc: o.draEndpointListen, + } + + 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) + } + d.pluginServer = pluginServer + } + + 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..74177a9a028 100644 --- a/test/e2e_node/dra_test.go +++ b/test/e2e_node/dra_test.go @@ -378,6 +378,74 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami return kubeletPlugin.CountCalls("/NodePrepareResources") }).WithTimeout(retryTestTimeout).Should(gomega.Equal(calls)) }) + + functionalListenAfterRegistration := func(ctx context.Context, socketPath string) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + 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, socketPath) + + 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) + + 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")), + ) + + functionalAfterServiceReconnect := func(ctx context.Context, socketPath string) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + 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, 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}) + + 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, 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() { @@ -513,18 +581,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 +599,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,14 +616,100 @@ 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") }) + + removedIfPluginStopsAfterRegistration := func(ctx context.Context, socketPath string) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + 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, socketPath) + + 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") + } + 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) + + 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") + }) + + testRemoveIfRestartsQuickly := func(ctx context.Context, socketPath string) { + 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") + 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") + } + 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")), + ) }) }) @@ -622,10 +764,102 @@ 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, 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) + 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, 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) + + // 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") + 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") + + 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", + }, + }, + }}, + }, + }, + }, + }, + opts..., + ) + 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 +933,10 @@ func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, node RestartPolicy: v1.RestartPolicyNever, }, } + if podName == "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 +1010,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)) +}