Merge pull request #3427 from chenharryhua/0.18

0.18
chenharryhua · Nov 15, 2024 · c50bef0 · c50bef0
2 parents 31678f9 + e5e8eb5
commit c50bef0
Show file tree

Hide file tree

Showing 34 changed files with 313 additions and 350 deletions.
diff --git a/example/src/main/scala/example/observers.scala b/example/src/main/scala/example/observers.scala
@@ -32,7 +32,7 @@ object observers {
   val cloudwatch: CloudWatchObserver[IO] =
     CloudWatchObserver(CloudWatch[IO](_.region(Region.AP_SOUTHEAST_2)))
       .includeHistogram(_.withP50.withP95.withMax)
-      .includeDimensions(_.withServiceID.withServiceName.withDigest)
+      .includeDimensions(_.withServiceID.withServiceName)
       .unifyMeasurementUnit(_.withInfoUnit(_.BYTES))
 
   val sqsObserver: SqsObserver[IO] =

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/Batch.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/Batch.scala
@@ -118,7 +118,7 @@ object Batch {
       for {
         meas <- measure(jobs.size, BatchKind.Quasi, mode)
         case (fd, details) <- Resource.eval(exec(meas))
-      } yield List(QuasiResult(fd.toJava, mode, details.sortBy(_.job.index)))
+      } yield List(QuasiResult(metrics.metricLabel, fd.toJava, mode, details.sortBy(_.job.index)))
     }
 
     override val fully: Resource[F, List[A]] = {
@@ -156,6 +156,7 @@ object Batch {
         details <- Resource.eval(exec(meas))
       } yield List(
         QuasiResult(
+          label = metrics.metricLabel,
           spent = details.map(_.took).foldLeft(Duration.ZERO)(_ plus _),
           mode = mode,
           details = details.sortBy(_.job.index)

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/BatchJob.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/BatchJob.scala
@@ -1,6 +1,7 @@
 package com.github.chenharryhua.nanjin.guard.action
 import cats.Show
 import cats.syntax.all.*
+import com.github.chenharryhua.nanjin.guard.config.MetricLabel
 import com.github.chenharryhua.nanjin.guard.translator.fmt
 import io.circe.syntax.EncoderOps
 import io.circe.{Encoder, Json}
@@ -34,11 +35,12 @@ object BatchMode {
 final case class BatchJob(name: Option[String], index: Int)
 
 final case class Detail(job: BatchJob, took: Duration, done: Boolean)
-final case class QuasiResult(spent: Duration, mode: BatchMode, details: List[Detail])
+final case class QuasiResult(label: MetricLabel, spent: Duration, mode: BatchMode, details: List[Detail])
 object QuasiResult {
   implicit val encoderQuasiResult: Encoder[QuasiResult] = { (a: QuasiResult) =>
     val (done, fail) = a.details.partition(_.done)
     Json.obj(
+      "batch" -> Json.fromString(a.label.label),
       "mode" -> a.mode.asJson,
       "spent" -> Json.fromString(fmt.format(a.spent)),
       "done" -> Json.fromInt(done.length),

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/Herald.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/Herald.scala
@@ -71,7 +71,7 @@ object Herald {
 
     private def toText(sm: ServiceMessage): String = {
       val color = ColorScheme.decorate(sm).run(textHelper.consoleColor).value
-      val msg   = jsonHelper.jsonServiceMessage(sm).noSpaces
+      val msg   = jsonHelper.json_service_message(sm).noSpaces
       s"${sm.timestamp.format(fmt)} Console $color - $msg"
     }
 

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/Retry.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/action/Retry.scala
@@ -1,5 +1,6 @@
 package com.github.chenharryhua.nanjin.guard.action
 
+import cats.data.{EitherT, Kleisli}
 import cats.effect.Temporal
 import cats.implicits.{toFlatMapOps, toFunctorOps}
 import com.github.chenharryhua.nanjin.common.chrono.{Tick, TickStatus}
@@ -10,6 +11,8 @@ sealed trait Retry[F[_]] {
   def apply[A](arrow: (Tick, Option[Throwable]) => F[Either[Throwable, A]]): F[A]
   def apply[A](tfa: Tick => F[A]): F[A]
   def apply[A](fa: F[A]): F[A]
+
+  def apply[A](mt: Kleisli[EitherT[F, Throwable, *], (Tick, Option[Throwable]), A]): F[A]
 }
 
 object Retry {
@@ -44,5 +47,8 @@ object Retry {
 
     override def apply[A](arrow: (Tick, Option[Throwable]) => F[Either[Throwable, A]]): F[A] =
       comprehensive(arrow)
+
+    override def apply[A](mt: Kleisli[EitherT[F, Throwable, *], (Tick, Option[Throwable]), A]): F[A] =
+      comprehensive((t: Tick, o: Option[Throwable]) => mt.run((t, o)).value)
   }
 }
diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/config/MetricID.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/config/MetricID.scala
@@ -5,7 +5,6 @@ import com.github.chenharryhua.nanjin.guard.event.MeasurementUnit
 import enumeratum.{CirceEnum, Enum, EnumEntry}
 import io.circe.Encoder
 import io.circe.generic.JsonCodec
-import org.apache.commons.codec.digest.DigestUtils
 
 import java.util.UUID
 import scala.concurrent.duration.FiniteDuration
@@ -85,20 +84,7 @@ object MetricName {
 }
 
 @JsonCodec
-final case class MetricLabel private (label: String, digest: String, measurement: String)
-object MetricLabel {
-  def apply(serviceParams: ServiceParams, measurement: Measurement, label: String): MetricLabel = {
-    val full_name: List[String] =
-      serviceParams.taskName.value :: serviceParams.serviceName.value :: measurement.value :: label :: Nil
-    val digest = DigestUtils.sha256Hex(full_name.mkString("/")).take(8)
-
-    MetricLabel(
-      label = label,
-      digest = digest,
-      measurement = measurement.value
-    )
-  }
-}
+final case class MetricLabel(label: String, measurement: Measurement)
 
 @JsonCodec
 final case class MetricID(metricLabel: MetricLabel, metricName: MetricName, category: Category) {

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/event/MetricSnapshot.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/event/MetricSnapshot.scala
@@ -10,6 +10,7 @@ import org.typelevel.cats.time.instances.duration
 import squants.time.{Frequency, Hertz}
 
 import java.time.Duration
+import java.util.UUID
 import scala.jdk.CollectionConverters.*
 
 sealed trait Snapshot extends Product with Serializable { def metricId: MetricID }
@@ -94,12 +95,19 @@ final case class MetricSnapshot(
       histograms.map(_.metricId)
 
   def hasDuplication: Boolean = {
-    val stable = metricIDs.map(id => (id.metricLabel, id.metricName.name, id.category))
+    val stable = metricIDs.map(id => (id.metricLabel, id.metricName.name))
     stable.distinct.size != stable.size
   }
+
+  def lookupCount: Map[UUID, Long] = {
+    meters.map(m => m.metricId.metricName.uuid -> m.meter.sum) :::
+      timers.map(t => t.metricId.metricName.uuid -> t.timer.calls) :::
+      histograms.map(h => h.metricId.metricName.uuid -> h.histogram.updates)
+  }.toMap
 }
 
 object MetricSnapshot extends duration {
+  val empty: MetricSnapshot = MetricSnapshot(Nil, Nil, Nil, Nil, Nil)
 
   def counters(metricRegistry: MetricRegistry): List[Snapshot.Counter] =
     metricRegistry.getCounters().asScala.toList.mapFilter { case (name, counter) =>

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/event/NJEvent.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/event/NJEvent.scala
@@ -48,6 +48,7 @@ object NJEvent {
   final case class MetricReport(
     index: MetricIndex,
     serviceParams: ServiceParams,
+    previous: MetricSnapshot,
     snapshot: MetricSnapshot,
     timestamp: ZonedDateTime) // land time
       extends MetricEvent

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/event/eventFilters.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/event/eventFilters.scala
@@ -21,7 +21,7 @@ object eventFilters {
     */
   def sampling(interval: FiniteDuration)(evt: NJEvent): Boolean =
     evt match {
-      case MetricReport(mrt, sp, _, _) =>
+      case MetricReport(mrt, sp, _, _, _) =>
         mrt match {
           case MetricIndex.Adhoc(_) => true
           case MetricIndex.Periodic(tick) =>
@@ -42,7 +42,7 @@ object eventFilters {
     */
   def sampling(divisor: Refined[Int, Positive])(evt: NJEvent): Boolean =
     evt match {
-      case MetricReport(mrt, _, _, _) =>
+      case MetricReport(mrt, _, _, _, _) =>
         mrt match {
           case MetricIndex.Adhoc(_)       => true
           case MetricIndex.Periodic(tick) => (tick.index % divisor.value) === 0
@@ -54,7 +54,7 @@ object eventFilters {
     */
   def sampling(cronExpr: CronExpr)(evt: NJEvent): Boolean =
     evt match {
-      case MetricReport(mrt, _, _, _) =>
+      case MetricReport(mrt, _, _, _, _) =>
         mrt match {
           case MetricIndex.Adhoc(_) => true
           case MetricIndex.Periodic(tick) =>

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/Metrics.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/Metrics.scala
@@ -6,11 +6,8 @@ import cats.effect.kernel.{Async, Ref, Resource}
 import cats.syntax.all.*
 import com.codahale.metrics.MetricRegistry
 import com.github.chenharryhua.nanjin.guard.config.MetricLabel
-import com.github.chenharryhua.nanjin.guard.event.NJUnits
 import com.github.chenharryhua.nanjin.guard.translator.{decimal_fmt, fmt}
 
-import scala.concurrent.duration.DurationInt
-
 trait KleisliLike[F[_], A] {
   def run(a: A): F[Unit]
 
@@ -21,6 +18,7 @@ trait KleisliLike[F[_], A] {
 }
 
 sealed trait Metrics[F[_]] {
+  def metricLabel: MetricLabel
 
   def counter(name: String, f: Endo[NJCounter.Builder]): Resource[F, NJCounter[F]]
   final def counter(name: String): Resource[F, NJCounter[F]] = counter(name, identity)
@@ -58,46 +56,30 @@ sealed trait Metrics[F[_]] {
 }
 
 object Metrics {
-  private[guard] class Impl[F[_]](
-    metricLabel: MetricLabel,
-    metricRegistry: MetricRegistry,
-    isEnabled: Boolean)(implicit F: Async[F])
+  private[guard] class Impl[F[_]](val metricLabel: MetricLabel, metricRegistry: MetricRegistry)(implicit
+    F: Async[F])
       extends Metrics[F] {
 
-    override def counter(name: String, f: Endo[NJCounter.Builder]): Resource[F, NJCounter[F]] = {
-      val init = new NJCounter.Builder(isEnabled, false)
-      f(init).build[F](metricLabel, name, metricRegistry)
-    }
-
-    override def meter(name: String, f: Endo[NJMeter.Builder]): Resource[F, NJMeter[F]] = {
-      val init = new NJMeter.Builder(isEnabled, NJUnits.COUNT)
-      f(init).build[F](metricLabel, name, metricRegistry)
-    }
-
-    override def histogram(name: String, f: Endo[NJHistogram.Builder]): Resource[F, NJHistogram[F]] = {
-      val init = new NJHistogram.Builder(isEnabled, NJUnits.COUNT, None)
-      f(init).build[F](metricLabel, name, metricRegistry)
-    }
-
-    override def timer(name: String, f: Endo[NJTimer.Builder]): Resource[F, NJTimer[F]] = {
-      val init = new NJTimer.Builder(isEnabled, None)
-      f(init).build[F](metricLabel, name, metricRegistry)
-    }
-
-    override def healthCheck(name: String, f: Endo[NJHealthCheck.Builder]): NJHealthCheck[F] = {
-      val init = new NJHealthCheck.Builder(isEnabled, timeout = 5.seconds)
-      f(init).build[F](metricLabel, name, metricRegistry)
-    }
-
-    override def ratio(name: String, f: Endo[NJRatio.Builder]): Resource[F, NJRatio[F]] = {
-      val init = new NJRatio.Builder(isEnabled, NJRatio.translator)
-      f(init).build[F](metricLabel, name, metricRegistry)
-    }
-
-    override def gauge(name: String, f: Endo[NJGauge.Builder]): NJGauge[F] = {
-      val init = new NJGauge.Builder(isEnabled, timeout = 5.seconds)
-      f(init).build[F](metricLabel, name, metricRegistry)
-    }
+    override def counter(name: String, f: Endo[NJCounter.Builder]): Resource[F, NJCounter[F]] =
+      f(NJCounter.initial).build[F](metricLabel, name, metricRegistry)
+
+    override def meter(name: String, f: Endo[NJMeter.Builder]): Resource[F, NJMeter[F]] =
+      f(NJMeter.initial).build[F](metricLabel, name, metricRegistry)
+
+    override def histogram(name: String, f: Endo[NJHistogram.Builder]): Resource[F, NJHistogram[F]] =
+      f(NJHistogram.initial).build[F](metricLabel, name, metricRegistry)
+
+    override def timer(name: String, f: Endo[NJTimer.Builder]): Resource[F, NJTimer[F]] =
+      f(NJTimer.initial).build[F](metricLabel, name, metricRegistry)
+
+    override def healthCheck(name: String, f: Endo[NJHealthCheck.Builder]): NJHealthCheck[F] =
+      f(NJHealthCheck.initial).build[F](metricLabel, name, metricRegistry)
+
+    override def ratio(name: String, f: Endo[NJRatio.Builder]): Resource[F, NJRatio[F]] =
+      f(NJRatio.initial).build[F](metricLabel, name, metricRegistry)
+
+    override def gauge(name: String, f: Endo[NJGauge.Builder]): NJGauge[F] =
+      f(NJGauge.initial).build[F](metricLabel, name, metricRegistry)
 
     override def idleGauge(name: String, f: Endo[NJGauge.Builder]): Resource[F, NJIdleGauge[F]] =
       for {

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJCounter.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJCounter.scala
@@ -45,6 +45,8 @@ object NJCounter {
 
   }
 
+  val initial: Builder = new Builder(isEnabled = true, isRisk = false)
+
   final class Builder private[guard] (isEnabled: Boolean, isRisk: Boolean) extends EnableConfig[Builder] {
 
     def asRisk: Builder = new Builder(isEnabled, true)

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJGauge.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJGauge.scala
@@ -15,7 +15,7 @@ import org.apache.commons.lang3.StringUtils
 import org.apache.commons.lang3.exception.ExceptionUtils
 
 import java.time.ZoneId
-import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration.{DurationInt, FiniteDuration}
 import scala.util.Try
 
 sealed trait NJGauge[F[_]] {
@@ -78,6 +78,8 @@ object NJGauge {
     }
   }
 
+  val initial: Builder = new Builder(isEnabled = true, timeout = 5.seconds)
+
   final class Builder private[guard] (isEnabled: Boolean, timeout: FiniteDuration)
       extends EnableConfig[Builder] {
 

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJHealthCheck.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJHealthCheck.scala
@@ -11,7 +11,7 @@ import com.github.chenharryhua.nanjin.guard.config.*
 import com.github.chenharryhua.nanjin.guard.config.CategoryKind.GaugeKind
 
 import java.time.ZoneId
-import scala.concurrent.duration.FiniteDuration
+import scala.concurrent.duration.{DurationInt, FiniteDuration}
 import scala.util.Try
 
 sealed trait NJHealthCheck[F[_]] {
@@ -71,6 +71,8 @@ object NJHealthCheck {
     }
   }
 
+  val initial: Builder = new Builder(isEnabled = true, timeout = 5.seconds)
+
   final class Builder private[guard] (isEnabled: Boolean, timeout: FiniteDuration)
       extends EnableConfig[Builder] {
 

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJHistogram.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJHistogram.scala
@@ -51,6 +51,8 @@ object NJHistogram {
     val unregister: F[Unit] = F.delay(metricRegistry.remove(histogram_name)).void
   }
 
+  val initial: Builder = new Builder(isEnabled = true, unit = NJUnits.COUNT, reservoir = None)
+
   final class Builder private[guard] (isEnabled: Boolean, unit: MeasurementUnit, reservoir: Option[Reservoir])
       extends EnableConfig[Builder] {
 

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJMeter.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJMeter.scala
@@ -42,6 +42,8 @@ object NJMeter {
     val unregister: F[Unit] = F.delay(metricRegistry.remove(meter_name)).void
   }
 
+  val initial: Builder = new Builder(isEnabled = true, unit = NJUnits.COUNT)
+
   final class Builder private[guard] (isEnabled: Boolean, unit: MeasurementUnit)
       extends EnableConfig[Builder] {
 

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJRatio.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJRatio.scala
@@ -76,6 +76,8 @@ object NJRatio {
       }
   }
 
+  val initial: Builder = new Builder(isEnabled = true, translator = translator)
+
   final class Builder private[guard] (
     isEnabled: Boolean,
     translator: Ior[Long, Long] => Json

diff --git a/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJTimer.scala b/guard/src/main/scala/com/github/chenharryhua/nanjin/guard/metrics/NJTimer.scala
@@ -70,6 +70,8 @@ object NJTimer {
 
   }
 
+  val initial: Builder = new Builder(isEnabled = true, reservoir = None)
+
   final class Builder private[guard] (
     isEnabled: Boolean,
     reservoir: Option[Reservoir]