[ci] capture logs of pods in the test namespace and kube-system (#7341) (#7512)

* fix: path for pod logs dump and uploading them as artifacts

* fix: derive k8s namespace from random uuid

* fix: dump the logs of all pods and the containers statuses into an archive for the test namespace and kube-system

* fix: properly set the namespace for clusterole and clusterrole bindings in kustomize-based k8s integration tests

* fix: extend k8sCheckAgentStatus to catch container restarts

* test: make k8s integration tests be flaky

* fix: time box all exec in pods

* Revert "test: make k8s integration tests be flaky"

This reverts commit e776c64.

(cherry picked from commit 79bcfb7)

Co-authored-by: Panos Koutsovasilis <[email protected]>

Author: mergify[bot]

.buildkite/bk.integration.pipeline.yml

@@ -262,6 +262,7 @@ steps:
         artifact_paths:
           - build/**
           - build/diagnostics/**
+          - build/*.pod_logs_dump/*
         retry:
           automatic:
             limit: 1

.buildkite/scripts/buildkite-k8s-integration-tests.sh

@@ -90,7 +90,7 @@ EOF
   fully_qualified_group_name="${CLUSTER_NAME}_${TARGET_ARCH}_${variant}"
   outputXML="build/${fully_qualified_group_name}.integration.xml"
   outputJSON="build/${fully_qualified_group_name}.integration.out.json"
-  pod_logs_base="build/${fully_qualified_group_name}.integration"
+  pod_logs_base="${PWD}/build/${fully_qualified_group_name}.pod_logs_dump"
 
   set +e
   K8S_TESTS_POD_LOGS_BASE="${pod_logs_base}" AGENT_IMAGE="${image}" DOCKER_VARIANT="${variant}" gotestsum --hide-summary=skipped --format testname --no-color -f standard-quiet --junitfile "${outputXML}" --jsonfile "${outputJSON}" -- -tags kubernetes,integration -test.shuffle on -test.timeout 2h0m0s github.com/elastic/elastic-agent/testing/integration -v -args -integration.groups="${group_name}" -integration.sudo="false"

testing/integration/kubernetes_agent_standalone_test.go

@@ -7,6 +7,7 @@
 package integration
 
 import (
+	"archive/tar"
 	"bufio"
 	"bytes"
 	"context"
@@ -24,6 +25,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/gofrs/uuid/v5"
 	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/elastic-agent-libs/kibana"
@@ -791,8 +793,28 @@ func k8sCheckAgentStatus(ctx context.Context, client klient.Client, stdout *byte
 	namespace string, agentPodName string, containerName string, componentPresence map[string]bool,
 ) error {
 	command := []string{"elastic-agent", "status", "--output=json"}
+	stopCheck := errors.New("stop check")
+
+	// we will wait maximum 120 seconds for the agent to report healthy
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+	defer cancel()
 
 	checkStatus := func() error {
+		pod := corev1.Pod{}
+		if err := client.Resources(namespace).Get(ctx, agentPodName, namespace, &pod); err != nil {
+			return err
+		}
+
+		for _, container := range pod.Status.ContainerStatuses {
+			if container.Name != containerName {
+				continue
+			}
+
+			if restarts := container.RestartCount; restarts != 0 {
+				return fmt.Errorf("container %q of pod %q has restarted %d times: %w", containerName, agentPodName, restarts, stopCheck)
+			}
+		}
+
 		status := atesting.AgentStatusOutput{} // clear status output
 		stdout.Reset()
 		stderr.Reset()
@@ -823,16 +845,14 @@ func k8sCheckAgentStatus(ctx context.Context, client klient.Client, stdout *byte
 		}
 		return err
 	}
-
-	// we will wait maximum 120 seconds for the agent to report healthy
-	timeoutCtx, timeoutCancel := context.WithTimeout(ctx, 120*time.Second)
-	defer timeoutCancel()
 	for {
 		err := checkStatus()
 		if err == nil {
 			return nil
+		} else if errors.Is(err, stopCheck) {
+			return err
 		}
-		if timeoutCtx.Err() != nil {
+		if ctx.Err() != nil {
 			// timeout waiting for agent to become healthy
 			return errors.Join(err, errors.New("timeout waiting for agent to become healthy"))
 		}
@@ -848,7 +868,10 @@ func k8sGetAgentID(ctx context.Context, client klient.Client, stdout *bytes.Buff
 	status := atesting.AgentStatusOutput{} // clear status output
 	stdout.Reset()
 	stderr.Reset()
-	if err := client.Resources().ExecInPod(ctx, namespace, agentPodName, containerName, command, stdout, stderr); err != nil {
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
+	err := client.Resources().ExecInPod(ctx, namespace, agentPodName, containerName, command, stdout, stderr)
+	cancel()
+	if err != nil {
 		return "", err
 	}
 
@@ -869,58 +892,157 @@ func getAgentComponentState(status atesting.AgentStatusOutput, componentName str
 	return -1, false
 }
 
-// k8sDumpAllPodLogs dumps the logs of all pods in the given namespace to the given target directory
-func k8sDumpAllPodLogs(ctx context.Context, client klient.Client, testName string, namespace string, targetDir string) error {
-	podList := &corev1.PodList{}
+// k8sDumpPods creates an archive that contains logs of all pods in the given namespace and kube-system to the given target directory
+func k8sDumpPods(t *testing.T, ctx context.Context, client klient.Client, testName string, namespace string, targetDir string, testStartTime time.Time) {
+	// Create the tar file
+	archivePath := filepath.Join(targetDir, fmt.Sprintf("%s.tar.gz", namespace))
+	tarFile, err := os.Create(archivePath)
+	if err != nil {
+		t.Logf("failed to create archive at path %q", archivePath)
+		return
+	}
+	defer tarFile.Close()
+
+	t.Logf("archive %q contains the dump info for %q test", archivePath, testName)
+
+	// Create a new tar writer
+	tarWriter := tar.NewWriter(tarFile)
+	defer tarWriter.Close()
 
 	clientSet, err := kubernetes.NewForConfig(client.RESTConfig())
 	if err != nil {
-		return fmt.Errorf("error creating clientset: %w", err)
+		t.Logf("error creating clientset: %s", err)
+		return
 	}
 
-	err = client.Resources(namespace).List(ctx, podList)
+	podList := &corev1.PodList{}
+	err = client.Resources("").List(ctx, podList)
 	if err != nil {
-		return fmt.Errorf("error listing pods: %w", err)
+		t.Logf("error listing pods: %s", err)
+		return
 	}
 
-	var errs error
+	type containerPodState struct {
+		Namespace              string `json:"namespace"`
+		PodName                string `json:"pod_name"`
+		ContainerName          string `json:"container_name"`
+		RestartCount           int32  `json:"restart_count"`
+		LastTerminationReason  string `json:"last_termination_reason"`
+		LastTerminationMessage string `json:"last_termination_message"`
+	}
+
+	var statesDump []containerPodState
+
 	for _, pod := range podList.Items {
-		previous := false
-		for _, containerStatus := range pod.Status.ContainerStatuses {
-			if containerStatus.RestartCount > 0 {
-				previous = true
-				break
-			}
+		podNamespace := pod.GetNamespace()
+		if podNamespace != namespace && podNamespace != "kube-system" {
+			continue
 		}
 
 		for _, container := range pod.Spec.Containers {
-			logFilePath := filepath.Join(targetDir, fmt.Sprintf("%s-%s-%s.log", testName, pod.Name, container.Name))
-			logFile, err := os.Create(logFilePath)
-			if err != nil {
-				errs = errors.Join(fmt.Errorf("error creating log file: %w", err), errs)
-				continue
+			state := containerPodState{
+				Namespace:     podNamespace,
+				PodName:       pod.GetName(),
+				ContainerName: container.Name,
 			}
+			previous := false
 
-			req := clientSet.CoreV1().Pods(namespace).GetLogs(pod.Name, &corev1.PodLogOptions{
+			for _, containerStatus := range pod.Status.ContainerStatuses {
+				if container.Name != containerStatus.Name {
+					continue
+				}
+
+				state.RestartCount = containerStatus.RestartCount
+				if containerStatus.RestartCount == 0 {
+					break
+				}
+				// since we dump logs from pods that are expected to constantly run,
+				// namely kube-apiserver in kube-system namespace, we need to identify
+				// if a restart of such pod happened during the test to correctly if we
+				// want previous log
+				containerTerminated := containerStatus.LastTerminationState.Terminated
+				if containerTerminated != nil && containerTerminated.FinishedAt.After(testStartTime) {
+					previous = true
+					state.LastTerminationReason = containerTerminated.Reason
+					state.LastTerminationMessage = containerTerminated.Message
+				}
+				break
+			}
+			statesDump = append(statesDump, state)
+
+			var logFileName string
+			if previous {
+				logFileName = fmt.Sprintf("%s-%s-%s-previous.log", podNamespace, pod.Name, container.Name)
+			} else {
+				logFileName = fmt.Sprintf("%s-%s-%s.log", podNamespace, pod.Name, container.Name)
+			}
+
+			req := clientSet.CoreV1().Pods(podNamespace).GetLogs(pod.Name, &corev1.PodLogOptions{
 				Container: container.Name,
 				Previous:  previous,
+				SinceTime: &metav1.Time{Time: testStartTime},
 			})
-			podLogsStream, err := req.Stream(context.TODO())
+
+			streamCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+			podLogsStream, err := req.Stream(streamCtx)
 			if err != nil {
-				errs = errors.Join(fmt.Errorf("error getting container %s of pod %s logs: %w", container.Name, pod.Name, err), errs)
+				cancel()
+				t.Logf("error getting container %q of pod %q logs: %s", container.Name, pod.Name, err)
 				continue
 			}
 
-			_, err = io.Copy(logFile, podLogsStream)
+			b, err := io.ReadAll(podLogsStream)
+			_ = podLogsStream.Close()
+			cancel()
 			if err != nil {
-				errs = errors.Join(fmt.Errorf("error writing container %s of pod %s logs: %w", container.Name, pod.Name, err), errs)
+				t.Logf("error reading container %q logs of pod %q: %s", container.Name, pod.Name, err)
+				continue
 			}
 
-			_ = podLogsStream.Close()
+			header := &tar.Header{
+				Name:       logFileName,
+				Size:       int64(len(b)),
+				Mode:       0600,
+				ModTime:    time.Now(),
+				AccessTime: time.Now(),
+				ChangeTime: time.Now(),
+			}
+
+			if err := tarWriter.WriteHeader(header); err != nil {
+				t.Logf("error writing header of file %q in archive: %s", logFileName, err)
+				continue
+			}
+
+			if _, err := tarWriter.Write(b); err != nil {
+				t.Logf("error writing data of file %q in archive: %s", logFileName, err)
+			}
 		}
 	}
 
-	return errs
+	b, err := json.Marshal(statesDump)
+	if err != nil {
+		t.Logf("error marshalling pod states: %s", err)
+		return
+	}
+
+	statesDumpFile := "containerPodsStates.json"
+	header := &tar.Header{
+		Name:       statesDumpFile,
+		Size:       int64(len(b)),
+		Mode:       0600,
+		ModTime:    time.Now(),
+		AccessTime: time.Now(),
+		ChangeTime: time.Now(),
+	}
+
+	if err := tarWriter.WriteHeader(header); err != nil {
+		t.Logf("error writing header of file %q in archive: %s", statesDumpFile, err)
+		return
+	}
+
+	if _, err := tarWriter.Write(b); err != nil {
+		t.Logf("error writing data of file %q in archive: %s", statesDumpFile, err)
+	}
 }
 
 // k8sKustomizeAdjustObjects adjusts the namespace of given k8s objects and calls the given callbacks for the containers and the pod
@@ -1142,12 +1264,10 @@ func k8sCreateObjects(ctx context.Context, client klient.Client, opts k8sCreateO
 				for idx := range objWithType.Subjects {
 					objWithType.Subjects[idx].Namespace = opts.namespace
 				}
-				continue
 			case *rbacv1.RoleBinding:
 				for idx := range objWithType.Subjects {
 					objWithType.Subjects[idx].Namespace = opts.namespace
 				}
-				continue
 			}
 		}
 		if err := client.Resources().Create(ctx, obj); err != nil {
@@ -1260,12 +1380,18 @@ type k8sContext struct {
 	esAPIKey string
 	// enrollParams contains the information needed to enroll an agent with Fleet in the test
 	enrollParams *fleettools.EnrollParams
+	// createdAt is the time when the k8sContext was created
+	createdAt time.Time
 }
 
 // getNamespace returns a unique namespace for the current test
-func (k8sContext) getNamespace(t *testing.T) string {
+func (k k8sContext) getNamespace(t *testing.T) string {
+	nsUUID, err := uuid.NewV4()
+	if err != nil {
+		t.Fatalf("error generating namespace UUID: %v", err)
+	}
 	hasher := sha256.New()
-	hasher.Write([]byte(t.Name()))
+	hasher.Write([]byte(nsUUID.String()))
 	testNamespace := strings.ToLower(base64.URLEncoding.EncodeToString(hasher.Sum(nil)))
 	return noSpecialCharsRegexp.ReplaceAllString(testNamespace, "")
 }
@@ -1323,7 +1449,7 @@ func k8sGetContext(t *testing.T, info *define.Info) k8sContext {
 	testLogsBasePath := os.Getenv("K8S_TESTS_POD_LOGS_BASE")
 	require.NotEmpty(t, testLogsBasePath, "K8S_TESTS_POD_LOGS_BASE must be set")
 
-	err = os.MkdirAll(filepath.Join(testLogsBasePath, t.Name()), 0o755)
+	err = os.MkdirAll(testLogsBasePath, 0o755)
 	require.NoError(t, err, "failed to create test logs directory")
 
 	esHost := os.Getenv("ELASTICSEARCH_HOST")
@@ -1346,6 +1472,7 @@ func k8sGetContext(t *testing.T, info *define.Info) k8sContext {
 		esHost:         esHost,
 		esAPIKey:       esAPIKey,
 		enrollParams:   enrollParams,
+		createdAt:      time.Now(),
 	}
 }
 
@@ -1470,9 +1597,7 @@ func k8sStepDeployKustomize(kustomizePath string, containerName string, override
 
 		t.Cleanup(func() {
 			if t.Failed() {
-				if err := k8sDumpAllPodLogs(ctx, kCtx.client, namespace, namespace, kCtx.logsBasePath); err != nil {
-					t.Logf("failed to dump logs: %v", err)
-				}
+				k8sDumpPods(t, ctx, kCtx.client, t.Name(), namespace, kCtx.logsBasePath, kCtx.createdAt)
 			}
 
 			err := k8sDeleteObjects(ctx, kCtx.client, k8sDeleteOpts{wait: true}, objects...)
@@ -1481,7 +1606,7 @@ func k8sStepDeployKustomize(kustomizePath string, containerName string, override
 			}
 		})
 
-		err = k8sCreateObjects(ctx, kCtx.client, k8sCreateOpts{wait: true}, objects...)
+		err = k8sCreateObjects(ctx, kCtx.client, k8sCreateOpts{wait: true, namespace: namespace}, objects...)
 		require.NoError(t, err, "failed to create objects")
 	}
 }
@@ -1544,8 +1669,10 @@ func k8sStepRunInnerTests(agentPodLabelSelector string, expectedPodNumber int, c
 
 		for _, pod := range perNodePodList.Items {
 			var stdout, stderr bytes.Buffer
+			ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
 			err = kCtx.client.Resources().ExecInPod(ctx, namespace, pod.Name, containerName,
 				[]string{"/usr/share/elastic-agent/k8s-inner-tests", "-test.v"}, &stdout, &stderr)
+			cancel()
 			t.Logf("%s k8s-inner-tests output:", pod.Name)
 			t.Log(stdout.String())
 			if err != nil {
@@ -1590,9 +1717,7 @@ func k8sStepHelmDeploy(chartPath string, releaseName string, values map[string]a
 
 		t.Cleanup(func() {
 			if t.Failed() {
-				if err := k8sDumpAllPodLogs(ctx, kCtx.client, namespace, namespace, kCtx.logsBasePath); err != nil {
-					t.Logf("failed to dump logs: %v", err)
-				}
+				k8sDumpPods(t, ctx, kCtx.client, t.Name(), namespace, kCtx.logsBasePath, kCtx.createdAt)
 			}
 
 			uninstallAction := action.NewUninstall(actionConfig)
@@ -1709,7 +1834,9 @@ func k8sStepCheckRestrictUpgrade(agentPodLabelSelector string, expectedPodNumber
 			var stdout, stderr bytes.Buffer
 
 			command := []string{"elastic-agent", "upgrade", "1.0.0"}
+			ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
 			err := kCtx.client.Resources().ExecInPod(ctx, namespace, pod.Name, containerName, command, &stdout, &stderr)
+			cancel()
 			require.Error(t, err)
 			require.Contains(t, stderr.String(), coordinator.ErrNotUpgradable.Error())
 		}