/
kubeletclient.go
executable file
·113 lines (93 loc) · 3.4 KB
/
kubeletclient.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
package kubeletclient
import (
"os"
"path/filepath"
"time"
"github.com/archichris/multus-hc/checkpoint"
"github.com/archichris/multus-hc/logging"
"github.com/archichris/multus-hc/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"
defaultPodResourcesMaxSize = 1024 * 1024 * 16 // 16 Mb
)
var (
kubeletSocket string
defaultPodResourcesPath = "/var/lib/kubelet/pod-resources"
)
// 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.Verbosef("GetResourceClient(): using Kubelet resource API endpoint")
return getKubeletClient()
}
logging.Verbosef("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
kubeletAPISocket := filepath.Join(defaultPodResourcesPath, defaultKubeletSocketFile)
if _, err := os.Stat(kubeletAPISocket); err != nil {
logging.Verbosef("hasKubeletAPIEndpoint(): error looking up kubelet resource api socket file: %q", err)
return false
}
return true
}