Device Plugin Kubelet integration

pkg/kubelet/BUILD

@@ -43,6 +43,7 @@ go_library(
         "//pkg/fieldpath:go_default_library",
         "//pkg/kubelet/apis:go_default_library",
         "//pkg/kubelet/apis/cri:go_default_library",
+        "//pkg/kubelet/apis/deviceplugin/v1alpha1:go_default_library",
         "//pkg/kubelet/apis/kubeletconfig:go_default_library",
         "//pkg/kubelet/apis/kubeletconfig/v1alpha1:go_default_library",
         "//pkg/kubelet/cadvisor:go_default_library",

pkg/kubelet/cm/BUILD

@@ -7,6 +7,7 @@ go_library(
         "container_manager.go",
         "container_manager_stub.go",
         "container_manager_unsupported.go",
+        "device_plugin_handler.go",
         "helpers_unsupported.go",
         "pod_container_manager_stub.go",
         "pod_container_manager_unsupported.go",
@@ -27,8 +28,10 @@ go_library(
     }),
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/kubelet/apis/deviceplugin/v1alpha1:go_default_library",
         "//pkg/kubelet/apis/kubeletconfig:go_default_library",
         "//pkg/kubelet/cadvisor:go_default_library",
+        "//pkg/kubelet/deviceplugin:go_default_library",
         "//pkg/kubelet/eviction/api:go_default_library",
         "//pkg/util/mount:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",

pkg/kubelet/cm/container_manager.go

@@ -20,6 +20,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	// TODO: Migrate kubelet to either use its own internal objects or client library.
 	"k8s.io/api/core/v1"
+	pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1alpha1"
 	"k8s.io/kubernetes/pkg/kubelet/apis/kubeletconfig"
 	evictionapi "k8s.io/kubernetes/pkg/kubelet/eviction/api"
 
@@ -66,6 +67,8 @@ type ContainerManager interface {
 	// UpdateQOSCgroups performs housekeeping updates to ensure that the top
 	// level QoS containers have their desired state in a thread-safe way
 	UpdateQOSCgroups() error
+
+	InternalContainerLifecycle() InternalContainerLifecycle
 }
 
 type NodeConfig struct {

pkg/kubelet/cm/container_manager_linux.go

@@ -117,6 +117,8 @@ type containerManagerImpl struct {
 	recorder record.EventRecorder
 	// Interface for QoS cgroup management
 	qosContainerManager QOSContainerManager
+	// Interface for device plugin management.
+	devicePluginHdler DevicePluginHandler
 }
 
 type features struct {
@@ -820,3 +822,13 @@ func (cm *containerManagerImpl) GetCapacity() v1.ResourceList {
 	defer cm.RUnlock()
 	return cm.capacity
 }
+
+// GetDevicePluginHandler returns the DevicePluginHandler
+func (m *containerManagerImpl) GetDevicePluginHandler() DevicePluginHandler {
+	return m.devicePluginHdler
+}
+
+// SetDevicePluginHandler sets the DevicePluginHandler
+func (m *containerManagerImpl) SetDevicePluginHandler(d DevicePluginHandler) {
+	m.devicePluginHdler = d
+}

pkg/kubelet/cm/device_plugin_handler.go

@@ -0,0 +1,55 @@
+/*
+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 cm
+
+import (
+	"fmt"
+
+	"github.com/golang/glog"
+
+	pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1alpha1"
+	"k8s.io/kubernetes/pkg/kubelet/deviceplugin"
+)
+
+type DevicePluginHandlerImpl struct {
+	devicePluginManager deviceplugin.Manager
+}
+
+// NewDevicePluginHandler create a DevicePluginHandler
+func NewDevicePluginHandler() (*DevicePluginHandlerImpl, error) {
+	glog.V(2).Infof("Starting Device Plugin Handler")
+
+	mgr, err := deviceplugin.NewManagerImpl(pluginapi.DevicePluginPath,
+		func(r string, a, u, d []*pluginapi.Device) {})
+
+	if err != nil {
+		return nil, fmt.Errorf("Failed to initialize device plugin: %+v", err)
+	}
+
+	if err := mgr.Start(); err != nil {
+		return nil, err
+	}
+
+	return &DevicePluginHandlerImpl{
+		devicePluginManager: mgr,
+	}, nil
+}
+
+// TODO cache this
+func (h *DevicePluginHandlerImpl) Devices() map[string][]*pluginapi.Device {
+	return h.devicePluginManager.Devices()
+}

pkg/kubelet/kubelet.go

@@ -754,6 +754,13 @@ func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
 		return nil, err
 	}
 
+	devicePluginHdlr, err := cm.NewDevicePluginHandler()
+	if err != nil {
+		return nil, err
+	}
+
+	klet.containerManager.SetDevicePluginHandler(devicePluginHdlr)
+
 	// If the experimentalMounterPathFlag is set, we do not want to
 	// check node capabilities since the mount path is not the default
 	if len(kubeCfg.ExperimentalMounterPath) != 0 {

pkg/kubelet/kubelet_node_status.go

@@ -39,6 +39,7 @@ import (
 	"k8s.io/kubernetes/pkg/cloudprovider"
 	"k8s.io/kubernetes/pkg/features"
 	kubeletapis "k8s.io/kubernetes/pkg/kubelet/apis"
+	pluginapi "k8s.io/kubernetes/pkg/kubelet/apis/deviceplugin/v1alpha1"
 	"k8s.io/kubernetes/pkg/kubelet/cadvisor"
 	"k8s.io/kubernetes/pkg/kubelet/events"
 	"k8s.io/kubernetes/pkg/kubelet/util"
@@ -621,6 +622,27 @@ func (kl *Kubelet) setNodeStatusMachineInfo(node *v1.Node) {
 		}
 		node.Status.Allocatable[k] = value
 	}
+
+	hdlr := kl.containerManager.GetDevicePluginHandler()
+	if hdlr == nil {
+		return
+	}
+
+	for k, v := range hdlr.Devices() {
+		key := v1.ResourceName(v1.ResourceOpaqueIntPrefix + k)
+
+		var n int64
+		n = 0
+
+		for _, d := range v {
+			if d.Health == pluginapi.Unhealthy {
+				continue
+			}
+			n++
+		}
+
+		node.Status.Capacity[key] = *resource.NewQuantity(n, resource.DecimalSI)
+	}
 }
 
 // Set versioninfo for the node.