Add option to exit on missing Iglu schemas (#382)

Before this PR, the loader would generate a failed event if it failed to fetch a required schema from Iglu. However, all events have already passed validation in Enrich, so it is completely unexpected to have an Iglu failure. An Iglu error _probably_ means some type of configuration error or service outage. After this PR, the loader will crash and exit on an Iglu error, instead of creating a failed event. This is probably the preferred behaviour, while the pipeline operator addresses the underlying infrastructure problem. If an Iglu schema is genuinely now unavailable, then the pipeline operator can override the default behaviour by setting `exitOnMissingIgluSchema: false` in the configuration file or by listing the missing schema in `skipschemas`.
snowplow-incubator · Sep 5, 2024 · 32a1a0e · 32a1a0e
1 parent 531d02c
commit 32a1a0e
Show file tree

Hide file tree

Showing 13 changed files with 155 additions and 32 deletions.
diff --git a/config/config.azure.reference.hocon b/config/config.azure.reference.hocon
@@ -111,6 +111,12 @@
     "iglu:com.acme/legacy/jsonschema/2-*-*"
   ]
 
+  # -- Whether the loader should crash and exit if it fails to resolve an Iglu Schema.
+  # -- We recommend `true` because Snowplow enriched events have already passed validation, so a missing schema normally
+  # -- indicates an error that needs addressing.
+  # -- Change to `false` so events go the failed events stream instead of crashing the loader.
+  "exitOnMissingIgluSchema": true
+
   "monitoring": {
     "metrics": {
 

diff --git a/config/config.kinesis.reference.hocon b/config/config.kinesis.reference.hocon
@@ -133,6 +133,12 @@
     "iglu:com.acme/legacy/jsonschema/2-*-*"
   ]
 
+  # -- Whether the loader should crash and exit if it fails to resolve an Iglu Schema.
+  # -- We recommend `true` because Snowplow enriched events have already passed validation, so a missing schema normally
+  # -- indicates an error that needs addressing.
+  # -- Change to `false` so events go the failed events stream instead of crashing the loader.
+  "exitOnMissingIgluSchema": true
+
   "monitoring": {
     "metrics": {
 

diff --git a/config/config.pubsub.reference.hocon b/config/config.pubsub.reference.hocon
@@ -114,6 +114,12 @@
     "iglu:com.acme/legacy/jsonschema/2-*-*"
   ]
 
+  # -- Whether the loader should crash and exit if it fails to resolve an Iglu Schema.
+  # -- We recommend `true` because Snowplow enriched events have already passed validation, so a missing schema normally
+  # -- indicates an error that needs addressing.
+  # -- Change to `false` so events go the failed events stream instead of crashing the loader.
+  "exitOnMissingIgluSchema": true
+
   "monitoring": {
     "metrics": {
 

diff --git a/modules/core/src/main/resources/reference.conf b/modules/core/src/main/resources/reference.conf
@@ -42,6 +42,7 @@
 
   "skipSchemas": []
   "legacyColumns": []
+  "exitOnMissingIgluSchema": true
 
   "monitoring": {
     "metrics": {

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Config.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Config.scala
@@ -31,7 +31,8 @@ case class Config[+Source, +Sink](
   monitoring: Config.Monitoring,
   license: AcceptedLicense,
   skipSchemas: List[SchemaCriterion],
-  legacyColumns: List[SchemaCriterion]
+  legacyColumns: List[SchemaCriterion],
+  exitOnMissingIgluSchema: Boolean
 )
 
 object Config {

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Environment.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/Environment.scala
@@ -37,7 +37,8 @@ case class Environment[F[_]](
   batching: Config.Batching,
   badRowMaxSize: Int,
   schemasToSkip: List[SchemaCriterion],
-  legacyColumns: List[SchemaCriterion]
+  legacyColumns: List[SchemaCriterion],
+  exitOnMissingIgluSchema: Boolean
 )
 
 object Environment {
@@ -66,20 +67,21 @@ object Environment {
       writerBuilder <- Writer.builder(config.main.output.good, creds)
       writerProvider <- Writer.provider(writerBuilder, config.main.retries, appHealth)
     } yield Environment(
-      appInfo              = appInfo,
-      source               = sourceAndAck,
-      badSink              = badSink,
-      resolver             = resolver,
-      httpClient           = httpClient,
-      tableManager         = tableManagerWrapped,
-      writer               = writerProvider,
-      metrics              = metrics,
-      appHealth            = appHealth,
-      alterTableWaitPolicy = BigQueryRetrying.policyForAlterTableWait[F](config.main.retries),
-      batching             = config.main.batching,
-      badRowMaxSize        = config.main.output.bad.maxRecordSize,
-      schemasToSkip        = config.main.skipSchemas,
-      legacyColumns        = config.main.legacyColumns
+      appInfo                 = appInfo,
+      source                  = sourceAndAck,
+      badSink                 = badSink,
+      resolver                = resolver,
+      httpClient              = httpClient,
+      tableManager            = tableManagerWrapped,
+      writer                  = writerProvider,
+      metrics                 = metrics,
+      appHealth               = appHealth,
+      alterTableWaitPolicy    = BigQueryRetrying.policyForAlterTableWait[F](config.main.retries),
+      batching                = config.main.batching,
+      badRowMaxSize           = config.main.output.bad.maxRecordSize,
+      schemasToSkip           = config.main.skipSchemas,
+      legacyColumns           = config.main.legacyColumns,
+      exitOnMissingIgluSchema = config.main.exitOnMissingIgluSchema
     )
 
   private def enableSentry[F[_]: Sync](appInfo: AppInfo, config: Option[Config.Sentry]): Resource[F, Unit] =

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/RuntimeService.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/RuntimeService.scala
@@ -16,10 +16,12 @@ object RuntimeService {
 
   case object BigQueryClient extends RuntimeService
   case object BadSink extends RuntimeService
+  case object Iglu extends RuntimeService
 
   implicit val show: Show[RuntimeService] = Show.show {
     case BigQueryClient => "BigQuery client"
-    case BadSink        => "Bad sink"
+    case BadSink        => "Failed event sink"
+    case Iglu           => "Iglu repositories"
   }
 
 }
diff --git a/...s/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/processing/Processing.scala b/...s/core/src/main/scala/com.snowplowanalytics.snowplow.bigquery/processing/Processing.scala
@@ -13,6 +13,7 @@ import cats.{Applicative, Foldable}
 import cats.effect.{Async, Sync}
 import cats.effect.kernel.Unique
 import fs2.{Chunk, Pipe, Stream}
+import io.circe.syntax._
 import org.typelevel.log4cats.Logger
 import org.typelevel.log4cats.slf4j.Slf4jLogger
 import retry.{PolicyDecision, RetryDetails, RetryPolicy}
@@ -151,6 +152,7 @@ object Processing {
         _ <- Logger[F].debug(s"Processing batch of size ${events.size}")
         v2NonAtomicFields <- NonAtomicFields.resolveTypes[F](env.resolver, entities, env.schemasToSkip ::: env.legacyColumns)
         legacyFields <- LegacyColumns.resolveTypes[F](env.resolver, entities, env.legacyColumns)
+        _ <- possiblyExitOnMissingIgluSchema(env, v2NonAtomicFields, legacyFields)
         (moreBad, rows) <- transformBatch[F](badProcessor, loadTstamp, events, v2NonAtomicFields, legacyFields)
         fields = v2NonAtomicFields.fields.flatMap { tte =>
                    tte.mergedField :: tte.recoveries.map(_._2)
@@ -380,4 +382,19 @@ object Processing {
       a.countBytes
   }
 
+  private def possiblyExitOnMissingIgluSchema[F[_]: Sync](
+    env: Environment[F],
+    v2NonAtomicFields: NonAtomicFields.Result,
+    legacyFields: LegacyColumns.Result
+  ): F[Unit] =
+    if (env.exitOnMissingIgluSchema && (v2NonAtomicFields.igluFailures.nonEmpty || legacyFields.igluFailures.nonEmpty)) {
+      val base =
+        "Exiting because failed to resolve Iglu schemas.  Either check the configuration of the Iglu repos, or set the `skipSchemas` config option, or set `exitOnMissingIgluSchema` to false.\n"
+      val failures = v2NonAtomicFields.igluFailures.map(_.failure) ::: legacyFields.igluFailures.map(_.failure)
+      val msg      = failures.map(_.asJson.noSpaces).mkString(base, "\n", "")
+      env.appHealth.beUnhealthyForRuntimeService(RuntimeService.Iglu) *> Logger[F].error(base) *> Sync[F].raiseError(
+        new RuntimeException(msg)
+      )
+    } else Applicative[F].unit
+
 }
diff --git a/modules/core/src/test/scala/com.snowplowanalytics.snowplow.bigquery/MockEnvironment.scala b/modules/core/src/test/scala/com.snowplowanalytics.snowplow.bigquery/MockEnvironment.scala
@@ -86,9 +86,10 @@ object MockEnvironment {
           maxDelay              = 10.seconds,
           writeBatchConcurrency = 1
         ),
-        badRowMaxSize = 1000000,
-        schemasToSkip = List.empty,
-        legacyColumns = legacyColumns
+        badRowMaxSize           = 1000000,
+        schemasToSkip           = List.empty,
+        legacyColumns           = legacyColumns,
+        exitOnMissingIgluSchema = false
       )
       MockEnvironment(state, env)
     }

diff --git a/...re/src/test/scala/com.snowplowanalytics.snowplow.bigquery/processing/ProcessingSpec.scala b/...re/src/test/scala/com.snowplowanalytics.snowplow.bigquery/processing/ProcessingSpec.scala
@@ -49,6 +49,8 @@ class ProcessingSpec extends Specification with CatsEffect {
     Set the latency metric based off the message timestamp $e9
     Mark app as unhealthy when sinking badrows fails $e10
     Mark app as unhealthy when writing to the Writer fails with runtime exception $e11
+    Emit BadRows for an unknown schema, if exitOnMissingIgluSchema is false $e12 $e12Legacy
+    Crash and exit for an unknown schema, if exitOnMissingIgluSchema is true $e13 $e13Legacy
   """
 
   def e1 =
@@ -506,6 +508,79 @@ class ProcessingSpec extends Specification with CatsEffect {
 
   }
 
+  def e12Base(legacyColumns: Boolean) = {
+    val unstructEvent: UnstructEvent = json"""
+    {
+      "schema": "iglu:com.snowplowanalytics.snowplow/unstruct_event/jsonschema/1-0-0",
+      "data": {
+        "schema": "iglu:com.unkown/unknown_event/jsonschema/1-0-0",
+        "data": {
+          "abc": 50
+        }
+      }
+    }
+    """.as[UnstructEvent].fold(throw _, identity)
+
+    val legacyCriteria =
+      if (legacyColumns) List(SchemaCriterion("com.unknown", "unknown_event", "jsonschema", 1))
+      else Nil
+
+    runTest(inputEvents(count = 1, good(unstructEvent)), legacyCriteria = legacyCriteria) { case (inputs, control) =>
+      for {
+        _ <- Processing.stream(control.environment).compile.drain
+        state <- control.state.get
+      } yield state should beEqualTo(
+        Vector(
+          Action.CreatedTable,
+          Action.OpenedWriter,
+          Action.WroteRowsToBigQuery(1),
+          Action.SentToBad(1),
+          Action.AddedGoodCountMetric(1),
+          Action.AddedBadCountMetric(1),
+          Action.Checkpointed(List(inputs(0).ack))
+        )
+      )
+    }
+  }
+
+  def e12       = e12Base(legacyColumns = false)
+  def e12Legacy = e12Base(legacyColumns = true)
+
+  def e13Base(legacyColumns: Boolean) = {
+    val unstructEvent: UnstructEvent = json"""
+    {
+      "schema": "iglu:com.snowplowanalytics.snowplow/unstruct_event/jsonschema/1-0-0",
+      "data": {
+        "schema": "iglu:com.unkown/unknown_event/jsonschema/1-0-0",
+        "data": {
+          "abc": 50
+        }
+      }
+    }
+    """.as[UnstructEvent].fold(throw _, identity)
+
+    val legacyCriteria =
+      if (legacyColumns) List(SchemaCriterion("com.unknown", "unknown_event", "jsonschema", 1))
+      else Nil
+
+    runTest(inputEvents(count = 1, good(unstructEvent)), legacyCriteria = legacyCriteria) { case (_, control) =>
+      val environment = control.environment.copy(exitOnMissingIgluSchema = true)
+      for {
+        _ <- Processing.stream(environment).compile.drain.voidError
+        state <- control.state.get
+      } yield state should beEqualTo(
+        Vector(
+          Action.CreatedTable,
+          Action.OpenedWriter,
+          Action.BecameUnhealthy(RuntimeService.Iglu)
+        )
+      )
+    }
+  }
+
+  def e13       = e13Base(legacyColumns = false)
+  def e13Legacy = e13Base(legacyColumns = true)
+
 }
 
 object ProcessingSpec {

diff --git a/modules/kafka/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KafkaConfigSpec.scala b/modules/kafka/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KafkaConfigSpec.scala
@@ -123,9 +123,10 @@ object KafkaConfigSpec {
       healthProbe = Config.HealthProbe(port = Port.fromInt(8000).get, unhealthyLatency = 5.minutes),
       webhook     = Webhook.Config(endpoint = None, tags = Map.empty, heartbeat = 60.minutes)
     ),
-    license       = AcceptedLicense(),
-    skipSchemas   = List.empty,
-    legacyColumns = List.empty
+    license                 = AcceptedLicense(),
+    skipSchemas             = List.empty,
+    legacyColumns           = List.empty,
+    exitOnMissingIgluSchema = true
   )
 
   private val extendedConfig = Config[KafkaSourceConfig, KafkaSinkConfig](
@@ -217,6 +218,7 @@ object KafkaConfigSpec {
     legacyColumns = List(
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/1-*-*").get,
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/2-*-*").get
-    )
+    ),
+    exitOnMissingIgluSchema = true
   )
 }
diff --git a/...es/kinesis/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KinesisConfigSpec.scala b/...es/kinesis/src/test/scala/com/snowplowanalytics/snowplow/bigquery/KinesisConfigSpec.scala
@@ -121,9 +121,10 @@ object KinesisConfigSpec {
       healthProbe = Config.HealthProbe(port = Port.fromInt(8000).get, unhealthyLatency = 5.minutes),
       webhook     = Webhook.Config(endpoint = None, tags = Map.empty, heartbeat = 60.minutes)
     ),
-    license       = AcceptedLicense(),
-    skipSchemas   = List.empty,
-    legacyColumns = List.empty
+    license                 = AcceptedLicense(),
+    skipSchemas             = List.empty,
+    legacyColumns           = List.empty,
+    exitOnMissingIgluSchema = true
   )
 
   // workerIdentifer coming from "HOSTNAME" env variable set in BuildSettings
@@ -212,6 +213,7 @@ object KinesisConfigSpec {
     legacyColumns = List(
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/1-*-*").get,
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/2-*-*").get
-    )
+    ),
+    exitOnMissingIgluSchema = true
   )
 }
diff --git a/modules/pubsub/src/test/scala/com/snowplowanalytics/snowplow/bigquery/PubsubConfigSpec.scala b/modules/pubsub/src/test/scala/com/snowplowanalytics/snowplow/bigquery/PubsubConfigSpec.scala
@@ -118,9 +118,10 @@ object PubsubConfigSpec {
       healthProbe = Config.HealthProbe(port = Port.fromInt(8000).get, unhealthyLatency = 5.minutes),
       webhook     = Webhook.Config(endpoint = None, tags = Map.empty, heartbeat = 60.minutes)
     ),
-    license       = AcceptedLicense(),
-    skipSchemas   = List.empty,
-    legacyColumns = List.empty
+    license                 = AcceptedLicense(),
+    skipSchemas             = List.empty,
+    legacyColumns           = List.empty,
+    exitOnMissingIgluSchema = true
   )
 
   private val extendedConfig = Config[PubsubSourceConfig, PubsubSinkConfig](
@@ -205,6 +206,7 @@ object PubsubConfigSpec {
     legacyColumns = List(
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/1-*-*").get,
       SchemaCriterion.parse("iglu:com.acme/legacy/jsonschema/2-*-*").get
-    )
+    ),
+    exitOnMissingIgluSchema = true
   )
 }