Skip to content

K
k3s
Commit 1c86ee30 authored by Darren Shepherd's avatar Darren Shepherd
Browse files
Options

Delete deviceplugin, checkpoint, devicemanager, custommetrics, gpu, mount pod

parent 1cc99801
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 8 additions and 3028 deletions
pkg/kubelet/apis/deviceplugin/OWNERS deleted 100644 → 0
View file @ 1cc99801
reviewers:
- jiayingz
- mindprince
- RenaudWasTaken
- vishh
pkg/kubelet/apis/deviceplugin/v1alpha/BUILD deleted 100644 → 0
View file @ 1cc99801
package(default_visibility = ["//visibility:public"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
)
go_library(
name = "go_default_library",
srcs = [
"api.pb.go",
"constants.go",
],
importpath = "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1alpha",
deps = [
"//vendor/github.com/gogo/protobuf/gogoproto:go_default_library",
"//vendor/github.com/gogo/protobuf/proto:go_default_library",
"//vendor/github.com/gogo/protobuf/sortkeys:go_default_library",
"//vendor/golang.org/x/net/context:go_default_library",
"//vendor/google.golang.org/grpc:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
filegroup(
name = "go_default_library_protos",
srcs = ["api.proto"],
visibility = ["//visibility:public"],
)
pkg/kubelet/apis/deviceplugin/v1alpha/api.proto deleted 100644 → 0
View file @ 1cc99801
// To regenerate api.pb.go run hack/update-device-plugin.sh
syntax = 'proto3';
package deviceplugin;
import "github.com/gogo/protobuf/gogoproto/gogo.proto";
option (gogoproto.goproto_stringer_all) = false;
option (gogoproto.stringer_all) = true;
option (gogoproto.goproto_getters_all) = true;
option (gogoproto.marshaler_all) = true;
option (gogoproto.sizer_all) = true;
option (gogoproto.unmarshaler_all) = true;
option (gogoproto.goproto_unrecognized_all) = false;
// Registration is the service advertised by the Kubelet
// Only when Kubelet answers with a success code to a Register Request
// may Device Plugins start their service
// Registration may fail when device plugin version is not supported by
// Kubelet or the registered resourceName is already taken by another
// active device plugin. Device plugin is expected to terminate upon registration failure
service Registration {
rpc Register(RegisterRequest) returns (Empty) {}
}
message RegisterRequest {
// Version of the API the Device Plugin was built against
string version = 1;
// Name of the unix socket the device plugin is listening on
// PATH = path.Join(DevicePluginPath, endpoint)
string endpoint = 2;
// Schedulable resource name. As of now it's expected to be a DNS Label
string resource_name = 3;
}
message Empty {
}
// DevicePlugin is the service advertised by Device Plugins
service DevicePlugin {
// ListAndWatch returns a stream of List of Devices
// Whenever a Device state changes or a Device disappears, ListAndWatch
// returns the new list
rpc ListAndWatch(Empty) returns (stream ListAndWatchResponse) {}
// Allocate is called during container creation so that the Device
// Plugin can run device specific operations and instruct Kubelet
// of the steps to make the Device available in the container
rpc Allocate(AllocateRequest) returns (AllocateResponse) {}
}
// ListAndWatch returns a stream of List of Devices
// Whenever a Device state changes or a Device disappears, ListAndWatch
// returns the new list
message ListAndWatchResponse {
repeated Device devices = 1;
}
/* E.g:
* struct Device {
* ID: "GPU-fef8089b-4820-abfc-e83e-94318197576e",
* State: "Healthy",
*} */
message Device {
// A unique ID assigned by the device plugin used
// to identify devices during the communication
// Max length of this field is 63 characters
string ID = 1;
// Health of the device, can be healthy or unhealthy, see constants.go
string health = 2;
}
// - Allocate is expected to be called during pod creation since allocation
// failures for any container would result in pod startup failure.
// - Allocate allows kubelet to exposes additional artifacts in a pod's
// environment as directed by the plugin.
// - Allocate allows Device Plugin to run device specific operations on
// the Devices requested
message AllocateRequest {
repeated string devicesIDs = 1;
}
// AllocateResponse includes the artifacts that needs to be injected into
// a container for accessing 'deviceIDs' that were mentioned as part of
// 'AllocateRequest'.
// Failure Handling:
// if Kubelet sends an allocation request for dev1 and dev2.
// Allocation on dev1 succeeds but allocation on dev2 fails.
// The Device plugin should send a ListAndWatch update and fail the
// Allocation request
message AllocateResponse {
// List of environment variable to be set in the container to access one of more devices.
map<string, string> envs = 1;
// Mounts for the container.
repeated Mount mounts = 2;
// Devices for the container.
repeated DeviceSpec devices = 3;
// Container annotations to pass to the container runtime
map<string, string> annotations = 4;
}
// Mount specifies a host volume to mount into a container.
// where device library or tools are installed on host and container
message Mount {
// Path of the mount within the container.
string container_path = 1;
// Path of the mount on the host.
string host_path = 2;
// If set, the mount is read-only.
bool read_only = 3;
}
// DeviceSpec specifies a host device to mount into a container.
message DeviceSpec {
// Path of the device within the container.
string container_path = 1;
// Path of the device on the host.
string host_path = 2;
// Cgroups permissions of the device, candidates are one or more of
// * r - allows container to read from the specified device.
// * w - allows container to write to the specified device.
// * m - allows container to create device files that do not yet exist.
string permissions = 3;
}
pkg/kubelet/apis/deviceplugin/v1alpha/constants.go deleted 100644 → 0
View file @ 1cc99801
/*
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
const (
// Healthy means that the device is healthy
Healthy = "Healthy"
// Unhealthy means that the device is unhealthy
Unhealthy = "Unhealthy"
// Current version of the API supported by kubelet
Version = "v1alpha2"
// DevicePluginPath is the folder the Device Plugin is expecting sockets to be on
// Only privileged pods have access to this path
// Note: Placeholder until we find a "standard path"
DevicePluginPath = "/var/lib/kubelet/device-plugins/"
// KubeletSocket is the path of the Kubelet registry socket
KubeletSocket = DevicePluginPath + "kubelet.sock"
)
pkg/kubelet/apis/deviceplugin/v1beta1/BUILD deleted 100644 → 0
View file @ 1cc99801
package(default_visibility = ["//visibility:public"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
)
go_library(
name = "go_default_library",
srcs = [
"api.pb.go",
"constants.go",
],
importpath = "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1",
deps = [
"//vendor/github.com/gogo/protobuf/gogoproto:go_default_library",
"//vendor/github.com/gogo/protobuf/proto:go_default_library",
"//vendor/github.com/gogo/protobuf/sortkeys:go_default_library",
"//vendor/golang.org/x/net/context:go_default_library",
"//vendor/google.golang.org/grpc:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
filegroup(
name = "go_default_library_protos",
srcs = ["api.proto"],
)
pkg/kubelet/apis/deviceplugin/v1beta1/api.proto deleted 100644 → 0
View file @ 1cc99801
// To regenerate api.pb.go run hack/update-device-plugin.sh
syntax = 'proto3';
package v1beta1;
import "github.com/gogo/protobuf/gogoproto/gogo.proto";
option (gogoproto.goproto_stringer_all) = false;
option (gogoproto.stringer_all) = true;
option (gogoproto.goproto_getters_all) = true;
option (gogoproto.marshaler_all) = true;
option (gogoproto.sizer_all) = true;
option (gogoproto.unmarshaler_all) = true;
option (gogoproto.goproto_unrecognized_all) = false;
// Registration is the service advertised by the Kubelet
// Only when Kubelet answers with a success code to a Register Request
// may Device Plugins start their service
// Registration may fail when device plugin version is not supported by
// Kubelet or the registered resourceName is already taken by another
// active device plugin. Device plugin is expected to terminate upon registration failure
service Registration {
rpc Register(RegisterRequest) returns (Empty) {}
}
message DevicePluginOptions {
// Indicates if PreStartContainer call is required before each container start
bool pre_start_required = 1;
}
message RegisterRequest {
// Version of the API the Device Plugin was built against
string version = 1;
// Name of the unix socket the device plugin is listening on
// PATH = path.Join(DevicePluginPath, endpoint)
string endpoint = 2;
// Schedulable resource name. As of now it's expected to be a DNS Label
string resource_name = 3;
// Options to be communicated with Device Manager
DevicePluginOptions options = 4;
}
message Empty {
}
// DevicePlugin is the service advertised by Device Plugins
service DevicePlugin {
// GetDevicePluginOptions returns options to be communicated with Device
// Manager
rpc GetDevicePluginOptions(Empty) returns (DevicePluginOptions) {}
// ListAndWatch returns a stream of List of Devices
// Whenever a Device state change or a Device disapears, ListAndWatch
// returns the new list
rpc ListAndWatch(Empty) returns (stream ListAndWatchResponse) {}
// Allocate is called during container creation so that the Device
// Plugin can run device specific operations and instruct Kubelet
// of the steps to make the Device available in the container
rpc Allocate(AllocateRequest) returns (AllocateResponse) {}
// PreStartContainer is called, if indicated by Device Plugin during registeration phase,
// before each container start. Device plugin can run device specific operations
// such as reseting the device before making devices available to the container
rpc PreStartContainer(PreStartContainerRequest) returns (PreStartContainerResponse) {}
}
// ListAndWatch returns a stream of List of Devices
// Whenever a Device state change or a Device disapears, ListAndWatch
// returns the new list
message ListAndWatchResponse {
repeated Device devices = 1;
}
/* E.g:
* struct Device {
* ID: "GPU-fef8089b-4820-abfc-e83e-94318197576e",
* State: "Healthy",
*} */
message Device {
// A unique ID assigned by the device plugin used
// to identify devices during the communication
// Max length of this field is 63 characters
string ID = 1;
// Health of the device, can be healthy or unhealthy, see constants.go
string health = 2;
}
// - PreStartContainer is expected to be called before each container start if indicated by plugin during registration phase.
// - PreStartContainer allows kubelet to pass reinitialized devices to containers.
// - PreStartContainer allows Device Plugin to run device specific operations on
// the Devices requested
message PreStartContainerRequest {
repeated string devicesIDs = 1;
}
// PreStartContainerResponse will be send by plugin in response to PreStartContainerRequest
message PreStartContainerResponse {
}
// - Allocate is expected to be called during pod creation since allocation
// failures for any container would result in pod startup failure.
// - Allocate allows kubelet to exposes additional artifacts in a pod's
// environment as directed by the plugin.
// - Allocate allows Device Plugin to run device specific operations on
// the Devices requested
message AllocateRequest {
repeated ContainerAllocateRequest container_requests = 1;
}
message ContainerAllocateRequest {
repeated string devicesIDs = 1;
}
// AllocateResponse includes the artifacts that needs to be injected into
// a container for accessing 'deviceIDs' that were mentioned as part of
// 'AllocateRequest'.
// Failure Handling:
// if Kubelet sends an allocation request for dev1 and dev2.
// Allocation on dev1 succeeds but allocation on dev2 fails.
// The Device plugin should send a ListAndWatch update and fail the
// Allocation request
message AllocateResponse {
repeated ContainerAllocateResponse container_responses = 1;
}
message ContainerAllocateResponse {
// List of environment variable to be set in the container to access one of more devices.
map<string, string> envs = 1;
// Mounts for the container.
repeated Mount mounts = 2;
// Devices for the container.
repeated DeviceSpec devices = 3;
// Container annotations to pass to the container runtime
map<string, string> annotations = 4;
}
// Mount specifies a host volume to mount into a container.
// where device library or tools are installed on host and container
message Mount {
// Path of the mount within the container.
string container_path = 1;
// Path of the mount on the host.
string host_path = 2;
// If set, the mount is read-only.
bool read_only = 3;
}
// DeviceSpec specifies a host device to mount into a container.
message DeviceSpec {
// Path of the device within the container.
string container_path = 1;
// Path of the device on the host.
string host_path = 2;
// Cgroups permissions of the device, candidates are one or more of
// * r - allows container to read from the specified device.
// * w - allows container to write to the specified device.
// * m - allows container to create device files that do not yet exist.
string permissions = 3;
}
pkg/kubelet/apis/deviceplugin/v1beta1/constants.go deleted 100644 → 0
View file @ 1cc99801
/*
Copyright 2018 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 v1beta1
const (
// Healthy means that the device is healty
Healthy = "Healthy"
// UnHealthy means that the device is unhealthy
Unhealthy = "Unhealthy"
// Current version of the API supported by kubelet
Version = "v1beta1"
// DevicePluginPath is the folder the Device Plugin is expecting sockets to be on
// Only privileged pods have access to this path
// Note: Placeholder until we find a "standard path"
DevicePluginPath = "/var/lib/kubelet/device-plugins/"
// KubeletSocket is the path of the Kubelet registry socket
KubeletSocket = DevicePluginPath + "kubelet.sock"
// Timeout duration in secs for PreStartContainer RPC
KubeletPreStartContainerRPCTimeoutInSecs = 30
)
var SupportedVersions = [...]string{"v1beta1"}
pkg/kubelet/checkpoint/BUILD deleted 100644 → 0
View file @ 1cc99801
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = ["checkpoint.go"],
importpath = "k8s.io/kubernetes/pkg/kubelet/checkpoint",
visibility = ["//visibility:public"],
deps = [
"//pkg/apis/core:go_default_library",
"//pkg/volume/util:go_default_library",
"//vendor/github.com/dchest/safefile:go_default_library",
"//vendor/github.com/ghodss/yaml:go_default_library",
"//vendor/github.com/golang/glog:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
],
)
go_test(
name = "go_default_test",
srcs = ["checkpoint_test.go"],
embed = [":go_default_library"],
deps = [
"//pkg/apis/core:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
visibility = ["//visibility:public"],
)
pkg/kubelet/checkpoint/checkpoint.go deleted 100644 → 0
View file @ 1cc99801
/*
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 checkpoint
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"strings"
"sync"
"github.com/dchest/safefile"
"github.com/ghodss/yaml"
"github.com/golang/glog"
"k8s.io/api/core/v1"
"k8s.io/kubernetes/pkg/apis/core"
"k8s.io/kubernetes/pkg/volume/util"
)
const (
// Delimiter used on checkpoints written to disk
delimiter = "_"
podPrefix = "Pod"
)
// Manager is the interface used to manage checkpoints
// which involves writing resources to disk to recover
// during restart or failure scenarios.
// https://github.com/kubernetes/community/pull/1241/files
type Manager interface {
// LoadPods will load checkpointed Pods from disk
LoadPods() ([]*v1.Pod, error)
// WritePod will serialize a Pod to disk
WritePod(pod *v1.Pod) error
// Deletes the checkpoint of the given pod from disk
DeletePod(pod *v1.Pod) error
}
var instance Manager
var mutex = &sync.Mutex{}
// fileCheckPointManager - is a checkpointer that writes contents to disk
// The type information of the resource objects are encoded in the name
type fileCheckPointManager struct {
path string
}
// NewCheckpointManager will create a Manager that points to the following path
func NewCheckpointManager(path string) Manager {
// NOTE: This is a precaution; current implementation should not run
// multiple checkpoint managers.
mutex.Lock()
defer mutex.Unlock()
instance = &fileCheckPointManager{path: path}
return instance
}
// GetInstance will return the current Manager, there should be only one.
func GetInstance() Manager {
mutex.Lock()
defer mutex.Unlock()
return instance
}
// loadPod will load Pod Checkpoint yaml file.
func (fcp *fileCheckPointManager) loadPod(file string) (*v1.Pod, error) {
return util.LoadPodFromFile(file)
}
// checkAnnotations will validate the checkpoint annotations exist on the Pod
func (fcp *fileCheckPointManager) checkAnnotations(pod *v1.Pod) bool {
if podAnnotations := pod.GetAnnotations(); podAnnotations != nil {
if podAnnotations[core.BootstrapCheckpointAnnotationKey] == "true" {
return true
}
}
return false
}
// getPodPath returns the full qualified path for the pod checkpoint
func (fcp *fileCheckPointManager) getPodPath(pod *v1.Pod) string {
return fmt.Sprintf("%v/Pod%v%v.yaml", fcp.path, delimiter, pod.GetUID())
}
// LoadPods Loads All Checkpoints from disk
func (fcp *fileCheckPointManager) LoadPods() ([]*v1.Pod, error) {
checkpoints := make([]*v1.Pod, 0)
files, err := ioutil.ReadDir(fcp.path)
if err != nil {
return nil, err
}
for _, f := range files {
// get just the filename
_, fname := filepath.Split(f.Name())
// Get just the Resource from "Resource_Name"
fnfields := strings.Split(fname, delimiter)
switch fnfields[0] {
case podPrefix:
pod, err := fcp.loadPod(fmt.Sprintf("%s/%s", fcp.path, f.Name()))
if err != nil {
return nil, err
}
checkpoints = append(checkpoints, pod)
default:
glog.Warningf("Unsupported checkpoint file detected %v", f)
}
}
return checkpoints, nil
}
// Writes a checkpoint to a file on disk if annotation is present
func (fcp *fileCheckPointManager) WritePod(pod *v1.Pod) error {
var err error
if fcp.checkAnnotations(pod) {
if blob, err := yaml.Marshal(pod); err == nil {
err = safefile.WriteFile(fcp.getPodPath(pod), blob, 0644)
}
} else {
// This is to handle an edge where a pod update could remove
// an annotation and the checkpoint should then be removed.
err = fcp.DeletePod(pod)
}
return err
}
// Deletes a checkpoint from disk if present
func (fcp *fileCheckPointManager) DeletePod(pod *v1.Pod) error {
podPath := fcp.getPodPath(pod)
if err := os.Remove(podPath); !os.IsNotExist(err) {
return err
}
return nil
}
pkg/kubelet/checkpoint/checkpoint_test.go deleted 100644 → 0
View file @ 1cc99801
/*
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 checkpoint
import (
"io/ioutil"
"os"
"reflect"
"testing"
"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/kubernetes/pkg/apis/core"
)
// TestWriteLoadDeletePods validates all combinations of write, load, and delete
func TestWriteLoadDeletePods(t *testing.T) {
testPods := []struct {
pod *v1.Pod
written bool
}{
{
pod: &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "Foo",
Annotations: map[string]string{core.BootstrapCheckpointAnnotationKey: "true"},
UID: "1",
},
},
written: true,
},
{
pod: &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "Foo2",
Annotations: map[string]string{core.BootstrapCheckpointAnnotationKey: "true"},
UID: "2",
},
},
written: true,
},
{
pod: &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: "Bar",
UID: "3",
},
},
written: false,
},
}
dir, err := ioutil.TempDir("", "checkpoint")
if err != nil {
t.Errorf("Failed to allocate temp directory for TestWriteLoadDeletePods error=%v", err)
}
defer os.RemoveAll(dir)
cp := NewCheckpointManager(dir)
for _, p := range testPods {
// Write pods should always pass unless there is an fs error
if err := cp.WritePod(p.pod); err != nil {
t.Errorf("Failed to Write Pod: %v", err)
}
}
// verify the correct written files are loaded from disk
pods, err := cp.LoadPods()
if err != nil {
t.Errorf("Failed to Load Pods: %v", err)
}
// loop through contents and check make sure
// what was loaded matched the expected results.
for _, p := range testPods {
pname := p.pod.GetName()
var lpod *v1.Pod
for _, check := range pods {
if check.GetName() == pname {
lpod = check
break
}
}
if p.written {
if lpod != nil {
if !reflect.DeepEqual(p.pod, lpod) {
t.Errorf("expected %#v, \ngot %#v", p.pod, lpod)
}
} else {
t.Errorf("Got unexpected result for %v, should have been loaded", pname)
}
} else if lpod != nil {
t.Errorf("Got unexpected result for %v, should not have been loaded", pname)
}
err = cp.DeletePod(p.pod)
if err != nil {
t.Errorf("Failed to delete pod %v", pname)
}
}
// finally validate the contents of the directory is empty.
files, err := ioutil.ReadDir(dir)
if err != nil {
t.Errorf("Failed to read directory %v", dir)
}
if len(files) > 0 {
t.Errorf("Directory %v should be empty but found %#v", dir, files)
}
}
pkg/kubelet/cm/container_manager_linux.go
View file @ 1c86ee30
...@@ -45,7 +45,6 @@ import ( ...@@ -45,7 +45,6 @@ import (
internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri" internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
"k8s.io/kubernetes/pkg/kubelet/cadvisor" "k8s.io/kubernetes/pkg/kubelet/cadvisor"
"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager" "k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
"k8s.io/kubernetes/pkg/kubelet/cm/devicemanager"
cmutil "k8s.io/kubernetes/pkg/kubelet/cm/util" cmutil "k8s.io/kubernetes/pkg/kubelet/cm/util"
"k8s.io/kubernetes/pkg/kubelet/config" "k8s.io/kubernetes/pkg/kubelet/config"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
...@@ -128,8 +127,6 @@ type containerManagerImpl struct { ...@@ -128,8 +127,6 @@ type containerManagerImpl struct {
recorder record.EventRecorder recorder record.EventRecorder
// Interface for QoS cgroup management // Interface for QoS cgroup management
qosContainerManager QOSContainerManager qosContainerManager QOSContainerManager
// Interface for exporting and allocating devices reported by device plugins.
deviceManager devicemanager.Manager
// Interface for CPU affinity management. // Interface for CPU affinity management.
cpuManager cpumanager.Manager cpuManager cpumanager.Manager
} }
...@@ -263,16 +260,6 @@ func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.I ...@@ -263,16 +260,6 @@ func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.I
qosContainerManager: qosContainerManager, qosContainerManager: qosContainerManager,
} }
glog.Infof("Creating device plugin manager: %t", devicePluginEnabled)
if devicePluginEnabled {
cm.deviceManager, err = devicemanager.NewManagerImpl()
} else {
cm.deviceManager, err = devicemanager.NewManagerStub()
}
if err != nil {
return nil, err
}
// Initialize CPU manager // Initialize CPU manager
if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.CPUManager) { if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.CPUManager) {
cm.cpuManager, err = cpumanager.NewManager( cm.cpuManager, err = cpumanager.NewManager(
...@@ -589,34 +576,17 @@ func (cm *containerManagerImpl) Start(node *v1.Node, ...@@ -589,34 +576,17 @@ func (cm *containerManagerImpl) Start(node *v1.Node,
}, 5*time.Minute, wait.NeverStop) }, 5*time.Minute, wait.NeverStop)
} }
// Starts device manager.
if err := cm.deviceManager.Start(devicemanager.ActivePodsFunc(activePods), sourcesReady); err != nil {
return err
}
return nil return nil
} }
// TODO: move the GetResources logic to PodContainerManager. // TODO: move the GetResources logic to PodContainerManager.
func (cm *containerManagerImpl) GetResources(pod *v1.Pod, container *v1.Container) (*kubecontainer.RunContainerOptions, error) { func (cm *containerManagerImpl) GetResources(pod *v1.Pod, container *v1.Container) (*kubecontainer.RunContainerOptions, error) {
opts := &kubecontainer.RunContainerOptions{} 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)
if err != nil {
return nil, err
} else 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...)
opts.Annotations = append(opts.Annotations, devOpts.Annotations...)
return opts, nil return opts, nil
} }
func (cm *containerManagerImpl) UpdatePluginResources(node *schedulercache.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error { func (cm *containerManagerImpl) UpdatePluginResources(node *schedulercache.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error {
return cm.deviceManager.Allocate(node, attrs) return nil
} }
func (cm *containerManagerImpl) SystemCgroupsLimit() v1.ResourceList { func (cm *containerManagerImpl) SystemCgroupsLimit() v1.ResourceList {
...@@ -879,5 +849,5 @@ func (cm *containerManagerImpl) GetCapacity() v1.ResourceList { ...@@ -879,5 +849,5 @@ func (cm *containerManagerImpl) GetCapacity() v1.ResourceList {
} }
func (cm *containerManagerImpl) GetDevicePluginResourceCapacity() (v1.ResourceList, v1.ResourceList, []string) { func (cm *containerManagerImpl) GetDevicePluginResourceCapacity() (v1.ResourceList, v1.ResourceList, []string) {
return cm.deviceManager.GetCapacity() return nil, nil, nil
} }
pkg/kubelet/cm/devicemanager/BUILD deleted 100644 → 0
View file @ 1cc99801
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = [
"device_plugin_stub.go",
"endpoint.go",
"manager.go",
"manager_stub.go",
"pod_devices.go",
"types.go",
],
importpath = "k8s.io/kubernetes/pkg/kubelet/cm/devicemanager",
visibility = ["//visibility:public"],
deps = [
"//pkg/apis/core/v1/helper:go_default_library",
"//pkg/kubelet/apis/deviceplugin/v1beta1:go_default_library",
"//pkg/kubelet/config:go_default_library",
"//pkg/kubelet/container:go_default_library",
"//pkg/kubelet/lifecycle:go_default_library",
"//pkg/kubelet/metrics:go_default_library",
"//pkg/kubelet/util/store:go_default_library",
"//pkg/scheduler/schedulercache:go_default_library",
"//pkg/util/filesystem: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",
],
)
go_test(
name = "go_default_test",
srcs = [
"endpoint_test.go",
"manager_test.go",
],
embed = [":go_default_library"],
deps = [
"//pkg/kubelet/apis/deviceplugin/v1beta1:go_default_library",
"//pkg/kubelet/lifecycle:go_default_library",
"//pkg/kubelet/util/store:go_default_library",
"//pkg/scheduler/schedulercache:go_default_library",
"//pkg/util/filesystem: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",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
visibility = ["//visibility:public"],
)
pkg/kubelet/cm/devicemanager/OWNERS deleted 100644 → 0
View file @ 1cc99801
approvers:
- jiayingz
- vishh
reviewers:
- mindprince
- RenaudWasTaken
- vikaschoudhary16
pkg/kubelet/cm/devicemanager/device_plugin_stub.go deleted 100644 → 0
View file @ 1cc99801
/*
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 devicemanager
import (
"log"
"net"
"os"
"path"
"time"
"golang.org/x/net/context"
"google.golang.org/grpc"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
)
// Stub implementation for DevicePlugin.
type Stub struct {
devs []*pluginapi.Device
socket string
stop chan interface{}
update chan []*pluginapi.Device
server *grpc.Server
// allocFunc is used for handling allocation request
allocFunc stubAllocFunc
}
// stubAllocFunc is the function called when receive an allocation request from Kubelet
type stubAllocFunc func(r *pluginapi.AllocateRequest, devs map[string]pluginapi.Device) (*pluginapi.AllocateResponse, error)
func defaultAllocFunc(r *pluginapi.AllocateRequest, devs map[string]pluginapi.Device) (*pluginapi.AllocateResponse, error) {
var response pluginapi.AllocateResponse
return &response, nil
}
// NewDevicePluginStub returns an initialized DevicePlugin Stub.
func NewDevicePluginStub(devs []*pluginapi.Device, socket string) *Stub {
return &Stub{
devs: devs,
socket: socket,
stop: make(chan interface{}),
update: make(chan []*pluginapi.Device),
allocFunc: defaultAllocFunc,
}
}
// SetAllocFunc sets allocFunc of the device plugin
func (m *Stub) SetAllocFunc(f stubAllocFunc) {
m.allocFunc = f
}
// Start starts the gRPC server of the device plugin
func (m *Stub) Start() error {
err := m.cleanup()
if err != nil {
return err
}
sock, err := net.Listen("unix", m.socket)
if err != nil {
return err
}
m.server = grpc.NewServer([]grpc.ServerOption{}...)
pluginapi.RegisterDevicePluginServer(m.server, m)
go m.server.Serve(sock)
_, conn, err := dial(m.socket)
if err != nil {
return err
}
conn.Close()
log.Println("Starting to serve on", m.socket)
return nil
}
// Stop stops the gRPC server
func (m *Stub) Stop() error {
m.server.Stop()
close(m.stop)
return m.cleanup()
}
// Register registers the device plugin for the given resourceName with Kubelet.
func (m *Stub) Register(kubeletEndpoint, resourceName string, preStartContainerFlag bool) error {
conn, err := grpc.Dial(kubeletEndpoint, grpc.WithInsecure(), grpc.WithBlock(),
grpc.WithTimeout(10*time.Second),
grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
return net.DialTimeout("unix", addr, timeout)
}))
defer conn.Close()
if err != nil {
return err
}
client := pluginapi.NewRegistrationClient(conn)
reqt := &pluginapi.RegisterRequest{
Version: pluginapi.Version,
Endpoint: path.Base(m.socket),
ResourceName: resourceName,
Options: &pluginapi.DevicePluginOptions{PreStartRequired: preStartContainerFlag},
}
_, err = client.Register(context.Background(), reqt)
if err != nil {
return err
}
return nil
}
// GetDevicePluginOptions returns DevicePluginOptions settings for the device plugin.
func (m *Stub) GetDevicePluginOptions(ctx context.Context, e *pluginapi.Empty) (*pluginapi.DevicePluginOptions, error) {
return &pluginapi.DevicePluginOptions{}, nil
}
// PreStartContainer resets the devices received
func (m *Stub) PreStartContainer(ctx context.Context, r *pluginapi.PreStartContainerRequest) (*pluginapi.PreStartContainerResponse, error) {
log.Printf("PreStartContainer, %+v", 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 {
log.Println("ListAndWatch")
s.Send(&pluginapi.ListAndWatchResponse{Devices: m.devs})
for {
select {
case <-m.stop:
return nil
case updated := <-m.update:
s.Send(&pluginapi.ListAndWatchResponse{Devices: updated})
}
}
}
// Update allows the device plugin to send new devices through ListAndWatch
func (m *Stub) Update(devs []*pluginapi.Device) {
m.update <- devs
}
// Allocate does a mock allocation
func (m *Stub) Allocate(ctx context.Context, r *pluginapi.AllocateRequest) (*pluginapi.AllocateResponse, error) {
log.Printf("Allocate, %+v", r)
devs := make(map[string]pluginapi.Device)
for _, dev := range m.devs {
devs[dev.ID] = *dev
}
return m.allocFunc(r, devs)
}
func (m *Stub) cleanup() error {
if err := os.Remove(m.socket); err != nil && !os.IsNotExist(err) {
return err
}
return nil
}
pkg/kubelet/cm/devicemanager/endpoint.go deleted 100644 → 0
View file @ 1cc99801
/*
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 devicemanager
import (
"fmt"
"net"
"sync"
"time"
"github.com/golang/glog"
"golang.org/x/net/context"
"google.golang.org/grpc"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
)
// endpoint maps to a single registered device plugin. It is responsible
// for managing gRPC communications with the device plugin and caching
// device states reported by the device plugin.
type endpoint interface {
run()
stop()
allocate(devs []string) (*pluginapi.AllocateResponse, error)
preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error)
getDevices() []pluginapi.Device
callback(resourceName string, added, updated, deleted []pluginapi.Device)
isStopped() bool
stopGracePeriodExpired() bool
}
type endpointImpl struct {
client pluginapi.DevicePluginClient
clientConn *grpc.ClientConn
socketPath string
resourceName string
stopTime time.Time
devices map[string]pluginapi.Device
mutex sync.Mutex
cb monitorCallback
}
// newEndpoint creates a new endpoint for the given resourceName.
// This is to be used during normal device plugin registration.
func newEndpointImpl(socketPath, resourceName string, devices map[string]pluginapi.Device, callback monitorCallback) (*endpointImpl, error) {
client, c, err := dial(socketPath)
if err != nil {
glog.Errorf("Can't create new endpoint with path %s err %v", socketPath, err)
return nil, err
}
return &endpointImpl{
client: client,
clientConn: c,
socketPath: socketPath,
resourceName: resourceName,
devices: devices,
cb: callback,
}, nil
}
// newStoppedEndpointImpl creates a new endpoint for the given resourceName with stopTime set.
// This is to be used during Kubelet restart, before the actual device plugin re-registers.
func newStoppedEndpointImpl(resourceName string, devices map[string]pluginapi.Device) *endpointImpl {
return &endpointImpl{
resourceName: resourceName,
devices: devices,
stopTime: time.Now(),
}
}
func (e *endpointImpl) callback(resourceName string, added, updated, deleted []pluginapi.Device) {
e.cb(resourceName, added, updated, deleted)
}
func (e *endpointImpl) getDevices() []pluginapi.Device {
e.mutex.Lock()
defer e.mutex.Unlock()
var devs []pluginapi.Device
for _, d := range e.devices {
devs = append(devs, d)
}
return devs
}
// run initializes ListAndWatch gRPC call for the device plugin and
// blocks on receiving ListAndWatch gRPC stream updates. Each ListAndWatch
// stream update contains a new list of device states. listAndWatch compares the new
// device states with its cached states to get list of new, updated, and deleted devices.
// It then issues a callback to pass this information to the device manager which
// will adjust the resource available information accordingly.
func (e *endpointImpl) run() {
stream, err := e.client.ListAndWatch(context.Background(), &pluginapi.Empty{})
if err != nil {
glog.Errorf(errListAndWatch, e.resourceName, err)
return
}
devices := make(map[string]pluginapi.Device)
e.mutex.Lock()
for _, d := range e.devices {
devices[d.ID] = d
}
e.mutex.Unlock()
for {
response, err := stream.Recv()
if err != nil {
glog.Errorf(errListAndWatch, e.resourceName, err)
return
}
devs := response.Devices
glog.V(2).Infof("State pushed for device plugin %s", e.resourceName)
newDevs := make(map[string]*pluginapi.Device)
var added, updated []pluginapi.Device
for _, d := range devs {
dOld, ok := devices[d.ID]
newDevs[d.ID] = d
if !ok {
glog.V(2).Infof("New device for Endpoint %s: %v", e.resourceName, d)
devices[d.ID] = *d
added = append(added, *d)
continue
}
if d.Health == dOld.Health {
continue
}
if d.Health == pluginapi.Unhealthy {
glog.Errorf("Device %s is now Unhealthy", d.ID)
} else if d.Health == pluginapi.Healthy {
glog.V(2).Infof("Device %s is now Healthy", d.ID)
}
devices[d.ID] = *d
updated = append(updated, *d)
}
var deleted []pluginapi.Device
for id, d := range devices {
if _, ok := newDevs[id]; ok {
continue
}
glog.Errorf("Device %s was deleted", d.ID)
deleted = append(deleted, d)
delete(devices, id)
}
e.mutex.Lock()
// NOTE: Return a copy of 'devices' instead of returning a direct reference to local 'devices'
e.devices = make(map[string]pluginapi.Device)
for _, d := range devices {
e.devices[d.ID] = d
}
e.mutex.Unlock()
e.callback(e.resourceName, added, updated, deleted)
}
}
func (e *endpointImpl) isStopped() bool {
e.mutex.Lock()
defer e.mutex.Unlock()
return !e.stopTime.IsZero()
}
func (e *endpointImpl) stopGracePeriodExpired() bool {
e.mutex.Lock()
defer e.mutex.Unlock()
return !e.stopTime.IsZero() && time.Since(e.stopTime) > endpointStopGracePeriod
}
// used for testing only
func (e *endpointImpl) setStopTime(t time.Time) {
e.mutex.Lock()
defer e.mutex.Unlock()
e.stopTime = t
}
// allocate issues Allocate gRPC call to the device plugin.
func (e *endpointImpl) allocate(devs []string) (*pluginapi.AllocateResponse, error) {
if e.isStopped() {
return nil, fmt.Errorf(errEndpointStopped, e)
}
return e.client.Allocate(context.Background(), &pluginapi.AllocateRequest{
ContainerRequests: []*pluginapi.ContainerAllocateRequest{
{DevicesIDs: devs},
},
})
}
// preStartContainer issues PreStartContainer gRPC call to the device plugin.
func (e *endpointImpl) preStartContainer(devs []string) (*pluginapi.PreStartContainerResponse, error) {
if e.isStopped() {
return nil, fmt.Errorf(errEndpointStopped, e)
}
ctx, cancel := context.WithTimeout(context.Background(), pluginapi.KubeletPreStartContainerRPCTimeoutInSecs*time.Second)
defer cancel()
return e.client.PreStartContainer(ctx, &pluginapi.PreStartContainerRequest{
DevicesIDs: devs,
})
}
func (e *endpointImpl) stop() {
e.mutex.Lock()
defer e.mutex.Unlock()
if e.clientConn != nil {
e.clientConn.Close()
}
e.stopTime = time.Now()
}
// dial establishes the gRPC communication with the registered device plugin. https://godoc.org/google.golang.org/grpc#Dial
func dial(unixSocketPath string) (pluginapi.DevicePluginClient, *grpc.ClientConn, error) {
c, err := grpc.Dial(unixSocketPath, grpc.WithInsecure(), grpc.WithBlock(),
grpc.WithTimeout(10*time.Second),
grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
return net.DialTimeout("unix", addr, timeout)
}),
)
if err != nil {
return nil, nil, fmt.Errorf(errFailedToDialDevicePlugin+" %v", err)
}
return pluginapi.NewDevicePluginClient(c), c, nil
}
pkg/kubelet/cm/devicemanager/endpoint_test.go deleted 100644 → 0
View file @ 1cc99801
/*
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 devicemanager
import (
"path"
"testing"
"time"
"github.com/stretchr/testify/require"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
)
var (
esocketName = "mock.sock"
)
func TestNewEndpoint(t *testing.T) {
socket := path.Join("/tmp", esocketName)
devs := []*pluginapi.Device{
{ID: "ADeviceId", Health: pluginapi.Healthy},
}
p, e := esetup(t, devs, socket, "mock", func(n string, a, u, r []pluginapi.Device) {})
defer ecleanup(t, p, e)
}
func TestRun(t *testing.T) {
socket := path.Join("/tmp", esocketName)
devs := []*pluginapi.Device{
{ID: "ADeviceId", Health: pluginapi.Healthy},
{ID: "AnotherDeviceId", Health: pluginapi.Healthy},
{ID: "AThirdDeviceId", Health: pluginapi.Unhealthy},
}
updated := []*pluginapi.Device{
{ID: "ADeviceId", Health: pluginapi.Unhealthy},
{ID: "AThirdDeviceId", Health: pluginapi.Healthy},
{ID: "AFourthDeviceId", Health: pluginapi.Healthy},
}
callbackCount := 0
callbackChan := make(chan int)
callback := func(n string, a, u, r []pluginapi.Device) {
// Should be called twice:
// one for plugin registration, one for plugin update.
if callbackCount > 2 {
t.FailNow()
}
// Check plugin registration
if callbackCount == 0 {
require.Len(t, a, 3)
require.Len(t, u, 0)
require.Len(t, r, 0)
}
// Check plugin update
if callbackCount == 1 {
require.Len(t, a, 1)
require.Len(t, u, 2)
require.Len(t, r, 1)
require.Equal(t, a[0].ID, updated[2].ID)
require.Equal(t, u[0].ID, updated[0].ID)
require.Equal(t, u[0].Health, updated[0].Health)
require.Equal(t, u[1].ID, updated[1].ID)
require.Equal(t, u[1].Health, updated[1].Health)
require.Equal(t, r[0].ID, devs[1].ID)
}
callbackCount++
callbackChan <- callbackCount
}
p, e := esetup(t, devs, socket, "mock", callback)
defer ecleanup(t, p, e)
go e.run()
// Wait for the first callback to be issued.
<-callbackChan
p.Update(updated)
// Wait for the second callback to be issued.
<-callbackChan
e.mutex.Lock()
defer e.mutex.Unlock()
require.Len(t, e.devices, 3)
for _, dref := range updated {
d, ok := e.devices[dref.ID]
require.True(t, ok)
require.Equal(t, d.ID, dref.ID)
require.Equal(t, d.Health, dref.Health)
}
}
func TestAllocate(t *testing.T) {
socket := path.Join("/tmp", 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, a, u, r []pluginapi.Device) {
callbackCount++
callbackChan <- callbackCount
})
defer ecleanup(t, p, e)
resp := new(pluginapi.AllocateResponse)
contResp := new(pluginapi.ContainerAllocateResponse)
contResp.Devices = append(contResp.Devices, &pluginapi.DeviceSpec{
ContainerPath: "/dev/aaa",
HostPath: "/dev/aaa",
Permissions: "mrw",
})
contResp.Devices = append(contResp.Devices, &pluginapi.DeviceSpec{
ContainerPath: "/dev/bbb",
HostPath: "/dev/bbb",
Permissions: "mrw",
})
contResp.Mounts = append(contResp.Mounts, &pluginapi.Mount{
ContainerPath: "/container_dir1/file1",
HostPath: "host_dir1/file1",
ReadOnly: true,
})
resp.ContainerResponses = append(resp.ContainerResponses, contResp)
p.SetAllocFunc(func(r *pluginapi.AllocateRequest, devs map[string]pluginapi.Device) (*pluginapi.AllocateResponse, error) {
return resp, nil
})
go e.run()
// Wait for the callback to be issued.
select {
case <-callbackChan:
break
case <-time.After(time.Second):
t.FailNow()
}
respOut, err := e.allocate([]string{"ADeviceId"})
require.NoError(t, err)
require.Equal(t, resp, respOut)
}
func TestGetDevices(t *testing.T) {
e := endpointImpl{
devices: map[string]pluginapi.Device{
"ADeviceId": {ID: "ADeviceId", Health: pluginapi.Healthy},
},
}
devs := e.getDevices()
require.Len(t, devs, 1)
}
func esetup(t *testing.T, devs []*pluginapi.Device, socket, resourceName string, callback monitorCallback) (*Stub, *endpointImpl) {
p := NewDevicePluginStub(devs, socket)
err := p.Start()
require.NoError(t, err)
e, err := newEndpointImpl(socket, resourceName, make(map[string]pluginapi.Device), callback)
require.NoError(t, err)
return p, e
}
func ecleanup(t *testing.T, p *Stub, e *endpointImpl) {
p.Stop()
e.stop()
}
pkg/kubelet/cm/devicemanager/manager_stub.go deleted 100644 → 0
View file @ 1cc99801
/*
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 devicemanager
import (
"k8s.io/api/core/v1"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
"k8s.io/kubernetes/pkg/kubelet/config"
"k8s.io/kubernetes/pkg/kubelet/lifecycle"
"k8s.io/kubernetes/pkg/scheduler/schedulercache"
)
// ManagerStub provides a simple stub implementation for the Device Manager.
type ManagerStub struct{}
// NewManagerStub creates a ManagerStub.
func NewManagerStub() (*ManagerStub, error) {
return &ManagerStub{}, nil
}
// Start simply returns nil.
func (h *ManagerStub) Start(activePods ActivePodsFunc, sourcesReady config.SourcesReady) error {
return nil
}
// Stop simply returns nil.
func (h *ManagerStub) Stop() error {
return nil
}
// Devices returns an empty map.
func (h *ManagerStub) Devices() map[string][]pluginapi.Device {
return make(map[string][]pluginapi.Device)
}
// Allocate simply returns nil.
func (h *ManagerStub) Allocate(node *schedulercache.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error {
return nil
}
// GetDeviceRunContainerOptions simply returns nil.
func (h *ManagerStub) GetDeviceRunContainerOptions(pod *v1.Pod, container *v1.Container) (*DeviceRunContainerOptions, error) {
return nil, nil
}
// GetCapacity simply returns nil capacity and empty removed resource list.
func (h *ManagerStub) GetCapacity() (v1.ResourceList, v1.ResourceList, []string) {
return nil, nil, []string{}
}
pkg/kubelet/cm/devicemanager/pod_devices.go deleted 100644 → 0
View file @ 1cc99801
/*
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 devicemanager
import (
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/util/sets"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
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.ContainerAllocateResponse
}
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.ContainerAllocateResponse) {
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
}
// Populates allocatedResources with the device resources allocated to the specified <podUID, contName>.
func (pdev podDevices) addContainerAllocatedResources(podUID, contName string, allocatedResources map[string]sets.String) {
containers, exists := pdev[podUID]
if !exists {
return
}
resources, exists := containers[contName]
if !exists {
return
}
for resource, devices := range resources {
allocatedResources[resource] = allocatedResources[resource].Union(devices.deviceIds)
}
}
// Removes the device resources allocated to the specified <podUID, contName> from allocatedResources.
func (pdev podDevices) removeContainerAllocatedResources(podUID, contName string, allocatedResources map[string]sets.String) {
containers, exists := pdev[podUID]
if !exists {
return
}
resources, exists := containers[contName]
if !exists {
return
}
for resource, devices := range resources {
allocatedResources[resource] = allocatedResources[resource].Difference(devices.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()
}
if devices.allocResp != nil {
ret[resource] = ret[resource].Union(devices.deviceIds)
}
}
}
}
return ret
}
// podDevicesCheckpointEntry is used to record <pod, container> to device allocation information.
type podDevicesCheckpointEntry struct {
PodUID string
ContainerName string
ResourceName string
DeviceIDs []string
AllocResp []byte
}
// Turns podDevices to checkpointData.
func (pdev podDevices) toCheckpointData() []podDevicesCheckpointEntry {
var data []podDevicesCheckpointEntry
for podUID, containerDevices := range pdev {
for conName, resources := range containerDevices {
for resource, devices := range resources {
devIds := devices.deviceIds.UnsortedList()
if devices.allocResp == nil {
glog.Errorf("Can't marshal allocResp for %v %v %v: allocation response is missing", podUID, conName, resource)
continue
}
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 = append(data, podDevicesCheckpointEntry{podUID, conName, resource, devIds, allocResp})
}
}
}
return data
}
// Populates podDevices from the passed in checkpointData.
func (pdev podDevices) fromCheckpointData(data []podDevicesCheckpointEntry) {
for _, entry := range data {
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.ContainerAllocateResponse{}
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)
annotationsMap := 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
// Container annotations
// These artifacts are per resource per container.
// Updates RunContainerOptions.Envs.
for k, v := range resp.Envs {
if e, ok := envsMap[k]; ok {
glog.V(4).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(4).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(4).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,
})
}
// Updates for Annotations
for k, v := range resp.Annotations {
if e, ok := annotationsMap[k]; ok {
glog.V(4).Infof("Skip existing annotation %s %s", k, v)
if e != v {
glog.Errorf("Annotation %s has conflicting setting: %s and %s", k, e, v)
}
continue
}
glog.V(4).Infof("Add annotation %s %s", k, v)
annotationsMap[k] = v
opts.Annotations = append(opts.Annotations, kubecontainer.Annotation{Name: k, Value: v})
}
}
return opts
}
pkg/kubelet/cm/devicemanager/types.go deleted 100644 → 0
View file @ 1cc99801
/*
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 devicemanager
import (
"time"
"k8s.io/api/core/v1"
pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
"k8s.io/kubernetes/pkg/kubelet/config"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/lifecycle"
"k8s.io/kubernetes/pkg/scheduler/schedulercache"
)
// 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) error
// Devices is the map of devices that have registered themselves
// against the manager.
// The map key is the ResourceName of the device plugins.
Devices() map[string][]pluginapi.Device
// Allocate configures and assigns devices to pods. The pods are provided
// through the pod admission attributes in the attrs argument. From the
// requested device resources, Allocate will communicate with the owning
// device plugin to allow setup procedures to take place, and for the
// device plugin to provide runtime settings to use the device (environment
// variables, mount points and device files). The node object is provided
// for the device manager to update the node capacity to reflect the
// currently available devices.
Allocate(node *schedulercache.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error
// Stop stops the manager.
Stop() 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)
// GetCapacity returns the amount of available device plugin resource capacity, resource allocatable
// and inactive device plugin resources previously registered on the node.
GetCapacity() (v1.ResourceList, v1.ResourceList, []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
// The Annotations for the container
Annotations []kubecontainer.Annotation
}
// TODO: evaluate whether we need these error definitions.
const (
// errFailedToDialDevicePlugin is the error raised when the device plugin could not be
// reached on the registered socket
errFailedToDialDevicePlugin = "failed to dial device plugin:"
// errUnsupportedVersion is the error raised when the device plugin uses an API version not
// supported by the Kubelet registry
errUnsupportedVersion = "requested API version %q is not supported by kubelet. Supported versions are %q"
// errDevicePluginAlreadyExists is the error raised when a device plugin with the
// same Resource Name tries to register itself
errDevicePluginAlreadyExists = "another device plugin already registered this Resource Name"
// errInvalidResourceName is the error raised when a device plugin is registering
// itself with an invalid ResourceName
errInvalidResourceName = "the ResourceName %q is invalid"
// errEmptyResourceName is the error raised when the resource name field is empty
errEmptyResourceName = "invalid Empty ResourceName"
// errEndpointStopped indicates that the endpoint has been stopped
errEndpointStopped = "endpoint %v has been stopped"
// errBadSocket is the error raised when the registry socket path is not absolute
errBadSocket = "bad socketPath, must be an absolute path:"
// errRemoveSocket is the error raised when the registry could not remove the existing socket
errRemoveSocket = "failed to remove socket while starting device plugin registry, with error"
// errListenSocket is the error raised when the registry could not listen on the socket
errListenSocket = "failed to listen to socket while starting device plugin registry, with error"
// errListAndWatch is the error raised when ListAndWatch ended unsuccessfully
errListAndWatch = "listAndWatch ended unexpectedly for device plugin %s with error %v"
)
// endpointStopGracePeriod indicates the grace period after an endpoint is stopped
// because its device plugin fails. DeviceManager keeps the stopped endpoint in its
// cache during this grace period to cover the time gap for the capacity change to
// take effect.
const endpointStopGracePeriod = time.Duration(5) * time.Minute
pkg/kubelet/config/config.go
View file @ 1c86ee30
...@@ -26,7 +26,6 @@ import ( ...@@ -26,7 +26,6 @@ import (
"k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/sets"
"k8s.io/client-go/tools/record" "k8s.io/client-go/tools/record"
"k8s.io/kubernetes/pkg/kubelet/checkpoint"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/events" "k8s.io/kubernetes/pkg/kubelet/events"
kubetypes "k8s.io/kubernetes/pkg/kubelet/types" kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
...@@ -62,9 +61,8 @@ type PodConfig struct { ...@@ -62,9 +61,8 @@ type PodConfig struct {
updates chan kubetypes.PodUpdate updates chan kubetypes.PodUpdate
// contains the list of all configured sources // contains the list of all configured sources
sourcesLock sync.Mutex sourcesLock sync.Mutex
sources sets.String sources sets.String
checkpointManager checkpoint.Manager
} }
// NewPodConfig creates an object that can merge many configuration sources into a stream // NewPodConfig creates an object that can merge many configuration sources into a stream
...@@ -112,15 +110,7 @@ func (c *PodConfig) Sync() { ...@@ -112,15 +110,7 @@ func (c *PodConfig) Sync() {
// Restore restores pods from the checkpoint path, *once* // Restore restores pods from the checkpoint path, *once*
func (c *PodConfig) Restore(path string, updates chan<- interface{}) error { func (c *PodConfig) Restore(path string, updates chan<- interface{}) error {
var err error return nil
if c.checkpointManager == nil {
c.checkpointManager = checkpoint.NewCheckpointManager(path)
pods, err := c.checkpointManager.LoadPods()
if err == nil {
updates <- kubetypes.PodUpdate{Pods: pods, Op: kubetypes.RESTORE, Source: kubetypes.ApiserverSource}
}
}
return err
} }
// podStorage manages the current pod state at any point in time and ensures updates // podStorage manages the current pod state at any point in time and ensures updates
......
pkg/kubelet/custommetrics/BUILD deleted 100644 → 0
View file @ 1cc99801
package(default_visibility = ["//visibility:public"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
"go_test",
)
go_library(
name = "go_default_library",
srcs = ["custom_metrics.go"],
importpath = "k8s.io/kubernetes/pkg/kubelet/custommetrics",
deps = ["//vendor/k8s.io/api/core/v1:go_default_library"],
)
go_test(
name = "go_default_test",
srcs = ["custom_metrics_test.go"],
embed = [":go_default_library"],
deps = [
"//vendor/github.com/stretchr/testify/assert:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
pkg/kubelet/custommetrics/custom_metrics.go deleted 100644 → 0
View file @ 1cc99801
/*
Copyright 2015 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 custommetrics contains support for instrumenting cAdvisor to gather custom metrics from pods.
package custommetrics
import (
"path"
"k8s.io/api/core/v1"
)
const (
CustomMetricsDefinitionContainerFile = "definition.json"
CustomMetricsDefinitionDir = "/etc/custom-metrics"
)
// Alpha implementation.
// Returns a path to a cAdvisor-specific custom metrics configuration.
func GetCAdvisorCustomMetricsDefinitionPath(container *v1.Container) (*string, error) {
// Assumes that the container has Custom Metrics enabled if it has "/etc/custom-metrics" directory
// mounted as a volume. Custom Metrics definition is expected to be in "definition.json".
if container.VolumeMounts != nil {
for _, volumeMount := range container.VolumeMounts {
if path.Clean(volumeMount.MountPath) == path.Clean(CustomMetricsDefinitionDir) {
// TODO: add definition file validation.
definitionPath := path.Clean(path.Join(volumeMount.MountPath, CustomMetricsDefinitionContainerFile))
return &definitionPath, nil
}
}
}
// No Custom Metrics definition available.
return nil, nil
}
pkg/kubelet/custommetrics/custom_metrics_test.go deleted 100644 → 0
View file @ 1cc99801
/*
Copyright 2015 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 custommetrics
import (
"testing"
"github.com/stretchr/testify/assert"
"k8s.io/api/core/v1"
)
func TestGetCAdvisorCustomMetricsDefinitionPath(t *testing.T) {
regularContainer := &v1.Container{
Name: "test_container",
}
cmContainer := &v1.Container{
Name: "test_container",
VolumeMounts: []v1.VolumeMount{
{
Name: "cm",
MountPath: CustomMetricsDefinitionDir,
},
},
}
path, err := GetCAdvisorCustomMetricsDefinitionPath(regularContainer)
assert.Nil(t, path)
assert.NoError(t, err)
path, err = GetCAdvisorCustomMetricsDefinitionPath(cmContainer)
assert.NotEmpty(t, *path)
assert.NoError(t, err)
}
pkg/kubelet/gpu/BUILD deleted 100644 → 0
View file @ 1cc99801
package(default_visibility = ["//visibility:public"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
)
go_library(
name = "go_default_library",
srcs = [
"gpu_manager_stub.go",
"types.go",
],
importpath = "k8s.io/kubernetes/pkg/kubelet/gpu",
deps = ["//vendor/k8s.io/api/core/v1:go_default_library"],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [
":package-srcs",
"//pkg/kubelet/gpu/nvidia:all-srcs",
],
tags = ["automanaged"],
)
pkg/kubelet/gpu/OWNERS deleted 100644 → 0
View file @ 1cc99801
approvers:
- dchen1107
- derekwaynecarr
- vishh
- yujuhong
reviewers:
- cmluciano
- jiayingz
- mindprince
- RenaudWasTaken
- vishh
- sig-node-reviewers
pkg/kubelet/gpu/gpu_manager_stub.go deleted 100644 → 0
View file @ 1cc99801
/*
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 gpu
import (
"fmt"
"k8s.io/api/core/v1"
)
type gpuManagerStub struct{}
func (gms *gpuManagerStub) Start() error {
return nil
}
func (gms *gpuManagerStub) Capacity() v1.ResourceList {
return nil
}
func (gms *gpuManagerStub) AllocateGPU(_ *v1.Pod, _ *v1.Container) ([]string, error) {
return nil, fmt.Errorf("GPUs are not supported")
}
func NewGPUManagerStub() GPUManager {
return &gpuManagerStub{}
}
pkg/kubelet/gpu/types.go deleted 100644 → 0
View file @ 1cc99801
/*
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 gpu
import "k8s.io/api/core/v1"
// GPUManager manages GPUs on a local node.
// Implementations are expected to be thread safe.
type GPUManager interface {
// Start logically initializes GPUManager
Start() error
// Capacity returns the total number of GPUs on the node.
Capacity() v1.ResourceList
// AllocateGPU attempts to allocate GPUs for input container.
// Returns paths to allocated GPUs and nil on success.
// Returns an error on failure.
AllocateGPU(*v1.Pod, *v1.Container) ([]string, error)
}
pkg/kubelet/kubelet.go
View file @ 1c86ee30
...@@ -60,7 +60,6 @@ import ( ...@@ -60,7 +60,6 @@ import (
"k8s.io/kubernetes/pkg/kubelet/dockershim" "k8s.io/kubernetes/pkg/kubelet/dockershim"
"k8s.io/kubernetes/pkg/kubelet/events" "k8s.io/kubernetes/pkg/kubelet/events"
"k8s.io/kubernetes/pkg/kubelet/eviction" "k8s.io/kubernetes/pkg/kubelet/eviction"
"k8s.io/kubernetes/pkg/kubelet/gpu"
"k8s.io/kubernetes/pkg/kubelet/images" "k8s.io/kubernetes/pkg/kubelet/images"
"k8s.io/kubernetes/pkg/kubelet/kubeletconfig" "k8s.io/kubernetes/pkg/kubelet/kubeletconfig"
"k8s.io/kubernetes/pkg/kubelet/kuberuntime" "k8s.io/kubernetes/pkg/kubelet/kuberuntime"
...@@ -267,19 +266,7 @@ func makePodSourceConfig(kubeCfg *kubeletconfiginternal.KubeletConfiguration, ku ...@@ -267,19 +266,7 @@ func makePodSourceConfig(kubeCfg *kubeletconfiginternal.KubeletConfiguration, ku
config.NewSourceURL(kubeCfg.StaticPodURL, manifestURLHeader, nodeName, kubeCfg.HTTPCheckFrequency.Duration, cfg.Channel(kubetypes.HTTPSource)) config.NewSourceURL(kubeCfg.StaticPodURL, manifestURLHeader, nodeName, kubeCfg.HTTPCheckFrequency.Duration, cfg.Channel(kubetypes.HTTPSource))
} }
// Restore from the checkpoint path
// NOTE: This MUST happen before creating the apiserver source
// below, or the checkpoint would override the source of truth.
var updatechannel chan<- interface{} var updatechannel chan<- interface{}
if bootstrapCheckpointPath != "" {
glog.Infof("Adding checkpoint path: %v", bootstrapCheckpointPath)
updatechannel = cfg.Channel(kubetypes.ApiserverSource)
err := cfg.Restore(bootstrapCheckpointPath, updatechannel)
if err != nil {
return nil, err
}
}
if kubeDeps.KubeClient != nil { if kubeDeps.KubeClient != nil {
glog.Infof("Watching apiserver") glog.Infof("Watching apiserver")
...@@ -721,10 +708,6 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration, ...@@ -721,10 +708,6 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
klet.appArmorValidator = apparmor.NewValidator(containerRuntime) klet.appArmorValidator = apparmor.NewValidator(containerRuntime)
klet.softAdmitHandlers.AddPodAdmitHandler(lifecycle.NewAppArmorAdmitHandler(klet.appArmorValidator)) klet.softAdmitHandlers.AddPodAdmitHandler(lifecycle.NewAppArmorAdmitHandler(klet.appArmorValidator))
klet.softAdmitHandlers.AddPodAdmitHandler(lifecycle.NewNoNewPrivsAdmitHandler(klet.containerRuntime)) klet.softAdmitHandlers.AddPodAdmitHandler(lifecycle.NewNoNewPrivsAdmitHandler(klet.containerRuntime))
// Set GPU manager to a stub implementation if it is not enabled or cannot be supported.
if klet.gpuManager == nil {
klet.gpuManager = gpu.NewGPUManagerStub()
}
// Finally, put the most recent version of the config on the Kubelet, so // Finally, put the most recent version of the config on the Kubelet, so
// people can see how it was configured. // people can see how it was configured.
klet.kubeletConfiguration = *kubeCfg klet.kubeletConfiguration = *kubeCfg
...@@ -985,13 +968,6 @@ type Kubelet struct { ...@@ -985,13 +968,6 @@ type Kubelet struct {
// The handler serving CRI streaming calls (exec/attach/port-forward). // The handler serving CRI streaming calls (exec/attach/port-forward).
criHandler http.Handler criHandler http.Handler
// GPU Manager
gpuManager gpu.GPUManager
// dockerLegacyService contains some legacy methods for backward compatibility.
// It should be set only when docker is using non json-file logging driver.
dockerLegacyService dockershim.DockerLegacyService
// StatsProvider provides the node and the container stats. // StatsProvider provides the node and the container stats.
*stats.StatsProvider *stats.StatsProvider
...@@ -1092,11 +1068,6 @@ func (kl *Kubelet) initializeModules() error { ...@@ -1092,11 +1068,6 @@ func (kl *Kubelet) initializeModules() error {
return fmt.Errorf("Failed to start OOM watcher %v", err) return fmt.Errorf("Failed to start OOM watcher %v", err)
} }
// Initialize GPUs
if err := kl.gpuManager.Start(); err != nil {
glog.Errorf("Failed to start gpuManager %v", err)
}
// Start resource analyzer // Start resource analyzer
kl.resourceAnalyzer.Start() kl.resourceAnalyzer.Start()
......
pkg/kubelet/kubelet_node_status.go
View file @ 1c86ee30
...@@ -424,14 +424,6 @@ func (kl *Kubelet) setNodeStatusMachineInfo(node *v1.Node) { ...@@ -424,14 +424,6 @@ func (kl *Kubelet) setNodeStatusMachineInfo(node *v1.Node) {
node.Status.Capacity = v1.ResourceList{} node.Status.Capacity = v1.ResourceList{}
} }
// populate GPU capacity.
gpuCapacity := kl.gpuManager.Capacity()
if gpuCapacity != nil {
for k, v := range gpuCapacity {
node.Status.Capacity[k] = v
}
}
var devicePluginAllocatable v1.ResourceList var devicePluginAllocatable v1.ResourceList
var devicePluginCapacity v1.ResourceList var devicePluginCapacity v1.ResourceList
var removedDevicePlugins []string var removedDevicePlugins []string
......
pkg/kubelet/kubelet_pods.go
View file @ 1c86ee30
...@@ -98,21 +98,7 @@ func (kl *Kubelet) GetActivePods() []*v1.Pod { ...@@ -98,21 +98,7 @@ func (kl *Kubelet) GetActivePods() []*v1.Pod {
// makeGPUDevices determines the devices for the given container. // makeGPUDevices determines the devices for the given container.
// Experimental. // Experimental.
func (kl *Kubelet) makeGPUDevices(pod *v1.Pod, container *v1.Container) ([]kubecontainer.DeviceInfo, error) { func (kl *Kubelet) makeGPUDevices(pod *v1.Pod, container *v1.Container) ([]kubecontainer.DeviceInfo, error) {
if container.Resources.Limits.NvidiaGPU().IsZero() { return nil, nil
return nil, nil
}
nvidiaGPUPaths, err := kl.gpuManager.AllocateGPU(pod, container)
if err != nil {
return nil, err
}
var devices []kubecontainer.DeviceInfo
for _, path := range nvidiaGPUPaths {
// Devices have to be mapped one to one because of nvidia CUDA library requirements.
devices = append(devices, kubecontainer.DeviceInfo{PathOnHost: path, PathInContainer: path, Permissions: "mrw"})
}
return devices, nil
} }
// makeBlockVolumes maps the raw block devices specified in the path of the container // makeBlockVolumes maps the raw block devices specified in the path of the container
...@@ -1216,12 +1202,6 @@ func (kl *Kubelet) GetKubeletContainerLogs(podFullName, containerName string, lo ...@@ -1216,12 +1202,6 @@ func (kl *Kubelet) GetKubeletContainerLogs(podFullName, containerName string, lo
return err return err
} }
if kl.dockerLegacyService != nil {
// dockerLegacyService should only be non-nil when we actually need it, so
// inject it into the runtimeService.
// TODO(random-liu): Remove this hack after deprecating unsupported log driver.
return kl.dockerLegacyService.GetContainerLogs(pod, containerID, logOptions, stdout, stderr)
}
return kl.containerRuntime.GetContainerLogs(pod, containerID, logOptions, stdout, stderr) return kl.containerRuntime.GetContainerLogs(pod, containerID, logOptions, stdout, stderr)
} }
......
pkg/kubelet/mountpod/BUILD deleted 100644 → 0
View file @ 1cc99801
load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
go_library(
name = "go_default_library",
srcs = ["mount_pod.go"],
importpath = "k8s.io/kubernetes/pkg/kubelet/mountpod",
visibility = ["//visibility:public"],
deps = [
"//pkg/kubelet/config:go_default_library",
"//pkg/kubelet/pod:go_default_library",
"//pkg/util/strings:go_default_library",
"//vendor/k8s.io/api/core/v1:go_default_library",
],
)
go_test(
name = "go_default_test",
srcs = ["mount_pod_test.go"],
embed = [":go_default_library"],
deps = [
"//pkg/kubelet/configmap:go_default_library",
"//pkg/kubelet/pod:go_default_library",
"//pkg/kubelet/pod/testing:go_default_library",
"//pkg/kubelet/secret: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/apis/meta/v1:go_default_library",
"//vendor/k8s.io/client-go/util/testing:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
visibility = ["//visibility:public"],
)
pkg/kubelet/mountpod/mount_pod.go deleted 100644 → 0
View file @ 1cc99801
/*
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 mountpod
import (
"encoding/json"
"fmt"
"io/ioutil"
"os"
"path"
"k8s.io/api/core/v1"
"k8s.io/kubernetes/pkg/kubelet/config"
kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
"k8s.io/kubernetes/pkg/util/strings"
)
// Manager is an interface that tracks pods with mount utilities for individual
// volume plugins.
type Manager interface {
GetMountPod(pluginName string) (pod *v1.Pod, container string, err error)
}
// basicManager is simple implementation of Manager. Pods with mount utilities
// are registered by placing a JSON file into
// /var/lib/kubelet/plugin-containers/<plugin name>.json and this manager just
// finds them there.
type basicManager struct {
registrationDirectory string
podManager kubepod.Manager
}
// volumePluginRegistration specified format of the json files placed in
// /var/lib/kubelet/plugin-containers/
type volumePluginRegistration struct {
PodName string `json:"podName"`
PodNamespace string `json:"podNamespace"`
PodUID string `json:"podUID"`
ContainerName string `json:"containerName"`
}
// NewManager returns a new mount pod manager.
func NewManager(rootDirectory string, podManager kubepod.Manager) (Manager, error) {
regPath := path.Join(rootDirectory, config.DefaultKubeletPluginContainersDirName)
// Create the directory on startup
os.MkdirAll(regPath, 0700)
return &basicManager{
registrationDirectory: regPath,
podManager: podManager,
}, nil
}
func (m *basicManager) getVolumePluginRegistrationPath(pluginName string) string {
// sanitize plugin name so it does not escape directory
safePluginName := strings.EscapePluginName(pluginName) + ".json"
return path.Join(m.registrationDirectory, safePluginName)
}
func (m *basicManager) GetMountPod(pluginName string) (pod *v1.Pod, containerName string, err error) {
// Read /var/lib/kubelet/plugin-containers/<plugin name>.json
regPath := m.getVolumePluginRegistrationPath(pluginName)
regBytes, err := ioutil.ReadFile(regPath)
if err != nil {
if os.IsNotExist(err) {
// No pod is registered for this plugin
return nil, "", nil
}
return nil, "", fmt.Errorf("cannot read %s: %v", regPath, err)
}
// Parse json
var reg volumePluginRegistration
if err := json.Unmarshal(regBytes, &reg); err != nil {
return nil, "", fmt.Errorf("unable to parse %s: %s", regPath, err)
}
if len(reg.ContainerName) == 0 {
return nil, "", fmt.Errorf("unable to parse %s: \"containerName\" is not set", regPath)
}
if len(reg.PodUID) == 0 {
return nil, "", fmt.Errorf("unable to parse %s: \"podUID\" is not set", regPath)
}
if len(reg.PodNamespace) == 0 {
return nil, "", fmt.Errorf("unable to parse %s: \"podNamespace\" is not set", regPath)
}
if len(reg.PodName) == 0 {
return nil, "", fmt.Errorf("unable to parse %s: \"podName\" is not set", regPath)
}
pod, ok := m.podManager.GetPodByName(reg.PodNamespace, reg.PodName)
if !ok {
return nil, "", fmt.Errorf("unable to process %s: pod %s/%s not found", regPath, reg.PodNamespace, reg.PodName)
}
if string(pod.UID) != reg.PodUID {
return nil, "", fmt.Errorf("unable to process %s: pod %s/%s has unexpected UID", regPath, reg.PodNamespace, reg.PodName)
}
// make sure that reg.ContainerName exists in the pod
for i := range pod.Spec.Containers {
if pod.Spec.Containers[i].Name == reg.ContainerName {
return pod, reg.ContainerName, nil
}
}
return nil, "", fmt.Errorf("unable to process %s: pod %s/%s has no container named %q", regPath, reg.PodNamespace, reg.PodName, reg.ContainerName)
}
pkg/kubelet/mountpod/mount_pod_test.go deleted 100644 → 0
View file @ 1cc99801
/*
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 mountpod
import (
"io/ioutil"
"os"
"path"
"testing"
"github.com/golang/glog"
"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
utiltesting "k8s.io/client-go/util/testing"
"k8s.io/kubernetes/pkg/kubelet/configmap"
kubepod "k8s.io/kubernetes/pkg/kubelet/pod"
podtest "k8s.io/kubernetes/pkg/kubelet/pod/testing"
"k8s.io/kubernetes/pkg/kubelet/secret"
)
func TestGetVolumeExec(t *testing.T) {
// prepare PodManager
pods := []*v1.Pod{
{
ObjectMeta: metav1.ObjectMeta{
UID: "12345678",
Name: "foo",
Namespace: "bar",
},
Spec: v1.PodSpec{
Containers: []v1.Container{
{Name: "baz"},
},
},
},
}
fakeSecretManager := secret.NewFakeManager()
fakeConfigMapManager := configmap.NewFakeManager()
podManager := kubepod.NewBasicPodManager(
podtest.NewFakeMirrorClient(), fakeSecretManager, fakeConfigMapManager)
podManager.SetPods(pods)
// Prepare fake /var/lib/kubelet
basePath, err := utiltesting.MkTmpdir("kubelet")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(basePath)
regPath := path.Join(basePath, "plugin-containers")
mgr, err := NewManager(basePath, podManager)
if err != nil {
t.Fatal(err)
}
tests := []struct {
name string
json string
expectError bool
}{
{
"invalid json",
"{{{}",
true,
},
{
"missing json",
"", // this means no json file should be created
false,
},
{
"missing podNamespace",
`{"podName": "foo", "podUID": "87654321", "containerName": "baz"}`,
true,
},
{
"missing podName",
`{"podNamespace": "bar", "podUID": "87654321", "containerName": "baz"}`,
true,
},
{
"missing containerName",
`{"podNamespace": "bar", "podName": "foo", "podUID": "87654321"}`,
true,
},
{
"missing podUID",
`{"podNamespace": "bar", "podName": "foo", "containerName": "baz"}`,
true,
},
{
"missing pod",
`{"podNamespace": "bar", "podName": "non-existing-pod", "podUID": "12345678", "containerName": "baz"}`,
true,
},
{
"invalid uid",
`{"podNamespace": "bar", "podName": "foo", "podUID": "87654321", "containerName": "baz"}`,
true,
},
{
"invalid container",
`{"podNamespace": "bar", "podName": "foo", "podUID": "12345678", "containerName": "invalid"}`,
true,
},
{
"valid pod",
`{"podNamespace": "bar", "podName": "foo", "podUID": "12345678", "containerName": "baz"}`,
false,
},
}
for _, test := range tests {
p := path.Join(regPath, "kubernetes.io~glusterfs.json")
if len(test.json) > 0 {
if err := ioutil.WriteFile(p, []byte(test.json), 0600); err != nil {
t.Errorf("test %q: error writing %s: %v", test.name, p, err)
continue
}
} else {
// "" means no JSON file
os.Remove(p)
}
pod, container, err := mgr.GetMountPod("kubernetes.io/glusterfs")
if err != nil {
glog.V(5).Infof("test %q returned error %s", test.name, err)
}
if err == nil && test.expectError {
t.Errorf("test %q: expected error, got none", test.name)
}
if err != nil && !test.expectError {
t.Errorf("test %q: unexpected error: %v", test.name, err)
}
if err == nil {
// Pod must be returned when the json file was not empty
if pod == nil && len(test.json) != 0 {
t.Errorf("test %q: expected exec, got nil", test.name)
}
// Both pod and container must be returned
if pod != nil && len(container) == 0 {
t.Errorf("test %q: expected container name, got %q", test.name, container)
}
}
}
}
pkg/kubelet/pod/pod_manager.go
View file @ 1c86ee30
...@@ -19,11 +19,8 @@ package pod ...@@ -19,11 +19,8 @@ package pod
import ( import (
"sync" "sync"
"github.com/golang/glog"
"k8s.io/api/core/v1" "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/types"
"k8s.io/kubernetes/pkg/kubelet/checkpoint"
"k8s.io/kubernetes/pkg/kubelet/configmap" "k8s.io/kubernetes/pkg/kubelet/configmap"
kubecontainer "k8s.io/kubernetes/pkg/kubelet/container" kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/secret" "k8s.io/kubernetes/pkg/kubelet/secret"
...@@ -119,9 +116,8 @@ type basicManager struct { ...@@ -119,9 +116,8 @@ type basicManager struct {
translationByUID map[kubetypes.MirrorPodUID]kubetypes.ResolvedPodUID translationByUID map[kubetypes.MirrorPodUID]kubetypes.ResolvedPodUID
// basicManager is keeping secretManager and configMapManager up-to-date. // basicManager is keeping secretManager and configMapManager up-to-date.
secretManager secret.Manager secretManager secret.Manager
configMapManager configmap.Manager configMapManager configmap.Manager
checkpointManager checkpoint.Manager
// A mirror pod client to create/delete mirror pods. // A mirror pod client to create/delete mirror pods.
MirrorClient MirrorClient
...@@ -132,7 +128,6 @@ func NewBasicPodManager(client MirrorClient, secretManager secret.Manager, confi ...@@ -132,7 +128,6 @@ func NewBasicPodManager(client MirrorClient, secretManager secret.Manager, confi
pm := &basicManager{} pm := &basicManager{}
pm.secretManager = secretManager pm.secretManager = secretManager
pm.configMapManager = configMapManager pm.configMapManager = configMapManager
pm.checkpointManager = checkpoint.GetInstance()
pm.MirrorClient = client pm.MirrorClient = client
pm.SetPods(nil) pm.SetPods(nil)
return pm return pm
...@@ -160,11 +155,6 @@ func (pm *basicManager) UpdatePod(pod *v1.Pod) { ...@@ -160,11 +155,6 @@ func (pm *basicManager) UpdatePod(pod *v1.Pod) {
pm.lock.Lock() pm.lock.Lock()
defer pm.lock.Unlock() defer pm.lock.Unlock()
pm.updatePodsInternal(pod) pm.updatePodsInternal(pod)
if pm.checkpointManager != nil {
if err := pm.checkpointManager.WritePod(pod); err != nil {
glog.Errorf("Error writing checkpoint for pod: %v", pod.GetName())
}
}
} }
// updatePodsInternal replaces the given pods in the current state of the // updatePodsInternal replaces the given pods in the current state of the
...@@ -223,11 +213,6 @@ func (pm *basicManager) DeletePod(pod *v1.Pod) { ...@@ -223,11 +213,6 @@ func (pm *basicManager) DeletePod(pod *v1.Pod) {
delete(pm.podByUID, kubetypes.ResolvedPodUID(pod.UID)) delete(pm.podByUID, kubetypes.ResolvedPodUID(pod.UID))
delete(pm.podByFullName, podFullName) delete(pm.podByFullName, podFullName)
} }
if pm.checkpointManager != nil {
if err := pm.checkpointManager.DeletePod(pod); err != nil {
glog.Errorf("Error deleting checkpoint for pod: %v", pod.GetName())
}
}
} }
func (pm *basicManager) GetPods() []*v1.Pod { func (pm *basicManager) GetPods() []*v1.Pod {
......
pkg/kubelet/volume_host.go
View file @ 1c86ee30
...@@ -28,7 +28,6 @@ import ( ...@@ -28,7 +28,6 @@ import (
clientset "k8s.io/client-go/kubernetes" clientset "k8s.io/client-go/kubernetes"
"k8s.io/kubernetes/pkg/kubelet/configmap" "k8s.io/kubernetes/pkg/kubelet/configmap"
"k8s.io/kubernetes/pkg/kubelet/container" "k8s.io/kubernetes/pkg/kubelet/container"
"k8s.io/kubernetes/pkg/kubelet/mountpod"
"k8s.io/kubernetes/pkg/kubelet/secret" "k8s.io/kubernetes/pkg/kubelet/secret"
"k8s.io/kubernetes/pkg/util/io" "k8s.io/kubernetes/pkg/util/io"
"k8s.io/kubernetes/pkg/util/mount" "k8s.io/kubernetes/pkg/util/mount"
...@@ -49,16 +48,11 @@ func NewInitializedVolumePluginMgr( ...@@ -49,16 +48,11 @@ func NewInitializedVolumePluginMgr(
plugins []volume.VolumePlugin, plugins []volume.VolumePlugin,
prober volume.DynamicPluginProber) (*volume.VolumePluginMgr, error) { prober volume.DynamicPluginProber) (*volume.VolumePluginMgr, error) {
mountPodManager, err := mountpod.NewManager(kubelet.getRootDir(), kubelet.podManager)
if err != nil {
return nil, err
}
kvh := &kubeletVolumeHost{ kvh := &kubeletVolumeHost{
kubelet: kubelet, kubelet: kubelet,
volumePluginMgr: volume.VolumePluginMgr{}, volumePluginMgr: volume.VolumePluginMgr{},
secretManager: secretManager, secretManager: secretManager,
configMapManager: configMapManager, configMapManager: configMapManager,
mountPodManager: mountPodManager,
} }
if err := kvh.volumePluginMgr.InitPlugins(plugins, prober, kvh); err != nil { if err := kvh.volumePluginMgr.InitPlugins(plugins, prober, kvh); err != nil {
...@@ -82,7 +76,6 @@ type kubeletVolumeHost struct { ...@@ -82,7 +76,6 @@ type kubeletVolumeHost struct {
volumePluginMgr volume.VolumePluginMgr volumePluginMgr volume.VolumePluginMgr
secretManager secret.Manager secretManager secret.Manager
configMapManager configmap.Manager configMapManager configmap.Manager
mountPodManager mountpod.Manager
} }
func (kvh *kubeletVolumeHost) GetVolumeDevicePluginDir(pluginName string) string { func (kvh *kubeletVolumeHost) GetVolumeDevicePluginDir(pluginName string) string {
......
pkg/kubemark/.import-restrictions deleted 100644 → 0
View file @ 1cc99801
{
"Rules": [
{
"SelectorRegexp": "k8s[.]io/kubernetes/cmd",
"AllowedPrefixes": [
"k8s.io/kubernetes/cmd/kube-proxy/app",
"k8s.io/kubernetes/cmd/kubelet/app"
],
"ForbiddenPrefixes": [
]
}
]
}
\ No newline at end of file
pkg/kubemark/BUILD deleted 100644 → 0
View file @ 1cc99801
package(default_visibility = ["//visibility:public"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
)
go_library(
name = "go_default_library",
srcs = [
"controller.go",
"hollow_kubelet.go",
"hollow_proxy.go",
],
importpath = "k8s.io/kubernetes/pkg/kubemark",
deps = [
"//cmd/kube-proxy/app:go_default_library",
"//cmd/kubelet/app:go_default_library",
"//cmd/kubelet/app/options:go_default_library",
"//pkg/apis/core:go_default_library",
"//pkg/client/clientset_generated/internalclientset:go_default_library",
"//pkg/controller:go_default_library",
"//pkg/kubelet:go_default_library",
"//pkg/kubelet/apis/kubeletconfig:go_default_library",
"//pkg/kubelet/cadvisor:go_default_library",
"//pkg/kubelet/cm:go_default_library",
"//pkg/kubelet/container/testing:go_default_library",
"//pkg/kubelet/dockershim:go_default_library",
"//pkg/kubelet/types:go_default_library",
"//pkg/proxy:go_default_library",
"//pkg/proxy/config:go_default_library",
"//pkg/proxy/iptables:go_default_library",
"//pkg/util/io:go_default_library",
"//pkg/util/iptables:go_default_library",
"//pkg/util/mount:go_default_library",
"//pkg/util/node:go_default_library",
"//pkg/util/oom:go_default_library",
"//pkg/util/pointer:go_default_library",
"//pkg/util/sysctl:go_default_library",
"//pkg/volume/empty_dir:go_default_library",
"//pkg/volume/secret:go_default_library",
"//test/utils: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/apis/meta/v1:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/fields:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
"//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
"//vendor/k8s.io/client-go/informers:go_default_library",
"//vendor/k8s.io/client-go/informers/core/v1:go_default_library",
"//vendor/k8s.io/client-go/kubernetes:go_default_library",
"//vendor/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
"//vendor/k8s.io/client-go/listers/core/v1:go_default_library",
"//vendor/k8s.io/client-go/tools/cache:go_default_library",
"//vendor/k8s.io/client-go/tools/record:go_default_library",
"//vendor/k8s.io/utils/exec:go_default_library",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
pkg/kubemark/OWNERS deleted 100644 → 0
View file @ 1cc99801
reviewers:
- gmarek
- shyamjvs
- wojtek-t
approvers:
- gmarek
- shyamjvs
- wojtek-t
pkg/kubemark/hollow_kubelet.go deleted 100644 → 0
View file @ 1cc99801
/*
Copyright 2015 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 kubemark
import (
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
clientset "k8s.io/client-go/kubernetes"
kubeletapp "k8s.io/kubernetes/cmd/kubelet/app"
"k8s.io/kubernetes/cmd/kubelet/app/options"
"k8s.io/kubernetes/pkg/kubelet"
"k8s.io/kubernetes/pkg/kubelet/apis/kubeletconfig"
"k8s.io/kubernetes/pkg/kubelet/cadvisor"
"k8s.io/kubernetes/pkg/kubelet/cm"
containertest "k8s.io/kubernetes/pkg/kubelet/container/testing"
"k8s.io/kubernetes/pkg/kubelet/dockershim"
kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
kubeio "k8s.io/kubernetes/pkg/util/io"
"k8s.io/kubernetes/pkg/util/mount"
"k8s.io/kubernetes/pkg/util/oom"
"k8s.io/kubernetes/pkg/volume/empty_dir"
"k8s.io/kubernetes/pkg/volume/secret"
"k8s.io/kubernetes/test/utils"
"github.com/golang/glog"
)
type HollowKubelet struct {
KubeletFlags *options.KubeletFlags
KubeletConfiguration *kubeletconfig.KubeletConfiguration
KubeletDeps *kubelet.Dependencies
}
func NewHollowKubelet(
nodeName string,
client *clientset.Clientset,
cadvisorInterface cadvisor.Interface,
dockerClientConfig *dockershim.ClientConfig,
kubeletPort, kubeletReadOnlyPort int,
containerManager cm.ContainerManager,
maxPods int, podsPerCore int,
) *HollowKubelet {
// -----------------
// Static config
// -----------------
f, c := GetHollowKubeletConfig(nodeName, kubeletPort, kubeletReadOnlyPort, maxPods, podsPerCore)
// -----------------
// Injected objects
// -----------------
volumePlugins := empty_dir.ProbeVolumePlugins()
volumePlugins = append(volumePlugins, secret.ProbeVolumePlugins()...)
d := &kubelet.Dependencies{
KubeClient: client,
HeartbeatClient: client.CoreV1(),
DockerClientConfig: dockerClientConfig,
CAdvisorInterface: cadvisorInterface,
Cloud: nil,
OSInterface: &containertest.FakeOS{},
ContainerManager: containerManager,
VolumePlugins: volumePlugins,
TLSOptions: nil,
OOMAdjuster: oom.NewFakeOOMAdjuster(),
Writer: &kubeio.StdWriter{},
Mounter: mount.New("" /* default mount path */),
}
return &HollowKubelet{
KubeletFlags: f,
KubeletConfiguration: c,
KubeletDeps: d,
}
}
// Starts this HollowKubelet and blocks.
func (hk *HollowKubelet) Run() {
if err := kubeletapp.RunKubelet(hk.KubeletFlags, hk.KubeletConfiguration, hk.KubeletDeps, false); err != nil {
glog.Fatalf("Failed to run HollowKubelet: %v. Exiting.", err)
}
select {}
}
// Builds a KubeletConfiguration for the HollowKubelet, ensuring that the
// usual defaults are applied for fields we do not override.
func GetHollowKubeletConfig(
nodeName string,
kubeletPort int,
kubeletReadOnlyPort int,
maxPods int,
podsPerCore int) (*options.KubeletFlags, *kubeletconfig.KubeletConfiguration) {
testRootDir := utils.MakeTempDirOrDie("hollow-kubelet.", "")
podFilePath := utils.MakeTempDirOrDie("static-pods", testRootDir)
glog.Infof("Using %s as root dir for hollow-kubelet", testRootDir)
// Flags struct
f := options.NewKubeletFlags()
f.EnableServer = true
f.RootDirectory = testRootDir
f.HostnameOverride = nodeName
f.MinimumGCAge = metav1.Duration{Duration: 1 * time.Minute}
f.MaxContainerCount = 100
f.MaxPerPodContainerCount = 2
f.RegisterNode = true
f.RegisterSchedulable = true
// Config struct
c, err := options.NewKubeletConfiguration()
if err != nil {
panic(err)
}
c.StaticPodURL = ""
c.Address = "0.0.0.0" /* bind address */
c.Port = int32(kubeletPort)
c.ReadOnlyPort = int32(kubeletReadOnlyPort)
c.StaticPodPath = podFilePath
c.FileCheckFrequency.Duration = 20 * time.Second
c.HTTPCheckFrequency.Duration = 20 * time.Second
c.NodeStatusUpdateFrequency.Duration = 10 * time.Second
c.SyncFrequency.Duration = 10 * time.Second
c.EvictionPressureTransitionPeriod.Duration = 5 * time.Minute
c.MaxPods = int32(maxPods)
c.PodsPerCore = int32(podsPerCore)
c.ClusterDNS = []string{}
c.ImageGCHighThresholdPercent = 90
c.ImageGCLowThresholdPercent = 80
c.VolumeStatsAggPeriod.Duration = time.Minute
c.CgroupRoot = ""
c.CPUCFSQuota = true
c.EnableControllerAttachDetach = false
c.EnableDebuggingHandlers = true
c.CgroupsPerQOS = false
// hairpin-veth is used to allow hairpin packets. Note that this deviates from
// what the "real" kubelet currently does, because there's no way to
// set promiscuous mode on docker0.
c.HairpinMode = kubeletconfig.HairpinVeth
c.MaxOpenFiles = 1024
c.RegistryBurst = 10
c.RegistryPullQPS = 5.0
c.ResolverConfig = kubetypes.ResolvConfDefault
c.KubeletCgroups = "/kubelet"
c.SerializeImagePulls = true
c.SystemCgroups = ""
c.ProtectKernelDefaults = false
return f, c
}
pkg/kubemark/hollow_proxy.go deleted 100644 → 0
View file @ 1cc99801
/*
Copyright 2015 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 kubemark
import (
"fmt"
"net"
"time"
"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
"k8s.io/client-go/tools/record"
proxyapp "k8s.io/kubernetes/cmd/kube-proxy/app"
api "k8s.io/kubernetes/pkg/apis/core"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
"k8s.io/kubernetes/pkg/proxy"
proxyconfig "k8s.io/kubernetes/pkg/proxy/config"
"k8s.io/kubernetes/pkg/proxy/iptables"
utiliptables "k8s.io/kubernetes/pkg/util/iptables"
utilnode "k8s.io/kubernetes/pkg/util/node"
utilpointer "k8s.io/kubernetes/pkg/util/pointer"
utilsysctl "k8s.io/kubernetes/pkg/util/sysctl"
utilexec "k8s.io/utils/exec"
"github.com/golang/glog"
)
type HollowProxy struct {
ProxyServer *proxyapp.ProxyServer
}
type FakeProxier struct{}
func (*FakeProxier) Sync() {}
func (*FakeProxier) SyncLoop() {
select {}
}
func (*FakeProxier) OnServiceAdd(service *api.Service) {}
func (*FakeProxier) OnServiceUpdate(oldService, service *api.Service) {}
func (*FakeProxier) OnServiceDelete(service *api.Service) {}
func (*FakeProxier) OnServiceSynced() {}
func (*FakeProxier) OnEndpointsAdd(endpoints *api.Endpoints) {}
func (*FakeProxier) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints) {}
func (*FakeProxier) OnEndpointsDelete(endpoints *api.Endpoints) {}
func (*FakeProxier) OnEndpointsSynced() {}
func NewHollowProxyOrDie(
nodeName string,
client clientset.Interface,
eventClient v1core.EventsGetter,
iptInterface utiliptables.Interface,
sysctl utilsysctl.Interface,
execer utilexec.Interface,
broadcaster record.EventBroadcaster,
recorder record.EventRecorder,
useRealProxier bool,
) (*HollowProxy, error) {
// Create proxier and service/endpoint handlers.
var proxier proxy.ProxyProvider
var serviceHandler proxyconfig.ServiceHandler
var endpointsHandler proxyconfig.EndpointsHandler
if useRealProxier {
// Real proxier with fake iptables, sysctl, etc underneath it.
//var err error
proxierIPTables, err := iptables.NewProxier(
iptInterface,
sysctl,
execer,
30*time.Second,
5*time.Second,
false,
0,
"10.0.0.0/8",
nodeName,
getNodeIP(client, nodeName),
recorder,
nil,
[]string{},
)
if err != nil {
return nil, fmt.Errorf("unable to create proxier: %v", err)
}
proxier = proxierIPTables
serviceHandler = proxierIPTables
endpointsHandler = proxierIPTables
} else {
proxier = &FakeProxier{}
serviceHandler = &FakeProxier{}
endpointsHandler = &FakeProxier{}
}
// Create a Hollow Proxy instance.
nodeRef := &v1.ObjectReference{
Kind: "Node",
Name: nodeName,
UID: types.UID(nodeName),
Namespace: "",
}
return &HollowProxy{
ProxyServer: &proxyapp.ProxyServer{
Client: client,
EventClient: eventClient,
IptInterface: iptInterface,
Proxier: proxier,
Broadcaster: broadcaster,
Recorder: recorder,
ProxyMode: "fake",
NodeRef: nodeRef,
OOMScoreAdj: utilpointer.Int32Ptr(0),
ResourceContainer: "",
ConfigSyncPeriod: 30 * time.Second,
ServiceEventHandler: serviceHandler,
EndpointsEventHandler: endpointsHandler,
},
}, nil
}
func (hp *HollowProxy) Run() {
if err := hp.ProxyServer.Run(); err != nil {
glog.Fatalf("Error while running proxy: %v\n", err)
}
}
func getNodeIP(client clientset.Interface, hostname string) net.IP {
var nodeIP net.IP
node, err := client.Core().Nodes().Get(hostname, metav1.GetOptions{})
if err != nil {
glog.Warningf("Failed to retrieve node info: %v", err)
return nil
}
nodeIP, err = utilnode.InternalGetNodeHostIP(node)
if err != nil {
glog.Warningf("Failed to retrieve node IP: %v", err)
return nil
}
return nodeIP
}
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