github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-18 04:54:54 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #133401 from bart0sh/PR192-dra-e2e-node-test-connection-reuse

Browse Source 
e2e_node: test DRA plugin gRPC connection reuse
This commit is contained in:
Kubernetes Prow Robot
2025-09-17 11:26:10 -07:00
committed by GitHub
parent da92891c7c 871f87eaec
commit c23d1ac668
1 changed files with 155 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
test/e2e_node/dra_test.go
View File

@@ -511,6 +511,55 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami
ginkgo.Entry("for registrar", wrappedNewRegistrar, kubeletplugin.RegistrarListener),
ginkgo.Entry("for DRA service", newDRAService, kubeletplugin.PluginListener),
)
// This test verifies that the Kubelet establishes only a single gRPC connection
// with the DRA plugin throughout the plugin lifecycle.
//
// It does so by:
// - Using a custom gRPC listener that counts accepted connections.
// - Sequentially creating pods that trigger NodePrepareResources and NodeUnprepareResources.
// - Asserting that only one connection was accepted by the listener — since each
// call to net.Listener.Accept() corresponds to a new incoming connection —
// ensures that the plugin reused the same gRPC connection for all calls.
ginkgo.It("must use one gRPC connection for all service calls", func(ctx context.Context) {
var listener *countingListener
var err error
getListener := func(_ context.Context, socketPath string) (net.Listener, error) {
listener, err = newCountingListener(socketPath)
return listener, err
}
nodeName := getNodeName(ctx, f)
ginkgo.By("start DRA registrar")
registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for registration to complete")
gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered)
ginkgo.By("start DRA plugin service")
draService := newDRAService(ctx, f.ClientSet, f.Namespace.Name, nodeName, driverName, "", kubeletplugin.PluginListener(getListener))
for _, suffix := range []string{"-1", "-2"} {
draService.ResetGRPCCalls()
ginkgo.By("create test objects " + suffix)
pod := createTestObjects(ctx, f.ClientSet, nodeName, f.Namespace.Name, "draclass"+suffix, "external-claim"+suffix, "drapod"+suffix, true, []string{driverName})
ginkgo.By("wait for NodePrepareResources call to succeed " + suffix)
gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded)
ginkgo.By("wait for NodeUnprepareResources call to succeed " + suffix)
gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodeUnprepareResourcesSucceeded)
ginkgo.By("wait for pod to succeed " + suffix)
err = e2epod.WaitForPodSuccessInNamespace(ctx, f.ClientSet, pod.Name, f.Namespace.Name)
framework.ExpectNoError(err)
ginkgo.By("check that listener.Accept was called only once " + suffix)
gomega.Expect(listener.acceptCount()).To(gomega.Equal(1))
}
})
})
f.Context("Two resource Kubelet Plugins", f.WithSerial(), func() {
@@ -875,6 +924,72 @@ var _ = framework.SIGDescribe("node")(framework.WithLabel("DRA"), feature.Dynami
}).WithTimeout(60*time.Second).WithPolling(2*time.Second).Should(gomega.Equal("Healthy"), "Device health should recover and update to Healthy")
})
// This test verifies that the Kubelet establishes only a single gRPC connection
// with the DRA plugin throughout the plugin lifecycle.
//
// It does so by:
// - Using a custom gRPC listener that counts accepted connections.
// - Sequentially creating pods that trigger NodePrepareResources and NodeUnprepareResources.
// - Asserting that only one connection was accepted by the listener — since each
// call to net.Listener.Accept() corresponds to a new incoming connection —
// ensures that the plugin reused the same gRPC connection for all calls.
//
// NOTE: This is a copy of a similar test to test service connection reuse with enabled
// ResourceHealth. It should be merged with the original test when the ResourceHealth feature matures.
ginkgo.It("must reuse one gRPC connection for service and health-monitoring calls", func(ctx context.Context) {
var listener *countingListener
var err error
getListener := func(_ context.Context, socketPath string) (net.Listener, error) {
listener, err = newCountingListener(socketPath)
return listener, err
}
nodeName := getNodeName(ctx, f)
ginkgo.By("start DRA registrar")
registrar := newRegistrar(ctx, f.ClientSet, nodeName, driverName)
ginkgo.By("wait for registration to complete")
gomega.Eventually(registrar.GetGRPCCalls).WithTimeout(pluginRegistrationTimeout).Should(testdrivergomega.BeRegistered)
ginkgo.By("start DRA plugin service")
draService := newDRAService(ctx, f.ClientSet, f.Namespace.Name, nodeName, driverName, "", kubeletplugin.PluginListener(getListener))
for _, suffix := range []string{"-1", "-2"} {
className := "health-test-class" + suffix
claimName := "health-test-claim" + suffix
podName := "health-test-pod" + suffix
poolNameForTest := "pool-a" + suffix
deviceNameForTest := "dev-0" + suffix
draService.ResetGRPCCalls()
ginkgo.By("create test objects " + suffix)
pod := createHealthTestPodAndClaim(ctx, f, driverName, podName, claimName, className, poolNameForTest, deviceNameForTest)
ginkgo.By("wait for NodePrepareResources call to succeed " + suffix)
gomega.Eventually(draService.GetGRPCCalls).WithTimeout(retryTestTimeout).Should(testdrivergomega.NodePrepareResourcesSucceeded)
ginkgo.By("Waiting for the pod to be running")
framework.ExpectNoError(e2epod.WaitForPodRunningInNamespace(ctx, f.ClientSet, pod))
ginkgo.By("Forcing a 'Healthy' status update to establish a baseline")
draService.HealthControlChan <- testdriver.DeviceHealthUpdate{
PoolName: poolNameForTest,
DeviceName: deviceNameForTest,
Health: "Healthy",
}
ginkgo.By("Verifying device health is now Healthy in the pod status")
gomega.Eventually(ctx, func(ctx context.Context) (string, error) {
return getDeviceHealthFromAPIServer(f, pod.Namespace, pod.Name, driverName, claimName, poolNameForTest, deviceNameForTest)
}).WithTimeout(30*time.Second).WithPolling(1*time.Second).Should(gomega.Equal("Healthy"), "Device health should be Healthy after explicit update")
ginkgo.By("check that listener.Accept was called only once " + suffix)
gomega.Expect(listener.acceptCount()).To(gomega.Equal(1))
}
})
// Verifies that device health transitions to "Unknown" when a DRA plugin
// stops and recovers to "Healthy" upon plugin restart.
ginkgo.It("should update health to Unknown when plugin stops and recover upon restart", func(ctx context.Context) {
@@ -1540,3 +1655,43 @@ func driverNameLabelKey(element any) string {
el := element.(*testutil.Sample)
return string(el.Metric[testutil.LabelName("driver_name")])
}
// countingListener is a wrapper around net.Listener that counts the number of times
// Accept method is called. It embeds a net.Listener and uses atomic counters
// to safely track the number of accept operations.
type countingListener struct {
net.Listener
acceptCalls atomic.Int32
}
// newCountingListener creates a new Unix domain socket listener
// wrapped in a countingListener to track accepted connections.
func newCountingListener(socketPath string) (*countingListener, error) {
addr, err := net.ResolveUnixAddr("unix", socketPath)
if err != nil {
return nil, err
}
l, err := net.ListenUnix("unix", addr)
if err != nil {
return nil, err
}
return &countingListener{Listener: l}, nil
}
// Accept waits for and returns the accepted connection.
// It increments the acceptCalls counter each time the call succeeded.
func (l *countingListener) Accept() (net.Conn, error) {
conn, err := l.Listener.Accept()
if err == nil {
l.acceptCalls.Add(1)
}
return conn, err
}
// acceptCount returns the number of times the accept method
// has been called on the listener.
func (l *countingListener) acceptCount() int {
return int(l.acceptCalls.Load())
}