3683

Committed 13 Jun 2019 11:59AM UTC coverage: 71.038% (-0.4%) from 71.466%

Build # 3683

Build Type

push

travis-ci

Committed by

awels

Commit Message

Merge pull request #843 from igoihman/fix-route-watch (#853)

update IsOpenShift function with OCP4 changes

(cherry picked from commit cc49ec208)
Signed-off-by: Irit goihman <igoihman@redhat.com>

Run Details

11 of 11 new or added lines in 1 file covered. (100.0%)

4312 of 6070 relevant lines covered (71.04%)

0.78 hits per line

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

62.35

/pkg/controller/upload-controller.go

/*
Copyright 2018 The CDI 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 controller

import (
        "time"

        "github.com/pkg/errors"
        v1 "k8s.io/api/core/v1"
        k8serrors "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/apimachinery/pkg/util/runtime"
        "k8s.io/apimachinery/pkg/util/wait"
        coreinformers "k8s.io/client-go/informers/core/v1"
        "k8s.io/client-go/kubernetes"
        corelisters "k8s.io/client-go/listers/core/v1"
        "k8s.io/client-go/tools/cache"
        "k8s.io/client-go/util/cert/triple"
        "k8s.io/client-go/util/workqueue"
        "k8s.io/klog"
        "kubevirt.io/containerized-data-importer/pkg/keys"
        "kubevirt.io/containerized-data-importer/pkg/util"

        clientset "kubevirt.io/containerized-data-importer/pkg/client/clientset/versioned"
)

const (
        // AnnUploadRequest marks that a PVC should be made available for upload
        AnnUploadRequest = "cdi.kubevirt.io/storage.upload.target"

        annCreatedByUpload = "cdi.kubevirt.io/storage.createdByUploadController"

        // cert/key annotations

        uploadServerCASecret = "cdi-upload-server-ca-key"
        uploadServerCAName   = "server.upload.cdi.kubevirt.io"

        uploadServerClientCASecret = "cdi-upload-server-client-ca-key"
        uploadServerClientCAName   = "client.upload-server.cdi.kubevirt.io"

        uploadServerClientKeySecret = "cdi-upload-server-client-key"
        uploadProxyClientName       = "uploadproxy.client.upload-server.cdi.kebevirt.io"

        uploadProxyCASecret     = "cdi-upload-proxy-ca-key"
        uploadProxyServerSecret = "cdi-upload-proxy-server-key"
        uploadProxyCAName       = "proxy.upload.cdi.kubevirt.io"
)

// UploadController members
type UploadController struct {
        client                                    kubernetes.Interface
        cdiClient                                 clientset.Interface
        queue                                     workqueue.RateLimitingInterface
        pvcInformer, podInformer, serviceInformer cache.SharedIndexInformer
        pvcLister                                 corelisters.PersistentVolumeClaimLister
        podLister                                 corelisters.PodLister
        serviceLister                             corelisters.ServiceLister
        pvcsSynced                                cache.InformerSynced
        podsSynced                                cache.InformerSynced
        servicesSynced                            cache.InformerSynced
        uploadServiceImage                        string
        pullPolicy                                string // Options: IfNotPresent, Always, or Never
        verbose                                   string // verbose levels: 1, 2, ...
        serverCAKeyPair                           *triple.KeyPair
        clientCAKeyPair                           *triple.KeyPair
        uploadProxyServiceName                    string
}

// GetUploadResourceName returns the name given to upload services/pods
func GetUploadResourceName(pvcName string) string {
        return "cdi-upload-" + pvcName
}

// UploadPossibleForPVC is called by the api server to see whether to return an upload token
func UploadPossibleForPVC(pvc *v1.PersistentVolumeClaim) error {
        if _, ok := pvc.Annotations[AnnUploadRequest]; !ok {
                return errors.Errorf("PVC %s is not an upload target", pvc.Name)
        }

        return nil
}

// NewUploadController returns a new UploadController
func NewUploadController(client kubernetes.Interface,
        cdiClientSet clientset.Interface,
        pvcInformer coreinformers.PersistentVolumeClaimInformer,
        podInformer coreinformers.PodInformer,
        serviceInformer coreinformers.ServiceInformer,
        uploadServiceImage string,
        uploadProxyServiceName string,
        pullPolicy string,
        verbose string) *UploadController {
        c := &UploadController{
                client:                 client,
                cdiClient:              cdiClientSet,
                queue:                  workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
                pvcInformer:            pvcInformer.Informer(),
                podInformer:            podInformer.Informer(),
                serviceInformer:        serviceInformer.Informer(),
                pvcLister:              pvcInformer.Lister(),
                podLister:              podInformer.Lister(),
                serviceLister:          serviceInformer.Lister(),
                pvcsSynced:             pvcInformer.Informer().HasSynced,
                podsSynced:             podInformer.Informer().HasSynced,
                servicesSynced:         serviceInformer.Informer().HasSynced,
                uploadServiceImage:     uploadServiceImage,
                uploadProxyServiceName: uploadProxyServiceName,
                pullPolicy:             pullPolicy,
                verbose:                verbose,
        }

        // Bind the pvc SharedIndexInformer to the pvc queue
        c.pvcInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: c.enqueueObject,
                UpdateFunc: func(old, new interface{}) {
                        c.enqueueObject(new)
                },
        })

        // Bind the pod SharedIndexInformer to the pvc queue
        c.podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: c.handleObject,
                UpdateFunc: func(old, new interface{}) {
                        newDepl := new.(*v1.Pod)
                        oldDepl := old.(*v1.Pod)
                        if newDepl.ResourceVersion == oldDepl.ResourceVersion {
                                // Periodic resync will send update events for all known Pods.
                                // Two different versions of the same PVCs will always have different RVs.
                                return
                        }
                        c.handleObject(new)
                },
                DeleteFunc: c.handleObject,
        })

        // Bind the service SharedIndexInformer to the service queue
        c.serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
                AddFunc: c.handleObject,
                UpdateFunc: func(old, new interface{}) {
                        newDepl := new.(*v1.Service)
                        oldDepl := old.(*v1.Service)
                        if newDepl.ResourceVersion == oldDepl.ResourceVersion {
                                // Periodic resync will send update events for all known Services.
                                // Two different versions of the same Servicess will always have different RVs.
                                return
                        }
                        c.handleObject(new)
                },
                DeleteFunc: c.handleObject,
        })

        return c
}

// Init does synchronous initialization before being considered "ready"
func (c *UploadController) Init() error {
        klog.V(2).Infoln("Getting/creating certs")

        if err := c.initCerts(); err != nil {
                runtime.HandleError(err)
                return errors.Wrap(err, "Error initializing certificates")
        }

        return nil
}

// Run sets up UploadController state and executes main event loop
func (c *UploadController) Run(threadiness int, stopCh <-chan struct{}) error {
        defer runtime.HandleCrash()
        defer c.queue.ShutDown()

        klog.V(2).Infoln("Starting cdi upload controller Run loop")

        if threadiness < 1 {
                return errors.Errorf("expected >0 threads, got %d", threadiness)
        }

        klog.V(3).Info("Waiting for informer caches to sync")

        if ok := cache.WaitForCacheSync(stopCh, c.pvcsSynced, c.podsSynced, c.servicesSynced); !ok {
                return errors.New("failed to wait for caches to sync")
        }

        klog.V(3).Infoln("UploadController cache has synced")

        for i := 0; i < threadiness; i++ {
                go wait.Until(c.runWorker, time.Second, stopCh)
        }

        klog.Info("Started workers")
        <-stopCh
        klog.Info("Shutting down workers")

        return nil
}

func (c *UploadController) initCerts() error {
        var err error

        // CA for Upload Servers
        c.serverCAKeyPair, err = keys.GetOrCreateCA(c.client, util.GetNamespace(), uploadServerCASecret, uploadServerCAName)
        if err != nil {
                return errors.Wrap(err, "Couldn't get/create server CA")
        }

        // CA for Upload Client
        c.clientCAKeyPair, err = keys.GetOrCreateCA(c.client, util.GetNamespace(), uploadServerClientCASecret, uploadServerClientCAName)
        if err != nil {
                return errors.Wrap(err, "Couldn't get/create client CA")
        }

        // Upload Server Client Cert
        _, err = keys.GetOrCreateClientKeyPairAndCert(c.client,
                util.GetNamespace(),
                uploadServerClientKeySecret,
                c.clientCAKeyPair,
                c.serverCAKeyPair.Cert,
                uploadProxyClientName,
                []string{},
                nil,
        )
        if err != nil {
                return errors.Wrap(err, "Couldn't get/create client cert")
        }

        uploadProxyCAKeyPair, err := keys.GetOrCreateCA(c.client, util.GetNamespace(), uploadProxyCASecret, uploadProxyCAName)
        if err != nil {
                return errors.Wrap(err, "Couldn't create upload proxy server cert")
        }

        _, err = keys.GetOrCreateServerKeyPairAndCert(c.client,
                util.GetNamespace(),
                uploadProxyServerSecret,
                uploadProxyCAKeyPair,
                nil,
                c.uploadProxyServiceName+"."+util.GetNamespace(),
                c.uploadProxyServiceName,
                nil,
        )
        if err != nil {
                return errors.Wrap(err, "Error creating upload proxy server key pair")
        }

        return nil
}

func (c *UploadController) handleObject(obj interface{}) {
        var object metav1.Object
        var ok bool
        if object, ok = obj.(metav1.Object); !ok {
                tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
                if !ok {
                        runtime.HandleError(errors.Errorf("error decoding object, invalid type"))
                        return
                }
                object, ok = tombstone.Obj.(metav1.Object)
                if !ok {
                        runtime.HandleError(errors.Errorf("error decoding object tombstone, invalid type"))
                        return
                }
                klog.V(3).Infof("Recovered deleted object '%s' from tombstone", object.GetName())
        }
        klog.V(3).Infof("Processing object: %s", object.GetName())
        if ownerRef := metav1.GetControllerOf(object); ownerRef != nil {
                _, createdByUs := object.GetAnnotations()[annCreatedByUpload]

                if ownerRef.Kind != "PersistentVolumeClaim" || !createdByUs {
                        return
                }

                pvc, err := c.pvcLister.PersistentVolumeClaims(object.GetNamespace()).Get(ownerRef.Name)
                if err != nil {
                        klog.V(3).Infof("ignoring orphaned object '%s' of pvc '%s'", object.GetSelfLink(), ownerRef.Name)
                        return
                }

                klog.V(3).Infof("queueing pvc %+v!!", pvc)

                c.enqueueObject(pvc)
                return
        }
}

func (c *UploadController) enqueueObject(obj interface{}) {
        key, err := cache.MetaNamespaceKeyFunc(obj)
        if err != nil {
                runtime.HandleError(err)
                return
        }
        c.queue.AddRateLimited(key)
}

func (c *UploadController) runWorker() {
        for c.processNextWorkItem() {
        }
}

func (c *UploadController) processNextWorkItem() bool {
        obj, shutdown := c.queue.Get()

        if shutdown {
                return false
        }

        err := func(obj interface{}) error {
                defer c.queue.Done(obj)

                var key string
                var ok bool

                if key, ok = obj.(string); !ok {
                        c.queue.Forget(obj)
                        runtime.HandleError(errors.Errorf("expected string in workqueue but got %#v", obj))
                        return nil
                }

                if err := c.syncHandler(key); err != nil {
                        return errors.Errorf("error syncing '%s': %s", key, err.Error())
                }

                c.queue.Forget(obj)
                klog.Infof("Successfully synced '%s'", key)
                return nil

        }(obj)

        if err != nil {
                runtime.HandleError(err)
                return true
        }

        return true
}

func podPhaseFromPVC(pvc *v1.PersistentVolumeClaim) v1.PodPhase {
        phase, _ := pvc.ObjectMeta.Annotations[AnnPodPhase]
        return v1.PodPhase(phase)
}

func podSucceededFromPVC(pvc *v1.PersistentVolumeClaim) bool {
        return (podPhaseFromPVC(pvc) == v1.PodSucceeded)
}

func (c *UploadController) syncHandler(key string) error {
        namespace, name, err := cache.SplitMetaNamespaceKey(key)
        if err != nil {
                runtime.HandleError(errors.Errorf("invalid resource key: %s", key))
                return nil
        }

        pvc, err := c.pvcLister.PersistentVolumeClaims(namespace).Get(name)
        if err != nil {
                if k8serrors.IsNotFound(err) {
                        runtime.HandleError(errors.Errorf("PVC '%s' in work queue no longer exists", key))
                        return nil
                }
                return errors.Wrapf(err, "error getting PVC %s", key)
        }

        _, isUploadPod := pvc.ObjectMeta.Annotations[AnnUploadRequest]
        resourceName := GetUploadResourceName(pvc.Name)

        // force cleanup if PVC pending delete and pod running or the upload annotation was removed
        if !isUploadPod || (pvc.DeletionTimestamp != nil && !podSucceededFromPVC(pvc)) {
                // delete everything

                // delete service
                err = c.deleteService(pvc.Namespace, resourceName)
                if err != nil {
                        return errors.Wrapf(err, "Error deleting upload service for pvc: %s", key)
                }

                // delete pod
                // we're using a req struct for now until we can normalize the controllers a bit more and share things like lister, client etc
                // this way it's easy to stuff everything into an easy request struct, and can extend aditional behaviors if we want going forward
                // NOTE this is a special case where the user updated annotations on the pvc to abort the upload requests, we'll add another call
                // for this for the success case
                dReq := podDeleteRequest{
                        namespace: pvc.Namespace,
                        podName:   resourceName,
                        podLister: c.podLister,
                        k8sClient: c.client,
                }

                err = deletePod(dReq)
                if err != nil {
                        return errors.Wrapf(err, "Error deleting upload pod for pvc: %s", key)
                }

                return nil
        }

        var pod *v1.Pod
        if !podSucceededFromPVC(pvc) {
                pod, err = c.getOrCreateUploadPod(pvc, resourceName)
                if err != nil {
                        return errors.Wrapf(err, "Error creating upload pod for pvc: %s", key)
                }

                podPhase := pod.Status.Phase
                if podPhase != podPhaseFromPVC(pvc) {
                        var labels map[string]string
                        annotations := map[string]string{AnnPodPhase: string(podPhase)}
                        pvc, err = updatePVC(c.client, pvc, annotations, labels)
                        if err != nil {
                                return errors.Wrapf(err, "Error updating pvc %s, pod phase %s", key, podPhase)
                        }
                }
        }

        if podSucceededFromPVC(pvc) {
                // delete service
                if err = c.deleteService(pvc.Namespace, resourceName); err != nil {
                        return errors.Wrapf(err, "Error deleting upload service for pvc %s", key)
                }

                dReq := podDeleteRequest{
                        namespace: pvc.Namespace,
                        podName:   resourceName,
                        podLister: c.podLister,
                        k8sClient: c.client,
                }

                // delete the pod
                err = deletePod(dReq)
                if err != nil {
                        return errors.Wrapf(err, "Error deleting upload pod for pvc: %s", key)
                }
        } else {
                // make sure the service exists
                if _, err = c.getOrCreateUploadService(pvc, resourceName); err != nil {
                        return errors.Wrapf(err, "Error getting/creating service resource for PVC %s", key)
                }
        }

        return nil
}

func (c *UploadController) getOrCreateUploadPod(pvc *v1.PersistentVolumeClaim, name string) (*v1.Pod, error) {
        pod, err := c.podLister.Pods(pvc.Namespace).Get(name)

        if k8serrors.IsNotFound(err) {
                pod, err = CreateUploadPod(c.client, c.serverCAKeyPair, c.clientCAKeyPair.Cert, c.uploadServiceImage, c.verbose, c.pullPolicy, name, pvc, pvc.Name+"-scratch")
        }
        // Always try to get or create the scratch PVC for a pod that is not successful yet, if it exists nothing happens otherwise attempt to create.
        _, err = c.getOrCreateScratchPvc(pvc, pod)
        if err != nil {
                return nil, errors.Wrap(err, "Unable to create scratch space for upload")
        }

        if pod != nil && !metav1.IsControlledBy(pod, pvc) {
                return nil, errors.Errorf("%s pod not controlled by pvc %s", name, pvc.Name)
        }
        return pod, err
}

func (c *UploadController) getOrCreateScratchPvc(pvc *v1.PersistentVolumeClaim, pod *v1.Pod) (*v1.PersistentVolumeClaim, error) {
        scratchPvc, err := c.pvcLister.PersistentVolumeClaims(pvc.Namespace).Get(pvc.Name + "-scratch")
        if scratchPvc != nil {
                // Scratch PVC already exists, maybe the pod crashed and the pvc didn't get cleaned up, verify this pvc is owned by the pod.
                for _, ownerRef := range scratchPvc.OwnerReferences {
                        if ownerRef.UID == pod.UID {
                                return scratchPvc, nil
                        }
                }
                return nil, errors.New("Scratch PVC exists, but is not owned by the right pod")
        }
        storageClassName := GetScratchPvcStorageClass(c.client, c.cdiClient, pvc)

        // Scratch PVC doesn't exist yet, create it. Determine which storage class to use.
        scratchPvc, err = CreateScratchPersistentVolumeClaim(c.client, pvc, pod, storageClassName)
        if err != nil {
                return nil, err
        }
        return scratchPvc, nil
}

func (c *UploadController) getOrCreateUploadService(pvc *v1.PersistentVolumeClaim, name string) (*v1.Service, error) {
        service, err := c.serviceLister.Services(pvc.Namespace).Get(name)

        if k8serrors.IsNotFound(err) {
                service, err = CreateUploadService(c.client, name, pvc)
        }

        if service != nil && !metav1.IsControlledBy(service, pvc) {
                return nil, errors.Errorf("%s service not controlled by pvc %s", name, pvc.Name)
        }

        return service, err
}

func (c *UploadController) deleteService(namespace, name string) error {
        service, err := c.serviceLister.Services(namespace).Get(name)
        if k8serrors.IsNotFound(err) {
                return nil
        }
        if err == nil && service.DeletionTimestamp == nil {
                err = c.client.CoreV1().Services(namespace).Delete(name, &metav1.DeleteOptions{})
                if k8serrors.IsNotFound(err) {
                        return nil
                }
        }
        return err
}

1	/*
2	Copyright 2018 The CDI Authors.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package controller
18
19	import (
20	"time"
21
22	"github.com/pkg/errors"
23	v1 "k8s.io/api/core/v1"
24	k8serrors "k8s.io/apimachinery/pkg/api/errors"
25	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
26	"k8s.io/apimachinery/pkg/util/runtime"
27	"k8s.io/apimachinery/pkg/util/wait"
28	coreinformers "k8s.io/client-go/informers/core/v1"
29	"k8s.io/client-go/kubernetes"
30	corelisters "k8s.io/client-go/listers/core/v1"
31	"k8s.io/client-go/tools/cache"
32	"k8s.io/client-go/util/cert/triple"
33	"k8s.io/client-go/util/workqueue"
34	"k8s.io/klog"
35	"kubevirt.io/containerized-data-importer/pkg/keys"
36	"kubevirt.io/containerized-data-importer/pkg/util"
37
38	clientset "kubevirt.io/containerized-data-importer/pkg/client/clientset/versioned"
39	)
40
41	const (
42	// AnnUploadRequest marks that a PVC should be made available for upload
43	AnnUploadRequest = "cdi.kubevirt.io/storage.upload.target"
44
45	annCreatedByUpload = "cdi.kubevirt.io/storage.createdByUploadController"
46
47	// cert/key annotations
48
49	uploadServerCASecret = "cdi-upload-server-ca-key"
50	uploadServerCAName = "server.upload.cdi.kubevirt.io"
51
52	uploadServerClientCASecret = "cdi-upload-server-client-ca-key"
53	uploadServerClientCAName = "client.upload-server.cdi.kubevirt.io"
54
55	uploadServerClientKeySecret = "cdi-upload-server-client-key"
56	uploadProxyClientName = "uploadproxy.client.upload-server.cdi.kebevirt.io"
57
58	uploadProxyCASecret = "cdi-upload-proxy-ca-key"
59	uploadProxyServerSecret = "cdi-upload-proxy-server-key"
60	uploadProxyCAName = "proxy.upload.cdi.kubevirt.io"
61	)
62
63	// UploadController members
64	type UploadController struct {
65	client kubernetes.Interface
66	cdiClient clientset.Interface
67	queue workqueue.RateLimitingInterface
68	pvcInformer, podInformer, serviceInformer cache.SharedIndexInformer
69	pvcLister corelisters.PersistentVolumeClaimLister
70	podLister corelisters.PodLister
71	serviceLister corelisters.ServiceLister
72	pvcsSynced cache.InformerSynced
73	podsSynced cache.InformerSynced
74	servicesSynced cache.InformerSynced
75	uploadServiceImage string
76	pullPolicy string // Options: IfNotPresent, Always, or Never
77	verbose string // verbose levels: 1, 2, ...
78	serverCAKeyPair *triple.KeyPair
79	clientCAKeyPair *triple.KeyPair
80	uploadProxyServiceName string
81	}
82
83	// GetUploadResourceName returns the name given to upload services/pods
84	func GetUploadResourceName(pvcName string) string {	1✔
85	return "cdi-upload-" + pvcName	1✔
86	}	1✔
87
88	// UploadPossibleForPVC is called by the api server to see whether to return an upload token
89	func UploadPossibleForPVC(pvc *v1.PersistentVolumeClaim) error {	1✔
90	if _, ok := pvc.Annotations[AnnUploadRequest]; !ok {	2✔
91	return errors.Errorf("PVC %s is not an upload target", pvc.Name)	1✔
92	}	1✔
93
94	return nil	1✔
95	}
96
97	// NewUploadController returns a new UploadController
98	func NewUploadController(client kubernetes.Interface,
99	cdiClientSet clientset.Interface,
100	pvcInformer coreinformers.PersistentVolumeClaimInformer,
101	podInformer coreinformers.PodInformer,
102	serviceInformer coreinformers.ServiceInformer,
103	uploadServiceImage string,
104	uploadProxyServiceName string,
105	pullPolicy string,
106	verbose string) *UploadController {	1✔
107	c := &UploadController{	1✔
108	client: client,	1✔
109	cdiClient: cdiClientSet,	1✔
110	queue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),	1✔
111	pvcInformer: pvcInformer.Informer(),	1✔
112	podInformer: podInformer.Informer(),	1✔
113	serviceInformer: serviceInformer.Informer(),	1✔
114	pvcLister: pvcInformer.Lister(),	1✔
115	podLister: podInformer.Lister(),	1✔
116	serviceLister: serviceInformer.Lister(),	1✔
117	pvcsSynced: pvcInformer.Informer().HasSynced,	1✔
118	podsSynced: podInformer.Informer().HasSynced,	1✔
119	servicesSynced: serviceInformer.Informer().HasSynced,	1✔
120	uploadServiceImage: uploadServiceImage,	1✔
121	uploadProxyServiceName: uploadProxyServiceName,	1✔
122	pullPolicy: pullPolicy,	1✔
123	verbose: verbose,	1✔
124	}	1✔
125		1✔
126	// Bind the pvc SharedIndexInformer to the pvc queue	1✔
127	c.pvcInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{	1✔
128	AddFunc: c.enqueueObject,	1✔
129	UpdateFunc: func(old, new interface{}) {	2✔
130	c.enqueueObject(new)	1✔
131	},	1✔
132	})
133
134	// Bind the pod SharedIndexInformer to the pvc queue
135	c.podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{	1✔
136	AddFunc: c.handleObject,	1✔
137	UpdateFunc: func(old, new interface{}) {	2✔
138	newDepl := new.(*v1.Pod)	1✔
139	oldDepl := old.(*v1.Pod)	1✔
140	if newDepl.ResourceVersion == oldDepl.ResourceVersion {	2✔
141	// Periodic resync will send update events for all known Pods.	1✔
142	// Two different versions of the same PVCs will always have different RVs.	1✔
143	return	1✔
144	}	1✔
145	c.handleObject(new)	×
146	},
147	DeleteFunc: c.handleObject,
148	})
149
150	// Bind the service SharedIndexInformer to the service queue
151	c.serviceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{	1✔
152	AddFunc: c.handleObject,	1✔
153	UpdateFunc: func(old, new interface{}) {	1✔
154	newDepl := new.(*v1.Service)	×
155	oldDepl := old.(*v1.Service)	×
156	if newDepl.ResourceVersion == oldDepl.ResourceVersion {	×
157	// Periodic resync will send update events for all known Services.	×
158	// Two different versions of the same Servicess will always have different RVs.	×
159	return	×
160	}	×
161	c.handleObject(new)	×
162	},
163	DeleteFunc: c.handleObject,
164	})
165
166	return c	1✔
167	}
168
169	// Init does synchronous initialization before being considered "ready"
170	func (c *UploadController) Init() error {	×
171	klog.V(2).Infoln("Getting/creating certs")	×
172		×
173	if err := c.initCerts(); err != nil {	×
174	runtime.HandleError(err)	×
175	return errors.Wrap(err, "Error initializing certificates")	×
176	}	×
177
178	return nil	×
179	}
180
181	// Run sets up UploadController state and executes main event loop
182	func (c *UploadController) Run(threadiness int, stopCh <-chan struct{}) error {	×
183	defer runtime.HandleCrash()	×
184	defer c.queue.ShutDown()	×
185		×
186	klog.V(2).Infoln("Starting cdi upload controller Run loop")	×
187		×
188	if threadiness < 1 {	×
189	return errors.Errorf("expected >0 threads, got %d", threadiness)	×
190	}	×
191
192	klog.V(3).Info("Waiting for informer caches to sync")	×
193		×
194	if ok := cache.WaitForCacheSync(stopCh, c.pvcsSynced, c.podsSynced, c.servicesSynced); !ok {	×
195	return errors.New("failed to wait for caches to sync")	×
196	}	×
197
198	klog.V(3).Infoln("UploadController cache has synced")	×
199		×
200	for i := 0; i < threadiness; i++ {	×
201	go wait.Until(c.runWorker, time.Second, stopCh)	×
202	}	×
203
204	klog.Info("Started workers")	×
205	<-stopCh	×
206	klog.Info("Shutting down workers")	×
207		×
208	return nil	×
209	}
210
211	func (c *UploadController) initCerts() error {	1✔
212	var err error	1✔
213		1✔
214	// CA for Upload Servers	1✔
215	c.serverCAKeyPair, err = keys.GetOrCreateCA(c.client, util.GetNamespace(), uploadServerCASecret, uploadServerCAName)	1✔
216	if err != nil {	1✔
217	return errors.Wrap(err, "Couldn't get/create server CA")	×
218	}	×
219
220	// CA for Upload Client
221	c.clientCAKeyPair, err = keys.GetOrCreateCA(c.client, util.GetNamespace(), uploadServerClientCASecret, uploadServerClientCAName)	1✔
222	if err != nil {	1✔
223	return errors.Wrap(err, "Couldn't get/create client CA")	×
224	}	×
225
226	// Upload Server Client Cert
227	_, err = keys.GetOrCreateClientKeyPairAndCert(c.client,	1✔
228	util.GetNamespace(),	1✔
229	uploadServerClientKeySecret,	1✔
230	c.clientCAKeyPair,	1✔
231	c.serverCAKeyPair.Cert,	1✔
232	uploadProxyClientName,	1✔
233	[]string{},	1✔
234	nil,	1✔
235	)	1✔
236	if err != nil {	1✔
237	return errors.Wrap(err, "Couldn't get/create client cert")	×
238	}	×
239
240	uploadProxyCAKeyPair, err := keys.GetOrCreateCA(c.client, util.GetNamespace(), uploadProxyCASecret, uploadProxyCAName)	1✔
241	if err != nil {	1✔
242	return errors.Wrap(err, "Couldn't create upload proxy server cert")	×
243	}	×
244
245	_, err = keys.GetOrCreateServerKeyPairAndCert(c.client,	1✔
246	util.GetNamespace(),	1✔
247	uploadProxyServerSecret,	1✔
248	uploadProxyCAKeyPair,	1✔
249	nil,	1✔
250	c.uploadProxyServiceName+"."+util.GetNamespace(),	1✔
251	c.uploadProxyServiceName,	1✔
252	nil,	1✔
253	)	1✔
254	if err != nil {	1✔
255	return errors.Wrap(err, "Error creating upload proxy server key pair")	×
256	}	×
257
258	return nil	1✔
259	}
260
261	func (c *UploadController) handleObject(obj interface{}) {	1✔
262	var object metav1.Object	1✔
263	var ok bool	1✔
264	if object, ok = obj.(metav1.Object); !ok {	2✔
265	tombstone, ok := obj.(cache.DeletedFinalStateUnknown)	1✔
266	if !ok {	1✔
267	runtime.HandleError(errors.Errorf("error decoding object, invalid type"))	×
268	return	×
269	}	×
270	object, ok = tombstone.Obj.(metav1.Object)	1✔
271	if !ok {	1✔
272	runtime.HandleError(errors.Errorf("error decoding object tombstone, invalid type"))	×
273	return	×
274	}	×
275	klog.V(3).Infof("Recovered deleted object '%s' from tombstone", object.GetName())	1✔
276	}
277	klog.V(3).Infof("Processing object: %s", object.GetName())	1✔
278	if ownerRef := metav1.GetControllerOf(object); ownerRef != nil {	2✔
279	_, createdByUs := object.GetAnnotations()[annCreatedByUpload]	1✔
280		1✔
281	if ownerRef.Kind != "PersistentVolumeClaim" \|\| !createdByUs {	1✔
282	return	×
283	}	×
284
285	pvc, err := c.pvcLister.PersistentVolumeClaims(object.GetNamespace()).Get(ownerRef.Name)	1✔
286	if err != nil {	1✔
287	klog.V(3).Infof("ignoring orphaned object '%s' of pvc '%s'", object.GetSelfLink(), ownerRef.Name)	×
288	return	×
289	}	×
290
291	klog.V(3).Infof("queueing pvc %+v!!", pvc)	1✔
292		1✔
293	c.enqueueObject(pvc)	1✔
294	return	1✔
295	}
296	}
297
298	func (c *UploadController) enqueueObject(obj interface{}) {	1✔
299	key, err := cache.MetaNamespaceKeyFunc(obj)	1✔
300	if err != nil {	1✔
301	runtime.HandleError(err)	×
302	return	×
303	}	×
304	c.queue.AddRateLimited(key)	1✔
305	}
306
307	func (c *UploadController) runWorker() {	×
308	for c.processNextWorkItem() {	×
309	}	×
310	}
311
312	func (c *UploadController) processNextWorkItem() bool {	×
313	obj, shutdown := c.queue.Get()	×
314		×
315	if shutdown {	×
316	return false	×
317	}	×
318
319	err := func(obj interface{}) error {	×
320	defer c.queue.Done(obj)	×
321		×
322	var key string	×
323	var ok bool	×
324		×
325	if key, ok = obj.(string); !ok {	×
326	c.queue.Forget(obj)	×
327	runtime.HandleError(errors.Errorf("expected string in workqueue but got %#v", obj))	×
328	return nil	×
329	}	×
330
331	if err := c.syncHandler(key); err != nil {	×
332	return errors.Errorf("error syncing '%s': %s", key, err.Error())	×
333	}	×
334
335	c.queue.Forget(obj)	×
336	klog.Infof("Successfully synced '%s'", key)	×
337	return nil	×
338
339	}(obj)
340
341	if err != nil {	×
342	runtime.HandleError(err)	×
343	return true	×
344	}	×
345
346	return true	×
347	}
348
349	func podPhaseFromPVC(pvc *v1.PersistentVolumeClaim) v1.PodPhase {	1✔
350	phase, _ := pvc.ObjectMeta.Annotations[AnnPodPhase]	1✔
351	return v1.PodPhase(phase)	1✔
352	}	1✔
353
354	func podSucceededFromPVC(pvc *v1.PersistentVolumeClaim) bool {	1✔
355	return (podPhaseFromPVC(pvc) == v1.PodSucceeded)	1✔
356	}	1✔
357
358	func (c *UploadController) syncHandler(key string) error {	1✔
359	namespace, name, err := cache.SplitMetaNamespaceKey(key)	1✔
360	if err != nil {	1✔
361	runtime.HandleError(errors.Errorf("invalid resource key: %s", key))	×
362	return nil	×
363	}	×
364
365	pvc, err := c.pvcLister.PersistentVolumeClaims(namespace).Get(name)	1✔
366	if err != nil {	2✔
367	if k8serrors.IsNotFound(err) {	2✔
368	runtime.HandleError(errors.Errorf("PVC '%s' in work queue no longer exists", key))	1✔
369	return nil	1✔
370	}	1✔
371	return errors.Wrapf(err, "error getting PVC %s", key)	×
372	}
373
374	_, isUploadPod := pvc.ObjectMeta.Annotations[AnnUploadRequest]	1✔
375	resourceName := GetUploadResourceName(pvc.Name)	1✔
376		1✔
377	// force cleanup if PVC pending delete and pod running or the upload annotation was removed	1✔
378	if !isUploadPod \|\| (pvc.DeletionTimestamp != nil && !podSucceededFromPVC(pvc)) {	2✔
379	// delete everything	1✔
380		1✔
381	// delete service	1✔
382	err = c.deleteService(pvc.Namespace, resourceName)	1✔
383	if err != nil {	1✔
384	return errors.Wrapf(err, "Error deleting upload service for pvc: %s", key)	×
385	}	×
386
387	// delete pod
388	// we're using a req struct for now until we can normalize the controllers a bit more and share things like lister, client etc
389	// this way it's easy to stuff everything into an easy request struct, and can extend aditional behaviors if we want going forward
390	// NOTE this is a special case where the user updated annotations on the pvc to abort the upload requests, we'll add another call
391	// for this for the success case
392	dReq := podDeleteRequest{	1✔
393	namespace: pvc.Namespace,	1✔
394	podName: resourceName,	1✔
395	podLister: c.podLister,	1✔
396	k8sClient: c.client,	1✔
397	}	1✔
398		1✔
399	err = deletePod(dReq)	1✔
400	if err != nil {	1✔
401	return errors.Wrapf(err, "Error deleting upload pod for pvc: %s", key)	×
402	}	×
403
404	return nil	1✔
405	}
406
407	var pod *v1.Pod	1✔
408	if !podSucceededFromPVC(pvc) {	2✔
409	pod, err = c.getOrCreateUploadPod(pvc, resourceName)	1✔
410	if err != nil {	1✔
411	return errors.Wrapf(err, "Error creating upload pod for pvc: %s", key)	×
412	}	×
413
414	podPhase := pod.Status.Phase	1✔
415	if podPhase != podPhaseFromPVC(pvc) {	2✔
416	var labels map[string]string	1✔
417	annotations := map[string]string{AnnPodPhase: string(podPhase)}	1✔
418	pvc, err = updatePVC(c.client, pvc, annotations, labels)	1✔
419	if err != nil {	1✔
420	return errors.Wrapf(err, "Error updating pvc %s, pod phase %s", key, podPhase)	×
421	}	×
422	}
423	}
424
425	if podSucceededFromPVC(pvc) {	2✔
426	// delete service	1✔
427	if err = c.deleteService(pvc.Namespace, resourceName); err != nil {	1✔
428	return errors.Wrapf(err, "Error deleting upload service for pvc %s", key)	×
429	}	×
430
431	dReq := podDeleteRequest{	1✔
432	namespace: pvc.Namespace,	1✔
433	podName: resourceName,	1✔
434	podLister: c.podLister,	1✔
435	k8sClient: c.client,	1✔
436	}	1✔
437		1✔
438	// delete the pod	1✔
439	err = deletePod(dReq)	1✔
440	if err != nil {	1✔
441	return errors.Wrapf(err, "Error deleting upload pod for pvc: %s", key)	×
442	}	×
443	} else {	1✔
444	// make sure the service exists	1✔
445	if _, err = c.getOrCreateUploadService(pvc, resourceName); err != nil {	1✔
446	return errors.Wrapf(err, "Error getting/creating service resource for PVC %s", key)	×
447	}	×
448	}
449
450	return nil	1✔
451	}
452
453	func (c UploadController) getOrCreateUploadPod(pvc v1.PersistentVolumeClaim, name string) (*v1.Pod, error) {	1✔
454	pod, err := c.podLister.Pods(pvc.Namespace).Get(name)	1✔
455		1✔
456	if k8serrors.IsNotFound(err) {	2✔
457	pod, err = CreateUploadPod(c.client, c.serverCAKeyPair, c.clientCAKeyPair.Cert, c.uploadServiceImage, c.verbose, c.pullPolicy, name, pvc, pvc.Name+"-scratch")	1✔
458	}	1✔
459	// Always try to get or create the scratch PVC for a pod that is not successful yet, if it exists nothing happens otherwise attempt to create.
460	_, err = c.getOrCreateScratchPvc(pvc, pod)	1✔
461	if err != nil {	1✔
462	return nil, errors.Wrap(err, "Unable to create scratch space for upload")	×
463	}	×
464
465	if pod != nil && !metav1.IsControlledBy(pod, pvc) {	1✔
466	return nil, errors.Errorf("%s pod not controlled by pvc %s", name, pvc.Name)	×
467	}	×
468	return pod, err	1✔
469	}
470
471	func (c UploadController) getOrCreateScratchPvc(pvc v1.PersistentVolumeClaim, pod v1.Pod) (v1.PersistentVolumeClaim, error) {	1✔
472	scratchPvc, err := c.pvcLister.PersistentVolumeClaims(pvc.Namespace).Get(pvc.Name + "-scratch")	1✔
473	if scratchPvc != nil {	2✔
474	// Scratch PVC already exists, maybe the pod crashed and the pvc didn't get cleaned up, verify this pvc is owned by the pod.	1✔
475	for _, ownerRef := range scratchPvc.OwnerReferences {	2✔
476	if ownerRef.UID == pod.UID {	2✔
477	return scratchPvc, nil	1✔
478	}	1✔
479	}
480	return nil, errors.New("Scratch PVC exists, but is not owned by the right pod")	×
481	}
482	storageClassName := GetScratchPvcStorageClass(c.client, c.cdiClient, pvc)	1✔
483		1✔
484	// Scratch PVC doesn't exist yet, create it. Determine which storage class to use.	1✔
485	scratchPvc, err = CreateScratchPersistentVolumeClaim(c.client, pvc, pod, storageClassName)	1✔
486	if err != nil {	1✔
487	return nil, err	×
488	}	×
489	return scratchPvc, nil	1✔
490	}
491
492	func (c UploadController) getOrCreateUploadService(pvc v1.PersistentVolumeClaim, name string) (*v1.Service, error) {	1✔
493	service, err := c.serviceLister.Services(pvc.Namespace).Get(name)	1✔
494		1✔
495	if k8serrors.IsNotFound(err) {	2✔
496	service, err = CreateUploadService(c.client, name, pvc)	1✔
497	}	1✔
498
499	if service != nil && !metav1.IsControlledBy(service, pvc) {	1✔
500	return nil, errors.Errorf("%s service not controlled by pvc %s", name, pvc.Name)	×
501	}	×
502
503	return service, err	1✔
504	}
505
506	func (c *UploadController) deleteService(namespace, name string) error {	1✔
507	service, err := c.serviceLister.Services(namespace).Get(name)	1✔
508	if k8serrors.IsNotFound(err) {	2✔
509	return nil	1✔
510	}	1✔
511	if err == nil && service.DeletionTimestamp == nil {	2✔
512	err = c.client.CoreV1().Services(namespace).Delete(name, &metav1.DeleteOptions{})	1✔
513	if k8serrors.IsNotFound(err) {	1✔
514	return nil	×
515	}	×
516	}
517	return err	1✔
518	}

kubevirt / containerized-data-importer / 3683

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