fix(pkg/util/kubernetes/portforward.go): corner case scenario in port…

… forwarding for debug cmd is now handled properly.
lburgazzoli · May 8, 2024 · 6da16e2 · 6da16e2
1 parent 1819d29
commit 6da16e2
Show file tree

Hide file tree

Showing 2 changed files with 37 additions and 11 deletions.
diff --git a/e2e/advanced/debug_test.go b/e2e/advanced/debug_test.go
@@ -58,6 +58,7 @@ func TestKamelCLIDebug(t *testing.T) {
 
 			g.Eventually(portIsInUse("127.0.0.1", "5005"), TestTimeoutMedium, 5*time.Second).Should(BeTrue())
 			g.Expect(Kamel(t, ctx, "delete", "--all", "-n", ns).Execute()).To(Succeed())
+			g.Eventually(IntegrationPods(t, ctx, ns, "yaml"), TestTimeoutMedium, 5*time.Second).Should(HaveLen(0))
 		})
 
 		t.Run("debug local port check", func(t *testing.T) {
@@ -71,6 +72,7 @@ func TestKamelCLIDebug(t *testing.T) {
 
 			g.Eventually(portIsInUse("127.0.0.1", "5006"), TestTimeoutMedium, 5*time.Second).Should(BeTrue())
 			g.Expect(Kamel(t, ctx, "delete", "--all", "-n", ns).Execute()).To(Succeed())
+			g.Eventually(IntegrationPods(t, ctx, ns, "yaml"), TestTimeoutMedium, 5*time.Second).Should(HaveLen(0))
 		})
 
 		t.Run("debug logs check", func(t *testing.T) {
@@ -83,6 +85,7 @@ func TestKamelCLIDebug(t *testing.T) {
 
 			g.Eventually(IntegrationLogs(t, ctx, ns, "yaml"), TestTimeoutMedium).Should(ContainSubstring("Listening for transport dt_socket at address: 5005"))
 			g.Expect(Kamel(t, ctx, "delete", "--all", "-n", ns).Execute()).To(Succeed())
+			g.Eventually(IntegrationPods(t, ctx, ns, "yaml"), TestTimeoutMedium, 5*time.Second).Should(HaveLen(0))
 		})
 
 		t.Run("Pod config test", func(t *testing.T) {
@@ -98,9 +101,11 @@ func TestKamelCLIDebug(t *testing.T) {
 			}).Should(ContainSubstring("-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:5005"))
 			g.Expect(IntegrationPod(t, ctx, ns, "yaml")().GetLabels()["camel.apache.org/debug"]).To(Not(BeNil()))
 			g.Expect(Kamel(t, ctx, "delete", "--all", "-n", ns).Execute()).To(Succeed())
+			g.Eventually(IntegrationPods(t, ctx, ns, "yaml"), TestTimeoutMedium, 5*time.Second).Should(HaveLen(0))
 		})
 
 		g.Expect(Kamel(t, ctx, "delete", "--all", "-n", ns).Execute()).To(Succeed())
+		g.Eventually(IntegrationPods(t, ctx, ns, "yaml"), TestTimeoutMedium, 5*time.Second).Should(HaveLen(0))
 	})
 }
 

diff --git a/pkg/util/kubernetes/portforward.go b/pkg/util/kubernetes/portforward.go
@@ -37,13 +37,6 @@ import (
 
 func PortForward(ctx context.Context, c client.Client, ns, labelSelector string, localPort, remotePort uint, stdOut, stdErr io.Writer) error {
 	log.InitForCmd()
-	list, err := c.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{
-		LabelSelector: labelSelector,
-	})
-	if err != nil {
-		return err
-	}
-
 	var forwardPod *corev1.Pod
 	var forwardCtx context.Context
 	var forwardCtxCancel context.CancelFunc
@@ -52,19 +45,21 @@ func PortForward(ctx context.Context, c client.Client, ns, labelSelector string,
 		if forwardPod == nil && podReady(pod) {
 			forwardPod = pod
 			forwardCtx, forwardCtxCancel = context.WithCancel(ctx)
+			log.Debugf("Setting up Port Forward for pod with name: %q\n", forwardPod.Name)
 			if _, err := portFowardPod(forwardCtx, c.GetConfig(), ns, forwardPod.Name, localPort, remotePort, stdOut, stdErr); err != nil {
 				return err
 			}
 		}
 		return nil
 	}
 
-	if len(list.Items) > 0 {
-		if err := setupPortForward(&list.Items[0]); err != nil {
-			return err
-		}
+	log.Debugf("First attempt to bootstrap Port Forward with LabelSelector: %v\n", labelSelector)
+	list, err := bootstrapPortForward(ctx, c, ns, labelSelector, setupPortForward)
+	if err != nil {
+		return err
 	}
 
+	log.Debugf("Instantiating pod event watcher with LabelSelector: %v and ResourceVersion: %v in namespace: %v\n", labelSelector, list.ResourceVersion, ns)
 	watcher, err := c.CoreV1().Pods(ns).Watch(ctx, metav1.ListOptions{
 		LabelSelector:   labelSelector,
 		ResourceVersion: list.ResourceVersion,
@@ -90,6 +85,7 @@ func PortForward(ctx context.Context, c client.Client, ns, labelSelector string,
 				if !ok {
 					return fmt.Errorf("type assertion failed: %v", e.Object)
 				}
+				log.Debugf("Handling watch.Added event for pod with name: %v\n", pod.Name)
 				if err := setupPortForward(pod); err != nil {
 					return err
 				}
@@ -98,27 +94,52 @@ func PortForward(ctx context.Context, c client.Client, ns, labelSelector string,
 				if !ok {
 					return fmt.Errorf("type assertion failed: %v", e.Object)
 				}
+				log.Debugf("Handling watch.Modified event for pod with name: %v\n", pod.Name)
 				if err := setupPortForward(pod); err != nil {
 					return err
 				}
 			case watch.Deleted:
+				log.Debugf("Handling watch.Deleted event\n")
 				if forwardPod != nil && e.Object != nil {
 					deletedPod, ok := e.Object.(*corev1.Pod)
 					if !ok {
 						return fmt.Errorf("type assertion failed: %v", e.Object)
 					}
+					log.Debugf("Handling watch.Deleted event for pod with name: %v while Port Forward was active for pod with name: %v\n", deletedPod.Name, forwardPod.Name)
 					if deletedPod.Name == forwardPod.Name {
 						forwardCtxCancel()
 						forwardPod = nil
 						forwardCtx = nil
 						forwardCtxCancel = nil
+
+						log.Debugf("Handling watch.Deleted event, since the pod with Port Forward enabled has been deleted we try to bootstrap Port Forward with LabelSelector: %v\n", labelSelector)
+						_, err := bootstrapPortForward(ctx, c, ns, labelSelector, setupPortForward)
+						if err != nil {
+							return err
+						}
 					}
 				}
 			}
 		}
 	}
 }
 
+func bootstrapPortForward(ctx context.Context, c client.Client, ns string, labelSelector string, setupPortForward func(pod *corev1.Pod) error) (*corev1.PodList, error) {
+	list, err := c.CoreV1().Pods(ns).List(ctx, metav1.ListOptions{
+		LabelSelector: labelSelector,
+	})
+	if err != nil {
+		return nil, err
+	}
+	if len(list.Items) > 0 {
+		log.Debugf("Bootstrapping Port Forward for pod with name: %v\n", list.Items[0].Name)
+		if err := setupPortForward(&list.Items[0]); err != nil {
+			return nil, err
+		}
+	}
+	return list, nil
+}
+
 func portFowardPod(ctx context.Context, config *restclient.Config, ns, pod string, localPort, remotePort uint, stdOut, stdErr io.Writer) (string, error) {
 	c, err := corev1client.NewForConfig(config)
 	if err != nil {