test/e2e/dra/utils/deploy.go - master - githubmirror/kubernetes

test/e2e/dra/utils/deploy.go
1 260 строк46 KB
Omer Yahud
DRA: implement device compatibility groups with unit, integration, and e2e tests
24 июл 2026, 16:19
24 июл 2026, 16:195539307
/*
Copyright 2022 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 utils

import (
	"bytes"
	"context"
	"crypto/sha256"
	_ "embed"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"net"
	"net/url"
	"os"
	"path"
	"sort"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/google/go-cmp/cmp"
	"github.com/onsi/ginkgo/v2"
	"github.com/onsi/gomega"
	"github.com/onsi/gomega/format"
	"google.golang.org/grpc"

	appsv1 "k8s.io/api/apps/v1"
	v1 "k8s.io/api/core/v1"
	resourceapi "k8s.io/api/resource/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/meta"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/selection"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	watch "k8s.io/apimachinery/pkg/watch"
	"k8s.io/apiserver/pkg/authentication/serviceaccount"
	"k8s.io/client-go/discovery/cached/memory"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/kubernetes/scheme"
	cgoresource "k8s.io/client-go/kubernetes/typed/resource/v1"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/restmapper"
	"k8s.io/client-go/tools/cache"
	draclient "k8s.io/dynamic-resource-allocation/client"
	"k8s.io/dynamic-resource-allocation/kubeletplugin"
	"k8s.io/dynamic-resource-allocation/resourceslice"
	"k8s.io/klog/v2"
	"k8s.io/kubectl/pkg/cmd/exec"
	"k8s.io/kubernetes/test/e2e/dra/test-driver/app"
	"k8s.io/kubernetes/test/e2e/dra/test-driver/deploy/example"
	testdrivergomega "k8s.io/kubernetes/test/e2e/dra/test-driver/gomega"
	"k8s.io/kubernetes/test/e2e/framework"
	e2enode "k8s.io/kubernetes/test/e2e/framework/node"
	e2epod "k8s.io/kubernetes/test/e2e/framework/pod"
	e2ereplicaset "k8s.io/kubernetes/test/e2e/framework/replicaset"
	e2eskipper "k8s.io/kubernetes/test/e2e/framework/skipper"
	"k8s.io/kubernetes/test/e2e/storage/drivers/proxy"
	"k8s.io/kubernetes/test/e2e/storage/utils"
	"k8s.io/kubernetes/test/utils/client-go/ktesting"
	"k8s.io/kubernetes/test/utils/image"
	"k8s.io/utils/clock"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/yaml"
)

type Nodes struct {
	// NodeNames has the main set of node names.
	NodeNames []string
	tempDir   string
	// NumReservedNodes specifies the desired number of
	// extra nodes that get set aside. That many node names
	// will be stored in ExtraNodeNames.
	//
	// Must be <= the minimum number of requested nodes.
	NumReservedNodes int
	// ExtraNodeNames has exactly as many node names as
	// requested via NumReservedNodes. Those nodes are
	// different than the nodes listed in NodeNames.
	ExtraNodeNames []string
}

// NewNodes selects nodes to run the test on.
//
// Call this outside of ginkgo.It, then use the instance inside ginkgo.It.
func NewNodes(f *framework.Framework, minNodes, maxNodes int) *Nodes {
	nodes := &Nodes{}
	ginkgo.BeforeEach(func(ctx context.Context) {
		nodes.init(f.TContext(ctx), minNodes, maxNodes)
	})
	return nodes
}

// NewNodesNow is a variant of NewNodes which can be used inside a ginkgo.It
// or a Go unit test.
func NewNodesNow(tCtx ktesting.TContext, minNodes, maxNodes int) *Nodes {
	nodes := &Nodes{}
	nodes.init(tCtx, minNodes, maxNodes)
	return nodes
}

func (nodes *Nodes) init(tCtx ktesting.TContext, minNodes, maxNodes int) {
	nodes.tempDir = tCtx.TempDir()

	tCtx.Log("selecting nodes")
	// The kubelet plugin is harder. We deploy the builtin manifest
	// after patching in the driver name and all nodes on which we
	// want the plugin to run.
	//
	// Only a subset of the nodes are picked to avoid causing
	// unnecessary load on a big cluster.
	nodeList, err := e2enode.GetBoundedReadySchedulableNodes(tCtx, tCtx.Client(), maxNodes)
	tCtx.ExpectNoError(err, "get nodes")
	numNodes := int32(len(nodeList.Items))
	if int(numNodes) < minNodes {
		e2eskipper.Skipf("%d ready nodes required, only have %d", minNodes+nodes.NumReservedNodes, numNodes)
	}
	nodes.NodeNames = nil
	for i, node := range nodeList.Items {
		if i < nodes.NumReservedNodes {
			nodes.ExtraNodeNames = append(nodes.ExtraNodeNames, node.Name)
			continue
		}
		nodes.NodeNames = append(nodes.NodeNames, node.Name)
	}
	sort.Strings(nodes.NodeNames)
	tCtx.Logf("testing on nodes %v", nodes.NodeNames)

	// Watch claims in the namespace. This is useful for monitoring a test
	// and enables additional sanity checks.
	resourceClaimLogger := klog.LoggerWithName(klog.FromContext(tCtx), "ResourceClaimListWatch")
	var resourceClaimWatchCounter atomic.Int32
	resourceClient := draclient.New(tCtx.Client())
	claimInformer := cache.NewSharedIndexInformer(
		&cache.ListWatch{
			ListWithContextFunc: func(ctx context.Context, options metav1.ListOptions) (runtime.Object, error) {
				tCtx := tCtx.WithContext(ctx)
				slices, err := resourceClient.ResourceClaims(tCtx.Namespace()).List(tCtx, options)
				if err == nil {
					resourceClaimLogger.Info("Listed ResourceClaims", "resourceAPI", resourceClient.CurrentAPI(), "numClaims", len(slices.Items), "listMeta", slices.ListMeta)
				} else {
					resourceClaimLogger.Info("Listing ResourceClaims failed", "resourceAPI", resourceClient.CurrentAPI(), "err", err)
				}
				return slices, err
			},
			WatchFuncWithContext: func(ctx context.Context, options metav1.ListOptions) (watch.Interface, error) {
				tCtx := tCtx.WithContext(ctx)
				w, err := resourceClient.ResourceClaims(tCtx.Namespace()).Watch(tCtx, options)
				if err == nil {
					resourceClaimLogger.Info("Started watching ResourceClaims", "resourceAPI", resourceClient.CurrentAPI())
					wrapper := newWatchWrapper(klog.LoggerWithName(resourceClaimLogger, fmt.Sprintf("%d", resourceClaimWatchCounter.Load())), w)
					resourceClaimWatchCounter.Add(1)
					go wrapper.run()
					w = wrapper
				} else {
					resourceClaimLogger.Info("Watching ResourceClaims failed", "resourceAPI", resourceClient.CurrentAPI(), "err", err)
				}
				return w, err
			},
		},
		&resourceapi.ResourceClaim{},
		// No resync because all it would do is periodically trigger syncing pools
		// again by reporting all slices as updated with the object as old/new.
		0,
		nil,
	)
	cancelCtx, cancel := context.WithCancelCause(context.Background())
	var wg sync.WaitGroup
	tCtx.Cleanup(func() {
		cancel(errors.New("test has completed"))
		wg.Wait()
	})
	_, err = claimInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: func(obj any) {
			claim := obj.(*resourceapi.ResourceClaim)
			resourceClaimLogger.Info("New claim", "claim", format.Object(claim, 0))
			validateClaim(tCtx, claim)
		},
		UpdateFunc: func(oldObj, newObj any) {
			oldClaim := oldObj.(*resourceapi.ResourceClaim)
			newClaim := newObj.(*resourceapi.ResourceClaim)
			resourceClaimLogger.Info("Updated claim", "newClaim", format.Object(newClaim, 0), "diff", cmp.Diff(oldClaim, newClaim))
			validateClaim(tCtx, newClaim)
		},
		DeleteFunc: func(obj any) {
			if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
				obj = tombstone.Obj
			}
			claim := obj.(*resourceapi.ResourceClaim)
			resourceClaimLogger.Info("Deleted claim", "claim", format.Object(claim, 0))
		},
	})
	tCtx.ExpectNoError(err, "AddEventHandler")
	wg.Add(1)
	go func() {
		defer wg.Done()
		claimInformer.RunWithContext(cancelCtx)
	}()
}

type watchWrapper struct {
	logger     klog.Logger
	delegate   watch.Interface
	resultChan chan watch.Event
}

func newWatchWrapper(logger klog.Logger, delegate watch.Interface) *watchWrapper {
	return &watchWrapper{
		logger:     logger,
		delegate:   delegate,
		resultChan: make(chan watch.Event, 100),
	}
}

func (w *watchWrapper) run() {
	defer utilruntime.HandleCrashWithLogger(w.logger)
	defer close(w.resultChan)
	inputChan := w.delegate.ResultChan()
	for {
		event, ok := <-inputChan
		if !ok {
			w.logger.Info("Wrapped result channel was closed, stopping event forwarding")
			return
		}
		w.logger.Info("Received event", "event", event.Type, "content", fmt.Sprintf("%T", event.Object))
		w.resultChan <- event
	}
}

func (w *watchWrapper) Stop() {
	w.delegate.Stop()
}

func (w *watchWrapper) ResultChan() <-chan watch.Event {
	return w.resultChan
}

func validateClaim(tCtx ktesting.TContext, claim *resourceapi.ResourceClaim) {
	// The apiserver doesn't enforce that a claim always has a finalizer
	// while being allocated. This is a convention that whoever allocates a
	// claim has to follow to prevent using a claim that is at risk of
	// being deleted.
	if claim.Status.Allocation != nil && len(claim.Finalizers) == 0 {
		tCtx.Errorf("Invalid claim: allocated without any finalizer:\n%s", format.Object(claim, 1))
	}
}

const (
	// multiHostDriverResources identifies DriverResources that are associated with multiple devices, i.e.
	// is not managed by a driver. So any Pools and ResourceSlices will be published directly
	// to the cluster, rather than through the driver.
	multiHostDriverResources = "multi-host"

	// LongRollingUpdateDriverName is a 30-character *.sigs.k8s.io-style driver name.
	// With rolling updates, the legacy registration socket basename
	// ({driver}-{pod UID}-reg.sock) exceeds common AF_UNIX path limits; see
	// https://github.com/kubernetes/kubernetes/issues/139166.
	LongRollingUpdateDriverName = "gpu.dra-example-driver.sigs.k8s.io"
)

// driverResourcesGenFunc defines the callback that will be invoked by the driver to generate the
// DriverResources that will be used to construct the ResourceSlices.
type driverResourcesGenFunc func(nodes *Nodes) map[string]resourceslice.DriverResources

// driverResourcesMutatorFunc defines the function signature for mutators that will
// update the DriverResources after they have been generated.
type driverResourcesMutatorFunc func(map[string]resourceslice.DriverResources)

// NewDriver sets up controller (as client of the cluster) and
// kubelet plugin (via proxy) before the test runs. It cleans
// up after the test.
//
// Call this outside of ginkgo.It, then use the instance inside ginkgo.It.
func NewDriver(f *framework.Framework, nodes *Nodes, driverResourcesGenerator driverResourcesGenFunc, driverResourcesMutators ...driverResourcesMutatorFunc) *Driver {
	d := NewDriverInstance(ktesting.TContext{} /* no namespace yet, will be set later */)

	ginkgo.BeforeEach(func() {
		tCtx := f.TContext(context.Background())
		d.initName(tCtx)
		driverResources := driverResourcesGenerator(nodes)
		for _, mutator := range driverResourcesMutators {
			mutator(driverResources)
		}
		d.Run(tCtx, framework.TestContext.KubeletRootDir, nodes, driverResources)
	})
	return d
}

// NewDriverInstance is a variant of NewDriver where the driver is inactive and must
// be started explicitly with Run. May be used inside ginkgo.It or a Go unit test.
// The context is used to determine the test's and thus the driver's namespace.
func NewDriverInstance(tCtx ktesting.TContext) *Driver {
	d := &Driver{
		fail:       map[MethodInstance]bool{},
		callCounts: map[MethodInstance]int64{},
		// By default, test with all gRPC APIs.
		// TODO: should setting this be optional to test the actual helper defaults?
		NodeV1:      true,
		NodeV1beta1: true,
		// By default, assume that the kubelet supports DRA and that
		// the driver's removal causes ResourceSlice cleanup.
		WithKubelet:                true,
		WithRealNodes:              true,
		ExpectResourceSliceRemoval: true,
	}
	d.initName(tCtx)
	return d
}

// ClientV1 returns a wrapper for client-go which provides the V1 API on top of whatever is enabled in the cluster.
func (d *Driver) ClientV1(tCtx ktesting.TContext) cgoresource.ResourceV1Interface {
	return draclient.New(tCtx.Client())
}

func (d *Driver) Run(tCtx ktesting.TContext, kubeletRootDir string, nodes *Nodes, driverResources map[string]resourceslice.DriverResources) {
	d.SetUp(tCtx, kubeletRootDir, nodes, driverResources)
	tCtx.CleanupCtx(d.TearDown)
}

// PublishResources re-publishes the given per-node driver resources through the
// already-running kubelet plugins, replacing whatever was published before. It
// is used to reset the resourceslice controller's desired state after a feature
// gate has been toggled: while a gate is off the apiserver drops the gated
// fields, and the controller latches its desired state to the stored (stripped)
// result to avoid a hot update loop. Re-publishing the original resources once
// the gate is back on restores those fields. This only handles the per-node
// publishing path, not multi-host DriverResources.
func (d *Driver) PublishResources(tCtx ktesting.TContext, driverResources map[string]resourceslice.DriverResources) {
	for nodename, plugin := range d.Nodes {
		dr, ok := driverResources[nodename]
		if !ok {
			continue
		}
		tCtx.ExpectNoError(plugin.PublishResources(tCtx, dr), "re-publish resources for node %s", nodename)
	}
}

// SetExpectDroppedFields controls whether the driver's error handler tolerates
// the apiserver dropping fields from published ResourceSlices. It must be set to
// true before a feature gate that gates a ResourceSlice field is turned off, and
// reset to false once the gate is back on and the driver has re-published the
// affected slices. See the expectDroppedFields field for details.
func (d *Driver) SetExpectDroppedFields(expect bool) {
	d.expectDroppedFields.Store(expect)
}

// NewGetSlices generates a function for ktesting.Eventually/Consistently which
// returns the ResourceSliceList.
func (d *Driver) NewGetSlices() func(tCtx ktesting.TContext) *resourceapi.ResourceSliceList {
	return func(tCtx ktesting.TContext) *resourceapi.ResourceSliceList {
		slices, err := framework.ListObjects(d.ClientV1(tCtx).ResourceSlices().List, metav1.ListOptions{FieldSelector: resourceapi.ResourceSliceSelectorDriver + "=" + d.Name})(tCtx)
		tCtx.ExpectNoError(err, "list ResourceSlices")
		return slices
	}
}

type MethodInstance struct {
	NodeName   string
	FullMethod string
}

type Driver struct {
	cleanup            []func(ktesting.TContext) // executed first-in-first-out
	wg                 sync.WaitGroup
	serviceAccountName string

	// NameSuffix can be set while registering a test to deploy different
	// drivers in the same test namespace.
	NameSuffix string

	// InstanceSuffix can be set while registering a test to deploy two different
	// instances of the same driver. Used to generate unique objects in the API server.
	// The socket path is still the same.
	InstanceSuffix string

	// RollingUpdate can be set to true to enable using different socket names
	// for different pods and thus seamless upgrades. Must be supported by the kubelet!
	RollingUpdate bool

	// Normally, tearing down the driver should cause ResourceSlices to get removed eventually.
	// The exception is when the driver is part of a rolling update and is torn down first.
	ExpectResourceSliceRemoval bool

	// Name gets derived automatically from the current test namespace and
	// (if set) the NameSuffix while setting up the driver for a test.
	Name string

	// Nodes contains entries for each node selected for a test when the test runs.
	// In addition, there is one entry for a fictional node.
	Nodes map[string]KubeletPlugin

	// IsLocal can be set to true when using local-up-cluster.sh *and* ensuring
	// that /var/lib/kubelet/plugins, /var/lib/kubelet/plugins_registry and
	// /var/run/cdi are writable by the current user.
	IsLocal bool

	NodeV1      bool
	NodeV1beta1 bool

	// Register the DRA test driver with the kubelet and expect DRA to work (= feature.DynamicResourceAllocation).
	WithKubelet bool

	// UsePrivilegedClient lets the test driver publish cluster-wide ResourceSlices.
	// The default node-scoped client is intentionally restricted by admission to
	// ResourceSlices for its own node.
	UsePrivilegedClient bool

	// Run driver pods. If false, only set up slices and class.
	WithRealNodes bool

	EnableDeviceMetadata   bool
	DeviceMetadataVersions []schema.GroupVersion // Must be non-empty when EnableDeviceMetadata is true.

	// ReconcilePoolWithName configures the ResourceSlice controller in each
	// test driver plugin to reconcile only the pool with this name.
	ReconcilePoolWithName string

	// expectDroppedFields, when true, suppresses the test failure that the
	// driver's error handler otherwise raises when the apiserver drops fields
	// from a published ResourceSlice (a resourceslice.DroppedFieldsError). The
	// feature gate cycle test sets this while a gate is intentionally off,
	// because the apiserver is then expected to drop the gated fields. It is
	// read from the resourceslice controller's background goroutine, so access
	// goes through an atomic.
	expectDroppedFields atomic.Bool

	mutex      sync.Mutex
	fail       map[MethodInstance]bool
	callCounts map[MethodInstance]int64
}

type KubeletPlugin struct {
	*app.ExamplePlugin
	ClientSet kubernetes.Interface
}

func (d *Driver) initName(tCtx ktesting.TContext) {
	d.Name = tCtx.Namespace() + d.NameSuffix + ".k8s.io"
}

func (d *Driver) SetNameSuffix(tCtx ktesting.TContext, suffix string) {
	d.NameSuffix = suffix
	d.initName(tCtx)
}

// deploymentID returns an identifier for Kubernetes objects (ServiceAccount,
// ClusterRole, etc.) derived from the driver name and instance suffix. The full
// driver name is kept in d.Name for the API and kubelet plugin. When it is too
// long for object name limits a short hashed form is used.
func (d *Driver) deploymentID() string {
	base := d.Name + d.InstanceSuffix
	const maxLen = 40 // leaves room for "dra-kubelet-plugin-" and "-service-account"
	if len(base) <= maxLen {
		return base
	}
	sum := sha256.Sum256([]byte(base))
	return "dra-" + hex.EncodeToString(sum[:8])
}

func (d *Driver) SetUp(tCtx ktesting.TContext, kubeletRootDir string, nodes *Nodes, driverResources map[string]resourceslice.DriverResources) {
	tCtx.Logf("deploying driver %s on nodes %v", d.Name, nodes.NodeNames)
	d.Nodes = make(map[string]KubeletPlugin)

	tCtx = tCtx.WithCancel()
	logger := klog.FromContext(tCtx)
	logger = klog.LoggerWithValues(logger, "driverName", d.Name)
	if d.InstanceSuffix != "" {
		instance, _ := strings.CutPrefix(d.InstanceSuffix, "-")
		logger = klog.LoggerWithValues(logger, "instance", instance)
	}
	tCtx = tCtx.WithLogger(logger)
	d.cleanup = append(d.cleanup, func(ktesting.TContext) { tCtx.Cancel("cleaning up test") })

	// After shutdown, check that all ResourceSlices were removed, either by the kubelet
	// or our own test code. This runs last because it gets registered first.
	if d.ExpectResourceSliceRemoval {
		tCtx.CleanupCtx(d.IsGone)
	}

	driverResource, useMultiHostDriverResources := driverResources[multiHostDriverResources]
	if useMultiHostDriverResources || !d.WithKubelet || d.UsePrivilegedClient {
		// We have to remove ResourceSlices ourselves.
		// Otherwise the kubelet does it after unregistering the driver. A
		// privileged client can create cluster-wide slices which the kubelet
		// does not own and therefore cannot remove.
		tCtx.CleanupCtx(func(tCtx ktesting.TContext) {
			err := tCtx.Client().ResourceV1().ResourceSlices().DeleteCollection(tCtx, metav1.DeleteOptions{}, metav1.ListOptions{FieldSelector: resourceapi.ResourceSliceSelectorDriver + "=" + d.Name})
			tCtx.ExpectNoError(err, "delete ResourceSlices of the driver")
		})
	}

	// If found, we create ResourceSlices that are associated with multiple nodes
	// through the node selector. Thus, the ResourceSlices are published here
	// rather than through the driver on a specific node.
	if useMultiHostDriverResources {
		for poolName, pool := range driverResource.Pools {
			for i, slice := range pool.Slices {
				resourceSlice := &resourceapi.ResourceSlice{
					ObjectMeta: metav1.ObjectMeta{
						Name: fmt.Sprintf("%s-%d", d.Name, i), // globally unique
					},
					Spec: resourceapi.ResourceSliceSpec{
						Driver: d.Name,
						Pool: resourceapi.ResourcePool{
							Name:               poolName,
							Generation:         pool.Generation,
							ResourceSliceCount: int64(len(pool.Slices)),
						},
						NodeSelector:           pool.NodeSelector,
						Devices:                slice.Devices,
						SharedCounters:         slice.SharedCounters,
						PerDeviceNodeSelection: slice.PerDeviceNodeSelection,
						PartitionTypeAttribute: slice.PartitionTypeAttribute,
					},
				}
				_, err := tCtx.Client().ResourceV1().ResourceSlices().Create(tCtx, resourceSlice, metav1.CreateOptions{})
				tCtx.ExpectNoError(err)
			}
		}
	}

	if !d.WithRealNodes {
		// Slices have been created as usual.
		// We don't actually have nodes, so
		// running pods wouldn't work and can be skipped.
		return
	}

	// Create service account and corresponding RBAC rules.
	deploymentID := d.deploymentID()
	d.serviceAccountName = "dra-kubelet-plugin-" + deploymentID + "-service-account"
	content := example.PluginPermissions

	content = strings.ReplaceAll(content, "dra-kubelet-plugin-namespace", tCtx.Namespace())
	content = strings.ReplaceAll(content, "dra-kubelet-plugin-driver-name", d.Name)
	content = strings.ReplaceAll(content, "dra-kubelet-plugin", "dra-kubelet-plugin-"+deploymentID)
	d.createFromYAML(tCtx, []byte(content), tCtx.Namespace())

	// Figure out which hostpathplugin to use: basically the latest one
	// from the test/e2e/testing-manifests/storage-csi manifests. That is
	// where SIG Storage maintains the versions of the hostpath image which
	// are part of Kubernetes E2E testing. test/utils/image parses those files.
	//
	// We piggy-back on that instead of controlling the version ourselves
	// because it reduces effort, at the risk of unexpected
	// breakage. Another benefit is that -list-images and registry patching
	// via test/utils/image + KUBE_TEST_REPO_LIST work.
	hostPathImage := "registry.k8s.io/sig-storage/hostpathplugin"
	hostPathVersion := ""
	for _, config := range image.GetOriginalImageConfigs() {
		parts := strings.SplitN(config.GetE2EImage(), ":", 2)
		if len(parts) < 2 {
			continue
		}
		image, version := parts[0], parts[1]
		if image != hostPathImage {
			continue
		}
		// "Dumb" string comparison is good enough for e.g. v1.16.1 < v1.17.0.
		// It seems unlikely that any major/patch will need more than one digit
		// or that version grow beyond 99.
		if hostPathVersion == "" || hostPathVersion < version {
			hostPathVersion = version
		}
	}
	origImageURL := hostPathImage + ":" + hostPathVersion
	patchedImageURL, err := image.ReplaceRegistryInImageURL(origImageURL)
	tCtx.ExpectNoError(err, "look up E2E image")

	// Using a ReplicaSet instead of a DaemonSet has the advantage that we can control
	// the lifecycle explicitly, in particular run two pods per node long enough to
	// run checks.
	manifests := []string{
		// The code below matches the content of this manifest (ports,
		// container names, etc.).
		"test/e2e/testing-manifests/dra/dra-test-driver-proxy.yaml",
	}
	instanceKey := "app.kubernetes.io/instance"
	rsName := ""
	numNodes := int32(len(nodes.NodeNames))
	pluginDataDirectoryPath := path.Join(kubeletRootDir, "plugins", d.Name)
	registrarDirectoryPath := path.Join(kubeletRootDir, "plugins_registry")
	instanceName := d.Name + d.InstanceSuffix
	err = utils.CreateFromManifestsTCtx(tCtx, func(item interface{}) error {
		switch item := item.(type) {
		case *appsv1.ReplicaSet:
			item.Name += d.NameSuffix + d.InstanceSuffix
			rsName = item.Name
			item.Spec.Replicas = &numNodes
			item.Spec.Selector.MatchLabels[instanceKey] = instanceName
			item.Spec.Template.Labels[instanceKey] = instanceName
			item.Spec.Template.Spec.Containers[0].Image = patchedImageURL
			item.Spec.Template.Spec.ServiceAccountName = d.serviceAccountName
			item.Spec.Template.Spec.Affinity.PodAntiAffinity.RequiredDuringSchedulingIgnoredDuringExecution[0].LabelSelector.MatchLabels[instanceKey] = instanceName
			item.Spec.Template.Spec.Affinity.NodeAffinity = &v1.NodeAffinity{
				RequiredDuringSchedulingIgnoredDuringExecution: &v1.NodeSelector{
					NodeSelectorTerms: []v1.NodeSelectorTerm{
						{
							MatchExpressions: []v1.NodeSelectorRequirement{
								{
									Key:      "kubernetes.io/hostname",
									Operator: v1.NodeSelectorOpIn,
									Values:   nodes.NodeNames,
								},
							},
						},
					},
				},
			}
			if d.IsLocal {
				// Drop mounting of directories. All operations run locally.
				item.Spec.Template.Spec.Volumes = nil
				item.Spec.Template.Spec.Containers[0].VolumeMounts = nil
				// No privileges required either.
				item.Spec.Template.Spec.SecurityContext = nil
				item.Spec.Template.Spec.Containers[0].SecurityContext = nil
			} else {
				item.Spec.Template.Spec.Volumes[0].HostPath.Path = pluginDataDirectoryPath
				item.Spec.Template.Spec.Volumes[1].HostPath.Path = registrarDirectoryPath
				item.Spec.Template.Spec.Containers[0].VolumeMounts[0].MountPath = pluginDataDirectoryPath
				item.Spec.Template.Spec.Containers[0].VolumeMounts[1].MountPath = registrarDirectoryPath
			}
		}
		return nil
	}, manifests...)
	tCtx.ExpectNoError(err, "deploy kubelet plugin replicaset")

	rs, err := tCtx.Client().AppsV1().ReplicaSets(tCtx.Namespace()).Get(tCtx, rsName, metav1.GetOptions{})
	tCtx.ExpectNoError(err, "get replicaset")

	// Wait for all pods to be running.
	if err := e2ereplicaset.WaitForReplicaSetTargetAvailableReplicas(tCtx, tCtx.Client(), rs, numNodes); err != nil {
		tCtx.ExpectNoError(err, "all kubelet plugin proxies running")
	}
	requirement, err := labels.NewRequirement(instanceKey, selection.Equals, []string{instanceName})
	tCtx.ExpectNoError(err, "create label selector requirement")
	selector := labels.NewSelector().Add(*requirement)
	pods, err := tCtx.Client().CoreV1().Pods(tCtx.Namespace()).List(tCtx, metav1.ListOptions{LabelSelector: selector.String()})
	tCtx.ExpectNoError(err, "list proxy pods")
	tCtx.Expect(numNodes).To(gomega.Equal(int32(len(pods.Items))), "number of proxy pods")
	sort.Slice(pods.Items, func(i, j int) bool {
		return pods.Items[i].Spec.NodeName < pods.Items[j].Spec.NodeName
	})

	// Run registrar and plugin for each of the pods.
	for _, pod := range pods.Items {
		nodename := pod.Spec.NodeName

		// Authenticate the plugin so that it has the exact same
		// permissions as the daemonset pod. This includes RBAC and a
		// validating admission policy which limits writes to per-node
		// ResourceSlices.
		//
		// We could retrieve
		// /var/run/secrets/kubernetes.io/serviceaccount/token from
		// each pod and use it. That would check that
		// ServiceAccountTokenNodeBindingValidation works. But that's
		// better covered by a test owned by SIG Auth (like the one in
		// https://github.com/kubernetes/kubernetes/pull/124711).
		//
		// Here we merely use impersonation, which is faster.
		driverClient := d.ImpersonateKubeletPlugin(tCtx, &pod)
		if d.UsePrivilegedClient {
			driverClient = tCtx.Client()
		}

		logger := klog.LoggerWithValues(klog.LoggerWithName(logger, "kubelet-plugin"), "node", pod.Spec.NodeName, "pod", klog.KObj(&pod))
		loggerCtx := klog.NewContext(tCtx, logger)
		fileOps := app.FileOperations{
			Create: func(name string, content []byte) error {
				logger.Info("creating CDI file", "node", nodename, "filename", name, "content", string(content))
				if d.IsLocal {
					// Name starts with /cdi, which is how it is mapped in the container.
					// Here we need it under /var/run.
					// Try to create /var/run/cdi, it might not exist yet.
					name = path.Join("/var/run", name)
					if err := os.MkdirAll(path.Dir(name), 0700); err != nil {
						return fmt.Errorf("create CDI directory: %w", err)
					}
					if err := os.WriteFile(name, content, 0644); err != nil {
						return fmt.Errorf("write CDI file: %w", err)
					}
					return nil
				}
				return d.createFile(tCtx, &pod, name, content)
			},
			Remove: func(name string) error {
				logger.Info("deleting CDI file", "node", nodename, "filename", name)
				if d.IsLocal {
					name = path.Join("/var/run", name)
					// Ignore the file already being gone. NodeUnprepareResources
					// must be idempotent, and during a feature gate cycle the
					// kubelet restarts and can unprepare the same claim more than
					// once. This matches the default FileOperations.Remove.
					if err := os.Remove(name); err != nil && !os.IsNotExist(err) {
						return err
					}
					return nil
				}
				return d.removeFile(tCtx, &pod, name)
			},
			HandleError: func(ctx context.Context, err error, msg string) {
				// Record a failure, but don't kill the background goroutine.
				// TODO: add to TContext or do it in Error/Assert/etc?
				defer ginkgo.GinkgoRecover()

				// During tests when canceling the context it is possible to get all kinds of
				// follow-up errors for that, like:
				//   processing ResourceSlice objects: retrieve node "127.0.0.1": client rate limiter Wait returned an error: context canceled
				//
				// The "context canceled" error was not wrapped, so `errors.Is` doesn't work.
				// Instead of trying to detect errors which can be ignored, let's only
				// treat errors as failures which definitely shouldn't occur:
				var droppedFields *resourceslice.DroppedFieldsError
				if errors.As(err, &droppedFields) && !d.expectDroppedFields.Load() {
					tCtx.Errorf("driver %s: %v", d.Name, err)
				}
			},
		}

		if dr, ok := driverResources[nodename]; ok {
			fileOps.DriverResources = &dr
		}
		// All listeners running in this pod use a new unique local port number
		// by atomically incrementing this variable.
		var listenerPort atomic.Int32
		listenerPort.Store(9000)
		rollingUpdateUID := pod.UID
		serialize := true
		if !d.RollingUpdate {
			rollingUpdateUID = ""
			// A test might have to execute two gRPC calls in parallel, so only
			// serialize when we explicitly want to test a rolling update.
			serialize = false
		}

		pluginOpts := []any{
			app.Options{EnableHealthService: true},
			kubeletplugin.GRPCVerbosity(0),
			kubeletplugin.GRPCInterceptor(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
				return d.interceptor(nodename, ctx, req, info, handler)
			}),
			kubeletplugin.GRPCStreamInterceptor(func(srv interface{}, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) (err error) {
				return d.streamInterceptor(nodename, srv, ss, info, handler)
			}),
			kubeletplugin.NodeV1(d.NodeV1),
			kubeletplugin.NodeV1beta1(d.NodeV1beta1),

			kubeletplugin.RollingUpdate(rollingUpdateUID),
			kubeletplugin.Serialize(serialize),
			kubeletplugin.FlockDirectoryPath(nodes.tempDir),

			kubeletplugin.PluginDataDirectoryPath(pluginDataDirectoryPath),
			kubeletplugin.PluginListener(d.listen(tCtx, &pod, &listenerPort)),

			kubeletplugin.RegistrarDirectoryPath(registrarDirectoryPath),
			kubeletplugin.RegistrarListener(d.listen(tCtx, &pod, &listenerPort)),

			kubeletplugin.EnableDeviceMetadata(d.EnableDeviceMetadata, d.DeviceMetadataVersions),
		}
		if d.ReconcilePoolWithName != "" {
			pluginOpts = append(pluginOpts, kubeletplugin.ReconcilePoolWithName(d.ReconcilePoolWithName))
		}
		if d.EnableDeviceMetadata {
			if !d.IsLocal {
				pluginOpts = append(pluginOpts,
					kubeletplugin.MetadataFileOps(d.buildRemoteMetadataFileOps(tCtx, &pod)),
					kubeletplugin.CDIDirectory("/cdi"),
				)
			}
		}

		plugin, err := app.StartPlugin(loggerCtx, "/cdi", d.Name, driverClient, nodename, fileOps, pluginOpts...)
		tCtx.ExpectNoError(err, "start kubelet plugin for node %s", pod.Spec.NodeName)
		d.cleanup = append(d.cleanup, func(tCtx ktesting.TContext) {
			// Depends on cancel being called first.
			plugin.Stop()
		})
		d.Nodes[nodename] = KubeletPlugin{ExamplePlugin: plugin, ClientSet: driverClient}
	}

	// Scale down the proxy ReplicaSet after all per-node plugins have
	// been stopped.
	d.cleanup = append(d.cleanup, func(tCtx ktesting.TContext) {
		tCtx.Log("scaling down driver proxy pods for", d.Name)
		rs, err := tCtx.Client().AppsV1().ReplicaSets(tCtx.Namespace()).Get(tCtx, rsName, metav1.GetOptions{})
		tCtx.ExpectNoError(err, "get ReplicaSet for driver "+d.Name)
		rs.Spec.Replicas = ptr.To(int32(0))
		rs, err = tCtx.Client().AppsV1().ReplicaSets(tCtx.Namespace()).Update(tCtx, rs, metav1.UpdateOptions{})
		tCtx.ExpectNoError(err, "scale down ReplicaSet for driver "+d.Name)
		if err := e2ereplicaset.WaitForReplicaSetTargetAvailableReplicas(tCtx, tCtx.Client(), rs, 0); err != nil {
			tCtx.ExpectNoError(err, "all kubelet plugin proxies stopped")
		}
	})

	if !d.WithKubelet {
		return
	}

	// Wait for registration.
	tCtx.Log("wait for plugin registration")
	tCtx.Eventually(func(tCtx ktesting.TContext) map[string][]app.GRPCCall {
		notRegistered := make(map[string][]app.GRPCCall)
		for nodename, plugin := range d.Nodes {
			calls := plugin.GetGRPCCalls()
			if contains, err := testdrivergomega.BeRegistered.Match(calls); err != nil || !contains {
				notRegistered[nodename] = calls
			}
		}
		return notRegistered
	}).WithTimeout(time.Minute).Should(gomega.BeEmpty(), "hosts where the plugin has not been registered yet")
}

func (d *Driver) ImpersonateKubeletPlugin(tCtx ktesting.TContext, pod *v1.Pod) kubernetes.Interface {
	tCtx.Helper()
	driverUserInfo := (&serviceaccount.ServiceAccountInfo{
		Name:      d.serviceAccountName,
		Namespace: pod.Namespace,
		NodeName:  pod.Spec.NodeName,
		PodName:   pod.Name,
		PodUID:    string(pod.UID),
	}).UserInfo()
	driverClientConfig := tCtx.RESTConfig()
	driverClientConfig.Impersonate = rest.ImpersonationConfig{
		UserName: driverUserInfo.GetName(),
		Groups:   driverUserInfo.GetGroups(),
		Extra:    driverUserInfo.GetExtra(),
	}
	driverClient, err := kubernetes.NewForConfig(driverClientConfig)
	tCtx.ExpectNoError(err, "create client for driver")
	return driverClient
}

func (d *Driver) createFile(tCtx ktesting.TContext, pod *v1.Pod, name string, content []byte) error {
	buffer := bytes.NewBuffer(content)
	// Writing the content can be slow. Better create a temporary file and
	// move it to the final destination once it is complete.
	tmpName := name + ".tmp"
	if err := d.podIO(tCtx, pod).CreateFile(tmpName, buffer); err != nil {
		_ = d.podIO(tCtx, pod).RemoveAll(tmpName)
		return err
	}
	return d.podIO(tCtx, pod).Rename(tmpName, name)
}

func (d *Driver) removeFile(tCtx ktesting.TContext, pod *v1.Pod, name string) error {
	return d.podIO(tCtx, pod).RemoveAll(name)
}

func (d *Driver) createFromYAML(tCtx ktesting.TContext, content []byte, namespace string) {
	// Not caching the discovery result isn't very efficient, but good enough.
	discoveryCache := memory.NewMemCacheClientWithContext(tCtx.Client().Discovery())
	restMapper := restmapper.NewDeferredDiscoveryRESTMapperWithContext(discoveryCache)

	for _, content := range bytes.Split(content, []byte("---\n")) {
		if len(content) == 0 {
			continue
		}

		var obj *unstructured.Unstructured
		tCtx.ExpectNoError(yaml.UnmarshalStrict(content, &obj), fmt.Sprintf("Full YAML:\n%s\n", string(content)))

		gv, err := schema.ParseGroupVersion(obj.GetAPIVersion())
		tCtx.ExpectNoError(err, fmt.Sprintf("extract group+version from object %q", klog.KObj(obj)))
		gk := schema.GroupKind{Group: gv.Group, Kind: obj.GetKind()}

		mapping, err := restMapper.RESTMappingWithContext(tCtx, gk, gv.Version)
		tCtx.ExpectNoError(err, fmt.Sprintf("map %q to resource", gk))

		resourceClient := tCtx.Dynamic().Resource(mapping.Resource)
		options := metav1.CreateOptions{
			// If the YAML input is invalid, then we want the
			// apiserver to tell us via an error. This can
			// happen because decoding into an unstructured object
			// doesn't validate.
			FieldValidation: "Strict",
		}
		switch mapping.Scope.Name() {
		case meta.RESTScopeNameRoot:
			_, err = resourceClient.Create(tCtx, obj, options)
		case meta.RESTScopeNameNamespace:
			if namespace == "" {
				tCtx.Fatalf("need namespace for object type %s", gk)
			}
			_, err = resourceClient.Namespace(namespace).Create(tCtx, obj, options)
		}
		tCtx.ExpectNoError(err, "create object")
		tCtx.CleanupCtx(func(tCtx ktesting.TContext) {
			del := resourceClient.Delete
			if mapping.Scope.Name() == meta.RESTScopeNameNamespace {
				del = resourceClient.Namespace(namespace).Delete
			}
			err := del(tCtx, obj.GetName(), metav1.DeleteOptions{})
			if !apierrors.IsNotFound(err) {
				tCtx.ExpectNoError(err, fmt.Sprintf("deleting %s.%s %s", obj.GetKind(), obj.GetAPIVersion(), klog.KObj(obj)))
			}
		})
	}
}

func (d *Driver) podIO(tCtx ktesting.TContext, pod *v1.Pod) proxy.PodDirIO {
	logger := tCtx.Logger()
	return proxy.PodDirIO{
		TCtx:          tCtx,
		Namespace:     pod.Namespace,
		PodName:       pod.Name,
		ContainerName: pod.Spec.Containers[0].Name,
		Logger:        &logger,
	}
}

func (d *Driver) buildRemoteMetadataFileOps(tCtx ktesting.TContext, pod *v1.Pod) kubeletplugin.MetadataFileOperations {
	execInPod := func(command []string) (string, error) {
		stdout, stderr, err := e2epod.Exec(tCtx, e2epod.ExecOptions{
			Command:       command,
			Namespace:     pod.Namespace,
			PodName:       pod.Name,
			ContainerName: pod.Spec.Containers[0].Name,
			CaptureStdout: true,
			CaptureStderr: true,
			Quiet:         true,
		})
		if err != nil {
			return "", fmt.Errorf("%v: stderr=%q, %w", command, stderr, err)
		}
		return stdout, nil
	}

	return kubeletplugin.MetadataFileOperations{
		WriteFile: func(name string, data []byte, perm os.FileMode) error {
			return d.createFile(tCtx, pod, name, data)
		},
		ReadFile: func(name string) ([]byte, error) {
			stdout, err := execInPod([]string{"cat", name})
			if err != nil {
				return nil, err
			}
			return []byte(stdout), nil
		},
		MkdirAll: func(p string, perm os.FileMode) error {
			_, err := execInPod([]string{"mkdir", "-p", p})
			return err
		},
		RemoveAll: func(p string) error {
			return d.podIO(tCtx, pod).RemoveAll(p)
		},
		Remove: func(name string) error {
			_, err := execInPod([]string{"rm", "-f", name})
			return err
		},
		Glob: func(pattern string) ([]string, error) {
			stdout, err := execInPod([]string{"sh", "-c", fmt.Sprintf("ls -1d %s 2>/dev/null || true", pattern)})
			if err != nil {
				return nil, err
			}
			stdout = strings.TrimSpace(stdout)
			if stdout == "" {
				return nil, nil
			}
			return strings.Split(stdout, "\n"), nil
		},
	}
}

// errListenerDone is the special error that we use to shut down.
// It doesn't need to be logged.
var errListenerDone = errors.New("listener is shutting down")

// listen returns the function which the kubeletplugin helper needs to open a listening socket.
// For that it spins up hostpathplugin in the pod for the desired node
// and connects to hostpathplugin via port forwarding.
func (d *Driver) listen(tCtx ktesting.TContext, pod *v1.Pod, port *atomic.Int32) func(ctx context.Context, endpoint string) (net.Listener, error) {
	return func(ctx context.Context, endpoint string) (l net.Listener, e error) {
		// No need create sockets, the kubelet is not expected to use them.
		if !d.WithKubelet {
			return newNullListener(), nil
		}

		// Try opening the socket directly on the local host. Falls back to pod if that fails.
		// Closing the listener will unlink the socket.
		if d.IsLocal {
			dir := path.Dir(endpoint)
			if err := os.MkdirAll(dir, 0755); err != nil {
				return nil, err
			}
			return net.ListenUnix("unix", &net.UnixAddr{Name: endpoint, Net: "unix"})
		}

		// "Allocate" a new port by by bumping the per-pod counter by one.
		port := port.Add(1)

		logger := klog.FromContext(ctx)
		logger = klog.LoggerWithName(logger, "socket-listener")
		logger = klog.LoggerWithValues(logger, "endpoint", endpoint, "port", port)
		ctx = klog.NewContext(ctx, logger)

		// Start hostpathplugin in proxy mode and keep it running until the listener gets closed.
		req := tCtx.Client().CoreV1().RESTClient().Post().
			Resource("pods").
			Namespace(tCtx.Namespace()).
			Name(pod.Name).
			SubResource("exec").
			VersionedParams(&v1.PodExecOptions{
				Container: pod.Spec.Containers[0].Name,
				Command: []string{
					"/hostpathplugin",
					"--v=5",
					"--endpoint=" + endpoint,
					fmt.Sprintf("--proxy-endpoint=tcp://:%d", port),
				},
				Stdout: true,
				Stderr: true,
			}, scheme.ParameterCodec)
		var wg sync.WaitGroup
		wg.Add(1)
		cmdCtx, cmdCancel := context.WithCancelCause(ctx)
		go func() {
			defer wg.Done()
			cmdLogger := klog.LoggerWithName(logger, "hostpathplugin")
			cmdCtx := klog.NewContext(cmdCtx, cmdLogger)
			logger.V(1).Info("Starting...")
			defer logger.V(1).Info("Stopped")

			// This may fail temporarily, which is recoverable by executing again.
			delayFn := wait.Backoff{
				Duration: time.Second,
				Cap:      30 * time.Second,
				Steps:    30,
				Factor:   2.0,
				Jitter:   1.0,
			}.DelayWithReset(clock.RealClock{}, 5*time.Minute)
			runHostpathPlugin := func(ctx context.Context) (bool, error) {
				// errors.Is(err, listenerDoneErr) would be nicer, but we don't get
				// that error from remotecommand. Instead forgo logging when we already shut down.
				if err := execute(ctx, req.URL(), tCtx.RESTConfig(), 5); err != nil && ctx.Err() == nil {
					klog.FromContext(ctx).V(5).Info("execution failed, will retry", "err", err)
				}
				// There is no reason to stop except for context cancellation =>
				// condition always false, no fatal errors.
				return false, nil
			}
			_ = delayFn.Until(cmdCtx, true /* immediate */, true /* sliding */, runHostpathPlugin)

			// Killing hostpathplugin does not remove the socket. Need to do that manually.
			req := tCtx.Client().CoreV1().RESTClient().Post().
				Resource("pods").
				Namespace(tCtx.Namespace()).
				Name(pod.Name).
				SubResource("exec").
				VersionedParams(&v1.PodExecOptions{
					Container: pod.Spec.Containers[0].Name,
					Command: []string{
						"rm",
						"-f",
						endpoint,
					},
					Stdout: true,
					Stderr: true,
				}, scheme.ParameterCodec)
			cleanupLogger := klog.LoggerWithName(logger, "cleanup")
			cleanupCtx := klog.NewContext(ctx, cleanupLogger)
			if err := execute(cleanupCtx, req.URL(), tCtx.RESTConfig(), 0); err != nil {
				cleanupLogger.Error(err, "Socket removal failed")
			}
		}()
		defer func() {
			// If we don't return a functional listener, then clean up.
			if e != nil {
				cmdCancel(e)
			}
		}()
		stopHostpathplugin := func() {
			cmdCancel(errListenerDone)
			wg.Wait()
		}

		addr := proxy.Addr{
			Namespace:     tCtx.Namespace(),
			PodName:       pod.Name,
			ContainerName: pod.Spec.Containers[0].Name,
			Port:          int(port),
		}
		listener, err := proxy.Listen(ctx, tCtx.Client(), tCtx.RESTConfig(), addr)
		if err != nil {
			return nil, fmt.Errorf("listen for connections from %+v: %w", addr, err)
		}
		return &listenerWithClose{Listener: listener, close: stopHostpathplugin}, nil
	}
}

// listenerWithClose wraps Close so that it also shuts down hostpathplugin.
type listenerWithClose struct {
	net.Listener
	close func()
}

func (l *listenerWithClose) Close() error {
	// First close connections, then shut down the remote command.
	// Otherwise the connection code is unhappy and logs errors.
	err := l.Listener.Close()
	l.close()
	return err
}

func newNullListener() net.Listener {
	ctx, cancel := context.WithCancelCause(context.Background())
	return &nullListener{ctx: ctx, cancel: cancel}
}

// nullListener blocks all Accept calls until the listener is closed.
type nullListener struct {
	ctx    context.Context
	cancel func(err error)
}

func (l *nullListener) Accept() (net.Conn, error) {
	<-l.ctx.Done()
	return nil, context.Cause(l.ctx)
}

func (l *nullListener) Close() error {
	l.cancel(errors.New("listener was closed"))
	return nil
}

func (l *nullListener) Addr() net.Addr {
	return &net.UnixAddr{}
}

// execute runs a remote command with stdout/stderr redirected to log messages at the chosen verbosity level.
func execute(ctx context.Context, url *url.URL, config *rest.Config, verbosity int) error {
	// Stream output as long as we run, i.e. ignore cancellation.
	stdout := pipe(context.WithoutCancel(ctx), "STDOUT", verbosity)
	stderr := pipe(context.WithoutCancel(ctx), "STDERR", verbosity)
	defer func() { _ = stdout.Close() }()
	defer func() { _ = stderr.Close() }()

	executor := exec.DefaultRemoteExecutor{}
	return executor.ExecuteWithContext(ctx, url, config, nil, stdout, stderr, false, nil)
}

// pipe creates an in-memory pipe and starts logging whatever is sent through that pipe in the background.
func pipe(ctx context.Context, msg string, verbosity int) *io.PipeWriter {
	logger := klog.FromContext(ctx)

	reader, writer := io.Pipe()
	go func() {
		buffer := make([]byte, 10*1024)
		for {
			n, err := reader.Read(buffer)
			if n > 0 {
				logger.V(verbosity).Info(msg, "msg", string(buffer[0:n]))
			}
			if err != nil {
				if !errors.Is(err, io.EOF) {
					logger.Error(err, msg)
				}
				reader.CloseWithError(err)
				return
			}
			if ctx.Err() != nil {
				reader.CloseWithError(context.Cause(ctx))
				return
			}
		}
	}()
	return writer
}

func (d *Driver) TearDown(tCtx ktesting.TContext) {
	for _, c := range d.cleanup {
		c(tCtx)
	}
	d.cleanup = nil
	d.wg.Wait()
}

// IsGone checks that the kubelet is done with the driver.
// This is done by waiting for the kubelet to remove the
// driver's ResourceSlices, which takes at least 30 seconds
// because of the delay in the kubelet.
//
// Only use this in tests where kubelet support for DRA is guaranteed.
func (d *Driver) IsGone(tCtx ktesting.TContext) {
	tCtx.Helper()
	tCtx.Logf("Waiting for ResourceSlices of driver %s to be removed...", d.Name)
	tCtx.Eventually(d.NewGetSlices()).WithTimeout(2 * time.Minute).Should(gomega.HaveField("Items", gomega.BeEmpty()))
}

func (d *Driver) interceptor(nodename string, ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	m := MethodInstance{nodename, info.FullMethod}
	d.callCounts[m]++
	if d.fail[m] {
		return nil, errors.New("injected error")
	}

	return handler(ctx, req)
}

func (d *Driver) streamInterceptor(nodename string, srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
	// Stream calls block for a long time. We must not hold the lock while
	// they are running.
	d.mutex.Lock()
	m := MethodInstance{nodename, info.FullMethod}
	d.callCounts[m]++
	fail := d.fail[m]
	d.mutex.Unlock()

	if fail {
		return errors.New("injected error")
	}

	return handler(srv, stream)
}

func (d *Driver) Fail(m MethodInstance, injectError bool) {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	d.fail[m] = injectError
}

func (d *Driver) CallCount(m MethodInstance) int64 {
	d.mutex.Lock()
	defer d.mutex.Unlock()

	return d.callCounts[m]
}

func (d *Driver) Nodenames() (nodenames []string) {
	for nodename := range d.Nodes {
		nodenames = append(nodenames, nodename)
	}
	sort.Strings(nodenames)
	return
}