func PortForward()

in pkg/util/kubernetes/portforward.go [38:121]
77 lines of code
22 McCabe index (conditional complexity)

func PortForward(ctx context.Context, c client.Client, ns, labelSelector string, localPort, remotePort uint, stdOut, stdErr io.Writer) error {
	log.InitForCmd()
	var forwardPod *corev1.Pod
	forwardCtx, forwardCtxCancel := context.WithCancel(ctx)
	defer forwardCtxCancel()

	setupPortForward := func(pod *corev1.Pod) error {
		if forwardPod == nil && podReady(pod) {
			forwardPod = pod
			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
	}

	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,
	})
	if err != nil {
		return err
	}

	events := watcher.ResultChan()

	for {
		select {
		case <-ctx.Done():
			return nil
		case e, ok := <-events:
			if !ok {
				return nil
			}

			switch e.Type {
			case watch.Added:
				pod, ok := e.Object.(*corev1.Pod)
				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
				}
			case watch.Modified:
				pod, ok := e.Object.(*corev1.Pod)
				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 {
						forwardPod = 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
						}
					}
				}
			}
		}
	}
}