github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-08-08 23:37:11 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #132058 from pohly/dra-kubelet-connection-monitoring

Browse Source 
DRA kubelet: connection monitoring
This commit is contained in:
Kubernetes Prow Robot
2025-06-26 03:40:29 -07:00
committed by GitHub
parent c75d425ae2 6040344a65
commit dcefe0ef41
8 changed files with 850 additions and 406 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
pkg/kubelet/cm/dra/plugin/dra_plugin.go
View File

@@ -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

308
pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go
View File

@@ -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].

56
pkg/kubelet/cm/dra/plugin/dra_plugin_manager_test.go
View File

@@ -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, "")
}

88
pkg/kubelet/cm/dra/plugin/dra_plugin_test.go
View File

@@ -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)

273
pkg/kubelet/cm/dra/plugin/registration_test.go
View File

@@ -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")
}
})
}
}

52
pkg/kubelet/pluginmanager/pluginwatcher/README.md
View File

@@ -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

100
staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go
View File

@@ -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.

303
test/e2e_node/dra_test.go
View File

@@ -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))
}