-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement alerting using webhook (close #12)
- Loading branch information
Showing
16 changed files
with
595 additions
and
278 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
63 changes: 63 additions & 0 deletions
63
modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Alert.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| package com.snowplowanalytics.snowplow.snowflake | ||
|
|
||
| import cats.Show | ||
| import cats.implicits.showInterpolator | ||
| import com.snowplowanalytics.iglu.core.circe.implicits.igluNormalizeDataJson | ||
| import com.snowplowanalytics.iglu.core.{SchemaKey, SchemaVer, SelfDescribingData} | ||
| import com.snowplowanalytics.snowplow.runtime.AppInfo | ||
| import io.circe.Json | ||
| import io.circe.syntax.EncoderOps | ||
|
|
||
| import java.sql.SQLException | ||
|
|
||
| sealed trait Alert | ||
| object Alert { | ||
|
|
||
| /** Restrict the length of an alert message to be compliant with alert iglu schema */ | ||
| private val MaxAlertPayloadLength = 4096 | ||
|
|
||
| final case class FailedToCreateEventsTable(cause: Throwable) extends Alert | ||
| final case class FailedToAddColumns(columns: List[String], cause: Throwable) extends Alert | ||
| final case class FailedToOpenSnowflakeChannel(cause: Throwable) extends Alert | ||
|
|
||
| def toSelfDescribingJson( | ||
| alert: Alert, | ||
| appInfo: AppInfo, | ||
| tags: Map[String, String] | ||
| ): Json = | ||
| SelfDescribingData( | ||
| schema = SchemaKey("com.snowplowanalytics.monitoring.loader", "alert", "jsonschema", SchemaVer.Full(1, 0, 0)), | ||
| data = Json.obj( | ||
| "application" -> s"${appInfo.name}-${appInfo.version}".asJson, | ||
| "message" -> getMessage(alert).asJson, | ||
| "tags" -> tags.asJson | ||
| ) | ||
| ).normalize | ||
|
|
||
| private def getMessage(alert: Alert): String = { | ||
| val full = alert match { | ||
| case FailedToCreateEventsTable(cause) => show"Failed to create events table: $cause" | ||
| case FailedToAddColumns(columns, cause) => show"Failed to add columns: ${columns.mkString("[", ",", "]")}. Cause: $cause" | ||
| case FailedToOpenSnowflakeChannel(cause) => show"Failed to open Snowflake channel: $cause" | ||
| } | ||
|
|
||
| full.take(MaxAlertPayloadLength) | ||
| } | ||
|
|
||
| private implicit def throwableShow: Show[Throwable] = { | ||
| def go(acc: List[String], next: Throwable): String = { | ||
| val nextMessage = next match { | ||
| case t: SQLException => Some(s"${t.getMessage} = SqlState: ${t.getSQLState}") | ||
| case t => Option(t.getMessage) | ||
| } | ||
| val msgs = nextMessage.filterNot(msg => acc.headOption.contains(msg)) ++: acc | ||
|
|
||
| Option(next.getCause) match { | ||
| case Some(cause) => go(msgs, cause) | ||
| case None => msgs.reverse.mkString(": ") | ||
| } | ||
| } | ||
|
|
||
| Show.show(go(Nil, _)) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Monitoring.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package com.snowplowanalytics.snowplow.snowflake | ||
|
|
||
| import cats.effect.{Resource, Sync} | ||
| import cats.implicits._ | ||
| import com.snowplowanalytics.snowplow.runtime.AppInfo | ||
| import org.http4s.circe.jsonEncoder | ||
| import org.http4s.client.Client | ||
| import org.http4s.{EntityDecoder, Method, Request} | ||
| import org.typelevel.log4cats.Logger | ||
| import org.typelevel.log4cats.slf4j.Slf4jLogger | ||
|
|
||
| trait Monitoring[F[_]] { | ||
| def alert(message: Alert): F[Unit] | ||
| } | ||
|
|
||
| object Monitoring { | ||
|
|
||
| private implicit def logger[F[_]: Sync] = Slf4jLogger.getLogger[F] | ||
|
|
||
| def create[F[_]: Sync]( | ||
| config: Option[Config.Webhook], | ||
| appInfo: AppInfo, | ||
| httpClient: Client[F] | ||
| )(implicit E: EntityDecoder[F, String] | ||
| ): Resource[F, Monitoring[F]] = Resource.pure { | ||
| new Monitoring[F] { | ||
|
|
||
| override def alert(message: Alert): F[Unit] = | ||
| config match { | ||
| case Some(webhookConfig) => | ||
| val request = buildHttpRequest(webhookConfig, message) | ||
| executeHttpRequest(webhookConfig, httpClient, request) | ||
| case None => | ||
| Logger[F].debug("Webhook monitoring is not configured, skipping alert") | ||
| } | ||
|
|
||
| def buildHttpRequest(webhookConfig: Config.Webhook, alert: Alert): Request[F] = | ||
| Request[F](Method.POST, webhookConfig.endpoint) | ||
| .withEntity(Alert.toSelfDescribingJson(alert, appInfo, webhookConfig.tags)) | ||
|
|
||
| def executeHttpRequest( | ||
| webhookConfig: Config.Webhook, | ||
| httpClient: Client[F], | ||
| request: Request[F] | ||
| ): F[Unit] = | ||
| httpClient | ||
| .run(request) | ||
| .use { response => | ||
| if (response.status.isSuccess) Sync[F].unit | ||
| else { | ||
| response.as[String].flatMap(body => Logger[F].error(s"Webhook ${webhookConfig.endpoint} returned non-2xx response:\n$body")) | ||
| } | ||
| } | ||
| .handleErrorWith { e => | ||
| Logger[F].error(e)(s"Webhook ${webhookConfig.endpoint} resulted in exception without a response") | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
38 changes: 38 additions & 0 deletions
38
modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Sentry.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| package com.snowplowanalytics.snowplow.snowflake | ||
|
|
||
| import cats.effect.{Resource, Sync} | ||
| import cats.implicits.catsSyntaxApplyOps | ||
| import com.snowplowanalytics.snowplow.runtime.AppInfo | ||
| import io.sentry.{Sentry => JSentry, SentryOptions} | ||
|
|
||
| object Sentry { | ||
|
|
||
| def capturingAnyException[F[_]: Sync](appInfo: AppInfo, config: Option[Config.Sentry]): Resource[F, Unit] = | ||
| config match { | ||
| case Some(sentryConfig) => | ||
| initSentry(appInfo, sentryConfig) | ||
| case None => | ||
| Resource.unit[F] | ||
| } | ||
|
|
||
| private def initSentry[F[_]: Sync](appInfo: AppInfo, sentryConfig: Config.Sentry): Resource[F, Unit] = { | ||
| val acquire = Sync[F].delay(JSentry.init(createSentryOptions(appInfo, sentryConfig))) | ||
| val release = Sync[F].delay(JSentry.close()) | ||
|
|
||
| Resource.makeCase(acquire) { | ||
| case (_, Resource.ExitCase.Errored(e)) => Sync[F].delay(JSentry.captureException(e)) *> release | ||
| case _ => release | ||
|
|
||
| } | ||
| } | ||
|
|
||
| private def createSentryOptions(appInfo: AppInfo, sentryConfig: Config.Sentry): SentryOptions = { | ||
| val options = new SentryOptions | ||
| options.setDsn(sentryConfig.dsn) | ||
| options.setRelease(appInfo.version) | ||
| sentryConfig.tags.foreach { case (k, v) => | ||
| options.setTag(k, v) | ||
| } | ||
| options | ||
| } | ||
| } |
Oops, something went wrong.