Add pager duty integration

build.sbt

@@ -36,6 +36,9 @@ libraryDependencies ++= Seq(
   "org.scalatest" %% "scalatest" % scalaTestVersion % Test,
   "com.typesafe.akka" %% "akka-stream-testkit" % akkaVersion % Test,
   "com.typesafe.akka" %% "akka-http-testkit" % akkaHttpVersion % Test,
+
+  // Alert mangement dependencies
+  "com.github.dikhan" % "pagerduty-client" % "3.0.9",
 )
 
 dockerBaseImage := "openjdk:jre"

src/main/scala/alerts/AlertActor.scala

@@ -0,0 +1,41 @@
+package me.amanj.greenish.alerts
+
+import me.amanj.greenish.models.{JobStatus, AlertLevel, Normal, Critical}
+import java.io.File
+import me.amanj.greenish.checker.{MaybeAlert, debugFile}
+import scala.collection.{mutable => m}
+import akka.actor.{Actor, ActorRef, ActorLogging}
+import scala.io.Source
+import java.util.UUID
+
+class AlertActor(clients: Map[AlertLevel, AlertClient], outputDir: File)
+  extends Actor with ActorLogging {
+
+  private[this] val instanceId = UUID.randomUUID()
+  private[this] val jobAlerts: m.Map[(Int, Int), AlertLevel] = m.Map.empty
+
+  override def receive: Actor.Receive = {
+    case MaybeAlert(groupId, groupName, jobId, jobName, alertLevel) =>
+      val mapKey = (groupId, jobId)
+      val maybeAlerted = jobAlerts.get(mapKey)
+      val shouldAlert = maybeAlerted.map(_ == alertLevel).getOrElse(true)
+      if(shouldAlert) {
+        try {
+          val stdout = Source.fromFile(debugFile(outputDir, groupId,
+            jobId)).getLines.toSeq
+          val info = JobInfo(s"group-$groupId-job-${jobId}-$instanceId",
+            groupName, jobName, stdout)
+
+          clients.get(alertLevel).foreach { client =>
+            alertLevel match {
+              case Critical | Normal => client.resolve(info)
+              case _                 => client.alert(info, alertLevel)
+            }
+          }
+          jobAlerts += ((mapKey, alertLevel))
+        } catch {
+          case e: Exception => log.error(e.getMessage)
+        }
+      }
+  }
+}

src/main/scala/alerts/AlertClient.scala

@@ -0,0 +1,13 @@
+package me.amanj.greenish.alerts
+
+import com.typesafe.config.Config
+import me.amanj.greenish.models.AlertLevel
+
+case class JobInfo(id: String,
+  groupName: String, jobName: String, stdout: Seq[String])
+
+abstract class AlertClient(config: Config) {
+  def alert(job: JobInfo, level: AlertLevel): Unit
+
+  def resolve(job: JobInfo): Unit
+}

src/main/scala/alerts/PagerDutyClient.scala

@@ -0,0 +1,69 @@
+package me.amanj.greenish.alerts
+
+import com.typesafe.config.Config
+import me.amanj.greenish.models.{AlertLevel, Great, Normal, Warn, Critical}
+import com.github.dikhan.pagerduty.client.events.PagerDutyEventsClient
+import com.github.dikhan.pagerduty.client.events.domain.{Payload, Severity,
+  TriggerIncident, ResolveIncident}
+import java.time.OffsetDateTime
+
+class PagerDutyClient(config: Config) extends AlertClient(config) {
+  private[this] val eventApi = config.getString("event-api")
+
+  private[this] val proxyHost = if(config.hasPath("proxy-host")) {
+    Some(config.getString("proxy-host"))
+  } else None
+
+  private[this] val proxyPort = if(config.hasPath("proxy-port")) {
+    Some(config.getInt("proxy-port"))
+  } else None
+
+  private[this] val doRetries = if(config.hasPath("do-retries")) {
+    config.getBoolean("do-retries")
+  } else false
+
+  private[this] val routingKey = config.getString("routing-key")
+
+  private[this] val client = {
+    var tmp = new PagerDutyEventsClient.PagerDutyClientBuilder()
+      .withEventApi(eventApi)
+      .withDoRetries(doRetries)
+    tmp = proxyHost.map { host => tmp.withProxyHost(host) }.getOrElse(tmp)
+    tmp = proxyPort.map { port => tmp.withProxyPort(port) }.getOrElse(tmp)
+    tmp.build()
+  }
+
+  def alert(job: JobInfo, level: AlertLevel): Unit = {
+    val seveirtyLevel = level match {
+      case Great | Normal => Severity.INFO
+      case Warn => Severity.WARNING
+      case Critical => Severity.CRITICAL
+    }
+
+    val payload = Payload.Builder.newBuilder()
+      .setSummary(s"""|
+        |${job.groupName}-${job.jobName} has problems, details are below:
+        |
+        |${job.stdout.mkString("\n")}
+        |""".stripMargin)
+      .setSource(System.getenv("HOSTNAME"))
+      .setSeverity(seveirtyLevel)
+      .setTimestamp(OffsetDateTime.now())
+      .build()
+
+    val incident = TriggerIncident.TriggerIncidentBuilder
+      .newBuilder(routingKey, payload)
+      .setDedupKey(job.id)
+      .build()
+
+    client.trigger(incident)
+  }
+
+  def resolve(job: JobInfo): Unit = {
+    val resolve = ResolveIncident.ResolveIncidentBuilder
+      .newBuilder(routingKey, job.id)
+      .build();
+
+    client.resolve(resolve);
+  }
+}

src/main/scala/checker/Message.scala

@@ -1,7 +1,7 @@
 package me.amanj.greenish.checker
 
 import java.time.ZonedDateTime
-import me.amanj.greenish.models.{JobStatus, PeriodHealth}
+import me.amanj.greenish.models.{JobStatus, PeriodHealth, AlertLevel}
 
 sealed trait Message
 case class Refresh(now: () => ZonedDateTime) extends Message
@@ -20,3 +20,5 @@ case class BatchRun(cmd: String, periods: Seq[String],
   expireAt: Long) extends Message
 case class RunResult(periodHealth: Seq[PeriodHealth],
   groupId: Int, jobId: Int, clockCounter: Long) extends Message
+case class MaybeAlert(groupId: Int, groupName: String,
+  jobId: Int, jobName: String, alertLevel: AlertLevel) extends Message