-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Extend health probe to report unhealthy on more error scenarios (#69)
The health probe now reports unhealthy under these extra scenarios: - Bad row sink is unhealthy – cannot write to the sink - Fatal error happens when trying to write events to the lake
- Loading branch information
1 parent
8d3626c
commit d8692fd
Showing
7 changed files
with
370 additions
and
42 deletions.
There are no files selected for viewing
82 changes: 82 additions & 0 deletions
82
modules/core/src/main/scala/com.snowplowanalytics.snowplow.lakes/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,82 @@ | ||
| /** | ||
| * Copyright (c) 2013-present Snowplow Analytics Ltd. All rights reserved. | ||
| * | ||
| * This software is made available by Snowplow Analytics, Ltd., under the terms of the Snowplow | ||
| * Limited Use License Agreement, Version 1.0 located at | ||
| * https://docs.snowplow.io/limited-use-license-1.0 BY INSTALLING, DOWNLOADING, ACCESSING, USING OR | ||
| * DISTRIBUTING ANY PORTION OF THE SOFTWARE, YOU AGREE TO THE TERMS OF SUCH LICENSE AGREEMENT. | ||
| */ | ||
| package com.snowplowanalytics.snowplow.lakes | ||
|
|
||
| import cats.effect.{Concurrent, Ref} | ||
| import cats.implicits._ | ||
| import cats.{Monad, Monoid, Show} | ||
| import com.snowplowanalytics.snowplow.runtime.HealthProbe | ||
| import com.snowplowanalytics.snowplow.runtime.HealthProbe.{Healthy, Unhealthy} | ||
| import com.snowplowanalytics.snowplow.sources.SourceAndAck | ||
|
|
||
| import scala.concurrent.duration.FiniteDuration | ||
|
|
||
| final class AppHealth[F[_]: Monad]( | ||
| unhealthyLatency: FiniteDuration, | ||
| source: SourceAndAck[F], | ||
| appManagedServices: Ref[F, Map[AppHealth.Service, Boolean]] | ||
| ) { | ||
|
|
||
| def status: F[HealthProbe.Status] = | ||
| for { | ||
| sourceHealth <- getSourceHealth | ||
| servicesHealth <- getAppManagedServicesHealth | ||
| } yield (sourceHealth :: servicesHealth).combineAll | ||
|
|
||
| def setServiceHealth(service: AppHealth.Service, isHealthy: Boolean): F[Unit] = | ||
| appManagedServices.update { currentHealth => | ||
| currentHealth.updated(service, isHealthy) | ||
| } | ||
|
|
||
| private def getAppManagedServicesHealth: F[List[HealthProbe.Status]] = | ||
| appManagedServices.get.map { services => | ||
| services.map { | ||
| case (service, false) => HealthProbe.Unhealthy(show"$service is not healthy") | ||
| case _ => HealthProbe.Healthy | ||
| }.toList | ||
| } | ||
|
|
||
| private def getSourceHealth: F[HealthProbe.Status] = | ||
| source.isHealthy(unhealthyLatency).map { | ||
| case SourceAndAck.Healthy => HealthProbe.Healthy | ||
| case unhealthy: SourceAndAck.Unhealthy => HealthProbe.Unhealthy(unhealthy.show) | ||
| } | ||
|
|
||
| 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") | ||
| } | ||
|
|
||
| private implicit val healthMonoid: Monoid[HealthProbe.Status] = Monoid.instance(Healthy, combineHealth) | ||
| } | ||
|
|
||
| object AppHealth { | ||
|
|
||
| sealed trait Service | ||
|
|
||
| object Service { | ||
| case object SparkWriter extends Service | ||
| case object BadSink extends Service | ||
|
|
||
| implicit val show: Show[Service] = Show.show { | ||
| case SparkWriter => "Spark writer" | ||
| case BadSink => "Failed events sink" | ||
| } | ||
| } | ||
|
|
||
| def init[F[_]: Concurrent]( | ||
| unhealthyLatency: FiniteDuration, | ||
| source: SourceAndAck[F] | ||
| ): F[AppHealth[F]] = | ||
| Ref[F] | ||
| .of(Map[Service, Boolean](Service.SparkWriter -> false, Service.BadSink -> false)) | ||
| .map(appManaged => new AppHealth[F](unhealthyLatency, source, appManaged)) | ||
| } |
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
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
Oops, something went wrong.