Warn about bad rows even in DummyStream sink (close #64)

Author: istreeter

config/config.kinesis.reference.hocon

@@ -74,6 +74,9 @@
       # Max size of the batch in bytes before emitting
       # Default is 5MB
       "maxBatchBytes": 5000000
+
+      # Only used when "type" is "Noop" or missing. How often to log number of bad rows discarded.
+      "reportPeriod": 10 seconds
     }
   }
 

config/config.local.reference.hocon

@@ -42,6 +42,9 @@
       "type": "Local"
       # Path for bad row sink.
       "path": "./tmp/bad"
+
+      # Only used when "type" is "Noop" or missing. How often to log number of bad rows discarded.
+      "reportPeriod": 10 seconds
     }
   }
 

config/config.pubsub.reference.hocon

@@ -63,6 +63,9 @@
       # The number of threads used internally by library to process the callback after message delivery
       # Default is 1
       "numCallbackExecutors": 1
+
+      # Only used when "type" is "Noop" or missing. How often to log number of bad rows discarded.
+      "reportPeriod": 10 seconds
     }
   }
 

modules/common/src/main/scala/com/snowplowanalytics/snowplow/postgres/streaming/DummyStreamSink.scala

@@ -12,9 +12,33 @@
  */
 package com.snowplowanalytics.snowplow.postgres.streaming
 
-import cats.effect.{Resource, Concurrent}
+import org.log4s.getLogger
+
+import cats.effect.{Resource, Concurrent, Sync, Timer}
+import cats.effect.concurrent.Ref
+
+import fs2.Stream
+
+import scala.concurrent.duration.FiniteDuration
 
 object DummyStreamSink {
-  def create[F[_]: Concurrent]:Resource[F, StreamSink[F]] =
-    Resource.pure[F, StreamSink[F]](_ => Concurrent[F].pure(()))
+  def create[F[_]: Concurrent: Timer](period: FiniteDuration): Resource[F, StreamSink[F]] =
+    for {
+      counter <- Resource.eval(Ref.of(0))
+      _ <- Concurrent[F].background(reporter(counter, period))
+  } yield { _ =>
+    counter.update(_ + 1)
+  }
+
+  lazy val logger = getLogger
+
+  private def reporter[F[_]: Sync: Timer](counter: Ref[F, Int], period: FiniteDuration): F[Unit] =
+    Stream.awakeDelay[F](period)
+      .evalMap(_ => counter.getAndSet(0))
+      .evalMap { count =>
+        if (count > 0) Sync[F].delay(logger.info(s"Discarded $count bad rows during the last $period"))
+        else Sync[F].unit
+      }
+      .compile
+      .drain
 }

modules/loader/src/main/resources/application.conf

@@ -22,6 +22,7 @@
       }
       "bad": {
         "type": "Noop"
+        "reportPeriod": 30 seconds
         "delayThreshold": 200 milliseconds
         "maxBatchSize": 500
         "maxBatchBytes": 5000000

modules/loader/src/main/scala/com/snowplowanalytics/snowplow/postgres/config/LoaderConfig.scala

@@ -234,7 +234,7 @@ object LoaderConfig {
   sealed trait StreamSink extends Product with Serializable
   object StreamSink {
 
-    case object Noop extends StreamSink
+    case class Noop(reportPeriod: FiniteDuration) extends StreamSink
 
     case class Local(path: PathInfo) extends StreamSink
 

modules/loader/src/main/scala/com/snowplowanalytics/snowplow/postgres/env/Environment.scala

@@ -42,7 +42,7 @@ object Environment {
                                                           blocker: Blocker) =
     for {
       badSink <- config.output.bad match {
-        case LoaderConfig.StreamSink.Noop => DummyStreamSink.create
+        case LoaderConfig.StreamSink.Noop(period) => DummyStreamSink.create(period)
         case c: LoaderConfig.StreamSink.Kinesis => KinesisSink.create(c, config.monitoring, config.backoffPolicy, blocker)
         case c: LoaderConfig.StreamSink.PubSub => PubSubSink.create(c, config.backoffPolicy)
         case c: LoaderConfig.StreamSink.Local => LocalSink.create(c, blocker)

modules/loader/src/test/scala/com.snowplowanalytics.snowplow.postgres/config/CliSpec.scala

@@ -57,7 +57,7 @@ class CliSpec extends Specification {
             10,
             None
           ),
-          LoaderConfig.StreamSink.Noop
+          LoaderConfig.StreamSink.Noop(30.seconds)
         ),
         Purpose.Enriched,
         Monitoring(Monitoring.Metrics(true)),
@@ -136,7 +136,7 @@ class CliSpec extends Specification {
             10,
             None
           ),
-          LoaderConfig.StreamSink.Noop
+          LoaderConfig.StreamSink.Noop(30.seconds)
         ),
         Purpose.Enriched,
         Monitoring(Monitoring.Metrics(true)),
@@ -209,7 +209,7 @@ class CliSpec extends Specification {
             10,
             None
           ),
-          LoaderConfig.StreamSink.Noop
+          LoaderConfig.StreamSink.Noop(30.seconds)
         ),
         Purpose.Enriched,
         Monitoring(Monitoring.Metrics(true)),
@@ -271,7 +271,7 @@ class CliSpec extends Specification {
             10,
             None
           ),
-          LoaderConfig.StreamSink.Noop
+          LoaderConfig.StreamSink.Noop(30.seconds)
         ),
         Purpose.Enriched,
         Monitoring(Monitoring.Metrics(true)),

modules/loader/src/test/scala/com.snowplowanalytics.snowplow.postgres/config/LocalSourceSpec.scala

@@ -57,7 +57,7 @@ class LocalSourceSpec extends Database {
             10,
             None
           ),
-          LoaderConfig.StreamSink.Noop
+          LoaderConfig.StreamSink.Noop(30.seconds)
         ),
         Purpose.Enriched,
         Monitoring(Monitoring.Metrics(false)),