Remove devicemanager and pluginwatcher

cmd/kubelet/app/server.go

@@ -613,8 +613,6 @@ func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan
 			return err
 		}
 
-		devicePluginEnabled := utilfeature.DefaultFeatureGate.Enabled(features.DevicePlugins)
-
 		kubeDeps.ContainerManager, err = cm.NewContainerManager(
 			kubeDeps.Mounter,
 			kubeDeps.CAdvisorInterface,
@@ -644,7 +642,6 @@ func run(s *options.KubeletServer, kubeDeps *kubelet.Dependencies, stopCh <-chan
 				CPUCFSQuotaPeriod:                     s.CPUCFSQuotaPeriod.Duration,
 			},
 			s.FailSwapOn,
-			devicePluginEnabled,
 			kubeDeps.Recorder)
 
 		if err != nil {

pkg/features/kube_features.go

@@ -46,12 +46,6 @@ const (
 	// SYS_TIME). This should only be enabled if user namespace remapping is enabled in the docker daemon.
 	ExperimentalHostUserNamespaceDefaultingGate utilfeature.Feature = "ExperimentalHostUserNamespaceDefaulting"
 
-	// owner: @jiayingz
-	// beta: v1.10
-	//
-	// Enables support for Device Plugins
-	DevicePlugins utilfeature.Feature = "DevicePlugins"
-
 	// owner: @msau42
 	// alpha: v1.7
 	//
@@ -200,13 +194,6 @@ const (
 	PodReadinessGates utilfeature.Feature = "PodReadinessGates"
 
 	// owner: @vikaschoudhary16
-	// alpha: v1.11
-	//
-	//
-	// Enable probe based plugin watcher utility for discovering Kubelet plugins
-	KubeletPluginsWatcher utilfeature.Feature = "KubeletPluginsWatcher"
-
-	// owner: @vikaschoudhary16
 	// beta: v1.12
 	//
 	//
@@ -225,7 +212,6 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	AppArmor:             {Default: true, PreRelease: utilfeature.Beta},
 	DynamicKubeletConfig: {Default: true, PreRelease: utilfeature.Beta},
 	ExperimentalHostUserNamespaceDefaultingGate: {Default: false, PreRelease: utilfeature.Beta},
-	DevicePlugins:                               {Default: true, PreRelease: utilfeature.Beta},
 	PersistentLocalVolumes:                      {Default: true, PreRelease: utilfeature.Beta},
 	LocalStorageCapacityIsolation:               {Default: true, PreRelease: utilfeature.Beta},
 	HugePages:                                   {Default: true, PreRelease: utilfeature.Beta},
@@ -250,7 +236,6 @@ var defaultKubernetesFeatureGates = map[utilfeature.Feature]utilfeature.FeatureS
 	RunAsGroup:                                  {Default: true, PreRelease: utilfeature.Beta},
 	VolumeSubpath:                               {Default: true, PreRelease: utilfeature.GA},
 	PodReadinessGates:                           {Default: true, PreRelease: utilfeature.Beta},
-	KubeletPluginsWatcher:                       {Default: true, PreRelease: utilfeature.Beta},
 	ResourceQuotaScopeSelectors:                 {Default: true, PreRelease: utilfeature.Beta},
 
 	// inherited features from generic apiserver, relisted here to get a conflict if it is changed

pkg/kubelet/apis/deviceplugin/OWNERS

-reviewers:
-- jiayingz
-- mindprince
-- RenaudWasTaken
-- vishh

pkg/kubelet/apis/deviceplugin/v1alpha/BUILD

-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"],
-)

pkg/kubelet/apis/deviceplugin/v1alpha/api.proto

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

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

-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"],
-)

pkg/kubelet/apis/deviceplugin/v1beta1/api.proto

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

-/*
-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/apis/pluginregistration/v1alpha1/BUILD

-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/pluginregistration/v1alpha1",
-    deps = [
-        "//vendor/github.com/gogo/protobuf/gogoproto:go_default_library",
-        "//vendor/github.com/gogo/protobuf/proto: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"],
-)

pkg/kubelet/apis/pluginregistration/v1alpha1/api.proto

-// To regenerate api.pb.go run hack/update-generated-kubelet-plugin-registration.sh
-syntax = 'proto3';
-
-package pluginregistration;
-
-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;
-
-// PluginInfo is the message sent from a plugin to the Kubelet pluginwatcher for plugin registration
-message PluginInfo {
-	// Type of the Plugin. CSIPlugin or DevicePlugin
-	string type = 1;
-	// Plugin name that uniquely identifies the plugin for the given plugin type.
-	// For DevicePlugin, this is the resource name that the plugin manages and
-	// should follow the extended resource name convention.
-	// For CSI, this is the CSI driver registrar name.
-	string name = 2;
-	// Optional endpoint location. If found set by Kubelet component,
-	// Kubelet component will use this endpoint for specific requests.
-	// This allows the plugin to register using one endpoint and possibly use
-	// a different socket for control operations. CSI uses this model to delegate
-	// its registration external from the plugin.
-	string endpoint = 3;
-	// Plugin service API versions the plugin supports.
-	// For DevicePlugin, this maps to the deviceplugin API versions the
-	// plugin supports at the given socket.
-	// The Kubelet component communicating with the plugin should be able
-	// to choose any preferred version from this list, or returns an error
-	// if none of the listed versions is supported.
-	repeated string supported_versions = 4;
-}
-
-// RegistrationStatus is the message sent from Kubelet pluginwatcher to the plugin for notification on registration status
-message RegistrationStatus {
-	// True if plugin gets registered successfully at Kubelet
-	bool plugin_registered  = 1;
-	// Error message in case plugin fails to register, empty string otherwise
-	string error  = 2;
-}
-
-// RegistrationStatusResponse is sent by plugin to kubelet in response to RegistrationStatus RPC
-message RegistrationStatusResponse {
-}
-
-// InfoRequest is the empty request message from Kubelet
-message InfoRequest {
-}
-
-// Registration is the service advertised by the Plugins.
-service Registration {
-	rpc GetInfo(InfoRequest) returns (PluginInfo) {}
-	rpc NotifyRegistrationStatus(RegistrationStatus) returns (RegistrationStatusResponse) {}
-}

pkg/kubelet/apis/pluginregistration/v1alpha1/constants.go

-/*
-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 pluginregistration
-
-const (
-	CSIPlugin    = "CSIPlugin"
-	DevicePlugin = "DevicePlugin"
-)

pkg/kubelet/apis/pluginregistration/v1beta1/BUILD

-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/pluginregistration/v1beta1",
-    deps = [
-        "//vendor/github.com/gogo/protobuf/gogoproto:go_default_library",
-        "//vendor/github.com/gogo/protobuf/proto: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"],
-)

pkg/kubelet/apis/pluginregistration/v1beta1/api.proto

-// To regenerate api.pb.go run hack/update-generated-kubelet-plugin-registration.sh
-syntax = 'proto3';
-
-package pluginregistration;
-
-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;
-
-// PluginInfo is the message sent from a plugin to the Kubelet pluginwatcher for plugin registration
-message PluginInfo {
-	// Type of the Plugin. CSIPlugin or DevicePlugin
-	string type = 1;
-	// Plugin name that uniquely identifies the plugin for the given plugin type.
-	// For DevicePlugin, this is the resource name that the plugin manages and
-	// should follow the extended resource name convention.
-	// For CSI, this is the CSI driver registrar name.
-	string name = 2;
-	// Optional endpoint location. If found set by Kubelet component,
-	// Kubelet component will use this endpoint for specific requests.
-	// This allows the plugin to register using one endpoint and possibly use
-	// a different socket for control operations. CSI uses this model to delegate
-	// its registration external from the plugin.
-	string endpoint = 3;
-	// Plugin service API versions the plugin supports.
-	// For DevicePlugin, this maps to the deviceplugin API versions the
-	// plugin supports at the given socket.
-	// The Kubelet component communicating with the plugin should be able
-	// to choose any preferred version from this list, or returns an error
-	// if none of the listed versions is supported.
-	repeated string supported_versions = 4;
-}
-
-// RegistrationStatus is the message sent from Kubelet pluginwatcher to the plugin for notification on registration status
-message RegistrationStatus {
-	// True if plugin gets registered successfully at Kubelet
-	bool plugin_registered  = 1;
-	// Error message in case plugin fails to register, empty string otherwise
-	string error  = 2;
-}
-
-// RegistrationStatusResponse is sent by plugin to kubelet in response to RegistrationStatus RPC
-message RegistrationStatusResponse {
-}
-
-// InfoRequest is the empty request message from Kubelet
-message InfoRequest {
-}
-
-// Registration is the service advertised by the Plugins.
-service Registration {
-	rpc GetInfo(InfoRequest) returns (PluginInfo) {}
-	rpc NotifyRegistrationStatus(RegistrationStatus) returns (RegistrationStatusResponse) {}
-}

pkg/kubelet/apis/pluginregistration/v1beta1/constants.go

-/*
-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 pluginregistration
-
-const (
-	CSIPlugin    = "CSIPlugin"
-	DevicePlugin = "DevicePlugin"
-)

pkg/kubelet/cm/container_manager.go

@@ -20,18 +20,13 @@ import (
 	"time"
 
 	"k8s.io/apimachinery/pkg/util/sets"
+	"fmt"
 	// TODO: Migrate kubelet to either use its own internal objects or client library.
 	"k8s.io/api/core/v1"
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
 	"k8s.io/kubernetes/pkg/kubelet/config"
-	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	evictionapi "k8s.io/kubernetes/pkg/kubelet/eviction/api"
-	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
 	"k8s.io/kubernetes/pkg/kubelet/status"
-	"k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher"
-	schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
-
-	"fmt"
 	"strconv"
 	"strings"
 )
@@ -71,35 +66,14 @@ type ContainerManager interface {
 	// GetCapacity returns the amount of compute resources tracked by container manager available on the node.
 	GetCapacity() v1.ResourceList
 
-	// GetDevicePluginResourceCapacity returns the node capacity (amount of total device plugin resources),
-	// node allocatable (amount of total healthy resources reported by device plugin),
-	// and inactive device plugin resources previously registered on the node.
-	GetDevicePluginResourceCapacity() (v1.ResourceList, v1.ResourceList, []string)
-
 	// UpdateQOSCgroups performs housekeeping updates to ensure that the top
 	// level QoS containers have their desired state in a thread-safe way
 	UpdateQOSCgroups() error
 
-	// GetResources returns RunContainerOptions with devices, mounts, and env fields populated for
-	// extended resources required by container.
-	GetResources(pod *v1.Pod, container *v1.Container) (*kubecontainer.RunContainerOptions, error)
-
-	// UpdatePluginResources calls Allocate of device plugin handler for potential
-	// requests for device plugin resources, and returns an error if fails.
-	// Otherwise, it updates allocatableResource in nodeInfo if necessary,
-	// to make sure it is at least equal to the pod's requested capacity for
-	// any registered device plugin resource
-	UpdatePluginResources(*schedulercache.NodeInfo, *lifecycle.PodAdmitAttributes) error
-
 	InternalContainerLifecycle() InternalContainerLifecycle
 
 	// GetPodCgroupRoot returns the cgroup which contains all pods.
 	GetPodCgroupRoot() string
-
-	// GetPluginRegistrationHandler returns a plugin registration handler
-	// The pluginwatcher's Handlers allow to have a single module for handling
-	// registration.
-	GetPluginRegistrationHandler() pluginwatcher.PluginHandler
 }
 
 type NodeConfig struct {

pkg/kubelet/cm/container_manager_linux.go

@@ -45,15 +45,10 @@ import (
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
-	"k8s.io/kubernetes/pkg/kubelet/cm/devicemanager"
 	cmutil "k8s.io/kubernetes/pkg/kubelet/cm/util"
 	"k8s.io/kubernetes/pkg/kubelet/config"
-	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
-	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
 	"k8s.io/kubernetes/pkg/kubelet/qos"
 	"k8s.io/kubernetes/pkg/kubelet/status"
-	"k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher"
-	schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
 	utilfile "k8s.io/kubernetes/pkg/util/file"
 	"k8s.io/kubernetes/pkg/util/mount"
 	"k8s.io/kubernetes/pkg/util/oom"
@@ -129,8 +124,6 @@ type containerManagerImpl struct {
 	recorder record.EventRecorder
 	// Interface for QoS cgroup management
 	qosContainerManager QOSContainerManager
-	// Interface for exporting and allocating devices reported by device plugins.
-	deviceManager devicemanager.Manager
 	// Interface for CPU affinity management.
 	cpuManager cpumanager.Manager
 }
@@ -195,7 +188,7 @@ func validateSystemRequirements(mountUtil mount.Interface) (features, error) {
 // TODO(vmarmol): Add limits to the system containers.
 // Takes the absolute name of the specified containers.
 // Empty container name disables use of the specified container.
-func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.Interface, nodeConfig NodeConfig, failSwapOn bool, devicePluginEnabled bool, recorder record.EventRecorder) (ContainerManager, error) {
+func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.Interface, nodeConfig NodeConfig, failSwapOn bool, recorder record.EventRecorder) (ContainerManager, error) {
 	subsystems, err := GetCgroupSubsystems()
 	if err != nil {
 		return nil, fmt.Errorf("failed to get mounted cgroup subsystems: %v", err)
@@ -268,16 +261,6 @@ func NewContainerManager(mountUtil mount.Interface, cadvisorInterface cadvisor.I
 		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
 	if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.CPUManager) {
 		cm.cpuManager, err = cpumanager.NewManager(
@@ -597,40 +580,9 @@ func (cm *containerManagerImpl) Start(node *v1.Node,
 		}, 5*time.Minute, wait.NeverStop)
 	}
 
-	// Starts device manager.
-	if err := cm.deviceManager.Start(devicemanager.ActivePodsFunc(activePods), sourcesReady); err != nil {
-		return err
-	}
-
 	return nil
 }
 
-func (cm *containerManagerImpl) GetPluginRegistrationHandler() pluginwatcher.PluginHandler {
-	return cm.deviceManager.GetWatcherHandler()
-}
-
-// TODO: move the GetResources logic to PodContainerManager.
-func (cm *containerManagerImpl) GetResources(pod *v1.Pod, container *v1.Container) (*kubecontainer.RunContainerOptions, error) {
-	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
-}
-
-func (cm *containerManagerImpl) UpdatePluginResources(node *schedulercache.NodeInfo, attrs *lifecycle.PodAdmitAttributes) error {
-	return cm.deviceManager.Allocate(node, attrs)
-}
-
 func (cm *containerManagerImpl) SystemCgroupsLimit() v1.ResourceList {
 	cpuLimit := int64(0)
 
@@ -703,34 +655,6 @@ func getPidsForProcess(name, pidFile string) ([]int, error) {
 	return []int{}, err
 }
 
-// Ensures that the Docker daemon is in the desired container.
-// Temporarily export the function to be used by dockershim.
-// TODO(yujuhong): Move this function to dockershim once kubelet migrates to
-// dockershim as the default.
-func EnsureDockerInContainer(dockerAPIVersion *utilversion.Version, oomScoreAdj int, manager *fs.Manager) error {
-	type process struct{ name, file string }
-	dockerProcs := []process{{dockerProcessName, dockerPidFile}}
-	if dockerAPIVersion.AtLeast(containerdAPIVersion) {
-		dockerProcs = append(dockerProcs, process{containerdProcessName, containerdPidFile})
-	}
-	var errs []error
-	for _, proc := range dockerProcs {
-		pids, err := getPidsForProcess(proc.name, proc.file)
-		if err != nil {
-			errs = append(errs, fmt.Errorf("failed to get pids for %q: %v", proc.name, err))
-			continue
-		}
-
-		// Move if the pid is not already in the desired container.
-		for _, pid := range pids {
-			if err := ensureProcessInContainerWithOOMScore(pid, oomScoreAdj, manager); err != nil {
-				errs = append(errs, fmt.Errorf("errors moving %q pid: %v", proc.name, err))
-			}
-		}
-	}
-	return utilerrors.NewAggregate(errs)
-}
-
 func ensureProcessInContainerWithOOMScore(pid int, oomScoreAdj int, manager *fs.Manager) error {
 	if runningInHost, err := isProcessRunningInHost(pid); err != nil {
 		// Err on the side of caution. Avoid moving the docker daemon unless we are able to identify its context.
@@ -874,7 +798,3 @@ func isKernelPid(pid int) bool {
 func (cm *containerManagerImpl) GetCapacity() v1.ResourceList {
 	return cm.capacity
 }
-
-func (cm *containerManagerImpl) GetDevicePluginResourceCapacity() (v1.ResourceList, v1.ResourceList, []string) {
-	return cm.deviceManager.GetCapacity()
-}

pkg/kubelet/cm/container_manager_stub.go

@@ -24,10 +24,8 @@ import (
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
 	"k8s.io/kubernetes/pkg/kubelet/cm/cpumanager"
 	"k8s.io/kubernetes/pkg/kubelet/config"
-	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
 	"k8s.io/kubernetes/pkg/kubelet/status"
-	"k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher"
 	schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
 )
 
@@ -77,22 +75,10 @@ func (cm *containerManagerStub) GetCapacity() v1.ResourceList {
 	return c
 }
 
-func (cm *containerManagerStub) GetPluginRegistrationHandler() pluginwatcher.PluginHandler {
-	return nil
-}
-
-func (cm *containerManagerStub) GetDevicePluginResourceCapacity() (v1.ResourceList, v1.ResourceList, []string) {
-	return nil, nil, []string{}
-}
-
 func (cm *containerManagerStub) NewPodContainerManager() PodContainerManager {
 	return &podContainerManagerStub{}
 }
 
-func (cm *containerManagerStub) GetResources(pod *v1.Pod, container *v1.Container) (*kubecontainer.RunContainerOptions, error) {
-	return &kubecontainer.RunContainerOptions{}, nil
-}
-
 func (cm *containerManagerStub) UpdatePluginResources(*schedulercache.NodeInfo, *lifecycle.PodAdmitAttributes) error {
 	return nil
 }

pkg/kubelet/cm/devicemanager/BUILD

-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/apis/pluginregistration/v1alpha1:go_default_library",
-        "//pkg/kubelet/checkpointmanager:go_default_library",
-        "//pkg/kubelet/checkpointmanager/errors:go_default_library",
-        "//pkg/kubelet/cm/devicemanager/checkpoint: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/pluginwatcher:go_default_library",
-        "//pkg/scheduler/cache:go_default_library",
-        "//staging/src/k8s.io/api/core/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
-        "//vendor/github.com/golang/glog:go_default_library",
-        "//vendor/google.golang.org/grpc: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/apis/pluginregistration/v1alpha1:go_default_library",
-        "//pkg/kubelet/checkpointmanager:go_default_library",
-        "//pkg/kubelet/lifecycle:go_default_library",
-        "//pkg/kubelet/util/pluginwatcher:go_default_library",
-        "//pkg/scheduler/cache:go_default_library",
-        "//staging/src/k8s.io/api/core/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
-        "//staging/src/k8s.io/apimachinery/pkg/util/uuid:go_default_library",
-        "//vendor/github.com/stretchr/testify/assert:go_default_library",
-        "//vendor/github.com/stretchr/testify/require:go_default_library",
-    ],
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [
-        ":package-srcs",
-        "//pkg/kubelet/cm/devicemanager/checkpoint:all-srcs",
-    ],
-    tags = ["automanaged"],
-    visibility = ["//visibility:public"],
-)

pkg/kubelet/cm/devicemanager/OWNERS

-approvers:
-- jiayingz
-- vishh
-reviewers:
-- mindprince
-- RenaudWasTaken
-- vikaschoudhary16

pkg/kubelet/cm/devicemanager/checkpoint/BUILD

-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-
-go_library(
-    name = "go_default_library",
-    srcs = ["checkpoint.go"],
-    importpath = "k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/checkpoint",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/kubelet/checkpointmanager:go_default_library",
-        "//pkg/kubelet/checkpointmanager/checksum: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/checkpoint/checkpoint.go

-/*
-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 (
-	"encoding/json"
-
-	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager"
-	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager/checksum"
-)
-
-type DeviceManagerCheckpoint interface {
-	checkpointmanager.Checkpoint
-	GetData() ([]PodDevicesEntry, map[string][]string)
-}
-
-type PodDevicesEntry struct {
-	PodUID        string
-	ContainerName string
-	ResourceName  string
-	DeviceIDs     []string
-	AllocResp     []byte
-}
-
-// checkpointData struct is used to store pod to device allocation information
-// in a checkpoint file.
-// TODO: add version control when we need to change checkpoint format.
-type checkpointData struct {
-	PodDeviceEntries  []PodDevicesEntry
-	RegisteredDevices map[string][]string
-}
-
-type Data struct {
-	Data     checkpointData
-	Checksum checksum.Checksum
-}
-
-// NewDeviceManagerCheckpoint returns an instance of Checkpoint
-func New(devEntries []PodDevicesEntry,
-	devices map[string][]string) DeviceManagerCheckpoint {
-	return &Data{
-		Data: checkpointData{
-			PodDeviceEntries:  devEntries,
-			RegisteredDevices: devices,
-		},
-	}
-}
-
-// MarshalCheckpoint returns marshalled data
-func (cp *Data) MarshalCheckpoint() ([]byte, error) {
-	cp.Checksum = checksum.New(cp.Data)
-	return json.Marshal(*cp)
-}
-
-// UnmarshalCheckpoint returns unmarshalled data
-func (cp *Data) UnmarshalCheckpoint(blob []byte) error {
-	return json.Unmarshal(blob, cp)
-}
-
-// VerifyChecksum verifies that passed checksum is same as calculated checksum
-func (cp *Data) VerifyChecksum() error {
-	return cp.Checksum.Verify(cp.Data)
-}
-
-func (cp *Data) GetData() ([]PodDevicesEntry, map[string][]string) {
-	return cp.Data.PodDeviceEntries, cp.Data.RegisteredDevices
-}

pkg/kubelet/cm/devicemanager/device_plugin_stub.go

-/*
-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 (
-	"context"
-	"log"
-	"net"
-	"os"
-	"path"
-	"sync"
-	"time"
-
-	"google.golang.org/grpc"
-
-	pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1beta1"
-	watcherapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1"
-)
-
-// Stub implementation for DevicePlugin.
-type Stub struct {
-	devs                  []*pluginapi.Device
-	socket                string
-	resourceName          string
-	preStartContainerFlag bool
-
-	stop   chan interface{}
-	wg     sync.WaitGroup
-	update chan []*pluginapi.Device
-
-	server *grpc.Server
-
-	// allocFunc is used for handling allocation request
-	allocFunc stubAllocFunc
-
-	registrationStatus chan watcherapi.RegistrationStatus // for testing
-	endpoint           string                             // for testing
-
-}
-
-// 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, name string, preStartContainerFlag bool) *Stub {
-	return &Stub{
-		devs:                  devs,
-		socket:                socket,
-		resourceName:          name,
-		preStartContainerFlag: preStartContainerFlag,
-
-		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. Can only
-// be called once.
-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.wg.Add(1)
-	m.server = grpc.NewServer([]grpc.ServerOption{}...)
-	pluginapi.RegisterDevicePluginServer(m.server, m)
-	watcherapi.RegisterRegistrationServer(m.server, m)
-
-	go func() {
-		defer m.wg.Done()
-		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. Can be called without a prior Start
-// and more than once. Not safe to be called concurrently by different
-// goroutines!
-func (m *Stub) Stop() error {
-	if m.server == nil {
-		return nil
-	}
-	m.server.Stop()
-	m.wg.Wait()
-	m.server = nil
-	close(m.stop) // This prevents re-starting the server.
-
-	return m.cleanup()
-}
-
-// GetInfo is the RPC which return pluginInfo
-func (m *Stub) GetInfo(ctx context.Context, req *watcherapi.InfoRequest) (*watcherapi.PluginInfo, error) {
-	log.Println("GetInfo")
-	return &watcherapi.PluginInfo{
-		Type:              watcherapi.DevicePlugin,
-		Name:              m.resourceName,
-		Endpoint:          m.endpoint,
-		SupportedVersions: []string{pluginapi.Version}}, nil
-}
-
-// NotifyRegistrationStatus receives the registration notification from watcher
-func (m *Stub) NotifyRegistrationStatus(ctx context.Context, status *watcherapi.RegistrationStatus) (*watcherapi.RegistrationStatusResponse, error) {
-	if m.registrationStatus != nil {
-		m.registrationStatus <- *status
-	}
-	if !status.PluginRegistered {
-		log.Println("Registration failed: ", status.Error)
-	}
-	return &watcherapi.RegistrationStatusResponse{}, nil
-}
-
-// Register registers the device plugin for the given resourceName with Kubelet.
-func (m *Stub) Register(kubeletEndpoint, resourceName string, pluginSockDir string) error {
-	if pluginSockDir != "" {
-		if _, err := os.Stat(pluginSockDir + "DEPRECATION"); err == nil {
-			log.Println("Deprecation file found. Skip registration.")
-			return nil
-		}
-	}
-	log.Println("Deprecation file not found. Invoke registration")
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-
-	conn, err := grpc.DialContext(ctx, kubeletEndpoint, grpc.WithInsecure(), grpc.WithBlock(),
-		grpc.WithDialer(func(addr string, timeout time.Duration) (net.Conn, error) {
-			return net.DialTimeout("unix", addr, timeout)
-		}))
-	if err != nil {
-		return err
-	}
-	defer conn.Close()
-	client := pluginapi.NewRegistrationClient(conn)
-	reqt := &pluginapi.RegisterRequest{
-		Version:      pluginapi.Version,
-		Endpoint:     path.Base(m.socket),
-		ResourceName: resourceName,
-		Options:      &pluginapi.DevicePluginOptions{PreStartRequired: m.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{PreStartRequired: m.preStartContainerFlag}, 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

-/*
-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 (
-	"context"
-	"fmt"
-	"net"
-	"sync"
-	"time"
-
-	"github.com/golang/glog"
-	"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)
-	callback(resourceName string, devices []pluginapi.Device)
-	isStopped() bool
-	stopGracePeriodExpired() bool
-}
-
-type endpointImpl struct {
-	client     pluginapi.DevicePluginClient
-	clientConn *grpc.ClientConn
-
-	socketPath   string
-	resourceName string
-	stopTime     time.Time
-
-	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, 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,
-
-		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) *endpointImpl {
-	return &endpointImpl{
-		resourceName: resourceName,
-		stopTime:     time.Now(),
-	}
-}
-
-func (e *endpointImpl) callback(resourceName string, devices []pluginapi.Device) {
-	e.cb(resourceName, devices)
-}
-
-// 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.
-// 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
-	}
-
-	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)
-
-		var newDevs []pluginapi.Device
-		for _, d := range devs {
-			newDevs = append(newDevs, *d)
-		}
-
-		e.callback(e.resourceName, newDevs)
-	}
-}
-
-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) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	defer cancel()
-
-	c, err := grpc.DialContext(ctx, unixSocketPath, grpc.WithInsecure(), grpc.WithBlock(),
-		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

-/*
-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, d []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, devices []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, devices, 3)
-			require.Equal(t, devices[0].ID, devs[0].ID)
-			require.Equal(t, devices[1].ID, devs[1].ID)
-			require.Equal(t, devices[2].ID, devs[2].ID)
-			require.Equal(t, devices[0].Health, devs[0].Health)
-			require.Equal(t, devices[1].Health, devs[1].Health)
-			require.Equal(t, devices[2].Health, devs[2].Health)
-		}
-
-		// Check plugin update
-		if callbackCount == 1 {
-			require.Len(t, devices, 3)
-			require.Equal(t, devices[0].ID, updated[0].ID)
-			require.Equal(t, devices[1].ID, updated[1].ID)
-			require.Equal(t, devices[2].ID, updated[2].ID)
-			require.Equal(t, devices[0].Health, updated[0].Health)
-			require.Equal(t, devices[1].Health, updated[1].Health)
-			require.Equal(t, devices[2].Health, updated[2].Health)
-		}
-
-		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
-
-	require.Equal(t, callbackCount, 2)
-}
-
-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, d []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 esetup(t *testing.T, devs []*pluginapi.Device, socket, resourceName string, callback monitorCallback) (*Stub, *endpointImpl) {
-	p := NewDevicePluginStub(devs, socket, resourceName, false)
-
-	err := p.Start()
-	require.NoError(t, err)
-
-	e, err := newEndpointImpl(socket, resourceName, 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

-/*
-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"
-	"k8s.io/kubernetes/pkg/kubelet/config"
-	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
-	"k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher"
-	schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
-)
-
-// 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
-}
-
-// 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{}
-}
-
-// GetWatcherHandler returns plugin watcher interface
-func (h *ManagerStub) GetWatcherHandler() pluginwatcher.PluginHandler {
-	return nil
-}

pkg/kubelet/cm/devicemanager/pod_devices.go

-/*
-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"
-	"k8s.io/kubernetes/pkg/kubelet/cm/devicemanager/checkpoint"
-	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
-}
-
-// Turns podDevices to checkpointData.
-func (pdev podDevices) toCheckpointData() []checkpoint.PodDevicesEntry {
-	var data []checkpoint.PodDevicesEntry
-	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, checkpoint.PodDevicesEntry{
-					PodUID:        podUID,
-					ContainerName: conName,
-					ResourceName:  resource,
-					DeviceIDs:     devIds,
-					AllocResp:     allocResp})
-			}
-		}
-	}
-	return data
-}
-
-// Populates podDevices from the passed in checkpointData.
-func (pdev podDevices) fromCheckpointData(data []checkpoint.PodDevicesEntry) {
-	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

-/*
-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"
-	"k8s.io/kubernetes/pkg/kubelet/config"
-	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
-	"k8s.io/kubernetes/pkg/kubelet/lifecycle"
-	watcher "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher"
-	schedulercache "k8s.io/kubernetes/pkg/scheduler/cache"
-)
-
-// 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
-
-	// 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)
-	GetWatcherHandler() watcher.PluginHandler
-}
-
-// 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 version is %q"
-	// errInvalidResourceName is the error raised when a device plugin is registering
-	// itself with an invalid ResourceName
-	errInvalidResourceName = "the ResourceName %q is invalid"
-	// 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:"
-	// 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
-
-// kubeletDeviceManagerCheckpoint is the file name of device plugin checkpoint
-const kubeletDeviceManagerCheckpoint = "kubelet_internal_checkpoint"

pkg/kubelet/kubelet.go

@@ -56,7 +56,6 @@ import (
 	"k8s.io/kubernetes/pkg/features"
 	kubeletconfiginternal "k8s.io/kubernetes/pkg/kubelet/apis/config"
 	internalapi "k8s.io/kubernetes/pkg/kubelet/apis/cri"
-	pluginwatcherapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1"
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/checkpointmanager"
 	"k8s.io/kubernetes/pkg/kubelet/cm"
@@ -90,7 +89,6 @@ import (
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/kubelet/util/format"
 	"k8s.io/kubernetes/pkg/kubelet/util/manager"
-	"k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher"
 	"k8s.io/kubernetes/pkg/kubelet/util/queue"
 	"k8s.io/kubernetes/pkg/kubelet/util/sliceutils"
 	"k8s.io/kubernetes/pkg/kubelet/volumemanager"
@@ -103,7 +101,6 @@ import (
 	nodeutil "k8s.io/kubernetes/pkg/util/node"
 	"k8s.io/kubernetes/pkg/util/oom"
 	"k8s.io/kubernetes/pkg/volume"
-	"k8s.io/kubernetes/pkg/volume/csi"
 	utilexec "k8s.io/utils/exec"
 )
 
@@ -506,7 +503,6 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		experimentalHostUserNamespaceDefaulting: utilfeature.DefaultFeatureGate.Enabled(features.ExperimentalHostUserNamespaceDefaultingGate),
 		keepTerminatedPodVolumes:                keepTerminatedPodVolumes,
 		nodeStatusMaxImages:                     nodeStatusMaxImages,
-		enablePluginsWatcher:                    utilfeature.DefaultFeatureGate.Enabled(features.KubeletPluginsWatcher),
 	}
 
 	var secretManager secret.Manager
@@ -684,9 +680,6 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	if err != nil {
 		return nil, err
 	}
-	if klet.enablePluginsWatcher {
-		klet.pluginWatcher = pluginwatcher.NewWatcher(klet.getPluginsDir())
-	}
 
 	// If the experimentalMounterPathFlag is set, we do not want to
 	// check node capabilities since the mount path is not the default
@@ -752,7 +745,7 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 	klet.AddPodSyncHandler(activeDeadlineHandler)
 
 	criticalPodAdmissionHandler := preemption.NewCriticalPodAdmissionHandler(klet.GetActivePods, killPodNow(klet.podWorkers, kubeDeps.Recorder), kubeDeps.Recorder)
-	klet.admitHandlers.AddPodAdmitHandler(lifecycle.NewPredicateAdmitHandler(klet.getNodeAnyWay, criticalPodAdmissionHandler, klet.containerManager.UpdatePluginResources))
+	klet.admitHandlers.AddPodAdmitHandler(lifecycle.NewPredicateAdmitHandler(klet.getNodeAnyWay, criticalPodAdmissionHandler))
 	// apply functional Option's
 	for _, opt := range kubeDeps.Options {
 		opt(klet)
@@ -1069,16 +1062,8 @@ type Kubelet struct {
 	// This can be useful for debugging volume related issues.
 	keepTerminatedPodVolumes bool // DEPRECATED
 
-	// pluginwatcher is a utility for Kubelet to register different types of node-level plugins
-	// such as device plugins or CSI plugins. It discovers plugins by monitoring inotify events under the
-	// directory returned by kubelet.getPluginsDir()
-	pluginWatcher *pluginwatcher.Watcher
-
 	// This flag sets a maximum number of images to report in the node status.
 	nodeStatusMaxImages int32
-
-	//  This flag indicates that kubelet should start plugin watcher utility server for discovering Kubelet plugins
-	enablePluginsWatcher bool
 }
 
 // setupDataDirs creates:
@@ -1210,18 +1195,6 @@ func (kl *Kubelet) initializeRuntimeDependentModules() {
 	// container log manager must start after container runtime is up to retrieve information from container runtime
 	// and inform container to reopen log file after log rotation.
 	kl.containerLogManager.Start()
-	if kl.enablePluginsWatcher {
-		// Adding Registration Callback function for CSI Driver
-		kl.pluginWatcher.AddHandler("CSIPlugin", pluginwatcher.PluginHandler(csi.PluginHandler))
-		// Adding Registration Callback function for Device Manager
-		kl.pluginWatcher.AddHandler(pluginwatcherapi.DevicePlugin, kl.containerManager.GetPluginRegistrationHandler())
-		// Start the plugin watcher
-		glog.V(4).Infof("starting watcher")
-		if err := kl.pluginWatcher.Start(); err != nil {
-			kl.recorder.Eventf(kl.nodeRef, v1.EventTypeWarning, events.KubeletSetupFailed, err.Error())
-			glog.Fatalf("failed to start Plugin Watcher. err: %v", err)
-		}
-	}
 }
 
 // Run starts the kubelet reacting to config updates

pkg/kubelet/kubelet_node_status.go

@@ -435,7 +435,7 @@ func (kl *Kubelet) defaultNodeStatusFuncs() []func(*v1.Node) error {
 	setters = append(setters,
 		nodestatus.NodeAddress(kl.nodeIP, kl.nodeIPValidator, kl.hostname, kl.hostnameOverridden, false, nodeAddressesFunc),
 		nodestatus.MachineInfo(string(kl.nodeName), kl.maxPods, kl.podsPerCore, kl.GetCachedMachineInfo, kl.containerManager.GetCapacity,
-			kl.containerManager.GetDevicePluginResourceCapacity, kl.containerManager.GetNodeAllocatableReservation, kl.recordEvent),
+			kl.containerManager.GetNodeAllocatableReservation, kl.recordEvent),
 		nodestatus.VersionInfo(kl.cadvisor.VersionInfo, kl.containerRuntime.Type, kl.containerRuntime.Version),
 		nodestatus.DaemonEndpoints(kl.daemonEndpoints),
 		nodestatus.Images(kl.nodeStatusMaxImages, kl.imageManager.GetImageList),

pkg/kubelet/kubelet_pods.go

@@ -423,10 +423,7 @@ func (kl *Kubelet) GetPodCgroupParent(pod *v1.Pod) string {
 // GenerateRunContainerOptions generates the RunContainerOptions, which can be used by
 // the container runtime to set parameters for launching a container.
 func (kl *Kubelet) GenerateRunContainerOptions(pod *v1.Pod, container *v1.Container, podIP string) (*kubecontainer.RunContainerOptions, func(), error) {
-	opts, err := kl.containerManager.GetResources(pod, container)
-	if err != nil {
-		return nil, nil, err
-	}
+	opts := &kubecontainer.RunContainerOptions{}
 
 	hostname, hostDomainName, err := kl.GeneratePodHostNameAndDomain(pod)
 	if err != nil {

pkg/kubelet/lifecycle/predicate.go

@@ -31,8 +31,6 @@ import (
 
 type getNodeAnyWayFuncType func() (*v1.Node, error)
 
-type pluginResourceUpdateFuncType func(*schedulercache.NodeInfo, *PodAdmitAttributes) error
-
 // AdmissionFailureHandler is an interface which defines how to deal with a failure to admit a pod.
 // This allows for the graceful handling of pod admission failure.
 type AdmissionFailureHandler interface {
@@ -41,16 +39,14 @@ type AdmissionFailureHandler interface {
 
 type predicateAdmitHandler struct {
 	getNodeAnyWayFunc        getNodeAnyWayFuncType
-	pluginResourceUpdateFunc pluginResourceUpdateFuncType
 	admissionFailureHandler  AdmissionFailureHandler
 }
 
 var _ PodAdmitHandler = &predicateAdmitHandler{}
 
-func NewPredicateAdmitHandler(getNodeAnyWayFunc getNodeAnyWayFuncType, admissionFailureHandler AdmissionFailureHandler, pluginResourceUpdateFunc pluginResourceUpdateFuncType) *predicateAdmitHandler {
+func NewPredicateAdmitHandler(getNodeAnyWayFunc getNodeAnyWayFuncType, admissionFailureHandler AdmissionFailureHandler) *predicateAdmitHandler {
 	return &predicateAdmitHandler{
 		getNodeAnyWayFunc,
-		pluginResourceUpdateFunc,
 		admissionFailureHandler,
 	}
 }
@@ -69,16 +65,6 @@ func (w *predicateAdmitHandler) Admit(attrs *PodAdmitAttributes) PodAdmitResult
 	pods := attrs.OtherPods
 	nodeInfo := schedulercache.NewNodeInfo(pods...)
 	nodeInfo.SetNode(node)
-	// ensure the node has enough plugin resources for that required in pods
-	if err = w.pluginResourceUpdateFunc(nodeInfo, attrs); err != nil {
-		message := fmt.Sprintf("Update plugin resources failed due to %v, which is unexpected.", err)
-		glog.Warningf("Failed to admit pod %v - %s", format.Pod(admitPod), message)
-		return PodAdmitResult{
-			Admit:   false,
-			Reason:  "UnexpectedAdmissionError",
-			Message: message,
-		}
-	}
 
 	// Remove the requests of the extended resources that are missing in the
 	// node info. This is required to support cluster-level resources, which

pkg/kubelet/nodestatus/setters.go

@@ -131,7 +131,6 @@ func MachineInfo(nodeName string,
 	podsPerCore int,
 	machineInfoFunc func() (*cadvisorapiv1.MachineInfo, error), // typically Kubelet.GetCachedMachineInfo
 	capacityFunc func() v1.ResourceList, // typically Kubelet.containerManager.GetCapacity
-	devicePluginResourceCapacityFunc func() (v1.ResourceList, v1.ResourceList, []string), // typically Kubelet.containerManager.GetDevicePluginResourceCapacity
 	nodeAllocatableReservationFunc func() v1.ResourceList, // typically Kubelet.containerManager.GetNodeAllocatableReservation
 	recordEventFunc func(eventType, event, message string), // typically Kubelet.recordEvent
 ) Setter {
@@ -190,7 +189,6 @@ func MachineInfo(nodeName string,
 				}
 			}
 
-			devicePluginCapacity, devicePluginAllocatable, removedDevicePlugins = devicePluginResourceCapacityFunc()
 			if devicePluginCapacity != nil {
 				for k, v := range devicePluginCapacity {
 					if old, ok := node.Status.Capacity[k]; !ok || old.Value() != v.Value() {

pkg/kubelet/util/pluginwatcher/BUILD

-load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
-
-go_library(
-    name = "go_default_library",
-    srcs = [
-        "example_handler.go",
-        "example_plugin.go",
-        "plugin_watcher.go",
-        "types.go",
-    ],
-    importpath = "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//pkg/kubelet/apis/pluginregistration/v1alpha1:go_default_library",
-        "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1:go_default_library",
-        "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2:go_default_library",
-        "//pkg/util/filesystem:go_default_library",
-        "//vendor/github.com/fsnotify/fsnotify:go_default_library",
-        "//vendor/github.com/golang/glog:go_default_library",
-        "//vendor/github.com/pkg/errors:go_default_library",
-        "//vendor/golang.org/x/net/context:go_default_library",
-        "//vendor/google.golang.org/grpc:go_default_library",
-    ],
-)
-
-go_test(
-    name = "go_default_test",
-    srcs = ["plugin_watcher_test.go"],
-    embed = [":go_default_library"],
-    deps = [
-        "//pkg/kubelet/apis/pluginregistration/v1alpha1:go_default_library",
-        "//vendor/github.com/stretchr/testify/require:go_default_library",
-    ],
-)
-
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [
-        ":package-srcs",
-        "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1:all-srcs",
-        "//pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2:all-srcs",
-    ],
-    tags = ["automanaged"],
-    visibility = ["//visibility:public"],
-)

pkg/kubelet/util/pluginwatcher/README

-This folder contains a utility, pluginwatcher, for Kubelet to register
-different types of node-level plugins such as device plugins or CSI plugins.
-It discovers plugins by monitoring inotify events under the directory returned by
-kubelet.getPluginsDir(). Lets refer this directory as PluginsSockDir.
-For any discovered plugin, pluginwatcher issues Registration.GetInfo grpc call
-to get plugin type, name and supported service API versions. For any registered plugin type,
-pluginwatcher calls the registered callback function with the received plugin
-name, supported service API versions, and the full socket path. The Kubelet
-component that receives this callback can acknowledge or reject the plugin
-according to its own logic, and use the socket path to establish its service
-communication with any API version supported by the plugin.
-
-Here are the general rules that Kubelet plugin developers should follow:
-- Run as 'root' user. Currently creating socket under PluginsSockDir, a root owned directory, requires
-  plugin process to be running as 'root'.
-
-- Implements the Registration service specified in
-  pkg/kubelet/apis/pluginregistration/v*/api.proto.
-
-- The plugin name sent during Registration.GetInfo grpc should be unique
-  for the given plugin type (CSIPlugin or DevicePlugin).
-
-- The socket path needs to be unique within one directory, in normal case, 
-  each plugin type has its own sub directory, but the design does support socket file
-  under any sub directory of PluginSockDir.
- 
-- A plugin should clean up its own socket upon exiting or when a new instance
-  comes up. A plugin should NOT remove any sockets belonging to other plugins.
-
-- A plugin should make sure it has service ready for any supported service API
-  version listed in the PluginInfo.
-
-- For an example plugin implementation, take a look at example_plugin.go
-  included in this directory.

pkg/kubelet/util/pluginwatcher/example_handler.go

-/*
-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 pluginwatcher
-
-import (
-	"errors"
-	"fmt"
-	"reflect"
-	"sync"
-	"time"
-
-	"github.com/golang/glog"
-	"golang.org/x/net/context"
-
-	v1beta1 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1"
-	v1beta2 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2"
-)
-
-type exampleHandler struct {
-	SupportedVersions []string
-	ExpectedNames     map[string]int
-
-	eventChans map[string]chan examplePluginEvent // map[pluginName]eventChan
-
-	m     sync.Mutex
-	count int
-}
-
-type examplePluginEvent int
-
-const (
-	exampleEventValidate   examplePluginEvent = 0
-	exampleEventRegister   examplePluginEvent = 1
-	exampleEventDeRegister examplePluginEvent = 2
-	exampleEventError      examplePluginEvent = 3
-)
-
-// NewExampleHandler provide a example handler
-func NewExampleHandler(supportedVersions []string) *exampleHandler {
-	return &exampleHandler{
-		SupportedVersions: supportedVersions,
-		ExpectedNames:     make(map[string]int),
-
-		eventChans: make(map[string]chan examplePluginEvent),
-	}
-}
-
-func (p *exampleHandler) ValidatePlugin(pluginName string, endpoint string, versions []string) error {
-	p.SendEvent(pluginName, exampleEventValidate)
-
-	n, ok := p.DecreasePluginCount(pluginName)
-	if !ok && n > 0 {
-		return fmt.Errorf("pluginName('%s') wasn't expected (count is %d)", pluginName, n)
-	}
-
-	if !reflect.DeepEqual(versions, p.SupportedVersions) {
-		return fmt.Errorf("versions('%v') != supported versions('%v')", versions, p.SupportedVersions)
-	}
-
-	// this handler expects non-empty endpoint as an example
-	if len(endpoint) == 0 {
-		return errors.New("expecting non empty endpoint")
-	}
-
-	return nil
-}
-
-func (p *exampleHandler) RegisterPlugin(pluginName, endpoint string) error {
-	p.SendEvent(pluginName, exampleEventRegister)
-
-	// Verifies the grpcServer is ready to serve services.
-	_, conn, err := dial(endpoint, time.Second)
-	if err != nil {
-		return fmt.Errorf("Failed dialing endpoint (%s): %v", endpoint, err)
-	}
-	defer conn.Close()
-
-	// The plugin handler should be able to use any listed service API version.
-	v1beta1Client := v1beta1.NewExampleClient(conn)
-	v1beta2Client := v1beta2.NewExampleClient(conn)
-
-	// Tests v1beta1 GetExampleInfo
-	_, err = v1beta1Client.GetExampleInfo(context.Background(), &v1beta1.ExampleRequest{})
-	if err != nil {
-		return fmt.Errorf("Failed GetExampleInfo for v1beta2Client(%s): %v", endpoint, err)
-	}
-
-	// Tests v1beta1 GetExampleInfo
-	_, err = v1beta2Client.GetExampleInfo(context.Background(), &v1beta2.ExampleRequest{})
-	if err != nil {
-		return fmt.Errorf("Failed GetExampleInfo for v1beta2Client(%s): %v", endpoint, err)
-	}
-
-	return nil
-}
-
-func (p *exampleHandler) DeRegisterPlugin(pluginName string) {
-	p.SendEvent(pluginName, exampleEventDeRegister)
-}
-
-func (p *exampleHandler) EventChan(pluginName string) chan examplePluginEvent {
-	return p.eventChans[pluginName]
-}
-
-func (p *exampleHandler) SendEvent(pluginName string, event examplePluginEvent) {
-	glog.V(2).Infof("Sending %v for plugin %s over chan %v", event, pluginName, p.eventChans[pluginName])
-	p.eventChans[pluginName] <- event
-}
-
-func (p *exampleHandler) AddPluginName(pluginName string) {
-	p.m.Lock()
-	defer p.m.Unlock()
-
-	v, ok := p.ExpectedNames[pluginName]
-	if !ok {
-		p.eventChans[pluginName] = make(chan examplePluginEvent)
-		v = 1
-	}
-
-	p.ExpectedNames[pluginName] = v
-}
-
-func (p *exampleHandler) DecreasePluginCount(pluginName string) (old int, ok bool) {
-	p.m.Lock()
-	defer p.m.Unlock()
-
-	v, ok := p.ExpectedNames[pluginName]
-	if !ok {
-		v = -1
-	}
-
-	return v, ok
-}

pkg/kubelet/util/pluginwatcher/example_plugin.go

-/*
-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 pluginwatcher
-
-import (
-	"errors"
-	"fmt"
-	"net"
-	"os"
-	"sync"
-	"time"
-
-	"github.com/golang/glog"
-	"golang.org/x/net/context"
-	"google.golang.org/grpc"
-
-	registerapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1"
-	v1beta1 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1"
-	v1beta2 "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2"
-)
-
-// examplePlugin is a sample plugin to work with plugin watcher
-type examplePlugin struct {
-	grpcServer         *grpc.Server
-	wg                 sync.WaitGroup
-	registrationStatus chan registerapi.RegistrationStatus // for testing
-	endpoint           string                              // for testing
-	pluginName         string
-	pluginType         string
-	versions           []string
-}
-
-type pluginServiceV1Beta1 struct {
-	server *examplePlugin
-}
-
-func (s *pluginServiceV1Beta1) GetExampleInfo(ctx context.Context, rqt *v1beta1.ExampleRequest) (*v1beta1.ExampleResponse, error) {
-	glog.Infof("GetExampleInfo v1beta1field: %s", rqt.V1Beta1Field)
-	return &v1beta1.ExampleResponse{}, nil
-}
-
-func (s *pluginServiceV1Beta1) RegisterService() {
-	v1beta1.RegisterExampleServer(s.server.grpcServer, s)
-}
-
-type pluginServiceV1Beta2 struct {
-	server *examplePlugin
-}
-
-func (s *pluginServiceV1Beta2) GetExampleInfo(ctx context.Context, rqt *v1beta2.ExampleRequest) (*v1beta2.ExampleResponse, error) {
-	glog.Infof("GetExampleInfo v1beta2_field: %s", rqt.V1Beta2Field)
-	return &v1beta2.ExampleResponse{}, nil
-}
-
-func (s *pluginServiceV1Beta2) RegisterService() {
-	v1beta2.RegisterExampleServer(s.server.grpcServer, s)
-}
-
-// NewExamplePlugin returns an initialized examplePlugin instance
-func NewExamplePlugin() *examplePlugin {
-	return &examplePlugin{}
-}
-
-// NewTestExamplePlugin returns an initialized examplePlugin instance for testing
-func NewTestExamplePlugin(pluginName string, pluginType string, endpoint string, advertisedVersions ...string) *examplePlugin {
-	return &examplePlugin{
-		pluginName:         pluginName,
-		pluginType:         pluginType,
-		endpoint:           endpoint,
-		versions:           advertisedVersions,
-		registrationStatus: make(chan registerapi.RegistrationStatus),
-	}
-}
-
-// GetInfo is the RPC invoked by plugin watcher
-func (e *examplePlugin) GetInfo(ctx context.Context, req *registerapi.InfoRequest) (*registerapi.PluginInfo, error) {
-	return &registerapi.PluginInfo{
-		Type:              e.pluginType,
-		Name:              e.pluginName,
-		Endpoint:          e.endpoint,
-		SupportedVersions: e.versions,
-	}, nil
-}
-
-func (e *examplePlugin) NotifyRegistrationStatus(ctx context.Context, status *registerapi.RegistrationStatus) (*registerapi.RegistrationStatusResponse, error) {
-	glog.Errorf("Registration is: %v\n", status)
-
-	if e.registrationStatus != nil {
-		e.registrationStatus <- *status
-	}
-
-	return &registerapi.RegistrationStatusResponse{}, nil
-}
-
-// Serve starts a pluginwatcher server and one or more of the plugin services
-func (e *examplePlugin) Serve(services ...string) error {
-	glog.Infof("starting example server at: %s\n", e.endpoint)
-	lis, err := net.Listen("unix", e.endpoint)
-	if err != nil {
-		return err
-	}
-
-	glog.Infof("example server started at: %s\n", e.endpoint)
-	e.grpcServer = grpc.NewServer()
-
-	// Registers kubelet plugin watcher api.
-	registerapi.RegisterRegistrationServer(e.grpcServer, e)
-
-	for _, service := range services {
-		switch service {
-		case "v1beta1":
-			v1beta1 := &pluginServiceV1Beta1{server: e}
-			v1beta1.RegisterService()
-			break
-		case "v1beta2":
-			v1beta2 := &pluginServiceV1Beta2{server: e}
-			v1beta2.RegisterService()
-			break
-		default:
-			return fmt.Errorf("Unsupported service: '%s'", service)
-		}
-	}
-
-	// Starts service
-	e.wg.Add(1)
-	go func() {
-		defer e.wg.Done()
-		// Blocking call to accept incoming connections.
-		if err := e.grpcServer.Serve(lis); err != nil {
-			glog.Errorf("example server stopped serving: %v", err)
-		}
-	}()
-
-	return nil
-}
-
-func (e *examplePlugin) Stop() error {
-	glog.Infof("Stopping example server at: %s\n", e.endpoint)
-
-	e.grpcServer.Stop()
-	c := make(chan struct{})
-	go func() {
-		defer close(c)
-		e.wg.Wait()
-	}()
-
-	select {
-	case <-c:
-		break
-	case <-time.After(time.Second):
-		return errors.New("Timed out on waiting for stop completion")
-	}
-
-	if err := os.Remove(e.endpoint); err != nil && !os.IsNotExist(err) {
-		return err
-	}
-
-	return nil
-}

pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/BUILD

-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-
-go_library(
-    name = "go_default_library",
-    srcs = ["api.pb.go"],
-    importpath = "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//vendor/github.com/gogo/protobuf/gogoproto:go_default_library",
-        "//vendor/github.com/gogo/protobuf/proto: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"],
-    visibility = ["//visibility:public"],
-)

pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta1/api.proto

-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;
-
-message ExampleRequest {
-	string request = 1;
-	string v1beta1_field = 2;
-}
-
-message ExampleResponse {
-	string error  = 1;
-}
-
-// Example is a simple example service for general reference on the recommended
-// kubelet plugin model and plugin watcher testing.
-service Example {
-	rpc GetExampleInfo(ExampleRequest) returns (ExampleResponse) {}
-}

pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/BUILD

-load("@io_bazel_rules_go//go:def.bzl", "go_library")
-
-go_library(
-    name = "go_default_library",
-    srcs = ["api.pb.go"],
-    importpath = "k8s.io/kubernetes/pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2",
-    visibility = ["//visibility:public"],
-    deps = [
-        "//vendor/github.com/gogo/protobuf/gogoproto:go_default_library",
-        "//vendor/github.com/gogo/protobuf/proto: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"],
-    visibility = ["//visibility:public"],
-)

pkg/kubelet/util/pluginwatcher/example_plugin_apis/v1beta2/api.proto

-syntax = 'proto3';
-
-package v1beta2;
-
-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;
-
-// Renames a field from v1beta1 ExampleRequest.
-message ExampleRequest {
-	string request = 1;
-	string v1beta2_field = 2;
-}
-
-message ExampleResponse {
-	string error  = 1;
-}
-
-// Example is a simple example service for general reference on the recommended
-// kubelet plugin model and plugin watcher testing.
-service Example {
-	rpc GetExampleInfo(ExampleRequest) returns (ExampleResponse) {}
-}

pkg/kubelet/util/pluginwatcher/plugin_watcher_test.go

-/*
-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 pluginwatcher
-
-import (
-	"flag"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/require"
-
-	registerapi "k8s.io/kubernetes/pkg/kubelet/apis/pluginregistration/v1alpha1"
-)
-
-var (
-	socketDir string
-
-	supportedVersions = []string{"v1beta1", "v1beta2"}
-)
-
-func init() {
-	var logLevel string
-
-	flag.Set("alsologtostderr", fmt.Sprintf("%t", true))
-	flag.StringVar(&logLevel, "logLevel", "6", "test")
-	flag.Lookup("v").Value.Set(logLevel)
-
-	d, err := ioutil.TempDir("", "plugin_test")
-	if err != nil {
-		panic(fmt.Sprintf("Could not create a temp directory: %s", d))
-	}
-
-	socketDir = d
-}
-
-func cleanup(t *testing.T) {
-	require.NoError(t, os.RemoveAll(socketDir))
-	os.MkdirAll(socketDir, 0755)
-}
-
-func TestPluginRegistration(t *testing.T) {
-	defer cleanup(t)
-
-	hdlr := NewExampleHandler(supportedVersions)
-	w := newWatcherWithHandler(t, hdlr)
-	defer func() { require.NoError(t, w.Stop()) }()
-
-	for i := 0; i < 10; i++ {
-		socketPath := fmt.Sprintf("%s/plugin-%d.sock", socketDir, i)
-		pluginName := fmt.Sprintf("example-plugin-%d", i)
-
-		hdlr.AddPluginName(pluginName)
-
-		p := NewTestExamplePlugin(pluginName, registerapi.DevicePlugin, socketPath, supportedVersions...)
-		require.NoError(t, p.Serve("v1beta1", "v1beta2"))
-
-		require.True(t, waitForEvent(t, exampleEventValidate, hdlr.EventChan(p.pluginName)))
-		require.True(t, waitForEvent(t, exampleEventRegister, hdlr.EventChan(p.pluginName)))
-
-		require.True(t, waitForPluginRegistrationStatus(t, p.registrationStatus))
-
-		require.NoError(t, p.Stop())
-		require.True(t, waitForEvent(t, exampleEventDeRegister, hdlr.EventChan(p.pluginName)))
-	}
-}
-
-func TestPluginReRegistration(t *testing.T) {
-	defer cleanup(t)
-
-	pluginName := fmt.Sprintf("example-plugin")
-	hdlr := NewExampleHandler(supportedVersions)
-
-	w := newWatcherWithHandler(t, hdlr)
-	defer func() { require.NoError(t, w.Stop()) }()
-
-	plugins := make([]*examplePlugin, 10)
-
-	for i := 0; i < 10; i++ {
-		socketPath := fmt.Sprintf("%s/plugin-%d.sock", socketDir, i)
-		hdlr.AddPluginName(pluginName)
-
-		p := NewTestExamplePlugin(pluginName, registerapi.DevicePlugin, socketPath, supportedVersions...)
-		require.NoError(t, p.Serve("v1beta1", "v1beta2"))
-
-		require.True(t, waitForEvent(t, exampleEventValidate, hdlr.EventChan(p.pluginName)))
-		require.True(t, waitForEvent(t, exampleEventRegister, hdlr.EventChan(p.pluginName)))
-
-		require.True(t, waitForPluginRegistrationStatus(t, p.registrationStatus))
-
-		plugins[i] = p
-	}
-
-	plugins[len(plugins)-1].Stop()
-	require.True(t, waitForEvent(t, exampleEventDeRegister, hdlr.EventChan(pluginName)))
-
-	close(hdlr.EventChan(pluginName))
-	for i := 0; i < len(plugins)-1; i++ {
-		plugins[i].Stop()
-	}
-}
-
-func TestPluginRegistrationAtKubeletStart(t *testing.T) {
-	defer cleanup(t)
-
-	hdlr := NewExampleHandler(supportedVersions)
-	plugins := make([]*examplePlugin, 10)
-
-	for i := 0; i < len(plugins); i++ {
-		socketPath := fmt.Sprintf("%s/plugin-%d.sock", socketDir, i)
-		pluginName := fmt.Sprintf("example-plugin-%d", i)
-		hdlr.AddPluginName(pluginName)
-
-		p := NewTestExamplePlugin(pluginName, registerapi.DevicePlugin, socketPath, supportedVersions...)
-		require.NoError(t, p.Serve("v1beta1", "v1beta2"))
-		defer func(p *examplePlugin) { require.NoError(t, p.Stop()) }(p)
-
-		plugins[i] = p
-	}
-
-	w := newWatcherWithHandler(t, hdlr)
-	defer func() { require.NoError(t, w.Stop()) }()
-
-	var wg sync.WaitGroup
-	for i := 0; i < len(plugins); i++ {
-		wg.Add(1)
-		go func(p *examplePlugin) {
-			defer wg.Done()
-
-			require.True(t, waitForEvent(t, exampleEventValidate, hdlr.EventChan(p.pluginName)))
-			require.True(t, waitForEvent(t, exampleEventRegister, hdlr.EventChan(p.pluginName)))
-
-			require.True(t, waitForPluginRegistrationStatus(t, p.registrationStatus))
-		}(plugins[i])
-	}
-
-	c := make(chan struct{})
-	go func() {
-		defer close(c)
-		wg.Wait()
-	}()
-
-	select {
-	case <-c:
-		return
-	case <-time.After(2 * time.Second):
-		t.Fatalf("Timeout while waiting for the plugin registration status")
-	}
-}
-
-func TestPluginRegistrationFailureWithUnsupportedVersion(t *testing.T) {
-	defer cleanup(t)
-
-	pluginName := fmt.Sprintf("example-plugin")
-	socketPath := socketDir + "/plugin.sock"
-
-	hdlr := NewExampleHandler(supportedVersions)
-	hdlr.AddPluginName(pluginName)
-
-	w := newWatcherWithHandler(t, hdlr)
-	defer func() { require.NoError(t, w.Stop()) }()
-
-	// Advertise v1beta3 but don't serve anything else than the plugin service
-	p := NewTestExamplePlugin(pluginName, registerapi.DevicePlugin, socketPath, "v1beta3")
-	require.NoError(t, p.Serve())
-	defer func() { require.NoError(t, p.Stop()) }()
-
-	require.True(t, waitForEvent(t, exampleEventValidate, hdlr.EventChan(p.pluginName)))
-	require.False(t, waitForPluginRegistrationStatus(t, p.registrationStatus))
-}
-
-func TestPlugiRegistrationFailureWithUnsupportedVersionAtKubeletStart(t *testing.T) {
-	defer cleanup(t)
-
-	pluginName := fmt.Sprintf("example-plugin")
-	socketPath := socketDir + "/plugin.sock"
-
-	// Advertise v1beta3 but don't serve anything else than the plugin service
-	p := NewTestExamplePlugin(pluginName, registerapi.DevicePlugin, socketPath, "v1beta3")
-	require.NoError(t, p.Serve())
-	defer func() { require.NoError(t, p.Stop()) }()
-
-	hdlr := NewExampleHandler(supportedVersions)
-	hdlr.AddPluginName(pluginName)
-
-	w := newWatcherWithHandler(t, hdlr)
-	defer func() { require.NoError(t, w.Stop()) }()
-
-	require.True(t, waitForEvent(t, exampleEventValidate, hdlr.EventChan(p.pluginName)))
-	require.False(t, waitForPluginRegistrationStatus(t, p.registrationStatus))
-}
-
-func waitForPluginRegistrationStatus(t *testing.T, statusChan chan registerapi.RegistrationStatus) bool {
-	select {
-	case status := <-statusChan:
-		return status.PluginRegistered
-	case <-time.After(10 * time.Second):
-		t.Fatalf("Timed out while waiting for registration status")
-	}
-	return false
-}
-
-func waitForEvent(t *testing.T, expected examplePluginEvent, eventChan chan examplePluginEvent) bool {
-	select {
-	case event := <-eventChan:
-		return event == expected
-	case <-time.After(2 * time.Second):
-		t.Fatalf("Timed out while waiting for registration status %v", expected)
-	}
-
-	return false
-}
-
-func newWatcherWithHandler(t *testing.T, hdlr PluginHandler) *Watcher {
-	w := NewWatcher(socketDir)
-
-	w.AddHandler(registerapi.DevicePlugin, hdlr)
-	require.NoError(t, w.Start())
-
-	return w
-}

pkg/kubelet/util/pluginwatcher/types.go

-/*
-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 pluginwatcher
-
-// PluginHandler is an interface a client of the pluginwatcher API needs to implement in
-// order to consume plugins
-// The PluginHandler follows the simple following state machine:
-//
-//                         +--------------------------------------+
-//                         |            ReRegistration            |
-//                         | Socket created with same plugin name |
-//                         |                                      |
-//                         |                                      |
-//    Socket Created       v                                      +        Socket Deleted
-// +------------------> Validate +---------------------------> Register +------------------> DeRegister
-//                         +                                      +                              +
-//                         |                                      |                              |
-//                         | Error                                | Error                        |
-//                         |                                      |                              |
-//                         v                                      v                              v
-//                        Out                                    Out                            Out
-//
-// The pluginwatcher module follows strictly and sequentially this state machine for each *plugin name*.
-// e.g: If you are Registering a plugin foo, you cannot get a DeRegister call for plugin foo
-//      until the Register("foo") call returns. Nor will you get a Validate("foo", "Different endpoint", ...)
-//      call until the Register("foo") call returns.
-//
-// ReRegistration: Socket created with same plugin name, usually for a plugin update
-// e.g: plugin with name foo registers at foo.com/foo-1.9.7 later a plugin with name foo
-//      registers at foo.com/foo-1.9.9
-//
-// DeRegistration: When ReRegistration happens only the deletion of the new socket will trigger a DeRegister call
-
-type PluginHandler interface {
-	// Validate returns an error if the information provided by
-	// the potential plugin is erroneous (unsupported version, ...)
-	ValidatePlugin(pluginName string, endpoint string, versions []string) error
-	// RegisterPlugin is called so that the plugin can be register by any
-	// plugin consumer
-	// Error encountered here can still be Notified to the plugin.
-	RegisterPlugin(pluginName, endpoint string) error
-	// DeRegister is called once the pluginwatcher observes that the socket has
-	// been deleted.
-	DeRegisterPlugin(pluginName string)
-}

pkg/volume/csi/csi_client.go

@@ -27,8 +27,6 @@ import (
 	"github.com/golang/glog"
 	"google.golang.org/grpc"
 	api "k8s.io/api/core/v1"
-	utilfeature "k8s.io/apiserver/pkg/util/feature"
-	"k8s.io/kubernetes/pkg/features"
 )
 
 type csiClient interface {
@@ -295,14 +293,6 @@ func newGrpcConn(driverName string) (*grpc.ClientConn, error) {
 		return nil, fmt.Errorf("driver name is empty")
 	}
 	addr := fmt.Sprintf(csiAddrTemplate, driverName)
-	// TODO once KubeletPluginsWatcher graduates to beta, remove FeatureGate check
-	if utilfeature.DefaultFeatureGate.Enabled(features.KubeletPluginsWatcher) {
-		driver, ok := csiDrivers.driversMap[driverName]
-		if !ok {
-			return nil, fmt.Errorf("driver name %s not found in the list of registered CSI drivers", driverName)
-		}
-		addr = driver.driverEndpoint
-	}
 	network := "unix"
 	glog.V(4).Infof(log("creating new gRPC connection for [%s://%s]", network, addr))