Skip to content

K
k3s
Commit ff4e8d42 authored by Jiaying Zhang's avatar Jiaying Zhang
Browse files
Options

Device plugin code refactoring to cope with file move.

While moving device_plugin_handler_test.go from pkg/kubelet/cm/ to pkg/kubelet/cm/deviceplugin/, we can no longer uses cm in its tests because that would cause a cycle dependency. To solve this problem, I moved the main cm GetResources functionality as well as part of the current device plugin handler Allocate functionality into a new device plugin handler function, GetDeviceRunContainerOptions(). This refactoring is also needed by another PR 51895 that moves device allocation into admission phase. Now device plugin handler Allocate() first checks whether there is cached device runtime state and only issues Allocate grpc call if there is no cached state available. The new GetDeviceRunContainerOptions() function simply returns device runtime config from the cached state. To support this change, extended the podDevices struct and checkpoint data structure with device runtime state.
parent 796f4887
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 63 additions and 93 deletions
pkg/kubelet/BUILD
View file @ ff4e8d42
......@@ -260,7 +260,6 @@ filegroup(
"//pkg/kubelet/configmap:all-srcs",
"//pkg/kubelet/container:all-srcs",
"//pkg/kubelet/custommetrics:all-srcs",
"//pkg/kubelet/deviceplugin:all-srcs",
"//pkg/kubelet/dockershim:all-srcs",
"//pkg/kubelet/envvars:all-srcs",
"//pkg/kubelet/events:all-srcs",
......
pkg/kubelet/cm/BUILD
View file @ ff4e8d42
......@@ -7,8 +7,6 @@ go_library(
"container_manager.go",
"container_manager_stub.go",
"container_manager_unsupported.go",
"device_plugin_handler.go",
"device_plugin_handler_stub.go",
"fake_internal_container_lifecycle.go",
"helpers_unsupported.go",
"internal_container_lifecycle.go",
......@@ -34,18 +32,15 @@ go_library(
deps = [
"//pkg/features:go_default_library",
"//pkg/kubelet/apis/cri:go_default_library",
"//pkg/kubelet/apis/deviceplugin/v1alpha:go_default_library",
"//pkg/kubelet/apis/kubeletconfig:go_default_library",
"//pkg/kubelet/cadvisor:go_default_library",
"//pkg/kubelet/cm/cpumanager:go_default_library",
"//pkg/kubelet/container:go_default_library",
"//pkg/kubelet/deviceplugin:go_default_library",
"//pkg/kubelet/eviction/api:go_default_library",
"//pkg/kubelet/status:go_default_library",
"//pkg/util/mount:go_default_library",
"//vendor/github.com/golang/glog:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
"//vendor/k8s.io/apiserver/pkg/util/feature:go_default_library",
......@@ -55,6 +50,7 @@ go_library(
"//pkg/api/v1/helper:go_default_library",
"//pkg/api/v1/helper/qos:go_default_library",
"//pkg/api/v1/resource:go_default_library",
"//pkg/kubelet/cm/deviceplugin:go_default_library",
"//pkg/kubelet/cm/util:go_default_library",
"//pkg/kubelet/events:go_default_library",
"//pkg/kubelet/metrics:go_default_library",
......@@ -69,6 +65,7 @@ go_library(
"//vendor/github.com/opencontainers/runc/libcontainer/cgroups/fs:go_default_library",
"//vendor/github.com/opencontainers/runc/libcontainer/cgroups/systemd:go_default_library",
"//vendor/github.com/opencontainers/runc/libcontainer/configs:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/errors:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
],
......@@ -80,7 +77,6 @@ go_test(
name = "go_default_test",
srcs = [
"container_manager_unsupported_test.go",
"device_plugin_handler_test.go",
] + select({
"@io_bazel_rules_go//go/platform:linux_amd64": [
"cgroup_manager_linux_test.go",
......@@ -94,19 +90,15 @@ go_test(
importpath = "k8s.io/kubernetes/pkg/kubelet/cm",
library = ":go_default_library",
deps = [
"//pkg/kubelet/apis/deviceplugin/v1alpha:go_default_library",
"//pkg/util/mount:go_default_library",
"//vendor/github.com/stretchr/testify/assert:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/uuid:go_default_library",
] + select({
"@io_bazel_rules_go//go/platform:linux_amd64": [
"//pkg/kubelet/apis/kubeletconfig:go_default_library",
"//pkg/kubelet/eviction/api:go_default_library",
"//vendor/github.com/stretchr/testify/assert:go_default_library",
"//vendor/github.com/stretchr/testify/require:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
],
"//conditions:default": [],
}),
......@@ -125,6 +117,7 @@ filegroup(
":package-srcs",
"//pkg/kubelet/cm/cpumanager:all-srcs",
"//pkg/kubelet/cm/cpuset:all-srcs",
"//pkg/kubelet/cm/deviceplugin:all-srcs",
"//pkg/kubelet/cm/util:all-srcs",
],
tags = ["automanaged"],
......
pkg/kubelet/cm/container_manager_linux.go
View file @ ff4e8d42
......@@ -45,6 +45,7 @@ import (
internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
"k8s.io/kubernetes/pkg/kubelet/cadvisor"
"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
"k8s.io/kubernetes/pkg/kubelet/cm/deviceplugin"
cmutil "k8s.io/kubernetes/pkg/kubelet/cm/util"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/qos"
......@@ -125,7 +126,7 @@ type containerManagerImpl struct {
// Interface for QoS cgroup management
qosContainerManager QOSContainerManager
// Interface for exporting and allocating devices reported by device plugins.
devicePluginHandler DevicePluginHandler
devicePluginHandler deviceplugin.Handler
// Interface for CPU affinity management.
cpuManager cpumanager.Manager
}
......@@ -273,9 +274,9 @@ func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.I
glog.Infof("Creating device plugin handler: %t", devicePluginEnabled)
if devicePluginEnabled {
cm.devicePluginHandler, err = NewDevicePluginHandlerImpl(updateDeviceCapacityFunc)
cm.devicePluginHandler, err = deviceplugin.NewHandlerImpl(updateDeviceCapacityFunc)
} else {
cm.devicePluginHandler, err = NewDevicePluginHandlerStub()
cm.devicePluginHandler, err = deviceplugin.NewHandlerStub()
}
if err != nil {
return nil, err
......@@ -618,72 +619,17 @@ func (cm *containerManagerImpl) GetResources(pod *v1.Pod, container *v1.Containe
opts := &kubecontainer.RunContainerOptions{}
// Gets devices, mounts, and envs from device plugin handler.
glog.V(3).Infof("Calling devicePluginHandler AllocateDevices")
// Maps to detect duplicate settings.
devsMap := make(map[string]string)
mountsMap := make(map[string]string)
envsMap := make(map[string]string)
allocResps, err := cm.devicePluginHandler.Allocate(pod, container, activePods)
err := cm.devicePluginHandler.Allocate(pod, container, activePods)
if err != nil {
return opts, err
}
// Loops through AllocationResponses of all required extended resources.
for _, resp := range allocResps {
// Each Allocate response has the following artifacts.
// Environment variables
// Mount points
// Device files
// These artifacts are per resource per container.
// Updates RunContainerOptions.Envs.
for k, v := range resp.Envs {
if e, ok := envsMap[k]; ok {
glog.V(3).Infof("skip existing envs %s %s", k, v)
if e != v {
glog.Errorf("Environment variable %s has conflicting setting: %s and %s", k, e, v)
}
continue
}
envsMap[k] = v
opts.Envs = append(opts.Envs, kubecontainer.EnvVar{Name: k, Value: v})
}
// Updates RunContainerOptions.Devices.
for _, dev := range resp.Devices {
if d, ok := devsMap[dev.ContainerPath]; ok {
glog.V(3).Infof("skip existing device %s %s", dev.ContainerPath, dev.HostPath)
if d != dev.HostPath {
glog.Errorf("Container device %s has conflicting mapping host devices: %s and %s",
dev.ContainerPath, d, dev.HostPath)
}
continue
}
devsMap[dev.ContainerPath] = dev.HostPath
opts.Devices = append(opts.Devices, kubecontainer.DeviceInfo{
PathOnHost: dev.HostPath,
PathInContainer: dev.ContainerPath,
Permissions: dev.Permissions,
})
}
// Updates RunContainerOptions.Mounts.
for _, mount := range resp.Mounts {
if m, ok := mountsMap[mount.ContainerPath]; ok {
glog.V(3).Infof("skip existing mount %s %s", mount.ContainerPath, mount.HostPath)
if m != mount.HostPath {
glog.Errorf("Container mount %s has conflicting mapping host mounts: %s and %s",
mount.ContainerPath, m, mount.HostPath)
}
continue
}
mountsMap[mount.ContainerPath] = mount.HostPath
opts.Mounts = append(opts.Mounts, kubecontainer.Mount{
Name: mount.ContainerPath,
ContainerPath: mount.ContainerPath,
HostPath: mount.HostPath,
ReadOnly: mount.ReadOnly,
// TODO: This may need to be part of Device plugin API.
SELinuxRelabel: false,
})
}
devOpts := cm.devicePluginHandler.GetDeviceRunContainerOptions(pod, container)
if devOpts == nil {
return opts, nil
}
opts.Devices = append(opts.Devices, devOpts.Devices...)
opts.Mounts = append(opts.Mounts, devOpts.Mounts...)
opts.Envs = append(opts.Envs, devOpts.Envs...)
return opts, nil
}
......
pkg/kubelet/cm/deviceplugin/BUILD
View file @ ff4e8d42
......@@ -9,19 +9,25 @@ load(
go_library(
name = "go_default_library",
srcs = [
"device_plugin_handler.go",
"device_plugin_handler_stub.go",
"device_plugin_stub.go",
"endpoint.go",
"manager.go",
"pod_devices.go",
"types.go",
],
importpath = "k8s.io/kubernetes/pkg/kubelet/deviceplugin",
importpath = "k8s.io/kubernetes/pkg/kubelet/cm/deviceplugin",
deps = [
"//pkg/api/v1/helper:go_default_library",
"//pkg/kubelet/apis/deviceplugin/v1alpha:go_default_library",
"//pkg/kubelet/container:go_default_library",
"//vendor/github.com/golang/glog:go_default_library",
"//vendor/golang.org/x/net/context:go_default_library",
"//vendor/google.golang.org/grpc:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
],
)
......@@ -41,13 +47,20 @@ filegroup(
go_test(
name = "go_default_test",
srcs = [
"device_plugin_handler_test.go",
"endpoint_test.go",
"manager_test.go",
],
importpath = "k8s.io/kubernetes/pkg/kubelet/deviceplugin",
importpath = "k8s.io/kubernetes/pkg/kubelet/cm/deviceplugin",
library = ":go_default_library",
deps = [
"//pkg/kubelet/apis/deviceplugin/v1alpha:go_default_library",
"//vendor/github.com/stretchr/testify/assert:go_default_library",
"//vendor/github.com/stretchr/testify/require:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/api/resource:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/sets:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/util/uuid:go_default_library",
],
)
pkg/kubelet/cm/deviceplugin/device_plugin_handler_stub.go
View file @ ff4e8d42
......@@ -14,29 +14,37 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package cm
package deviceplugin
import (
"k8s.io/api/core/v1"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1alpha"
)
// A simple stub implementation for DevicePluginHandler.
type DevicePluginHandlerStub struct{}
// HandlerStub provides a simple stub implementation for Handler.
type HandlerStub struct{}
func NewDevicePluginHandlerStub() (*DevicePluginHandlerStub, error) {
return &DevicePluginHandlerStub{}, nil
// NewHandlerStub creates a HandlerStub.
func NewHandlerStub() (*HandlerStub, error) {
return &HandlerStub{}, nil
}
func (h *DevicePluginHandlerStub) Start() error {
// Start simply returns nil.
func (h *HandlerStub) Start() error {
return nil
}
func (h *DevicePluginHandlerStub) Devices() map[string][]pluginapi.Device {
// Devices returns an empty map.
func (h *HandlerStub) Devices() map[string][]pluginapi.Device {
return make(map[string][]pluginapi.Device)
}
func (h *DevicePluginHandlerStub) Allocate(pod *v1.Pod, container *v1.Container, activePods []*v1.Pod) ([]*pluginapi.AllocateResponse, error) {
var ret []*pluginapi.AllocateResponse
return ret, nil
// Allocate simply returns nil.
func (h *HandlerStub) Allocate(pod *v1.Pod, container *v1.Container, activePods []*v1.Pod) error {
return nil
}
// GetDeviceRunContainerOptions simply returns nil.
func (h *HandlerStub) GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Container) *DeviceRunContainerOptions {
return nil
}
pkg/kubelet/cm/deviceplugin/manager.go
View file @ ff4e8d42
......@@ -141,7 +141,7 @@ func (m *ManagerImpl) Devices() map[string][]pluginapi.Device {
devs := make(map[string][]pluginapi.Device)
for k, e := range m.endpoints {
glog.V(3).Infof("Endpoint: %+v: %+v", k, e)
glog.V(3).Infof("Endpoint: %+v: %p", k, e)
devs[k] = e.getDevices()
}
......@@ -221,8 +221,8 @@ func (m *ManagerImpl) addEndpoint(r *pluginapi.RegisterRequest) {
m.mutex.Lock()
old, ok := m.endpoints[r.ResourceName]
m.endpoints[r.ResourceName] = e
m.mutex.Unlock()
glog.V(2).Infof("Registered endpoint %v", e)
m.mutex.Unlock()
if ok && old != nil {
old.stop()
......
pkg/kubelet/cm/deviceplugin/types.go
View file @ ff4e8d42
......@@ -18,6 +18,7 @@ package deviceplugin
import (
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1alpha"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
)
// MonitorCallback is the function called when a device's health state changes,
......@@ -46,6 +47,16 @@ type Manager interface {
CheckpointFile() string
}
// DeviceRunContainerOptions contains the combined container runtime settings to consume its allocated devices.
type DeviceRunContainerOptions struct {
// The environment variables list.
Envs []kubecontainer.EnvVar
// The mounts for the container.
Mounts []kubecontainer.Mount
// The host devices mapped into the container.
Devices []kubecontainer.DeviceInfo
}
// TODO: evaluate whether we need these error definitions.
const (
// errFailedToDialDevicePlugin is the error raised when the device plugin could not be
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment