#372 Add global notification targets

README.md

@@ -2624,7 +2624,32 @@ pramen.operations = [
 ```
 </details>
 
-## Notifications
+## Pipeline Notifications
+Custom pipeline notification targets allow execution arbitrary actions after the pipeline is finished. Usually, it is 
+used to send custom notifications to external systems. A pipeline notification target can be created by implementing
+`PipelineNotificationTarget` interface:
+```scala
+package com.example
+
+import com.typesafe.config.Config
+import za.co.absa.pramen.api.PipelineNotificationTarget
+
+class MyPipelineNotificationTarget(conf: Config) extends PipelineNotificationTarget {
+  override def sendNotification(pipelineStarted: Instant,
+                                applicationId: Option[String],
+                                appException: Option[Throwable],
+                                tasksCompleted: Seq[TaskNotification]): Unit = ???
+
+  override def config: Config = conf
+}
+```
+
+Pipeline notification targets can be registered in the workflow configuration:
+```hocon
+pramen.pipeline.notification.targets = [ "com.example.MyPipelineNotificationTarget" ]
+```
+
+## Job Notifications
 If you need to react on a completion event of any job, you can do it using notification targets. A notification target 
 is a component that you can implement and register for any operation or table. The notification target will be called 
 when the operation or job completes. Usually it is used to send an event to trigger actions outside the Pramen pipeline.

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/FieldChange.scala → pramen/api/src/main/scala/za/co/absa/pramen/api/FieldChange.scala

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.notify.pipeline
+package za.co.absa.pramen.api
 
 sealed trait FieldChange
 

pramen/api/src/main/scala/za/co/absa/pramen/api/PipelineNotificationTarget.scala

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.api
+
+import java.time.Instant
+
+trait PipelineNotificationTarget extends ExternalChannel {
+  /** Sends a notification after completion of the pipeline. */
+  def sendNotification(pipelineStarted: Instant,
+                       applicationId: Option[String],
+                       appException: Option[Throwable],
+                       tasksCompleted: Seq[TaskNotification]): Unit
+}

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/SchemaDifference.scala → pramen/api/src/main/scala/za/co/absa/pramen/api/SchemaDifference.scala

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package za.co.absa.pramen.core.notify.pipeline
+package za.co.absa.pramen.api
 
 import java.time.LocalDate
 

pramen/api/src/main/scala/za/co/absa/pramen/api/TaskNotification.scala

@@ -20,10 +20,14 @@ import java.time.{Instant, LocalDate}
 
 case class TaskNotification(
                              tableName: String,
-                             infoDate: LocalDate,
-                             started: Instant,
-                             finished: Instant,
+                             infoDate: Option[LocalDate],
+                             started: Option[Instant],
+                             finished: Option[Instant],
                              status: TaskStatus,
                              applicationId: String,
+                             isTransient: Boolean,
+                             isRawFilesJob: Boolean,
+                             schemaChanges: Seq[SchemaDifference],
+                             dependencyWarningTables: Seq[String],
                              options: Map[String, String]
                            )

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/PipelineNotificationTargetFactory.scala

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.core.notify
+
+import com.typesafe.config.Config
+import za.co.absa.pramen.api.PipelineNotificationTarget
+import za.co.absa.pramen.core.utils.ClassLoaderUtils
+
+import scala.collection.JavaConverters._
+
+object PipelineNotificationTargetFactory {
+  val PIPELINE_NOTIFICATION_FACTORIES_KEY = "pramen.pipeline.notification.targets"
+
+  def fromConfig(conf: Config): Seq[PipelineNotificationTarget] = {
+    if (conf.hasPath(PIPELINE_NOTIFICATION_FACTORIES_KEY)) {
+      val factories = conf.getStringList(PIPELINE_NOTIFICATION_FACTORIES_KEY).asScala
+      factories.map(clazz => createPipelineNotificationTarget(clazz, conf)).toSeq
+    } else {
+      Seq.empty
+    }
+  }
+
+  private[core] def createPipelineNotificationTarget(clazz: String, appConfig: Config): PipelineNotificationTarget = {
+    ClassLoaderUtils.loadConfigurableClass[PipelineNotificationTarget](clazz, appConfig)
+  }
+
+}

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotification.scala

@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core.notify.pipeline
 
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
-import za.co.absa.pramen.core.runner.task.TaskResult
+import za.co.absa.pramen.core.runner.task.{PipelineNotificationFailure, TaskResult}
 
 import java.time.Instant
 
@@ -29,6 +29,7 @@ case class PipelineNotification(
                                  started: Instant,
                                  finished: Instant,
                                  tasksCompleted: List[TaskResult],
+                                 pipelineNotificationFailures: List[PipelineNotificationFailure],
                                  customEntries: List[NotificationEntry],
                                  customSignature: List[TextElement]
                                )

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilder.scala

@@ -17,7 +17,7 @@
 package za.co.absa.pramen.core.notify.pipeline
 
 import za.co.absa.pramen.api.notification.{NotificationEntry, TextElement}
-import za.co.absa.pramen.core.runner.task.TaskResult
+import za.co.absa.pramen.core.runner.task.{PipelineNotificationFailure, TaskResult}
 
 import java.time.Instant
 
@@ -40,6 +40,8 @@ trait PipelineNotificationBuilder {
 
   def addCompletedTask(completedTask: TaskResult): Unit
 
+  def addPipelineNotificationFailure(failure: PipelineNotificationFailure): Unit
+
   def addCustomEntries(entries: Seq[NotificationEntry]): Unit
 
   def addSignature(signature: TextElement*): Unit

pramen/core/src/main/scala/za/co/absa/pramen/core/notify/pipeline/PipelineNotificationBuilderHtml.scala

@@ -18,14 +18,15 @@ package za.co.absa.pramen.core.notify.pipeline
 
 import com.typesafe.config.Config
 import org.slf4j.LoggerFactory
+import za.co.absa.pramen.api.{FieldChange, SchemaDifference}
 import za.co.absa.pramen.api.notification._
 import za.co.absa.pramen.core.config.Keys.TIMEZONE
 import za.co.absa.pramen.core.exceptions.{CmdFailedException, ProcessFailedException}
 import za.co.absa.pramen.core.notify.message._
 import za.co.absa.pramen.core.notify.pipeline.PipelineNotificationBuilderHtml.{MIN_MEGABYTES, MIN_RPS_JOB_DURATION_SECONDS, MIN_RPS_RECORDS}
 import za.co.absa.pramen.core.pipeline.TaskRunReason
 import za.co.absa.pramen.core.runner.task.RunStatus._
-import za.co.absa.pramen.core.runner.task.{NotificationFailure, RunStatus, TaskResult}
+import za.co.absa.pramen.core.runner.task.{NotificationFailure, PipelineNotificationFailure, RunStatus, TaskResult}
 import za.co.absa.pramen.core.utils.JvmUtils.getShortExceptionDescription
 import za.co.absa.pramen.core.utils.StringUtils.renderThrowable
 import za.co.absa.pramen.core.utils.{BuildPropertyUtils, ConfigUtils, StringUtils, TimeUtils}
@@ -63,6 +64,7 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
   var customSignature = Seq.empty[TextElement]
 
   val completedTasks = new ListBuffer[TaskResult]
+  val pipelineNotificationFailures = new ListBuffer[PipelineNotificationFailure]
   val customEntries = new ListBuffer[NotificationEntry]
 
   override def addFailureException(ex: Throwable): Unit = {
@@ -103,6 +105,10 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
     completedTasks += completedTask
   }
 
+  override def addPipelineNotificationFailure(failure: PipelineNotificationFailure): Unit = {
+    pipelineNotificationFailures += failure
+  }
+
   override def addCustomEntries(entries: Seq[NotificationEntry]): Unit = customEntries ++= entries
 
   override def addSignature(signature: TextElement*): Unit = customSignature = signature
@@ -145,6 +151,8 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
         }
       })
 
+    renderPipelineNotificationFailures(builder)
+
     val notificationTargetErrors = completedTasks.flatMap(_.notificationTargetErrors)
 
     if (notificationTargetErrors.nonEmpty) {
@@ -253,8 +261,9 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
       val hasTaskWarnings = task.runStatus.isInstanceOf[RunStatus.Succeeded] && task.runStatus.asInstanceOf[RunStatus.Succeeded].warnings.nonEmpty
       val hasSkippedWithWarnings = task.runStatus.isInstanceOf[RunStatus.Skipped] && task.runStatus.asInstanceOf[RunStatus.Skipped].isWarning
       val hasSchemaChanges = task.schemaChanges.nonEmpty
+      val hasPipelineNotificationFailures = pipelineNotificationFailures.nonEmpty
 
-      hasDependencyWarnings || hasNotificationErrors || hasTaskWarnings || hasSkippedWithWarnings || hasSchemaChanges
+      hasDependencyWarnings || hasNotificationErrors || hasTaskWarnings || hasSkippedWithWarnings || hasSchemaChanges || hasPipelineNotificationFailures
     }
   }
 
@@ -412,6 +421,16 @@ class PipelineNotificationBuilderHtml(implicit conf: Config) extends PipelineNot
     builder.withTable(tableBuilder)
   }
 
+  private[core] def renderPipelineNotificationFailures(builder: MessageBuilderHtml): MessageBuilder = {
+    pipelineNotificationFailures.foreach { failure =>
+      val notificationErrorsParagraph = ParagraphBuilder()
+        .withText(s"Failed to send pipeline notification via '${failure.notificationTarget}': ", Style.Exception)
+      builder.withParagraph(notificationErrorsParagraph)
+      renderException(builder, failure.ex)
+    }
+    builder
+  }
+
   private[core] def renderNotificationTargetErrors(builder: MessageBuilderHtml, notificationTargetErrors: ListBuffer[NotificationFailure]): MessageBuilder = {
     val tableBuilder = new TableBuilderHtml
 

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/PipelineNotificationFailure.scala

@@ -0,0 +1,22 @@
+/*
+ * Copyright 2022 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package za.co.absa.pramen.core.runner.task
+
+case class PipelineNotificationFailure(
+                                        notificationTarget: String,
+                                        ex: Throwable
+                                      )

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskResult.scala

@@ -16,7 +16,7 @@
 
 package za.co.absa.pramen.core.runner.task
 
-import za.co.absa.pramen.core.notify.pipeline.SchemaDifference
+import za.co.absa.pramen.api.SchemaDifference
 import za.co.absa.pramen.core.pipeline.{DependencyWarning, Job}
 
 case class TaskResult(

pramen/core/src/main/scala/za/co/absa/pramen/core/runner/task/TaskRunnerBase.scala

@@ -20,7 +20,7 @@ import com.typesafe.config.Config
 import org.apache.spark.sql.DataFrame
 import org.apache.spark.sql.functions.lit
 import org.slf4j.LoggerFactory
-import za.co.absa.pramen.api.{DataFormat, Reason, TaskNotification}
+import za.co.absa.pramen.api.{DataFormat, Reason, SchemaDifference, TaskNotification}
 import za.co.absa.pramen.core.app.config.RuntimeConfig
 import za.co.absa.pramen.core.bookkeeper.Bookkeeper
 import za.co.absa.pramen.core.exceptions.{FatalErrorWrapper, ReasonException}
@@ -30,7 +30,6 @@ import za.co.absa.pramen.core.lock.TokenLockFactory
 import za.co.absa.pramen.core.metastore.MetaTableStats
 import za.co.absa.pramen.core.metastore.model.MetaTable
 import za.co.absa.pramen.core.notify.NotificationTargetManager
-import za.co.absa.pramen.core.notify.pipeline.SchemaDifference
 import za.co.absa.pramen.core.pipeline.JobPreRunStatus._
 import za.co.absa.pramen.core.pipeline._
 import za.co.absa.pramen.core.state.PipelineState
@@ -394,11 +393,15 @@ abstract class TaskRunnerBase(conf: Config,
       NotificationTargetManager.runStatusToTaskStatus(result.runStatus).foreach { taskStatus =>
         val notification = TaskNotification(
           task.job.outputTable.name,
-          task.infoDate,
-          result.runInfo.get.started,
-          result.runInfo.get.finished,
+          Option(task.infoDate),
+          result.runInfo.map(_.started),
+          result.runInfo.map(_.finished),
           taskStatus,
           result.applicationId,
+          result.isTransient,
+          result.isRawFilesJob,
+          result.schemaChanges,
+          result.dependencyWarnings.map(_.table),
           notificationTarget.options
         )