From d45a4557c1b9486e06a2247e955e02aae5254bcf Mon Sep 17 00:00:00 2001 From: Ed Bartosh Date: Fri, 26 Sep 2025 16:47:33 +0300 Subject: [PATCH] Migrate device manager to contextual logging --- hack/golangci-hints.yaml | 1 + hack/golangci.yaml | 1 + hack/logcheck.conf | 1 + pkg/kubelet/cm/container_manager_linux.go | 4 +- pkg/kubelet/cm/container_manager_windows.go | 7 +- pkg/kubelet/cm/devicemanager/endpoint.go | 18 +- pkg/kubelet/cm/devicemanager/endpoint_test.go | 68 +++--- pkg/kubelet/cm/devicemanager/manager.go | 174 ++++++++------- pkg/kubelet/cm/devicemanager/manager_test.go | 201 ++++++++++-------- .../cm/devicemanager/plugin/v1beta1/api.go | 11 +- .../cm/devicemanager/plugin/v1beta1/client.go | 42 ++-- .../devicemanager/plugin/v1beta1/handler.go | 66 +++--- .../cm/devicemanager/plugin/v1beta1/server.go | 43 ++-- .../cm/devicemanager/plugin/v1beta1/stub.go | 80 +++---- pkg/kubelet/cm/devicemanager/pod_devices.go | 46 ++-- .../cm/devicemanager/pod_devices_test.go | 4 +- .../cm/devicemanager/topology_hints.go | 22 +- .../cm/devicemanager/topology_hints_test.go | 7 +- pkg/kubelet/cm/devicemanager/types.go | 8 +- 19 files changed, 450 insertions(+), 354 deletions(-) diff --git a/hack/golangci-hints.yaml b/hack/golangci-hints.yaml index 005f869436d..648d044a5e0 100644 --- a/hack/golangci-hints.yaml +++ b/hack/golangci-hints.yaml @@ -236,6 +236,7 @@ linters: contextual k8s.io/kubernetes/pkg/securitycontext/.* contextual k8s.io/kubernetes/test/e2e/dra/.* contextual k8s.io/kubernetes/pkg/kubelet/certificate/.* + contextual k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/.* contextual k8s.io/kubernetes/pkg/kubelet/cm/dra/.* contextual k8s.io/kubernetes/pkg/kubelet/cm/memorymanager/.* contextual k8s.io/kubernetes/pkg/kubelet/lifecycle/.* diff --git a/hack/golangci.yaml b/hack/golangci.yaml index 3efcb8035e6..5d18a12f8d9 100644 --- a/hack/golangci.yaml +++ b/hack/golangci.yaml @@ -250,6 +250,7 @@ linters: contextual k8s.io/kubernetes/pkg/securitycontext/.* contextual k8s.io/kubernetes/test/e2e/dra/.* contextual k8s.io/kubernetes/pkg/kubelet/certificate/.* + contextual k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/.* contextual k8s.io/kubernetes/pkg/kubelet/cm/dra/.* contextual k8s.io/kubernetes/pkg/kubelet/cm/memorymanager/.* contextual k8s.io/kubernetes/pkg/kubelet/lifecycle/.* diff --git a/hack/logcheck.conf b/hack/logcheck.conf index 6b8a624b5fe..70d664d7da4 100644 --- a/hack/logcheck.conf +++ b/hack/logcheck.conf @@ -63,6 +63,7 @@ contextual k8s.io/kubernetes/pkg/security/.* contextual k8s.io/kubernetes/pkg/securitycontext/.* contextual k8s.io/kubernetes/test/e2e/dra/.* contextual k8s.io/kubernetes/pkg/kubelet/certificate/.* +contextual k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/.* contextual k8s.io/kubernetes/pkg/kubelet/cm/dra/.* contextual k8s.io/kubernetes/pkg/kubelet/cm/memorymanager/.* contextual k8s.io/kubernetes/pkg/kubelet/lifecycle/.* diff --git a/pkg/kubelet/cm/container_manager_linux.go b/pkg/kubelet/cm/container_manager_linux.go index c65053cd678..d15673839b8 100644 --- a/pkg/kubelet/cm/container_manager_linux.go +++ b/pkg/kubelet/cm/container_manager_linux.go @@ -689,7 +689,7 @@ func (cm *containerManagerImpl) Start(ctx context.Context, node *v1.Node, } // Starts device manager. - if err := cm.deviceManager.Start(devicemanager.ActivePodsFunc(activePods), sourcesReady, containerMap.Clone(), containerRunningSet); err != nil { + if err := cm.deviceManager.Start(klog.FromContext(ctx), devicemanager.ActivePodsFunc(activePods), sourcesReady, containerMap.Clone(), containerRunningSet); err != nil { return err } @@ -726,7 +726,7 @@ func (cm *containerManagerImpl) GetResources(ctx context.Context, pod *v1.Pod, c } // Allocate should already be called during predicateAdmitHandler.Admit(), // just try to fetch device runtime information from cached state here - devOpts, err := cm.deviceManager.GetDeviceRunContainerOptions(pod, container) + devOpts, err := cm.deviceManager.GetDeviceRunContainerOptions(ctx, pod, container) if err != nil { return nil, err } else if devOpts == nil { diff --git a/pkg/kubelet/cm/container_manager_windows.go b/pkg/kubelet/cm/container_manager_windows.go index 41f977168da..771b405c26f 100644 --- a/pkg/kubelet/cm/container_manager_windows.go +++ b/pkg/kubelet/cm/container_manager_windows.go @@ -80,7 +80,8 @@ func (cm *containerManagerImpl) Start(ctx context.Context, node *v1.Node, podStatusProvider status.PodStatusProvider, runtimeService internalapi.RuntimeService, localStorageCapacityIsolation bool) error { - klog.V(2).InfoS("Starting Windows container manager") + logger := klog.FromContext(ctx) + logger.V(2).Info("Starting Windows container manager") cm.nodeInfo = node @@ -110,7 +111,7 @@ func (cm *containerManagerImpl) Start(ctx context.Context, node *v1.Node, } // Starts device manager. - if err := cm.deviceManager.Start(devicemanager.ActivePodsFunc(activePods), sourcesReady, containerMap.Clone(), containerRunningSet); err != nil { + if err := cm.deviceManager.Start(logger, devicemanager.ActivePodsFunc(activePods), sourcesReady, containerMap.Clone(), containerRunningSet); err != nil { return err } @@ -265,7 +266,7 @@ func (cm *containerManagerImpl) GetResources(ctx context.Context, pod *v1.Pod, c opts := &kubecontainer.RunContainerOptions{} // Allocate should already be called during predicateAdmitHandler.Admit(), // just try to fetch device runtime information from cached state here - devOpts, err := cm.deviceManager.GetDeviceRunContainerOptions(pod, container) + devOpts, err := cm.deviceManager.GetDeviceRunContainerOptions(ctx, pod, container) if err != nil { return nil, err } else if devOpts == nil { diff --git a/pkg/kubelet/cm/devicemanager/endpoint.go b/pkg/kubelet/cm/devicemanager/endpoint.go index 31575812879..f9f0f2845be 100644 --- a/pkg/kubelet/cm/devicemanager/endpoint.go +++ b/pkg/kubelet/cm/devicemanager/endpoint.go @@ -30,9 +30,9 @@ import ( // for managing gRPC communications with the device plugin and caching // device states reported by the device plugin. type endpoint interface { - getPreferredAllocation(available, mustInclude []string, size int) (*pluginapi.PreferredAllocationResponse, error) - allocate(devs []string) (*pluginapi.AllocateResponse, error) - preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error) + getPreferredAllocation(ctx context.Context, available, mustInclude []string, size int) (*pluginapi.PreferredAllocationResponse, error) + allocate(ctx context.Context, devs []string) (*pluginapi.AllocateResponse, error) + preStartContainer(ctx context.Context, devs []string) (*pluginapi.PreStartContainerResponse, error) setStopTime(t time.Time) isStopped() bool stopGracePeriodExpired() bool @@ -83,11 +83,11 @@ func (e *endpointImpl) setStopTime(t time.Time) { } // getPreferredAllocation issues GetPreferredAllocation gRPC call to the device plugin. -func (e *endpointImpl) getPreferredAllocation(available, mustInclude []string, size int) (*pluginapi.PreferredAllocationResponse, error) { +func (e *endpointImpl) getPreferredAllocation(ctx context.Context, available, mustInclude []string, size int) (*pluginapi.PreferredAllocationResponse, error) { if e.isStopped() { return nil, fmt.Errorf(errEndpointStopped, e) } - return e.api.GetPreferredAllocation(context.Background(), &pluginapi.PreferredAllocationRequest{ + return e.api.GetPreferredAllocation(ctx, &pluginapi.PreferredAllocationRequest{ ContainerRequests: []*pluginapi.ContainerPreferredAllocationRequest{ { AvailableDeviceIDs: available, @@ -99,11 +99,11 @@ func (e *endpointImpl) getPreferredAllocation(available, mustInclude []string, s } // allocate issues Allocate gRPC call to the device plugin. -func (e *endpointImpl) allocate(devs []string) (*pluginapi.AllocateResponse, error) { +func (e *endpointImpl) allocate(ctx context.Context, devs []string) (*pluginapi.AllocateResponse, error) { if e.isStopped() { return nil, fmt.Errorf(errEndpointStopped, e) } - return e.api.Allocate(context.Background(), &pluginapi.AllocateRequest{ + return e.api.Allocate(ctx, &pluginapi.AllocateRequest{ ContainerRequests: []*pluginapi.ContainerAllocateRequest{ {DevicesIds: devs}, }, @@ -111,11 +111,11 @@ func (e *endpointImpl) allocate(devs []string) (*pluginapi.AllocateResponse, err } // preStartContainer issues PreStartContainer gRPC call to the device plugin. -func (e *endpointImpl) preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error) { +func (e *endpointImpl) preStartContainer(ctx context.Context, devs []string) (*pluginapi.PreStartContainerResponse, error) { if e.isStopped() { return nil, fmt.Errorf(errEndpointStopped, e) } - ctx, cancel := context.WithTimeout(context.Background(), pluginapi.KubeletPreStartContainerRPCTimeoutInSecs*time.Second) + ctx, cancel := context.WithTimeout(ctx, pluginapi.KubeletPreStartContainerRPCTimeoutInSecs*time.Second) defer cancel() return e.api.PreStartContainer(ctx, &pluginapi.PreStartContainerRequest{ DevicesIds: devs, diff --git a/pkg/kubelet/cm/devicemanager/endpoint_test.go b/pkg/kubelet/cm/devicemanager/endpoint_test.go index 530ccd3bd3a..d703fc8db28 100644 --- a/pkg/kubelet/cm/devicemanager/endpoint_test.go +++ b/pkg/kubelet/cm/devicemanager/endpoint_test.go @@ -17,6 +17,7 @@ limitations under the License. package devicemanager import ( + "context" "fmt" "os" "path/filepath" @@ -27,20 +28,22 @@ import ( "github.com/stretchr/testify/require" "google.golang.org/protobuf/proto" + "k8s.io/klog/v2" pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" plugin "k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/plugin/v1beta1" + "k8s.io/kubernetes/test/utils/ktesting" ) // monitorCallback is the function called when a device's health state changes, // or new devices are reported, or old devices are deleted. // Updated contains the most recent state of the Device. -type monitorCallback func(resourceName string, devices []*pluginapi.Device) +type monitorCallback func(logger klog.Logger, resourceName string, devices []*pluginapi.Device) func newMockPluginManager() *mockPluginManager { return &mockPluginManager{ func(string) error { return nil }, func(string, plugin.DevicePlugin) error { return nil }, - func(string) {}, + func(klog.Logger, string) {}, func(string, *pluginapi.ListAndWatchResponse) {}, } } @@ -48,7 +51,7 @@ func newMockPluginManager() *mockPluginManager { type mockPluginManager struct { cleanupPluginDirectory func(string) error pluginConnected func(string, plugin.DevicePlugin) error - pluginDisconnected func(string) + pluginDisconnected func(klog.Logger, string) pluginListAndWatchReceiver func(string, *pluginapi.ListAndWatchResponse) } @@ -56,15 +59,15 @@ func (m *mockPluginManager) CleanupPluginDirectory(r string) error { return m.cleanupPluginDirectory(r) } -func (m *mockPluginManager) PluginConnected(r string, p plugin.DevicePlugin) error { +func (m *mockPluginManager) PluginConnected(_ context.Context, r string, p plugin.DevicePlugin) error { return m.pluginConnected(r, p) } -func (m *mockPluginManager) PluginDisconnected(r string) { - m.pluginDisconnected(r) +func (m *mockPluginManager) PluginDisconnected(logger klog.Logger, r string) { + m.pluginDisconnected(logger, r) } -func (m *mockPluginManager) PluginListAndWatchReceiver(r string, lr *pluginapi.ListAndWatchResponse) { +func (m *mockPluginManager) PluginListAndWatchReceiver(_ klog.Logger, r string, lr *pluginapi.ListAndWatchResponse) { m.pluginListAndWatchReceiver(r, lr) } @@ -73,17 +76,19 @@ func esocketName() string { } func TestNewEndpoint(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) socket := filepath.Join(os.TempDir(), esocketName()) devs := []*pluginapi.Device{ {ID: "ADeviceId", Health: pluginapi.Healthy}, } - p, e := esetup(t, devs, socket, "mock", func(n string, d []*pluginapi.Device) {}) - defer ecleanup(t, p, e) + p, e := esetup(tCtx, t, devs, socket, "mock", func(logger klog.Logger, n string, d []*pluginapi.Device) {}) + defer ecleanup(logger, p, e) } func TestRun(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) socket := filepath.Join(os.TempDir(), esocketName()) devs := []*pluginapi.Device{ @@ -100,7 +105,7 @@ func TestRun(t *testing.T) { callbackCount := 0 callbackChan := make(chan int) - callback := func(n string, devices []*pluginapi.Device) { + callback := func(_ klog.Logger, n string, devices []*pluginapi.Device) { // Should be called twice: // one for plugin registration, one for plugin update. if callbackCount > 2 { @@ -133,10 +138,10 @@ func TestRun(t *testing.T) { callbackChan <- callbackCount } - p, e := esetup(t, devs, socket, "mock", callback) - defer ecleanup(t, p, e) + p, e := esetup(tCtx, t, devs, socket, "mock", callback) + defer ecleanup(logger, p, e) - go e.client.Run() + go e.client.Run(tCtx) // Wait for the first callback to be issued. <-callbackChan @@ -149,17 +154,18 @@ func TestRun(t *testing.T) { } func TestAllocate(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) socket := filepath.Join(os.TempDir(), esocketName()) devs := []*pluginapi.Device{ {ID: "ADeviceId", Health: pluginapi.Healthy}, } callbackCount := 0 callbackChan := make(chan int) - p, e := esetup(t, devs, socket, "mock", func(n string, d []*pluginapi.Device) { + p, e := esetup(tCtx, t, devs, socket, "mock", func(_ klog.Logger, n string, d []*pluginapi.Device) { callbackCount++ callbackChan <- callbackCount }) - defer ecleanup(t, p, e) + defer ecleanup(logger, p, e) resp := new(pluginapi.AllocateResponse) contResp := new(pluginapi.ContainerAllocateResponse) @@ -187,7 +193,7 @@ func TestAllocate(t *testing.T) { return resp, nil }) - go e.client.Run() + go e.client.Run(tCtx) // Wait for the callback to be issued. select { case <-callbackChan: @@ -196,20 +202,21 @@ func TestAllocate(t *testing.T) { t.FailNow() } - respOut, err := e.allocate([]string{"ADeviceId"}) + respOut, err := e.allocate(tCtx, []string{"ADeviceId"}) require.NoError(t, err) require.True(t, proto.Equal(resp, respOut)) } func TestGetPreferredAllocation(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) socket := filepath.Join(os.TempDir(), esocketName()) callbackCount := 0 callbackChan := make(chan int) - p, e := esetup(t, []*pluginapi.Device{}, socket, "mock", func(n string, d []*pluginapi.Device) { + p, e := esetup(tCtx, t, []*pluginapi.Device{}, socket, "mock", func(_ klog.Logger, n string, d []*pluginapi.Device) { callbackCount++ callbackChan <- callbackCount }) - defer ecleanup(t, p, e) + defer ecleanup(logger, p, e) resp := &pluginapi.PreferredAllocationResponse{ ContainerResponses: []*pluginapi.ContainerPreferredAllocationResponse{ @@ -221,7 +228,7 @@ func TestGetPreferredAllocation(t *testing.T) { return resp, nil }) - go e.client.Run() + go e.client.Run(tCtx) // Wait for the callback to be issued. select { case <-callbackChan: @@ -230,12 +237,13 @@ func TestGetPreferredAllocation(t *testing.T) { t.FailNow() } - respOut, err := e.getPreferredAllocation([]string{}, []string{}, -1) + respOut, err := e.getPreferredAllocation(tCtx, []string{}, []string{}, -1) require.NoError(t, err) require.True(t, proto.Equal(resp, respOut)) } -func esetup(t *testing.T, devs []*pluginapi.Device, socket, resourceName string, callback monitorCallback) (*plugin.Stub, *endpointImpl) { +func esetup(ctx context.Context, t *testing.T, devs []*pluginapi.Device, socket, resourceName string, callback monitorCallback) (*plugin.Stub, *endpointImpl) { + logger := klog.FromContext(ctx) m := newMockPluginManager() m.pluginListAndWatchReceiver = func(r string, resp *pluginapi.ListAndWatchResponse) { @@ -243,7 +251,7 @@ func esetup(t *testing.T, devs []*pluginapi.Device, socket, resourceName string, for _, d := range resp.Devices { newDevs = append(newDevs, d) } - callback(resourceName, newDevs) + callback(klog.FromContext(ctx), resourceName, newDevs) } var dp plugin.DevicePlugin @@ -255,12 +263,12 @@ func esetup(t *testing.T, devs []*pluginapi.Device, socket, resourceName string, return nil } - p := plugin.NewDevicePluginStub(devs, socket, resourceName, false, false) - err := p.Start() + p := plugin.NewDevicePluginStub(logger, devs, socket, resourceName, false, false) + err := p.Start(ctx) require.NoError(t, err) c := plugin.NewPluginClient(resourceName, socket, m) - err = c.Connect() + err = c.Connect(ctx) require.NoError(t, err) wg.Wait() @@ -268,14 +276,14 @@ func esetup(t *testing.T, devs []*pluginapi.Device, socket, resourceName string, e := newEndpointImpl(dp) e.client = c - m.pluginDisconnected = func(r string) { + m.pluginDisconnected = func(logger klog.Logger, r string) { e.setStopTime(time.Now()) } return p, e } -func ecleanup(t *testing.T, p *plugin.Stub, e *endpointImpl) { - p.Stop() - e.client.Disconnect() +func ecleanup(logger klog.Logger, p *plugin.Stub, e *endpointImpl) { + p.Stop(logger) + e.client.Disconnect(logger) } diff --git a/pkg/kubelet/cm/devicemanager/manager.go b/pkg/kubelet/cm/devicemanager/manager.go index 8074fa18e41..dd478cbee1e 100644 --- a/pkg/kubelet/cm/devicemanager/manager.go +++ b/pkg/kubelet/cm/devicemanager/manager.go @@ -130,15 +130,18 @@ func (s *sourcesReadyStub) AllReady() bool { return true } // NewManagerImpl creates a new manager. func NewManagerImpl(topology []cadvisorapi.Node, topologyAffinityStore topologymanager.Store) (*ManagerImpl, error) { + // Use klog.TODO() because we currently do not have a proper logger to pass in. + // Replace this with an appropriate context when refactoring this function to accept a logger parameter. + logger := klog.TODO() socketPath := pluginapi.KubeletSocket if runtime.GOOS == "windows" { socketPath = os.Getenv("SYSTEMDRIVE") + pluginapi.KubeletSocketWindows } - return newManagerImpl(socketPath, topology, topologyAffinityStore) + return newManagerImpl(logger, socketPath, topology, topologyAffinityStore) } -func newManagerImpl(socketPath string, topology []cadvisorapi.Node, topologyAffinityStore topologymanager.Store) (*ManagerImpl, error) { - klog.V(2).InfoS("Creating Device Plugin manager", "path", socketPath) +func newManagerImpl(logger klog.Logger, socketPath string, topology []cadvisorapi.Node, topologyAffinityStore topologymanager.Store) (*ManagerImpl, error) { + logger.V(2).Info("Creating Device Plugin manager", "path", socketPath) var numaNodes []int for _, node := range topology { @@ -159,7 +162,7 @@ func newManagerImpl(socketPath string, topology []cadvisorapi.Node, topologyAffi update: make(chan resourceupdates.Update, 100), } - server, err := plugin.NewServer(socketPath, manager, manager) + server, err := plugin.NewServer(logger, socketPath, manager, manager) if err != nil { return nil, fmt.Errorf("failed to create plugin server: %v", err) } @@ -186,7 +189,7 @@ func (m *ManagerImpl) Updates() <-chan resourceupdates.Update { // CleanupPluginDirectory is to remove all existing unix sockets // from /var/lib/kubelet/device-plugins on Device Plugin Manager start -func (m *ManagerImpl) CleanupPluginDirectory(dir string) error { +func (m *ManagerImpl) CleanupPluginDirectory(logger klog.Logger, dir string) error { d, err := os.Open(dir) if err != nil { return err @@ -204,7 +207,7 @@ func (m *ManagerImpl) CleanupPluginDirectory(dir string) error { } stat, err := os.Stat(filePath) if err != nil { - klog.ErrorS(err, "Failed to stat file", "path", filePath) + logger.Error(err, "Failed to stat file", "path", filePath) continue } if stat.IsDir() || stat.Mode()&os.ModeSocket == 0 { @@ -213,7 +216,7 @@ func (m *ManagerImpl) CleanupPluginDirectory(dir string) error { err = os.RemoveAll(filePath) if err != nil { errs = append(errs, err) - klog.ErrorS(err, "Failed to remove file", "path", filePath) + logger.Error(err, "Failed to remove file", "path", filePath) continue } } @@ -222,8 +225,9 @@ func (m *ManagerImpl) CleanupPluginDirectory(dir string) error { // PluginConnected is to connect a plugin to a new endpoint. // This is done as part of device plugin registration. -func (m *ManagerImpl) PluginConnected(resourceName string, p plugin.DevicePlugin) error { - options, err := p.API().GetDevicePluginOptions(context.Background(), &pluginapi.Empty{}) +func (m *ManagerImpl) PluginConnected(ctx context.Context, resourceName string, p plugin.DevicePlugin) error { + logger := klog.FromContext(ctx) + options, err := p.API().GetDevicePluginOptions(ctx, &pluginapi.Empty{}) if err != nil { return fmt.Errorf("failed to get device plugin options: %v", err) } @@ -234,19 +238,19 @@ func (m *ManagerImpl) PluginConnected(resourceName string, p plugin.DevicePlugin defer m.mutex.Unlock() m.endpoints[resourceName] = endpointInfo{e, options} - klog.V(2).InfoS("Device plugin connected", "resourceName", resourceName) + logger.V(2).Info("Device plugin connected", "resourceName", resourceName) return nil } // PluginDisconnected is to disconnect a plugin from an endpoint. // This is done as part of device plugin deregistration. -func (m *ManagerImpl) PluginDisconnected(resourceName string) { +func (m *ManagerImpl) PluginDisconnected(logger klog.Logger, resourceName string) { m.mutex.Lock() defer m.mutex.Unlock() if ep, exists := m.endpoints[resourceName]; exists { - m.markResourceUnhealthy(resourceName) - klog.V(2).InfoS("Endpoint became unhealthy", "resourceName", resourceName, "endpoint", ep) + m.markResourceUnhealthy(logger, resourceName) + logger.V(2).Info("Endpoint became unhealthy", "resourceName", resourceName, "endpoint", ep) ep.e.setStopTime(time.Now()) } @@ -256,11 +260,11 @@ func (m *ManagerImpl) PluginDisconnected(resourceName string) { // and ensures that an upto date state (e.g. number of devices and device health) // is captured. Also, registered device and device to container allocation // information is checkpointed to the disk. -func (m *ManagerImpl) PluginListAndWatchReceiver(resourceName string, resp *pluginapi.ListAndWatchResponse) { - m.genericDeviceUpdateCallback(resourceName, resp.Devices) +func (m *ManagerImpl) PluginListAndWatchReceiver(logger klog.Logger, resourceName string, resp *pluginapi.ListAndWatchResponse) { + m.genericDeviceUpdateCallback(logger, resourceName, resp.Devices) } -func (m *ManagerImpl) genericDeviceUpdateCallback(resourceName string, devices []*pluginapi.Device) { +func (m *ManagerImpl) genericDeviceUpdateCallback(logger klog.Logger, resourceName string, devices []*pluginapi.Device) { healthyCount := 0 m.mutex.Lock() m.healthyDevices[resourceName] = sets.New[string]() @@ -304,15 +308,15 @@ func (m *ManagerImpl) genericDeviceUpdateCallback(resourceName string, devices [ select { case m.update <- resourceupdates.Update{PodUIDs: podsToUpdate.UnsortedList()}: default: - klog.ErrorS(goerrors.New("device update channel is full"), "discard pods info", "podsToUpdate", podsToUpdate.UnsortedList()) + logger.Error(goerrors.New("device update channel is full"), "discard pods info", "podsToUpdate", podsToUpdate.UnsortedList()) } } } - if err := m.writeCheckpoint(); err != nil { - klog.ErrorS(err, "Writing checkpoint encountered") + if err := m.writeCheckpoint(logger); err != nil { + logger.Error(err, "Writing checkpoint encountered") } - klog.V(2).InfoS("Processed device updates for resource", "resourceName", resourceName, "totalCount", len(devices), "healthyCount", healthyCount) + logger.V(2).Info("Processed device updates for resource", "resourceName", resourceName, "totalCount", len(devices), "healthyCount", healthyCount) } // GetWatcherHandler returns the plugin handler @@ -333,8 +337,8 @@ func (m *ManagerImpl) checkpointFile() string { // Start starts the Device Plugin Manager and start initialization of // podDevices and allocatedDevices information from checkpointed state and // starts device plugin registration service. -func (m *ManagerImpl) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, initialContainers containermap.ContainerMap, initialContainerRunningSet sets.Set[string]) error { - klog.V(2).InfoS("Starting Device Plugin manager") +func (m *ManagerImpl) Start(logger klog.Logger, activePods ActivePodsFunc, sourcesReady config.SourcesReady, initialContainers containermap.ContainerMap, initialContainerRunningSet sets.Set[string]) error { + logger.V(2).Info("Starting Device Plugin manager") m.activePods = activePods m.sourcesReady = sourcesReady @@ -342,24 +346,27 @@ func (m *ManagerImpl) Start(activePods ActivePodsFunc, sourcesReady config.Sourc m.containerRunningSet = initialContainerRunningSet // Loads in allocatedDevices information from disk. - err := m.readCheckpoint() + err := m.readCheckpoint(logger) if err != nil { - klog.ErrorS(err, "Continue after failing to read checkpoint file. Device allocation info may NOT be up-to-date") + logger.Error(err, "Continue after failing to read checkpoint file. Device allocation info may NOT be up-to-date") } - return m.server.Start() + return m.server.Start(logger) } // Stop is the function that can stop the plugin server. // Can be called concurrently, more than once, and is safe to call // without a prior Start. -func (m *ManagerImpl) Stop() error { - return m.server.Stop() +func (m *ManagerImpl) Stop(logger klog.Logger) error { + return m.server.Stop(logger) } // Allocate is the call that you can use to allocate a set of devices // from the registered device plugins. func (m *ManagerImpl) Allocate(pod *v1.Pod, container *v1.Container) error { + // Use context.TODO() because we currently do not have a proper context to pass in. + // Replace this with an appropriate context when refactoring this function to accept a context parameter. + ctx := context.TODO() if _, ok := m.devicesToReuse[string(pod.UID)]; !ok { m.devicesToReuse[string(pod.UID)] = make(map[string]sets.Set[string]) } @@ -374,7 +381,8 @@ func (m *ManagerImpl) Allocate(pod *v1.Pod, container *v1.Container) error { // ever change those semantics, this logic will need to be amended. for _, initContainer := range pod.Spec.InitContainers { if container.Name == initContainer.Name { - if err := m.allocateContainerResources(pod, container, m.devicesToReuse[string(pod.UID)]); err != nil { + + if err := m.allocateContainerResources(ctx, pod, container, m.devicesToReuse[string(pod.UID)]); err != nil { return err } if !podutil.IsRestartableInitContainer(&initContainer) { @@ -388,7 +396,7 @@ func (m *ManagerImpl) Allocate(pod *v1.Pod, container *v1.Container) error { return nil } } - if err := m.allocateContainerResources(pod, container, m.devicesToReuse[string(pod.UID)]); err != nil { + if err := m.allocateContainerResources(ctx, pod, container, m.devicesToReuse[string(pod.UID)]); err != nil { return err } m.podDevices.removeContainerAllocatedResources(string(pod.UID), container.Name, m.devicesToReuse[string(pod.UID)]) @@ -408,8 +416,8 @@ func (m *ManagerImpl) UpdatePluginResources(node *schedulerframework.NodeInfo, a return nil } -func (m *ManagerImpl) markResourceUnhealthy(resourceName string) { - klog.V(2).InfoS("Mark all resources Unhealthy for resource", "resourceName", resourceName) +func (m *ManagerImpl) markResourceUnhealthy(logger klog.Logger, resourceName string) { + logger.V(2).Info("Mark all resources Unhealthy for resource", "resourceName", resourceName) healthyDevices := sets.New[string]() if _, ok := m.healthyDevices[resourceName]; ok { healthyDevices = m.healthyDevices[resourceName] @@ -434,6 +442,9 @@ func (m *ManagerImpl) markResourceUnhealthy(resourceName string) { // capacity for already allocated pods so that they can continue to run. However, new pods // requiring device plugin resources will not be scheduled till device plugin re-registers. func (m *ManagerImpl) GetCapacity() (v1.ResourceList, v1.ResourceList, []string) { + // Use logger.TODO() because we currently do not have a proper logger to pass in. + // Replace this with an appropriate logger when refactoring this function to accept a logger parameter. + logger := klog.TODO() needsUpdateCheckpoint := false var capacity = v1.ResourceList{} var allocatable = v1.ResourceList{} @@ -446,7 +457,7 @@ func (m *ManagerImpl) GetCapacity() (v1.ResourceList, v1.ResourceList, []string) // should always be consistent. Otherwise, we run with the risk // of failing to garbage collect non-existing resources or devices. if !ok { - klog.InfoS("Unexpected: healthyDevices and endpoints are out of sync") + logger.Info("Unexpected: healthyDevices and endpoints are out of sync") } delete(m.endpoints, resourceName) delete(m.healthyDevices, resourceName) @@ -461,7 +472,7 @@ func (m *ManagerImpl) GetCapacity() (v1.ResourceList, v1.ResourceList, []string) eI, ok := m.endpoints[resourceName] if (ok && eI.e.stopGracePeriodExpired()) || !ok { if !ok { - klog.InfoS("Unexpected: unhealthyDevices and endpoints became out of sync") + logger.Info("Unexpected: unhealthyDevices and endpoints became out of sync") } delete(m.endpoints, resourceName) delete(m.unhealthyDevices, resourceName) @@ -476,41 +487,41 @@ func (m *ManagerImpl) GetCapacity() (v1.ResourceList, v1.ResourceList, []string) } m.mutex.Unlock() if needsUpdateCheckpoint { - if err := m.writeCheckpoint(); err != nil { - klog.ErrorS(err, "Failed to write checkpoint file") + if err := m.writeCheckpoint(logger); err != nil { + logger.Error(err, "Failed to write checkpoint file") } } return capacity, allocatable, deletedResources.UnsortedList() } // Checkpoints device to container allocation information to disk. -func (m *ManagerImpl) writeCheckpoint() error { +func (m *ManagerImpl) writeCheckpoint(logger klog.Logger) error { m.mutex.Lock() registeredDevs := make(map[string][]string) for resource, devices := range m.healthyDevices { registeredDevs[resource] = devices.UnsortedList() } - data := checkpoint.New(m.podDevices.toCheckpointData(), + data := checkpoint.New(m.podDevices.toCheckpointData(logger), registeredDevs) m.mutex.Unlock() err := m.checkpointManager.CreateCheckpoint(kubeletDeviceManagerCheckpoint, data) if err != nil { err2 := fmt.Errorf("failed to write checkpoint file %q: %v", kubeletDeviceManagerCheckpoint, err) - klog.ErrorS(err, "Failed to write checkpoint file") + logger.Error(err, "Failed to write checkpoint file") return err2 } - klog.V(4).InfoS("Checkpoint file written", "checkpoint", kubeletDeviceManagerCheckpoint) + logger.V(4).Info("Checkpoint file written", "checkpoint", kubeletDeviceManagerCheckpoint) return nil } // Reads device to container allocation information from disk, and populates // m.allocatedDevices accordingly. -func (m *ManagerImpl) readCheckpoint() error { +func (m *ManagerImpl) readCheckpoint(logger klog.Logger) error { cp, err := m.getCheckpoint() if err != nil { if err == errors.ErrCheckpointNotFound { // no point in trying anything else - klog.ErrorS(err, "Failed to read data from checkpoint", "checkpoint", kubeletDeviceManagerCheckpoint) + logger.Error(err, "Failed to read data from checkpoint", "checkpoint", kubeletDeviceManagerCheckpoint) return nil } return err @@ -519,7 +530,7 @@ func (m *ManagerImpl) readCheckpoint() error { m.mutex.Lock() defer m.mutex.Unlock() podDevices, registeredDevs := cp.GetData() - m.podDevices.fromCheckpointData(podDevices) + m.podDevices.fromCheckpointData(logger, podDevices) m.allocatedDevices = m.podDevices.devices() for resource := range registeredDevs { // During start up, creates empty healthyDevices list so that the resource capacity @@ -529,7 +540,7 @@ func (m *ManagerImpl) readCheckpoint() error { m.endpoints[resource] = endpointInfo{e: newStoppedEndpointImpl(resource), opts: nil} } - klog.V(4).InfoS("Read data from checkpoint file", "checkpoint", kubeletDeviceManagerCheckpoint) + logger.V(4).Info("Read data from checkpoint file", "checkpoint", kubeletDeviceManagerCheckpoint) return nil } @@ -543,6 +554,9 @@ func (m *ManagerImpl) getCheckpoint() (checkpoint.DeviceManagerCheckpoint, error // UpdateAllocatedDevices frees any Devices that are bound to terminated pods. func (m *ManagerImpl) UpdateAllocatedDevices() { + // Use klog.TODO() because we currently do not have a proper logger to pass in. + // Replace this with an appropriate context when refactoring this function to accept a logger parameter. + logger := klog.TODO() activePods := m.activePods() if !m.sourcesReady.AllReady() { return @@ -556,7 +570,7 @@ func (m *ManagerImpl) UpdateAllocatedDevices() { if len(podsToBeRemoved) <= 0 { return } - klog.V(3).InfoS("Pods to be removed", "podUIDs", sets.List(podsToBeRemoved)) + logger.V(3).Info("Pods to be removed", "podUIDs", sets.List(podsToBeRemoved)) m.podDevices.delete(sets.List(podsToBeRemoved)) // Regenerated allocatedDevices after we update pod allocation information. m.allocatedDevices = m.podDevices.devices() @@ -564,7 +578,8 @@ func (m *ManagerImpl) UpdateAllocatedDevices() { // Returns list of device Ids we need to allocate with Allocate rpc call. // Returns empty list in case we don't need to issue the Allocate rpc call. -func (m *ManagerImpl) devicesToAllocate(podUID, contName, resource string, required int, reusableDevices sets.Set[string]) (sets.Set[string], error) { +func (m *ManagerImpl) devicesToAllocate(ctx context.Context, podUID, contName, resource string, required int, reusableDevices sets.Set[string]) (sets.Set[string], error) { + logger := klog.FromContext(ctx) m.mutex.Lock() defer m.mutex.Unlock() needed := required @@ -572,7 +587,7 @@ func (m *ManagerImpl) devicesToAllocate(podUID, contName, resource string, requi // This can happen if a container restarts for example. devices := m.podDevices.containerDevices(podUID, contName, resource) if devices != nil { - klog.V(3).InfoS("Found pre-allocated devices for resource on pod", "resourceName", resource, "containerName", contName, "podUID", podUID, "devices", sets.List(devices)) + logger.V(3).Info("Found pre-allocated devices for resource on pod", "resourceName", resource, "containerName", contName, "podUID", podUID, "devices", sets.List(devices)) needed = needed - devices.Len() // A pod's resource is not expected to change once admitted by the API server, // so just fail loudly here. We can revisit this part if this no longer holds. @@ -591,13 +606,13 @@ func (m *ManagerImpl) devicesToAllocate(podUID, contName, resource string, requi // Is this a simple kubelet restart (scenario 2)? To distinguish, we use the information we got for runtime. If we are asked to allocate devices for containers reported // running, then it can only be a kubelet restart. On node reboot the runtime and the containers were also shut down. Then, if the container was running, it can only be // because it already has access to all the required devices, so we got nothing to do and we can bail out. - if !m.sourcesReady.AllReady() && m.isContainerAlreadyRunning(podUID, contName) { - klog.V(3).InfoS("Container detected running, nothing to do", "deviceNumber", needed, "resourceName", resource, "podUID", podUID, "containerName", contName) + if !m.sourcesReady.AllReady() && m.isContainerAlreadyRunning(logger, podUID, contName) { + logger.V(3).Info("Container detected running, nothing to do", "deviceNumber", needed, "resourceName", resource, "podUID", podUID, "containerName", contName) return nil, nil } // We dealt with scenario 2. If we got this far it's either scenario 3 (node reboot) or scenario 1 (steady state, normal flow). - klog.V(3).InfoS("Need devices to allocate for pod", "deviceNumber", needed, "resourceName", resource, "podUID", podUID, "containerName", contName) + logger.V(3).Info("Need devices to allocate for pod", "deviceNumber", needed, "resourceName", resource, "podUID", podUID, "containerName", contName) healthyDevices, hasRegistered := m.healthyDevices[resource] // The following checks are expected to fail only happen on scenario 3 (node reboot). @@ -623,7 +638,7 @@ func (m *ManagerImpl) devicesToAllocate(podUID, contName, resource string, requi // We handled the known error paths in scenario 3 (node reboot), so from now on we can fall back in a common path. // We cover container restart on kubelet steady state with the same flow. if needed == 0 { - klog.V(3).InfoS("No devices needed, nothing to do", "deviceNumber", needed, "resourceName", resource, "podUID", podUID, "containerName", contName) + logger.V(3).Info("No devices needed, nothing to do", "deviceNumber", needed, "resourceName", resource, "podUID", podUID, "containerName", contName) // No change, no work. return nil, nil } @@ -673,7 +688,7 @@ func (m *ManagerImpl) devicesToAllocate(podUID, contName, resource string, requi // give the plugin the chance to influence which ones to allocate from that set. if needed < aligned.Len() { // First allocate from the preferred devices list (if available). - preferred, err := m.callGetPreferredAllocationIfAvailable(podUID, contName, resource, aligned.Union(allocated), allocated, required) + preferred, err := m.callGetPreferredAllocationIfAvailable(ctx, podUID, contName, resource, aligned.Union(allocated), allocated, required) if err != nil { return nil, err } @@ -698,7 +713,7 @@ func (m *ManagerImpl) devicesToAllocate(podUID, contName, resource string, requi // Then give the plugin the chance to influence the decision on any // remaining devices to allocate. - preferred, err := m.callGetPreferredAllocationIfAvailable(podUID, contName, resource, available.Union(allocated), allocated, required) + preferred, err := m.callGetPreferredAllocationIfAvailable(ctx, podUID, contName, resource, available.Union(allocated), allocated, required) if err != nil { return nil, err } @@ -820,7 +835,8 @@ func (m *ManagerImpl) filterByAffinity(podUID, contName, resource string, availa // plugin resources for the input container, issues an Allocate rpc request // for each new device resource requirement, processes their AllocateResponses, // and updates the cached containerDevices on success. -func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Container, devicesToReuse map[string]sets.Set[string]) error { +func (m *ManagerImpl) allocateContainerResources(ctx context.Context, pod *v1.Pod, container *v1.Container, devicesToReuse map[string]sets.Set[string]) error { + logger := klog.FromContext(ctx) podUID := string(pod.UID) contName := container.Name allocatedDevicesUpdated := false @@ -832,7 +848,7 @@ func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Cont for k, v := range container.Resources.Limits { resource := string(k) needed := int(v.Value()) - klog.V(3).InfoS("Looking for needed resources", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "needed", needed) + logger.V(3).Info("Looking for needed resources", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "needed", needed) if !m.isDevicePluginResource(resource) { continue } @@ -842,7 +858,7 @@ func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Cont m.UpdateAllocatedDevices() allocatedDevicesUpdated = true } - allocDevices, err := m.devicesToAllocate(podUID, contName, resource, needed, devicesToReuse[resource]) + allocDevices, err := m.devicesToAllocate(ctx, podUID, contName, resource, needed, devicesToReuse[resource]) if err != nil { return err } @@ -878,8 +894,8 @@ func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Cont devs := allocDevices.UnsortedList() // TODO: refactor this part of code to just append a ContainerAllocationRequest // in a passed in AllocateRequest pointer, and issues a single Allocate call per pod. - klog.V(4).InfoS("Making allocation request for device plugin", "devices", devs, "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name) - resp, err := eI.e.allocate(devs) + logger.V(4).Info("Making allocation request for device plugin", "devices", devs, "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name) + resp, err := eI.e.allocate(ctx, devs) metrics.DevicePluginAllocationDuration.WithLabelValues(resource).Observe(metrics.SinceInSeconds(startRPCTime)) if err != nil { // In case of allocation failure, we want to restore m.allocatedDevices @@ -912,7 +928,7 @@ func (m *ManagerImpl) allocateContainerResources(pod *v1.Pod, container *v1.Cont } if needsUpdateCheckpoint { - return m.writeCheckpoint() + return m.writeCheckpoint(logger) } return nil @@ -933,7 +949,8 @@ func (m *ManagerImpl) checkPodActive(pod *v1.Pod) bool { // GetDeviceRunContainerOptions checks whether we have cached containerDevices // for the passed-in and returns its DeviceRunContainerOptions // for the found one. An empty struct is returned in case no cached state is found. -func (m *ManagerImpl) GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Container) (*DeviceRunContainerOptions, error) { +func (m *ManagerImpl) GetDeviceRunContainerOptions(ctx context.Context, pod *v1.Pod, container *v1.Container) (*DeviceRunContainerOptions, error) { + logger := klog.FromContext(ctx) podUID := string(pod.UID) contName := container.Name needsReAllocate := false @@ -942,13 +959,13 @@ func (m *ManagerImpl) GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Co if !m.isDevicePluginResource(resource) || v.Value() == 0 { continue } - err := m.callPreStartContainerIfNeeded(podUID, contName, resource) + err := m.callPreStartContainerIfNeeded(ctx, podUID, contName, resource) if err != nil { return nil, err } if !m.checkPodActive(pod) { - klog.V(5).InfoS("Pod deleted from activePods, skip to reAllocate", "pod", klog.KObj(pod), "podUID", podUID, "containerName", container.Name) + logger.V(5).Info("Pod deleted from activePods, skip to reAllocate", "pod", klog.KObj(pod), "podUID", podUID, "containerName", container.Name) continue } @@ -960,17 +977,18 @@ func (m *ManagerImpl) GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Co } } if needsReAllocate { - klog.V(2).InfoS("Needs to re-allocate device plugin resources for pod", "pod", klog.KObj(pod), "containerName", container.Name) + logger.V(2).Info("Needs to re-allocate device plugin resources for pod", "pod", klog.KObj(pod), "containerName", container.Name) if err := m.Allocate(pod, container); err != nil { return nil, err } } - return m.podDevices.deviceRunContainerOptions(string(pod.UID), container.Name), nil + return m.podDevices.deviceRunContainerOptions(logger, string(pod.UID), container.Name), nil } // callPreStartContainerIfNeeded issues PreStartContainer grpc call for device plugin resource // with PreStartRequired option set. -func (m *ManagerImpl) callPreStartContainerIfNeeded(podUID, contName, resource string) error { +func (m *ManagerImpl) callPreStartContainerIfNeeded(ctx context.Context, podUID, contName, resource string) error { + logger := klog.FromContext(ctx) m.mutex.Lock() eI, ok := m.endpoints[resource] if !ok { @@ -980,7 +998,7 @@ func (m *ManagerImpl) callPreStartContainerIfNeeded(podUID, contName, resource s if eI.opts == nil || !eI.opts.PreStartRequired { m.mutex.Unlock() - klog.V(5).InfoS("Plugin options indicate to skip PreStartContainer for resource", "podUID", podUID, "resourceName", resource, "containerName", contName) + logger.V(5).Info("Plugin options indicate to skip PreStartContainer for resource", "podUID", podUID, "resourceName", resource, "containerName", contName) return nil } @@ -992,8 +1010,8 @@ func (m *ManagerImpl) callPreStartContainerIfNeeded(podUID, contName, resource s m.mutex.Unlock() devs := devices.UnsortedList() - klog.V(4).InfoS("Issuing a PreStartContainer call for container", "containerName", contName, "podUID", podUID) - _, err := eI.e.preStartContainer(devs) + logger.V(4).Info("Issuing a PreStartContainer call for container", "containerName", contName, "podUID", podUID) + _, err := eI.e.preStartContainer(ctx, devs) if err != nil { return fmt.Errorf("device plugin PreStartContainer rpc failed with err: %v", err) } @@ -1003,20 +1021,21 @@ func (m *ManagerImpl) callPreStartContainerIfNeeded(podUID, contName, resource s // callGetPreferredAllocationIfAvailable issues GetPreferredAllocation grpc // call for device plugin resource with GetPreferredAllocationAvailable option set. -func (m *ManagerImpl) callGetPreferredAllocationIfAvailable(podUID, contName, resource string, available, mustInclude sets.Set[string], size int) (sets.Set[string], error) { +func (m *ManagerImpl) callGetPreferredAllocationIfAvailable(ctx context.Context, podUID, contName, resource string, available, mustInclude sets.Set[string], size int) (sets.Set[string], error) { + logger := klog.FromContext(ctx) eI, ok := m.endpoints[resource] if !ok { return nil, fmt.Errorf("endpoint not found in cache for a registered resource: %s", resource) } if eI.opts == nil || !eI.opts.GetPreferredAllocationAvailable { - klog.V(5).InfoS("Plugin options indicate to skip GetPreferredAllocation for resource", "resourceName", resource, "podUID", podUID, "containerName", contName) + logger.V(5).Info("Plugin options indicate to skip GetPreferredAllocation for resource", "resourceName", resource, "podUID", podUID, "containerName", contName) return nil, nil } m.mutex.Unlock() - klog.V(4).InfoS("Issuing a GetPreferredAllocation call for container", "resourceName", resource, "containerName", contName, "podUID", podUID) - resp, err := eI.e.getPreferredAllocation(available.UnsortedList(), mustInclude.UnsortedList(), size) + logger.V(4).Info("Issuing a GetPreferredAllocation call for container", "resourceName", resource, "containerName", contName, "podUID", podUID) + resp, err := eI.e.getPreferredAllocation(ctx, available.UnsortedList(), mustInclude.UnsortedList(), size) m.mutex.Lock() if err != nil { return nil, fmt.Errorf("device plugin GetPreferredAllocation rpc failed with err: %v", err) @@ -1073,10 +1092,13 @@ func (m *ManagerImpl) isDevicePluginResource(resource string) bool { // GetAllocatableDevices returns information about all the healthy devices known to the manager func (m *ManagerImpl) GetAllocatableDevices() ResourceDeviceInstances { + // Use klog.TODO() because we currently do not have a proper logger to pass in. + // Replace this with an appropriate context when refactoring this function to accept a logger parameter. + logger := klog.TODO() m.mutex.Lock() defer m.mutex.Unlock() resp := m.allDevices.Filter(m.healthyDevices) - klog.V(4).InfoS("GetAllocatableDevices", "known", len(m.allDevices), "allocatable", len(resp)) + logger.V(4).Info("GetAllocatableDevices", "known", len(m.allDevices), "allocatable", len(resp)) return resp } @@ -1160,10 +1182,10 @@ func (m *ManagerImpl) ShouldResetExtendedResourceCapacity() bool { return len(checkpoints) == 0 } -func (m *ManagerImpl) isContainerAlreadyRunning(podUID, cntName string) bool { +func (m *ManagerImpl) isContainerAlreadyRunning(logger klog.Logger, podUID, cntName string) bool { cntID, err := m.containerMap.GetContainerID(podUID, cntName) if err != nil { - klog.ErrorS(err, "Container not found in the initial map, assumed NOT running", "podUID", podUID, "containerName", cntName) + logger.Error(err, "Container not found in the initial map, assumed NOT running", "podUID", podUID, "containerName", cntName) return false } @@ -1171,11 +1193,11 @@ func (m *ManagerImpl) isContainerAlreadyRunning(podUID, cntName string) bool { // so on kubelet restart containers will again fail admission, hitting https://github.com/kubernetes/kubernetes/issues/118559 again. // This scenario should however be rare enough. if !m.containerRunningSet.Has(cntID) { - klog.V(4).InfoS("Container not present in the initial running set", "podUID", podUID, "containerName", cntName, "containerID", cntID) + logger.V(4).Info("Container not present in the initial running set", "podUID", podUID, "containerName", cntName, "containerID", cntID) return false } // Once we make it here we know we have a running container. - klog.V(4).InfoS("Container found in the initial set, assumed running", "podUID", podUID, "containerName", cntName, "containerID", cntID) + logger.V(4).Info("Container found in the initial set, assumed running", "podUID", podUID, "containerName", cntName, "containerID", cntID) return true } diff --git a/pkg/kubelet/cm/devicemanager/manager_test.go b/pkg/kubelet/cm/devicemanager/manager_test.go index e6e0f279dbb..13dc2d159de 100644 --- a/pkg/kubelet/cm/devicemanager/manager_test.go +++ b/pkg/kubelet/cm/devicemanager/manager_test.go @@ -42,6 +42,7 @@ import ( utilfeature "k8s.io/apiserver/pkg/util/feature" "k8s.io/client-go/tools/record" featuregatetesting "k8s.io/component-base/featuregate/testing" + "k8s.io/klog/v2" pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" watcherapi "k8s.io/kubelet/pkg/apis/pluginregistration/v1" "k8s.io/kubernetes/pkg/features" @@ -63,24 +64,24 @@ const ( testResourceName = "fake-domain/resource" ) -func newWrappedManagerImpl(socketPath string, manager *ManagerImpl) *wrappedManagerImpl { +func newWrappedManagerImpl(logger klog.Logger, socketPath string, manager *ManagerImpl) *wrappedManagerImpl { w := &wrappedManagerImpl{ ManagerImpl: manager, callback: manager.genericDeviceUpdateCallback, } w.socketdir, _ = filepath.Split(socketPath) - w.server, _ = plugin.NewServer(socketPath, w, w) + w.server, _ = plugin.NewServer(logger, socketPath, w, w) return w } type wrappedManagerImpl struct { *ManagerImpl socketdir string - callback func(string, []*pluginapi.Device) + callback func(klog.Logger, string, []*pluginapi.Device) } -func (m *wrappedManagerImpl) PluginListAndWatchReceiver(r string, resp *pluginapi.ListAndWatchResponse) { - m.callback(r, resp.Devices) +func (m *wrappedManagerImpl) PluginListAndWatchReceiver(logger klog.Logger, r string, resp *pluginapi.ListAndWatchResponse) { + m.callback(logger, r, resp.Devices) } func tmpSocketDir() (socketDir, socketName, pluginSocketName string, err error) { @@ -95,37 +96,41 @@ func tmpSocketDir() (socketDir, socketName, pluginSocketName string, err error) } func TestNewManagerImpl(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) socketDir, socketName, _, err := tmpSocketDir() topologyStore := topologymanager.NewFakeManager() require.NoError(t, err) defer os.RemoveAll(socketDir) - _, err = newManagerImpl(socketName, nil, topologyStore) + _, err = newManagerImpl(logger, socketName, nil, topologyStore) require.NoError(t, err) os.RemoveAll(socketDir) } func TestNewManagerImplStart(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) socketDir, socketName, pluginSocketName, err := tmpSocketDir() require.NoError(t, err) defer os.RemoveAll(socketDir) - m, _, p := setup(t, []*pluginapi.Device{}, func(n string, d []*pluginapi.Device) {}, socketName, pluginSocketName) - cleanup(t, m, p) + m, _, p := setup(tCtx, t, []*pluginapi.Device{}, func(_ klog.Logger, n string, d []*pluginapi.Device) {}, socketName, pluginSocketName) + cleanup(logger, m, p) // Stop should tolerate being called more than once. - cleanup(t, m, p) + cleanup(logger, m, p) } func TestNewManagerImplStartProbeMode(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) socketDir, socketName, pluginSocketName, err := tmpSocketDir() require.NoError(t, err) defer os.RemoveAll(socketDir) - m, _, p, _ := setupInProbeMode(t, []*pluginapi.Device{}, func(n string, d []*pluginapi.Device) {}, socketName, pluginSocketName) - cleanup(t, m, p) + m, _, p, _ := setupInProbeMode(tCtx, t, []*pluginapi.Device{}, func(_ klog.Logger, n string, d []*pluginapi.Device) {}, socketName, pluginSocketName) + cleanup(logger, m, p) } // Tests that the device plugin manager correctly handles registration and re-registration by // making sure that after registration, devices are correctly updated and if a re-registration // happens, we will NOT delete devices; and no orphaned devices left. func TestDevicePluginReRegistration(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) // TODO: Remove skip once https://github.com/kubernetes/kubernetes/pull/115269 merges. if goruntime.GOOS == "windows" { t.Skip("Skipping test on Windows.") @@ -142,8 +147,8 @@ func TestDevicePluginReRegistration(t *testing.T) { } for _, preStartContainerFlag := range []bool{false, true} { for _, getPreferredAllocationFlag := range []bool{false, true} { - m, ch, p1 := setup(t, devs, nil, socketName, pluginSocketName) - p1.Register(socketName, testResourceName, "") + m, ch, p1 := setup(tCtx, t, devs, nil, socketName, pluginSocketName) + p1.Register(tCtx, socketName, testResourceName, "") select { case <-ch: @@ -156,10 +161,10 @@ func TestDevicePluginReRegistration(t *testing.T) { require.Equal(t, resourceCapacity.Value(), resourceAllocatable.Value(), "capacity should equal to allocatable") require.Equal(t, int64(2), resourceAllocatable.Value(), "Devices are not updated.") - p2 := plugin.NewDevicePluginStub(devs, pluginSocketName+".new", testResourceName, preStartContainerFlag, getPreferredAllocationFlag) - err = p2.Start() + p2 := plugin.NewDevicePluginStub(logger, devs, pluginSocketName+".new", testResourceName, preStartContainerFlag, getPreferredAllocationFlag) + err = p2.Start(tCtx) require.NoError(t, err) - p2.Register(socketName, testResourceName, "") + p2.Register(tCtx, socketName, testResourceName, "") select { case <-ch: @@ -173,10 +178,10 @@ func TestDevicePluginReRegistration(t *testing.T) { require.Equal(t, int64(2), resourceAllocatable.Value(), "Devices shouldn't change.") // Test the scenario that a plugin re-registers with different devices. - p3 := plugin.NewDevicePluginStub(devsForRegistration, pluginSocketName+".third", testResourceName, preStartContainerFlag, getPreferredAllocationFlag) - err = p3.Start() + p3 := plugin.NewDevicePluginStub(logger, devsForRegistration, pluginSocketName+".third", testResourceName, preStartContainerFlag, getPreferredAllocationFlag) + err = p3.Start(tCtx) require.NoError(t, err) - p3.Register(socketName, testResourceName, "") + p3.Register(tCtx, socketName, testResourceName, "") select { case <-ch: @@ -188,9 +193,9 @@ func TestDevicePluginReRegistration(t *testing.T) { resourceAllocatable = allocatable[v1.ResourceName(testResourceName)] require.Equal(t, resourceCapacity.Value(), resourceAllocatable.Value(), "capacity should equal to allocatable") require.Equal(t, int64(1), resourceAllocatable.Value(), "Devices of plugin previously registered should be removed.") - p2.Stop() - p3.Stop() - cleanup(t, m, p1) + p2.Stop(logger) + p3.Stop(logger) + cleanup(logger, m, p1) } } } @@ -201,6 +206,7 @@ func TestDevicePluginReRegistration(t *testing.T) { // While testing above scenario, plugin discovery and registration will be done using // Kubelet probe based mechanism func TestDevicePluginReRegistrationProbeMode(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) // TODO: Remove skip once https://github.com/kubernetes/kubernetes/pull/115269 merges. if goruntime.GOOS == "windows" { t.Skip("Skipping test on Windows.") @@ -216,7 +222,7 @@ func TestDevicePluginReRegistrationProbeMode(t *testing.T) { {ID: "Dev3", Health: pluginapi.Healthy}, } - m, ch, p1, _ := setupInProbeMode(t, devs, nil, socketName, pluginSocketName) + m, ch, p1, _ := setupInProbeMode(tCtx, t, devs, nil, socketName, pluginSocketName) // Wait for the first callback to be issued. select { @@ -230,8 +236,8 @@ func TestDevicePluginReRegistrationProbeMode(t *testing.T) { require.Equal(t, resourceCapacity.Value(), resourceAllocatable.Value(), "capacity should equal to allocatable") require.Equal(t, int64(2), resourceAllocatable.Value(), "Devices are not updated.") - p2 := plugin.NewDevicePluginStub(devs, pluginSocketName+".new", testResourceName, false, false) - err = p2.Start() + p2 := plugin.NewDevicePluginStub(logger, devs, pluginSocketName+".new", testResourceName, false, false) + err = p2.Start(tCtx) require.NoError(t, err) // Wait for the second callback to be issued. select { @@ -247,8 +253,8 @@ func TestDevicePluginReRegistrationProbeMode(t *testing.T) { require.Equal(t, int64(2), resourceAllocatable.Value(), "Devices are not updated.") // Test the scenario that a plugin re-registers with different devices. - p3 := plugin.NewDevicePluginStub(devsForRegistration, pluginSocketName+".third", testResourceName, false, false) - err = p3.Start() + p3 := plugin.NewDevicePluginStub(logger, devsForRegistration, pluginSocketName+".third", testResourceName, false, false) + err = p3.Start(tCtx) require.NoError(t, err) // Wait for the third callback to be issued. select { @@ -262,26 +268,26 @@ func TestDevicePluginReRegistrationProbeMode(t *testing.T) { resourceAllocatable = allocatable[v1.ResourceName(testResourceName)] require.Equal(t, resourceCapacity.Value(), resourceAllocatable.Value(), "capacity should equal to allocatable") require.Equal(t, int64(1), resourceAllocatable.Value(), "Devices of previous registered should be removed") - p2.Stop() - p3.Stop() - cleanup(t, m, p1) + p2.Stop(logger) + p3.Stop(logger) + cleanup(logger, m, p1) } func setupDeviceManager(t *testing.T, devs []*pluginapi.Device, callback monitorCallback, socketName string, - topology []cadvisorapi.Node) (Manager, <-chan interface{}) { + topology []cadvisorapi.Node, logger klog.Logger) (Manager, <-chan interface{}) { topologyStore := topologymanager.NewFakeManager() - m, err := newManagerImpl(socketName, topology, topologyStore) + m, err := newManagerImpl(logger, socketName, topology, topologyStore) require.NoError(t, err) updateChan := make(chan interface{}) - w := newWrappedManagerImpl(socketName, m) + w := newWrappedManagerImpl(logger, socketName, m) if callback != nil { w.callback = callback } originalCallback := w.callback - w.callback = func(resourceName string, devices []*pluginapi.Device) { - originalCallback(resourceName, devices) + w.callback = func(logger klog.Logger, resourceName string, devices []*pluginapi.Device) { + originalCallback(logger, resourceName, devices) updateChan <- new(interface{}) } activePods := func() []*v1.Pod { @@ -290,15 +296,15 @@ func setupDeviceManager(t *testing.T, devs []*pluginapi.Device, callback monitor // test steady state, initialization where sourcesReady, containerMap and containerRunningSet // are relevant will be tested with a different flow - err = w.Start(activePods, &sourcesReadyStub{}, containermap.NewContainerMap(), sets.New[string]()) + err = w.Start(logger, activePods, &sourcesReadyStub{}, containermap.NewContainerMap(), sets.New[string]()) require.NoError(t, err) return w, updateChan } -func setupDevicePlugin(t *testing.T, devs []*pluginapi.Device, pluginSocketName string) *plugin.Stub { - p := plugin.NewDevicePluginStub(devs, pluginSocketName, testResourceName, false, false) - err := p.Start() +func setupDevicePlugin(ctx context.Context, t *testing.T, devs []*pluginapi.Device, pluginSocketName string) *plugin.Stub { + p := plugin.NewDevicePluginStub(klog.FromContext(ctx), devs, pluginSocketName, testResourceName, false, false) + err := p.Start(ctx) require.NoError(t, err) return p } @@ -321,30 +327,33 @@ func runPluginManager(ctx context.Context, pluginManager pluginmanager.PluginMan go pluginManager.Run(ctx, sourcesReady, wait.NeverStop) } -func setup(t *testing.T, devs []*pluginapi.Device, callback monitorCallback, socketName string, pluginSocketName string) (Manager, <-chan interface{}, *plugin.Stub) { - m, updateChan := setupDeviceManager(t, devs, callback, socketName, nil) - p := setupDevicePlugin(t, devs, pluginSocketName) +func setup(ctx context.Context, t *testing.T, devs []*pluginapi.Device, callback monitorCallback, socketName string, pluginSocketName string) (Manager, <-chan interface{}, *plugin.Stub) { + logger := klog.FromContext(ctx) + m, updateChan := setupDeviceManager(t, devs, callback, socketName, nil, logger) + p := setupDevicePlugin(ctx, t, devs, pluginSocketName) return m, updateChan, p } -func setupInProbeMode(t *testing.T, devs []*pluginapi.Device, callback monitorCallback, socketName string, pluginSocketName string) (Manager, <-chan interface{}, *plugin.Stub, pluginmanager.PluginManager) { - m, updateChan := setupDeviceManager(t, devs, callback, socketName, nil) - p := setupDevicePlugin(t, devs, pluginSocketName) +func setupInProbeMode(ctx context.Context, t *testing.T, devs []*pluginapi.Device, callback monitorCallback, socketName string, pluginSocketName string) (Manager, <-chan interface{}, *plugin.Stub, pluginmanager.PluginManager) { + logger := klog.FromContext(ctx) + m, updateChan := setupDeviceManager(t, devs, callback, socketName, nil, logger) + p := setupDevicePlugin(ctx, t, devs, pluginSocketName) pm := setupPluginManager(t, pluginSocketName, m) return m, updateChan, p, pm } -func cleanup(t *testing.T, m Manager, p *plugin.Stub) { - p.Stop() - m.Stop() +func cleanup(logger klog.Logger, m Manager, p *plugin.Stub) { + p.Stop(logger) + m.Stop(logger) } func TestUpdateCapacityAllocatable(t *testing.T) { + logger, tCtx := ktesting.NewTestContext(t) socketDir, socketName, _, err := tmpSocketDir() topologyStore := topologymanager.NewFakeManager() require.NoError(t, err) defer os.RemoveAll(socketDir) - testManager, err := newManagerImpl(socketName, nil, topologyStore) + testManager, err := newManagerImpl(logger, socketName, nil, topologyStore) as := assert.New(t) as.NotNil(testManager) as.NoError(err) @@ -361,7 +370,7 @@ func TestUpdateCapacityAllocatable(t *testing.T) { resourceName1 := "domain1.com/resource1" e1 := &endpointImpl{} testManager.endpoints[resourceName1] = endpointInfo{e: e1, opts: nil} - callback(resourceName1, devs) + callback(logger, resourceName1, devs) capacity, allocatable, removedResources := testManager.GetCapacity() resource1Capacity, ok := capacity[v1.ResourceName(resourceName1)] as.True(ok) @@ -373,7 +382,7 @@ func TestUpdateCapacityAllocatable(t *testing.T) { // Deletes an unhealthy device should NOT change allocatable but change capacity. devs1 := devs[:len(devs)-1] - callback(resourceName1, devs1) + callback(logger, resourceName1, devs1) capacity, allocatable, removedResources = testManager.GetCapacity() resource1Capacity, ok = capacity[v1.ResourceName(resourceName1)] as.True(ok) @@ -385,7 +394,7 @@ func TestUpdateCapacityAllocatable(t *testing.T) { // Updates a healthy device to unhealthy should reduce allocatable by 1. devs[1].Health = pluginapi.Unhealthy - callback(resourceName1, devs) + callback(logger, resourceName1, devs) capacity, allocatable, removedResources = testManager.GetCapacity() resource1Capacity, ok = capacity[v1.ResourceName(resourceName1)] as.True(ok) @@ -397,7 +406,7 @@ func TestUpdateCapacityAllocatable(t *testing.T) { // Deletes a healthy device should reduce capacity and allocatable by 1. devs2 := devs[1:] - callback(resourceName1, devs2) + callback(logger, resourceName1, devs2) capacity, allocatable, removedResources = testManager.GetCapacity() resource1Capacity, ok = capacity[v1.ResourceName(resourceName1)] as.True(ok) @@ -412,7 +421,7 @@ func TestUpdateCapacityAllocatable(t *testing.T) { e2 := &endpointImpl{} e2.client = plugin.NewPluginClient(resourceName2, socketName, testManager) testManager.endpoints[resourceName2] = endpointInfo{e: e2, opts: nil} - callback(resourceName2, devs) + callback(logger, resourceName2, devs) capacity, allocatable, removedResources = testManager.GetCapacity() as.Len(capacity, 2) resource2Capacity, ok := capacity[v1.ResourceName(resourceName2)] @@ -440,15 +449,15 @@ func TestUpdateCapacityAllocatable(t *testing.T) { // Stops resourceName2 endpoint. Verifies its stopTime is set, allocate and // preStartContainer calls return errors. - e2.client.Disconnect() + e2.client.Disconnect(logger) as.False(e2.stopTime.IsZero()) - _, err = e2.allocate([]string{"Device1"}) + _, err = e2.allocate(tCtx, []string{"Device1"}) reflect.DeepEqual(err, fmt.Errorf(errEndpointStopped, e2)) - _, err = e2.preStartContainer([]string{"Device1"}) + _, err = e2.preStartContainer(tCtx, []string{"Device1"}) reflect.DeepEqual(err, fmt.Errorf(errEndpointStopped, e2)) // Marks resourceName2 unhealthy and verifies its capacity/allocatable are // correctly updated. - testManager.markResourceUnhealthy(resourceName2) + testManager.markResourceUnhealthy(logger, resourceName2) capacity, allocatable, removed = testManager.GetCapacity() val, ok = capacity[v1.ResourceName(resourceName2)] as.True(ok) @@ -462,11 +471,11 @@ func TestUpdateCapacityAllocatable(t *testing.T) { // it as a DevicePlugin resource. This makes sure any pod that was scheduled // during the time of propagating capacity change to the scheduler will be // properly rejected instead of being incorrectly started. - err = testManager.writeCheckpoint() + err = testManager.writeCheckpoint(logger) as.NoError(err) testManager.healthyDevices = make(map[string]sets.Set[string]) testManager.unhealthyDevices = make(map[string]sets.Set[string]) - err = testManager.readCheckpoint() + err = testManager.readCheckpoint(logger) as.NoError(err) as.Len(testManager.endpoints, 1) as.Contains(testManager.endpoints, resourceName2) @@ -482,11 +491,12 @@ func TestUpdateCapacityAllocatable(t *testing.T) { } func TestGetAllocatableDevicesMultipleResources(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) socketDir, socketName, _, err := tmpSocketDir() topologyStore := topologymanager.NewFakeManager() require.NoError(t, err) defer os.RemoveAll(socketDir) - testManager, err := newManagerImpl(socketName, nil, topologyStore) + testManager, err := newManagerImpl(logger, socketName, nil, topologyStore) as := assert.New(t) as.NotNil(testManager) as.NoError(err) @@ -499,7 +509,7 @@ func TestGetAllocatableDevicesMultipleResources(t *testing.T) { resourceName1 := "domain1.com/resource1" e1 := &endpointImpl{} testManager.endpoints[resourceName1] = endpointInfo{e: e1, opts: nil} - testManager.genericDeviceUpdateCallback(resourceName1, resource1Devs) + testManager.genericDeviceUpdateCallback(logger, resourceName1, resource1Devs) resource2Devs := []*pluginapi.Device{ {ID: "R2Device1", Health: pluginapi.Healthy}, @@ -507,7 +517,7 @@ func TestGetAllocatableDevicesMultipleResources(t *testing.T) { resourceName2 := "other.domain2.org/resource2" e2 := &endpointImpl{} testManager.endpoints[resourceName2] = endpointInfo{e: e2, opts: nil} - testManager.genericDeviceUpdateCallback(resourceName2, resource2Devs) + testManager.genericDeviceUpdateCallback(logger, resourceName2, resource2Devs) allocatableDevs := testManager.GetAllocatableDevices() as.Len(allocatableDevs, 2) @@ -523,11 +533,12 @@ func TestGetAllocatableDevicesMultipleResources(t *testing.T) { } func TestGetAllocatableDevicesHealthTransition(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) socketDir, socketName, _, err := tmpSocketDir() topologyStore := topologymanager.NewFakeManager() require.NoError(t, err) defer os.RemoveAll(socketDir) - testManager, err := newManagerImpl(socketName, nil, topologyStore) + testManager, err := newManagerImpl(logger, socketName, nil, topologyStore) as := assert.New(t) as.NotNil(testManager) as.NoError(err) @@ -544,7 +555,7 @@ func TestGetAllocatableDevicesHealthTransition(t *testing.T) { e1 := &endpointImpl{} testManager.endpoints[resourceName1] = endpointInfo{e: e1, opts: nil} - testManager.genericDeviceUpdateCallback(resourceName1, resource1Devs) + testManager.genericDeviceUpdateCallback(logger, resourceName1, resource1Devs) allocatableDevs := testManager.GetAllocatableDevices() as.Len(allocatableDevs, 1) @@ -558,7 +569,7 @@ func TestGetAllocatableDevicesHealthTransition(t *testing.T) { {ID: "R1Device2", Health: pluginapi.Healthy}, {ID: "R1Device3", Health: pluginapi.Healthy}, } - testManager.genericDeviceUpdateCallback(resourceName1, resource1Devs) + testManager.genericDeviceUpdateCallback(logger, resourceName1, resource1Devs) allocatableDevs = testManager.GetAllocatableDevices() as.Len(allocatableDevs, 1) @@ -667,6 +678,7 @@ func (b *containerAllocateResponseBuilder) Build() *pluginapi.ContainerAllocateR } func TestCheckpoint(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) resourceName1 := "domain1.com/resource1" resourceName2 := "domain2.com/resource2" resourceName3 := "domain2.com/resource3" @@ -739,11 +751,11 @@ func TestCheckpoint(t *testing.T) { expectedAllocatedDevices := testManager.podDevices.devices() expectedAllDevices := testManager.healthyDevices - err = testManager.writeCheckpoint() + err = testManager.writeCheckpoint(logger) as.NoError(err) testManager.podDevices = newPodDevices() - err = testManager.readCheckpoint() + err = testManager.readCheckpoint(logger) as.NoError(err) as.Equal(expectedPodDevices.size(), testManager.podDevices.size()) @@ -753,8 +765,8 @@ func TestCheckpoint(t *testing.T) { expDevices := expectedPodDevices.containerDevices(podUID, conName, resource) testDevices := testManager.podDevices.containerDevices(podUID, conName, resource) as.True(reflect.DeepEqual(expDevices, testDevices)) - opts1 := expectedPodDevices.deviceRunContainerOptions(podUID, conName) - opts2 := testManager.podDevices.deviceRunContainerOptions(podUID, conName) + opts1 := expectedPodDevices.deviceRunContainerOptions(logger, podUID, conName) + opts2 := testManager.podDevices.deviceRunContainerOptions(logger, podUID, conName) as.Equal(len(opts1.Envs), len(opts2.Envs)) as.Equal(len(opts1.Mounts), len(opts2.Mounts)) as.Equal(len(opts1.Devices), len(opts2.Devices)) @@ -783,19 +795,19 @@ type MockEndpoint struct { initChan chan []string } -func (m *MockEndpoint) preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error) { +func (m *MockEndpoint) preStartContainer(_ context.Context, devs []string) (*pluginapi.PreStartContainerResponse, error) { m.initChan <- devs return &pluginapi.PreStartContainerResponse{}, nil } -func (m *MockEndpoint) getPreferredAllocation(available, mustInclude []string, size int) (*pluginapi.PreferredAllocationResponse, error) { +func (m *MockEndpoint) getPreferredAllocation(_ context.Context, available, mustInclude []string, size int) (*pluginapi.PreferredAllocationResponse, error) { if m.getPreferredAllocationFunc != nil { return m.getPreferredAllocationFunc(available, mustInclude, size) } return nil, nil } -func (m *MockEndpoint) allocate(devs []string) (*pluginapi.AllocateResponse, error) { +func (m *MockEndpoint) allocate(ctx context.Context, devs []string) (*pluginapi.AllocateResponse, error) { if m.allocateFunc != nil { return m.allocateFunc(devs) } @@ -826,7 +838,7 @@ func makePod(limits v1.ResourceList) *v1.Pod { } func getTestManager(tmpDir string, activePods ActivePodsFunc, testRes []TestResource) (*wrappedManagerImpl, error) { - monitorCallback := func(resourceName string, devices []*pluginapi.Device) {} + monitorCallback := func(logger klog.Logger, resourceName string, devices []*pluginapi.Device) {} ckm, err := checkpointmanager.NewCheckpointManager(tmpDir) if err != nil { return nil, err @@ -986,6 +998,7 @@ func TestFilterByAffinity(t *testing.T) { } func TestPodContainerDeviceAllocation(t *testing.T) { + tCtx := ktesting.Init(t) res1 := TestResource{ resourceName: "domain1.com/resource1", resourceQuantity: *resource.NewQuantity(int64(2), resource.DecimalSI), @@ -1064,7 +1077,7 @@ func TestPodContainerDeviceAllocation(t *testing.T) { t.Errorf("DevicePluginManager error (%v). expected error: %v but got: %v", testCase.description, testCase.expErr, err) } - runContainerOpts, err := testManager.GetDeviceRunContainerOptions(pod, &pod.Spec.Containers[0]) + runContainerOpts, err := testManager.GetDeviceRunContainerOptions(tCtx, pod, &pod.Spec.Containers[0]) if testCase.expErr == nil { as.NoError(err) } @@ -1082,6 +1095,7 @@ func TestPodContainerDeviceAllocation(t *testing.T) { } func TestPodContainerDeviceToAllocate(t *testing.T) { + tCtx := ktesting.Init(t) resourceName1 := "domain1.com/resource1" resourceName2 := "domain2.com/resource2" resourceName3 := "domain2.com/resource3" @@ -1175,7 +1189,7 @@ func TestPodContainerDeviceToAllocate(t *testing.T) { } for _, testCase := range testCases { - allocDevices, err := testManager.devicesToAllocate(testCase.podUID, testCase.contName, testCase.resource, testCase.required, testCase.reusableDevices) + allocDevices, err := testManager.devicesToAllocate(tCtx, testCase.podUID, testCase.contName, testCase.resource, testCase.required, testCase.reusableDevices) if !reflect.DeepEqual(err, testCase.expErr) { t.Errorf("devicePluginManager error (%v). expected error: %v but got: %v", testCase.description, testCase.expErr, err) @@ -1189,6 +1203,7 @@ func TestPodContainerDeviceToAllocate(t *testing.T) { } func TestDevicesToAllocateConflictWithUpdateAllocatedDevices(t *testing.T) { + tCtx := ktesting.Init(t) podToAllocate := "podToAllocate" containerToAllocate := "containerToAllocate" podToRemove := "podToRemove" @@ -1200,7 +1215,7 @@ func TestDevicesToAllocateConflictWithUpdateAllocatedDevices(t *testing.T) { devs := []*pluginapi.Device{ {ID: deviceID, Health: pluginapi.Healthy}, } - p, e := esetup(t, devs, socket, resourceName, func(n string, d []*pluginapi.Device) {}) + p, e := esetup(tCtx, t, devs, socket, resourceName, func(logger klog.Logger, n string, d []*pluginapi.Device) {}) waitUpdateAllocatedDevicesChan := make(chan struct{}) waitSetGetPreferredAllocChan := make(chan struct{}) @@ -1243,12 +1258,13 @@ func TestDevicesToAllocateConflictWithUpdateAllocatedDevices(t *testing.T) { waitUpdateAllocatedDevicesChan <- struct{}{} }() - set, err := testManager.devicesToAllocate(podToAllocate, containerToAllocate, resourceName, 1, sets.New[string]()) + set, err := testManager.devicesToAllocate(tCtx, podToAllocate, containerToAllocate, resourceName, 1, sets.New[string]()) assert.NoError(t, err) assert.Equal(t, set, sets.New[string](deviceID)) } func TestGetDeviceRunContainerOptions(t *testing.T) { + tCtx := ktesting.Init(t) res1 := TestResource{ resourceName: "domain1.com/resource1", resourceQuantity: *resource.NewQuantity(int64(2), resource.DecimalSI), @@ -1295,7 +1311,7 @@ func TestGetDeviceRunContainerOptions(t *testing.T) { as.NoError(err) // when pod is in activePods, GetDeviceRunContainerOptions should return - runContainerOpts, err := testManager.GetDeviceRunContainerOptions(pod1, &pod1.Spec.Containers[0]) + runContainerOpts, err := testManager.GetDeviceRunContainerOptions(tCtx, pod1, &pod1.Spec.Containers[0]) as.NoError(err) as.Len(runContainerOpts.Devices, 3) as.Len(runContainerOpts.Mounts, 2) @@ -1306,7 +1322,7 @@ func TestGetDeviceRunContainerOptions(t *testing.T) { testManager.UpdateAllocatedDevices() // when pod is removed from activePods,G etDeviceRunContainerOptions should return error - runContainerOpts, err = testManager.GetDeviceRunContainerOptions(pod1, &pod1.Spec.Containers[0]) + runContainerOpts, err = testManager.GetDeviceRunContainerOptions(tCtx, pod1, &pod1.Spec.Containers[0]) as.NoError(err) as.Nil(runContainerOpts) } @@ -1550,7 +1566,7 @@ func TestUpdatePluginResources(t *testing.T) { devID2 := "dev2" as := assert.New(t) - monitorCallback := func(resourceName string, devices []*pluginapi.Device) {} + monitorCallback := func(logger klog.Logger, resourceName string, devices []*pluginapi.Device) {} tmpDir, err := os.MkdirTemp("", "checkpoint") as.NoError(err) defer os.RemoveAll(tmpDir) @@ -1596,6 +1612,7 @@ func TestUpdatePluginResources(t *testing.T) { } func TestDevicePreStartContainer(t *testing.T) { + tCtx := ktesting.Init(t) // Ensures that if device manager is indicated to invoke `PreStartContainer` RPC // by device plugin, then device manager invokes PreStartContainer at endpoint interface. // Also verifies that final allocation of mounts, envs etc is same as expected. @@ -1631,7 +1648,7 @@ func TestDevicePreStartContainer(t *testing.T) { podsStub.updateActivePods(activePods) err = testManager.Allocate(pod, &pod.Spec.Containers[0]) as.NoError(err) - runContainerOpts, err := testManager.GetDeviceRunContainerOptions(pod, &pod.Spec.Containers[0]) + runContainerOpts, err := testManager.GetDeviceRunContainerOptions(tCtx, pod, &pod.Spec.Containers[0]) as.NoError(err) var initializedDevs []string select { @@ -1659,7 +1676,7 @@ func TestDevicePreStartContainer(t *testing.T) { podsStub.updateActivePods(activePods) err = testManager.Allocate(pod2, &pod2.Spec.Containers[0]) as.NoError(err) - _, err = testManager.GetDeviceRunContainerOptions(pod2, &pod2.Spec.Containers[0]) + _, err = testManager.GetDeviceRunContainerOptions(tCtx, pod2, &pod2.Spec.Containers[0]) as.NoError(err) select { case <-time.After(time.Millisecond): @@ -1670,6 +1687,7 @@ func TestDevicePreStartContainer(t *testing.T) { } func TestResetExtendedResource(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) as := assert.New(t) tmpDir, err := os.MkdirTemp("", "checkpoint") as.NoError(err) @@ -1697,7 +1715,7 @@ func TestResetExtendedResource(t *testing.T) { testManager.healthyDevices[extendedResourceName] = sets.New[string]() testManager.healthyDevices[extendedResourceName].Insert("dev1") // checkpoint is present, indicating node hasn't been recreated - err = testManager.writeCheckpoint() + err = testManager.writeCheckpoint(logger) require.NoError(t, err) as.False(testManager.ShouldResetExtendedResourceCapacity()) @@ -1776,6 +1794,7 @@ func makeDevice(devOnNUMA checkpoint.DevicesPerNUMA, topology bool) map[string]* } func TestGetTopologyHintsWithUpdates(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) socketDir, socketName, _, err := tmpSocketDir() defer os.RemoveAll(socketDir) require.NoError(t, err) @@ -1823,8 +1842,8 @@ func TestGetTopologyHintsWithUpdates(t *testing.T) { for _, test := range testCases { t.Run(test.description, func(t *testing.T) { - m, _ := setupDeviceManager(t, nil, nil, socketName, topology) - defer m.Stop() + m, _ := setupDeviceManager(t, nil, nil, socketName, topology, logger) + defer m.Stop(logger) mimpl := m.(*wrappedManagerImpl) wg := sync.WaitGroup{} @@ -1836,7 +1855,7 @@ func TestGetTopologyHintsWithUpdates(t *testing.T) { defer wg.Done() for i := 0; i < test.count; i++ { // simulate the device plugin to send device updates - mimpl.genericDeviceUpdateCallback(testResourceName, devs) + mimpl.genericDeviceUpdateCallback(logger, testResourceName, devs) } updated.Store(true) }() @@ -1855,6 +1874,7 @@ func TestGetTopologyHintsWithUpdates(t *testing.T) { } } func TestUpdateAllocatedResourcesStatus(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) podUID := "test-pod-uid" containerName := "test-container" resourceName := "test-resource" @@ -1893,7 +1913,7 @@ func TestUpdateAllocatedResourcesStatus(t *testing.T) { ), ) - testManager.genericDeviceUpdateCallback(resourceName, []*pluginapi.Device{ + testManager.genericDeviceUpdateCallback(logger, resourceName, []*pluginapi.Device{ {ID: "dev1", Health: pluginapi.Healthy}, {ID: "dev2", Health: pluginapi.Unhealthy}, }) @@ -1958,6 +1978,7 @@ func sortContainerStatuses(statuses []v1.ContainerStatus) { } func TestFeatureGateResourceHealthStatus(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) tmpDir, err := os.MkdirTemp("", "checkpoint") require.NoError(t, err, "err should be nil") defer func() { @@ -2007,7 +2028,7 @@ func TestFeatureGateResourceHealthStatus(t *testing.T) { featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.ResourceHealthStatus, true) for i := 0; i < deviceUpdateNumber; i++ { - testManager.genericDeviceUpdateCallback(resourceName, []*pluginapi.Device{ + testManager.genericDeviceUpdateCallback(logger, resourceName, []*pluginapi.Device{ {ID: "dev1", Health: pluginapi.Healthy}, }) } @@ -2016,7 +2037,7 @@ func TestFeatureGateResourceHealthStatus(t *testing.T) { // update device status, assume all device unhealthy. for i := 0; i < deviceUpdateNumber; i++ { - testManager.genericDeviceUpdateCallback(resourceName, []*pluginapi.Device{ + testManager.genericDeviceUpdateCallback(logger, resourceName, []*pluginapi.Device{ {ID: fmt.Sprintf("dev%d", i), Health: pluginapi.Unhealthy}, }) } diff --git a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/api.go b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/api.go index f84cd551ffd..b400a212564 100644 --- a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/api.go +++ b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/api.go @@ -17,20 +17,23 @@ limitations under the License. package v1beta1 import ( + "context" + + "k8s.io/klog/v2" api "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" ) // RegistrationHandler is an interface for handling device plugin registration // and plugin directory cleanup. type RegistrationHandler interface { - CleanupPluginDirectory(string) error + CleanupPluginDirectory(klog.Logger, string) error } // ClientHandler is an interface for handling device plugin connections. type ClientHandler interface { - PluginConnected(string, DevicePlugin) error - PluginDisconnected(string) - PluginListAndWatchReceiver(string, *api.ListAndWatchResponse) + PluginConnected(context.Context, string, DevicePlugin) error + PluginDisconnected(klog.Logger, string) + PluginListAndWatchReceiver(klog.Logger, string, *api.ListAndWatchResponse) } // TODO: evaluate whether we need these error definitions. diff --git a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/client.go b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/client.go index 98cc3c2813a..751ebc17ebb 100644 --- a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/client.go +++ b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/client.go @@ -39,9 +39,9 @@ type DevicePlugin interface { // Client interface provides methods for establishing/closing gRPC connection and running the device plugin gRPC client. type Client interface { - Connect() error - Run() - Disconnect() error + Connect(context.Context) error + Run(context.Context) + Disconnect(klog.Logger) error } type client struct { @@ -63,51 +63,55 @@ func NewPluginClient(r string, socketPath string, h ClientHandler) Client { } // Connect is for establishing a gRPC connection between device manager and device plugin. -func (c *client) Connect() error { - client, conn, err := dial(c.socket) +func (c *client) Connect(ctx context.Context) error { + logger := klog.FromContext(ctx) + client, conn, err := dial(ctx, c.socket) if err != nil { - klog.ErrorS(err, "Unable to connect to device plugin client with socket path", "path", c.socket) + logger.Error(err, "Unable to connect to device plugin client with socket path", "path", c.socket) return err } c.mutex.Lock() c.grpc = conn c.client = client c.mutex.Unlock() - return c.handler.PluginConnected(c.resource, c) + return c.handler.PluginConnected(ctx, c.resource, c) } // Run is for running the device plugin gRPC client. -func (c *client) Run() { - stream, err := c.client.ListAndWatch(context.Background(), &api.Empty{}) +func (c *client) Run(ctx context.Context) { + logger := klog.FromContext(ctx) + // FIXME: passing real context to ListAndWatch results in + // failing TestDevicePluginReRegistration with "context cancelled" error + stream, err := c.client.ListAndWatch(context.TODO(), &api.Empty{}) if err != nil { - klog.ErrorS(err, "ListAndWatch ended unexpectedly for device plugin", "resource", c.resource) + logger.Error(err, "ListAndWatch ended unexpectedly for device plugin", "resource", c.resource) return } for { response, err := stream.Recv() if err != nil { - klog.ErrorS(err, "ListAndWatch ended unexpectedly for device plugin", "resource", c.resource) + logger.Error(err, "ListAndWatch ended unexpectedly for device plugin", "resource", c.resource) return } - klog.V(2).InfoS("State pushed for device plugin", "resource", c.resource, "resourceCapacity", len(response.Devices)) - c.handler.PluginListAndWatchReceiver(c.resource, response) + logger.V(2).Info("State pushed for device plugin", "resource", c.resource, "resourceCapacity", len(response.Devices)) + c.handler.PluginListAndWatchReceiver(logger, c.resource, response) } } // Disconnect is for closing gRPC connection between device manager and device plugin. -func (c *client) Disconnect() error { +func (c *client) Disconnect(logger klog.Logger) error { c.mutex.Lock() if c.grpc != nil { if err := c.grpc.Close(); err != nil { - klog.V(2).ErrorS(err, "Failed to close grpc connection", "resource", c.Resource()) + logger.V(2).Error(err, "Failed to close grpc connection", "resource", c.Resource()) } c.grpc = nil } c.mutex.Unlock() - c.handler.PluginDisconnected(c.resource) + c.handler.PluginDisconnected(logger, c.resource) - klog.V(2).InfoS("Device plugin disconnected", "resource", c.resource) + logger.V(2).Info("Device plugin disconnected", "resource", c.resource) return nil } @@ -124,8 +128,8 @@ func (c *client) SocketPath() string { } // dial establishes the gRPC communication with the registered device plugin. https://godoc.org/google.golang.org/grpc#Dial -func dial(unixSocketPath string) (api.DevicePluginClient, *grpc.ClientConn, error) { - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) +func dial(ctx context.Context, unixSocketPath string) (api.DevicePluginClient, *grpc.ClientConn, error) { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() c, err := grpc.DialContext(ctx, unixSocketPath, diff --git a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/handler.go b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/handler.go index 204afd3d1b1..764f520348a 100644 --- a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/handler.go +++ b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/handler.go @@ -17,6 +17,7 @@ limitations under the License. package v1beta1 import ( + "context" "fmt" "os" "time" @@ -29,30 +30,41 @@ import ( ) func (s *server) GetPluginHandler() cache.PluginHandler { + // Use context.TODO() because we currently do not have a proper context to pass in. + // Replace this with an appropriate context when refactoring this function to accept a context parameter. + logger := klog.FromContext(context.TODO()) if f, err := os.Create(s.socketDir + "DEPRECATION"); err != nil { - klog.ErrorS(err, "Failed to create deprecation file at socket dir", "path", s.socketDir) + logger.Error(err, "Failed to create deprecation file at socket dir", "path", s.socketDir) } else { f.Close() - klog.V(4).InfoS("Created deprecation file", "path", f.Name()) + logger.V(4).Info("Created deprecation file", "path", f.Name()) } return s } func (s *server) RegisterPlugin(pluginName string, endpoint string, versions []string, pluginClientTimeout *time.Duration) error { - klog.V(2).InfoS("Registering plugin at endpoint", "plugin", pluginName, "endpoint", endpoint) - return s.connectClient(pluginName, endpoint) + // Use context.TODO() because we currently do not have a proper context to pass in. + // Replace this with an appropriate context when refactoring this function to accept a context parameter. + ctx := context.TODO() + logger := klog.FromContext(ctx) + logger.V(2).Info("Registering plugin at endpoint", "plugin", pluginName, "endpoint", endpoint) + return s.connectClient(ctx, pluginName, endpoint) } func (s *server) DeRegisterPlugin(pluginName, endpoint string) { - klog.V(2).InfoS("Deregistering plugin", "plugin", pluginName, "endpoint", endpoint) + logger := klog.FromContext(context.TODO()) + logger.V(2).Info("Deregistering plugin", "plugin", pluginName, "endpoint", endpoint) client := s.getClient(pluginName) if client != nil { - s.disconnectClient(pluginName, client) + s.disconnectClient(logger, pluginName, client) } } func (s *server) ValidatePlugin(pluginName string, endpoint string, versions []string) error { - klog.V(2).InfoS("Got plugin at endpoint with versions", "plugin", pluginName, "endpoint", endpoint, "versions", versions) + // Use context.TODO() because we currently do not have a proper context to pass in. + // Replace this with an appropriate context when refactoring this function to accept a context parameter. + logger := klog.FromContext(context.TODO()) + logger.V(2).Info("Got plugin at endpoint with versions", "plugin", pluginName, "endpoint", endpoint, "versions", versions) if !s.isVersionCompatibleWithPlugin(versions...) { return fmt.Errorf("manager version, %s, is not among plugin supported versions %v", api.Version, versions) @@ -62,58 +74,60 @@ func (s *server) ValidatePlugin(pluginName string, endpoint string, versions []s return fmt.Errorf("invalid name of device plugin socket: %s", fmt.Sprintf(errInvalidResourceName, pluginName)) } - klog.V(2).InfoS("Device plugin validated", "plugin", pluginName, "endpoint", endpoint, "versions", versions) + logger.V(2).Info("Device plugin validated", "plugin", pluginName, "endpoint", endpoint, "versions", versions) return nil } -func (s *server) connectClient(name string, socketPath string) error { +func (s *server) connectClient(ctx context.Context, name string, socketPath string) error { + logger := klog.FromContext(ctx) c := NewPluginClient(name, socketPath, s.chandler) - s.registerClient(name, c) - if err := c.Connect(); err != nil { - s.deregisterClient(name) - klog.ErrorS(err, "Failed to connect to new client", "resource", name) + s.registerClient(logger, name, c) + if err := c.Connect(ctx); err != nil { + s.deregisterClient(logger, name) + logger.Error(err, "Failed to connect to new client", "resource", name) return err } - klog.V(2).InfoS("Connected to new client", "resource", name) + logger.V(2).Info("Connected to new client", "resource", name) go func() { - s.runClient(name, c) + s.runClient(ctx, name, c) }() return nil } -func (s *server) disconnectClient(name string, c Client) error { - s.deregisterClient(name) - return c.Disconnect() +func (s *server) disconnectClient(logger klog.Logger, name string, c Client) error { + s.deregisterClient(logger, name) + return c.Disconnect(logger) } -func (s *server) registerClient(name string, c Client) { +func (s *server) registerClient(logger klog.Logger, name string, c Client) { s.mutex.Lock() defer s.mutex.Unlock() s.clients[name] = c - klog.V(2).InfoS("Registered client", "name", name) + logger.V(2).Info("Registered client", "name", name) } -func (s *server) deregisterClient(name string) { +func (s *server) deregisterClient(logger klog.Logger, name string) { s.mutex.Lock() defer s.mutex.Unlock() delete(s.clients, name) - klog.V(2).InfoS("Deregistered client", "name", name) + logger.V(2).Info("Deregistered client", "name", name) } -func (s *server) runClient(name string, c Client) { - c.Run() +func (s *server) runClient(ctx context.Context, name string, c Client) { + logger := klog.FromContext(ctx) + c.Run(ctx) c = s.getClient(name) if c == nil { return } - if err := s.disconnectClient(name, c); err != nil { - klog.ErrorS(err, "Unable to disconnect client", "resource", name, "client", c) + if err := s.disconnectClient(logger, name, c); err != nil { + logger.Error(err, "Unable to disconnect client", "resource", name, "client", c) } } diff --git a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/server.go b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/server.go index 66c6d3777e1..2cf6c5e1b83 100644 --- a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/server.go +++ b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/server.go @@ -42,8 +42,8 @@ import ( type Server interface { cache.PluginHandler healthz.HealthChecker - Start() error - Stop() error + Start(klog.Logger) error + Stop(klog.Logger) error SocketPath() string } @@ -64,14 +64,14 @@ type server struct { } // NewServer returns an initialized device plugin registration server. -func NewServer(socketPath string, rh RegistrationHandler, ch ClientHandler) (Server, error) { +func NewServer(logger klog.Logger, socketPath string, rh RegistrationHandler, ch ClientHandler) (Server, error) { if socketPath == "" || !filepath.IsAbs(socketPath) { return nil, fmt.Errorf(errBadSocket+" %s", socketPath) } dir, name := filepath.Split(socketPath) - klog.V(2).InfoS("Creating device plugin registration server", "version", api.Version, "socket", socketPath) + logger.V(2).Info("Creating device plugin registration server", "version", api.Version, "socket", socketPath) s := &server{ socketName: name, socketDir: dir, @@ -83,31 +83,31 @@ func NewServer(socketPath string, rh RegistrationHandler, ch ClientHandler) (Ser return s, nil } -func (s *server) Start() error { - klog.V(2).InfoS("Starting device plugin registration server") +func (s *server) Start(logger klog.Logger) error { + logger.V(2).Info("Starting device plugin registration server") if err := os.MkdirAll(s.socketDir, 0750); err != nil { - klog.ErrorS(err, "Failed to create the device plugin socket directory", "directory", s.socketDir) + logger.Error(err, "Failed to create the device plugin socket directory", "directory", s.socketDir) return err } if selinux.GetEnabled() { if err := selinux.SetFileLabel(s.socketDir, kubeletconfig.KubeletPluginsDirSELinuxLabel); err != nil { - klog.ErrorS(err, "Unprivileged containerized plugins might not work. Could not set selinux context on socket dir", "path", s.socketDir) + logger.Error(err, "Unprivileged containerized plugins might not work. Could not set selinux context on socket dir", "path", s.socketDir) } } // For now, we leave cleanup of the *entire* directory up to the Handler // (even though we should in theory be able to just wipe the whole directory) // because the Handler stores its checkpoint file (amongst others) in here. - if err := s.rhandler.CleanupPluginDirectory(s.socketDir); err != nil { - klog.ErrorS(err, "Failed to cleanup the device plugin directory", "directory", s.socketDir) + if err := s.rhandler.CleanupPluginDirectory(logger, s.socketDir); err != nil { + logger.Error(err, "Failed to cleanup the device plugin directory", "directory", s.socketDir) return err } ln, err := net.Listen("unix", s.SocketPath()) if err != nil { - klog.ErrorS(err, "Failed to listen to socket while starting device plugin registry") + logger.Error(err, "Failed to listen to socket while starting device plugin registry") return err } @@ -120,17 +120,17 @@ func (s *server) Start() error { s.setHealthy() if err = s.grpc.Serve(ln); err != nil { s.setUnhealthy() - klog.ErrorS(err, "Error while serving device plugin registration grpc server") + logger.Error(err, "Error while serving device plugin registration grpc server") } }() return nil } -func (s *server) Stop() error { +func (s *server) Stop(logger klog.Logger) error { s.visitClients(func(r string, c Client) { - if err := s.disconnectClient(r, c); err != nil { - klog.ErrorS(err, "Failed to disconnect device plugin client", "resourceName", r) + if err := s.disconnectClient(logger, r, c); err != nil { + logger.Error(err, "Failed to disconnect device plugin client", "resourceName", r) } }) @@ -147,7 +147,7 @@ func (s *server) Stop() error { // During kubelet termination, we do not need the registration server, // and we consider the kubelet to be healthy even when it is down. s.setHealthy() - klog.V(2).InfoS("Stopping device plugin registration server") + logger.V(2).Info("Stopping device plugin registration server") return nil } @@ -157,23 +157,24 @@ func (s *server) SocketPath() string { } func (s *server) Register(ctx context.Context, r *api.RegisterRequest) (*api.Empty, error) { - klog.InfoS("Got registration request from device plugin with resource", "resourceName", r.ResourceName) + logger := klog.FromContext(ctx) + logger.Info("Got registration request from device plugin with resource", "resourceName", r.ResourceName) metrics.DevicePluginRegistrationCount.WithLabelValues(r.ResourceName).Inc() if !s.isVersionCompatibleWithPlugin(r.Version) { err := fmt.Errorf(errUnsupportedVersion, r.Version, api.SupportedVersions) - klog.ErrorS(err, "Bad registration request from device plugin with resource", "resourceName", r.ResourceName) + logger.Error(err, "Bad registration request from device plugin with resource", "resourceName", r.ResourceName) return &api.Empty{}, err } if !v1helper.IsExtendedResourceName(core.ResourceName(r.ResourceName)) { err := fmt.Errorf(errInvalidResourceName, r.ResourceName) - klog.ErrorS(err, "Bad registration request from device plugin") + logger.Error(err, "Bad registration request from device plugin") return &api.Empty{}, err } - if err := s.connectClient(r.ResourceName, filepath.Join(s.socketDir, r.Endpoint)); err != nil { - klog.ErrorS(err, "Error connecting to device plugin client") + if err := s.connectClient(ctx, r.ResourceName, filepath.Join(s.socketDir, r.Endpoint)); err != nil { + logger.Error(err, "Error connecting to device plugin client") return &api.Empty{}, err } diff --git a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/stub.go b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/stub.go index 1d271540a1d..ed949909329 100644 --- a/pkg/kubelet/cm/devicemanager/plugin/v1beta1/stub.go +++ b/pkg/kubelet/cm/devicemanager/plugin/v1beta1/stub.go @@ -92,11 +92,11 @@ func defaultRegisterControlFunc() bool { } // NewDevicePluginStub returns an initialized DevicePlugin Stub. -func NewDevicePluginStub(devs []*pluginapi.Device, socket string, name string, preStartContainerFlag bool, getPreferredAllocationFlag bool) *Stub { +func NewDevicePluginStub(logger klog.Logger, devs []*pluginapi.Device, socket string, name string, preStartContainerFlag bool, getPreferredAllocationFlag bool) *Stub { watcher, err := fsnotify.NewWatcher() if err != nil { - klog.ErrorS(err, "Watcher creation failed") + logger.Error(err, "Watcher creation failed") panic(err) } @@ -134,8 +134,9 @@ func (m *Stub) SetRegisterControlFunc(f stubRegisterControlFunc) { // Start starts the gRPC server of the device plugin. Can only // be called once. -func (m *Stub) Start() error { - klog.InfoS("Starting device plugin server") +func (m *Stub) Start(ctx context.Context) error { + logger := klog.FromContext(ctx) + logger.Info("Starting device plugin server") err := m.cleanup() if err != nil { return err @@ -153,21 +154,21 @@ func (m *Stub) Start() error { err = m.kubeletRestartWatcher.Add(filepath.Dir(m.socket)) if err != nil { - klog.ErrorS(err, "Failed to add watch", "devicePluginPath", pluginapi.DevicePluginPath) + logger.Error(err, "Failed to add watch", "devicePluginPath", pluginapi.DevicePluginPath) return err } go func() { defer m.wg.Done() if err = m.server.Serve(sock); err != nil { - klog.ErrorS(err, "Error while serving device plugin registration grpc server") + logger.Error(err, "Error while serving device plugin registration grpc server") } }() var lastDialErr error wait.PollImmediate(1*time.Second, 10*time.Second, func() (bool, error) { var conn *grpc.ClientConn - _, conn, lastDialErr = dial(m.socket) + _, conn, lastDialErr = dial(ctx, m.socket) if lastDialErr != nil { return false, nil } @@ -178,12 +179,12 @@ func (m *Stub) Start() error { return lastDialErr } - klog.InfoS("Starting to serve on socket", "socket", m.socket) + logger.Info("Starting to serve on socket", "socket", m.socket) return nil } -func (m *Stub) Restart() error { - klog.InfoS("Restarting Device Plugin server") +func (m *Stub) Restart(ctx context.Context) error { + klog.FromContext(ctx).Info("Restarting Device Plugin server") if m.server == nil { return nil } @@ -191,14 +192,14 @@ func (m *Stub) Restart() error { m.server.Stop() m.server = nil - return m.Start() + return m.Start(ctx) } // Stop stops the gRPC server. Can be called without a prior Start // and more than once. Not safe to be called concurrently by different // goroutines! -func (m *Stub) Stop() error { - klog.InfoS("Stopping device plugin server") +func (m *Stub) Stop(logger klog.Logger) error { + logger.Info("Stopping device plugin server") if m.server == nil { return nil } @@ -213,7 +214,8 @@ func (m *Stub) Stop() error { return m.cleanup() } -func (m *Stub) Watch(kubeletEndpoint, resourceName, pluginSockDir string) { +func (m *Stub) Watch(ctx context.Context, kubeletEndpoint, resourceName, pluginSockDir string) { + logger := klog.FromContext(ctx) for { select { // Detect a kubelet restart by watching for a newly created @@ -221,26 +223,26 @@ func (m *Stub) Watch(kubeletEndpoint, resourceName, pluginSockDir string) { // the device plugin server case event := <-m.kubeletRestartWatcher.Events: if event.Name == kubeletEndpoint && event.Op&fsnotify.Create == fsnotify.Create { - klog.InfoS("inotify: file created, restarting", "kubeletEndpoint", kubeletEndpoint) + logger.Info("inotify: file created, restarting", "kubeletEndpoint", kubeletEndpoint) var lastErr error - err := wait.PollUntilContextTimeout(context.Background(), 10*time.Second, 2*time.Minute, false, func(context.Context) (done bool, err error) { - restartErr := m.Restart() + err := wait.PollUntilContextTimeout(ctx, 10*time.Second, 2*time.Minute, false, func(context.Context) (done bool, err error) { + restartErr := m.Restart(ctx) if restartErr == nil { return true, nil } - klog.ErrorS(restartErr, "Retrying after error") + logger.Error(restartErr, "Retrying after error") lastErr = restartErr return false, nil }) if err != nil { - klog.ErrorS(err, "Unable to restart server: wait timed out", "lastErr", lastErr.Error()) + logger.Error(err, "Unable to restart server: wait timed out", "lastErr", lastErr.Error()) panic(err) } if ok := m.registerControlFunc(); ok { - if err := m.Register(kubeletEndpoint, resourceName, pluginSockDir); err != nil { - klog.ErrorS(err, "Unable to register to kubelet") + if err := m.Register(ctx, kubeletEndpoint, resourceName, pluginSockDir); err != nil { + logger.Error(err, "Unable to register to kubelet") panic(err) } } @@ -248,14 +250,14 @@ func (m *Stub) Watch(kubeletEndpoint, resourceName, pluginSockDir string) { // Watch for any other fs errors and log them. case err := <-m.kubeletRestartWatcher.Errors: - klog.ErrorS(err, "inotify error") + logger.Error(err, "inotify error") } } } // GetInfo is the RPC which return pluginInfo func (m *Stub) GetInfo(ctx context.Context, req *watcherapi.InfoRequest) (*watcherapi.PluginInfo, error) { - klog.InfoS("GetInfo") + klog.FromContext(ctx).Info("GetInfo") return &watcherapi.PluginInfo{ Type: watcherapi.DevicePlugin, Name: m.resourceName, @@ -265,30 +267,32 @@ func (m *Stub) GetInfo(ctx context.Context, req *watcherapi.InfoRequest) (*watch // NotifyRegistrationStatus receives the registration notification from watcher func (m *Stub) NotifyRegistrationStatus(ctx context.Context, status *watcherapi.RegistrationStatus) (*watcherapi.RegistrationStatusResponse, error) { + logger := klog.FromContext(ctx) if m.registrationStatus != nil { m.registrationStatus <- *status } if !status.PluginRegistered { - klog.InfoS("Registration failed", "err", status.Error) + logger.Info("Registration failed", "err", status.Error) } return &watcherapi.RegistrationStatusResponse{}, nil } // Register registers the device plugin for the given resourceName with Kubelet. -func (m *Stub) Register(kubeletEndpoint, resourceName string, pluginSockDir string) error { - klog.InfoS("Register", "kubeletEndpoint", kubeletEndpoint, "resourceName", resourceName, "socket", pluginSockDir) +func (m *Stub) Register(ctx context.Context, kubeletEndpoint, resourceName string, pluginSockDir string) error { + logger := klog.FromContext(ctx) + logger.Info("Register", "kubeletEndpoint", kubeletEndpoint, "resourceName", resourceName, "socket", pluginSockDir) if pluginSockDir != "" { if _, err := os.Stat(pluginSockDir + "DEPRECATION"); err == nil { - klog.InfoS("Deprecation file found. Skip registration") + logger.Info("Deprecation file found. Skip registration") return nil } } - klog.InfoS("Deprecation file not found. Invoke registration") - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + logger.Info("Deprecation file not found. Invoke registration") + ctxDial, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() - conn, err := grpc.DialContext(ctx, kubeletEndpoint, + conn, err := grpc.DialContext(ctxDial, kubeletEndpoint, grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithBlock(), grpc.WithContextDialer(func(ctx context.Context, addr string) (net.Conn, error) { @@ -309,14 +313,14 @@ func (m *Stub) Register(kubeletEndpoint, resourceName string, pluginSockDir stri }, } - _, err = client.Register(context.Background(), reqt) + _, err = client.Register(ctx, reqt) if err != nil { // Stop server m.server.Stop() - klog.ErrorS(err, "Client unable to register to kubelet") + logger.Error(err, "Client unable to register to kubelet") return err } - klog.InfoS("Device Plugin registered with the Kubelet") + logger.Info("Device Plugin registered with the Kubelet") return err } @@ -331,13 +335,15 @@ func (m *Stub) GetDevicePluginOptions(ctx context.Context, e *pluginapi.Empty) ( // PreStartContainer resets the devices received func (m *Stub) PreStartContainer(ctx context.Context, r *pluginapi.PreStartContainerRequest) (*pluginapi.PreStartContainerResponse, error) { - klog.InfoS("PreStartContainer", "request", r) + klog.FromContext(ctx).Info("PreStartContainer", "request", r) return &pluginapi.PreStartContainerResponse{}, nil } // ListAndWatch lists devices and update that list according to the Update call func (m *Stub) ListAndWatch(e *pluginapi.Empty, s pluginapi.DevicePlugin_ListAndWatchServer) error { - klog.InfoS("ListAndWatch") + // Use klog.TODO() because we currently do not have a proper logger to pass in. + // Replace this with an appropriate context when refactoring this function to accept a logger parameter. + klog.TODO().Info("ListAndWatch") s.Send(&pluginapi.ListAndWatchResponse{Devices: m.devs}) @@ -358,7 +364,7 @@ func (m *Stub) Update(devs []*pluginapi.Device) { // GetPreferredAllocation gets the preferred allocation from a set of available devices func (m *Stub) GetPreferredAllocation(ctx context.Context, r *pluginapi.PreferredAllocationRequest) (*pluginapi.PreferredAllocationResponse, error) { - klog.InfoS("GetPreferredAllocation", "request", r) + klog.FromContext(ctx).Info("GetPreferredAllocation", "request", r) devs := make(map[string]*pluginapi.Device) @@ -371,7 +377,7 @@ func (m *Stub) GetPreferredAllocation(ctx context.Context, r *pluginapi.Preferre // Allocate does a mock allocation func (m *Stub) Allocate(ctx context.Context, r *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) { - klog.InfoS("Allocate", "request", r) + klog.FromContext(ctx).Info("Allocate", "request", r) devs := make(map[string]*pluginapi.Device) diff --git a/pkg/kubelet/cm/devicemanager/pod_devices.go b/pkg/kubelet/cm/devicemanager/pod_devices.go index 2270a188a81..9ed59575bd3 100644 --- a/pkg/kubelet/cm/devicemanager/pod_devices.go +++ b/pkg/kubelet/cm/devicemanager/pod_devices.go @@ -198,7 +198,7 @@ func (pdev *podDevices) getPodAndContainerForDevice(deviceID string) (string, st } // Turns podDevices to checkpointData. -func (pdev *podDevices) toCheckpointData() []checkpoint.PodDevicesEntry { +func (pdev *podDevices) toCheckpointData(logger klog.Logger) []checkpoint.PodDevicesEntry { var data []checkpoint.PodDevicesEntry pdev.RLock() defer pdev.RUnlock() @@ -206,13 +206,13 @@ func (pdev *podDevices) toCheckpointData() []checkpoint.PodDevicesEntry { for conName, resources := range containerDevices { for resource, devices := range resources { if devices.allocResp == nil { - klog.ErrorS(nil, "Can't marshal allocResp, allocation response is missing", "podUID", podUID, "containerName", conName, "resourceName", resource) + logger.Error(nil, "Can't marshal allocResp, allocation response is missing", "podUID", podUID, "containerName", conName, "resourceName", resource) continue } allocResp, err := proto.Marshal(devices.allocResp) if err != nil { - klog.ErrorS(err, "Can't marshal allocResp", "podUID", podUID, "containerName", conName, "resourceName", resource) + logger.Error(err, "Can't marshal allocResp", "podUID", podUID, "containerName", conName, "resourceName", resource) continue } data = append(data, checkpoint.PodDevicesEntry{ @@ -228,15 +228,15 @@ func (pdev *podDevices) toCheckpointData() []checkpoint.PodDevicesEntry { } // Populates podDevices from the passed in checkpointData. -func (pdev *podDevices) fromCheckpointData(data []checkpoint.PodDevicesEntry) { +func (pdev *podDevices) fromCheckpointData(logger klog.Logger, data []checkpoint.PodDevicesEntry) { for _, entry := range data { - klog.V(2).InfoS("Get checkpoint entry", + logger.V(2).Info("Get checkpoint entry", "podUID", entry.PodUID, "containerName", entry.ContainerName, "resourceName", entry.ResourceName, "deviceIDs", entry.DeviceIDs, "allocated", entry.AllocResp) allocResp := &pluginapi.ContainerAllocateResponse{} err := proto.Unmarshal(entry.AllocResp, allocResp) if err != nil { - klog.ErrorS(err, "Can't unmarshal allocResp", "podUID", entry.PodUID, "containerName", entry.ContainerName, "resourceName", entry.ResourceName) + logger.Error(err, "Can't unmarshal allocResp", "podUID", entry.PodUID, "containerName", entry.ContainerName, "resourceName", entry.ResourceName) continue } pdev.insert(entry.PodUID, entry.ContainerName, entry.ResourceName, entry.DeviceIDs, allocResp) @@ -244,7 +244,7 @@ func (pdev *podDevices) fromCheckpointData(data []checkpoint.PodDevicesEntry) { } // Returns combined container runtime settings to consume the container's allocated devices. -func (pdev *podDevices) deviceRunContainerOptions(podUID, contName string) *DeviceRunContainerOptions { +func (pdev *podDevices) deviceRunContainerOptions(logger klog.Logger, podUID, contName string) *DeviceRunContainerOptions { pdev.RLock() defer pdev.RUnlock() @@ -277,13 +277,13 @@ func (pdev *podDevices) deviceRunContainerOptions(podUID, contName string) *Devi // Updates RunContainerOptions.Envs. for k, v := range resp.Envs { if e, ok := envsMap[k]; ok { - klog.V(4).InfoS("Skip existing env", "envKey", k, "envValue", v) + logger.V(4).Info("Skip existing env", "envKey", k, "envValue", v) if e != v { - klog.ErrorS(nil, "Environment variable has conflicting setting", "envKey", k, "expected", v, "got", e) + logger.Error(nil, "Environment variable has conflicting setting", "envKey", k, "expected", v, "got", e) } continue } - klog.V(4).InfoS("Add env", "envKey", k, "envValue", v) + logger.V(4).Info("Add env", "envKey", k, "envValue", v) envsMap[k] = v opts.Envs = append(opts.Envs, kubecontainer.EnvVar{Name: k, Value: v}) } @@ -291,14 +291,14 @@ func (pdev *podDevices) deviceRunContainerOptions(podUID, contName string) *Devi // Updates RunContainerOptions.Devices. for _, dev := range resp.Devices { if d, ok := devsMap[dev.ContainerPath]; ok { - klog.V(4).InfoS("Skip existing device", "containerPath", dev.ContainerPath, "hostPath", dev.HostPath) + logger.V(4).Info("Skip existing device", "containerPath", dev.ContainerPath, "hostPath", dev.HostPath) if d != dev.HostPath { - klog.ErrorS(nil, "Container device has conflicting mapping host devices", + logger.Error(nil, "Container device has conflicting mapping host devices", "containerPath", dev.ContainerPath, "got", d, "expected", dev.HostPath) } continue } - klog.V(4).InfoS("Add device", "containerPath", dev.ContainerPath, "hostPath", dev.HostPath) + logger.V(4).Info("Add device", "containerPath", dev.ContainerPath, "hostPath", dev.HostPath) devsMap[dev.ContainerPath] = dev.HostPath opts.Devices = append(opts.Devices, kubecontainer.DeviceInfo{ PathOnHost: dev.HostPath, @@ -310,14 +310,14 @@ func (pdev *podDevices) deviceRunContainerOptions(podUID, contName string) *Devi // Updates RunContainerOptions.Mounts. for _, mount := range resp.Mounts { if m, ok := mountsMap[mount.ContainerPath]; ok { - klog.V(4).InfoS("Skip existing mount", "containerPath", mount.ContainerPath, "hostPath", mount.HostPath) + logger.V(4).Info("Skip existing mount", "containerPath", mount.ContainerPath, "hostPath", mount.HostPath) if m != mount.HostPath { - klog.ErrorS(nil, "Container mount has conflicting mapping host mounts", + logger.Error(nil, "Container mount has conflicting mapping host mounts", "containerPath", mount.ContainerPath, "conflictingPath", m, "hostPath", mount.HostPath) } continue } - klog.V(4).InfoS("Add mount", "containerPath", mount.ContainerPath, "hostPath", mount.HostPath) + logger.V(4).Info("Add mount", "containerPath", mount.ContainerPath, "hostPath", mount.HostPath) mountsMap[mount.ContainerPath] = mount.HostPath opts.Mounts = append(opts.Mounts, kubecontainer.Mount{ Name: mount.ContainerPath, @@ -332,19 +332,19 @@ func (pdev *podDevices) deviceRunContainerOptions(podUID, contName string) *Devi // Updates for Annotations for k, v := range resp.Annotations { if e, ok := annotationsMap[k]; ok { - klog.V(4).InfoS("Skip existing annotation", "annotationKey", k, "annotationValue", v) + logger.V(4).Info("Skip existing annotation", "annotationKey", k, "annotationValue", v) if e != v { - klog.ErrorS(nil, "Annotation has conflicting setting", "annotationKey", k, "expected", e, "got", v) + logger.Error(nil, "Annotation has conflicting setting", "annotationKey", k, "expected", e, "got", v) } continue } - klog.V(4).InfoS("Add annotation", "annotationKey", k, "annotationValue", v) + logger.V(4).Info("Add annotation", "annotationKey", k, "annotationValue", v) annotationsMap[k] = v opts.Annotations = append(opts.Annotations, kubecontainer.Annotation{Name: k, Value: v}) } // Updates for CDI devices. - cdiDevices := getCDIDeviceInfo(resp, allCDIDevices) + cdiDevices := getCDIDeviceInfo(logger, resp, allCDIDevices) opts.CDIDevices = append(opts.CDIDevices, cdiDevices...) } @@ -352,14 +352,14 @@ func (pdev *podDevices) deviceRunContainerOptions(podUID, contName string) *Devi } // getCDIDeviceInfo returns CDI devices from an allocate response -func getCDIDeviceInfo(resp *pluginapi.ContainerAllocateResponse, knownCDIDevices sets.Set[string]) []kubecontainer.CDIDevice { +func getCDIDeviceInfo(logger klog.Logger, resp *pluginapi.ContainerAllocateResponse, knownCDIDevices sets.Set[string]) []kubecontainer.CDIDevice { var cdiDevices []kubecontainer.CDIDevice for _, cdiDevice := range resp.CdiDevices { if knownCDIDevices.Has(cdiDevice.Name) { - klog.V(4).InfoS("Skip existing CDI Device", "name", cdiDevice.Name) + logger.V(4).Info("Skip existing CDI Device", "name", cdiDevice.Name) continue } - klog.V(4).InfoS("Add CDI device", "name", cdiDevice.Name) + logger.V(4).Info("Add CDI device", "name", cdiDevice.Name) knownCDIDevices.Insert(cdiDevice.Name) device := kubecontainer.CDIDevice{ diff --git a/pkg/kubelet/cm/devicemanager/pod_devices_test.go b/pkg/kubelet/cm/devicemanager/pod_devices_test.go index b0c71086975..0e4d9f982d0 100644 --- a/pkg/kubelet/cm/devicemanager/pod_devices_test.go +++ b/pkg/kubelet/cm/devicemanager/pod_devices_test.go @@ -27,6 +27,7 @@ import ( pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" "k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/checkpoint" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" + "k8s.io/kubernetes/test/utils/ktesting" ) func TestGetContainerDevices(t *testing.T) { @@ -157,6 +158,7 @@ func expectResourceDeviceInstances(t *testing.T, resp ResourceDeviceInstances, e } func TestDeviceRunContainerOptions(t *testing.T) { + logger, _ := ktesting.NewTestContext(t) const ( podUID = "pod" containerName = "container" @@ -239,7 +241,7 @@ func TestDeviceRunContainerOptions(t *testing.T) { response, ) } - opts := podDevices.deviceRunContainerOptions(podUID, containerName) + opts := podDevices.deviceRunContainerOptions(logger, podUID, containerName) // The exact ordering of the options depends on the order of the resources in the map. // We therefore use `ElementsMatch` instead of `Equal` on the member slices. diff --git a/pkg/kubelet/cm/devicemanager/topology_hints.go b/pkg/kubelet/cm/devicemanager/topology_hints.go index f4d7fa73a8a..8846122a3d7 100644 --- a/pkg/kubelet/cm/devicemanager/topology_hints.go +++ b/pkg/kubelet/cm/devicemanager/topology_hints.go @@ -31,6 +31,9 @@ import ( // ensures the Device Manager is consulted when Topology Aware Hints for each // container are created. func (m *ManagerImpl) GetTopologyHints(pod *v1.Pod, container *v1.Container) map[string][]topologymanager.TopologyHint { + // Use klog.TODO() because we currently do not have a proper logger to pass in. + // Replace this with an appropriate logger when refactoring this function to accept a logger parameter. + logger := klog.TODO() // Garbage collect any stranded device resources before providing TopologyHints m.UpdateAllocatedDevices() @@ -43,7 +46,7 @@ func (m *ManagerImpl) GetTopologyHints(pod *v1.Pod, container *v1.Container) map for resource, requested := range accumulatedResourceRequests { // Only consider devices that actually contain topology information. if aligned := m.deviceHasTopologyAlignment(resource); !aligned { - klog.InfoS("Resource does not have a topology preference", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "request", requested) + logger.Info("Resource does not have a topology preference", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "request", requested) deviceHints[resource] = nil continue } @@ -54,11 +57,11 @@ func (m *ManagerImpl) GetTopologyHints(pod *v1.Pod, container *v1.Container) map allocated := m.podDevices.containerDevices(string(pod.UID), container.Name, resource) if allocated.Len() > 0 { if allocated.Len() != requested { - klog.InfoS("Resource already allocated to pod with different number than request", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "request", requested, "allocated", allocated.Len()) + logger.Info("Resource already allocated to pod with different number than request", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "request", requested, "allocated", allocated.Len()) deviceHints[resource] = []topologymanager.TopologyHint{} continue } - klog.InfoS("Regenerating TopologyHints for resource already allocated to pod", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name) + logger.Info("Regenerating TopologyHints for resource already allocated to pod", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name) deviceHints[resource] = m.generateDeviceTopologyHints(resource, allocated, sets.Set[string]{}, requested) continue } @@ -67,7 +70,7 @@ func (m *ManagerImpl) GetTopologyHints(pod *v1.Pod, container *v1.Container) map available := m.getAvailableDevices(resource) reusable := m.devicesToReuse[string(pod.UID)][resource] if available.Union(reusable).Len() < requested { - klog.InfoS("Unable to generate topology hints: requested number of devices unavailable", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "request", requested, "available", available.Union(reusable).Len()) + logger.Info("Unable to generate topology hints: requested number of devices unavailable", "resourceName", resource, "pod", klog.KObj(pod), "containerName", container.Name, "request", requested, "available", available.Union(reusable).Len()) deviceHints[resource] = []topologymanager.TopologyHint{} continue } @@ -83,6 +86,9 @@ func (m *ManagerImpl) GetTopologyHints(pod *v1.Pod, container *v1.Container) map // GetPodTopologyHints implements the topologymanager.HintProvider Interface which // ensures the Device Manager is consulted when Topology Aware Hints for Pod are created. func (m *ManagerImpl) GetPodTopologyHints(pod *v1.Pod) map[string][]topologymanager.TopologyHint { + // Use klog.TODO() because we currently do not have a proper logger to pass in. + // Replace this with an appropriate logger when refactoring this function to accept a logger parameter. + logger := klog.TODO() // Garbage collect any stranded device resources before providing TopologyHints m.UpdateAllocatedDevices() @@ -94,7 +100,7 @@ func (m *ManagerImpl) GetPodTopologyHints(pod *v1.Pod) map[string][]topologymana for resource, requested := range accumulatedResourceRequests { // Only consider devices that actually contain topology information. if aligned := m.deviceHasTopologyAlignment(resource); !aligned { - klog.InfoS("Resource does not have a topology preference", "resourceName", resource, "pod", klog.KObj(pod), "request", requested) + logger.Info("Resource does not have a topology preference", "resourceName", resource, "pod", klog.KObj(pod), "request", requested) deviceHints[resource] = nil continue } @@ -105,11 +111,11 @@ func (m *ManagerImpl) GetPodTopologyHints(pod *v1.Pod) map[string][]topologymana allocated := m.podDevices.podDevices(string(pod.UID), resource) if allocated.Len() > 0 { if allocated.Len() != requested { - klog.InfoS("Resource already allocated to pod with different number than request", "resourceName", resource, "pod", klog.KObj(pod), "request", requested, "allocated", allocated.Len()) + logger.Info("Resource already allocated to pod with different number than request", "resourceName", resource, "pod", klog.KObj(pod), "request", requested, "allocated", allocated.Len()) deviceHints[resource] = []topologymanager.TopologyHint{} continue } - klog.InfoS("Regenerating TopologyHints for resource already allocated to pod", "resourceName", resource, "pod", klog.KObj(pod), "allocated", allocated.Len()) + logger.Info("Regenerating TopologyHints for resource already allocated to pod", "resourceName", resource, "pod", klog.KObj(pod), "allocated", allocated.Len()) deviceHints[resource] = m.generateDeviceTopologyHints(resource, allocated, sets.Set[string]{}, requested) continue } @@ -117,7 +123,7 @@ func (m *ManagerImpl) GetPodTopologyHints(pod *v1.Pod) map[string][]topologymana // Get the list of available devices, for which TopologyHints should be generated. available := m.getAvailableDevices(resource) if available.Len() < requested { - klog.InfoS("Unable to generate topology hints: requested number of devices unavailable", "resourceName", resource, "pod", klog.KObj(pod), "request", requested, "available", available.Len()) + logger.Info("Unable to generate topology hints: requested number of devices unavailable", "resourceName", resource, "pod", klog.KObj(pod), "request", requested, "available", available.Len()) deviceHints[resource] = []topologymanager.TopologyHint{} continue } diff --git a/pkg/kubelet/cm/devicemanager/topology_hints_test.go b/pkg/kubelet/cm/devicemanager/topology_hints_test.go index 2ab1e957fcf..53ea645a3af 100644 --- a/pkg/kubelet/cm/devicemanager/topology_hints_test.go +++ b/pkg/kubelet/cm/devicemanager/topology_hints_test.go @@ -29,6 +29,7 @@ import ( pluginapi "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" "k8s.io/kubernetes/pkg/kubelet/cm/topologymanager" "k8s.io/kubernetes/pkg/kubelet/cm/topologymanager/bitmask" + "k8s.io/kubernetes/test/utils/ktesting" ) type mockAffinityStore struct { @@ -109,6 +110,7 @@ func TestGetTopologyHints(t *testing.T) { } func TestTopologyAlignedAllocation(t *testing.T) { + tCtx := ktesting.Init(t) tcases := []struct { description string resource string @@ -441,7 +443,7 @@ func TestTopologyAlignedAllocation(t *testing.T) { } } - allocated, err := m.devicesToAllocate("podUID", "containerName", tc.resource, tc.request, sets.New[string]()) + allocated, err := m.devicesToAllocate(tCtx, "podUID", "containerName", tc.resource, tc.request, sets.New[string]()) if err != nil { t.Errorf("Unexpected error: %v", err) continue @@ -471,6 +473,7 @@ func TestTopologyAlignedAllocation(t *testing.T) { } func TestGetPreferredAllocationParameters(t *testing.T) { + tCtx := ktesting.Init(t) tcases := []struct { description string resource string @@ -639,7 +642,7 @@ func TestGetPreferredAllocationParameters(t *testing.T) { opts: &pluginapi.DevicePluginOptions{GetPreferredAllocationAvailable: true}, } - _, err := m.devicesToAllocate("podUID", "containerName", tc.resource, tc.request, sets.New[string](tc.reusableDevices...)) + _, err := m.devicesToAllocate(tCtx, "podUID", "containerName", tc.resource, tc.request, sets.New[string](tc.reusableDevices...)) if err != nil { t.Errorf("Unexpected error: %v", err) continue diff --git a/pkg/kubelet/cm/devicemanager/types.go b/pkg/kubelet/cm/devicemanager/types.go index bdf0b27d673..b377443c434 100644 --- a/pkg/kubelet/cm/devicemanager/types.go +++ b/pkg/kubelet/cm/devicemanager/types.go @@ -17,11 +17,13 @@ limitations under the License. package devicemanager import ( + "context" "time" v1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apiserver/pkg/server/healthz" + "k8s.io/klog/v2" "k8s.io/kubernetes/pkg/kubelet/cm/containermap" "k8s.io/kubernetes/pkg/kubelet/cm/resourceupdates" "k8s.io/kubernetes/pkg/kubelet/cm/topologymanager" @@ -35,7 +37,7 @@ import ( // Manager manages all the Device Plugins running on a node. type Manager interface { // Start starts device plugin registration service. - Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady, initialContainers containermap.ContainerMap, initialContainerRunningSet sets.Set[string]) error + Start(logger klog.Logger, activePods ActivePodsFunc, sourcesReady config.SourcesReady, initialContainers containermap.ContainerMap, initialContainerRunningSet sets.Set[string]) error // Allocate configures and assigns devices to a container in a pod. From // the requested device resources, Allocate will communicate with the @@ -50,12 +52,12 @@ type Manager interface { UpdatePluginResources(node *schedulerframework.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error // Stop stops the manager. - Stop() error + Stop(logger klog.Logger) error // GetDeviceRunContainerOptions checks whether we have cached containerDevices // for the passed-in and returns its DeviceRunContainerOptions // for the found one. An empty struct is returned in case no cached state is found. - GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Container) (*DeviceRunContainerOptions, error) + GetDeviceRunContainerOptions(ctx context.Context, pod *v1.Pod, container *v1.Container) (*DeviceRunContainerOptions, error) // GetCapacity returns the amount of available device plugin resource capacity, resource allocatable // and inactive device plugin resources previously registered on the node.