feat: add "node" destination for NodeDiagnostic (#58)

Author: cdirubbio

e2e/setup/e2e.go

@@ -169,4 +169,8 @@ func TestWrapper(t *testing.T, Testenv env.Environment) {
 	// log collection runs at the end, which effectively makes it collects logs
 	// and data from prior tests.
 	Testenv.Test(t, nodediagnostic.LogCollection(awsCfg))
+	Testenv.Test(t,
+		nodediagnostic.LogCollection(awsCfg),
+		nodediagnostic.NodeDestination(),
+	)
 }

e2e/suites/nodediagnostic/nodediagnostic.go

@@ -199,3 +199,93 @@ func assertLogsValid(t *testing.T, reader io.Reader) {
 		t.Logf("found the following paths from the log archive: %s", strings.Join(fileNames, ","))
 	}
 }
+
+func NodeDestination() types.Feature {
+	var nodeDiagnostics []v1alpha1.NodeDiagnostic
+
+	return features.New("NodeDestination").
+		WithLabel("type", "log-collection").
+		Setup(func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+			client, err := cfg.NewClient()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := v1alpha1.SchemeBuilder.AddToScheme(client.Resources().GetScheme()); err != nil {
+				t.Fatal(err)
+			}
+			var nodes corev1.NodeList
+			if err := client.Resources().List(ctx, &nodes); err != nil {
+				t.Fatal(err)
+			}
+			if len(nodes.Items) == 0 {
+				t.Fatal("no nodes were found in the cluster")
+			}
+			for _, node := range nodes.Items {
+				if node.DeletionTimestamp != nil {
+					t.Logf("skipping node %q because it is being deleted", node.Name)
+					continue
+				}
+
+				nodeDiagnostic := v1alpha1.NodeDiagnostic{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: node.Name,
+					},
+					Spec: v1alpha1.NodeDiagnosticSpec{
+						LogCapture: &v1alpha1.LogCapture{
+							UploadDestination: "node",
+						},
+					},
+				}
+				t.Logf("creating NodeDiagnostic for node %q with node destination", node.Name)
+				if err := client.Resources().Create(ctx, &nodeDiagnostic); err != nil {
+					t.Fatal(err)
+				}
+				nodeDiagnostics = append(nodeDiagnostics, nodeDiagnostic)
+			}
+			if len(nodeDiagnostics) == 0 {
+				t.Fatal("no non-terminating nodes were found in the cluster")
+			}
+			return ctx
+		}).
+		Assess("CollectLogs", func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+			for _, nodeDiagnostic := range nodeDiagnostics {
+				t.Run(nodeDiagnostic.Name, func(t *testing.T) {
+					if err := cfg.Client().Resources().Get(ctx, nodeDiagnostic.Name, nodeDiagnostic.Namespace, &nodeDiagnostic); err != nil {
+						t.Fatal(err)
+					}
+
+					if err := wait.For(
+						conditions.New(cfg.Client().Resources()).ResourceMatch(&nodeDiagnostic, func(object k8s.Object) bool {
+							nd := object.(*v1alpha1.NodeDiagnostic)
+							return len(nd.Status.CaptureStatuses) > 0 && nd.Status.CaptureStatuses[0].State.Completed != nil
+						}),
+						wait.WithTimeout(time.Minute),
+						wait.WithContext(ctx),
+					); err != nil {
+						t.Error(err)
+					}
+					for _, status := range nodeDiagnostic.Status.CaptureStatuses {
+						if status.State.Completed == nil {
+							t.Errorf("capture was not complete: %+v", status.State)
+						} else if status.State.Completed.Reason == v1alpha1.CaptureStateFailure {
+							t.Errorf("capture failed with reason: %s, message: %s", status.State.Completed.Reason, status.State.Completed.Message)
+						} else {
+							t.Logf("capture succeeded with reason: %s, message: %s", status.State.Completed.Reason, status.State.Completed.Message)
+						}
+					}
+				})
+			}
+			return ctx
+		}).
+		Teardown(func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+			for _, nodeDiagnostic := range nodeDiagnostics {
+				t.Run(nodeDiagnostic.Name, func(t *testing.T) {
+					if err := cfg.Client().Resources().Delete(ctx, &nodeDiagnostic); err != nil {
+						t.Fatal(err)
+					}
+				})
+			}
+			return ctx
+		}).
+		Feature()
+}

pkg/controllers/nodediagnostic.go

@@ -9,6 +9,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	controllerruntime "sigs.k8s.io/controller-runtime"
@@ -100,6 +101,83 @@ func (c *nodeDiagnosticController) Reconcile(ctx context.Context, nodeDiagnostic
 		}
 
 		log.Info("uploading logs", "url", nodeDiagnostic.Spec.UploadDestination)
+
+		if nodeDiagnostic.Spec.UploadDestination == "node" {
+			log.Info("saving logs to /var/log/support for download via node proxy")
+			supportDir := filepath.Join(config.HostRoot(), "var/log/support")
+			if err := os.MkdirAll(supportDir, 0600); err != nil {
+				log.Error(err, "failed to create support directory")
+				captureStatus.State = v1alpha1.CaptureState{
+					Completed: &v1alpha1.CaptureStateCompleted{
+						Reason:     v1alpha1.CaptureStateFailure,
+						Message:    "failed to create support directory",
+						StartedAt:  captureStatus.State.Running.StartedAt,
+						FinishedAt: metav1.Now(),
+					},
+				}
+				stored := nodeDiagnostic.DeepCopy()
+				nodeDiagnostic.Status.SetCaptureStatus(captureStatus)
+				return reconcile.Result{}, c.kubeClient.Status().Patch(ctx, nodeDiagnostic, client.MergeFrom(stored))
+			}
+			destPath := filepath.Join(supportDir, fmt.Sprintf("%s-logs.tar.gz", c.nodeName))
+			destFile, err := os.Create(destPath)
+			if err != nil {
+				log.Error(err, "failed to create log file")
+				captureStatus.State = v1alpha1.CaptureState{
+					Completed: &v1alpha1.CaptureStateCompleted{
+						Reason:     v1alpha1.CaptureStateFailure,
+						Message:    "failed to create log file",
+						StartedAt:  captureStatus.State.Running.StartedAt,
+						FinishedAt: metav1.Now(),
+					},
+				}
+				stored := nodeDiagnostic.DeepCopy()
+				nodeDiagnostic.Status.SetCaptureStatus(captureStatus)
+				return reconcile.Result{}, c.kubeClient.Status().Patch(ctx, nodeDiagnostic, client.MergeFrom(stored))
+			}
+			defer destFile.Close()
+			if _, err := io.Copy(destFile, archiveReader); err != nil {
+				log.Error(err, "failed to write logs to file")
+				captureStatus.State = v1alpha1.CaptureState{
+					Completed: &v1alpha1.CaptureStateCompleted{
+						Reason:     v1alpha1.CaptureStateFailure,
+						Message:    "failed to write logs to file",
+						StartedAt:  captureStatus.State.Running.StartedAt,
+						FinishedAt: metav1.Now(),
+					},
+				}
+				stored := nodeDiagnostic.DeepCopy()
+				nodeDiagnostic.Status.SetCaptureStatus(captureStatus)
+				return reconcile.Result{}, c.kubeClient.Status().Patch(ctx, nodeDiagnostic, client.MergeFrom(stored))
+			}
+			log.Info("logs saved successfully", "path", destPath)
+			captureStatus.State = v1alpha1.CaptureState{
+				Completed: &v1alpha1.CaptureStateCompleted{
+					Reason:     v1alpha1.CaptureStateSuccess,
+					Message:    fmt.Sprintf("successfully saved logs to %s", destPath),
+					StartedAt:  captureStatus.State.Running.StartedAt,
+					FinishedAt: metav1.Now(),
+				},
+			}
+			if issueCount > 0 {
+				captureStatus.State.Completed.Reason = v1alpha1.CaptureStateSuccessWithErrors
+				captureStatus.State.Completed.Message = fmt.Sprintf("successfully saved logs to %s with some errors", destPath)
+			}
+			stored := nodeDiagnostic.DeepCopy()
+			nodeDiagnostic.Status.SetCaptureStatus(captureStatus)
+			// Delete file from /var/log/support after 10 minutes
+			go func() {
+				time.Sleep(600 * time.Second)
+				if err := os.Remove(destPath); err != nil {
+					if !os.IsNotExist(err) {
+						log.Error(err, "failed to delete log file after timeout", "path", destPath)
+					}
+				} else {
+					log.Info("successfully deleted log file after timeout", "path", destPath)
+				}
+			}()
+			return reconcile.Result{}, c.kubeClient.Status().Patch(ctx, nodeDiagnostic, client.MergeFrom(stored))
+		}
 		// wrapping this setup into one function to avoid redundant failure code
 		doUpload := func() error {
 			uploadRequest, err := http.NewRequestWithContext(ctx, http.MethodPut, string(nodeDiagnostic.Spec.UploadDestination), archiveReader)