merge main

Signed-off-by: Julie Vogelman <[email protected]>

.github/workflows/ci.yaml

@@ -160,4 +160,13 @@ jobs:
         env:
           GOPATH: /home/runner/go
         run: KUBECONFIG=~/.kube/config VERSION=${{ github.sha }} DOCKER_PUSH=true DATA_LOSS_PREVENTION=${{matrix.datalossprevention}} make test-e2e
+
+      - name: Archive controller logs
+        uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: controller-logs-${{matrix.datalossprevention}}
+          path: |
+            tests/e2e/output/
+          retention-days: 7
 

.gitignore

@@ -33,3 +33,5 @@ config/crd/bases/_.yaml
 .idea
 
 config/crd/external/
+
+tests/e2e/output

extensions/argo/numa-rollout/numa-rollout-extension/ui/src/rollout/MonovertexRollout.tsx

@@ -6,6 +6,8 @@ import { SquareCancelIcon } from "../utils/SquareCancelIcon";
 
 export const MonovertexRollout = () => {
   const { props, kindToNodeMap } = useContext(RolloutComponentContext);
+  const conditions = props?.resource?.status?.conditions
+  const hasChildResourcesHealthy = conditions.some(condition => condition.type === 'ChildResourcesHealthy');
 
   return (
     <Box>
@@ -28,8 +30,7 @@ export const MonovertexRollout = () => {
                 }}
               >
                 Monovertex Status:{" "}
-                {props?.resource?.status?.conditions?.[0].type ===
-                "ChildResourcesHealthy" ? (
+                {hasChildResourcesHealthy ? (
                   <SquareCheckIcon tooltipTitle={node?.name} />
                 ) : (
                   <SquareCancelIcon tooltipTitle={node?.name} />

extensions/argo/numa-rollout/numa-rollout-extension/ui/src/rollout/PipelineRollout.tsx

@@ -6,6 +6,8 @@ import { SquareCancelIcon } from "../utils/SquareCancelIcon";
 
 export const PipelineRollout = () => {
   const { props, kindToNodeMap } = useContext(RolloutComponentContext);
+  const conditions = props?.resource?.status?.conditions
+  const hasChildResourcesHealthy = conditions.some(condition => condition.type === 'ChildResourcesHealthy');
 
   return (
     <Box>
@@ -28,8 +30,7 @@ export const PipelineRollout = () => {
                 }}
               >
                 Pipeline Status:{" "}
-                {props?.resource?.status?.conditions?.[0].type ===
-                "ChildResourcesHealthy" ? (
+                {hasChildResourcesHealthy ? (
                   <SquareCheckIcon tooltipTitle={node?.name} />
                 ) : (
                   <SquareCancelIcon tooltipTitle={node?.name} />

internal/common/common.go

@@ -9,6 +9,10 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
+// UpgradeState is the enum to track the possible state of
+// a resource upgrade, it can only be `promoted` or `in-progress`.
+type UpgradeState string
+
 const (
 	// SSAManager is the default numaplane manager name used by server-side apply syncs
 	SSAManager = "numaplane-controller"
@@ -43,6 +47,18 @@ const (
 	// LabelKeyPipelineRolloutForPipeline is the label key used to identify the PipelineRollout a Pipeline is managed by
 	// This is useful as a Label to quickly locate all Pipelines of a given PipelineRollout
 	LabelKeyPipelineRolloutForPipeline = "numaplane.numaproj.io/pipeline-rollout-name"
+
+	// LabelKeyUpgradeState is the label key used to identify the upgrade state of a resource that is managed by
+	// a NumaRollout.
+	LabelKeyUpgradeState = "numaplane.numaproj.io/upgrade-state"
+
+	// LabelValueUpgradePromoted is the label value indicating that the resource managed by a NumaRollout is promoted
+	// after an upgrade.
+	LabelValueUpgradePromoted UpgradeState = "promoted"
+
+	// LabelValueUpgradeInProgress is the label value indicating that the resource managed by a NumaRollout is in progress
+	// of upgrade.
+	LabelValueUpgradeInProgress UpgradeState = "in-progress"
 )
 
 var (

internal/controller/pipelinerollout_controller.go

@@ -1002,6 +1002,7 @@ func pipelineLabels(pipelineRollout *apiv1.PipelineRollout) (map[string]string,
 	}
 
 	labelMapping[common.LabelKeyPipelineRolloutForPipeline] = pipelineRollout.Name
+	labelMapping[common.LabelKeyUpgradeState] = string(common.LabelValueUpgradePromoted)
 
 	return labelMapping, nil
 }

internal/controller/pipelinerollout_controller_test.go

@@ -21,23 +21,20 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
-	"time"
-
-	"github.com/numaproj/numaplane/internal/common"
-
 	"testing"
-
-	"github.com/stretchr/testify/assert"
+	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 
 	numaflowv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
+	"github.com/numaproj/numaplane/internal/common"
 	"github.com/numaproj/numaplane/internal/util/kubernetes"
 	apiv1 "github.com/numaproj/numaplane/pkg/apis/numaplane/v1alpha1"
 )
@@ -144,6 +141,7 @@ var _ = Describe("PipelineRollout Controller", Ordered, func() {
 
 			By("Verifying the label of the pipeline")
 			Expect(createdResource.Labels[common.LabelKeyPipelineRolloutForPipeline]).Should(Equal(pipelineRollout.Name))
+			Expect(createdResource.Labels[common.LabelKeyUpgradeState]).Should(Equal(string(common.LabelValueUpgradePromoted)))
 		})
 
 		It("Should have the PipelineRollout Status Phase has Deployed and ObservedGeneration matching Generation", func() {
@@ -679,6 +677,10 @@ func TestPipelineLabels(t *testing.T) {
 				if labels[common.LabelKeyPipelineRolloutForPipeline] != pipelineRolloutName {
 					t.Errorf("pipelineLabels() = %v, expected %v", common.LabelKeyPipelineRolloutForPipeline, pipelineRolloutName)
 				}
+
+				if labels[common.LabelKeyUpgradeState] != string(common.LabelValueUpgradePromoted) {
+					t.Errorf("pipelineLabels() = %v, expected %v", common.LabelKeyUpgradeState, string(common.LabelValueUpgradePromoted))
+				}
 			}
 		})
 	}

tests/e2e/common.go

@@ -3,6 +3,8 @@ package e2e
 import (
 	"context"
 	"fmt"
+	"io"
+	"os"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -19,6 +21,7 @@ import (
 	"github.com/numaproj/numaplane/internal/util/kubernetes"
 	apiv1 "github.com/numaproj/numaplane/pkg/apis/numaplane/v1alpha1"
 	planepkg "github.com/numaproj/numaplane/pkg/client/clientset/versioned/typed/numaplane/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
 )
 
 var (
@@ -41,6 +44,12 @@ var (
 
 const (
 	Namespace = "numaplane-system"
+
+	NumaplaneCtrlLogs = "output/numaplane-controller.log"
+	NumaflowCtrlLogs  = "output/numaflow-controller.log"
+
+	NumaplaneLabel = "app.kubernetes.io/part-of=numaplane"
+	NumaflowLabel  = "app.kubernetes.io/part-of=numaflow, app.kubernetes.io/component=controller-manager"
 )
 
 // document for Ginkgo framework and print to console
@@ -94,16 +103,31 @@ func getNumaflowResourceStatus(u *unstructured.Unstructured) (kubernetes.Generic
 	return status, err
 }
 
-// commenting out to please Lint, but leaving here because it could be useful later
-/*
-func printPodLogs(client clientgo.Interface, namespace, podName, containerName string) {
-	podLogOptions := &apiv1.PodLogOptions{Container: containerName}
-	stream, err := client.CoreV1().Pods(namespace).GetLogs(podName, podLogOptions).Stream(ctx)
+func getPodLogs(client clientgo.Interface, namespace, labelSelector, containerName, fileName string) {
+
+	ctx := context.Background()
+	podLogOptions := &corev1.PodLogOptions{Container: containerName}
+
+	podList, err := client.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{LabelSelector: labelSelector})
 	if err != nil {
-		fmt.Printf("Error getting Pod logs: namespace=%q, pod=%q, container=%q\n", namespace, podName, containerName)
+		fmt.Printf("Error listing pods: %v\n", err)
 		return
 	}
-	defer stream.Close()
-	logBytes, _ := io.ReadAll(stream)
-	fmt.Printf("Printing Log for namespace=%q, pod=%q, container=%q:\n%s\n", namespace, podName, containerName, string(logBytes))
-}*/
+
+	for _, pod := range podList.Items {
+		stream, err := client.CoreV1().Pods(namespace).GetLogs(pod.Name, podLogOptions).Stream(ctx)
+		if err != nil {
+			fmt.Printf("Error getting pods logs: %v\n", err)
+			return
+		}
+		defer stream.Close()
+		logBytes, _ := io.ReadAll(stream)
+
+		err = os.WriteFile(fileName, logBytes, 0644)
+		if err != nil {
+			fmt.Printf("Error writing pod logs to file: %v\n", err)
+			return
+		}
+	}
+
+}

tests/e2e/suite_test.go

@@ -45,14 +45,17 @@ func TestE2E(t *testing.T) {
 
 var _ = BeforeSuite(func() {
 
+	var err error
+	err = os.Mkdir("output", os.ModePerm)
+	Expect(err).NotTo(HaveOccurred())
+
 	dataLossPrevention = os.Getenv("DATA_LOSS_PREVENTION")
 
 	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
 
 	By("bootstrapping test environment")
 	ctx, cancel = context.WithTimeout(context.Background(), suiteTimeout) // Note: if we start seeing "client rate limiter: context deadline exceeded", we need to increase this value
 
-	var err error
 	scheme := runtime.NewScheme()
 	err = apiv1.AddToScheme(scheme)
 	Expect(err).NotTo(HaveOccurred())
@@ -103,7 +106,19 @@ var _ = AfterSuite(func() {
 
 	cancel()
 	By("tearing down test environment")
+	getPodLogs(kubeClient, Namespace, NumaplaneLabel, "manager", NumaplaneCtrlLogs)
 	err := testEnv.Stop()
 	Expect(err).NotTo(HaveOccurred())
 
 })
+
+var _ = AfterEach(func() {
+
+	report := CurrentSpecReport()
+	if report.Failed() {
+		getPodLogs(kubeClient, Namespace, NumaplaneLabel, "manager", NumaplaneCtrlLogs)
+		getPodLogs(kubeClient, Namespace, NumaflowLabel, "controller-manager", NumaflowCtrlLogs)
+		AbortSuite("Test spec has failed, aborting suite run")
+	}
+
+})