intel / multus-cni / 715

Committed 21 Mar 2019 04:28PM UTC coverage: 53.084% (+1.5%) from 51.624%

Build # 715

Build Type

Pull #231

travis-ci

Committed by

web-flow

Commit Message

add kubelet client for Pod resource info

This change introduces kubelet client to get allocated device
information of a Pod from newly added Kubelet grpc service.
For more information please see:
[kubernetes/kubernetes#70508](https://github.com/kubernetes/kubernetes/pull/70508)

Change-Id: I11e58ccdd
Signed-off-by: Abdul Halim <abdul.halim@intel.com>

Pull Request Pull Request #231: add kubelet client for Pod resource info

Run Details

77 of 77 new or added lines in 3 files covered. (100.0%)

654 of 1232 relevant lines covered (53.08%)

3.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.95

/kubeletclient/kubeletclient.go

package kubeletclient

import (
        "os"
        "path/filepath"
        "time"

        "github.com/intel/multus-cni/checkpoint"
        "github.com/intel/multus-cni/logging"
        "github.com/intel/multus-cni/types"
        "golang.org/x/net/context"
        v1 "k8s.io/api/core/v1"
        "k8s.io/kubernetes/pkg/kubelet/apis/podresources"
        podresourcesapi "k8s.io/kubernetes/pkg/kubelet/apis/podresources/v1alpha1"
        "k8s.io/kubernetes/pkg/kubelet/util"
)

const (
        defaultKubeletSocketFile   = "kubelet.sock"
        defaultPodResourcesPath    = "/var/lib/kubelet/pod-resources"
        defaultPodResourcesMaxSize = 1024 * 1024 * 16 // 16 Mb
)

var (
        kubeletSocket string
)

// GetResourceClient returns an instance of ResourceClient interface initialized with Pod resource information
func GetResourceClient() (types.ResourceClient, error) {
        // If Kubelet resource API endpoint exist use that by default
        // Or else fallback with checkpoint file
        if hasKubeletAPIEndpoint() {
                logging.Printf(logging.VerboseLevel, "GetResourceClient(): using Kubelet resource API endpoint")
                return getKubeletClient()
        } else {
                logging.Printf(logging.VerboseLevel, "GetResourceClient(): using Kubelet device plugin checkpoint")
                return checkpoint.GetCheckpoint()
        }
}

func getKubeletClient() (types.ResourceClient, error) {
        newClient := &kubeletClient{}
        if kubeletSocket == "" {
                kubeletSocket = util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)
        }

        client, conn, err := podresources.GetClient(kubeletSocket, 10*time.Second, defaultPodResourcesMaxSize)
        if err != nil {
                return nil, logging.Errorf("GetResourceClient(): error getting grpc client: %v\n", err)
        }
        defer conn.Close()

        if err := newClient.getPodResources(client); err != nil {
                return nil, logging.Errorf("GetResourceClient(): error getting resource client: %v\n", err)
        }

        return newClient, nil
}

type kubeletClient struct {
        resources []*podresourcesapi.PodResources
}

func (rc *kubeletClient) getPodResources(client podresourcesapi.PodResourcesListerClient) error {

        ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
        defer cancel()

        resp, err := client.List(ctx, &podresourcesapi.ListPodResourcesRequest{})
        if err != nil {
                return logging.Errorf("getPodResources(): %v.Get(_) = _, %v", client, err)
        }

        rc.resources = resp.PodResources
        return nil
}

// GetPodResourceMap returns an instance of a map of Pod ResourceInfo given a (Pod name, namespace) tuple
func (rc *kubeletClient) GetPodResourceMap(pod *v1.Pod) (map[string]*types.ResourceInfo, error) {
        resourceMap := make(map[string]*types.ResourceInfo)

        name := pod.Name
        ns := pod.Namespace

        if name == "" || ns == "" {
                return nil, logging.Errorf("GetPodResourcesMap(): Pod name or namespace cannot be empty")
        }

        for _, pr := range rc.resources {
                if pr.Name == name && pr.Namespace == ns {
                        for _, cnt := range pr.Containers {
                                for _, dev := range cnt.Devices {
                                        if rInfo, ok := resourceMap[dev.ResourceName]; ok {
                                                rInfo.DeviceIDs = append(rInfo.DeviceIDs, dev.DeviceIds...)
                                        } else {
                                                resourceMap[dev.ResourceName] = &types.ResourceInfo{DeviceIDs: dev.DeviceIds}
                                        }
                                }
                        }
                }
        }
        return resourceMap, nil
}

func hasKubeletAPIEndpoint() bool {
        // Check for kubelet resource API socket file
        kubeletSocket = filepath.Join(defaultPodResourcesPath, defaultKubeletSocketFile)
        if _, err := os.Stat(kubeletSocket); err != nil {
                logging.Errorf("hasKubeletAPIEndpoint(): error looking up kubelet resource api socket file: %q", err)
                return false
        }
        return true
}

1	package kubeletclient
2
3	import (
4	"os"
5	"path/filepath"
6	"time"
7
8	"github.com/intel/multus-cni/checkpoint"
9	"github.com/intel/multus-cni/logging"
10	"github.com/intel/multus-cni/types"
11	"golang.org/x/net/context"
12	v1 "k8s.io/api/core/v1"
13	"k8s.io/kubernetes/pkg/kubelet/apis/podresources"
14	podresourcesapi "k8s.io/kubernetes/pkg/kubelet/apis/podresources/v1alpha1"
15	"k8s.io/kubernetes/pkg/kubelet/util"
16	)
17
18	const (
19	defaultKubeletSocketFile = "kubelet.sock"
20	defaultPodResourcesPath = "/var/lib/kubelet/pod-resources"
21	defaultPodResourcesMaxSize = 1024 * 1024 * 16 // 16 Mb
22	)
23
24	var (
25	kubeletSocket string
26	)
27
28	// GetResourceClient returns an instance of ResourceClient interface initialized with Pod resource information
29	func GetResourceClient() (types.ResourceClient, error) {	1✔
30	// If Kubelet resource API endpoint exist use that by default	1✔
31	// Or else fallback with checkpoint file	1✔
32	if hasKubeletAPIEndpoint() {	1✔
33	logging.Printf(logging.VerboseLevel, "GetResourceClient(): using Kubelet resource API endpoint")	×
34	return getKubeletClient()	×
35	} else {	1✔
36	logging.Printf(logging.VerboseLevel, "GetResourceClient(): using Kubelet device plugin checkpoint")	1✔
37	return checkpoint.GetCheckpoint()	1✔
38	}	1✔
39	}
40
41	func getKubeletClient() (types.ResourceClient, error) {	4✔
42	newClient := &kubeletClient{}	4✔
43	if kubeletSocket == "" {	4✔
44	kubeletSocket = util.LocalEndpoint(defaultPodResourcesPath, podresources.Socket)	×
45	}	×
46
47	client, conn, err := podresources.GetClient(kubeletSocket, 10*time.Second, defaultPodResourcesMaxSize)	4✔
48	if err != nil {	4✔
49	return nil, logging.Errorf("GetResourceClient(): error getting grpc client: %v\n", err)	×
50	}	×
51	defer conn.Close()	4✔
52		4✔
53	if err := newClient.getPodResources(client); err != nil {	4✔
54	return nil, logging.Errorf("GetResourceClient(): error getting resource client: %v\n", err)	×
55	}	×
56
57	return newClient, nil	4✔
58	}
59
60	type kubeletClient struct {
61	resources []*podresourcesapi.PodResources
62	}
63
64	func (rc *kubeletClient) getPodResources(client podresourcesapi.PodResourcesListerClient) error {	4✔
65		4✔
66	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)	4✔
67	defer cancel()	4✔
68		4✔
69	resp, err := client.List(ctx, &podresourcesapi.ListPodResourcesRequest{})	4✔
70	if err != nil {	4✔
71	return logging.Errorf("getPodResources(): %v.Get(_) = _, %v", client, err)	×
72	}	×
73
74	rc.resources = resp.PodResources	4✔
75	return nil	4✔
76	}
77
78	// GetPodResourceMap returns an instance of a map of Pod ResourceInfo given a (Pod name, namespace) tuple
79	func (rc kubeletClient) GetPodResourceMap(pod v1.Pod) (map[string]*types.ResourceInfo, error) {	4✔
80	resourceMap := make(map[string]*types.ResourceInfo)	4✔
81		4✔
82	name := pod.Name	4✔
83	ns := pod.Namespace	4✔
84		4✔
85	if name == "" \|\| ns == "" {	6✔
86	return nil, logging.Errorf("GetPodResourcesMap(): Pod name or namespace cannot be empty")	2✔
87	}	2✔
88
89	for _, pr := range rc.resources {	4✔
90	if pr.Name == name && pr.Namespace == ns {	3✔
91	for _, cnt := range pr.Containers {	2✔
92	for _, dev := range cnt.Devices {	2✔
93	if rInfo, ok := resourceMap[dev.ResourceName]; ok {	1✔
94	rInfo.DeviceIDs = append(rInfo.DeviceIDs, dev.DeviceIds...)	×
95	} else {	1✔
96	resourceMap[dev.ResourceName] = &types.ResourceInfo{DeviceIDs: dev.DeviceIds}	1✔
97	}	1✔
98	}
99	}
100	}
101	}
102	return resourceMap, nil	2✔
103	}
104
105	func hasKubeletAPIEndpoint() bool {	1✔
106	// Check for kubelet resource API socket file	1✔
107	kubeletSocket = filepath.Join(defaultPodResourcesPath, defaultKubeletSocketFile)	1✔
108	if _, err := os.Stat(kubeletSocket); err != nil {	2✔
109	logging.Errorf("hasKubeletAPIEndpoint(): error looking up kubelet resource api socket file: %q", err)	1✔
110	return false	1✔
111	}	1✔
112	return true	×
113	}

intel / multus-cni / 715

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous