Add ability to disable CloudWatch metric reportings (close #11)

Author: istreeter

config/config.hocon.sample

@@ -16,12 +16,15 @@
     # Either TRIM_HORIZON or LATEST
     "initialPosition": "TRIM_HORIZON"
 
-    # Use either enhanced fan out or polling mode for retrieving records
-    "retrievalMode": "FanOut"
+    # Optional, set the  polling mode for retrieving records. Default is FanOut
+    # "retrievalMode": "FanOut"
     # "retrievalMode": {
     #   "type": "Polling"
     #   "maxRecords": 1000
     # }
+
+    # Optional override to disable cloudwatch metrics
+    # "disableCloudWatch": true
   }
 
   "storage" : {

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

@@ -1,3 +1,4 @@
 snowplow {
-  input.retrievalMode = FAN_OUT
+  input.retrievalMode = FanOut
+  input.disableCloudWatch = false
 }

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

@@ -107,7 +107,8 @@ object LoaderConfig {
                        streamName: String,
                        region: Region,
                        initialPosition: InitPosition,
-                       retrievalMode: Kinesis.Retrieval) extends Source
+                       retrievalMode: Kinesis.Retrieval,
+                       disableCloudWatch: Boolean) extends Source
     case class PubSub(projectId: String, subscriptionId: String) extends Source
 
     object Kinesis {

modules/loader/src/main/scala/com/snowplowanalytics/snowplow/postgres/streaming/source.scala

@@ -45,6 +45,7 @@ import com.permutive.pubsub.consumer.decoder.MessageDecoder
 import software.amazon.awssdk.regions.Region
 import software.amazon.kinesis.common.ConfigsBuilder
 import software.amazon.kinesis.coordinator.Scheduler
+import software.amazon.kinesis.metrics.MetricsLevel
 import software.amazon.kinesis.processor.ShardRecordProcessorFactory
 import software.amazon.kinesis.retrieval.polling.PollingConfig
 import software.amazon.kinesis.retrieval.fanout.FanOutConfig
@@ -171,12 +172,16 @@ object source {
             }
           }
 
+      val metricsConfig = configsBuilder.metricsConfig.metricsLevel {
+        if (config.disableCloudWatch) MetricsLevel.NONE else MetricsLevel.DETAILED
+      }
+
       new Scheduler(
         configsBuilder.checkpointConfig,
         configsBuilder.coordinatorConfig,
         configsBuilder.leaseManagementConfig,
         configsBuilder.lifecycleConfig,
-        configsBuilder.metricsConfig,
+        metricsConfig,
         configsBuilder.processorConfig,
         retrievalConfig
       )

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

@@ -19,9 +19,10 @@ import cats.effect.{Clock, IO}
 
 import com.snowplowanalytics.snowplow.postgres.config.LoaderConfig.{InitPosition, Purpose, Source}
 
-import org.specs2.mutable.Specification
 import software.amazon.awssdk.regions.Region
 
+import org.specs2.mutable.Specification
+
 class CliSpec extends Specification {
   implicit val ioClock: Clock[IO] = Clock.create[IO]
 
@@ -34,7 +35,14 @@ class CliSpec extends Specification {
       val expected = LoaderConfig(
         "Acme Ltd. Snowplow Postgres",
         UUID.fromString("5c5e4353-4eeb-43da-98f8-2de6dc7fa947"),
-        Source.Kinesis("acme-postgres-loader", "enriched-events", Region.EU_CENTRAL_1, InitPosition.TrimHorizon, Source.Kinesis.Retrieval.FanOut),
+        Source.Kinesis(
+          "acme-postgres-loader",
+          "enriched-events",
+          Region.EU_CENTRAL_1,
+          InitPosition.TrimHorizon,
+          Source.Kinesis.Retrieval.FanOut,
+          false
+        ),
         DBConfig(
           "localhost",
           5432,