From 4ee7374b24e6cd788e96fc0362ca5f08b686dc04 Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Wed, 26 Mar 2025 20:05:48 +0200 Subject: [PATCH] DRA kubelet: add connection monitoring This ensures that ResourceSlices get removed also when a plugin becomes unresponsive without removing the registration socket. Tests are from https://github.com/kubernetes/kubernetes/pull/131073 by Ed with some modifications, the implementation is new. --- .../cm/dra/plugin/dra_plugin_manager.go | 117 +++++++- pkg/kubelet/cm/dra/plugin/dra_plugin_test.go | 34 +-- .../cm/dra/plugin/registration_test.go | 242 ++++++++++++----- .../pluginmanager/pluginwatcher/README.md | 52 ++++ .../kubeletplugin/draplugin.go | 73 +++-- test/e2e_node/dra_test.go | 252 ++++++++++++++++-- 6 files changed, 636 insertions(+), 134 deletions(-) diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go index 307edebb35f..50eedce9b1a 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_manager.go @@ -26,6 +26,8 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + grpcstats "google.golang.org/grpc/stats" + v1 "k8s.io/api/core/v1" resourceapi "k8s.io/api/resource/v1beta1" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -58,11 +60,21 @@ type DRAPluginManager struct { getNode func() (*v1.Node, error) wipingDelay time.Duration + // TODO: replace pendingWipes with some kind of workqueue. + // As it stands, WaitGroup suffers from a data race: + // - Queueing a new wiping creates a goroutine and adds to + // to wg. + // - Concurrently, wg.Wait reads from it. + // + // This is not allowed, all wg.Adds must come before wg.Wait. + // + // This race can be triggered with + // go test -count=10 -race ./... wg sync.WaitGroup mutex sync.RWMutex // driver name -> DRAPlugin in the order in which they got added - store map[string][]*DRAPlugin + store map[string][]*monitoredPlugin // pendingWipes maps a driver name to a cancel function for // wiping of that plugin's ResourceSlices. Entries get added @@ -76,6 +88,54 @@ type DRAPluginManager struct { var _ cache.PluginHandler = &DRAPluginManager{} +// monitoredPlugin tracks whether the gRPC connection of a plugin is +// currently connected. Fot that it implements the [grpcstats.Handler] +// interface. +// +// The tagging functions might be useful for contextual logging. But +// for now all that matters is HandleConn. +type monitoredPlugin struct { + *DRAPlugin + pm *DRAPluginManager + + // connected is protected by store.mutex. + connected bool +} + +var _ grpcstats.Handler = &monitoredPlugin{} + +func (m *monitoredPlugin) TagRPC(ctx context.Context, info *grpcstats.RPCTagInfo) context.Context { + return ctx +} + +func (m *monitoredPlugin) HandleRPC(context.Context, grpcstats.RPCStats) { +} + +func (m *monitoredPlugin) TagConn(ctx context.Context, info *grpcstats.ConnTagInfo) context.Context { + return ctx +} + +func (m *monitoredPlugin) HandleConn(_ context.Context, stats grpcstats.ConnStats) { + connected := false + switch stats.(type) { + case *grpcstats.ConnBegin: + connected = true + case *grpcstats.ConnEnd: + // We have to ask for a reconnect, otherwise gRPC wouldn't try and + // thus we wouldn't be notified about a restart of the plugin. + m.conn.Connect() + default: + return + } + + logger := klog.FromContext(m.pm.backgroundCtx) + m.pm.mutex.Lock() + defer m.pm.mutex.Unlock() + logger.V(2).Info("Connection changed", "driverName", m.driverName, "endpoint", m.endpoint, "connected", connected) + m.connected = connected + m.pm.sync(m.driverName) +} + // NewDRAPluginManager creates a new DRAPluginManager, with support for wiping ResourceSlices // when the plugin(s) for a DRA driver are not available too long. // @@ -216,14 +276,29 @@ func (pm *DRAPluginManager) get(driverName string) *DRAPlugin { pm.mutex.RLock() defer pm.mutex.RUnlock() + logger := klog.FromContext(pm.backgroundCtx) + plugins := pm.store[driverName] if len(plugins) == 0 { + logger.V(5).Info("No plugin registered", "driverName", driverName) return nil } + // Heuristic: pick the most recent one. It's most likely // the newest, except when kubelet got restarted and registered // all running plugins in random order. - return plugins[len(plugins)-1] + // + // Prefer plugins which are connected, otherwise also + // disconnected ones. + for i := len(plugins) - 1; i >= 0; i-- { + if plugin := plugins[i]; plugin.connected { + logger.V(5).Info("Preferring connected plugin", "driverName", driverName, "endpoint", plugin.endpoint) + return plugin.DRAPlugin + } + } + plugin := plugins[len(plugins)-1] + logger.V(5).Info("No plugin connected, using latest one", "driverName", driverName, "endpoint", plugin.endpoint) + return plugin.DRAPlugin } // RegisterPlugin implements [cache.PluginHandler]. @@ -259,8 +334,14 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic pm.mutex.Lock() defer pm.mutex.Unlock() + p := &DRAPlugin{ + driverName: driverName, + endpoint: endpoint, + chosenService: chosenService, + clientCallTimeout: clientCallTimeout, + } if pm.store == nil { - pm.store = make(map[string][]*DRAPlugin) + pm.store = make(map[string][]*monitoredPlugin) } for _, oldP := range pm.store[driverName] { if oldP.endpoint == endpoint { @@ -271,6 +352,11 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic logger := klog.FromContext(pm.backgroundCtx) + mp := &monitoredPlugin{ + DRAPlugin: p, + pm: pm, + } + // The gRPC connection gets created once. gRPC then connects to the gRPC server on demand. target := "unix:" + endpoint logger.V(4).Info("Creating new gRPC connection", "target", target) @@ -278,20 +364,20 @@ func (pm *DRAPluginManager) add(driverName string, endpoint string, chosenServic target, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithChainUnaryInterceptor(newMetricsInterceptor(driverName)), + grpc.WithStatsHandler(mp), ) if err != nil { return fmt.Errorf("create gRPC connection to DRA driver %s plugin at endpoint %s: %w", driverName, endpoint, err) } + p.conn = conn - p := &DRAPlugin{ - driverName: driverName, - endpoint: endpoint, - conn: conn, - chosenService: chosenService, - clientCallTimeout: clientCallTimeout, - } + // Ensure that gRPC tries to connect even if we don't call any gRPC method. + // This is necessary to detect early whether a plugin is really available. + // This is currently an experimental gRPC method. Should it be removed we + // would need to do something else, like sending a fake gRPC method call. + conn.Connect() - pm.store[p.driverName] = append(pm.store[p.driverName], p) + pm.store[p.driverName] = append(pm.store[p.driverName], mp) logger.V(3).Info("Registered DRA plugin", "driverName", p.driverName, "endpoint", p.endpoint, "chosenService", p.chosenService, "numPlugins", len(pm.store[p.driverName])) pm.sync(p.driverName) return nil @@ -312,7 +398,7 @@ func (pm *DRAPluginManager) remove(driverName, endpoint string) { defer pm.mutex.Unlock() plugins := pm.store[driverName] - i := slices.IndexFunc(plugins, func(p *DRAPlugin) bool { return p.driverName == driverName && p.endpoint == endpoint }) + i := slices.IndexFunc(plugins, func(mp *monitoredPlugin) bool { return mp.driverName == driverName && mp.endpoint == endpoint }) if i == -1 { return } @@ -382,7 +468,12 @@ func (pm *DRAPluginManager) sync(driverName string) { // usable returns true if at least one endpoint is ready to handle gRPC calls for the DRA driver. // Must be called while holding the mutex. func (pm *DRAPluginManager) usable(driverName string) bool { - return len(pm.store[driverName]) > 0 + for _, mp := range pm.store[driverName] { + if mp.connected { + return true + } + } + return false } // ValidatePlugin implements [cache.PluginHandler]. diff --git a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go index 2644661cd8a..9d506577032 100644 --- a/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go +++ b/pkg/kubelet/cm/dra/plugin/dra_plugin_test.go @@ -21,7 +21,7 @@ import ( "fmt" "net" "os" - "path/filepath" + "path" "strings" "sync" "testing" @@ -56,18 +56,13 @@ func (f *fakeGRPCServer) NodeUnprepareResources(ctx context.Context, in *drapbv1 return &drapbv1beta1.NodeUnprepareResourcesResponse{}, nil } +// tearDown is an idempotent cleanup function. type tearDown func() -func setupFakeGRPCServer(service string) (string, tearDown, error) { - p, err := os.MkdirTemp("", "dra_plugin") - if err != nil { - return "", nil, err - } - - closeCh := make(chan struct{}) - addr := filepath.Join(p, "server.sock") +func setupFakeGRPCServer(service, addr string) (tearDown, error) { + ctx, cancel := context.WithCancel(context.Background()) teardown := func() { - close(closeCh) + cancel() if err := os.RemoveAll(addr); err != nil { panic(err) } @@ -76,7 +71,7 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) { listener, err := net.Listen("unix", addr) if err != nil { teardown() - return "", nil, err + return nil, err } s := grpc.NewServer() @@ -87,7 +82,7 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) { case drapbv1alpha4.NodeService: drapbv1alpha4.RegisterNodeServer(s, drapbv1alpha4.V1Beta1ServerWrapper{DRAPluginServer: fakeGRPCServer}) default: - return "", nil, fmt.Errorf("unsupported gRPC service: %s", service) + return nil, fmt.Errorf("unsupported gRPC service: %s", service) } go func() { @@ -96,17 +91,18 @@ func setupFakeGRPCServer(service string) (string, tearDown, error) { panic(err) } }() - <-closeCh + <-ctx.Done() s.GracefulStop() }() - return addr, teardown, nil + return teardown, nil } func TestGRPCConnIsReused(t *testing.T) { tCtx := ktesting.Init(t) service := drapbv1beta1.DRAPluginService - addr, teardown, err := setupFakeGRPCServer(service) + addr := path.Join(t.TempDir(), "dra.sock") + teardown, err := setupFakeGRPCServer(service, addr) if err != nil { t.Fatal(err) } @@ -212,27 +208,23 @@ func TestGetDRAPlugin(t *testing.T) { func TestGRPCMethods(t *testing.T) { for _, test := range []struct { description string - serverSetup func(string) (string, tearDown, error) service string chosenService string expectError string }{ { description: "v1alpha4", - serverSetup: setupFakeGRPCServer, service: drapbv1alpha4.NodeService, chosenService: drapbv1alpha4.NodeService, }, { description: "v1beta1", - serverSetup: setupFakeGRPCServer, service: drapbv1beta1.DRAPluginService, chosenService: drapbv1beta1.DRAPluginService, }, { // In practice, such a mismatch between plugin and kubelet should not happen. description: "mismatch", - serverSetup: setupFakeGRPCServer, service: drapbv1beta1.DRAPluginService, chosenService: drapbv1alpha4.NodeService, expectError: "unknown service v1alpha3.Node", @@ -240,7 +232,6 @@ func TestGRPCMethods(t *testing.T) { { // In practice, kubelet wouldn't choose an invalid service. description: "internal-error", - serverSetup: setupFakeGRPCServer, service: drapbv1beta1.DRAPluginService, chosenService: "some-other-service", expectError: "unsupported chosen service", @@ -248,7 +239,8 @@ func TestGRPCMethods(t *testing.T) { } { t.Run(test.description, func(t *testing.T) { tCtx := ktesting.Init(t) - addr, teardown, err := setupFakeGRPCServer(test.service) + addr := path.Join(t.TempDir(), "dra.sock") + teardown, err := setupFakeGRPCServer(test.service, addr) if err != nil { t.Fatal(err) } diff --git a/pkg/kubelet/cm/dra/plugin/registration_test.go b/pkg/kubelet/cm/dra/plugin/registration_test.go index 8012a060799..6cd505df934 100644 --- a/pkg/kubelet/cm/dra/plugin/registration_test.go +++ b/pkg/kubelet/cm/dra/plugin/registration_test.go @@ -17,11 +17,13 @@ limitations under the License. package plugin import ( + "path" "sort" "strings" "testing" "time" + "github.com/onsi/gomega" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -39,17 +41,78 @@ import ( ) const ( - nodeName = "worker" - pluginA = "pluginA" - endpointA = "endpointA" - pluginB = "pluginB" - endpointB = "endpointB" + nodeName = "worker" + pluginA = "pluginA" + pluginB = "pluginB" ) func getFakeNode() (*v1.Node, error) { return &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: nodeName}}, nil } +func getSlice(name string) *resourceapi.ResourceSlice { + return &resourceapi.ResourceSlice{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: resourceapi.ResourceSliceSpec{ + NodeName: nodeName, + }, + } +} + +func getFakeClient(t *testing.T, nodeName, driverName string, slice *resourceapi.ResourceSlice) kubernetes.Interface { + expectedSliceFields := fields.Set{"spec.nodeName": nodeName} + fakeClient := fake.NewClientset(slice) + fakeClient.AddReactor("delete-collection", "resourceslices", func(action cgotesting.Action) (bool, runtime.Object, error) { + deleteAction := action.(cgotesting.DeleteCollectionAction) + restrictions := deleteAction.GetListRestrictions() + fieldsSelector := fields.SelectorFromSet(expectedSliceFields) + // The order of field requirements is random because it comes + // from a map. We need to sort. + normalize := func(selector string) string { + requirements := strings.Split(selector, ",") + sort.Strings(requirements) + return strings.Join(requirements, ",") + } + assert.Empty(t, restrictions.Labels.String(), "label selector in DeleteCollection") + assert.Equal(t, normalize(fieldsSelector.String()), normalize(restrictions.Fields.String()), "field selector in DeleteCollection") + + // There's only one object that could get matched, so delete it. + // Delete doesn't return an error if already deleted, which is what + // we need here (no error when nothing to delete). + err := fakeClient.Tracker().Delete(resourceapi.SchemeGroupVersion.WithResource("resourceslices"), "", slice.Name) + + // Set expected slice fields for the next call of this reactor. + // The reactor will be called next time when resourceslices object is deleted + // by the kubelet after plugin deregistration. + switch len(expectedSliceFields) { + case 1: + // Startup cleanup done, now expect cleanup for test plugin. + expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": driverName} + case 2: + // Test plugin cleanup done, now expect cleanup for the other plugin. + otherPlugin := pluginA + if otherPlugin == driverName { + otherPlugin = pluginB + } + expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": otherPlugin} + } + return true, nil, err + }) + return fakeClient +} + +func requireNoSlices(tCtx ktesting.TContext) { + tCtx.Helper() + ktesting.Eventually(tCtx, func(tCtx ktesting.TContext) error { + slices, err := tCtx.Client().ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{}) + if err != nil { + return err + } + assert.Empty(tCtx, slices.Items, "slices") + return nil + }).Should(gomega.Succeed(), "there should be no slices") +} + func TestRegistrationHandler(t *testing.T) { slice := &resourceapi.ResourceSlice{ ObjectMeta: metav1.ObjectMeta{Name: "test-slice"}, @@ -58,6 +121,10 @@ func TestRegistrationHandler(t *testing.T) { }, } + tmp := t.TempDir() + endpointA := path.Join(tmp, "dra-plugin-a.sock") + endpointB := path.Join(tmp, "dra-plugin-b.sock") + for _, test := range []struct { description string driverName string @@ -121,71 +188,43 @@ func TestRegistrationHandler(t *testing.T) { t.Run(test.description, func(t *testing.T) { tCtx := ktesting.Init(t) + // Run GRPC services for both plugins. + // + // This is necessary because otherwise connection + // monitoring will start wiping slices, regardless + // of whether the plugin is registered or not. + // + // Here we are only interested in registration. + // In TestConnectionHandling we check detection + // of the connection state. + + service := drapb.DRAPluginService + teardownA, err := setupFakeGRPCServer(service, endpointA) + require.NoError(t, err) + tCtx.Cleanup(teardownA) + + teardown, err := setupFakeGRPCServer(service, test.endpoint) + require.NoError(t, err) + tCtx.Cleanup(teardown) + // Stand-alone kubelet has no connection to an // apiserver, so faking one is optional. var client kubernetes.Interface if test.withClient { - expectedSliceFields := fields.Set{"spec.nodeName": nodeName} - fakeClient := fake.NewClientset(slice) - fakeClient.AddReactor("delete-collection", "resourceslices", func(action cgotesting.Action) (bool, runtime.Object, error) { - deleteAction := action.(cgotesting.DeleteCollectionAction) - restrictions := deleteAction.GetListRestrictions() - fieldsSelector := fields.SelectorFromSet(expectedSliceFields) - // The order of field requirements is random because it comes - // from a map. We need to sort. - normalize := func(selector string) string { - requirements := strings.Split(selector, ",") - sort.Strings(requirements) - return strings.Join(requirements, ",") - } - assert.Equal(t, "", restrictions.Labels.String(), "label selector in DeleteCollection") - assert.Equal(t, normalize(fieldsSelector.String()), normalize(restrictions.Fields.String()), "field selector in DeleteCollection") - - // There's only one object that could get matched, so delete it. - // Delete doesn't return an error if already deleted, which is what - // we need here (no error when nothing to delete). - err := fakeClient.Tracker().Delete(resourceapi.SchemeGroupVersion.WithResource("resourceslices"), "", slice.Name) - - // Set expected slice fields for the next call of this reactor. - // The reactor will be called next time when resourceslices object is deleted - // by the kubelet after plugin deregistration. - switch len(expectedSliceFields) { - case 1: - // Startup cleanup done, now expect cleanup for test plugin. - expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": test.driverName} - case 2: - // Test plugin cleanup done, now expect cleanup for the other plugin. - otherPlugin := pluginA - if otherPlugin == test.driverName { - otherPlugin = pluginB - } - expectedSliceFields = fields.Set{"spec.nodeName": nodeName, "spec.driver": otherPlugin} - } - return true, nil, err - }) + fakeClient := getFakeClient(t, nodeName, test.driverName, getSlice("test-slice")) client = fakeClient + tCtx = ktesting.WithClients(tCtx, nil, nil, client, nil, nil) } // The DRAPluginManager wipes all slices at startup. draPlugins := NewDRAPluginManager(tCtx, client, getFakeNode, time.Second /* very short wiping delay for testing */) tCtx.Cleanup(draPlugins.Stop) - requireNoSlices := func() { - t.Helper() - if client == nil { - return - } - require.EventuallyWithT(t, func(t *assert.CollectT) { - slices, err := client.ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{}) - if !assert.NoError(t, err, "list slices") { - return - } - assert.Empty(t, slices.Items, "slices") - }, time.Minute, time.Second) + if test.withClient { + requireNoSlices(tCtx) } - requireNoSlices() // Simulate one existing plugin A. - err := draPlugins.RegisterPlugin(pluginA, endpointA, []string{drapb.DRAPluginService}, nil) + err = draPlugins.RegisterPlugin(pluginA, endpointA, []string{drapb.DRAPluginService}, nil) require.NoError(t, err) t.Cleanup(func() { tCtx.Logf("Removing plugin %s", pluginA) @@ -225,11 +264,94 @@ func TestRegistrationHandler(t *testing.T) { draPlugins.DeRegisterPlugin(test.driverName, test.endpoint) // Nop. draPlugins.DeRegisterPlugin(test.driverName, test.endpoint) - - requireNoSlices() + if test.withClient { + requireNoSlices(tCtx) + } }) - assert.Equal(t, test.endpoint, plugin.endpoint, "plugin endpoint") + // Which plugin was chosen is random in this test: it depends on which plugin was detected as connected, + // which can be both, one, or none at this point. Some attributes are common to both. + assert.Equal(t, test.driverName, plugin.driverName, "DRA driver driver name") assert.Equal(t, test.chosenService, plugin.chosenService, "chosen service") }) } } + +// TestConnectionHandling checks the reaction to state changes of the service connection. +func TestConnectionHandling(t *testing.T) { + t.Parallel() + for description, test := range map[string]struct { + delay time.Duration + requireSliceRemoval bool + }{ + "wipe-slices-on-disconnect": { + delay: time.Second, // very short wiping delay for testing + requireSliceRemoval: true, + }, + "no-wipe-slices-on-reconnect": { + delay: time.Hour, // long delay to avoid wiping while the test runs + requireSliceRemoval: false, + }, + } { + t.Run(description, func(t *testing.T) { + t.Parallel() + tCtx := ktesting.Init(t) + + service := drapb.DRAPluginService + driverName := "test-plugin" + sliceName := "test-slice" + + slice := getSlice(sliceName) + client := getFakeClient(t, nodeName, driverName, slice) + tCtx = ktesting.WithClients(tCtx, nil, nil, client, nil, nil) + + // The handler wipes all slices at startup. + draPlugins := NewDRAPluginManager(tCtx, client, getFakeNode, test.delay) + tCtx.Cleanup(draPlugins.Stop) + requireNoSlices(tCtx) + + // Run GRPC service. + endpoint := path.Join(t.TempDir(), "dra-plugin-test.sock") + teardown, err := setupFakeGRPCServer(service, endpoint) + require.NoError(t, err) + defer teardown() + + err = draPlugins.RegisterPlugin(driverName, endpoint, []string{service}, nil) + require.NoError(t, err) + + plugin := draPlugins.get(driverName) + assert.NotNil(t, plugin, "plugin should be present in the plugin store") + + // Create the slice as if the plugin had done that while it runs. + _, err = client.ResourceV1beta1().ResourceSlices().Create(tCtx, slice, metav1.CreateOptions{}) + require.NoError(t, err, "recreate slice") + + // Stop gRPC server. + tCtx.Log("Stopping plugin gRPC server") + teardown() + + if test.requireSliceRemoval { + // Slice should get removed. + requireNoSlices(tCtx) + } else { + // Start up gRPC server again. + tCtx.Log("Restarting plugin gRPC server") + teardown, err = setupFakeGRPCServer(service, endpoint) + require.NoError(t, err) + defer teardown() + + // There shouldn't be any pending wipes for the plugin. + require.Eventuallyf(t, func() bool { + draPlugins.mutex.Lock() + defer draPlugins.mutex.Unlock() + _, ok := draPlugins.pendingWipes[driverName] + return ok + }, time.Minute, time.Second, "wiping should be stopped for plugin %s", driverName) + + // Slice should still be there + slices, err := client.ResourceV1beta1().ResourceSlices().List(tCtx, metav1.ListOptions{}) + require.NoError(t, err, "list slices") + assert.Len(t, slices.Items, 1, "slices") + } + }) + } +} diff --git a/pkg/kubelet/pluginmanager/pluginwatcher/README.md b/pkg/kubelet/pluginmanager/pluginwatcher/README.md index 9403829a2fb..9fe366923a1 100644 --- a/pkg/kubelet/pluginmanager/pluginwatcher/README.md +++ b/pkg/kubelet/pluginmanager/pluginwatcher/README.md @@ -22,6 +22,58 @@ should end with a DNS domain that is unique for the plugin. Each time a plugin starts, it has to delete old sockets if they exist and listen anew under the same filename. +## Monitoring Plugin Connection + +**Warning**: Monitoring the plugin connection is only supported +for DRA at the moment. + +The Kubelet monitors the gRPC connection to a plugin's **service socket** using +a [gRPC stats handler](https://github.com/grpc/grpc-go/blob/master/examples/features/stats_monitoring/README.md). + +This enables the Kubelet to: +- Detect when the plugin process has crashed, exited, or restarted +- Trigger cleanup of the plugin’s resources on connection drop +- Cancel pending cleanup if the connection is restored + + +The **registration socket** is used by the plugin manager. It registers the +plugin when registration socket is created by the plugin and the GetInfo gRPC +call succeeds. It deregisters the plugin when the socket is removed. +The plugin should be ready to handle gRPC requests over the **service socket** +that it returned in response to the GetInfo call because the kubelet might try +to use the service immediately. Monitoring this service socket is therefore +more accurate for detecting the real availability of the plugin. + +### How It Works + +Internally, the plugin client configures a gRPC stats handler to observe +`ConnBegin` and `ConnEnd` events on the service socket connection. The +connection is established over a Unix Domain Socket, which provides reliable +semantics - if the connection drops, it definitively indicates that the plugin +closed its end (e.g., due to crash or shutdown). + +1. During plugin registration, the Kubelet connects to plugin's + **service socket** and attaches the stats handler. +2. A long-lived gRPC connection is established and actively monitored. +3. If the plugin process exits and the connection drops, the stats + handler observes a `ConnEnd` event. +4. This triggers a check whether cleanup is necessary: as long as at + least one gRPC connection is connected, the plugin is usable and + no cleanup is required. +5. A gRPC reconnect is initiated immediately after connection loss. +6. When the plugin resumes serving on the same service socket, the connection + is re-established and a `ConnBegin` event is observed. +7. This cancels any in-progress resource cleanup and restores communication. + +### Key Properties + +- The plugin is **not** deregistered when the connection drops. +- This model supports multi-container plugin deployments (e.g., CSI-style + sidecar setups) where the service container may restart independently of + the registrar container. +- Cleanup is only executed if the connection is not restored before a + grace period expires. + ## Seamless Upgrade To avoid downtime of a plugin on a node, it would be nice to support running an diff --git a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go index 97ec6842105..4f3ae832c28 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go +++ b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go @@ -397,6 +397,26 @@ func FlockDirectoryPath(path string) Option { } } +// RegistrationService controls whether the kubelet plugin gRPC service +// is started. It's on by default. This is meant for testing, normal +// DRA drivers should use the default. +func RegistrationService(enabled bool) Option { + return func(o *options) error { + o.registrationService = enabled + return nil + } +} + +// DRAService controls whether the DRA gRPC service +// is started. It's on by default. This is meant for testing, normal +// DRA drivers should use the default. +func DRAService(enabled bool) Option { + return func(o *options) error { + o.draService = enabled + return nil + } +} + type options struct { logger klog.Logger grpcVerbosity int @@ -413,6 +433,8 @@ type options struct { serialize bool flockDirectoryPath string nodeV1beta1 bool + registrationService bool + draService bool } // Helper combines the kubelet registration service and the DRA node plugin @@ -442,8 +464,9 @@ type Helper struct { resourceSliceController *resourceslice.Controller } -// Start sets up two gRPC servers (one for registration, one for the DRA node -// client) and implements them by calling a [DRAPlugin] implementation. +// Start sets up all enabled gRPC servers (by default, one for registration, +// one for the DRA node client) and implements them by calling a [DRAPlugin] +// implementation. // // The context and/or DRAPlugin.Stop can be used to stop all background activity. // Stop also blocks. A logger can be stored in the context to add values or @@ -462,6 +485,8 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe pluginRegistrationEndpoint: endpoint{ dir: KubeletRegistryDir, }, + draService: true, + registrationService: true, } for _, option := range opts { if err := option(&o); err != nil { @@ -530,34 +555,42 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe } }() - // Run the node plugin gRPC server first to ensure that it is ready. var supportedServices []string + if o.nodeV1beta1 { + logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service") + supportedServices = append(supportedServices, drapb.DRAPluginService) + } + if len(supportedServices) == 0 { + return nil, errors.New("no supported DRA gRPC API is implemented and enabled") + } draEndpoint := endpoint{ dir: o.pluginDataDirectoryPath, file: "dra" + uidPart + ".sock", // "dra" is hard-coded. The directory is unique, so we get a unique full path also without the UID. listenFunc: o.draEndpointListen, } - pluginServer, err := startGRPCServer(klog.LoggerWithName(logger, "dra"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, draEndpoint, func(grpcServer *grpc.Server) { - if o.nodeV1beta1 { - logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service") - drapb.RegisterDRAPluginServer(grpcServer, &nodePluginImplementation{Helper: d}) - supportedServices = append(supportedServices, drapb.DRAPluginService) + + if o.draService { + // Run the node plugin gRPC server first to ensure that it is ready. + pluginServer, err := startGRPCServer(klog.LoggerWithName(logger, "dra"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, draEndpoint, func(grpcServer *grpc.Server) { + if o.nodeV1beta1 { + logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service") + drapb.RegisterDRAPluginServer(grpcServer, &nodePluginImplementation{Helper: d}) + } + }) + if err != nil { + return nil, fmt.Errorf("start DRA service: %w", err) } - }) - if err != nil { - return nil, fmt.Errorf("start node client: %v", err) - } - d.pluginServer = pluginServer - if len(supportedServices) == 0 { - return nil, errors.New("no supported DRA gRPC API is implemented and enabled") + d.pluginServer = pluginServer } - // Now make it available to kubelet. - registrar, err := startRegistrar(klog.LoggerWithName(logger, "registrar"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, draEndpoint.path(), o.pluginRegistrationEndpoint) - if err != nil { - return nil, fmt.Errorf("start registrar: %v", err) + if o.registrationService { + // Now make it available to kubelet. + registrar, err := startRegistrar(klog.LoggerWithName(logger, "registrar"), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, draEndpoint.path(), o.pluginRegistrationEndpoint) + if err != nil { + return nil, fmt.Errorf("start registrar: %w", err) + } + d.registrar = registrar } - d.registrar = registrar // startGRPCServer and startRegistrar don't implement cancellation // themselves, we add that for both here. diff --git a/test/e2e_node/dra_test.go b/test/e2e_node/dra_test.go index 09ade51d071..57a59dc1c6b 100644 --- a/test/e2e_node/dra_test.go +++ b/test/e2e_node/dra_test.go @@ -378,6 +378,62 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami return kubeletPlugin.CountCalls("/NodePrepareResources") }).WithTimeout(retryTestTimeout).Should(gomega.Equal(calls)) }) + + ginkgo.It("must be functional when plugin starts to listen on a service socket after registration", func(ctx context.Context) { + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, getNodeName(ctx, f), driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + draService := newDRAService(ctx, f.ClientSet, getNodeName(ctx, f), driverName) + + pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drapod", false, []string{driverName}) + + ginkgo.By("wait for NodePrepareResources call to succeed") + gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded) + + ginkgo.By("wait for pod to succeed") + err := e2epod.WaitForPodSuccessInNamespace(ctx, f.ClientSet, pod.Name, f.Namespace.Name) + framework.ExpectNoError(err) + }) + + ginkgo.It("must be functional after reconnect", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + draService := newDRAService(ctx, f.ClientSet, nodeName, driverName) + + pod := createTestObjects(ctx, f.ClientSet, getNodeName(ctx, f), f.Namespace.Name, "draclass", "external-claim", "drasleeppod" /* enables sleeping */, false /* pod is deleted below */, []string{driverName}) + + ginkgo.By("wait for NodePrepareResources call to succeed") + gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded) + + ginkgo.By("stop plugin") + draService.Stop() + + ginkgo.By("waiting for pod to run") + err := e2epod.WaitForPodRunningInNamespace(ctx, f.ClientSet, pod) + framework.ExpectNoError(err) + + ginkgo.By("wait for ResourceSlice removal, indicating detection of disconnect") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin") + + ginkgo.By("restarting plugin") + draService = newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("stopping pod") + err = f.ClientSet.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{}) + framework.ExpectNoError(err) + gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodeUnprepareResourcesSucceeded) + }) }) f.Context("Two resource Kubelet Plugins", f.WithSerial(), func() { @@ -513,18 +569,6 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami }) f.Context("ResourceSlice", f.WithSerial(), func() { - listResources := func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { - slices, err := f.ClientSet.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{}) - if err != nil { - return nil, err - } - return slices.Items, nil - } - - matchResourcesByNodeName := func(nodeName string) types.GomegaMatcher { - return gomega.HaveField("Spec.NodeName", gomega.Equal(nodeName)) - } - f.It("must be removed on kubelet startup", f.WithDisruptive(), func(ctx context.Context) { ginkgo.By("stop kubelet") restartKubelet := mustStopKubelet(ctx, f) @@ -543,15 +587,15 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami matchAll := gomega.ConsistOf(matchResourcesByNodeName(nodeName), matchResourcesByNodeName(otherNodeName)) matchOtherNode := gomega.ConsistOf(matchResourcesByNodeName(otherNodeName)) - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchAll, "ResourceSlices without kubelet") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchAll, "ResourceSlices without kubelet") ginkgo.By("restart kubelet") restartKubelet(ctx) restartKubelet = nil ginkgo.By("wait for exactly the node's ResourceSlice to get deleted") - gomega.Eventually(ctx, listResources).Should(matchOtherNode, "ResourceSlices with kubelet") - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchOtherNode, "ResourceSlices with kubelet") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchOtherNode, "ResourceSlices with kubelet") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchOtherNode, "ResourceSlices with kubelet") }) f.It("must be removed after plugin unregistration", func(ctx context.Context) { @@ -560,13 +604,89 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami ginkgo.By("start plugin and wait for ResourceSlice") kubeletPlugin := newKubeletPlugin(ctx, f.ClientSet, getNodeName(ctx, f), driverName) - gomega.Eventually(ctx, listResources).Should(matchNode, "ResourceSlice from kubelet plugin") - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlice from kubelet plugin") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlice from kubelet plugin") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlice from kubelet plugin") ginkgo.By("stop plugin and wait for ResourceSlice removal") kubeletPlugin.Stop() - gomega.Eventually(ctx, listResources).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") - gomega.Consistently(ctx, listResources).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices with no plugin") + }) + + f.It("must be removed if plugin stops after registration", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for ResourceSlice to be created by plugin") + matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName)) + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlices") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlices") + + ginkgo.By("stop plugin") + kubeletPlugin.Stop() + + ginkgo.By("wait for ResourceSlice removal") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices") + }) + + f.It("must be removed if plugin is unresponsive after registration", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("create a ResourceSlice") + createTestResourceSlice(ctx, f.ClientSet, nodeName, driverName) + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.ConsistOf(matchResourcesByNodeName(nodeName)), "ResourceSlices without plugin") + + ginkgo.By("wait for ResourceSlice removal") + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(gomega.BeEmpty(), "ResourceSlices without plugin") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(5*time.Second).Should(gomega.BeEmpty(), "ResourceSlices without plugin") + }) + + f.It("must not be removed if plugin restarts quickly enough", func(ctx context.Context) { + nodeName := getNodeName(ctx, f) + + ginkgo.By("start DRA registrar") + registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for registration to complete") + gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered) + + ginkgo.By("start DRA plugin service") + kubeletPlugin := newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("wait for ResourceSlice to be created by plugin") + matchNode := gomega.ConsistOf(matchResourcesByNodeName(nodeName)) + gomega.Eventually(ctx, listResources(f.ClientSet)).Should(matchNode, "ResourceSlices") + var slices []resourceapi.ResourceSlice + gomega.Consistently(ctx, listAndStoreResources(f.ClientSet, &slices)).WithTimeout(5*time.Second).Should(matchNode, "ResourceSlices") + + ginkgo.By("stop plugin") + kubeletPlugin.Stop() + + // We know from the "must be removed if plugin is unresponsive after registration" that the kubelet + // eventually notices the dropped connection. We cannot observe when that happens, we would need + // a new metric for that ("registered DRA plugins"). Let's give it a few seconds, which is significantly + // less than the wiping delay. + time.Sleep(5 * time.Second) + + ginkgo.By("restarting plugin") + kubeletPlugin = newDRAService(ctx, f.ClientSet, nodeName, driverName) + + ginkgo.By("ensuring unchanged ResourceSlices") + gomega.Consistently(ctx, listResources(f.ClientSet)).WithTimeout(time.Minute).Should(gomega.Equal(slices), "ResourceSlices") }) }) }) @@ -622,10 +742,72 @@ func newKubeletPlugin(ctx context.Context, clientSet kubernetes.Interface, nodeN return plugin } +// newRegistrar starts a registrar for the specified DRA driver, without the DRA gRPC service. +func newRegistrar(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin { + ginkgo.By("start only Kubelet plugin registrar") + logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin registrar "+driverName)) + ctx = klog.NewContext(ctx, logger) + registrar, err := testdriver.StartPlugin(ctx, cdiDir, driverName, clientSet, nodeName, testdriver.FileOperations{}, kubeletplugin.DRAService(false)) + framework.ExpectNoError(err, "start only Kubelet plugin registrar") + return registrar +} + +// newDRAService starts the DRA gRPC service for the specified DRA driver, without the registrar. +func newDRAService(ctx context.Context, clientSet kubernetes.Interface, nodeName, driverName string) *testdriver.ExamplePlugin { + ginkgo.By("start only Kubelet plugin") + logger := klog.LoggerWithValues(klog.LoggerWithName(klog.Background(), "kubelet plugin "+driverName), "node", nodeName) + ctx = klog.NewContext(ctx, logger) + + // Ensure that directories exist, creating them if necessary. We want + // to know early if there is a setup problem that would prevent + // creating those directories. + err := os.MkdirAll(cdiDir, os.FileMode(0750)) + framework.ExpectNoError(err, "create CDI directory") + datadir := path.Join(kubeletplugin.KubeletPluginsDir, driverName) // The default, not set below. + err = os.MkdirAll(datadir, 0750) + framework.ExpectNoError(err, "create DRA socket directory") + + plugin, err := testdriver.StartPlugin( + ctx, + cdiDir, + driverName, + clientSet, + nodeName, + testdriver.FileOperations{ + DriverResources: &resourceslice.DriverResources{ + Pools: map[string]resourceslice.Pool{ + nodeName: { + Slices: []resourceslice.Slice{{ + Devices: []resourceapiv1beta2.Device{ + { + Name: "device-00", + }, + }, + }}, + }, + }, + }, + }, + kubeletplugin.RegistrationService(false), + ) + framework.ExpectNoError(err) + + ginkgo.DeferCleanup(func(ctx context.Context) { + // kubelet should do this eventually, but better make sure. + // A separate test checks this explicitly. + framework.ExpectNoError(clientSet.ResourceV1beta1().ResourceSlices().DeleteCollection(ctx, metav1.DeleteOptions{}, metav1.ListOptions{FieldSelector: resourceapi.ResourceSliceSelectorDriver + "=" + driverName})) + }) + ginkgo.DeferCleanup(plugin.Stop) + + return plugin +} + // createTestObjects creates objects required by the test // NOTE: as scheduler and controller manager are not running by the Node e2e, // the objects must contain all required data to be processed correctly by the API server -// and placed on the node without involving the scheduler and the DRA controller +// and placed on the node without involving the scheduler and the DRA controller. +// +// Instead adding more parameters, the podName determines what the pod does. func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, nodename, namespace, className, claimName, podName string, deferPodDeletion bool, driverNames []string) *v1.Pod { // DeviceClass class := &resourceapi.DeviceClass{ @@ -699,6 +881,11 @@ func createTestObjects(ctx context.Context, clientSet kubernetes.Interface, node RestartPolicy: v1.RestartPolicyNever, }, } + switch podName { + case "drasleeppod": + // As above, plus infinite sleep. + pod.Spec.Containers[0].Command[2] += "&& sleep 100000" + } createdPod, err := clientSet.CoreV1().Pods(namespace).Create(ctx, pod, metav1.CreateOptions{}) framework.ExpectNoError(err) @@ -772,3 +959,28 @@ func createTestResourceSlice(ctx context.Context, clientSet kubernetes.Interface } }) } + +func listResources(client kubernetes.Interface) func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + return func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + slices, err := client.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, err + } + return slices.Items, nil + } +} + +func listAndStoreResources(client kubernetes.Interface, lastSlices *[]resourceapi.ResourceSlice) func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + return func(ctx context.Context) ([]resourceapi.ResourceSlice, error) { + slices, err := client.ResourceV1beta1().ResourceSlices().List(ctx, metav1.ListOptions{}) + if err != nil { + return nil, err + } + *lastSlices = slices.Items + return *lastSlices, nil + } +} + +func matchResourcesByNodeName(nodeName string) types.GomegaMatcher { + return gomega.HaveField("Spec.NodeName", gomega.Equal(nodeName)) +}