DRA kubeletplugin: turn helper into wrapper

The goal is to simplify writing DRA drivers. This is also a first step towards
supporting seamless upgrades.

DRA drivers no longer need to implement the kubelet plugin API
directly. Instead, the helper wraps an implementation of an interface. The
helper then provides common functionality:

- retrieve and validate ResourceClaims
- serialize gRPC calls (enabled by default, can be opted out)
- gRPC logging

The definition of that interface is meant to be comprehensive enough that a
correct DRA driver can be implemented by following the documentation of the
package, without having to cross-reference KEPs.

The DRAPlugin interface used to be the abstract API of the helper. Now it's
what the DRA driver kubelet plugin needs to implement. The helper is a concrete
Server struct with no exported fields. It only exports the methods that
drivers need when using the helper.

While at it, support for the v1alpha4 API gets removed from the helper, which
implies removing the corresponding E2E tests. The kubelet implementation will
be dropped separately.

Author: pohly

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

@@ -0,0 +1,50 @@
+/*
+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 (
+	"k8s.io/apimachinery/pkg/types"
+)
+
+// NamespacedObject comprises a resource name with a mandatory namespace
+// and optional UID. It gets rendered as "<namespace>/<name>:[<uid>]"
+// (text output) or as an object (JSON output).
+type NamespacedObject struct {
+	types.NamespacedName
+	UID types.UID
+}
+
+// String returns the general purpose string representation
+func (n NamespacedObject) String() string {
+	if n.UID != "" {
+		return n.Namespace + string(types.Separator) + n.Name + ":" + string(n.UID)
+	}
+	return n.Namespace + string(types.Separator) + n.Name
+}
+
+// MarshalLog emits a struct containing required key/value pair
+func (n NamespacedObject) MarshalLog() interface{} {
+	return struct {
+		Name      string    `json:"name"`
+		Namespace string    `json:"namespace,omitempty"`
+		UID       types.UID `json:"uid,omitempty"`
+	}{
+		Name:      n.Name,
+		Namespace: n.Namespace,
+		UID:       n.UID,
+	}
+}

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

@@ -82,14 +82,18 @@ func startGRPCServer(valueCtx context.Context, grpcVerbosity int, unaryIntercept
 
 	// Run a gRPC server. It will close the listening socket when
 	// shutting down, so we don't need to do that.
+	// The per-context logger always gets initialized because
+	// there might be log output inside the method implementations.
 	var opts []grpc.ServerOption
-	finalUnaryInterceptors := []grpc.UnaryServerInterceptor{unaryContextInterceptor(valueCtx)}
-	finalStreamInterceptors := []grpc.StreamServerInterceptor{streamContextInterceptor(valueCtx)}
-	if grpcVerbosity >= 0 {
-		finalUnaryInterceptors = append(finalUnaryInterceptors, s.interceptor)
-		finalStreamInterceptors = append(finalStreamInterceptors, s.streamInterceptor)
+	finalUnaryInterceptors := []grpc.UnaryServerInterceptor{
+		unaryContextInterceptor(valueCtx),
+		s.interceptor,
 	}
 	finalUnaryInterceptors = append(finalUnaryInterceptors, unaryInterceptors...)
+	finalStreamInterceptors := []grpc.StreamServerInterceptor{
+		streamContextInterceptor(valueCtx),
+		s.streamInterceptor,
+	}
 	finalStreamInterceptors = append(finalStreamInterceptors, streamInterceptors...)
 	opts = append(opts, grpc.ChainUnaryInterceptor(finalUnaryInterceptors...))
 	opts = append(opts, grpc.ChainStreamInterceptor(finalStreamInterceptors...))
@@ -164,7 +168,6 @@ func (m mergeCtx) Value(i interface{}) interface{} {
 // sequentially increasing request ID and adds that logger to the context. It
 // also logs request and response.
 func (s *grpcServer) interceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
-
 	requestID := atomic.AddInt64(&requestID, 1)
 	logger := klog.FromContext(ctx)
 	logger = klog.LoggerWithValues(logger, "requestID", requestID, "method", info.FullMethod)

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

@@ -430,7 +430,6 @@ func (d *Driver) SetUp(nodes *Nodes, resources Resources, devicesPerNode ...map[
 			kubeletplugin.PluginListener(listen(ctx, d.f, pod.Name, "plugin", 9001)),
 			kubeletplugin.RegistrarListener(listen(ctx, d.f, pod.Name, "registrar", 9000)),
 			kubeletplugin.KubeletPluginSocketPath(draAddr),
-			kubeletplugin.NodeV1alpha4(d.NodeV1alpha4),
 			kubeletplugin.NodeV1beta1(d.NodeV1beta1),
 		)
 		framework.ExpectNoError(err, "start kubelet plugin for node %s", pod.Spec.NodeName)

test/e2e/dra/deploy.go

@@ -1737,15 +1737,13 @@ var _ = framework.SIGDescribe("node")("DRA", feature.DynamicResourceAllocation,
 		})
 	})
 
-	multipleDrivers := func(nodeV1alpha4, nodeV1beta1 bool) {
+	multipleDrivers := func(nodeV1beta1 bool) {
 		nodes := NewNodes(f, 1, 4)
 		driver1 := NewDriver(f, nodes, perNode(2, nodes))
-		driver1.NodeV1alpha4 = nodeV1alpha4
 		driver1.NodeV1beta1 = nodeV1beta1
 		b1 := newBuilder(f, driver1)
 
 		driver2 := NewDriver(f, nodes, perNode(2, nodes))
-		driver2.NodeV1alpha4 = nodeV1alpha4
 		driver2.NodeV1beta1 = nodeV1beta1
 		driver2.NameSuffix = "-other"
 		b2 := newBuilder(f, driver2)
@@ -1769,16 +1767,14 @@ var _ = framework.SIGDescribe("node")("DRA", feature.DynamicResourceAllocation,
 			b1.testPod(ctx, f, pod)
 		})
 	}
-	multipleDriversContext := func(prefix string, nodeV1alpha4, nodeV1beta1 bool) {
+	multipleDriversContext := func(prefix string, nodeV1beta1 bool) {
 		ginkgo.Context(prefix, func() {
-			multipleDrivers(nodeV1alpha4, nodeV1beta1)
+			multipleDrivers(nodeV1beta1)
 		})
 	}
 
 	ginkgo.Context("multiple drivers", func() {
-		multipleDriversContext("using only drapbv1alpha4", true, false)
-		multipleDriversContext("using only drapbv1beta1", false, true)
-		multipleDriversContext("using both drav1alpha4 and drapbv1beta1", true, true)
+		multipleDriversContext("using only drapbv1beta1", true)
 	})
 
 	ginkgo.It("runs pod after driver starts", func(ctx context.Context) {

test/e2e/dra/dra.go

@@ -60,7 +60,11 @@ RUNTIME_CONFIG="resource.k8s.io/v1alpha3" FEATURE_GATES=DynamicResourceAllocatio
 
 In another:
 ```
-sudo mkdir -p /var/run/cdi && sudo chmod a+rwx /var/run/cdi /var/lib/kubelet/plugins_registry
+sudo mkdir -p /var/run/cdi
+sudo mkdir -p /var/lib/kubelet/plugins
+sudo mkdir -p /var/lib/kubelet/plugins_registry
+sudo chmod a+rx /var/lib/kubelet /var/lib/kubelet/plugins
+sudo chmod a+rwx /var/run/cdi /var/lib/kubelet/plugins_registry
 KUBECONFIG=/var/run/kubernetes/admin.kubeconfig go run ./test/e2e/dra/test-driver -v=5 kubelet-plugin --node-name=127.0.0.1
 ```