feat(collectors): Store all pod logs in cluster-resources directory (#821)

* feat(collectors): Store all pod logs in cluster-resources directory

All pod logs collected by the logs collector will now be stored in
/cluster-resources/pods/logs/[namespace]/[pod]/[container].log. This
will provide consistency and allow sbctl to find the logs when we run
`kubectl logs <pod>`. To allow backwards compatibility, symlinks of the
log files will be created in the current expected locations.

Closes: #744

Author: banjoh

.gitignore

@@ -38,3 +38,6 @@ sbom/
 
 # Ignore local pre-commit config
 .pre-commit-config.yaml
+
+# Ignore generated support bundles
+*.tar.gz

examples/preflight/e2e.yaml

@@ -9,7 +9,7 @@ spec:
         data: "5"
     - run:
         collectorName: "static-hi"
-        image: 'alpine:3.5'
+        image: 'alpine:3'
         command: ["echo", "hi static!"]
   analyzers:
     - clusterVersion:

examples/support-bundle/e2e.yaml

@@ -11,10 +11,10 @@ spec:
         data: "5"
     - runPod:
         collectorName: "static-hi"
-        podSpec: 
+        podSpec:
           containers:
           - name: static-hi
-            image: alpine:3.5
+            image: alpine:3
             command: ["echo", "hi static!"]
   analyzers:
     - clusterVersion:

examples/support-bundle/sample-supportbundle.yaml

@@ -12,6 +12,16 @@ spec:
         limits:
           maxAge: 720h # 30*24
           maxLines: 10000
+    - logs:
+        collectorName: all-logs
+        name: all-logs
+    - runPod:
+        collectorName: "static-hi"
+        podSpec:
+          containers:
+          - name: static-hi
+            image: alpine:3
+            command: ["echo", "hi static!"]
   analyzers:
     - clusterVersion:
         outcomes:
@@ -100,3 +110,12 @@ spec:
               message: The API deployment has only a single ready replica.
           - pass:
               message: There are multiple replicas of the API deployment ready.
+    - textAnalyze:
+        checkName: Said hi!
+        fileName: /static-hi.log
+        regex: 'hi static'
+        outcomes:
+          - fail:
+              message: Didn't say hi.
+          - pass:
+              message: Said hi!

pkg/collect/cluster_resources.go

@@ -131,7 +131,7 @@ func (c *CollectClusterResources) Collect(progressChan chan<- interface{}) (Coll
 			limits := &troubleshootv1beta2.LogLimits{
 				MaxLines: 500,
 			}
-			podLogs, err := savePodLogs(ctx, logsRoot, client, pod, "", container.Name, limits, false)
+			podLogs, err := savePodLogs(ctx, logsRoot, client, &pod, "", container.Name, limits, false)
 			if err != nil {
 				errPath := filepath.Join("cluster-resources", "pods", "logs", pod.Namespace, pod.Name, fmt.Sprintf("%s-logs-errors.log", container.Name))
 				output.SaveResult(c.BundlePath, errPath, bytes.NewBuffer([]byte(err.Error())))

pkg/collect/logs.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"path/filepath"
 	"strings"
 	"time"
 
@@ -70,10 +71,7 @@ func (c *CollectLogs) Collect(progressChan chan<- interface{}) (CollectorResult,
 				}
 
 				for _, containerName := range containerNames {
-					if len(containerNames) == 1 {
-						containerName = "" // if there was only one container, use the old behavior of not including the container name in the path
-					}
-					podLogs, err := savePodLogs(ctx, c.BundlePath, client, pod, c.Collector.Name, containerName, c.Collector.Limits, false)
+					podLogs, err := savePodLogs(ctx, c.BundlePath, client, &pod, c.Collector.Name, containerName, c.Collector.Limits, false)
 					if err != nil {
 						key := fmt.Sprintf("%s/%s-errors.json", c.Collector.Name, pod.Name)
 						if containerName != "" {
@@ -91,7 +89,7 @@ func (c *CollectLogs) Collect(progressChan chan<- interface{}) (CollectorResult,
 				}
 			} else {
 				for _, container := range c.Collector.ContainerNames {
-					containerLogs, err := savePodLogs(ctx, c.BundlePath, client, pod, c.Collector.Name, container, c.Collector.Limits, false)
+					containerLogs, err := savePodLogs(ctx, c.BundlePath, client, &pod, c.Collector.Name, container, c.Collector.Limits, false)
 					if err != nil {
 						key := fmt.Sprintf("%s/%s/%s-errors.json", c.Collector.Name, pod.Name, container)
 						err := output.SaveResult(c.BundlePath, key, marshalErrors([]string{err.Error()}))
@@ -111,7 +109,7 @@ func (c *CollectLogs) Collect(progressChan chan<- interface{}) (CollectorResult,
 	return output, nil
 }
 
-func listPodsInSelectors(ctx context.Context, client *kubernetes.Clientset, namespace string, selector []string) ([]corev1.Pod, []string) {
+func listPodsInSelectors(ctx context.Context, client kubernetes.Interface, namespace string, selector []string) ([]corev1.Pod, []string) {
 	serializedLabelSelector := strings.Join(selector, ",")
 
 	listOptions := metav1.ListOptions{
@@ -126,20 +124,54 @@ func listPodsInSelectors(ctx context.Context, client *kubernetes.Clientset, name
 	return pods.Items, nil
 }
 
-func savePodLogs(ctx context.Context, bundlePath string, client *kubernetes.Clientset, pod corev1.Pod, name, container string, limits *troubleshootv1beta2.LogLimits, follow bool) (CollectorResult, error) {
+func savePodLogs(
+	ctx context.Context,
+	bundlePath string,
+	client *kubernetes.Clientset,
+	pod *corev1.Pod,
+	collectorName, container string,
+	limits *troubleshootv1beta2.LogLimits,
+	follow bool,
+) (CollectorResult, error) {
+	return savePodLogsWithInterface(ctx, bundlePath, client, pod, collectorName, container, limits, follow)
+}
+
+func savePodLogsWithInterface(
+	ctx context.Context,
+	bundlePath string,
+	client kubernetes.Interface,
+	pod *corev1.Pod,
+	collectorName, container string,
+	limits *troubleshootv1beta2.LogLimits,
+	follow bool,
+) (CollectorResult, error) {
 	podLogOpts := corev1.PodLogOptions{
 		Follow:    follow,
 		Container: container,
 	}
 
-	setLogLimits(&podLogOpts, limits, convertMaxAgeToTime)
+	result := NewResult()
 
-	fileKey := fmt.Sprintf("%s/%s", name, pod.Name)
+	// TODO: Abstract away hard coded directory structure paths
+	// Maybe create a FS provider or something similar
+	filePathPrefix := filepath.Join(
+		"cluster-resources", "pods", "logs", pod.Namespace, pod.Name, pod.Spec.Containers[0].Name,
+	)
+
+	// TODO: If collectorName is empty, the path is stored with a leading slash
+	// Retain this behavior otherwise analysers in the wild may break
+	// Analysers that need to find a file in the root of the bundle should
+	// prefix the path with a slash e.g /file.txt. This behavior should be
+	// properly deprecated in the future.
+	linkRelPathPrefix := fmt.Sprintf("%s/%s", collectorName, pod.Name)
 	if container != "" {
-		fileKey = fmt.Sprintf("%s/%s/%s", name, pod.Name, container)
+		linkRelPathPrefix = fmt.Sprintf("%s/%s/%s", collectorName, pod.Name, container)
+		filePathPrefix = filepath.Join(
+			"cluster-resources", "pods", "logs", pod.Namespace, pod.Name, container,
+		)
 	}
 
-	result := NewResult()
+	setLogLimits(&podLogOpts, limits, convertMaxAgeToTime)
 
 	req := client.CoreV1().Pods(pod.Namespace).GetLogs(pod.Name, &podLogOpts)
 	podLogs, err := req.Stream(ctx)
@@ -148,11 +180,13 @@ func savePodLogs(ctx context.Context, bundlePath string, client *kubernetes.Clie
 	}
 	defer podLogs.Close()
 
-	logWriter, err := result.GetWriter(bundlePath, fileKey+".log")
+	logWriter, err := result.GetWriter(bundlePath, filePathPrefix+".log")
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get log writer")
 	}
-	defer result.CloseWriter(bundlePath, fileKey+".log", logWriter)
+	// NOTE: deferred calls are executed in LIFO order i.e called in reverse order
+	defer result.SymLinkResult(bundlePath, linkRelPathPrefix+".log", filePathPrefix+".log")
+	defer result.CloseWriter(bundlePath, filePathPrefix+".log", logWriter)
 
 	_, err = io.Copy(logWriter, podLogs)
 	if err != nil {
@@ -168,11 +202,13 @@ func savePodLogs(ctx context.Context, bundlePath string, client *kubernetes.Clie
 	}
 	defer podLogs.Close()
 
-	prevLogWriter, err := result.GetWriter(bundlePath, fileKey+"-previous.log")
+	prevLogWriter, err := result.GetWriter(bundlePath, filePathPrefix+"-previous.log")
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get previous log writer")
 	}
-	defer result.CloseWriter(bundlePath, fileKey+"-previous.log", logWriter)
+	// NOTE: deferred calls are executed in LIFO order i.e called in reverse order
+	defer result.SymLinkResult(bundlePath, linkRelPathPrefix+"-previous.log", filePathPrefix+"-previous.log")
+	defer result.CloseWriter(bundlePath, filePathPrefix+"-previous.log", logWriter)
 
 	_, err = io.Copy(prevLogWriter, podLogs)
 	if err != nil {

pkg/collect/logs_test.go

@@ -1,6 +1,7 @@
 package collect
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -9,6 +10,7 @@ import (
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	testclient "k8s.io/client-go/kubernetes/fake"
 )
 
 func Test_setLogLimits(t *testing.T) {
@@ -77,3 +79,74 @@ func Test_setLogLimits(t *testing.T) {
 		})
 	}
 }
+
+func Test_savePodLogs(t *testing.T) {
+	tests := []struct {
+		name              string
+		withContainerName bool
+		collectorName     string
+		want              CollectorResult
+	}{
+		{
+			name:              "with container name",
+			withContainerName: true,
+			collectorName:     "all-logs",
+			want: CollectorResult{
+				"all-logs/test-pod/nginx.log":                                          []byte("fake logs"),
+				"all-logs/test-pod/nginx-previous.log":                                 []byte("fake logs"),
+				"cluster-resources/pods/logs/my-namespace/test-pod/nginx.log":          []byte("fake logs"),
+				"cluster-resources/pods/logs/my-namespace/test-pod/nginx-previous.log": []byte("fake logs"),
+			},
+		},
+		{
+			name:              "without container name",
+			withContainerName: false,
+			collectorName:     "all-logs",
+			want: CollectorResult{
+				"all-logs/test-pod.log":                                                []byte("fake logs"),
+				"all-logs/test-pod-previous.log":                                       []byte("fake logs"),
+				"cluster-resources/pods/logs/my-namespace/test-pod/nginx.log":          []byte("fake logs"),
+				"cluster-resources/pods/logs/my-namespace/test-pod/nginx-previous.log": []byte("fake logs"),
+			},
+		},
+		{
+			name:              "without container or collector names",
+			withContainerName: false,
+			want: CollectorResult{
+				"/test-pod.log":          []byte("fake logs"),
+				"/test-pod-previous.log": []byte("fake logs"),
+				"cluster-resources/pods/logs/my-namespace/test-pod/nginx.log":          []byte("fake logs"),
+				"cluster-resources/pods/logs/my-namespace/test-pod/nginx-previous.log": []byte("fake logs"),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.TODO()
+			containerName := "nginx"
+			client := testclient.NewSimpleClientset()
+			limits := &troubleshootv1beta2.LogLimits{
+				MaxLines: 500,
+			}
+			pod, err := client.CoreV1().Pods("my-namespace").Create(ctx, &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-pod",
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: containerName,
+						},
+					},
+				},
+			}, metav1.CreateOptions{})
+			assert.NoError(t, err)
+			if !tt.withContainerName {
+				containerName = ""
+			}
+			got, err := savePodLogsWithInterface(ctx, "", client, pod, tt.collectorName, containerName, limits, false)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

pkg/collect/redact.go

@@ -17,9 +17,28 @@ import (
 
 func RedactResult(bundlePath string, input CollectorResult, additionalRedactors []*troubleshootv1beta2.Redact) error {
 	for k, v := range input {
+		file := k
+
 		var reader io.Reader
 		if v == nil {
-			r, err := input.GetReader(bundlePath, k)
+			// Collected contents are in a file. Get a reader to the file.
+			info, err := os.Lstat(filepath.Join(bundlePath, file))
+			if err != nil {
+				if os.IsNotExist(errors.Cause(err)) {
+					// File not found, moving on.
+					continue
+				}
+				return errors.Wrap(err, "failed to stat file")
+			}
+
+			// Redact the target file of a symlink
+			if info.Mode().Type() == os.ModeSymlink {
+				file, err = os.Readlink(filepath.Join(bundlePath, file))
+				if err != nil {
+					return errors.Wrap(err, "failed to read symlink")
+				}
+			}
+			r, err := input.GetReader(bundlePath, file)
 			if err != nil {
 				if os.IsNotExist(errors.Cause(err)) {
 					continue
@@ -30,19 +49,20 @@ func RedactResult(bundlePath string, input CollectorResult, additionalRedactors
 
 			reader = r
 		} else {
+			// Collected contents are in memory. Get a reader to the memory buffer.
 			reader = bytes.NewBuffer(v)
 		}
 
 		//If the file is .tar, .tgz or .tar.gz, it must not be redacted. Instead it is decompressed and each file inside the
 		//tar is decompressed, redacted and compressed back into the tar.
-		if filepath.Ext(k) == ".tar" || filepath.Ext(k) == ".tgz" || strings.HasSuffix(k, ".tar.gz") {
+		if filepath.Ext(file) == ".tar" || filepath.Ext(file) == ".tgz" || strings.HasSuffix(file, ".tar.gz") {
 			tmpDir, err := ioutil.TempDir("", "troubleshoot-subresult-")
 			if err != nil {
 				return errors.Wrap(err, "failed to create temp dir")
 			}
 			defer os.RemoveAll(tmpDir)
 
-			subResult, tarHeaders, err := decompressFile(tmpDir, reader, k)
+			subResult, tarHeaders, err := decompressFile(tmpDir, reader, file)
 			if err != nil {
 				return errors.Wrap(err, "failed to decompress file")
 			}
@@ -51,7 +71,7 @@ func RedactResult(bundlePath string, input CollectorResult, additionalRedactors
 				return errors.Wrap(err, "failed to redact file")
 			}
 
-			dstFilename := filepath.Join(bundlePath, k)
+			dstFilename := filepath.Join(bundlePath, file)
 			err = compressFiles(tmpDir, subResult, tarHeaders, dstFilename)
 			if err != nil {
 				return errors.Wrap(err, "failed to re-compress file")
@@ -63,12 +83,12 @@ func RedactResult(bundlePath string, input CollectorResult, additionalRedactors
 			continue
 		}
 
-		redacted, err := redact.Redact(reader, k, additionalRedactors)
+		redacted, err := redact.Redact(reader, file, additionalRedactors)
 		if err != nil {
-			return errors.Wrap(err, "failed to redact")
+			return errors.Wrap(err, "failed to redact io stream")
 		}
 
-		err = input.ReplaceResult(bundlePath, k, redacted)
+		err = input.ReplaceResult(bundlePath, file, redacted)
 		if err != nil {
 			return errors.Wrap(err, "failed to create redacted result")
 		}

pkg/collect/remote_collector.go

@@ -109,7 +109,7 @@ func (c *RemoteCollector) RunCollectorSync(globalRedactors []*troubleshootv1beta
 
 	if err = RedactResult("", result, globalRedactors); err != nil {
 		// Returning result on error to be consistent with local collector.
-		return result, errors.Wrap(err, "failed to redact")
+		return result, errors.Wrap(err, "failed to redact remote collector results")
 	}
 	return result, nil
 }