best case

Author: oguzhanunlu

modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Metrics.scala

@@ -21,6 +21,8 @@ trait Metrics[F[_]] {
   def addGood(count: Int): F[Unit]
   def addBad(count: Int): F[Unit]
   def setLatencyMillis(latencyMillis: Long): F[Unit]
+  def setLatencyCollectorToTargetMillis(latencyMillis: Long): F[Unit]
+  def setLatencyCollectorToTargetPessimisticMillis(latencyMillis: Long): F[Unit]
 
   def report: Stream[F, Nothing]
 }
@@ -33,18 +35,22 @@ object Metrics {
   private case class State(
     good: Int,
     bad: Int,
-    latencyMillis: Long
+    latencyMillis: Long,
+    latencyCollectorToTargetMillis: Long,
+    latencyCollectorToTargetPessimisticMillis: Long
   ) extends CommonMetrics.State {
     def toKVMetrics: List[CommonMetrics.KVMetric] =
       List(
         KVMetric.CountGood(good),
         KVMetric.CountBad(bad),
-        KVMetric.LatencyMillis(latencyMillis)
+        KVMetric.LatencyMillis(latencyMillis),
+        KVMetric.LatencyCollectorToTargetMillis(latencyCollectorToTargetMillis),
+        KVMetric.LatencyCollectorToTargetPessimisticMillis(latencyCollectorToTargetPessimisticMillis)
       )
   }
 
   private object State {
-    def empty: State = State(0, 0, 0L)
+    def empty: State = State(0, 0, 0L, 0L, 0L)
   }
 
   private def impl[F[_]: Async](config: Config.Metrics, ref: Ref[F, State]): Metrics[F] =
@@ -55,6 +61,10 @@ object Metrics {
         ref.update(s => s.copy(bad = s.bad + count))
       def setLatencyMillis(latencyMillis: Long): F[Unit] =
         ref.update(s => s.copy(latencyMillis = s.latencyMillis.max(latencyMillis)))
+      def setLatencyCollectorToTargetMillis(latencyMillis: Long): F[Unit] =
+        ref.update(s => s.copy(latencyMillis = s.latencyMillis.min(latencyMillis)))
+      def setLatencyCollectorToTargetPessimisticMillis(latencyMillis: Long): F[Unit] =
+        ref.update(s => s.copy(latencyMillis = s.latencyMillis.max(latencyMillis)))
     }
 
   private object KVMetric {
@@ -77,5 +87,16 @@ object Metrics {
       val metricType = CommonMetrics.MetricType.Gauge
     }
 
+    final case class LatencyCollectorToTargetMillis(v: Long) extends CommonMetrics.KVMetric {
+      val key        = "latency_collector_to_target_millis"
+      val value      = v.toString
+      val metricType = CommonMetrics.MetricType.Gauge
+    }
+
+    final case class LatencyCollectorToTargetPessimisticMillis(v: Long) extends CommonMetrics.KVMetric {
+      val key        = "latency_collector_to_target_pessimistic_millis"
+      val value      = v.toString
+      val metricType = CommonMetrics.MetricType.Gauge
+    }
   }
 }

modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/processing/Processing.scala

@@ -21,8 +21,7 @@ import org.typelevel.log4cats.Logger
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 
 import java.nio.charset.StandardCharsets
-import java.time.OffsetDateTime
-
+import java.time.{Instant, OffsetDateTime}
 import com.snowplowanalytics.iglu.schemaddl.parquet.Caster
 import com.snowplowanalytics.snowplow.analytics.scalasdk.Event
 import com.snowplowanalytics.snowplow.badrows.{BadRow, Payload => BadPayload, Processor => BadRowProcessor}
@@ -86,7 +85,9 @@ object Processing {
     origBatchBytes: Long,
     origBatchCount: Int,
     badAccumulated: ListOfList[BadRow],
-    tokens: Vector[Unique.Token]
+    tokens: Vector[Unique.Token],
+    maxCollectorTstamp: Instant,
+    minCollectorTstamp: Instant
   )
 
   /**
@@ -246,6 +247,18 @@ object Processing {
       }
   }
 
+  private def setLatencyCollectorToTargetMillis[F[_]: Sync](
+    metrics: Metrics[F],
+    collectorTstamp: Instant,
+    optimistic: Boolean
+  ): F[Unit] =
+    for {
+      now <- Sync[F].realTime
+      latencyMillis = now.toMillis - collectorTstamp.toEpochMilli
+      _ <- if (optimistic) metrics.setLatencyCollectorToTargetMillis(latencyMillis)
+           else metrics.setLatencyCollectorToTargetPessimisticMillis(latencyMillis)
+    } yield ()
+
   /**
    * First attempt to write events with the Snowflake SDK
    *
@@ -258,8 +271,14 @@ object Processing {
     batch: BatchAfterTransform
   ): F[BatchAfterTransform] =
     withWriteAttempt(env, batch) { notWritten =>
+      val setLatency =
+        if (notWritten.isEmpty) {
+          setLatencyCollectorToTargetMillis(env.metrics, batch.maxCollectorTstamp, optimistic   = true) >>
+            setLatencyCollectorToTargetMillis(env.metrics, batch.minCollectorTstamp, optimistic = false)
+        } else Sync[F].unit
       val parsedResult = ParsedWriteResult.buildFrom(batch.toBeInserted, notWritten)
       for {
+        _ <- setLatency
         _ <- abortIfFatalException[F](parsedResult.unexpectedFailures)
         _ <- handleSchemaEvolution(env, parsedResult.extraColsRequired)
       } yield {
@@ -285,13 +304,18 @@ object Processing {
     batch: BatchAfterTransform
   ): F[BatchAfterTransform] =
     withWriteAttempt(env, batch) { notWritten =>
+      val setLatency =
+        if (notWritten.isEmpty) {
+          setLatencyCollectorToTargetMillis(env.metrics, batch.maxCollectorTstamp, optimistic   = true) >>
+            setLatencyCollectorToTargetMillis(env.metrics, batch.minCollectorTstamp, optimistic = false)
+        } else Sync[F].unit
       val mapped = notWritten match {
         case Nil => Nil
         case more =>
           val indexed = batch.toBeInserted.copyToIndexedSeq
           more.map(f => (fastGetByIndex(indexed, f.index)._1, f.cause))
       }
-      abortIfFatalException[F](mapped).as {
+      setLatency >> abortIfFatalException[F](mapped).as {
         val moreBad = mapped.map { case (event, sfe) =>
           badRowFromEnqueueFailure(badProcessor, event, sfe)
         }
@@ -385,15 +409,25 @@ object Processing {
 
   private def fastGetByIndex[A](items: IndexedSeq[A], index: Long): A = items(index.toInt)
 
+  private implicit class MaxCollectorTstamp(events: List[EventWithTransform]) {
+    def maxCollectorTstamp(): Instant = events.map(_._1).map(_.collector_tstamp).max
+  }
+
+  private implicit class MinCollectorTstamp(events: List[EventWithTransform]) {
+    def minCollectorTstamp(): Instant = events.map(_._1).map(_.collector_tstamp).min
+  }
+
   private implicit def batchable: BatchUp.Batchable[TransformedBatch, BatchAfterTransform] =
     new BatchUp.Batchable[TransformedBatch, BatchAfterTransform] {
       def combine(b: BatchAfterTransform, a: TransformedBatch): BatchAfterTransform =
         BatchAfterTransform(
-          toBeInserted   = b.toBeInserted.prepend(a.events),
-          origBatchBytes = b.origBatchBytes + a.countBytes,
-          origBatchCount = b.origBatchCount + a.countItems,
-          badAccumulated = b.badAccumulated.prepend(a.parseFailures).prepend(a.transformFailures),
-          tokens         = b.tokens :+ a.token
+          toBeInserted       = b.toBeInserted.prepend(a.events),
+          origBatchBytes     = b.origBatchBytes + a.countBytes,
+          origBatchCount     = b.origBatchCount + a.countItems,
+          badAccumulated     = b.badAccumulated.prepend(a.parseFailures).prepend(a.transformFailures),
+          tokens             = b.tokens :+ a.token,
+          maxCollectorTstamp = List(b.maxCollectorTstamp, a.events.maxCollectorTstamp()).max,
+          minCollectorTstamp = List(b.maxCollectorTstamp, a.events.maxCollectorTstamp()).min
         )
 
       def single(a: TransformedBatch): BatchAfterTransform =
@@ -402,7 +436,9 @@ object Processing {
           a.countBytes,
           a.countItems,
           ListOfList.ofLists(a.parseFailures, a.transformFailures),
-          Vector(a.token)
+          Vector(a.token),
+          a.events.maxCollectorTstamp(),
+          a.events.minCollectorTstamp()
         )
 
       def weightOf(a: TransformedBatch): Long =