Add message to event and update status condition when job fails

Signed-off-by: Brad Davidson <brad.davidson@rancher.com>

Author: brandond

e2e/framework/controller/deployment.go

@@ -51,6 +51,9 @@ func NewDeployment(name string, opt ...DeploymentOption) *appsv1.Deployment {
 }, {
 Name: "SYSTEM_UPGRADE_CONTROLLER_LEADER_ELECT",
 Value: "true",
+}, {
+Name: "SYSTEM_UPGRADE_CONTROLLER_DEBUG",
+Value: "true",
 }, {
 Name: "SYSTEM_UPGRADE_CONTROLLER_NAMESPACE",
 ValueFrom: &corev1.EnvVarSource{

e2e/suite/channel_resolve_test.go

@@ -65,5 +65,6 @@ var _ = Describe("Resolve channel", func() {
 Expect(err).To(HaveOccurred())
 Expect(latest).To(BeEmpty())
 })
+AfterEach(CollectLogsOnFailure(e2e))
 })
 })

e2e/suite/job_generate_test.go

@@ -8,13 +8,17 @@ import (
 
 . "github.com/onsi/ginkgo/v2"
 . "github.com/onsi/gomega"
+format "github.com/onsi/gomega/format"
 batchv1 "k8s.io/api/batch/v1"
 v1 "k8s.io/api/core/v1"
+"k8s.io/utils/pointer"
 "k8s.io/utils/ptr"
 
 "github.com/rancher/system-upgrade-controller/e2e/framework"
 upgradeapiv1 "github.com/rancher/system-upgrade-controller/pkg/apis/upgrade.cattle.io/v1"
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+_ "k8s.io/kubernetes/test/utils/format"
 )
 
 var _ = Describe("Job Generation", func() {
@@ -137,29 +141,7 @@ var _ = Describe("Job Generation", func() {
 Expect(jobs[0].Spec.Template.Spec.InitContainers[0].Args).To(ContainElement(ContainSubstring("!upgrade.cattle.io/controller")))
 Expect(jobs[0].Spec.Template.Spec.InitContainers[0].Args).To(ContainElement(ContainSubstring("component notin (sonobuoy)")))
 })
-AfterEach(func() {
-if CurrentSpecReport().Failed() {
-podList, _ := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).List(context.Background(), metav1.ListOptions{})
-for _, pod := range podList.Items {
-containerNames := []string{}
-for _, container := range pod.Spec.InitContainers {
-containerNames = append(containerNames, container.Name)
-}
-for _, container := range pod.Spec.Containers {
-containerNames = append(containerNames, container.Name)
-}
-for _, container := range containerNames {
-reportName := fmt.Sprintf("podlogs-%s-%s", pod.Name, container)
-logs := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).GetLogs(pod.Name, &v1.PodLogOptions{Container: container})
-if logStreamer, err := logs.Stream(context.Background()); err == nil {
-if podLogs, err := io.ReadAll(logStreamer); err == nil {
-AddReportEntry(reportName, string(podLogs))
-}
-}
-}
-}
-}
-})
+AfterEach(CollectLogsOnFailure(e2e))
 })
 
 When("fails because of invalid time window", func() {
@@ -206,29 +188,7 @@ var _ = Describe("Job Generation", func() {
 Expect(jobs[0].Status.Active).To(BeNumerically("==", 0))
 Expect(jobs[0].Status.Failed).To(BeNumerically("==", 0))
 })
-AfterEach(func() {
-if CurrentSpecReport().Failed() {
-podList, _ := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).List(context.Background(), metav1.ListOptions{})
-for _, pod := range podList.Items {
-containerNames := []string{}
-for _, container := range pod.Spec.InitContainers {
-containerNames = append(containerNames, container.Name)
-}
-for _, container := range pod.Spec.Containers {
-containerNames = append(containerNames, container.Name)
-}
-for _, container := range containerNames {
-reportName := fmt.Sprintf("podlogs-%s-%s", pod.Name, container)
-logs := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).GetLogs(pod.Name, &v1.PodLogOptions{Container: container})
-if logStreamer, err := logs.Stream(context.Background()); err == nil {
-if podLogs, err := io.ReadAll(logStreamer); err == nil {
-AddReportEntry(reportName, string(podLogs))
-}
-}
-}
-}
-}
-})
+AfterEach(CollectLogsOnFailure(e2e))
 })
 
 When("fails because of invalid post complete delay", func() {
@@ -275,32 +235,10 @@ var _ = Describe("Job Generation", func() {
 Expect(jobs[0].Status.Active).To(BeNumerically("==", 0))
 Expect(jobs[0].Status.Failed).To(BeNumerically("==", 0))
 })
-AfterEach(func() {
-if CurrentSpecReport().Failed() {
-podList, _ := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).List(context.Background(), metav1.ListOptions{})
-for _, pod := range podList.Items {
-containerNames := []string{}
-for _, container := range pod.Spec.InitContainers {
-containerNames = append(containerNames, container.Name)
-}
-for _, container := range pod.Spec.Containers {
-containerNames = append(containerNames, container.Name)
-}
-for _, container := range containerNames {
-reportName := fmt.Sprintf("podlogs-%s-%s", pod.Name, container)
-logs := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).GetLogs(pod.Name, &v1.PodLogOptions{Container: container})
-if logStreamer, err := logs.Stream(context.Background()); err == nil {
-if podLogs, err := io.ReadAll(logStreamer); err == nil {
-AddReportEntry(reportName, string(podLogs))
-}
-}
-}
-}
-}
-})
+AfterEach(CollectLogsOnFailure(e2e))
 })
 
-When("updated secret should not change hash", func() {
+When("updated secret does not change hash", func() {
 var (
 err error
 plan *upgradeapiv1.Plan
@@ -347,5 +285,82 @@ var _ = Describe("Job Generation", func() {
 It("hash should be equal", func() {
 Expect(plan.Status.LatestHash).Should(Equal(hash))
 })
+AfterEach(CollectLogsOnFailure(e2e))
+})
+
+When("job failure message is reflected in plan status condition", func() {
+var (
+err error
+plan *upgradeapiv1.Plan
+jobs []batchv1.Job
+)
+BeforeEach(func() {
+plan = e2e.NewPlan("job-deadline-", "library/alpine:3.18", []string{"sh", "-c"}, "sleep 3600")
+plan.Spec.JobActiveDeadlineSecs = pointer.Int64(15)
+plan.Spec.Version = "latest"
+plan.Spec.Concurrency = 1
+plan.Spec.ServiceAccountName = e2e.Namespace.Name
+plan.Spec.NodeSelector = &metav1.LabelSelector{
+MatchExpressions: []metav1.LabelSelectorRequirement{{
+Key: "node-role.kubernetes.io/control-plane",
+Operator: metav1.LabelSelectorOpDoesNotExist,
+}},
+}
+plan, err = e2e.CreatePlan(plan)
+Expect(err).ToNot(HaveOccurred())
+
+plan, err = e2e.WaitForPlanCondition(plan.Name, upgradeapiv1.PlanLatestResolved, 30*time.Second)
+Expect(err).ToNot(HaveOccurred())
+})
+It("message should contain deadline reason and message", func() {
+jobs, err = e2e.WaitForPlanJobs(plan, 1, 120*time.Second)
+Expect(err).ToNot(HaveOccurred())
+Expect(jobs).To(HaveLen(1))
+Expect(jobs[0].Status.Succeeded).To(BeNumerically("==", 0))
+Expect(jobs[0].Status.Active).To(BeNumerically("==", 0))
+Expect(jobs[0].Status.Failed).To(BeNumerically(">=", 1))
+
+Eventually(e2e.GetPlan).
+WithArguments(plan.Name, metav1.GetOptions{}).
+WithTimeout(45 * time.Second).
+Should(SatisfyAll(
+WithTransform(upgradeapiv1.PlanComplete.IsTrue, BeFalse()),
+WithTransform(upgradeapiv1.PlanComplete.GetReason, Equal("JobFailed")),
+WithTransform(upgradeapiv1.PlanComplete.GetMessage, ContainSubstring("DeadlineExceeded: Job was active longer than specified deadline")),
+))
+})
+AfterEach(CollectLogsOnFailure(e2e))
 })
 })
+
+func CollectLogsOnFailure(e2e *framework.Client) func() {
+return func() {
+if CurrentSpecReport().Failed() {
+planList, _ := e2e.UpgradeClientSet.UpgradeV1().Plans(e2e.Namespace.Name).List(context.Background(), metav1.ListOptions{})
+AddReportEntry("plans", format.Object(planList, 0))
+
+jobList, _ := e2e.ClientSet.BatchV1().Jobs(e2e.Namespace.Name).List(context.Background(), metav1.ListOptions{})
+AddReportEntry("jobs", format.Object(jobList, 0))
+
+podList, _ := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).List(context.Background(), metav1.ListOptions{})
+for _, pod := range podList.Items {
+containerNames := []string{}
+for _, container := range pod.Spec.InitContainers {
+containerNames = append(containerNames, container.Name)
+}
+for _, container := range pod.Spec.Containers {
+containerNames = append(containerNames, container.Name)
+}
+for _, container := range containerNames {
+reportName := fmt.Sprintf("podlogs-%s-%s", pod.Name, container)
+logs := e2e.ClientSet.CoreV1().Pods(e2e.Namespace.Name).GetLogs(pod.Name, &v1.PodLogOptions{Container: container})
+if logStreamer, err := logs.Stream(context.Background()); err == nil {
+if podLogs, err := io.ReadAll(logStreamer); err == nil {
+AddReportEntry(reportName, string(podLogs))
+}
+}
+}
+}
+}
+}
+}

e2e/suite/plan_create_test.go

@@ -22,5 +22,6 @@ var _ = Describe("Plan Creation", func() {
 It("should return an error if upgrade in nil", func() {
 Expect(err).To(HaveOccurred())
 })
+AfterEach(CollectLogsOnFailure(e2e))
 })
 })

e2e/suite/plan_resolve_test.go

@@ -33,6 +33,7 @@ var _ = Describe("Plan Resolution", func() {
 Expect(plan.Status.LatestVersion).To(BeEmpty())
 Expect(plan.Status.LatestHash).To(BeEmpty())
 })
+AfterEach(CollectLogsOnFailure(e2e))
 })
 
 When("has version", func() {
@@ -56,6 +57,7 @@ var _ = Describe("Plan Resolution", func() {
 Expect(plan.Status.LatestVersion).To(Equal(plan.Spec.Version))
 Expect(plan.Status.LatestHash).ToNot(BeEmpty())
 })
+AfterEach(CollectLogsOnFailure(e2e))
 })
 
 When("has version with semver+metadata", func() {
@@ -82,6 +84,7 @@ var _ = Describe("Plan Resolution", func() {
 It("should munge the semver", func() {
 Expect(plan.Status.LatestVersion).ToNot(ContainSubstring(`+`))
 })
+AfterEach(CollectLogsOnFailure(e2e))
 })
 
 When("has channel", func() {
@@ -114,6 +117,7 @@ var _ = Describe("Plan Resolution", func() {
 Expect(plan.Status.LatestVersion).To(Equal(channelTag))
 Expect(plan.Status.LatestHash).ToNot(BeEmpty())
 })
+AfterEach(CollectLogsOnFailure(e2e))
 })
 
 When("has channel with semver+metadata", func() {
@@ -148,5 +152,6 @@ var _ = Describe("Plan Resolution", func() {
 It("should munge the semver", func() {
 Expect(plan.Status.LatestVersion).ToNot(ContainSubstring(`+`))
 })
+AfterEach(CollectLogsOnFailure(e2e))
 })
 })

pkg/upgrade/handle_batch.go

@@ -2,18 +2,20 @@ package upgrade
 
 import (
 "context"
+"errors"
 "fmt"
 "sort"
 "strconv"
 "time"
 
 upgradeapi "github.com/rancher/system-upgrade-controller/pkg/apis/upgrade.cattle.io"
+upgradeapiv1 "github.com/rancher/system-upgrade-controller/pkg/apis/upgrade.cattle.io/v1"
 upgradejob "github.com/rancher/system-upgrade-controller/pkg/upgrade/job"
 batchctlv1 "github.com/rancher/wrangler/v3/pkg/generated/controllers/batch/v1"
 "github.com/sirupsen/logrus"
 batchv1 "k8s.io/api/batch/v1"
 corev1 "k8s.io/api/core/v1"
-"k8s.io/apimachinery/pkg/api/errors"
+apierrors "k8s.io/apimachinery/pkg/api/errors"
 metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 "k8s.io/apimachinery/pkg/labels"
 )
@@ -50,7 +52,7 @@ func (ctl *Controller) handleJobs(ctx context.Context) error {
 // get the plan being applied
 plan, err := plans.Get(obj.Namespace, planName, metav1.GetOptions{})
 switch {
-case errors.IsNotFound(err):
+case apierrors.IsNotFound(err):
 // plan is gone, delete
 return obj, deleteJob(jobs, obj, metav1.DeletePropagationBackground)
 case err != nil:
@@ -73,7 +75,7 @@ func (ctl *Controller) handleJobs(ctx context.Context) error {
 // get the node that the plan is being applied to
 node, err := nodes.Cache().Get(nodeName)
 switch {
-case errors.IsNotFound(err):
+case apierrors.IsNotFound(err):
 // node is gone, delete
 return obj, deleteJob(jobs, obj, metav1.DeletePropagationBackground)
 case err != nil:
@@ -85,7 +87,12 @@ func (ctl *Controller) handleJobs(ctx context.Context) error {
 if failedTime.IsZero() {
 return obj, fmt.Errorf("condition %q missing field %q", upgradejob.ConditionFailed, "LastTransitionTime")
 }
-ctl.recorder.Eventf(plan, corev1.EventTypeWarning, "JobFailed", "Job failed on Node %s", node.Name)
+message := fmt.Sprintf("Job failed on Node %s: %s: %s", nodeName, upgradejob.ConditionFailed.GetReason(obj), upgradejob.ConditionFailed.GetMessage(obj))
+ctl.recorder.Eventf(plan, corev1.EventTypeWarning, "JobFailed", message)
+upgradeapiv1.PlanComplete.SetError(plan, "JobFailed", errors.New(message))
+if plan, err = plans.UpdateStatus(plan); err != nil {
+return obj, err
+}
 return obj, enqueueOrDelete(jobs, obj, failedTime)
 }
 // if the job has completed tag the node then enqueue-or-delete depending on the TTL window