address feedback

leverages must delete functions during creation
changes log.fatal to panic since log fatal will immediately exit, skipping all defers
leverages wait for daemonset instead of wait for pods
adds retry parameter to exec cmd on pod, adjusting associated calls
incorporates exec cmd on pod error into lrp test

Author: QxBytes

test/integration/lrp/lrp_fqdn_test.go

@@ -48,50 +48,57 @@ func TestLRPFQDN(t *testing.T) {
 		command                []string
 		expectedMsgContains    string
 		expectedErrMsgContains string
+		shouldError            bool
 		countIncreases         bool
 	}{
 		{
 			name:                "nslookup google succeeds",
 			command:             []string{"nslookup", "www.google.com", "10.0.0.10"},
-			expectedMsgContains: "Server:",
+			expectedMsgContains: "answer:",
 			countIncreases:      true,
+			shouldError:         false,
 		},
 		{
 			name:                   "wget google succeeds",
 			command:                []string{"wget", "-O", "index.html", "www.google.com", "--timeout=5"},
 			expectedErrMsgContains: "saved",
 			countIncreases:         true,
+			shouldError:            false,
 		},
 		{
 			name:                "nslookup bing succeeds",
 			command:             []string{"nslookup", "www.bing.com", "10.0.0.10"},
-			expectedMsgContains: "Server:",
+			expectedMsgContains: "answer:",
 			countIncreases:      true,
+			shouldError:         false,
 		},
 		{
 			name:                   "wget bing fails but dns succeeds",
 			command:                []string{"wget", "-O", "index.html", "www.bing.com", "--timeout=5"},
 			expectedErrMsgContains: "timed out",
 			countIncreases:         true,
+			shouldError:            true,
 		},
 		{
 			name:                "nslookup example fails",
 			command:             []string{"nslookup", "www.example.com", "10.0.0.10"},
 			expectedMsgContains: "REFUSED",
 			countIncreases:      false,
+			shouldError:         true,
 		},
 		{
 			// won't be able to nslookup, let alone query the website
 			name:                   "wget example fails",
 			command:                []string{"wget", "-O", "index.html", "www.example.com", "--timeout=5"},
 			expectedErrMsgContains: "bad address",
 			countIncreases:         false,
+			shouldError:            true,
 		},
 	}
 	for _, tt := range tests {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
-			testLRPCase(t, ctx, *selectedPod, tt.command, tt.expectedMsgContains, tt.expectedErrMsgContains, tt.countIncreases)
+			testLRPCase(t, ctx, *selectedPod, tt.command, tt.expectedMsgContains, tt.expectedErrMsgContains, tt.shouldError, tt.countIncreases)
 		})
 	}
 }

test/integration/lrp/lrp_test.go

@@ -105,7 +105,7 @@ func setupLRP(t *testing.T, ctx context.Context) (*corev1.Pod, func()) {
 	cleanUpFns = append(cleanUpFns, cleanupService)
 	nodeLocalDNSDS, cleanupNodeLocalDNS := kubernetes.MustSetupDaemonset(ctx, cs, tempNodeLocalDNSDaemonsetPath)
 	cleanUpFns = append(cleanUpFns, cleanupNodeLocalDNS)
-	err = kubernetes.WaitForPodsRunning(ctx, cs, nodeLocalDNSDS.Namespace, nodeLocalDNSLabelSelector)
+	kubernetes.WaitForPodDaemonset(ctx, cs, nodeLocalDNSDS.Namespace, nodeLocalDNSDS.Name, nodeLocalDNSLabelSelector)
 	require.NoError(t, err)
 	// select a local dns pod after they start running
 	pods, err := kubernetes.GetPodsByNode(ctx, cs, nodeLocalDNSDS.Namespace, nodeLocalDNSLabelSelector, selectedNode)
@@ -119,7 +119,7 @@ func setupLRP(t *testing.T, ctx context.Context) (*corev1.Pod, func()) {
 	// create client pods
 	clientDS, cleanupClient := kubernetes.MustSetupDaemonset(ctx, cs, clientPath)
 	cleanUpFns = append(cleanUpFns, cleanupClient)
-	err = kubernetes.WaitForPodsRunning(ctx, cs, clientDS.Namespace, clientLabelSelector)
+	kubernetes.WaitForPodDaemonset(ctx, cs, clientDS.Namespace, clientDS.Name, clientLabelSelector)
 	require.NoError(t, err)
 	// select a client pod after they start running
 	clientPods, err := kubernetes.GetPodsByNode(ctx, cs, clientDS.Namespace, clientLabelSelector, selectedNode)
@@ -153,7 +153,9 @@ func setupLRP(t *testing.T, ctx context.Context) (*corev1.Pod, func()) {
 	return &selectedClientPod, cleanupFn
 }
 
-func testLRPCase(t *testing.T, ctx context.Context, clientPod corev1.Pod, clientCmd []string, expectResponse, expectErrMsg string, countShouldIncrease bool) {
+func testLRPCase(t *testing.T, ctx context.Context, clientPod corev1.Pod, clientCmd []string, expectResponse, expectErrMsg string,
+	shouldError, countShouldIncrease bool) {
+
 	config := kubernetes.MustGetRestConfig()
 	cs := kubernetes.MustGetClientset()
 
@@ -171,10 +173,15 @@ func testLRPCase(t *testing.T, ctx context.Context, clientPod corev1.Pod, client
 
 	t.Log("calling command from client")
 
-	val, errMsg, err := kubernetes.ExecCmdOnPodOnce(ctx, cs, clientPod.Namespace, clientPod.Name, clientContainer, clientCmd, config)
+	val, errMsg, err := kubernetes.ExecCmdOnPod(ctx, cs, clientPod.Namespace, clientPod.Name, clientContainer, clientCmd, config, false)
 
 	require.Contains(t, string(val), expectResponse)
 	require.Contains(t, string(errMsg), expectErrMsg)
+	if shouldError {
+		require.Error(t, err)
+	} else {
+		require.NoError(t, err)
+	}
 
 	// in case there is time to propagate
 	time.Sleep(500 * time.Millisecond)
@@ -205,7 +212,7 @@ func TestLRP(t *testing.T) {
 
 	testLRPCase(t, ctx, *selectedPod, []string{
 		"nslookup", "google.com", "10.0.0.10",
-	}, "Server:", "", true)
+	}, "Server:", "", false, true)
 }
 
 // TakeOne takes one item from the slice randomly; if empty, it returns the empty value for the type

test/internal/datapath/datapath_win.go

@@ -19,7 +19,7 @@ var ipv6PrefixPolicy = []string{"powershell", "-c", "curl.exe", "-6", "-v", "www
 
 func podTest(ctx context.Context, clientset *kubernetes.Clientset, srcPod *apiv1.Pod, cmd []string, rc *restclient.Config, passFunc func(string) error) error {
 	logrus.Infof("podTest() - %v %v", srcPod.Name, cmd)
-	output, err := acnk8s.ExecCmdOnPod(ctx, clientset, srcPod.Namespace, srcPod.Name, "", cmd, rc)
+	output, _, err := acnk8s.ExecCmdOnPod(ctx, clientset, srcPod.Namespace, srcPod.Name, "", cmd, rc, true)
 	if err != nil {
 		return errors.Wrapf(err, "failed to execute command on pod: %v", srcPod.Name)
 	}

test/internal/kubernetes/utils.go

@@ -233,37 +233,37 @@ func MustSetupDeployment(ctx context.Context, clientset *kubernetes.Clientset, d
 
 func MustSetupServiceAccount(ctx context.Context, clientset *kubernetes.Clientset, serviceAccountPath string) (corev1.ServiceAccount, func()) { // nolint
 	sa := mustParseServiceAccount(serviceAccountPath)
-	sas := clientset.CoreV1().ServiceAccounts(sa.Namespace)
-	mustCreateServiceAccount(ctx, sas, sa)
+	saClient := clientset.CoreV1().ServiceAccounts(sa.Namespace)
+	mustCreateServiceAccount(ctx, saClient, sa)
 	return sa, func() {
-		MustDeleteServiceAccount(ctx, sas, sa)
+		MustDeleteServiceAccount(ctx, saClient, sa)
 	}
 }
 
 func MustSetupService(ctx context.Context, clientset *kubernetes.Clientset, servicePath string) (corev1.Service, func()) { // nolint
 	svc := mustParseService(servicePath)
-	svcs := clientset.CoreV1().Services(svc.Namespace)
-	mustCreateService(ctx, svcs, svc)
+	svcClient := clientset.CoreV1().Services(svc.Namespace)
+	mustCreateService(ctx, svcClient, svc)
 	return svc, func() {
-		MustDeleteService(ctx, svcs, svc)
+		MustDeleteService(ctx, svcClient, svc)
 	}
 }
 
 func MustSetupLRP(ctx context.Context, clientset *cilium.Clientset, lrpPath string) (ciliumv2.CiliumLocalRedirectPolicy, func()) { // nolint
 	lrp := mustParseLRP(lrpPath)
-	lrps := clientset.CiliumV2().CiliumLocalRedirectPolicies(lrp.Namespace)
-	mustCreateCiliumLocalRedirectPolicy(ctx, lrps, lrp)
+	lrpClient := clientset.CiliumV2().CiliumLocalRedirectPolicies(lrp.Namespace)
+	mustCreateCiliumLocalRedirectPolicy(ctx, lrpClient, lrp)
 	return lrp, func() {
-		MustDeleteCiliumLocalRedirectPolicy(ctx, lrps, lrp)
+		MustDeleteCiliumLocalRedirectPolicy(ctx, lrpClient, lrp)
 	}
 }
 
 func MustSetupCNP(ctx context.Context, clientset *cilium.Clientset, cnpPath string) (ciliumv2.CiliumNetworkPolicy, func()) { // nolint
 	cnp := mustParseCNP(cnpPath)
-	cnps := clientset.CiliumV2().CiliumNetworkPolicies(cnp.Namespace)
-	mustCreateCiliumNetworkPolicy(ctx, cnps, cnp)
+	cnpClient := clientset.CiliumV2().CiliumNetworkPolicies(cnp.Namespace)
+	mustCreateCiliumNetworkPolicy(ctx, cnpClient, cnp)
 	return cnp, func() {
-		MustDeleteCiliumNetworkPolicy(ctx, cnps, cnp)
+		MustDeleteCiliumNetworkPolicy(ctx, cnpClient, cnp)
 	}
 }
 
@@ -488,59 +488,60 @@ func writeToFile(dir, fileName, str string) error {
 	return errors.Wrap(err, "failed to write string")
 }
 
-func ExecCmdOnPod(ctx context.Context, clientset *kubernetes.Clientset, namespace, podName, containerName string, cmd []string, config *rest.Config) ([]byte, error) {
+// ExecCmdOnPod runs the specified command on a particular pod and retries the command on failure if doRetry is set to true
+// The function returns the standard output, standard error, and error (if any) in that order
+func ExecCmdOnPod(ctx context.Context, clientset *kubernetes.Clientset, namespace, podName, containerName string, cmd []string, config *rest.Config, doRetry bool) ([]byte, []byte, error) {
 	var result []byte
+	var errResult []byte
 	execCmdOnPod := func() error {
-		output, _, err := ExecCmdOnPodOnce(ctx, clientset, namespace, podName, containerName, cmd, config)
-		result = output
-		return err
-	}
-	retrier := retry.Retrier{Attempts: ShortRetryAttempts, Delay: RetryDelay}
-	err := retrier.Do(ctx, execCmdOnPod)
-	return result, errors.Wrapf(err, "could not execute the cmd %s on %s", cmd, podName)
-}
-
-// ExecCmdOnPodOnce runs a command on the specified pod and returns its standard output, standard error output, and error in that order
-// The command does not retry when the command fails (ex: due to timeout), unlike ExecCmdOnPod
-func ExecCmdOnPodOnce(ctx context.Context, clientset *kubernetes.Clientset, namespace, podName, containerName string, cmd []string, config *rest.Config) ([]byte, []byte, error) { // nolint
-	req := clientset.CoreV1().RESTClient().Post().
-		Resource("pods").
-		Name(podName).
-		Namespace(namespace).
-		SubResource("exec").
-		VersionedParams(&corev1.PodExecOptions{
-			Command:   cmd,
-			Container: containerName,
-			Stdin:     false,
-			Stdout:    true,
-			Stderr:    true,
-			TTY:       false,
-		}, scheme.ParameterCodec)
-
-	exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
-	if err != nil {
-		return nil, nil, errors.Wrapf(err, "error in creating executor for req %s", req.URL())
-	}
+		req := clientset.CoreV1().RESTClient().Post().
+			Resource("pods").
+			Name(podName).
+			Namespace(namespace).
+			SubResource("exec").
+			VersionedParams(&corev1.PodExecOptions{
+				Command:   cmd,
+				Container: containerName,
+				Stdin:     false,
+				Stdout:    true,
+				Stderr:    true,
+				TTY:       false,
+			}, scheme.ParameterCodec)
+
+		exec, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
+		if err != nil {
+			return errors.Wrapf(err, "error in creating executor for req %s", req.URL())
+		}
 
-	var stdout, stderr bytes.Buffer
-	err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
-		Stdin:  nil,
-		Stdout: &stdout,
-		Stderr: &stderr,
-		Tty:    false,
-	})
+		var stdout, stderr bytes.Buffer
+		err = exec.StreamWithContext(ctx, remotecommand.StreamOptions{
+			Stdin:  nil,
+			Stdout: &stdout,
+			Stderr: &stderr,
+			Tty:    false,
+		})
 
-	result := stdout.Bytes()
-	errResult := stderr.Bytes()
+		result = stdout.Bytes()
+		errResult = stderr.Bytes()
 
-	if err != nil {
-		log.Printf("Error: %v had error %v from command - %v", podName, err, cmd)
-		return result, errResult, errors.Wrapf(err, "error in executing command %s", cmd)
+		if err != nil {
+			log.Printf("Error: %v had error %v from command - %v", podName, err, cmd)
+			return errors.Wrapf(err, "error in executing command %s", cmd)
+		}
+		if len(stdout.Bytes()) == 0 {
+			log.Printf("Warning: %v had 0 bytes returned from command - %v", podName, cmd)
+		}
+		return nil
 	}
-	if len(stdout.Bytes()) == 0 {
-		log.Printf("Warning: %v had 0 bytes returned from command - %v", podName, cmd)
+
+	var err error
+	if doRetry {
+		retrier := retry.Retrier{Attempts: ShortRetryAttempts, Delay: RetryDelay}
+		err = retrier.Do(ctx, execCmdOnPod)
+	} else {
+		err = execCmdOnPod()
 	}
-	return result, errResult, nil
+	return result, errResult, errors.Wrapf(err, "could not execute the cmd %s on %s", cmd, podName)
 }
 
 func NamespaceExists(ctx context.Context, clientset *kubernetes.Clientset, namespace string) (bool, error) {
@@ -655,7 +656,7 @@ func RestartKubeProxyService(ctx context.Context, clientset *kubernetes.Clientse
 		}
 		privilegedPod := pod.Items[0]
 		// exec into the pod and restart kubeproxy
-		_, err = ExecCmdOnPod(ctx, clientset, privilegedNamespace, privilegedPod.Name, "", restartKubeProxyCmd, config)
+		_, _, err = ExecCmdOnPod(ctx, clientset, privilegedNamespace, privilegedPod.Name, "", restartKubeProxyCmd, config, true)
 		if err != nil {
 			return errors.Wrapf(err, "failed to exec into privileged pod %s on node %s", privilegedPod.Name, node.Name)
 		}

test/internal/kubernetes/utils_create.go

@@ -166,38 +166,26 @@ func mustCreateConfigMap(ctx context.Context, cmi typedcorev1.ConfigMapInterface
 }
 
 func mustCreateService(ctx context.Context, svci typedcorev1.ServiceInterface, svc corev1.Service) {
-	if err := svci.Delete(ctx, svc.Name, metav1.DeleteOptions{}); err != nil {
-		if !apierrors.IsNotFound(err) {
-			log.Fatal(errors.Wrap(err, "failed to delete service"))
-		}
-	}
+	MustDeleteService(ctx, svci, svc)
 	log.Printf("Creating Service %v", svc.Name)
 	if _, err := svci.Create(ctx, &svc, metav1.CreateOptions{}); err != nil {
-		log.Fatal(errors.Wrap(err, "failed to create service"))
+		panic(errors.Wrap(err, "failed to create service"))
 	}
 }
 
 func mustCreateCiliumLocalRedirectPolicy(ctx context.Context, lrpClient typedciliumv2.CiliumLocalRedirectPolicyInterface, clrp ciliumv2.CiliumLocalRedirectPolicy) {
-	if err := lrpClient.Delete(ctx, clrp.Name, metav1.DeleteOptions{}); err != nil {
-		if !apierrors.IsNotFound(err) {
-			log.Fatal(errors.Wrap(err, "failed to delete cilium local redirect policy"))
-		}
-	}
+	MustDeleteCiliumLocalRedirectPolicy(ctx, lrpClient, clrp)
 	log.Printf("Creating CiliumLocalRedirectPolicy %v", clrp.Name)
 	if _, err := lrpClient.Create(ctx, &clrp, metav1.CreateOptions{}); err != nil {
-		log.Fatal(errors.Wrap(err, "failed to create cilium local redirect policy"))
+		panic(errors.Wrap(err, "failed to create cilium local redirect policy"))
 	}
 }
 
 func mustCreateCiliumNetworkPolicy(ctx context.Context, cnpClient typedciliumv2.CiliumNetworkPolicyInterface, cnp ciliumv2.CiliumNetworkPolicy) {
-	if err := cnpClient.Delete(ctx, cnp.Name, metav1.DeleteOptions{}); err != nil {
-		if !apierrors.IsNotFound(err) {
-			log.Fatal(errors.Wrap(err, "failed to delete cilium network policy"))
-		}
-	}
+	MustDeleteCiliumNetworkPolicy(ctx, cnpClient, cnp)
 	log.Printf("Creating CiliumNetworkPolicy %v", cnp.Name)
 	if _, err := cnpClient.Create(ctx, &cnp, metav1.CreateOptions{}); err != nil {
-		log.Fatal(errors.Wrap(err, "failed to create cilium network policy"))
+		panic(errors.Wrap(err, "failed to create cilium network policy"))
 	}
 }
 

test/validate/linux_validate.go

@@ -326,7 +326,7 @@ func (v *Validator) validateRestartNetwork(ctx context.Context) error {
 		}
 		privilegedPod := pod.Items[0]
 		// exec into the pod to get the state file
-		_, err = acnk8s.ExecCmdOnPod(ctx, v.clientset, privilegedNamespace, privilegedPod.Name, "", restartNetworkCmd, v.config)
+		_, _, err = acnk8s.ExecCmdOnPod(ctx, v.clientset, privilegedNamespace, privilegedPod.Name, "", restartNetworkCmd, v.config, true)
 		if err != nil {
 			return errors.Wrapf(err, "failed to exec into privileged pod %s on node %s", privilegedPod.Name, node.Name)
 		}

test/validate/validate.go

@@ -142,7 +142,7 @@ func (v *Validator) validateIPs(ctx context.Context, stateFileIps stateFileIpsFu
 		podName := pod.Items[0].Name
 		// exec into the pod to get the state file
 		log.Printf("Executing command %s on pod %s, container %s", cmd, podName, containerName)
-		result, err := acnk8s.ExecCmdOnPod(ctx, v.clientset, namespace, podName, containerName, cmd, v.config)
+		result, _, err := acnk8s.ExecCmdOnPod(ctx, v.clientset, namespace, podName, containerName, cmd, v.config, true)
 		if err != nil {
 			return errors.Wrapf(err, "failed to exec into privileged pod - %s", podName)
 		}

test/validate/windows_validate.go

@@ -269,7 +269,7 @@ func validateHNSNetworkState(ctx context.Context, nodes *corev1.NodeList, client
 		}
 		podName := pod.Items[0].Name
 		// exec into the pod to get the state file
-		result, err := acnk8s.ExecCmdOnPod(ctx, clientset, privilegedNamespace, podName, "", hnsNetworkCmd, restConfig)
+		result, _, err := acnk8s.ExecCmdOnPod(ctx, clientset, privilegedNamespace, podName, "", hnsNetworkCmd, restConfig, true)
 		if err != nil {
 			return errors.Wrap(err, "failed to exec into privileged pod")
 		}