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

DRA kubelet: add connection monitoring

Browse Source 
This ensures that ResourceSlices get removed also when a plugin becomes
unresponsive without removing the registration socket.

Tests are from https://github.com/kubernetes/kubernetes/pull/131073 by Ed
with some modifications, the implementation is new.
This commit is contained in:
Ed Bartosh
2025-03-26 20:05:48 +02:00
committed by Patrick Ohly
parent f927cd0108
commit 4ee7374b24
6 changed files with 636 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,6 +26,8 @@ import (
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
grpcstats "google.golang.org/grpc/stats"
v1 "k8s.io/api/core/v1"
resourceapi "k8s.io/api/resource/v1beta1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -58,11 +60,21 @@ type DRAPluginManager struct {
getNode func() (*v1.Node, error)
wipingDelay time.Duration
// TODO: replace pendingWipes with some kind of workqueue.
// As it stands, WaitGroup suffers from a data race:
// - Queueing a new wiping creates a goroutine and adds to
// to wg.
// - Concurrently, wg.Wait reads from it.
//
// This is not allowed, all wg.Adds must come before wg.Wait.
//
// This race can be triggered with
// go test -count=10 -race ./...
wg sync.WaitGroup
mutex sync.RWMutex
// driver name -> DRAPlugin in the order in which they got added
store map[string][]*DRAPlugin
store map[string][]*monitoredPlugin
// pendingWipes maps a driver name to a cancel function for
// wiping of that plugin's ResourceSlices. Entries get added
@@ -76,6 +88,54 @@ type DRAPluginManager struct {
var _ cache.PluginHandler = &DRAPluginManager{}
// monitoredPlugin tracks whether the gRPC connection of a plugin is
// currently connected. Fot that it implements the [grpcstats.Handler]
// interface.
//
// The tagging functions might be useful for contextual logging. But
// for now all that matters is HandleConn.
type monitoredPlugin struct {
*DRAPlugin
pm *DRAPluginManager
// connected is protected by store.mutex.
connected bool
}
var _ grpcstats.Handler = &monitoredPlugin{}
func (m *monitoredPlugin) TagRPC(ctx context.Context, info *grpcstats.RPCTagInfo) context.Context {
return ctx
}
func (m *monitoredPlugin) HandleRPC(context.Context, grpcstats.RPCStats) {
}
func (m *monitoredPlugin) TagConn(ctx context.Context, info *grpcstats.ConnTagInfo) context.Context {
return ctx
}
func (m *monitoredPlugin) HandleConn(_ context.Context, stats grpcstats.ConnStats) {
connected := false
switch stats.(type) {
case *grpcstats.ConnBegin:
connected = true
case *grpcstats.ConnEnd:
// We have to ask for a reconnect, otherwise gRPC wouldn't try and
// thus we wouldn't be notified about a restart of the plugin.
m.conn.Connect()
default:
return
}
logger := klog.FromContext(m.pm.backgroundCtx)
m.pm.mutex.Lock()
defer m.pm.mutex.Unlock()
logger.V(2).Info("Connection changed", "driverName", m.driverName, "endpoint", m.endpoint, "connected", connected)
m.connected = connected
m.pm.sync(m.driverName)
}
// NewDRAPluginManager creates a new DRAPluginManager, with support for wiping ResourceSlices
// when the plugin(s) for a DRA driver are not available too long.
//
@@ -216,14 +276,29 @@ func (pm *DRAPluginManager) get(driverName string) *DRAPlugin {
pm.mutex.RLock()
defer pm.mutex.RUnlock()
logger := klog.FromContext(pm.backgroundCtx)
plugins := pm.store[driverName]
if len(plugins) == 0 {
logger.V(5).Info("No plugin registered", "driverName", driverName)
return nil
}
// Heuristic: pick the most recent one. It's most likely
// the newest, except when kubelet got restarted and registered
// all running plugins in random order.
return plugins[len(plugins)-1]
//
// Prefer plugins which are connected, otherwise also
// disconnected ones.
for i := len(plugins) - 1; i >= 0; i-- {
if plugin := plugins[i]; plugin.connected {
logger.V(5).Info("Preferring connected plugin", "driverName", driverName, "endpoint", plugin.endpoint)
return plugin.DRAPlugin
}
}
plugin := plugins[len(plugins)-1]
logger.V(5).Info("No plugin connected, using latest one", "driverName", driverName, "endpoint", plugin.endpoint)
return plugin.DRAPlugin
}
// RegisterPlugin implements [cache.PluginHandler].
@@ -259,8 +334,14 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic
pm.mutex.Lock()
defer pm.mutex.Unlock()
p := &DRAPlugin{
driverName: driverName,
endpoint: endpoint,
chosenService: chosenService,
clientCallTimeout: clientCallTimeout,
}
if pm.store == nil {
pm.store = make(map[string][]*DRAPlugin)
pm.store = make(map[string][]*monitoredPlugin)
}
for _, oldP := range pm.store[driverName] {
if oldP.endpoint == endpoint {
@@ -271,6 +352,11 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic
logger := klog.FromContext(pm.backgroundCtx)
mp := &monitoredPlugin{
DRAPlugin: p,
pm: pm,
}
// The gRPC connection gets created once. gRPC then connects to the gRPC server on demand.
target := "unix:" + endpoint
logger.V(4).Info("Creating new gRPC connection", "target", target)
@@ -278,20 +364,20 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic
target,
grpc.WithTransportCredentials(insecure.NewCredentials()),
grpc.WithChainUnaryInterceptor(newMetricsInterceptor(driverName)),
grpc.WithStatsHandler(mp),
)
if err != nil {
return fmt.Errorf("create gRPC connection to DRA driver %s plugin at endpoint %s: %w", driverName, endpoint, err)
}
p.conn = conn
p := &DRAPlugin{
driverName: driverName,
endpoint: endpoint,
conn: conn,
chosenService: chosenService,
clientCallTimeout: clientCallTimeout,
}
// Ensure that gRPC tries to connect even if we don't call any gRPC method.
// This is necessary to detect early whether a plugin is really available.
// This is currently an experimental gRPC method. Should it be removed we
// would need to do something else, like sending a fake gRPC method call.
conn.Connect()
pm.store[p.driverName] = append(pm.store[p.driverName], p)
pm.store[p.driverName] = append(pm.store[p.driverName], mp)
logger.V(3).Info("Registered DRA plugin", "driverName", p.driverName, "endpoint", p.endpoint, "chosenService", p.chosenService, "numPlugins", len(pm.store[p.driverName]))
pm.sync(p.driverName)
return nil
@@ -312,7 +398,7 @@ func (pm *DRAPluginManager) remove(driverName, endpoint string) {
defer pm.mutex.Unlock()
plugins := pm.store[driverName]
i := slices.IndexFunc(plugins, func(p *DRAPlugin) bool { return p.driverName == driverName && p.endpoint == endpoint })
i := slices.IndexFunc(plugins, func(mp *monitoredPlugin) bool { return mp.driverName == driverName && mp.endpoint == endpoint })
if i == -1 {
return
}
@@ -382,7 +468,12 @@ func (pm *DRAPluginManager) sync(driverName string) {
// usable returns true if at least one endpoint is ready to handle gRPC calls for the DRA driver.
// Must be called while holding the mutex.
func (pm *DRAPluginManager) usable(driverName string) bool {
return len(pm.store[driverName]) > 0
for _, mp := range pm.store[driverName] {
if mp.connected {
return true
}
}
return false
}
// ValidatePlugin implements [cache.PluginHandler].

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

@@ -21,7 +21,7 @@ import (
"fmt"
"net"
"os"
"path/filepath"
"path"
"strings"
"sync"
"testing"
@@ -56,18 +56,13 @@ func (f *fakeGRPCServer) NodeUnprepareResources(ctx context.Context, in *drapbv1
return &drapbv1beta1.NodeUnprepareResourcesResponse{}, nil
}
// tearDown is an idempotent cleanup function.
type tearDown func()
func setupFakeGRPCServer(service string) (string, tearDown, error) {
p, err := os.MkdirTemp("", "dra_plugin")
if err != nil {
return "", nil, err
}
closeCh := make(chan struct{})
addr := filepath.Join(p, "server.sock")
func setupFakeGRPCServer(service, addr string) (tearDown, error) {
ctx, cancel := context.WithCancel(context.Background())
teardown := func() {
close(closeCh)
cancel()
if err := os.RemoveAll(addr); err != nil {
panic(err)
}
@@ -76,7 +71,7 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) {
listener, err := net.Listen("unix", addr)
if err != nil {
teardown()
return "", nil, err
return nil, err
}
s := grpc.NewServer()
@@ -87,7 +82,7 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) {
case drapbv1alpha4.NodeService:
drapbv1alpha4.RegisterNodeServer(s, drapbv1alpha4.V1Beta1ServerWrapper{DRAPluginServer: fakeGRPCServer})
default:
return "", nil, fmt.Errorf("unsupported gRPC service: %s", service)
return nil, fmt.Errorf("unsupported gRPC service: %s", service)
}
go func() {
@@ -96,17 +91,18 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) {
panic(err)
}
}()
<-closeCh
<-ctx.Done()
s.GracefulStop()
}()
return addr, teardown, nil
return teardown, nil
}
func TestGRPCConnIsReused(t *testing.T) {
tCtx := ktesting.Init(t)
service := drapbv1beta1.DRAPluginService
addr, teardown, err := setupFakeGRPCServer(service)
addr := path.Join(t.TempDir(), "dra.sock")
teardown, err := setupFakeGRPCServer(service, addr)
if err != nil {
t.Fatal(err)
}
@@ -212,27 +208,23 @@ func TestGetDRAPlugin(t *testing.T) {
func TestGRPCMethods(t *testing.T) {
for _, test := range []struct {
description string
serverSetup func(string) (string, tearDown, error)
service string
chosenService string
expectError string
}{
{
description: "v1alpha4",
serverSetup: setupFakeGRPCServer,
service: drapbv1alpha4.NodeService,
chosenService: drapbv1alpha4.NodeService,
},
{
description: "v1beta1",
serverSetup: setupFakeGRPCServer,
service: drapbv1beta1.DRAPluginService,
chosenService: drapbv1beta1.DRAPluginService,
},
{
// In practice, such a mismatch between plugin and kubelet should not happen.
description: "mismatch",
serverSetup: setupFakeGRPCServer,
service: drapbv1beta1.DRAPluginService,
chosenService: drapbv1alpha4.NodeService,
expectError: "unknown service v1alpha3.Node",
@@ -240,7 +232,6 @@ func TestGRPCMethods(t *testing.T) {
{
// In practice, kubelet wouldn't choose an invalid service.
description: "internal-error",
serverSetup: setupFakeGRPCServer,
service: drapbv1beta1.DRAPluginService,
chosenService: "some-other-service",
expectError: "unsupported chosen service",
@@ -248,7 +239,8 @@ func TestGRPCMethods(t *testing.T) {
} {
t.Run(test.description, func(t *testing.T) {
tCtx := ktesting.Init(t)
addr, teardown, err := setupFakeGRPCServer(test.service)
addr := path.Join(t.TempDir(), "dra.sock")
teardown, err := setupFakeGRPCServer(test.service, addr)
if err != nil {
t.Fatal(err)
}

242
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"
@@ -39,17 +41,78 @@ import (
)
const (
nodeName = "worker"
pluginA = "pluginA"
endpointA = "endpointA"
pluginB = "pluginB"
endpointB = "endpointB"
nodeName = "worker"
pluginA = "pluginA"
pluginB = "pluginB"
)
func getFakeNode() (*v1.Node, error) {
return &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}, nil
}
func getSlice(name string) *resourceapi.ResourceSlice {
return &resourceapi.ResourceSlice{
ObjectMeta: metav1.ObjectMeta{Name: name},
Spec: resourceapi.ResourceSliceSpec{
NodeName: nodeName,
},
}
}
func getFakeClient(t *testing.T, nodeName, driverName string, slice *resourceapi.ResourceSlice) kubernetes.Interface {
expectedSliceFields := fields.Set{"spec.nodeName": nodeName}
fakeClient := fake.NewClientset(slice)
fakeClient.AddReactor("delete-collection", "resourceslices", func(action cgotesting.Action) (bool, runtime.Object, error) {
deleteAction := action.(cgotesting.DeleteCollectionAction)
restrictions := deleteAction.GetListRestrictions()
fieldsSelector := fields.SelectorFromSet(expectedSliceFields)
// The order of field requirements is random because it comes
// from a map. We need to sort.
normalize := func(selector string) string {
requirements := strings.Split(selector, ",")
sort.Strings(requirements)
return strings.Join(requirements, ",")
}
assert.Empty(t, restrictions.Labels.String(), "label selector in DeleteCollection")
assert.Equal(t, normalize(fieldsSelector.String()), normalize(restrictions.Fields.String()), "field selector in DeleteCollection")
// There's only one object that could get matched, so delete it.
// Delete doesn't return an error if already deleted, which is what
// we need here (no error when nothing to delete).
err := fakeClient.Tracker().Delete(resourceapi.SchemeGroupVersion.WithResource("resourceslices"), "", slice.Name)
// Set expected slice fields for the next call of this reactor.
// The reactor will be called next time when resourceslices object is deleted
// by the kubelet after plugin deregistration.
switch len(expectedSliceFields) {
case 1:
// Startup cleanup done, now expect cleanup for test plugin.
expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": driverName}
case 2:
// Test plugin cleanup done, now expect cleanup for the other plugin.
otherPlugin := pluginA
if otherPlugin == driverName {
otherPlugin = pluginB
}
expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": otherPlugin}
}
return true, nil, err
})
return fakeClient
}
func requireNoSlices(tCtx ktesting.TContext) {
tCtx.Helper()
ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) error {
slices, err := tCtx.Client().ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{})
if err != nil {
return err
}
assert.Empty(tCtx, slices.Items, "slices")
return nil
}).Should(gomega.Succeed(), "there should be no slices")
}
func TestRegistrationHandler(t *testing.T) {
slice := &resourceapi.ResourceSlice{
ObjectMeta: metav1.ObjectMeta{Name: "test-slice"},
@@ -58,6 +121,10 @@ func TestRegistrationHandler(t *testing.T) {
},
}
tmp := t.TempDir()
endpointA := path.Join(tmp, "dra-plugin-a.sock")
endpointB := path.Join(tmp, "dra-plugin-b.sock")
for _, test := range []struct {
description string
driverName string
@@ -121,71 +188,43 @@ func TestRegistrationHandler(t *testing.T) {
t.Run(test.description, func(t *testing.T) {
tCtx := ktesting.Init(t)
// Run GRPC services for both plugins.
//
// This is necessary because otherwise connection
// monitoring will start wiping slices, regardless
// of whether the plugin is registered or not.
//
// Here we are only interested in registration.
// In TestConnectionHandling we check detection
// of the connection state.
service := drapb.DRAPluginService
teardownA, err := setupFakeGRPCServer(service, endpointA)
require.NoError(t, err)
tCtx.Cleanup(teardownA)
teardown, err := setupFakeGRPCServer(service, test.endpoint)
require.NoError(t, err)
tCtx.Cleanup(teardown)
// Stand-alone kubelet has no connection to an
// apiserver, so faking one is optional.
var client kubernetes.Interface
if test.withClient {
expectedSliceFields := fields.Set{"spec.nodeName": nodeName}
fakeClient := fake.NewClientset(slice)
fakeClient.AddReactor("delete-collection", "resourceslices", func(action cgotesting.Action) (bool, runtime.Object, error) {
deleteAction := action.(cgotesting.DeleteCollectionAction)
restrictions := deleteAction.GetListRestrictions()
fieldsSelector := fields.SelectorFromSet(expectedSliceFields)
// The order of field requirements is random because it comes
// from a map. We need to sort.
normalize := func(selector string) string {
requirements := strings.Split(selector, ",")
sort.Strings(requirements)
return strings.Join(requirements, ",")
}
assert.Equal(t, "", restrictions.Labels.String(), "label selector in DeleteCollection")
assert.Equal(t, normalize(fieldsSelector.String()), normalize(restrictions.Fields.String()), "field selector in DeleteCollection")
// There's only one object that could get matched, so delete it.
// Delete doesn't return an error if already deleted, which is what
// we need here (no error when nothing to delete).
err := fakeClient.Tracker().Delete(resourceapi.SchemeGroupVersion.WithResource("resourceslices"), "", slice.Name)
// Set expected slice fields for the next call of this reactor.
// The reactor will be called next time when resourceslices object is deleted
// by the kubelet after plugin deregistration.
switch len(expectedSliceFields) {
case 1:
// Startup cleanup done, now expect cleanup for test plugin.
expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": test.driverName}
case 2:
// Test plugin cleanup done, now expect cleanup for the other plugin.
otherPlugin := pluginA
if otherPlugin == test.driverName {
otherPlugin = pluginB
}
expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": otherPlugin}
}
return true, nil, err
})
fakeClient := getFakeClient(t, nodeName, test.driverName, getSlice("test-slice"))
client = fakeClient
tCtx = ktesting.WithClients(tCtx, nil, nil, client, nil, nil)
}
// The DRAPluginManager wipes all slices at startup.
draPlugins := NewDRAPluginManager(tCtx, client, getFakeNode, time.Second /* very short wiping delay for testing */)
tCtx.Cleanup(draPlugins.Stop)
requireNoSlices := func() {
t.Helper()
if client == nil {
return
}
require.EventuallyWithT(t, func(t *assert.CollectT) {
slices, err := client.ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{})
if !assert.NoError(t, err, "list slices") {
return
}
assert.Empty(t, slices.Items, "slices")
}, time.Minute, time.Second)
if test.withClient {
requireNoSlices(tCtx)
}
requireNoSlices()
// Simulate one existing plugin A.
err := draPlugins.RegisterPlugin(pluginA, endpointA, []string{drapb.DRAPluginService}, nil)
err = draPlugins.RegisterPlugin(pluginA, endpointA, []string{drapb.DRAPluginService}, nil)
require.NoError(t, err)
t.Cleanup(func() {
tCtx.Logf("Removing plugin %s", pluginA)
@@ -225,11 +264,94 @@ func TestRegistrationHandler(t *testing.T) {
draPlugins.DeRegisterPlugin(test.driverName, test.endpoint)
// Nop.
draPlugins.DeRegisterPlugin(test.driverName, test.endpoint)
requireNoSlices()
if test.withClient {
requireNoSlices(tCtx)
}
})
assert.Equal(t, test.endpoint, plugin.endpoint, "plugin endpoint")
// Which plugin was chosen is random in this test: it depends on which plugin was detected as connected,
// which can be both, one, or none at this point. Some attributes are common to both.
assert.Equal(t, test.driverName, plugin.driverName, "DRA driver driver name")
assert.Equal(t, test.chosenService, plugin.chosenService, "chosen service")
})
}
}
// TestConnectionHandling checks the reaction to state changes of the service connection.
func TestConnectionHandling(t *testing.T) {
t.Parallel()
for description, test := range map[string]struct {
delay time.Duration
requireSliceRemoval bool
}{
"wipe-slices-on-disconnect": {
delay: time.Second, // very short wiping delay for testing
requireSliceRemoval: true,
},
"no-wipe-slices-on-reconnect": {
delay: time.Hour, // long delay to avoid wiping while the test runs
requireSliceRemoval: false,
},
} {
t.Run(description, func(t *testing.T) {
t.Parallel()
tCtx := ktesting.Init(t)
service := drapb.DRAPluginService
driverName := "test-plugin"
sliceName := "test-slice"
slice := getSlice(sliceName)
client := getFakeClient(t, nodeName, driverName, slice)
tCtx = ktesting.WithClients(tCtx, nil, nil, client, nil, nil)
// The handler wipes all slices at startup.
draPlugins := NewDRAPluginManager(tCtx, client, getFakeNode, test.delay)
tCtx.Cleanup(draPlugins.Stop)
requireNoSlices(tCtx)
// Run GRPC service.
endpoint := path.Join(t.TempDir(), "dra-plugin-test.sock")
teardown, err := setupFakeGRPCServer(service, endpoint)
require.NoError(t, err)
defer teardown()
err = draPlugins.RegisterPlugin(driverName, endpoint, []string{service}, nil)
require.NoError(t, err)
plugin := draPlugins.get(driverName)
assert.NotNil(t, plugin, "plugin should be present in the plugin store")
// Create the slice as if the plugin had done that while it runs.
_, err = client.ResourceV1beta1().ResourceSlices().Create(tCtx, slice, metav1.CreateOptions{})
require.NoError(t, err, "recreate slice")
// Stop gRPC server.
tCtx.Log("Stopping plugin gRPC server")
teardown()
if test.requireSliceRemoval {
// Slice should get removed.
requireNoSlices(tCtx)
} else {
// Start up gRPC server again.
tCtx.Log("Restarting plugin gRPC server")
teardown, err = setupFakeGRPCServer(service, endpoint)
require.NoError(t, err)
defer teardown()
// There shouldn't be any pending wipes for the plugin.
require.Eventuallyf(t, func() bool {
draPlugins.mutex.Lock()
defer draPlugins.mutex.Unlock()
_, ok := draPlugins.pendingWipes[driverName]
return ok
}, time.Minute, time.Second, "wiping should be stopped for plugin %s", driverName)
// Slice should still be there
slices, err := client.ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{})
require.NoError(t, err, "list slices")
assert.Len(t, slices.Items, 1, "slices")
}
})
}
}

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

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

@@ -397,6 +397,26 @@ func FlockDirectoryPath(path string) Option {
}
}
// RegistrationService controls whether the kubelet plugin gRPC service
// is started. It's on by default. This is meant for testing, normal
// DRA drivers should use the default.
func RegistrationService(enabled bool) Option {
return func(o *options) error {
o.registrationService = enabled
return nil
}
}
// DRAService controls whether the DRA gRPC service
// is started. It's on by default. This is meant for testing, normal
// DRA drivers should use the default.
func DRAService(enabled bool) Option {
return func(o *options) error {
o.draService = enabled
return nil
}
}
type options struct {
logger klog.Logger
grpcVerbosity int
@@ -413,6 +433,8 @@ type options struct {
serialize bool
flockDirectoryPath string
nodeV1beta1 bool
registrationService bool
draService bool
}
// Helper combines the kubelet registration service and the DRA node plugin
@@ -442,8 +464,9 @@ type Helper struct {
resourceSliceController *resourceslice.Controller
}
// Start sets up two gRPC servers (one for registration, one for the DRA node
// client) and implements them by calling a [DRAPlugin] implementation.
// Start sets up all enabled gRPC servers (by default, one for registration,
// one for the DRA node client) and implements them by calling a [DRAPlugin]
// implementation.
//
// The context and/or DRAPlugin.Stop can be used to stop all background activity.
// Stop also blocks. A logger can be stored in the context to add values or
@@ -462,6 +485,8 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe
pluginRegistrationEndpoint: endpoint{
dir: KubeletRegistryDir,
},
draService: true,
registrationService: true,
}
for _, option := range opts {
if err := option(&o); err != nil {
@@ -530,34 +555,42 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe
}
}()
// Run the node plugin gRPC server first to ensure that it is ready.
var supportedServices []string
if o.nodeV1beta1 {
logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service")
supportedServices = append(supportedServices, drapb.DRAPluginService)
}
if len(supportedServices) == 0 {
return nil, errors.New("no supported DRA gRPC API is implemented and enabled")
}
draEndpoint := endpoint{
dir: o.pluginDataDirectoryPath,
file: "dra" + uidPart + ".sock", // "dra" is hard-coded. The directory is unique, so we get a unique full path also without the UID.
listenFunc: o.draEndpointListen,
}
pluginServer, err := startGRPCServer(klog.LoggerWithName(logger, "dra"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, draEndpoint, func(grpcServer *grpc.Server) {
if o.nodeV1beta1 {
logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service")
drapb.RegisterDRAPluginServer(grpcServer, &nodePluginImplementation{Helper: d})
supportedServices = append(supportedServices, drapb.DRAPluginService)
if o.draService {
// Run the node plugin gRPC server first to ensure that it is ready.
pluginServer, err := startGRPCServer(klog.LoggerWithName(logger, "dra"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, draEndpoint, func(grpcServer *grpc.Server) {
if o.nodeV1beta1 {
logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service")
drapb.RegisterDRAPluginServer(grpcServer, &nodePluginImplementation{Helper: d})
}
})
if err != nil {
return nil, fmt.Errorf("start DRA service: %w", err)
}
})
if err != nil {
return nil, fmt.Errorf("start node client: %v", err)
}
d.pluginServer = pluginServer
if len(supportedServices) == 0 {
return nil, errors.New("no supported DRA gRPC API is implemented and enabled")
d.pluginServer = pluginServer
}
// Now make it available to kubelet.
registrar, err := startRegistrar(klog.LoggerWithName(logger, "registrar"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, draEndpoint.path(), o.pluginRegistrationEndpoint)
if err != nil {
return nil, fmt.Errorf("start registrar: %v", err)
if o.registrationService {
// Now make it available to kubelet.
registrar, err := startRegistrar(klog.LoggerWithName(logger, "registrar"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, draEndpoint.path(), o.pluginRegistrationEndpoint)
if err != nil {
return nil, fmt.Errorf("start registrar: %w", err)
}
d.registrar = registrar
}
d.registrar = registrar
// startGRPCServer and startRegistrar don't implement cancellation
// themselves, we add that for both here.

252
test/e2e_node/dra_test.go
View File

@@ -378,6 +378,62 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami
return kubeletPlugin.CountCalls("/NodePrepareResources")
}).WithTimeout(retryTestTimeout).Should(gomega.Equal(calls))
})
ginkgo.It("must be functional when plugin starts to listen on a service socket after registration", func(ctx context.Context) {
ginkgo.By("start DRA registrar")
registrar := newRegistrar(ctx, f.ClientSet, getNodeName(ctx, f), driverName)
ginkgo.By("wait for registration to complete")
gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered)
ginkgo.By("start DRA plugin service")
draService := newDRAService(ctx, f.ClientSet, getNodeName(ctx, f), driverName)
pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drapod", false, []string{driverName})
ginkgo.By("wait for NodePrepareResources call to succeed")
gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded)
ginkgo.By("wait for pod to succeed")
err := e2epod.WaitForPodSuccessInNamespace(ctx, f.ClientSet, pod.Name, f.Namespace.Name)
framework.ExpectNoError(err)
})
ginkgo.It("must be functional after reconnect", func(ctx context.Context) {
nodeName := getNodeName(ctx, f)
ginkgo.By("start DRA registrar")
registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for registration to complete")
gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered)
ginkgo.By("start DRA plugin service")
draService := newDRAService(ctx, f.ClientSet, nodeName, driverName)
pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drasleeppod" /* enables sleeping */, false /* pod is deleted below */, []string{driverName})
ginkgo.By("wait for NodePrepareResources call to succeed")
gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded)
ginkgo.By("stop plugin")
draService.Stop()
ginkgo.By("waiting for pod to run")
err := e2epod.WaitForPodRunningInNamespace(ctx, f.ClientSet, pod)
framework.ExpectNoError(err)
ginkgo.By("wait for ResourceSlice removal, indicating detection of disconnect")
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin")
ginkgo.By("restarting plugin")
draService = newDRAService(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("stopping pod")
err = f.ClientSet.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})
framework.ExpectNoError(err)
gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodeUnprepareResourcesSucceeded)
})
})
f.Context("Two resource Kubelet Plugins", f.WithSerial(), func() {
@@ -513,18 +569,6 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami
})
f.Context("ResourceSlice", f.WithSerial(), func() {
listResources := func(ctx context.Context) ([]resourceapi.ResourceSlice, error) {
slices, err := f.ClientSet.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{})
if err != nil {
return nil, err
}
return slices.Items, nil
}
matchResourcesByNodeName := func(nodeName string) types.GomegaMatcher {
return gomega.HaveField("Spec.NodeName", gomega.Equal(nodeName))
}
f.It("must be removed on kubelet startup", f.WithDisruptive(), func(ctx context.Context) {
ginkgo.By("stop kubelet")
restartKubelet := mustStopKubelet(ctx, f)
@@ -543,15 +587,15 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami
matchAll := gomega.ConsistOf(matchResourcesByNodeName(nodeName), matchResourcesByNodeName(otherNodeName))
matchOtherNode := gomega.ConsistOf(matchResourcesByNodeName(otherNodeName))
gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchAll, "ResourceSlices without kubelet")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchAll, "ResourceSlices without kubelet")
ginkgo.By("restart kubelet")
restartKubelet(ctx)
restartKubelet = nil
ginkgo.By("wait for exactly the node's ResourceSlice to get deleted")
gomega.Eventually(ctx, listResources).Should(matchOtherNode, "ResourceSlices with kubelet")
gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchOtherNode, "ResourceSlices with kubelet")
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchOtherNode, "ResourceSlices with kubelet")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchOtherNode, "ResourceSlices with kubelet")
})
f.It("must be removed after plugin unregistration", func(ctx context.Context) {
@@ -560,13 +604,89 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami
ginkgo.By("start plugin and wait for ResourceSlice")
kubeletPlugin := newKubeletPlugin(ctx, f.ClientSet, getNodeName(ctx, f), driverName)
gomega.Eventually(ctx, listResources).Should(matchNode, "ResourceSlice from kubelet plugin")
gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlice from kubelet plugin")
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlice from kubelet plugin")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlice from kubelet plugin")
ginkgo.By("stop plugin and wait for ResourceSlice removal")
kubeletPlugin.Stop()
gomega.Eventually(ctx, listResources).Should(gomega.BeEmpty(), "ResourceSlices with no plugin")
gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices with no plugin")
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices with no plugin")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices with no plugin")
})
f.It("must be removed if plugin stops after registration", func(ctx context.Context) {
nodeName := getNodeName(ctx, f)
ginkgo.By("start DRA registrar")
registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for registration to complete")
gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered)
ginkgo.By("start DRA plugin service")
kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for ResourceSlice to be created by plugin")
matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName))
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlices")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlices")
ginkgo.By("stop plugin")
kubeletPlugin.Stop()
ginkgo.By("wait for ResourceSlice removal")
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices")
})
f.It("must be removed if plugin is unresponsive after registration", func(ctx context.Context) {
nodeName := getNodeName(ctx, f)
ginkgo.By("start DRA registrar")
registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for registration to complete")
gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered)
ginkgo.By("create a ResourceSlice")
createTestResourceSlice(ctx, f.ClientSet, nodeName, driverName)
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.ConsistOf(matchResourcesByNodeName(nodeName)), "ResourceSlices without plugin")
ginkgo.By("wait for ResourceSlice removal")
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices without plugin")
})
f.It("must not be removed if plugin restarts quickly enough", func(ctx context.Context) {
nodeName := getNodeName(ctx, f)
ginkgo.By("start DRA registrar")
registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for registration to complete")
gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered)
ginkgo.By("start DRA plugin service")
kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for ResourceSlice to be created by plugin")
matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName))
gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlices")
var slices []resourceapi.ResourceSlice
gomega.Consistently(ctx, listAndStoreResources(f.ClientSet, &slices)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlices")
ginkgo.By("stop plugin")
kubeletPlugin.Stop()
// We know from the "must be removed if plugin is unresponsive after registration" that the kubelet
// eventually notices the dropped connection. We cannot observe when that happens, we would need
// a new metric for that ("registered DRA plugins"). Let's give it a few seconds, which is significantly
// less than the wiping delay.
time.Sleep(5 * time.Second)
ginkgo.By("restarting plugin")
kubeletPlugin = newDRAService(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("ensuring unchanged ResourceSlices")
gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(time.Minute).Should(gomega.Equal(slices), "ResourceSlices")
})
})
})
@@ -622,10 +742,72 @@ func newKubeletPlugin(ctx context.Context, clientSet kubernetes.Interface, nodeN
return plugin
}
// newRegistrar starts a registrar for the specified DRA driver, without the DRA gRPC service.
func newRegistrar(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin {
ginkgo.By("start only Kubelet plugin registrar")
logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin registrar "+driverName))
ctx = klog.NewContext(ctx, logger)
registrar, err := testdriver.StartPlugin(ctx, cdiDir, driverName, clientSet, nodeName, testdriver.FileOperations{}, kubeletplugin.DRAService(false))
framework.ExpectNoError(err, "start only Kubelet plugin registrar")
return registrar
}
// newDRAService starts the DRA gRPC service for the specified DRA driver, without the registrar.
func newDRAService(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin {
ginkgo.By("start only Kubelet plugin")
logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin "+driverName), "node", nodeName)
ctx = klog.NewContext(ctx, logger)
// Ensure that directories exist, creating them if necessary. We want
// to know early if there is a setup problem that would prevent
// creating those directories.
err := os.MkdirAll(cdiDir, os.FileMode(0750))
framework.ExpectNoError(err, "create CDI directory")
datadir := path.Join(kubeletplugin.KubeletPluginsDir, driverName) // The default, not set below.
err = os.MkdirAll(datadir, 0750)
framework.ExpectNoError(err, "create DRA socket directory")
plugin, err := testdriver.StartPlugin(
ctx,
cdiDir,
driverName,
clientSet,
nodeName,
testdriver.FileOperations{
DriverResources: &resourceslice.DriverResources{
Pools: map[string]resourceslice.Pool{
nodeName: {
Slices: []resourceslice.Slice{{
Devices: []resourceapiv1beta2.Device{
{
Name: "device-00",
},
},
}},
},
},
},
},
kubeletplugin.RegistrationService(false),
)
framework.ExpectNoError(err)
ginkgo.DeferCleanup(func(ctx context.Context) {
// kubelet should do this eventually, but better make sure.
// A separate test checks this explicitly.
framework.ExpectNoError(clientSet.ResourceV1beta1().ResourceSlices().DeleteCollection(ctx, metav1.DeleteOptions{}, metav1.ListOptions{FieldSelector: resourceapi.ResourceSliceSelectorDriver + "=" + driverName}))
})
ginkgo.DeferCleanup(plugin.Stop)
return plugin
}
// createTestObjects creates objects required by the test
// NOTE: as scheduler and controller manager are not running by the Node e2e,
// the objects must contain all required data to be processed correctly by the API server
// and placed on the node without involving the scheduler and the DRA controller
// and placed on the node without involving the scheduler and the DRA controller.
//
// Instead adding more parameters, the podName determines what the pod does.
func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, nodename, namespace, className, claimName, podName string, deferPodDeletion bool, driverNames []string) *v1.Pod {
// DeviceClass
class := &resourceapi.DeviceClass{
@@ -699,6 +881,11 @@ func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, node
RestartPolicy: v1.RestartPolicyNever,
},
}
switch podName {
case "drasleeppod":
// As above, plus infinite sleep.
pod.Spec.Containers[0].Command[2] += "&& sleep 100000"
}
createdPod, err := clientSet.CoreV1().Pods(namespace).Create(ctx, pod, metav1.CreateOptions{})
framework.ExpectNoError(err)
@@ -772,3 +959,28 @@ func createTestResourceSlice(ctx context.Context, clientSet kubernetes.Interface
}
})
}
func listResources(client kubernetes.Interface) func(ctx context.Context) ([]resourceapi.ResourceSlice, error) {
return func(ctx context.Context) ([]resourceapi.ResourceSlice, error) {
slices, err := client.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{})
if err != nil {
return nil, err
}
return slices.Items, nil
}
}
func listAndStoreResources(client kubernetes.Interface, lastSlices *[]resourceapi.ResourceSlice) func(ctx context.Context) ([]resourceapi.ResourceSlice, error) {
return func(ctx context.Context) ([]resourceapi.ResourceSlice, error) {
slices, err := client.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{})
if err != nil {
return nil, err
}
*lastSlices = slices.Items
return *lastSlices, nil
}
}
func matchResourcesByNodeName(nodeName string) types.GomegaMatcher {
return gomega.HaveField("Spec.NodeName", gomega.Equal(nodeName))
}