DRA kubelet: list supported gRPC services during registration

Listing supported gRPC services (e.g. drav1alpha3.Node, drav1beta1.DRAPlugin)
during registration enables the kubelet to determine in advance which methods
it can call.

Versioning by Kubernetes release makes less sense because it doesn't say
anything about which gRPC service is supported. New ones might get added and
obsolete ones removed. Some services might be optional.

In the past, this versioning support wasn't really used. At least one version
had to be provided and kubelet tried to use the plugin with the highest
version. This version comparison gets dropped. In the unlikely situation
that different plugins register under the same name, the most recent one is
used.

Because advertising gRPC services is a new convention, plugins only reporting
some version are treated as providing the old alpha gRPC service.

pkg/kubelet/cm/dra/manager_test.go

@@ -581,7 +581,7 @@ func TestPrepareResources(t *testing.T) {
 			defer draServerInfo.teardownFn()
 
 			plg := plugin.NewRegistrationHandler(nil, getFakeNode)
-			if err := plg.RegisterPlugin(test.driverName, draServerInfo.socketName, []string{"1.27"}, pluginClientTimeout); err != nil {
+			if err := plg.RegisterPlugin(test.driverName, draServerInfo.socketName, []string{drapb.DRAPluginService}, pluginClientTimeout); err != nil {
 				t.Fatalf("failed to register plugin %s, err: %v", test.driverName, err)
 			}
 			defer plg.DeRegisterPlugin(test.driverName) // for sake of next tests
@@ -718,7 +718,7 @@ func TestUnprepareResources(t *testing.T) {
 			defer draServerInfo.teardownFn()
 
 			plg := plugin.NewRegistrationHandler(nil, getFakeNode)
-			if err := plg.RegisterPlugin(test.driverName, draServerInfo.socketName, []string{"1.27"}, pluginClientTimeout); err != nil {
+			if err := plg.RegisterPlugin(test.driverName, draServerInfo.socketName, []string{drapb.DRAPluginService}, pluginClientTimeout); err != nil {
 				t.Fatalf("failed to register plugin %s, err: %v", test.driverName, err)
 			}
 			defer plg.DeRegisterPlugin(test.driverName) // for sake of next tests
@@ -888,7 +888,7 @@ func TestParallelPrepareUnprepareResources(t *testing.T) {
 	defer draServerInfo.teardownFn()
 
 	plg := plugin.NewRegistrationHandler(nil, getFakeNode)
-	if err := plg.RegisterPlugin(driverName, draServerInfo.socketName, []string{"1.27"}, nil); err != nil {
+	if err := plg.RegisterPlugin(driverName, draServerInfo.socketName, []string{drapb.DRAPluginService}, nil); err != nil {
 		t.Fatalf("failed to register plugin %s, err: %v", driverName, err)
 	}
 	defer plg.DeRegisterPlugin(driverName)

pkg/kubelet/cm/dra/plugin/plugin.go

@@ -25,12 +25,10 @@ import (
 	"time"
 
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
 
-	utilversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/klog/v2"
 	drapbv1alpha4 "k8s.io/kubelet/pkg/apis/dra/v1alpha4"
 	drapbv1beta1 "k8s.io/kubelet/pkg/apis/dra/v1beta1"
@@ -59,27 +57,19 @@ type Plugin struct {
 	backgroundCtx context.Context
 	cancel        func(cause error)
 
-	mutex                   sync.Mutex
-	conn                    *grpc.ClientConn
-	supportedAPI            apiVersion
-	endpoint                string
-	highestSupportedVersion *utilversion.Version
-	clientCallTimeout       time.Duration
+	mutex             sync.Mutex
+	conn              *grpc.ClientConn
+	endpoint          string
+	chosenService     string // e.g. drapbv1beta1.DRAPluginService
+	clientCallTimeout time.Duration
 }
 
-type apiVersion string
-
-const (
-	apiV1alpha4 = apiVersion("v1alpha4")
-	apiV1beta1  = apiVersion("v1beta1")
-)
-
-func (p *Plugin) getOrCreateGRPCConn() (*grpc.ClientConn, apiVersion, error) {
+func (p *Plugin) getOrCreateGRPCConn() (*grpc.ClientConn, error) {
 	p.mutex.Lock()
 	defer p.mutex.Unlock()
 
 	if p.conn != nil {
-		return p.conn, p.supportedAPI, nil
+		return p.conn, nil
 	}
 
 	ctx := p.backgroundCtx
@@ -101,18 +91,18 @@ func (p *Plugin) getOrCreateGRPCConn() (*grpc.ClientConn, apiVersion, error) {
 		grpc.WithChainUnaryInterceptor(newMetricsInterceptor(p.name)),
 	)
 	if err != nil {
-		return nil, "", err
+		return nil, err
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 
 	if ok := conn.WaitForStateChange(ctx, connectivity.Connecting); !ok {
-		return nil, "", errors.New("timed out waiting for gRPC connection to be ready")
+		return nil, errors.New("timed out waiting for gRPC connection to be ready")
 	}
 
 	p.conn = conn
-	return p.conn, "", nil
+	return p.conn, nil
 }
 
 func (p *Plugin) NodePrepareResources(
@@ -123,7 +113,7 @@ func (p *Plugin) NodePrepareResources(
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("Calling NodePrepareResources rpc", "request", req)
 
-	conn, supportedAPI, err := p.getOrCreateGRPCConn()
+	conn, err := p.getOrCreateGRPCConn()
 	if err != nil {
 		return nil, err
 	}
@@ -132,29 +122,17 @@ func (p *Plugin) NodePrepareResources(
 	defer cancel()
 
 	var response *drapbv1beta1.NodePrepareResourcesResponse
-	switch supportedAPI {
-	case apiV1beta1:
+	switch p.chosenService {
+	case drapbv1beta1.DRAPluginService:
 		nodeClient := drapbv1beta1.NewDRAPluginClient(conn)
 		response, err = nodeClient.NodePrepareResources(ctx, req)
-	case apiV1alpha4:
+	case drapbv1alpha4.NodeService:
 		nodeClient := drapbv1alpha4.NewNodeClient(conn)
 		response, err = nodeClient.NodePrepareResources(ctx, req)
 	default:
-		// Try it, fall back if necessary.
-		supportedAPI = apiV1beta1
-		nodeClient := drapbv1beta1.NewDRAPluginClient(conn)
-		response, err = nodeClient.NodePrepareResources(ctx, req)
-		if err != nil && status.Convert(err).Code() == codes.Unimplemented {
-			supportedAPI = apiV1alpha4
-			nodeClient := drapbv1alpha4.NewNodeClient(conn)
-			response, err = nodeClient.NodePrepareResources(ctx, req)
-		}
-		if err == nil || status.Convert(err).Code() != codes.Unimplemented {
-			// Store discovered version for future use.
-			p.mutex.Lock()
-			p.supportedAPI = supportedAPI
-			p.mutex.Unlock()
-		}
+		// Shouldn't happen, validateSupportedServices should only
+		// return services we support here.
+		return nil, fmt.Errorf("internal error: unsupported chosen service: %q", p.chosenService)
 	}
 	logger.V(4).Info("Done calling NodePrepareResources rpc", "response", response, "err", err)
 	return response, err
@@ -168,7 +146,7 @@ func (p *Plugin) NodeUnprepareResources(
 	logger := klog.FromContext(ctx)
 	logger.V(4).Info("Calling NodeUnprepareResource rpc", "request", req)
 
-	conn, supportedAPI, err := p.getOrCreateGRPCConn()
+	conn, err := p.getOrCreateGRPCConn()
 	if err != nil {
 		return nil, err
 	}
@@ -177,29 +155,17 @@ func (p *Plugin) NodeUnprepareResources(
 	defer cancel()
 
 	var response *drapbv1beta1.NodeUnprepareResourcesResponse
-	switch supportedAPI {
-	case apiV1beta1:
+	switch p.chosenService {
+	case drapbv1beta1.DRAPluginService:
 		nodeClient := drapbv1beta1.NewDRAPluginClient(conn)
 		response, err = nodeClient.NodeUnprepareResources(ctx, req)
-	case apiV1alpha4:
+	case drapbv1alpha4.NodeService:
 		nodeClient := drapbv1alpha4.NewNodeClient(conn)
 		response, err = nodeClient.NodeUnprepareResources(ctx, req)
 	default:
-		// Try it, fall back if necessary.
-		supportedAPI = apiV1beta1
-		nodeClient := drapbv1beta1.NewDRAPluginClient(conn)
-		response, err = nodeClient.NodeUnprepareResources(ctx, req)
-		if err != nil && status.Convert(err).Code() == codes.Unimplemented {
-			supportedAPI = apiV1alpha4
-			nodeClient := drapbv1alpha4.NewNodeClient(conn)
-			response, err = nodeClient.NodeUnprepareResources(ctx, req)
-		}
-		if err == nil || status.Convert(err).Code() != codes.Unimplemented {
-			// Store discovered version for future use.
-			p.mutex.Lock()
-			p.supportedAPI = supportedAPI
-			p.mutex.Unlock()
-		}
+		// Shouldn't happen, validateSupportedServices should only
+		// return services we support here.
+		return nil, fmt.Errorf("internal error: unsupported chosen service: %q", p.chosenService)
 	}
 	logger.V(4).Info("Done calling NodeUnprepareResources rpc", "response", response, "err", err)
 	return response, err

pkg/kubelet/cm/dra/plugin/plugin_test.go

@@ -27,23 +27,20 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc"
-	drapb "k8s.io/kubelet/pkg/apis/dra/v1beta1"
+	drapbv1alpha4 "k8s.io/kubelet/pkg/apis/dra/v1alpha4"
+	drapbv1beta1 "k8s.io/kubelet/pkg/apis/dra/v1beta1"
 	"k8s.io/kubernetes/test/utils/ktesting"
 )
 
-const (
-	v1alpha4Version = "v1alpha4"
-)
-
-type fakeV1alpha4GRPCServer struct {
-	drapb.UnimplementedDRAPluginServer
+type fakeGRPCServer struct {
+	drapbv1beta1.UnimplementedDRAPluginServer
 }
 
-var _ drapb.DRAPluginServer = &fakeV1alpha4GRPCServer{}
+var _ drapbv1beta1.DRAPluginServer = &fakeGRPCServer{}
 
-func (f *fakeV1alpha4GRPCServer) NodePrepareResources(ctx context.Context, in *drapb.NodePrepareResourcesRequest) (*drapb.NodePrepareResourcesResponse, error) {
-	return &drapb.NodePrepareResourcesResponse{Claims: map[string]*drapb.NodePrepareResourceResponse{"claim-uid": {
-		Devices: []*drapb.Device{
+func (f *fakeGRPCServer) NodePrepareResources(ctx context.Context, in *drapbv1beta1.NodePrepareResourcesRequest) (*drapbv1beta1.NodePrepareResourcesResponse, error) {
+	return &drapbv1beta1.NodePrepareResourcesResponse{Claims: map[string]*drapbv1beta1.NodePrepareResourceResponse{"claim-uid": {
+		Devices: []*drapbv1beta1.Device{
 			{
 				RequestNames: []string{"test-request"},
 				CDIDeviceIDs: []string{"test-cdi-id"},
@@ -52,14 +49,14 @@ func (f *fakeV1alpha4GRPCServer) NodePrepareResources(ctx context.Context, in *d
 	}}}, nil
 }
 
-func (f *fakeV1alpha4GRPCServer) NodeUnprepareResources(ctx context.Context, in *drapb.NodeUnprepareResourcesRequest) (*drapb.NodeUnprepareResourcesResponse, error) {
+func (f *fakeGRPCServer) NodeUnprepareResources(ctx context.Context, in *drapbv1beta1.NodeUnprepareResourcesRequest) (*drapbv1beta1.NodeUnprepareResourcesResponse, error) {
 
-	return &drapb.NodeUnprepareResourcesResponse{}, nil
+	return &drapbv1beta1.NodeUnprepareResourcesResponse{}, nil
 }
 
 type tearDown func()
 
-func setupFakeGRPCServer(version string) (string, tearDown, error) {
+func setupFakeGRPCServer(service string) (string, tearDown, error) {
 	p, err := os.MkdirTemp("", "dra_plugin")
 	if err != nil {
 		return "", nil, err
@@ -81,12 +78,14 @@ func setupFakeGRPCServer(version string) (string, tearDown, error) {
 	}
 
 	s := grpc.NewServer()
-	switch version {
-	case v1alpha4Version:
-		fakeGRPCServer := &fakeV1alpha4GRPCServer{}
-		drapb.RegisterDRAPluginServer(s, fakeGRPCServer)
+	fakeGRPCServer := &fakeGRPCServer{}
+	switch service {
+	case drapbv1beta1.DRAPluginService:
+		drapbv1beta1.RegisterDRAPluginServer(s, fakeGRPCServer)
+	case drapbv1alpha4.NodeService:
+		drapbv1alpha4.RegisterNodeServer(s, fakeGRPCServer)
 	default:
-		return "", nil, fmt.Errorf("unsupported version: %s", version)
+		return "", nil, fmt.Errorf("unsupported gRPC service: %s", service)
 	}
 
 	go func() {
@@ -104,7 +103,8 @@ func setupFakeGRPCServer(version string) (string, tearDown, error) {
 
 func TestGRPCConnIsReused(t *testing.T) {
 	tCtx := ktesting.Init(t)
-	addr, teardown, err := setupFakeGRPCServer(v1alpha4Version)
+	service := drapbv1beta1.DRAPluginService
+	addr, teardown, err := setupFakeGRPCServer(service)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -119,10 +119,11 @@ func TestGRPCConnIsReused(t *testing.T) {
 		name:              pluginName,
 		backgroundCtx:     tCtx,
 		endpoint:          addr,
+		chosenService:     service,
 		clientCallTimeout: defaultClientCallTimeout,
 	}
 
-	conn, _, err := p.getOrCreateGRPCConn()
+	conn, err := p.getOrCreateGRPCConn()
 	defer func() {
 		err := conn.Close()
 		if err != nil {
@@ -148,8 +149,8 @@ func TestGRPCConnIsReused(t *testing.T) {
 				return
 			}
 
-			req := &drapb.NodePrepareResourcesRequest{
-				Claims: []*drapb.Claim{
+			req := &drapbv1beta1.NodePrepareResourcesRequest{
+				Claims: []*drapbv1beta1.Claim{
 					{
 						Namespace: "dummy-namespace",
 						UID:       "dummy-uid",
@@ -233,21 +234,27 @@ func TestNewDRAPluginClient(t *testing.T) {
 
 func TestNodeUnprepareResources(t *testing.T) {
 	for _, test := range []struct {
-		description   string
-		serverSetup   func(string) (string, tearDown, error)
-		serverVersion string
-		request       *drapb.NodeUnprepareResourcesRequest
+		description string
+		serverSetup func(string) (string, tearDown, error)
+		service     string
+		request     *drapbv1beta1.NodeUnprepareResourcesRequest
 	}{
 		{
-			description:   "server supports v1alpha4",
-			serverSetup:   setupFakeGRPCServer,
-			serverVersion: v1alpha4Version,
-			request:       &drapb.NodeUnprepareResourcesRequest{},
+			description: "server supports v1alpha4",
+			serverSetup: setupFakeGRPCServer,
+			service:     drapbv1alpha4.NodeService,
+			request:     &drapbv1beta1.NodeUnprepareResourcesRequest{},
+		},
+		{
+			description: "server supports v1beta1",
+			serverSetup: setupFakeGRPCServer,
+			service:     drapbv1beta1.DRAPluginService,
+			request:     &drapbv1beta1.NodeUnprepareResourcesRequest{},
 		},
 	} {
 		t.Run(test.description, func(t *testing.T) {
 			tCtx := ktesting.Init(t)
-			addr, teardown, err := setupFakeGRPCServer(test.serverVersion)
+			addr, teardown, err := setupFakeGRPCServer(test.service)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -258,10 +265,11 @@ func TestNodeUnprepareResources(t *testing.T) {
 				name:              pluginName,
 				backgroundCtx:     tCtx,
 				endpoint:          addr,
+				chosenService:     test.service,
 				clientCallTimeout: defaultClientCallTimeout,
 			}
 
-			conn, _, err := p.getOrCreateGRPCConn()
+			conn, err := p.getOrCreateGRPCConn()
 			defer func() {
 				err := conn.Close()
 				if err != nil {

pkg/kubelet/cm/dra/plugin/registration.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"time"
 
 	v1 "k8s.io/api/core/v1"
@@ -27,10 +28,11 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
-	utilversion "k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog/v2"
+	drapbv1alpha4 "k8s.io/kubelet/pkg/apis/dra/v1alpha4"
+	drapbv1beta1 "k8s.io/kubelet/pkg/apis/dra/v1beta1"
 	"k8s.io/kubernetes/pkg/kubelet/pluginmanager/cache"
 )
 
@@ -130,7 +132,13 @@ func (h *RegistrationHandler) wipeResourceSlices(driver string) {
 }
 
 // RegisterPlugin is called when a plugin can be registered.
-func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string, versions []string, pluginClientTimeout *time.Duration) error {
+//
+// DRA uses the version array in the registration API to enumerate all gRPC
+// services that the plugin provides, using the "<gRPC package name>.<service
+// name>" format (e.g. "v1beta1.DRAPlugin"). This allows kubelet to determine
+// in advance which version to use resp. which optional services the plugin
+// supports.
+func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string, supportedServices []string, pluginClientTimeout *time.Duration) error {
 	// Prepare a context with its own logger for the plugin.
 	//
 	// The lifecycle of the plugin's background activities is tied to our
@@ -145,7 +153,7 @@ func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string,
 
 	logger.V(3).Info("Register new DRA plugin", "endpoint", endpoint)
 
-	highestSupportedVersion, err := h.validateVersions(pluginName, versions)
+	chosenService, err := h.validateSupportedServices(pluginName, supportedServices)
 	if err != nil {
 		return fmt.Errorf("version check of plugin %s failed: %w", pluginName, err)
 	}
@@ -160,13 +168,13 @@ func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string,
 	ctx, cancel := context.WithCancelCause(ctx)
 
 	pluginInstance := &Plugin{
-		name:                    pluginName,
-		backgroundCtx:           ctx,
-		cancel:                  cancel,
-		conn:                    nil,
-		endpoint:                endpoint,
-		highestSupportedVersion: highestSupportedVersion,
-		clientCallTimeout:       timeout,
+		name:              pluginName,
+		backgroundCtx:     ctx,
+		cancel:            cancel,
+		conn:              nil,
+		endpoint:          endpoint,
+		chosenService:     chosenService,
+		clientCallTimeout: timeout,
 	}
 
 	// Storing endpoint of newly registered DRA Plugin into the map, where plugin name will be the key
@@ -178,30 +186,35 @@ func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string,
 	return nil
 }
 
-func (h *RegistrationHandler) validateVersions(
-	pluginName string,
-	versions []string,
-) (*utilversion.Version, error) {
-	if len(versions) == 0 {
-		return nil, errors.New("empty list for supported versions")
+// validateSupportedServices identifies the highest supported gRPC service for
+// NodePrepareResources and NodeUnprepareResources and returns its name
+// (e.g. [drapbv1beta1.DRAPluginService]). An error is returned if the plugin
+// is unusable.
+func (h *RegistrationHandler) validateSupportedServices(pluginName string, supportedServices []string) (string, error) {
+	if len(supportedServices) == 0 {
+		return "", errors.New("empty list of supported gRPC services (aka supported versions)")
 	}
 
-	// Validate version
-	newPluginHighestVersion, err := utilversion.HighestSupportedVersion(versions)
-	if err != nil {
-		// HighestSupportedVersion includes the list of versions in its error
-		// if relevant, no need to repeat it here.
-		return nil, fmt.Errorf("none of the versions are supported: %w", err)
+	// Pick most recent version if available.
+	chosenService := ""
+	for _, service := range []string{
+		// Sorted by most recent first, oldest last.
+		drapbv1beta1.DRAPluginService,
+		drapbv1alpha4.NodeService,
+	} {
+		if slices.Contains(supportedServices, service) {
+			chosenService = service
+			break
+		}
 	}
 
-	existingPlugin := draPlugins.get(pluginName)
-	if existingPlugin == nil {
-		return newPluginHighestVersion, nil
+	// Fall back to alpha if necessary because
+	// plugins at that time didn't advertise gRPC services.
+	if chosenService == "" {
+		chosenService = drapbv1alpha4.NodeService
 	}
-	if existingPlugin.highestSupportedVersion.LessThan(newPluginHighestVersion) {
-		return newPluginHighestVersion, nil
-	}
-	return nil, fmt.Errorf("another plugin instance is already registered with a higher supported version: %q < %q", newPluginHighestVersion, existingPlugin.highestSupportedVersion)
+
+	return chosenService, nil
 }
 
 // DeRegisterPlugin is called when a plugin has removed its socket,
@@ -225,8 +238,8 @@ func (h *RegistrationHandler) DeRegisterPlugin(pluginName string) {
 
 // ValidatePlugin is called by kubelet's plugin watcher upon detection
 // of a new registration socket opened by DRA plugin.
-func (h *RegistrationHandler) ValidatePlugin(pluginName string, endpoint string, versions []string) error {
-	_, err := h.validateVersions(pluginName, versions)
+func (h *RegistrationHandler) ValidatePlugin(pluginName string, endpoint string, supportedServices []string) error {
+	_, err := h.validateSupportedServices(pluginName, supportedServices)
 	if err != nil {
 		return fmt.Errorf("invalid versions of plugin %s: %w", pluginName, err)
 	}

pkg/kubelet/cm/dra/plugin/registration_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	drapb "k8s.io/kubelet/pkg/apis/dra/v1beta1"
 )
 
 func getFakeNode() (*v1.Node, error) {
@@ -34,12 +35,12 @@ func TestRegistrationHandler_ValidatePlugin(t *testing.T) {
 	}
 
 	for _, test := range []struct {
-		description string
-		handler     func() *RegistrationHandler
-		pluginName  string
-		endpoint    string
-		versions    []string
-		shouldError bool
+		description       string
+		handler           func() *RegistrationHandler
+		pluginName        string
+		endpoint          string
+		supportedServices []string
+		shouldError       bool
 	}{
 		{
 			description: "no versions provided",
@@ -47,34 +48,15 @@ func TestRegistrationHandler_ValidatePlugin(t *testing.T) {
 			shouldError: true,
 		},
 		{
-			description: "unsupported version",
-			handler:     newRegistrationHandler,
-			versions:    []string{"v2.0.0"},
-			shouldError: true,
-		},
-		{
-			description: "plugin already registered with a higher supported version",
-			handler: func() *RegistrationHandler {
-				handler := newRegistrationHandler()
-				if err := handler.RegisterPlugin("this-plugin-already-exists-and-has-a-long-name-so-it-doesnt-collide", "", []string{"v1.1.0"}, nil); err != nil {
-					t.Fatal(err)
-				}
-				return handler
-			},
-			pluginName:  "this-plugin-already-exists-and-has-a-long-name-so-it-doesnt-collide",
-			versions:    []string{"v1.0.0"},
-			shouldError: true,
-		},
-		{
-			description: "should validate the plugin",
-			handler:     newRegistrationHandler,
-			pluginName:  "this-is-a-dummy-plugin-with-a-long-name-so-it-doesnt-collide",
-			versions:    []string{"v1.3.0"},
+			description:       "should validate the plugin",
+			handler:           newRegistrationHandler,
+			pluginName:        "this-is-a-dummy-plugin-with-a-long-name-so-it-doesnt-collide",
+			supportedServices: []string{drapb.DRAPluginService},
 		},
 	} {
 		t.Run(test.description, func(t *testing.T) {
 			handler := test.handler()
-			err := handler.ValidatePlugin(test.pluginName, test.endpoint, test.versions)
+			err := handler.ValidatePlugin(test.pluginName, test.endpoint, test.supportedServices)
 			if test.shouldError {
 				assert.Error(t, err)
 			} else {

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

@@ -330,29 +330,29 @@ func Start(ctx context.Context, nodeServer interface{}, opts ...Option) (result
 	}()
 
 	// Run the node plugin gRPC server first to ensure that it is ready.
-	implemented := false
+	var supportedServices []string
 	plugin, err := startGRPCServer(klog.NewContext(ctx, klog.LoggerWithName(logger, "dra")), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.draEndpoint, func(grpcServer *grpc.Server) {
 		if nodeServer, ok := nodeServer.(drapbv1alpha4.NodeServer); ok && o.nodeV1alpha4 {
 			logger.V(5).Info("registering v1alpha4.Node gGRPC service")
 			drapbv1alpha4.RegisterNodeServer(grpcServer, nodeServer)
-			implemented = true
+			supportedServices = append(supportedServices, drapbv1alpha4.NodeService)
 		}
 		if nodeServer, ok := nodeServer.(drapbv1beta1.DRAPluginServer); ok && o.nodeV1beta1 {
 			logger.V(5).Info("registering v1beta1.DRAPlugin gRPC service")
 			drapbv1beta1.RegisterDRAPluginServer(grpcServer, nodeServer)
-			implemented = true
+			supportedServices = append(supportedServices, drapbv1beta1.DRAPluginService)
 		}
 	})
 	if err != nil {
 		return nil, fmt.Errorf("start node client: %v", err)
 	}
 	d.plugin = plugin
-	if !implemented {
+	if len(supportedServices) == 0 {
 		return nil, errors.New("no supported DRA gRPC API is implemented and enabled")
 	}
 
 	// Now make it available to kubelet.
-	registrar, err := startRegistrar(klog.NewContext(ctx, klog.LoggerWithName(logger, "registrar")), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, o.draAddress, o.pluginRegistrationEndpoint)
+	registrar, err := startRegistrar(klog.NewContext(ctx, klog.LoggerWithName(logger, "registrar")), o.grpcVerbosity, o.unaryInterceptors, o.streamInterceptors, o.driverName, supportedServices, o.draAddress, o.pluginRegistrationEndpoint)
 	if err != nil {
 		return nil, fmt.Errorf("start registrar: %v", err)
 	}

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

@@ -32,12 +32,12 @@ type nodeRegistrar struct {
 // startRegistrar returns a running instance.
 //
 // The context is only used for additional values, cancellation is ignored.
-func startRegistrar(valueCtx context.Context, grpcVerbosity int, interceptors []grpc.UnaryServerInterceptor, streamInterceptors []grpc.StreamServerInterceptor, driverName string, endpoint string, pluginRegistrationEndpoint endpoint) (*nodeRegistrar, error) {
+func startRegistrar(valueCtx context.Context, grpcVerbosity int, interceptors []grpc.UnaryServerInterceptor, streamInterceptors []grpc.StreamServerInterceptor, driverName string, supportedServices []string, endpoint string, pluginRegistrationEndpoint endpoint) (*nodeRegistrar, error) {
 	n := &nodeRegistrar{
 		registrationServer: registrationServer{
 			driverName:        driverName,
 			endpoint:          endpoint,
-			supportedVersions: []string{"1.0.0"}, // TODO: is this correct?
+			supportedVersions: supportedServices, // DRA uses this field to describe provided services (e.g. "v1beta1.DRAPlugin").
 		},
 	}
 	s, err := startGRPCServer(valueCtx, grpcVerbosity, interceptors, streamInterceptors, pluginRegistrationEndpoint, func(grpcServer *grpc.Server) {

staging/src/k8s.io/kubelet/pkg/apis/dra/v1alpha4/types.go

@@ -0,0 +1,30 @@
+/*
+Copyright 2024 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 v1alpha4
+
+const (
+	// NodeService should be listed in the "supported versions"
+	// array during plugin registration by a DRA plugin which provides
+	// an implementation of the v1alpha3 Node service.
+	//
+	// This convention was introduced in Kubernetes 1.32. Older DRA
+	// plugins provide the implementation without advertising it.
+	//
+	// For historic reasons (= a mistake...) there is a mismatch between
+	// the package name and gRPC version.
+	NodeService = "v1alpha3.NodeService"
+)

staging/src/k8s.io/kubelet/pkg/apis/dra/v1beta1/types.go

@@ -14,10 +14,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// Package v1beta1 has the same Go API  as v1alpha4. A DRA driver implementing
-// [v1beta1.NodeServer] also implements [v1alpha4.NodeServer] and vice versa.
+// Package v1beta1 has the same Go API as v1alpha4. A DRA driver implementing
+// [v1beta1.DRAPluginServer] also implements [v1alpha4.NodeServer] and vice versa.
 //
-// The k8s.io/dynamic-resource-allocation/kubeletplugin helper will
+// The [k8s.io/dynamic-resource-allocation/kubeletplugin] helper will
 // automatically register both API versions unless explicitly configured
 // otherwise.
 package v1beta1
@@ -37,5 +37,12 @@ type (
 	Claim                          = v1alpha4.Claim
 )
 
+const (
+	// DRAPluginService needs to be listed in the "supported versions"
+	// array during plugin registration by a DRA plugin which provides
+	// an implementation of the v1beta1 DRAPlugin service.
+	DRAPluginService = "v1beta1.DRAPlugin"
+)
+
 // Ensure that the interfaces are equivalent.
 var _ DRAPluginServer = v1alpha4.NodeServer(nil)