diff --git a/pkg/kubelet/cm/dra/manager_test.go b/pkg/kubelet/cm/dra/manager_test.go index fb5ab19b1f1..bc2448690a8 100644 --- a/pkg/kubelet/cm/dra/manager_test.go +++ b/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) diff --git a/pkg/kubelet/cm/dra/plugin/plugin.go b/pkg/kubelet/cm/dra/plugin/plugin.go index cbc961dbe75..cfa3a05cde3 100644 --- a/pkg/kubelet/cm/dra/plugin/plugin.go +++ b/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 diff --git a/pkg/kubelet/cm/dra/plugin/plugin_test.go b/pkg/kubelet/cm/dra/plugin/plugin_test.go index 9ee32a9b8f5..87c3cef13f3 100644 --- a/pkg/kubelet/cm/dra/plugin/plugin_test.go +++ b/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 { diff --git a/pkg/kubelet/cm/dra/plugin/registration.go b/pkg/kubelet/cm/dra/plugin/registration.go index 0af6f24a382..03865b09338 100644 --- a/pkg/kubelet/cm/dra/plugin/registration.go +++ b/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 "." 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) } diff --git a/pkg/kubelet/cm/dra/plugin/registration_test.go b/pkg/kubelet/cm/dra/plugin/registration_test.go index 91258be8e69..880d134824b 100644 --- a/pkg/kubelet/cm/dra/plugin/registration_test.go +++ b/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 { 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 f12276bd6d4..813c3ce1c40 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/draplugin.go +++ b/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) } diff --git a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/noderegistrar.go b/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/noderegistrar.go index abcc6df2170..6384ca0c180 100644 --- a/staging/src/k8s.io/dynamic-resource-allocation/kubeletplugin/noderegistrar.go +++ b/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) { diff --git a/staging/src/k8s.io/kubelet/pkg/apis/dra/v1alpha4/types.go b/staging/src/k8s.io/kubelet/pkg/apis/dra/v1alpha4/types.go new file mode 100644 index 00000000000..aec6992ee8f --- /dev/null +++ b/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" +) diff --git a/staging/src/k8s.io/kubelet/pkg/apis/dra/v1beta1/types.go b/staging/src/k8s.io/kubelet/pkg/apis/dra/v1beta1/types.go index c6562f8b090..4658eb51530 100644 --- a/staging/src/k8s.io/kubelet/pkg/apis/dra/v1beta1/types.go +++ b/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)