SlackNotification . This fixes #8

Author: simplesteph

CHANGELOG.md

@@ -5,6 +5,10 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+
+
+## [0.1] - 08/03/2017
+- Kafka 1.0.1
 - Initial Release
 - Travis CI automation
 - Docker Hub release
@@ -14,4 +18,5 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 - Extract ACLs from Kafka. 
 - Tests including with Kafka running 
 - GitHub Enterprise Support
-- GitHub authentication Support
+- GitHub authentication Support
+- Slack Notification Support

src/main/resources/application.conf

@@ -58,6 +58,13 @@ notification {
 
   }
   slack {
-
+    // https://hooks.slack.com/services/etc/etc/etc
+    webhook = ${?NOTIFICATION_SLACK_WEBHOOK}
+    username = "Kafka Security Manager"
+    username = ${?NOTIFICATION_SLACK_USERNAME}
+    icon = "https://cdn.iconscout.com/public/images/icon/free/png-512/kafka-logo-brand-3f2f9aed5b9ae1e2-512x512.png"
+    icon = ${?NOTIFICATION_SLACK_ICON}
+    channel = "general"
+    channel = ${?NOTIFICATION_SLACK_CHANNEL}
   }
 }

src/main/scala/com/github/simplesteph/ksm/notification/ConsoleNotification.scala

@@ -21,13 +21,7 @@ case class ConsoleNotification() extends Notification {
   override def configure(config: Config): Unit = ()
 
   override def notifyErrors(errs: List[Try[Throwable]]): Unit = {
-    if (errs.nonEmpty) {
-      errs.foreach(e =>
-        e.get match {
-          case cPE: CsvParserException => log.error(s"${cPE.getLocalizedMessage} | Row: ${cPE.printRow()}")
-          case _ => log.error("error while parsing ACL source", e.get)
-        })
-    }
+    NotificationUtils.errorsToString(errs).foreach(println)
   }
 
   override protected def notifyOne(action: String, acls: Set[(Resource, Acl)]): Unit = {

src/main/scala/com/github/simplesteph/ksm/notification/Notification.scala

@@ -52,6 +52,6 @@ object Notification {
 
   // helper to pretty print a long ACL, as the toString provided by Kafka is not suitable here
   def printAcl(acl: Acl, resource: Resource): String = {
-    s"${acl.principal}, $resource, ${acl.operation}, ${acl.permissionType}, ${acl.host}"
+    s"${acl.principal}, $resource, Operation: ${acl.operation}, Permission: ${acl.permissionType}, Host: ${acl.host}"
   }
 }

src/main/scala/com/github/simplesteph/ksm/notification/NotificationUtils.scala

@@ -0,0 +1,17 @@
+package com.github.simplesteph.ksm.notification
+
+import com.github.simplesteph.ksm.parser.CsvParserException
+
+import scala.util.Try
+
+object NotificationUtils {
+
+  def errorsToString(errs: List[Try[Throwable]]): List[String] = {
+    errs.map(e =>
+      e.get match {
+        case cPE: CsvParserException => s"${cPE.getLocalizedMessage} | Row: ${cPE.printRow()}"
+        case _ => s"error while parsing ACL source: ${e.get}"
+      })
+  }
+
+}

src/main/scala/com/github/simplesteph/ksm/notification/SlackNotification.scala

@@ -1,6 +1,10 @@
 package com.github.simplesteph.ksm.notification
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.github.simplesteph.ksm.parser.CsvParserException
 import com.typesafe.config.Config
 import kafka.security.auth.{ Acl, Resource }
+import org.slf4j.LoggerFactory
+import skinny.http.HTTP
 
 import scala.util.Try
 
@@ -10,14 +14,77 @@ class SlackNotification extends Notification {
    */
   override val CONFIG_PREFIX: String = "slack"
 
+  private val log = LoggerFactory.getLogger(classOf[SlackNotification])
+
+  final val WEBHOOK_CONFIG = "webhook"
+  final val USERNAME_CONFIG = "username"
+  final val ICON_CONFIG = "icon"
+  final val CHANNEL_CONFIG = "channel"
+
+  val objectMapper: ObjectMapper = new ObjectMapper()
+  var webhook: String = _
+  var username: String = _
+  var icon: String = _
+  var channel: String = _
+
   /**
    * internal config definition for the module
    */
-  override def configure(config: Config): Unit = ???
+  override def configure(config: Config): Unit = {
+    webhook = config.getString(WEBHOOK_CONFIG)
+    username = config.getString(USERNAME_CONFIG)
+    icon = config.getString(ICON_CONFIG)
+    channel = config.getString(CHANNEL_CONFIG)
+  }
+
+  override def notifyOne(action: String, acls: Set[(Resource, Acl)]): Unit = {
+    if (acls.nonEmpty) {
+      val messages = acls.map {
+        case (resource, acl) =>
+          val message = Notification.printAcl(acl, resource)
+          s"$action $message"
+      }.toList
+
+      sendToSlack(messages)
+    }
+  }
+
+  def sendToSlack(messages: List[String], retries: Int = 5): Unit = {
+    if (retries > 0) {
+      messages.grouped(50).foreach(msgChunks => {
+        val text =
+          s"""```
+             |${msgChunks.mkString("\n")}
+             |```
+           """.stripMargin
+
+        val payload = objectMapper.createObjectNode()
+          .put("text", text)
+          .put("username", username)
+          .put("icon_url", icon)
+          .put("channel", channel)
+
+        val response = HTTP.post(webhook, payload.toString)
+
+        response.status match {
+          case 200 => ()
+          case _ =>
+            log.warn(response.asString)
+            if (retries > 1) log.warn("Retrying...")
+            Thread.sleep(300)
+            sendToSlack(msgChunks, retries - 1)
+        }
+      })
+    } else {
+      log.error("Can't send notification to Slack after retries")
+    }
+  }
 
-  override def notifyOne(action: String, acls: Set[(Resource, Acl)]): Unit = ???
+  override def notifyErrors(errs: List[Try[Throwable]]): Unit = {
+    sendToSlack(NotificationUtils.errorsToString(errs))
+  }
 
-  override def notifyErrors(errs: List[Try[Throwable]]): Unit = ???
+  override def close(): Unit = {
 
-  override def close(): Unit = ???
+  }
 }