Watch for new pods when logs --follow

redhat-developer · Jun 19, 2023 · 233d8ce · 233d8ce
1 parent 4f46fe9
commit 233d8ce
Show file tree

Hide file tree

Showing 7 changed files with 108 additions and 17 deletions.
diff --git a/pkg/exec/exec_test.go b/pkg/exec/exec_test.go
@@ -9,6 +9,7 @@ import (
 	"github.com/google/go-cmp/cmp"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/watch"
 )
 
 const (
@@ -48,6 +49,10 @@ func (o fakePlatform) GetPodUsingComponentName(componentName string) (*corev1.Po
 	panic("not implemented yet")
 }
 
+func (o fakePlatform) PodWatcher(ctx context.Context, selector string) (watch.Interface, error) {
+	return nil, nil
+}
+
 func TestExecuteCommand(t *testing.T) {
 	for _, tt := range []struct {
 		name               string

diff --git a/pkg/logs/logs.go b/pkg/logs/logs.go
@@ -6,6 +6,7 @@ import (
 	"io"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/watch"
 
 	odolabels "github.com/redhat-developer/odo/pkg/labels"
 	odocontext "github.com/redhat-developer/odo/pkg/odo/context"
@@ -17,8 +18,9 @@ type LogsClient struct {
 }
 
 type ContainerLogs struct {
-	Name string
-	Logs io.ReadCloser
+	PodName       string
+	ContainerName string
+	Logs          io.ReadCloser
 }
 
 type Events struct {
@@ -80,7 +82,11 @@ func (o *LogsClient) getLogsForMode(
 					if err != nil {
 						events.Err <- fmt.Errorf("failed to get logs for container %s; error: %v", container.Name, err)
 					}
-					events.Logs <- ContainerLogs{container.Name, containerLogs}
+					events.Logs <- ContainerLogs{
+						PodName:       pod.GetName(),
+						ContainerName: container.Name,
+						Logs:          containerLogs,
+					}
 				}
 			case err := <-errChan:
 				events.Err <- err
@@ -92,18 +98,29 @@ func (o *LogsClient) getLogsForMode(
 
 	appname := odocontext.GetApplication(ctx)
 
-	if mode == odolabels.ComponentDevMode || mode == odolabels.ComponentAnyMode {
-		selector = odolabels.GetSelector(componentName, appname, odolabels.ComponentDevMode, false)
-		err := o.getPodsForSelector(selector, namespace, podChan)
+	if follow {
+		podWatcher, err := o.platformClient.PodWatcher(ctx, "")
 		if err != nil {
 			errChan <- err
 		}
-	}
-	if mode == odolabels.ComponentDeployMode || mode == odolabels.ComponentAnyMode {
-		selector = odolabels.GetSelector(componentName, appname, odolabels.ComponentDeployMode, false)
-		err := o.getPodsForSelector(selector, namespace, podChan)
-		if err != nil {
-			errChan <- err
+		for ev := range podWatcher.ResultChan() {
+			switch ev.Type {
+			case watch.Added, watch.Modified:
+				if mode == odolabels.ComponentDevMode || mode == odolabels.ComponentAnyMode {
+					selector = odolabels.GetSelector(componentName, appname, odolabels.ComponentDevMode, false)
+					err := o.getPodsForSelector(selector, namespace, podChan)
+					if err != nil {
+						errChan <- err
+					}
+				}
+				if mode == odolabels.ComponentDeployMode || mode == odolabels.ComponentAnyMode {
+					selector = odolabels.GetSelector(componentName, appname, odolabels.ComponentDeployMode, false)
+					err := o.getPodsForSelector(selector, namespace, podChan)
+					if err != nil {
+						errChan <- err
+					}
+				}
+			}
 		}
 	}
 
@@ -125,7 +142,9 @@ func (o *LogsClient) getPodsForSelector(
 		return err
 	}
 	for _, pod := range podList.Items {
-		pods[pod.GetName()] = struct{}{}
+		if pod.Status.Phase == "Running" {
+			pods[pod.GetName()] = struct{}{}
+		}
 	}
 
 	// get all pods in the namespace
@@ -139,11 +158,15 @@ func (o *LogsClient) getPodsForSelector(
 			// Pod's logs have already been displayed to user
 			continue
 		}
-		podList.Items = append(podList.Items, pod)
+		if pod.Status.Phase == "Running" {
+			podList.Items = append(podList.Items, pod)
+		}
 	}
 
 	for _, pod := range podList.Items {
-		podChan <- pod
+		if pod.Status.Phase == "Running" {
+			podChan <- pod
+		}
 	}
 
 	return nil

diff --git a/pkg/odo/cli/logs/logs.go b/pkg/odo/cli/logs/logs.go
@@ -152,14 +152,36 @@ func (o *LogsOptions) Run(ctx context.Context) error {
 	errChan := make(chan error)          // errors are put on this channel
 	var mu sync.Mutex
 
+	displayedLogs := map[string]struct{}{}
 	for {
 		select {
 		case containerLogs := <-events.Logs:
-			uniqueName := getUniqueContainerName(containerLogs.Name, uniqueContainerNames)
+			podContainerName := fmt.Sprintf("%s-%s", containerLogs.PodName, containerLogs.ContainerName)
+			if _, ok := displayedLogs[podContainerName]; ok {
+				continue
+			}
+			displayedLogs[podContainerName] = struct{}{}
+
+			uniqueName := getUniqueContainerName(containerLogs.ContainerName, uniqueContainerNames)
 			uniqueContainerNames[uniqueName] = struct{}{}
 			colour := log.ColorPicker()
 			logs := containerLogs.Logs
 
+			func() {
+				mu.Lock()
+				defer mu.Unlock()
+				color.Set(colour)
+				defer color.Unset()
+				help := ""
+				if uniqueName != containerLogs.ContainerName {
+					help = fmt.Sprintf(" (%s)", uniqueName)
+				}
+				_, err = fmt.Fprintf(o.out, "--> Logs for %s / %s%s\n", containerLogs.PodName, containerLogs.ContainerName, help)
+				if err != nil {
+					errChan <- err
+				}
+			}()
+
 			if o.follow {
 				atomic.AddInt64(&goroutines.count, 1)
 				go func(out io.Writer) {
@@ -183,7 +205,7 @@ func (o *LogsOptions) Run(ctx context.Context) error {
 		case err = <-events.Err:
 			return err
 		case <-events.Done:
-			if goroutines.count == 0 {
+			if !o.follow && goroutines.count == 0 {
 				if len(uniqueContainerNames) == 0 {
 					// This will be the case when:
 					// 1. user specifies --dev flag, but the component's running in Deploy mode

diff --git a/pkg/platform/interface.go b/pkg/platform/interface.go
@@ -6,6 +6,7 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/watch"
 )
 
 // Client is the interface that wraps operations that can be performed on any supported platform.
@@ -33,4 +34,6 @@ type Client interface {
 	GetRunningPodFromSelector(selector string) (*corev1.Pod, error)
 
 	GetPodUsingComponentName(componentName string) (*corev1.Pod, error)
+
+	PodWatcher(ctx context.Context, selector string) (watch.Interface, error)
 }
diff --git a/pkg/platform/mock.go b/pkg/platform/mock.go
diff --git a/pkg/podman/mock.go b/pkg/podman/mock.go
diff --git a/pkg/podman/pods.go b/pkg/podman/pods.go
@@ -1,13 +1,15 @@
 package podman
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os/exec"
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/klog"
 
 	"github.com/redhat-developer/odo/pkg/platform"
@@ -129,3 +131,7 @@ func (o *PodmanCli) getPodsFromSelector(selector string) ([]ListPodsReport, erro
 	}
 	return list, nil
 }
+
+func (o *PodmanCli) PodWatcher(ctx context.Context, selector string) (watch.Interface, error) {
+	return nil, nil
+}