#423 Add pipeline failures to the pipeline info object.

AbsaOSS · Jun 20, 2024 · 3bf4ec1 · 3bf4ec1
1 parent 7732a4c
commit 3bf4ec1
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 7 deletions.
diff --git a/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala b/pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineInfo.scala
@@ -16,7 +16,7 @@
 
 package za.co.absa.pramen.api
 
-import za.co.absa.pramen.api.status.RuntimeInfo
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, RuntimeInfo}
 
 import java.time.Instant
 
@@ -27,5 +27,6 @@ case class PipelineInfo(
                          startedAt: Instant,
                          finishedAt: Option[Instant],
                          sparkApplicationId: Option[String],
-                         failureException: Option[Throwable]
+                         failureException: Option[Throwable],
+                         pipelineNotificationFailures: Seq[PipelineNotificationFailure]
                        )
diff --git a/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala b/pramen/core/src/main/scala/za/co/absa/pramen/core/state/PipelineStateImpl.scala
@@ -106,7 +106,8 @@ class PipelineStateImpl(implicit conf: Config, notificationBuilder: Notification
         startedInstant,
         finishedInstant,
         sparkAppId,
-        appException
+        appException,
+        pipelineNotificationFailures.toSeq
       ),
       isFinished,
       exitedNormally,

diff --git a/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala b/pramen/core/src/test/scala/za/co/absa/pramen/core/mocks/PipelineInfoFactory.scala
@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core.mocks
 
 import za.co.absa.pramen.api.PipelineInfo
-import za.co.absa.pramen.api.status.RuntimeInfo
+import za.co.absa.pramen.api.status.{PipelineNotificationFailure, RuntimeInfo}
 
 import java.time.Instant
 
@@ -28,7 +28,8 @@ object PipelineInfoFactory {
                            startedAt: Instant = Instant.ofEpochSecond(1718609409),
                            finishedAt: Option[Instant] = None,
                            sparkApplicationId: Option[String] = Some("testid-12345"),
-                           failureException: Option[Throwable] = None): PipelineInfo = {
-    PipelineInfo(pipelineName, environment, runtimeInfo, startedAt, finishedAt, sparkApplicationId, failureException)
+                           failureException: Option[Throwable] = None,
+                           pipelineNotificationFailures: Seq[PipelineNotificationFailure] = Seq.empty): PipelineInfo = {
+    PipelineInfo(pipelineName, environment, runtimeInfo, startedAt, finishedAt, sparkApplicationId, failureException, pipelineNotificationFailures)
   }
 }
diff --git a/...test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala b/...test/scala/za/co/absa/pramen/extras/notification/EcsPipelineNotificationTargetSuite.scala
@@ -55,7 +55,7 @@ class EcsPipelineNotificationTargetSuite extends AnyWordSpec {
       val task3 = TestPrototypes.taskNotification.copy(jobName = "Job 3", outputTable = metaTableDef3)
 
       notificationTarget.sendNotification(
-        PipelineInfo("Dummy", "DEV", RuntimeInfo(), Instant.now, None, None, None),
+        PipelineInfo("Dummy", "DEV", RuntimeInfo(), Instant.now, None, None, None, Seq.empty),
         Seq(task1, task2, task3),
         CustomNotification(Seq.empty, Seq.empty)
       )