intel / multus-cni / 379

Committed 17 Sep 2018 01:37PM UTC coverage: 26.471% (-18.5%) from 45.0%

Build # 379

Build Type

Pull #151

travis-ci

Committed by

web-flow

Commit Message

parse kubelet checkpoint file for pod devices

Enabling kubelete checkpoint file  parsing to get Pod device info
so that these device information can be passed into CNI plugins
that need specific device information to work on.

Change-Id: I6630f56ad

Pull Request Pull Request #151: parse kubelet checkpoint file for pod devices

Run Details

45 of 45 new or added lines in 2 files covered. (100.0%)

216 of 816 relevant lines covered (26.47%)

0.73 hits per line

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

8.98

/k8sclient/k8sclient.go

// Copyright (c) 2017 Intel Corporation
//
// 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 k8sclient

import (
        "encoding/json"
        "fmt"
        "os"
        "regexp"
        "strings"

        v1 "k8s.io/api/core/v1"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/rest"
        "k8s.io/client-go/tools/clientcmd"
        "k8s.io/client-go/util/retry"

        "github.com/containernetworking/cni/libcni"
        "github.com/containernetworking/cni/pkg/skel"
        cnitypes "github.com/containernetworking/cni/pkg/types"
        "github.com/intel/multus-cni/checkpoint"
        "github.com/intel/multus-cni/logging"
        "github.com/intel/multus-cni/types"
)

const (
        resourceNameAnnot = "k8s.v1.cni.cncf.io/resourceName"
)

// NoK8sNetworkError indicates error, no network in kubernetes
type NoK8sNetworkError struct {
        message string
}

type clientInfo struct {
        Client       KubeClient
        Podnamespace string
        Podname      string
}

func (e *NoK8sNetworkError) Error() string { return string(e.message) }

type defaultKubeClient struct {
        client kubernetes.Interface
}

// defaultKubeClient implements KubeClient
var _ KubeClient = &defaultKubeClient{}

func (d *defaultKubeClient) GetRawWithPath(path string) ([]byte, error) {
        return d.client.ExtensionsV1beta1().RESTClient().Get().AbsPath(path).DoRaw()
}

func (d *defaultKubeClient) GetPod(namespace, name string) (*v1.Pod, error) {
        return d.client.Core().Pods(namespace).Get(name, metav1.GetOptions{})
}

func (d *defaultKubeClient) UpdatePodStatus(pod *v1.Pod) (*v1.Pod, error) {
        return d.client.Core().Pods(pod.Namespace).UpdateStatus(pod)
}

func setKubeClientInfo(c *clientInfo, client KubeClient, k8sArgs *types.K8sArgs) {
        logging.Debugf("setKubeClientInfo: %v, %v, %v", c, client, k8sArgs)
        c.Client = client
        c.Podnamespace = string(k8sArgs.K8S_POD_NAMESPACE)
        c.Podname = string(k8sArgs.K8S_POD_NAME)
}

func SetNetworkStatus(k *clientInfo, netStatus []*types.NetworkStatus) error {

        logging.Debugf("SetNetworkStatus: %v, %v", k, netStatus)
        pod, err := k.Client.GetPod(k.Podnamespace, k.Podname)
        if err != nil {
                return logging.Errorf("SetNetworkStatus: failed to query the pod %v in out of cluster comm: %v", k.Podname, err)
        }

        var ns string
        if netStatus != nil {
                var networkStatus []string
                for _, nets := range netStatus {
                        data, err := json.MarshalIndent(nets, "", "    ")
                        if err != nil {
                                return logging.Errorf("SetNetworkStatus: error with Marshal Indent: %v", err)
                        }
                        networkStatus = append(networkStatus, string(data))
                }

                ns = fmt.Sprintf("[%s]", strings.Join(networkStatus, ","))
        }
        _, err = setPodNetworkAnnotation(k.Client, k.Podnamespace, pod, ns)
        if err != nil {
                return logging.Errorf("SetNetworkStatus: failed to update the pod %v in out of cluster comm: %v", k.Podname, err)
        }

        return nil
}

func setPodNetworkAnnotation(client KubeClient, namespace string, pod *v1.Pod, networkstatus string) (*v1.Pod, error) {
        logging.Debugf("setPodNetworkAnnotation: %v, %s, %v, %s", client, namespace, pod, networkstatus)
        //if pod annotations is empty, make sure it allocatable
        if len(pod.Annotations) == 0 {
                pod.Annotations = make(map[string]string)
        }

        pod.Annotations["k8s.v1.cni.cncf.io/networks-status"] = networkstatus

        pod = pod.DeepCopy()
        var err error
        if resultErr := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
                if err != nil {
                        // Re-get the pod unless it's the first attempt to update
                        pod, err = client.GetPod(pod.Namespace, pod.Name)
                        if err != nil {
                                return err
                        }
                }

                pod, err = client.UpdatePodStatus(pod)
                return err
        }); resultErr != nil {
                return nil, logging.Errorf("status update failed for pod %s/%s: %v", pod.Namespace, pod.Name, resultErr)
        }
        return pod, nil
}

func getPodNetworkAnnotation(client KubeClient, k8sArgs *types.K8sArgs) (string, string, string, error) {
        var err error

        logging.Debugf("getPodNetworkAnnotation: %v, %v", client, k8sArgs)
        pod, err := client.GetPod(string(k8sArgs.K8S_POD_NAMESPACE), string(k8sArgs.K8S_POD_NAME))
        if err != nil {
                return "", "", "", logging.Errorf("getPodNetworkAnnotation: failed to query the pod %v in out of cluster comm: %v", string(k8sArgs.K8S_POD_NAME), err)
        }

        return pod.Annotations["k8s.v1.cni.cncf.io/networks"], pod.ObjectMeta.Namespace, string(pod.UID), nil
}

func parsePodNetworkObjectName(podnetwork string) (string, string, string, error) {
        var netNsName string
        var netIfName string
        var networkName string

        logging.Debugf("parsePodNetworkObjectName: %s", podnetwork)
        slashItems := strings.Split(podnetwork, "/")
        if len(slashItems) == 2 {
                netNsName = strings.TrimSpace(slashItems[0])
                networkName = slashItems[1]
        } else if len(slashItems) == 1 {
                networkName = slashItems[0]
        } else {
                return "", "", "", logging.Errorf("Invalid network object (failed at '/')")
        }

        atItems := strings.Split(networkName, "@")
        networkName = strings.TrimSpace(atItems[0])
        if len(atItems) == 2 {
                netIfName = strings.TrimSpace(atItems[1])
        } else if len(atItems) != 1 {
                return "", "", "", logging.Errorf("Invalid network object (failed at '@')")
        }

        // Check and see if each item matches the specification for valid attachment name.
        // "Valid attachment names must be comprised of units of the DNS-1123 label format"
        // [a-z0-9]([-a-z0-9]*[a-z0-9])?
        // And we allow at (@), and forward slash (/) (units separated by commas)
        // It must start and end alphanumerically.
        allItems := []string{netNsName, networkName, netIfName}
        for i := range allItems {
                matched, _ := regexp.MatchString("^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", allItems[i])
                if !matched && len([]rune(allItems[i])) > 0 {
                        return "", "", "", logging.Errorf(fmt.Sprintf("Failed to parse: one or more items did not match comma-delimited format (must consist of lower case alphanumeric characters). Must start and end with an alphanumeric character), mismatch @ '%v'", allItems[i]))
                }
        }

        logging.Debugf("parsePodNetworkObjectName: parsed: %s, %s, %s", netNsName, networkName, netIfName)
        return netNsName, networkName, netIfName, nil
}

func parsePodNetworkAnnotation(podNetworks, defaultNamespace string) ([]*types.NetworkSelectionElement, error) {
        var networks []*types.NetworkSelectionElement

        logging.Debugf("parsePodNetworkAnnotation: %s, %s", podNetworks, defaultNamespace)
        if podNetworks == "" {
                return nil, logging.Errorf("parsePodNetworkAnnotation: pod annotation not having \"network\" as key, refer Multus README.md for the usage guide")
        }

        if strings.IndexAny(podNetworks, "[{\"") >= 0 {
                if err := json.Unmarshal([]byte(podNetworks), &networks); err != nil {
                        return nil, logging.Errorf("parsePodNetworkAnnotation: failed to parse pod Network Attachment Selection Annotation JSON format: %v", err)
                }
        } else {
                // Comma-delimited list of network attachment object names
                for _, item := range strings.Split(podNetworks, ",") {
                        // Remove leading and trailing whitespace.
                        item = strings.TrimSpace(item)

                        // Parse network name (i.e. <namespace>/<network name>@<ifname>)
                        netNsName, networkName, netIfName, err := parsePodNetworkObjectName(item)
                        if err != nil {
                                return nil, logging.Errorf("parsePodNetworkAnnotation: %v", err)
                        }

                        networks = append(networks, &types.NetworkSelectionElement{
                                Name:             networkName,
                                Namespace:        netNsName,
                                InterfaceRequest: netIfName,
                        })
                }
        }

        for _, net := range networks {
                if net.Namespace == "" {
                        net.Namespace = defaultNamespace
                }
        }

        return networks, nil
}

func getCNIConfigFromFile(name string, confdir string) ([]byte, error) {
        logging.Debugf("getCNIConfigFromFile: %s, %s", name, confdir)

        // In the absence of valid keys in a Spec, the runtime (or
        // meta-plugin) should load and execute a CNI .configlist
        // or .config (in that order) file on-disk whose JSON
        // “name” key matches this Network object’s name.

        // In part, adapted from K8s pkg/kubelet/dockershim/network/cni/cni.go#getDefaultCNINetwork
        files, err := libcni.ConfFiles(confdir, []string{".conf", ".json", ".conflist"})
        switch {
        case err != nil:
                return nil, logging.Errorf("No networks found in %s", confdir)
        case len(files) == 0:
                return nil, logging.Errorf("No networks found in %s", confdir)
        }

        for _, confFile := range files {
                var confList *libcni.NetworkConfigList
                if strings.HasSuffix(confFile, ".conflist") {
                        confList, err = libcni.ConfListFromFile(confFile)
                        if err != nil {
                                return nil, logging.Errorf("Error loading CNI conflist file %s: %v", confFile, err)
                        }

                        if confList.Name == name {
                                return confList.Bytes, nil
                        }

                } else {
                        conf, err := libcni.ConfFromFile(confFile)
                        if err != nil {
                                return nil, logging.Errorf("Error loading CNI config file %s: %v", confFile, err)
                        }

                        if conf.Network.Name == name {
                                // Ensure the config has a "type" so we know what plugin to run.
                                // Also catches the case where somebody put a conflist into a conf file.
                                if conf.Network.Type == "" {
                                        return nil, logging.Errorf("Error loading CNI config file %s: no 'type'; perhaps this is a .conflist?", confFile)
                                }
                                return conf.Bytes, nil
                        }
                }
        }

        return nil, logging.Errorf("no network available in the name %s in cni dir %s", name, confdir)
}

// getCNIConfigFromSpec reads a CNI JSON configuration from the NetworkAttachmentDefinition
// object's Spec.Config field and fills in any missing details like the network name
func getCNIConfigFromSpec(configData, netName string) ([]byte, error) {
        var rawConfig map[string]interface{}
        var err error

        logging.Debugf("getCNIConfigFromSpec: %s, %s", configData, netName)
        configBytes := []byte(configData)
        err = json.Unmarshal(configBytes, &rawConfig)
        if err != nil {
                return nil, logging.Errorf("getCNIConfigFromSpec: failed to unmarshal Spec.Config: %v", err)
        }

        // Inject network name if missing from Config for the thick plugin case
        if n, ok := rawConfig["name"]; !ok || n == "" {
                rawConfig["name"] = netName
                configBytes, err = json.Marshal(rawConfig)
                if err != nil {
                        return nil, logging.Errorf("getCNIConfigFromSpec: failed to re-marshal Spec.Config: %v", err)
                }
        }

        return configBytes, nil
}

func cniConfigFromNetworkResource(customResource *types.NetworkAttachmentDefinition, confdir string) ([]byte, error) {
        var config []byte
        var err error

        logging.Debugf("cniConfigFromNetworkResource: %v, %s", customResource, confdir)
        emptySpec := types.NetworkAttachmentDefinitionSpec{}
        if customResource.Spec == emptySpec {
                // Network Spec empty; generate delegate from CNI JSON config
                // from the configuration directory that has the same network
                // name as the custom resource
                config, err = getCNIConfigFromFile(customResource.Metadata.Name, confdir)
                if err != nil {
                        return nil, logging.Errorf("cniConfigFromNetworkResource: err in getCNIConfigFromFile: %v", err)
                }
        } else {
                // Config contains a standard JSON-encoded CNI configuration
                // or configuration list which defines the plugin chain to
                // execute.
                config, err = getCNIConfigFromSpec(customResource.Spec.Config, customResource.Metadata.Name)
                if err != nil {
                        return nil, logging.Errorf("cniConfigFromNetworkResource: err in getCNIConfigFromSpec: %v", err)
                }
        }

        return config, nil
}

func getKubernetesDelegate(client KubeClient, net *types.NetworkSelectionElement, confdir string, resourceMap map[string]*types.ResourceInfo) (*types.DelegateNetConf, error) {

        logging.Debugf("getKubernetesDelegate: %v, %v, %s", client, net, confdir)
        rawPath := fmt.Sprintf("/apis/k8s.cni.cncf.io/v1/namespaces/%s/network-attachment-definitions/%s", net.Namespace, net.Name)
        netData, err := client.GetRawWithPath(rawPath)
        if err != nil {
                return nil, logging.Errorf("getKubernetesDelegate: failed to get network resource, refer Multus README.md for the usage guide: %v", err)
        }

        customResource := &types.NetworkAttachmentDefinition{}
        if err := json.Unmarshal(netData, customResource); err != nil {
                return nil, logging.Errorf("getKubernetesDelegate: failed to get the netplugin data: %v", err)
        }

        // Get resourceName annotation from NetDefinition
        deviceID := ""
        resourceName, ok := customResource.Metadata.Annotations[resourceNameAnnot]
        if ok {
                // ResourceName annotation is found; try to get device info from resourceMap
                entry, ok := resourceMap[resourceName]
                if ok {
                        if idCount := len(entry.DeviceIDs); idCount > 0 && idCount > entry.Index {
                                deviceID = entry.DeviceIDs[entry.Index]
                                entry.Index++ // increment Index for next delegate
                        }
                }
        }

        configBytes, err := cniConfigFromNetworkResource(customResource, confdir)
        if err != nil {
                return nil, err
        }

        delegate, err := types.LoadDelegateNetConf(configBytes, net.InterfaceRequest, deviceID)
        if err != nil {
                return nil, err
        }

        return delegate, nil
}

type KubeClient interface {
        GetRawWithPath(path string) ([]byte, error)
        GetPod(namespace, name string) (*v1.Pod, error)
        UpdatePodStatus(pod *v1.Pod) (*v1.Pod, error)
}

func GetK8sArgs(args *skel.CmdArgs) (*types.K8sArgs, error) {
        k8sArgs := &types.K8sArgs{}

        logging.Debugf("GetK8sNetwork: %v", args)
        err := cnitypes.LoadArgs(args.Args, k8sArgs)
        if err != nil {
                return nil, err
        }

        return k8sArgs, nil
}

// Attempts to load Kubernetes-defined delegates and add them to the Multus config.
// Returns the number of Kubernetes-defined delegates added or an error.
func TryLoadK8sDelegates(k8sArgs *types.K8sArgs, conf *types.NetConf, kubeClient KubeClient) (int, *clientInfo, error) {
        var err error
        clientInfo := &clientInfo{}

        logging.Debugf("TryLoadK8sDelegates: %v, %v, %v", k8sArgs, conf, kubeClient)
        kubeClient, err = GetK8sClient(conf.Kubeconfig, kubeClient)
        if err != nil {
                return 0, nil, err
        }

        if kubeClient == nil {
                if len(conf.Delegates) == 0 {
                        // No available kube client and no delegates, we can't do anything
                        return 0, nil, logging.Errorf("must have either Kubernetes config or delegates, refer Multus README.md for the usage guide")
                }
                return 0, nil, nil
        }

        setKubeClientInfo(clientInfo, kubeClient, k8sArgs)
        delegates, err := GetK8sNetwork(kubeClient, k8sArgs, conf.ConfDir)
        if err != nil {
                if _, ok := err.(*NoK8sNetworkError); ok {
                        return 0, clientInfo, nil
                }
                return 0, nil, logging.Errorf("Multus: Err in getting k8s network from pod: %v", err)
        }

        if err = conf.AddDelegates(delegates); err != nil {
                return 0, nil, err
        }

        return len(delegates), clientInfo, nil
}

func GetK8sClient(kubeconfig string, kubeClient KubeClient) (KubeClient, error) {
        logging.Debugf("GetK8sClient: %s, %v", kubeconfig, kubeClient)
        // If we get a valid kubeClient (eg from testcases) just return that
        // one.
        if kubeClient != nil {
                return kubeClient, nil
        }

        var err error
        var config *rest.Config

        // Otherwise try to create a kubeClient from a given kubeConfig
        if kubeconfig != "" {
                // uses the current context in kubeconfig
                config, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
                if err != nil {
                        return nil, logging.Errorf("GetK8sClient: failed to get context for the kubeconfig %v, refer Multus README.md for the usage guide: %v", kubeconfig, err)
                }
        } else if os.Getenv("KUBERNETES_SERVICE_HOST") != "" && os.Getenv("KUBERNETES_SERVICE_PORT") != "" {
                // Try in-cluster config where multus might be running in a kubernetes pod
                config, err = rest.InClusterConfig()
                if err != nil {
                        return nil, logging.Errorf("createK8sClient: failed to get context for in-cluster kube config, refer Multus README.md for the usage guide: %v", err)
                }
        } else {
                // No kubernetes config; assume we shouldn't talk to Kube at all
                return nil, nil
        }

        // creates the clientset
        client, err := kubernetes.NewForConfig(config)
        if err != nil {
                return nil, err
        }

        return &defaultKubeClient{client: client}, nil
}

func GetK8sNetwork(k8sclient KubeClient, k8sArgs *types.K8sArgs, confdir string) ([]*types.DelegateNetConf, error) {
        logging.Debugf("GetK8sNetwork: %v, %v, %v", k8sclient, k8sArgs, confdir)

        netAnnot, defaultNamespace, podID, err := getPodNetworkAnnotation(k8sclient, k8sArgs)
        if err != nil {
                return nil, err
        }

        // Get Pod ComputeDevices info
        resourceMap, err := checkpoint.GetComputeDeviceMap(podID)
        if err != nil {
                return nil, logging.Errorf("GetK8sNetwork: failed to get resourceMap for PodUID: %v %v", podID, err)
        }

        if len(netAnnot) == 0 {
                return nil, &NoK8sNetworkError{"no kubernetes network found"}
        }

        networks, err := parsePodNetworkAnnotation(netAnnot, defaultNamespace)
        if err != nil {
                return nil, err
        }

        // Read all network objects referenced by 'networks'
        var delegates []*types.DelegateNetConf
        for _, net := range networks {
                delegate, err := getKubernetesDelegate(k8sclient, net, confdir, resourceMap)
                if err != nil {
                        return nil, logging.Errorf("GetK8sNetwork: failed getting the delegate: %v", err)
                }
                delegates = append(delegates, delegate)
        }

        return delegates, nil
}

1	// Copyright (c) 2017 Intel Corporation
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package k8sclient
16
17	import (
18	"encoding/json"
19	"fmt"
20	"os"
21	"regexp"
22	"strings"
23
24	v1 "k8s.io/api/core/v1"
25	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
26	"k8s.io/client-go/kubernetes"
27	"k8s.io/client-go/rest"
28	"k8s.io/client-go/tools/clientcmd"
29	"k8s.io/client-go/util/retry"
30
31	"github.com/containernetworking/cni/libcni"
32	"github.com/containernetworking/cni/pkg/skel"
33	cnitypes "github.com/containernetworking/cni/pkg/types"
34	"github.com/intel/multus-cni/checkpoint"
35	"github.com/intel/multus-cni/logging"
36	"github.com/intel/multus-cni/types"
37	)
38
39	const (
40	resourceNameAnnot = "k8s.v1.cni.cncf.io/resourceName"
41	)
42
43	// NoK8sNetworkError indicates error, no network in kubernetes
44	type NoK8sNetworkError struct {
45	message string
46	}
47
48	type clientInfo struct {
49	Client KubeClient
50	Podnamespace string
51	Podname string
52	}
53
54	func (e *NoK8sNetworkError) Error() string { return string(e.message) }	×
55
56	type defaultKubeClient struct {
57	client kubernetes.Interface
58	}
59
60	// defaultKubeClient implements KubeClient
61	var _ KubeClient = &defaultKubeClient{}
62
63	func (d *defaultKubeClient) GetRawWithPath(path string) ([]byte, error) {	×
64	return d.client.ExtensionsV1beta1().RESTClient().Get().AbsPath(path).DoRaw()	×
65	}	×
66
67	func (d defaultKubeClient) GetPod(namespace, name string) (v1.Pod, error) {	×
68	return d.client.Core().Pods(namespace).Get(name, metav1.GetOptions{})	×
69	}	×
70
71	func (d defaultKubeClient) UpdatePodStatus(pod v1.Pod) (*v1.Pod, error) {	×
72	return d.client.Core().Pods(pod.Namespace).UpdateStatus(pod)	×
73	}	×
74
75	func setKubeClientInfo(c clientInfo, client KubeClient, k8sArgs types.K8sArgs) {	×
76	logging.Debugf("setKubeClientInfo: %v, %v, %v", c, client, k8sArgs)	×
77	c.Client = client	×
78	c.Podnamespace = string(k8sArgs.K8S_POD_NAMESPACE)	×
79	c.Podname = string(k8sArgs.K8S_POD_NAME)	×
80	}	×
81
82	func SetNetworkStatus(k clientInfo, netStatus []types.NetworkStatus) error {	×
83		×
84	logging.Debugf("SetNetworkStatus: %v, %v", k, netStatus)	×
85	pod, err := k.Client.GetPod(k.Podnamespace, k.Podname)	×
86	if err != nil {	×
87	return logging.Errorf("SetNetworkStatus: failed to query the pod %v in out of cluster comm: %v", k.Podname, err)	×
88	}	×
89
90	var ns string	×
91	if netStatus != nil {	×
92	var networkStatus []string	×
93	for _, nets := range netStatus {	×
94	data, err := json.MarshalIndent(nets, "", " ")	×
95	if err != nil {	×
96	return logging.Errorf("SetNetworkStatus: error with Marshal Indent: %v", err)	×
97	}	×
98	networkStatus = append(networkStatus, string(data))	×
99	}
100
101	ns = fmt.Sprintf("[%s]", strings.Join(networkStatus, ","))	×
102	}
103	_, err = setPodNetworkAnnotation(k.Client, k.Podnamespace, pod, ns)	×
104	if err != nil {	×
105	return logging.Errorf("SetNetworkStatus: failed to update the pod %v in out of cluster comm: %v", k.Podname, err)	×
106	}	×
107
108	return nil	×
109	}
110
111	func setPodNetworkAnnotation(client KubeClient, namespace string, pod v1.Pod, networkstatus string) (v1.Pod, error) {	×
112	logging.Debugf("setPodNetworkAnnotation: %v, %s, %v, %s", client, namespace, pod, networkstatus)	×
113	//if pod annotations is empty, make sure it allocatable	×
114	if len(pod.Annotations) == 0 {	×
115	pod.Annotations = make(map[string]string)	×
116	}	×
117
118	pod.Annotations["k8s.v1.cni.cncf.io/networks-status"] = networkstatus	×
119		×
120	pod = pod.DeepCopy()	×
121	var err error	×
122	if resultErr := retry.RetryOnConflict(retry.DefaultBackoff, func() error {	×
123	if err != nil {	×
124	// Re-get the pod unless it's the first attempt to update	×
125	pod, err = client.GetPod(pod.Namespace, pod.Name)	×
126	if err != nil {	×
127	return err	×
128	}	×
129	}
130
131	pod, err = client.UpdatePodStatus(pod)	×
132	return err	×
133	}); resultErr != nil {	×
134	return nil, logging.Errorf("status update failed for pod %s/%s: %v", pod.Namespace, pod.Name, resultErr)	×
135	}	×
136	return pod, nil	×
137	}
138
139	func getPodNetworkAnnotation(client KubeClient, k8sArgs *types.K8sArgs) (string, string, string, error) {	7✔
140	var err error	7✔
141		7✔
142	logging.Debugf("getPodNetworkAnnotation: %v, %v", client, k8sArgs)	7✔
143	pod, err := client.GetPod(string(k8sArgs.K8S_POD_NAMESPACE), string(k8sArgs.K8S_POD_NAME))	7✔
144	if err != nil {	7✔
145	return "", "", "", logging.Errorf("getPodNetworkAnnotation: failed to query the pod %v in out of cluster comm: %v", string(k8sArgs.K8S_POD_NAME), err)	×
146	}	×
147
148	return pod.Annotations["k8s.v1.cni.cncf.io/networks"], pod.ObjectMeta.Namespace, string(pod.UID), nil	7✔
149	}
150
151	func parsePodNetworkObjectName(podnetwork string) (string, string, string, error) {	×
152	var netNsName string	×
153	var netIfName string	×
154	var networkName string	×
155		×
156	logging.Debugf("parsePodNetworkObjectName: %s", podnetwork)	×
157	slashItems := strings.Split(podnetwork, "/")	×
158	if len(slashItems) == 2 {	×
159	netNsName = strings.TrimSpace(slashItems[0])	×
160	networkName = slashItems[1]	×
161	} else if len(slashItems) == 1 {	×
162	networkName = slashItems[0]	×
163	} else {	×
164	return "", "", "", logging.Errorf("Invalid network object (failed at '/')")	×
165	}	×
166
167	atItems := strings.Split(networkName, "@")	×
168	networkName = strings.TrimSpace(atItems[0])	×
169	if len(atItems) == 2 {	×
170	netIfName = strings.TrimSpace(atItems[1])	×
171	} else if len(atItems) != 1 {	×
172	return "", "", "", logging.Errorf("Invalid network object (failed at '@')")	×
173	}	×
174
175	// Check and see if each item matches the specification for valid attachment name.
176	// "Valid attachment names must be comprised of units of the DNS-1123 label format"
177	// [a-z0-9]([-a-z0-9]*[a-z0-9])?
178	// And we allow at (@), and forward slash (/) (units separated by commas)
179	// It must start and end alphanumerically.
180	allItems := []string{netNsName, networkName, netIfName}	×
181	for i := range allItems {	×
182	matched, _ := regexp.MatchString("^[a-z0-9]([-a-z0-9]*[a-z0-9])?$", allItems[i])	×
183	if !matched && len([]rune(allItems[i])) > 0 {	×
184	return "", "", "", logging.Errorf(fmt.Sprintf("Failed to parse: one or more items did not match comma-delimited format (must consist of lower case alphanumeric characters). Must start and end with an alphanumeric character), mismatch @ '%v'", allItems[i]))	×
185	}	×
186	}
187
188	logging.Debugf("parsePodNetworkObjectName: parsed: %s, %s, %s", netNsName, networkName, netIfName)	×
189	return netNsName, networkName, netIfName, nil	×
190	}
191
192	func parsePodNetworkAnnotation(podNetworks, defaultNamespace string) ([]*types.NetworkSelectionElement, error) {	×
193	var networks []*types.NetworkSelectionElement	×
194		×
195	logging.Debugf("parsePodNetworkAnnotation: %s, %s", podNetworks, defaultNamespace)	×
196	if podNetworks == "" {	×
197	return nil, logging.Errorf("parsePodNetworkAnnotation: pod annotation not having \"network\" as key, refer Multus README.md for the usage guide")	×
198	}	×
199
200	if strings.IndexAny(podNetworks, "[{\"") >= 0 {	×
201	if err := json.Unmarshal([]byte(podNetworks), &networks); err != nil {	×
202	return nil, logging.Errorf("parsePodNetworkAnnotation: failed to parse pod Network Attachment Selection Annotation JSON format: %v", err)	×
203	}	×
204	} else {	×
205	// Comma-delimited list of network attachment object names	×
206	for _, item := range strings.Split(podNetworks, ",") {	×
207	// Remove leading and trailing whitespace.	×
208	item = strings.TrimSpace(item)	×
209		×
210	// Parse network name (i.e. <namespace>/<network name>@<ifname>)	×
211	netNsName, networkName, netIfName, err := parsePodNetworkObjectName(item)	×
212	if err != nil {	×
213	return nil, logging.Errorf("parsePodNetworkAnnotation: %v", err)	×
214	}	×
215
216	networks = append(networks, &types.NetworkSelectionElement{	×
217	Name: networkName,	×
218	Namespace: netNsName,	×
219	InterfaceRequest: netIfName,	×
220	})	×
221	}
222	}
223
224	for _, net := range networks {	×
225	if net.Namespace == "" {	×
226	net.Namespace = defaultNamespace	×
227	}	×
228	}
229
230	return networks, nil	×
231	}
232
233	func getCNIConfigFromFile(name string, confdir string) ([]byte, error) {	×
234	logging.Debugf("getCNIConfigFromFile: %s, %s", name, confdir)	×
235		×
236	// In the absence of valid keys in a Spec, the runtime (or	×
237	// meta-plugin) should load and execute a CNI .configlist	×
238	// or .config (in that order) file on-disk whose JSON	×
239	// “name” key matches this Network object’s name.	×
240		×
241	// In part, adapted from K8s pkg/kubelet/dockershim/network/cni/cni.go#getDefaultCNINetwork	×
242	files, err := libcni.ConfFiles(confdir, []string{".conf", ".json", ".conflist"})	×
243	switch {	×
244	case err != nil:	×
245	return nil, logging.Errorf("No networks found in %s", confdir)	×
246	case len(files) == 0:	×
247	return nil, logging.Errorf("No networks found in %s", confdir)	×
248	}
249
250	for _, confFile := range files {	×
251	var confList *libcni.NetworkConfigList	×
252	if strings.HasSuffix(confFile, ".conflist") {	×
253	confList, err = libcni.ConfListFromFile(confFile)	×
254	if err != nil {	×
255	return nil, logging.Errorf("Error loading CNI conflist file %s: %v", confFile, err)	×
256	}	×
257
258	if confList.Name == name {	×
259	return confList.Bytes, nil	×
260	}	×
261
262	} else {	×
263	conf, err := libcni.ConfFromFile(confFile)	×
264	if err != nil {	×
265	return nil, logging.Errorf("Error loading CNI config file %s: %v", confFile, err)	×
266	}	×
267
268	if conf.Network.Name == name {	×
269	// Ensure the config has a "type" so we know what plugin to run.	×
270	// Also catches the case where somebody put a conflist into a conf file.	×
271	if conf.Network.Type == "" {	×
272	return nil, logging.Errorf("Error loading CNI config file %s: no 'type'; perhaps this is a .conflist?", confFile)	×
273	}	×
274	return conf.Bytes, nil	×
275	}
276	}
277	}
278
279	return nil, logging.Errorf("no network available in the name %s in cni dir %s", name, confdir)	×
280	}
281
282	// getCNIConfigFromSpec reads a CNI JSON configuration from the NetworkAttachmentDefinition
283	// object's Spec.Config field and fills in any missing details like the network name
284	func getCNIConfigFromSpec(configData, netName string) ([]byte, error) {	×
285	var rawConfig map[string]interface{}	×
286	var err error	×
287		×
288	logging.Debugf("getCNIConfigFromSpec: %s, %s", configData, netName)	×
289	configBytes := []byte(configData)	×
290	err = json.Unmarshal(configBytes, &rawConfig)	×
291	if err != nil {	×
292	return nil, logging.Errorf("getCNIConfigFromSpec: failed to unmarshal Spec.Config: %v", err)	×
293	}	×
294
295	// Inject network name if missing from Config for the thick plugin case
296	if n, ok := rawConfig["name"]; !ok \|\| n == "" {	×
297	rawConfig["name"] = netName	×
298	configBytes, err = json.Marshal(rawConfig)	×
299	if err != nil {	×
300	return nil, logging.Errorf("getCNIConfigFromSpec: failed to re-marshal Spec.Config: %v", err)	×
301	}	×
302	}
303
304	return configBytes, nil	×
305	}
306
307	func cniConfigFromNetworkResource(customResource *types.NetworkAttachmentDefinition, confdir string) ([]byte, error) {	×
308	var config []byte	×
309	var err error	×
310		×
311	logging.Debugf("cniConfigFromNetworkResource: %v, %s", customResource, confdir)	×
312	emptySpec := types.NetworkAttachmentDefinitionSpec{}	×
313	if customResource.Spec == emptySpec {	×
314	// Network Spec empty; generate delegate from CNI JSON config	×
315	// from the configuration directory that has the same network	×
316	// name as the custom resource	×
317	config, err = getCNIConfigFromFile(customResource.Metadata.Name, confdir)	×
318	if err != nil {	×
319	return nil, logging.Errorf("cniConfigFromNetworkResource: err in getCNIConfigFromFile: %v", err)	×
320	}	×
321	} else {	×
322	// Config contains a standard JSON-encoded CNI configuration	×
323	// or configuration list which defines the plugin chain to	×
324	// execute.	×
325	config, err = getCNIConfigFromSpec(customResource.Spec.Config, customResource.Metadata.Name)	×
326	if err != nil {	×
327	return nil, logging.Errorf("cniConfigFromNetworkResource: err in getCNIConfigFromSpec: %v", err)	×
328	}	×
329	}
330
331	return config, nil	×
332	}
333
334	func getKubernetesDelegate(client KubeClient, net types.NetworkSelectionElement, confdir string, resourceMap map[string]types.ResourceInfo) (*types.DelegateNetConf, error) {	×
335		×
336	logging.Debugf("getKubernetesDelegate: %v, %v, %s", client, net, confdir)	×
337	rawPath := fmt.Sprintf("/apis/k8s.cni.cncf.io/v1/namespaces/%s/network-attachment-definitions/%s", net.Namespace, net.Name)	×
338	netData, err := client.GetRawWithPath(rawPath)	×
339	if err != nil {	×
340	return nil, logging.Errorf("getKubernetesDelegate: failed to get network resource, refer Multus README.md for the usage guide: %v", err)	×
341	}	×
342
343	customResource := &types.NetworkAttachmentDefinition{}	×
344	if err := json.Unmarshal(netData, customResource); err != nil {	×
345	return nil, logging.Errorf("getKubernetesDelegate: failed to get the netplugin data: %v", err)	×
346	}	×
347
348	// Get resourceName annotation from NetDefinition
349	deviceID := ""	×
350	resourceName, ok := customResource.Metadata.Annotations[resourceNameAnnot]	×
351	if ok {	×
352	// ResourceName annotation is found; try to get device info from resourceMap	×
353	entry, ok := resourceMap[resourceName]	×
354	if ok {	×
355	if idCount := len(entry.DeviceIDs); idCount > 0 && idCount > entry.Index {	×
356	deviceID = entry.DeviceIDs[entry.Index]	×
357	entry.Index++ // increment Index for next delegate	×
358	}	×
359	}
360	}
361
362	configBytes, err := cniConfigFromNetworkResource(customResource, confdir)	×
363	if err != nil {	×
364	return nil, err	×
365	}	×
366
367	delegate, err := types.LoadDelegateNetConf(configBytes, net.InterfaceRequest, deviceID)	×
368	if err != nil {	×
369	return nil, err	×
370	}	×
371
372	return delegate, nil	×
373	}
374
375	type KubeClient interface {
376	GetRawWithPath(path string) ([]byte, error)
377	GetPod(namespace, name string) (*v1.Pod, error)
378	UpdatePodStatus(pod v1.Pod) (v1.Pod, error)
379	}
380
381	func GetK8sArgs(args skel.CmdArgs) (types.K8sArgs, error) {	7✔
382	k8sArgs := &types.K8sArgs{}	7✔
383		7✔
384	logging.Debugf("GetK8sNetwork: %v", args)	7✔
385	err := cnitypes.LoadArgs(args.Args, k8sArgs)	7✔
386	if err != nil {	7✔
387	return nil, err	×
388	}	×
389
390	return k8sArgs, nil	7✔
391	}
392
393	// Attempts to load Kubernetes-defined delegates and add them to the Multus config.
394	// Returns the number of Kubernetes-defined delegates added or an error.
395	func TryLoadK8sDelegates(k8sArgs types.K8sArgs, conf types.NetConf, kubeClient KubeClient) (int, *clientInfo, error) {	×
396	var err error	×
397	clientInfo := &clientInfo{}	×
398		×
399	logging.Debugf("TryLoadK8sDelegates: %v, %v, %v", k8sArgs, conf, kubeClient)	×
400	kubeClient, err = GetK8sClient(conf.Kubeconfig, kubeClient)	×
401	if err != nil {	×
402	return 0, nil, err	×
403	}	×
404
405	if kubeClient == nil {	×
406	if len(conf.Delegates) == 0 {	×
407	// No available kube client and no delegates, we can't do anything	×
408	return 0, nil, logging.Errorf("must have either Kubernetes config or delegates, refer Multus README.md for the usage guide")	×
409	}	×
410	return 0, nil, nil	×
411	}
412
413	setKubeClientInfo(clientInfo, kubeClient, k8sArgs)	×
414	delegates, err := GetK8sNetwork(kubeClient, k8sArgs, conf.ConfDir)	×
415	if err != nil {	×
416	if _, ok := err.(*NoK8sNetworkError); ok {	×
417	return 0, clientInfo, nil	×
418	}	×
419	return 0, nil, logging.Errorf("Multus: Err in getting k8s network from pod: %v", err)	×
420	}
421
422	if err = conf.AddDelegates(delegates); err != nil {	×
423	return 0, nil, err	×
424	}	×
425
426	return len(delegates), clientInfo, nil	×
427	}
428
429	func GetK8sClient(kubeconfig string, kubeClient KubeClient) (KubeClient, error) {	7✔
430	logging.Debugf("GetK8sClient: %s, %v", kubeconfig, kubeClient)	7✔
431	// If we get a valid kubeClient (eg from testcases) just return that	7✔
432	// one.	7✔
433	if kubeClient != nil {	14✔
434	return kubeClient, nil	7✔
435	}	7✔
436
437	var err error	×
438	var config *rest.Config	×
439		×
440	// Otherwise try to create a kubeClient from a given kubeConfig	×
441	if kubeconfig != "" {	×
442	// uses the current context in kubeconfig	×
443	config, err = clientcmd.BuildConfigFromFlags("", kubeconfig)	×
444	if err != nil {	×
445	return nil, logging.Errorf("GetK8sClient: failed to get context for the kubeconfig %v, refer Multus README.md for the usage guide: %v", kubeconfig, err)	×
446	}	×
447	} else if os.Getenv("KUBERNETES_SERVICE_HOST") != "" && os.Getenv("KUBERNETES_SERVICE_PORT") != "" {	×
448	// Try in-cluster config where multus might be running in a kubernetes pod	×
449	config, err = rest.InClusterConfig()	×
450	if err != nil {	×
451	return nil, logging.Errorf("createK8sClient: failed to get context for in-cluster kube config, refer Multus README.md for the usage guide: %v", err)	×
452	}	×
453	} else {	×
454	// No kubernetes config; assume we shouldn't talk to Kube at all	×
455	return nil, nil	×
456	}	×
457
458	// creates the clientset
459	client, err := kubernetes.NewForConfig(config)	×
460	if err != nil {	×
461	return nil, err	×
462	}	×
463
464	return &defaultKubeClient{client: client}, nil	×
465	}
466
467	func GetK8sNetwork(k8sclient KubeClient, k8sArgs types.K8sArgs, confdir string) ([]types.DelegateNetConf, error) {	7✔
468	logging.Debugf("GetK8sNetwork: %v, %v, %v", k8sclient, k8sArgs, confdir)	7✔
469		7✔
470	netAnnot, defaultNamespace, podID, err := getPodNetworkAnnotation(k8sclient, k8sArgs)	7✔
471	if err != nil {	7✔
472	return nil, err	×
473	}	×
474
475	// Get Pod ComputeDevices info
476	resourceMap, err := checkpoint.GetComputeDeviceMap(podID)	7✔
477	if err != nil {	14✔
478	return nil, logging.Errorf("GetK8sNetwork: failed to get resourceMap for PodUID: %v %v", podID, err)	7✔
479	}	7✔
480
481	if len(netAnnot) == 0 {	×
482	return nil, &NoK8sNetworkError{"no kubernetes network found"}	×
483	}	×
484
485	networks, err := parsePodNetworkAnnotation(netAnnot, defaultNamespace)	×
486	if err != nil {	×
487	return nil, err	×
488	}	×
489
490	// Read all network objects referenced by 'networks'
491	var delegates []*types.DelegateNetConf	×
492	for _, net := range networks {	×
493	delegate, err := getKubernetesDelegate(k8sclient, net, confdir, resourceMap)	×
494	if err != nil {	×
495	return nil, logging.Errorf("GetK8sNetwork: failed getting the delegate: %v", err)	×
496	}	×
497	delegates = append(delegates, delegate)	×
498	}
499
500	return delegates, nil	×
501	}

intel / multus-cni / 379

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