Merge pull request #481 from j-griffith/modify_container_restart_policy

Set pod restart policy to "OnFailure" and cleanup succesful pods

Author: awels

pkg/controller/clone-controller.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"fmt"
+
 	"github.com/golang/glog"
 	"github.com/pkg/errors"
 	"k8s.io/api/core/v1"
@@ -156,6 +157,7 @@ func (cc *CloneController) processPvcItem(pvc *v1.PersistentVolumeClaim) error {
 	//add the following annotation only if the pod pahse is succeeded, meaning job is completed
 	if phase == string(v1.PodSucceeded) {
 		anno[AnnCloneOf] = "true"
+		defer cc.deleteClonePods(sourcePod.Namespace, sourcePod.Name, targetPod.Name)
 	}
 	var lab map[string]string
 	if !checkIfLabelExists(pvc, common.CDILabelKey, common.CDILabelValue) {
@@ -168,6 +170,23 @@ func (cc *CloneController) processPvcItem(pvc *v1.PersistentVolumeClaim) error {
 	return nil
 }
 
+func (cc *CloneController) deleteClonePods(namespace, srcName, tgtName string) {
+	srcReq := podDeleteRequest{
+		namespace: namespace,
+		podName:   srcName,
+		podLister: cc.Controller.podLister,
+		k8sClient: cc.Controller.clientset,
+	}
+	tgtReq := podDeleteRequest{
+		namespace: namespace,
+		podName:   tgtName,
+		podLister: cc.Controller.podLister,
+		k8sClient: cc.Controller.clientset,
+	}
+	deletePod(srcReq)
+	deletePod(tgtReq)
+}
+
 func (c *Controller) initializeExpectations(pvcKey string) error {
 	return c.podExpectations.SetExpectations(pvcKey, 0, 0)
 }

pkg/controller/import-controller.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/client-go/tools/cache"
 
 	"fmt"
+
 	"kubevirt.io/containerized-data-importer/pkg/common"
 )
 
@@ -118,6 +119,16 @@ func (ic *ImportController) processPvcItem(pvc *v1.PersistentVolumeClaim) error
 	if pod != nil {
 		anno[AnnImportPod] = string(pod.Name)
 		anno[AnnPodPhase] = string(pod.Status.Phase)
+		if pod.Status.Phase == "Succeeded" {
+			dReq := podDeleteRequest{
+				namespace: pod.Namespace,
+				podName:   pod.Name,
+				podLister: ic.Controller.podLister,
+				k8sClient: ic.Controller.clientset,
+			}
+			// just use defer here so we make sure our pvc updates get written prior to actual deletion
+			defer deletePod(dReq)
+		}
 	}
 
 	var lab map[string]string

pkg/controller/upload-controller.go

@@ -370,7 +370,18 @@ func (c *UploadController) syncHandler(key string) error {
 		}
 
 		// delete pod
-		err = c.deletePod(pvc.Namespace, resourceName)
+		// 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)
 		}
@@ -410,6 +421,20 @@ func (c *UploadController) syncHandler(key string) error {
 		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 {
@@ -448,20 +473,6 @@ func (c *UploadController) getOrCreateUploadService(pvc *v1.PersistentVolumeClai
 	return service, err
 }
 
-func (c *UploadController) deletePod(namespace, name string) error {
-	pod, err := c.podLister.Pods(namespace).Get(name)
-	if k8serrors.IsNotFound(err) {
-		return nil
-	}
-	if err == nil && pod.DeletionTimestamp == nil {
-		err = c.client.CoreV1().Pods(namespace).Delete(name, &metav1.DeleteOptions{})
-		if k8serrors.IsNotFound(err) {
-			return nil
-		}
-	}
-	return err
-}
-
 func (c *UploadController) deleteService(namespace, name string) error {
 	service, err := c.serviceLister.Services(namespace).Get(name)
 	if k8serrors.IsNotFound(err) {

pkg/controller/upload-controller_test.go

@@ -355,7 +355,7 @@ func TestUploadComplete(t *testing.T) {
 
 	f.expectUpdatePvcAction(updatedPVC)
 	f.expectDeleteServiceAction(service)
-
+	f.expectDeletePodAction(pod)
 	f.run(getPvcKey(pvc, t))
 }
 

pkg/controller/util.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 	"time"
 
+	corelisters "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/util/cert/triple"
 
 	"github.com/golang/glog"
@@ -27,6 +28,13 @@ const DataVolName = "cdi-data-vol"
 const ImagePathName = "image-path"
 const socketPathName = "socket-path"
 
+type podDeleteRequest struct {
+	namespace string
+	podName   string
+	podLister corelisters.PodLister
+	k8sClient kubernetes.Interface
+}
+
 func checkPVC(pvc *v1.PersistentVolumeClaim, annotation string) bool {
 	if pvc.DeletionTimestamp != nil {
 		return false
@@ -216,7 +224,7 @@ func MakeImporterPodSpec(image, verbose, pullPolicy, ep, secret string, pvc *v1.
 					Args: []string{"-v=" + verbose},
 				},
 			},
-			RestartPolicy: v1.RestartPolicyNever,
+			RestartPolicy: v1.RestartPolicyOnFailure,
 			Volumes: []v1.Volume{
 				{
 					Name: DataVolName,
@@ -721,3 +729,20 @@ func MakeUploadServiceSpec(name string, pvc *v1.PersistentVolumeClaim) *v1.Servi
 	}
 	return service
 }
+
+func deletePod(req podDeleteRequest) error {
+	pod, err := req.podLister.Pods(req.namespace).Get(req.podName)
+	if k8serrors.IsNotFound(err) {
+		return nil
+	}
+	if err == nil && pod.DeletionTimestamp == nil {
+		err = req.k8sClient.CoreV1().Pods(req.namespace).Delete(req.podName, &metav1.DeleteOptions{})
+		if k8serrors.IsNotFound(err) {
+			return nil
+		}
+	}
+	if err != nil {
+		glog.V(1).Infof("error encountered deleting pod (%s): %s", req.podName, err.Error())
+	}
+	return err
+}

pkg/controller/util_test.go

@@ -796,7 +796,7 @@ func createPod(pvc *v1.PersistentVolumeClaim, dvname string) *v1.Pod {
 					Args: []string{"-v=5"},
 				},
 			},
-			RestartPolicy: v1.RestartPolicyNever,
+			RestartPolicy: v1.RestartPolicyOnFailure,
 			Volumes: []v1.Volume{
 				{
 					Name: dvname,

tests/cloner_test.go

@@ -25,6 +25,7 @@ const (
 	fillCommand           = "echo \"" + fillData + "\" >> " + testFile
 	assertionPollInterval = 2 * time.Second
 	cloneCompleteTimeout  = 60 * time.Second
+	shortTimeout          = 2 * time.Second
 )
 
 var _ = Describe(testSuiteName, func() {
@@ -69,6 +70,7 @@ func doCloneTest(f *framework.Framework, targetNs *v1.Namespace) {
 		map[string]string{controller.AnnCloneRequest: f.Namespace.Name + "/" + sourcePVCName},
 		nil))
 	Expect(err).ToNot(HaveOccurred())
+	fmt.Fprintf(GinkgoWriter, "INFO: wait for PVC claim phase: %s\n", targetPvc.Name)
 	utils.WaitForPersistentVolumeClaimPhase(f.K8sClient, targetNs.Name, v1.ClaimBound, targetPvc.Name)
 
 	By("Find cloner pods")
@@ -94,11 +96,6 @@ func doCloneTest(f *framework.Framework, targetNs *v1.Namespace) {
 		return srcNode == tgtNode
 	}, cloneCompleteTimeout, assertionPollInterval).Should(BeTrue())
 
-	err = f.WaitTimeoutForPodStatus(sourcePod.Name, v1.PodSucceeded, cloneCompleteTimeout)
-	Expect(err).ToNot(HaveOccurred())
-	err = utils.WaitTimeoutForPodStatus(f.K8sClient, targetPod.Name, targetNs.Name, v1.PodSucceeded, cloneCompleteTimeout)
-	Expect(err).ToNot(HaveOccurred())
-
 	By("Verify the clone annotation is on the target PVC")
 	_, cloneAnnotationFound, err := utils.WaitForPVCAnnotation(f.K8sClient, targetNs.Name, targetPvc, controller.AnnCloneOf)
 	Expect(err).ToNot(HaveOccurred())
@@ -119,4 +116,14 @@ func doCloneTest(f *framework.Framework, targetNs *v1.Namespace) {
 		err = utils.DeletePVC(f.K8sClient, targetNs.Name, targetPvc)
 		Expect(err).ToNot(HaveOccurred())
 	}
+
+	By("Verify source pod deleted")
+	deleted, err := utils.WaitPodDeleted(f.K8sClient, sourcePod.Name, targetPod.Namespace, time.Second*40)
+	Expect(err).ToNot(HaveOccurred())
+	Expect(deleted).To(BeTrue())
+
+	By("Verify target pod deleted")
+	deleted, err = utils.WaitPodDeleted(f.K8sClient, targetPod.Name, targetNs.Name, time.Second*40)
+	Expect(err).ToNot(HaveOccurred())
+	Expect(deleted).To(BeTrue())
 }

tests/framework/pvc.go

@@ -41,7 +41,6 @@ func VerifyPVCIsEmpty(f *Framework, pvc *k8sv1.PersistentVolumeClaim) bool {
 	err = f.WaitTimeoutForPodReady(executorPod.Name, utils.PodWaitForTime)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred())
 	output := f.ExecShellInPod(executorPod.Name, f.Namespace.Name, "ls -1 /pvc | wc -l")
-	f.DeletePod(executorPod)
 	return strings.Compare("0", output) == 0
 }
 
@@ -67,7 +66,6 @@ func (f *Framework) VerifyTargetPVCContent(namespace *k8sv1.Namespace, pvc *k8sv
 	err = utils.WaitTimeoutForPodReady(f.K8sClient, executorPod.Name, namespace.Name, utils.PodWaitForTime)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred())
 	output := f.ExecShellInPod(executorPod.Name, namespace.Name, "cat "+fileName)
-	f.DeletePod(executorPod)
 	return strings.Compare(expectedData, output) == 0
 }
 
@@ -78,6 +76,5 @@ func (f *Framework) VerifyTargetPVCContentMD5(namespace *k8sv1.Namespace, pvc *k
 	err = utils.WaitTimeoutForPodReady(f.K8sClient, executorPod.Name, namespace.Name, utils.PodWaitForTime)
 	gomega.Expect(err).ToNot(gomega.HaveOccurred())
 	output := f.ExecShellInPod(executorPod.Name, namespace.Name, "md5sum "+fileName)
-	f.DeletePod(executorPod)
 	return strings.Compare(expectedHash, output[:32]) == 0
 }

tests/transport_test.go

@@ -71,7 +71,6 @@ var _ = Describe("Transport Tests", func() {
 		if err != nil {
 			PrintPodLog(f, importer.Name, ns)
 		}
-		Expect(err).NotTo(HaveOccurred(), fmt.Sprintf("Unable to get importer pod %q", ns+"/"+common.ImporterPodName))
 
 		By("Verifying PVC is not empty")
 		Expect(framework.VerifyPVCIsEmpty(f, pvc)).To(BeFalse(), fmt.Sprintf("Found 0 imported files on PVC %q", pvc.Namespace+"/"+pvc.Name))

tests/upload_test.go

@@ -6,8 +6,6 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
-	k8sv1 "k8s.io/api/core/v1"
-
 	"kubevirt.io/containerized-data-importer/tests/framework"
 	"kubevirt.io/containerized-data-importer/tests/utils"
 )
@@ -46,15 +44,11 @@ var _ = Describe("Upload tests", func() {
 		err = utils.UploadImageFromNode(f.K8sClient, f.GoCLIPath, token)
 		Expect(err).ToNot(HaveOccurred())
 
-		By("Wait pod succeeded")
-		err = f.WaitTimeoutForPodStatus(utils.UploadPodName(pvc), k8sv1.PodSucceeded, time.Second*20)
-		Expect(err).ToNot(HaveOccurred())
-
 		By("Verify content")
 		same := f.VerifyTargetPVCContentMD5(f.Namespace, pvc, testFile, utils.UploadFileMD5)
 		Expect(same).To(BeTrue())
 
-		By("Delete upoad PVC")
+		By("Delete upload PVC")
 		err = f.DeletePVC(pvc)
 		Expect(err).ToNot(HaveOccurred())
 

tests/utils.go

@@ -56,7 +56,7 @@ func PrintPodLog(f *framework.Framework, podName, namespace string) {
 	if err == nil {
 		fmt.Fprintf(ginkgo.GinkgoWriter, "INFO: Controller log\n%s\n", log)
 	} else {
-		fmt.Fprintf(ginkgo.GinkgoWriter, "INFO: Unable to get controller log")
+		fmt.Fprintf(ginkgo.GinkgoWriter, "INFO: Unable to get controller log\n")
 	}
 }