DRA kubeletplugin: revise socket path handling

When supporting rolling updates, we cannot use the same fixed socket paths for
old and new pod. With the revised API, the caller no longer specifies the full
socket paths, only directories. The logic about how to name sockets then can be
in the helper.

While at it, avoid passing a context to the gRPC helper code when
all that the helper code needs is a logger. That leads to confusion
about whether cancellation has an effect.

Author: pohly

staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/doc.go

@@ -16,4 +16,52 @@ limitations under the License.
 
 // Package kubeletplugin provides helper functions for running a dynamic
 // resource allocation kubelet plugin.
+//
+// A DRA driver using this package can be deployed as a DaemonSet on suitable
+// nodes. Node labeling, for example through NFD
+// (https://github.com/kubernetes-sigs/node-feature-discovery), can be used
+// to run the driver only on nodes which have the necessary hardware.
+//
+// The service account of the DaemonSet must have sufficient RBAC permissions
+// to read ResourceClaims and to create and update ResourceSlices, if
+// the driver intends to publish per-node ResourceSlices. It is good
+// security practice (but not required) to limit access to ResourceSlices
+// associated with the node a specific Pod is running on. This can be done
+// with a Validating Admission Policy (VAP). For more information,
+// see the deployment of the DRA example driver
+// (https://github.com/kubernetes-sigs/dra-example-driver/tree/main/deployments/helm/dra-example-driver/templates).
+//
+// Traditionally, the kubelet has not supported rolling updates of plugins.
+// Therefore the DaemonSet must not set `maxSurge` to a value larger than
+// zero. With the default `maxSurge: 0`, updating the DaemonSet of the driver
+// will first shut down the old driver Pod, then start the replacement.
+//
+// This leads to a short downtime for operations that need the driver:
+//   - Pods cannot start unless the claims they depend on were already
+//     prepared for use.
+//   - Cleanup after the last pod which used a claim gets delayed
+//     until the driver is available again. The pod is not marked
+//     as terminated. This prevents reusing the resources used by
+//     the pod for other pods.
+//   - Running pods are *not* affected as far as Kubernetes is
+//     concerned. However, a DRA driver might provide required runtime
+//     services. Vendors need to document this.
+//
+// Note that the second point also means that draining a node should
+// first evict normal pods, then the driver DaemonSet Pod.
+//
+// Starting with Kubernetes 1.33, the kubelet supports rolling updates
+// such that old and new Pod run at the same time for a short while
+// and hand over work gracefully, with no downtime.
+// However, there is no mechanism for determining in advance whether
+// the node the DaemonSet runs on supports that. Trying
+// to do a rolling update with a kubelet which does not support it yet
+// will fail because shutting down the old Pod unregisters the driver
+// even though the new Pod is running. See https://github.com/kubernetes/kubernetes/pull/129832
+// for details (TODO: link to doc after merging instead).
+//
+// A DRA driver can either require 1.33 as minimal Kubernetes version or
+// provide two variants of its DaemonSet deployment. In the variant with
+// support for rolling updates, `maxSurge` can be set to a non-zero
+// value. Administrators have to be careful about running the right variant.
 package kubeletplugin

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

@@ -21,6 +21,7 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"path"
 	"sync"
 
 	"google.golang.org/grpc"
@@ -36,6 +37,13 @@ import (
 	registerapi "k8s.io/kubelet/pkg/apis/pluginregistration/v1"
 )
 
+const (
+	// KubeletPluginsDir is the default directory for [PluginDataDirectoryPath].
+	KubeletPluginsDir = "/var/lib/kubelet/plugins"
+	// KubeletRegistryDir is the default for [RegistrarDirectoryPath]
+	KubeletRegistryDir = "/var/lib/kubelet/plugins_registry"
+)
+
 // DRAPlugin is the interface that needs to be implemented by a DRA driver to
 // use this helper package. The helper package then implements the gRPC
 // interface expected by the kubelet by wrapping the DRAPlugin implementation.
@@ -165,63 +173,77 @@ func GRPCVerbosity(level int) Option {
 	}
 }
 
-// RegistrarSocketPath sets the file path for a Unix domain socket.
-// If RegistrarListener is not used, then Start will remove
-// a file at that path, should one exist, and creates a socket
-// itself. Otherwise it uses the provided listener and only
-// removes the socket at the specified path during shutdown.
+// RegistrarDirectoryPath sets the path to the directory where the kubelet
+// expects to find registration sockets of plugins. Typically this is
+// /var/lib/kubelet/plugins_registry with /var/lib/kubelet being the kubelet's
+// data directory.
 //
-// At least one of these two options is required.
-func RegistrarSocketPath(path string) Option {
+// This is also the default. Some Kubernetes clusters may use a different data directory.
+// This path must be the same inside and outside of the driver's container.
+// The directory must exist.
+func RegistrarDirectoryPath(path string) Option {
 	return func(o *options) error {
-		o.pluginRegistrationEndpoint.path = path
+		o.pluginRegistrationEndpoint.dir = path
 		return nil
 	}
 }
 
-// RegistrarListener sets an already created listener for the plugin
-// registration API. Can be combined with RegistrarSocketPath.
+// RegistrarSocketFilename sets the name of the socket inside the directory where
+// the kubelet watches for registration sockets (see RegistrarDirectoryPath).
+//
+// Usually DRA drivers should not need this option. It is provided to
+// support updates from an installation which used an older release of
+// of the helper code.
 //
-// At least one of these two options is required.
-func RegistrarListener(listener net.Listener) Option {
+// The default is <driver name>-reg.sock. When rolling updates are enabled (not supported yet),
+// it is <driver name>-<uid>-reg.sock.
+func RegistrarSocketFilename(name string) Option {
 	return func(o *options) error {
-		o.pluginRegistrationEndpoint.listener = listener
+		o.pluginRegistrationEndpoint.file = name
 		return nil
 	}
 }
 
-// PluginSocketPath sets the file path for a Unix domain socket.
-// If PluginListener is not used, then Start will remove
-// a file at that path, should one exist, and creates a socket
-// itself. Otherwise it uses the provided listener and only
-// removes the socket at the specified path during shutdown.
+// RegistrarListener configures how to create the registrar socket.
+// The default is to remove the file if it exists and to then
+// create a socket.
 //
-// At least one of these two options is required.
-func PluginSocketPath(path string) Option {
+// This is used in Kubernetes for end-to-end testing. The default should
+// be fine for DRA drivers.
+func RegistrarListener(listen func(ctx context.Context, path string) (net.Listener, error)) Option {
 	return func(o *options) error {
-		o.draEndpoint.path = path
+		o.pluginRegistrationEndpoint.listenFunc = listen
 		return nil
 	}
 }
 
-// PluginListener sets an already created listener for the dynamic resource
-// allocation plugin API. Can be combined with PluginSocketPath.
+// PluginDataDirectoryPath sets the path where the DRA driver creates the
+// "dra.sock" socket that the kubelet connects to for the DRA-specific gRPC calls.
+// It is also used to coordinate between different Pods when using rolling
+// updates. It must not be shared with other kubelet plugins.
+//
+// The default is /var/lib/kubelet/plugins/<driver name>. This directory
+// does not need to be inside the kubelet data directory, as long as
+// the kubelet can access it.
 //
-// At least one of these two options is required.
-func PluginListener(listener net.Listener) Option {
+// This path must be the same inside and outside of the driver's container.
+// The directory must exist.
+func PluginDataDirectoryPath(path string) Option {
 	return func(o *options) error {
-		o.draEndpoint.listener = listener
+		o.pluginDataDirectoryPath = path
 		return nil
 	}
 }
 
-// KubeletPluginSocketPath defines how kubelet will connect to the dynamic
-// resource allocation plugin. This corresponds to PluginSocketPath, except
-// that PluginSocketPath defines the path in the filesystem of the caller and
-// KubeletPluginSocketPath in the filesystem of kubelet.
-func KubeletPluginSocketPath(path string) Option {
+// PluginListener configures how to create the registrar socket.
+// The default is to remove the file if it exists and to then
+// create a socket.
+//
+// This is used in Kubernetes for end-to-end testing. The default should
+// be fine for DRA drivers.
+func PluginListener(listen func(ctx context.Context, path string) (net.Listener, error)) Option {
 	return func(o *options) error {
-		o.draAddress = path
+		o.draEndpointListen = listen
 		return nil
 	}
 }
@@ -306,9 +328,9 @@ type options struct {
 	driverName                 string
 	nodeName                   string
 	nodeUID                    types.UID
-	draEndpoint                endpoint
-	draAddress                 string
 	pluginRegistrationEndpoint endpoint
+	pluginDataDirectoryPath    string
+	draEndpointListen          func(ctx context.Context, path string) (net.Listener, error)
 	unaryInterceptors          []grpc.UnaryServerInterceptor
 	streamInterceptors         []grpc.StreamServerInterceptor
 	kubeClient                 kubernetes.Interface
@@ -349,14 +371,17 @@ type Helper struct {
 // a name to all log entries.
 //
 // If the plugin will be used to publish resources, [KubeClient] and [NodeName]
-// options are mandatory.
+// options are mandatory. Otherwise only [DriverName] is mandatory.
 func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helper, finalErr error) {
 	logger := klog.FromContext(ctx)
 	o := options{
 		logger:        klog.Background(),
 		grpcVerbosity: 6, // Logs requests and responses, which can be large.
 		serialize:     true,
 		nodeV1beta1:   true,
+		pluginRegistrationEndpoint: endpoint{
+			dir: KubeletRegistryDir,
+		},
 	}
 	for _, option := range opts {
 		if err := option(&o); err != nil {
@@ -367,17 +392,12 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe
 	if o.driverName == "" {
 		return nil, errors.New("driver name must be set")
 	}
-	if o.draAddress == "" {
-		return nil, errors.New("DRA address must be set")
+	if o.pluginRegistrationEndpoint.file == "" {
+		o.pluginRegistrationEndpoint.file = o.driverName + "-reg.sock"
 	}
-	var emptyEndpoint endpoint
-	if o.draEndpoint == emptyEndpoint {
-		return nil, errors.New("a Unix domain socket path and/or listener must be set for the kubelet plugin")
+	if o.pluginDataDirectoryPath == "" {
+		o.pluginDataDirectoryPath = path.Join(KubeletPluginsDir, o.driverName)
 	}
-	if o.pluginRegistrationEndpoint == emptyEndpoint {
-		return nil, errors.New("a Unix domain socket path and/or listener must be set for the registrar")
-	}
-
 	d := &Helper{
 		driverName: o.driverName,
 		nodeName:   o.nodeName,
@@ -412,7 +432,12 @@ 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
-	pluginServer, err := startGRPCServer(klog.NewContext(ctx, klog.LoggerWithName(logger, "dra")), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.draEndpoint, func(grpcServer *grpc.Server) {
+	draEndpoint := endpoint{
+		dir:        o.pluginDataDirectoryPath,
+		file:       "dra.sock", // "dra" is hard-coded.
+		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})
@@ -428,7 +453,7 @@ func Start(ctx context.Context, plugin DRAPlugin, opts ...Option) (result *Helpe
 	}
 
 	// Now make it available to kubelet.
-	registrar, err := startRegistrar(klog.NewContext(ctx, klog.LoggerWithName(logger, "registrar")), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, o.draAddress, o.pluginRegistrationEndpoint)
+	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)
 	}
@@ -510,6 +535,11 @@ func (d *Helper) PublishResources(_ context.Context, resources resourceslice.Dri
 		// our background context, not the one passed into this
 		// function, and thus is connected to the lifecycle of the
 		// plugin.
+		//
+		// TODO: don't delete ResourceSlices, not even on a clean shutdown.
+		// We either support rolling updates and want to hand over seamlessly
+		// or don't and then perhaps restart the pod quickly enough that
+		// the kubelet hasn't deleted ResourceSlices yet.
 		controllerCtx := d.backgroundCtx
 		controllerLogger := klog.FromContext(controllerCtx)
 		controllerLogger = klog.LoggerWithName(controllerLogger, "ResourceSlice controller")

staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/endpoint.go

@@ -0,0 +1,83 @@
+/*
+Copyright 2025 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kubeletplugin
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"path"
+)
+
+// endpoint defines where and how to listen for incoming connections.
+// The listener always gets closed when shutting down.
+//
+// If the listen function is not set, a new listener for a Unix domain socket gets
+// created at the path.
+type endpoint struct {
+	dir, file  string
+	listenFunc func(ctx context.Context, socketpath string) (net.Listener, error)
+}
+
+func (e endpoint) path() string {
+	return path.Join(e.dir, e.file)
+}
+
+func (e endpoint) listen(ctx context.Context) (net.Listener, error) {
+	socketpath := e.path()
+
+	if e.listenFunc != nil {
+		return e.listenFunc(ctx, socketpath)
+	}
+
+	// Remove stale sockets, listen would fail otherwise.
+	if err := e.removeSocket(); err != nil {
+		return nil, err
+	}
+	cfg := net.ListenConfig{}
+	listener, err := cfg.Listen(ctx, "unix", socketpath)
+	if err != nil {
+		if removeErr := e.removeSocket(); removeErr != nil {
+			err = errors.Join(err, err)
+		}
+		return nil, err
+	}
+	return &unixListener{Listener: listener, endpoint: e}, nil
+}
+
+func (e endpoint) removeSocket() error {
+	if err := os.Remove(e.path()); err != nil && !os.IsNotExist(err) {
+		return fmt.Errorf("remove Unix domain socket: %w", err)
+	}
+	return nil
+}
+
+// unixListener adds removing of the Unix domain socket on Close.
+type unixListener struct {
+	net.Listener
+	endpoint endpoint
+}
+
+func (l *unixListener) Close() error {
+	err := l.Listener.Close()
+	if removeErr := l.endpoint.removeSocket(); removeErr != nil {
+		err = errors.Join(err, err)
+	}
+	return err
+}