github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-08-08 23:37:11 +00:00
Code Issues Projects Releases Wiki Activity

Migrate device manager to contextual logging

Browse Source
This commit is contained in:
Ed Bartosh
2025-09-26 16:47:33 +03:00
parent 243d8c000e
commit d45a4557c1
19 changed files with 450 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
hack/golangci-hints.yaml
View File

@@ -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/.*

1
hack/golangci.yaml
View File

@@ -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/.*

1
hack/logcheck.conf
View File

@@ -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/.*

4
pkg/kubelet/cm/container_manager_linux.go
View File

@@ -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 {

7
pkg/kubelet/cm/container_manager_windows.go
View File

@@ -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 {

18
pkg/kubelet/cm/devicemanager/endpoint.go
View File

@@ -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,

68
pkg/kubelet/cm/devicemanager/endpoint_test.go
View File

@@ -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)
}

174
pkg/kubelet/cm/devicemanager/manager.go
View File

@@ -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 <pod, container> 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
}

201
pkg/kubelet/cm/devicemanager/manager_test.go
View File

@@ -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},
})
}

11
pkg/kubelet/cm/devicemanager/plugin/v1beta1/api.go
View File

@@ -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.

42
pkg/kubelet/cm/devicemanager/plugin/v1beta1/client.go
View File

@@ -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,

66
pkg/kubelet/cm/devicemanager/plugin/v1beta1/handler.go
View File

@@ -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)
}
}

43
pkg/kubelet/cm/devicemanager/plugin/v1beta1/server.go
View File

@@ -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
}

80
pkg/kubelet/cm/devicemanager/plugin/v1beta1/stub.go
View File

@@ -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)

46
pkg/kubelet/cm/devicemanager/pod_devices.go
View File

@@ -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{

4
pkg/kubelet/cm/devicemanager/pod_devices_test.go
View File

@@ -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.

22
pkg/kubelet/cm/devicemanager/topology_hints.go
View File

@@ -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
}

7
pkg/kubelet/cm/devicemanager/topology_hints_test.go
View File

@@ -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

8
pkg/kubelet/cm/devicemanager/types.go
View File

@@ -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 <pod, container> 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.