Commit 7e31e70a authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub

Merge pull request #54019 from jiayingz/lifecycle

Automatic merge from submit-queue. If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>. Move device plugin related code to pkg/kubelet/cm/deviceplugin/. **What this PR does / why we need it**: Move device plugin related code to pkg/kubelet/cm/deviceplugin/. Inspired by PR 53258 proposed by @dolftax. The code structure looks much cleaner this way and it also makes future code refactoring easier. 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. We expect to clean up device plugin interface in follow up PRs. Currently both device plugin handler interface and manager interface are public. This is no longer necessary after this PR given that their code now live in the same package. We should clean up unnecessary public interfaces and functions, and/or consider to merge handler and manager code into a single layer. **Which issue this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close that issue when PR gets merged)*: fixes # **Special notes for your reviewer**: Part of https://github.com/kubernetes/kubernetes/issues/53170 but not fixes it. **Release note**: ```release-note ```
parents d274b67c e501f01d
......@@ -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",
......
......@@ -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"],
......
......@@ -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
}
......
......@@ -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",
],
)
......@@ -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
}
......@@ -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()
......
/*
Copyright 2017 The Kubernetes Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package deviceplugin
import (
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/util/sets"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1alpha"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
)
type deviceAllocateInfo struct {
// deviceIds contains device Ids allocated to this container for the given resourceName.
deviceIds sets.String
// allocResp contains cached rpc AllocateResponse.
allocResp *pluginapi.AllocateResponse
}
type resourceAllocateInfo map[string]deviceAllocateInfo // Keyed by resourceName.
type containerDevices map[string]resourceAllocateInfo // Keyed by containerName.
type podDevices map[string]containerDevices // Keyed by podUID.
func (pdev podDevices) pods() sets.String {
ret := sets.NewString()
for k := range pdev {
ret.Insert(k)
}
return ret
}
func (pdev podDevices) insert(podUID, contName, resource string, devices sets.String, resp *pluginapi.AllocateResponse) {
if _, podExists := pdev[podUID]; !podExists {
pdev[podUID] = make(containerDevices)
}
if _, contExists := pdev[podUID][contName]; !contExists {
pdev[podUID][contName] = make(resourceAllocateInfo)
}
pdev[podUID][contName][resource] = deviceAllocateInfo{
deviceIds: devices,
allocResp: resp,
}
}
func (pdev podDevices) delete(pods []string) {
for _, uid := range pods {
delete(pdev, uid)
}
}
// Returns list of device Ids allocated to the given container for the given resource.
// Returns nil if we don't have cached state for the given <podUID, contName, resource>.
func (pdev podDevices) containerDevices(podUID, contName, resource string) sets.String {
if _, podExists := pdev[podUID]; !podExists {
return nil
}
if _, contExists := pdev[podUID][contName]; !contExists {
return nil
}
devs, resourceExists := pdev[podUID][contName][resource]
if !resourceExists {
return nil
}
return devs.deviceIds
}
// Returns all of devices allocated to the pods being tracked, keyed by resourceName.
func (pdev podDevices) devices() map[string]sets.String {
ret := make(map[string]sets.String)
for _, containerDevices := range pdev {
for _, resources := range containerDevices {
for resource, devices := range resources {
if _, exists := ret[resource]; !exists {
ret[resource] = sets.NewString()
}
ret[resource] = ret[resource].Union(devices.deviceIds)
}
}
}
return ret
}
type checkpointEntry struct {
PodUID string
ContainerName string
ResourceName string
DeviceIDs []string
AllocResp []byte
}
// checkpointData struct is used to store pod to device allocation information
// in a checkpoint file.
// TODO: add version control when we need to change checkpoint format.
type checkpointData struct {
Entries []checkpointEntry
}
// Turns podDevices to checkpointData.
func (pdev podDevices) toCheckpointData() checkpointData {
var data checkpointData
for podUID, containerDevices := range pdev {
for conName, resources := range containerDevices {
for resource, devices := range resources {
devIds := devices.deviceIds.UnsortedList()
allocResp, err := devices.allocResp.Marshal()
if err != nil {
glog.Errorf("Can't marshal allocResp for %v %v %v: %v", podUID, conName, resource, err)
continue
}
data.Entries = append(data.Entries, checkpointEntry{podUID, conName, resource, devIds, allocResp})
}
}
}
return data
}
// Populates podDevices from the passed in checkpointData.
func (pdev podDevices) fromCheckpointData(data checkpointData) {
for _, entry := range data.Entries {
glog.V(2).Infof("Get checkpoint entry: %v %v %v %v %v\n",
entry.PodUID, entry.ContainerName, entry.ResourceName, entry.DeviceIDs, entry.AllocResp)
devIDs := sets.NewString()
for _, devID := range entry.DeviceIDs {
devIDs.Insert(devID)
}
allocResp := &pluginapi.AllocateResponse{}
err := allocResp.Unmarshal(entry.AllocResp)
if err != nil {
glog.Errorf("Can't unmarshal allocResp for %v %v %v: %v", entry.PodUID, entry.ContainerName, entry.ResourceName, err)
continue
}
pdev.insert(entry.PodUID, entry.ContainerName, entry.ResourceName, devIDs, allocResp)
}
}
// Returns combined container runtime settings to consume the container's allocated devices.
func (pdev podDevices) deviceRunContainerOptions(podUID, contName string) *DeviceRunContainerOptions {
containers, exists := pdev[podUID]
if !exists {
return nil
}
resources, exists := containers[contName]
if !exists {
return nil
}
opts := &DeviceRunContainerOptions{}
// Maps to detect duplicate settings.
devsMap := make(map[string]string)
mountsMap := make(map[string]string)
envsMap := make(map[string]string)
// Loops through AllocationResponses of all cached device resources.
for _, devices := range resources {
resp := devices.allocResp
// 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 env %s %s", k, v)
if e != v {
glog.Errorf("Environment variable %s has conflicting setting: %s and %s", k, e, v)
}
continue
}
glog.V(4).Infof("add env %s %s", k, v)
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
}
glog.V(4).Infof("add device %s %s", dev.ContainerPath, dev.HostPath)
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
}
glog.V(4).Infof("add mount %s %s", mount.ContainerPath, mount.HostPath)
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,
})
}
}
return opts
}
......@@ -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