-
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.
Don't crash app for Snowflake failures
- Loading branch information
Showing
6 changed files
with
126 additions
and
23 deletions.
There are no files selected for viewing
39 changes: 39 additions & 0 deletions
39
modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/AppHealth.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,39 @@ | ||
| package com.snowplowanalytics.snowplow.snowflake | ||
|
|
||
| import cats.implicits._ | ||
| import cats.{Functor, Monad, Monoid} | ||
| import com.snowplowanalytics.snowplow.runtime.HealthProbe | ||
| import com.snowplowanalytics.snowplow.runtime.HealthProbe.{Healthy, Unhealthy} | ||
| import com.snowplowanalytics.snowplow.snowflake.processing.SnowflakeHealth | ||
| import com.snowplowanalytics.snowplow.sources.SourceAndAck | ||
|
|
||
| object AppHealth { | ||
|
|
||
| def isHealthy[F[_]: Monad]( | ||
| config: Config.HealthProbe, | ||
| source: SourceAndAck[F], | ||
| snowflakeHealth: SnowflakeHealth[F] | ||
| ): F[HealthProbe.Status] = | ||
| List( | ||
| latencyHealth(config, source), | ||
| snowflakeHealth.state.get | ||
| ).sequence.map(_.combineAll) | ||
|
|
||
| private def latencyHealth[F[_]: Functor](config: Config.HealthProbe, source: SourceAndAck[F]): F[HealthProbe.Status] = | ||
| source.processingLatency.map { latency => | ||
| if (latency > config.unhealthyLatency) | ||
| Unhealthy(show"Processing latency is $latency") | ||
| else | ||
| Healthy | ||
| } | ||
|
|
||
| private val combineHealth: (HealthProbe.Status, HealthProbe.Status) => HealthProbe.Status = { | ||
| case (Healthy, Healthy) => Healthy | ||
| case (Healthy, unhealthy) => unhealthy | ||
| case (unhealthy, Healthy) => unhealthy | ||
| case (Unhealthy(first), Unhealthy(second)) => Unhealthy(reason = s"$first, $second") //TODO do we want to combine reasons like that...? | ||
| } | ||
|
|
||
| private implicit val healthMonoid: Monoid[HealthProbe.Status] = Monoid.instance(Healthy, combineHealth | ||
| ) | ||
| } |
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
45 changes: 45 additions & 0 deletions
45
...ain/scala/com.snowplowanalytics.snowplow.snowflake/processing/SnowflakeActionRunner.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,45 @@ | ||
| package com.snowplowanalytics.snowplow.snowflake.processing | ||
|
|
||
| import cats.effect.Sync | ||
| import cats.implicits._ | ||
| import net.snowflake.client.jdbc.SnowflakeSQLException | ||
| import org.typelevel.log4cats.Logger | ||
| import org.typelevel.log4cats.slf4j.Slf4jLogger | ||
| import retry._ | ||
| import retry.implicits.retrySyntaxError | ||
|
|
||
| import scala.concurrent.duration.DurationInt | ||
|
|
||
| trait SnowflakeActionRunner[F[_]] { | ||
| def run[A](action: F[A]): F[A] | ||
| } | ||
|
|
||
| object SnowflakeActionRunner { | ||
|
|
||
| private implicit def logger[F[_]: Sync] = Slf4jLogger.getLogger[F] | ||
|
|
||
| def retryingIndefinitely[F[_]: Sync: Sleep](snowflakeHealth: SnowflakeHealth[F]): SnowflakeActionRunner[F] = | ||
| new SnowflakeActionRunner[F] { | ||
| override def run[A](action: F[A]): F[A] = | ||
| retryUntilSuccessful(snowflakeHealth, action) <* | ||
| snowflakeHealth.setHealthy() | ||
| } | ||
|
|
||
| private def retryUntilSuccessful[F[_]: Sync: Sleep, A](snowflakeHealth: SnowflakeHealth[F], action: F[A]) = | ||
| action.retryingOnSomeErrors( | ||
| isWorthRetrying = isWorthRetrying(_).pure[F], | ||
| policy = RetryPolicies.exponentialBackoff[F](baseDelay = 1.minute), // TODO make it configurable | ||
| onError = (ex, _) => handleError(snowflakeHealth, ex) | ||
| ) | ||
|
|
||
| private def handleError[F[_]: Sync](snowflakeHealth: SnowflakeHealth[F], ex: Throwable): F[Unit] = | ||
| snowflakeHealth.setUnhealthy() *> | ||
| Logger[F].error(ex)("Executing Snowflake command failed") | ||
|
|
||
| private def isWorthRetrying(error: Throwable): Boolean = error match { | ||
| // TODO Retry for some specific error codes or simply for any snowflake exception? | ||
| case se: SnowflakeSQLException if se.getErrorCode === 2222 => | ||
| true | ||
| case _ => true | ||
| } | ||
| } |
17 changes: 17 additions & 0 deletions
17
.../src/main/scala/com.snowplowanalytics.snowplow.snowflake/processing/SnowflakeHealth.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,17 @@ | ||
| package com.snowplowanalytics.snowplow.snowflake.processing | ||
|
|
||
| import cats.effect.{Concurrent, Ref} | ||
| import cats.implicits._ | ||
| import com.snowplowanalytics.snowplow.runtime.HealthProbe | ||
|
|
||
| final case class SnowflakeHealth[F[_]](state: Ref[F, HealthProbe.Status]) { | ||
| def setUnhealthy(): F[Unit] = state.set(HealthProbe.Unhealthy("Snowflake is not healthy")) //TODO do we need more details here? | ||
| def setHealthy(): F[Unit] = state.set(HealthProbe.Healthy) | ||
| } | ||
|
|
||
| object SnowflakeHealth { | ||
| def initHealthy[F[_]: Concurrent]: F[SnowflakeHealth[F]] = | ||
| Ref | ||
| .of[F, HealthProbe.Status](HealthProbe.Healthy) | ||
| .map(SnowflakeHealth.apply) | ||
| } |
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