Switch bad row type from EnrichmentFailures to SchemaViolations for s…

…ome errors There are 3 places that currently produce an EnrichmentFailures bad row whereas the more appropriate type is SchemaViolations: 1. When the input fields of the HTTP requests are mapped to the atomic event. 2. When the enrichments contexts get validated. 3. When the atomic fields lengths get validated. For 1 and 3, the error should be mapped into an Iglu ValidationError with the atomic schema referenced. Before this change, if there was any error in the mapping of the atomic fields, a bad row would get emitted right away and we would not try to validate the entities and unstruct event. Now all these errors are wrapped inside a same SchemaViolations bad row. Likewise, before this change when an enrichment context was invalid, we were emitting a bad row right away and not checking the lengths of the atomic fields. Now all these errors are wrapped inside a same SchemaViolations bad row.
snowplow · Mar 8, 2024 · 70fd56d · 70fd56d
1 parent 02b20da
commit 70fd56d
Show file tree

Hide file tree

Showing 22 changed files with 546 additions and 583 deletions.
diff --git a/modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/EtlPipeline.scala b/modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/EtlPipeline.scala
@@ -10,9 +10,8 @@
  */
 package com.snowplowanalytics.snowplow.enrich.common
 
-import cats.Monad
 import cats.data.{Validated, ValidatedNel}
-import cats.effect.Clock
+import cats.effect.kernel.Sync
 import cats.implicits._
 
 import org.joda.time.DateTime
@@ -56,7 +55,7 @@ object EtlPipeline {
    * @return the ValidatedMaybeCanonicalOutput. Thanks to flatMap, will include any validation
    * errors contained within the ValidatedMaybeCanonicalInput
    */
-  def processEvents[F[_]: Clock: Monad](
+  def processEvents[F[_]: Sync](
     adapterRegistry: AdapterRegistry[F],
     enrichmentRegistry: EnrichmentRegistry[F],
     client: IgluCirceClient[F],
@@ -90,11 +89,11 @@ object EtlPipeline {
                   .toValidated
               }
             case Validated.Invalid(badRow) =>
-              Monad[F].pure(List(badRow.invalid[EnrichedEvent]))
+              Sync[F].pure(List(badRow.invalid[EnrichedEvent]))
           }
       case Validated.Invalid(badRows) =>
-        Monad[F].pure(badRows.map(_.invalid[EnrichedEvent])).map(_.toList)
+        Sync[F].pure(badRows.map(_.invalid[EnrichedEvent])).map(_.toList)
       case Validated.Valid(None) =>
-        Monad[F].pure(List.empty[Validated[BadRow, EnrichedEvent]])
+        Sync[F].pure(List.empty[Validated[BadRow, EnrichedEvent]])
     }
 }
diff --git a/...rc/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/AtomicFields.scala b/...rc/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/AtomicFields.scala
@@ -10,13 +10,24 @@
  */
 package com.snowplowanalytics.snowplow.enrich.common.enrichments
 
+import cats.data.NonEmptyList
+
+import com.snowplowanalytics.snowplow.badrows.FailureDetails
+
+import com.snowplowanalytics.iglu.client.ClientError.ValidationError
+import com.snowplowanalytics.iglu.client.validator.{ValidatorError, ValidatorReport}
+
+import com.snowplowanalytics.iglu.core.{SchemaKey, SchemaVer}
+
 import com.snowplowanalytics.snowplow.enrich.common.enrichments.AtomicFields.LimitedAtomicField
 import com.snowplowanalytics.snowplow.enrich.common.outputs.EnrichedEvent
 
 final case class AtomicFields(value: List[LimitedAtomicField])
 
 object AtomicFields {
 
+  val atomicSchema = SchemaKey("com.snowplowanalytics.snowplow", "atomic", "jsonschema", SchemaVer.Full(1, 0, 0))
+
   final case class AtomicField(
     name: String,
     enrichedValueExtractor: EnrichedEvent => String
@@ -121,4 +132,13 @@ object AtomicFields {
 
     AtomicFields(withLimits)
   }
+
+  def errorsToSchemaViolation(errors: NonEmptyList[ValidatorReport]): FailureDetails.SchemaViolation = {
+    val clientError = ValidationError(ValidatorError.InvalidData(errors), None)
+
+    FailureDetails.SchemaViolation.IgluError(
+      AtomicFields.atomicSchema,
+      clientError
+    )
+  }
 }
diff --git a/...om.snowplowanalytics.snowplow.enrich/common/enrichments/AtomicFieldsLengthValidator.scala b/...om.snowplowanalytics.snowplow.enrich/common/enrichments/AtomicFieldsLengthValidator.scala
@@ -14,14 +14,14 @@ import org.slf4j.LoggerFactory
 
 import cats.Monad
 import cats.data.Validated.{Invalid, Valid}
-import cats.data.{NonEmptyList, ValidatedNel}
+import cats.data.NonEmptyList
 
 import cats.implicits._
 
-import com.snowplowanalytics.snowplow.badrows.FailureDetails.EnrichmentFailure
-import com.snowplowanalytics.snowplow.badrows.{BadRow, FailureDetails, Processor}
+import com.snowplowanalytics.iglu.client.validator.ValidatorReport
+
+import com.snowplowanalytics.snowplow.badrows.FailureDetails
 
-import com.snowplowanalytics.snowplow.enrich.common.adapters.RawEvent
 import com.snowplowanalytics.snowplow.enrich.common.enrichments.AtomicFields.LimitedAtomicField
 import com.snowplowanalytics.snowplow.enrich.common.outputs.EnrichedEvent
 
@@ -35,66 +35,47 @@ object AtomicFieldsLengthValidator {
 
   def validate[F[_]: Monad](
     event: EnrichedEvent,
-    rawEvent: RawEvent,
-    processor: Processor,
     acceptInvalid: Boolean,
     invalidCount: F[Unit],
     atomicFields: AtomicFields
-  ): F[Either[BadRow, Unit]] =
+  ): F[Either[FailureDetails.SchemaViolation, Unit]] =
     atomicFields.value
-      .map(validateField(event))
+      .map(field => validateField(event, field).toValidatedNel)
       .combineAll match {
       case Invalid(errors) if acceptInvalid =>
-        handleAcceptableBadRow(invalidCount, event, errors) *> Monad[F].pure(Right(()))
+        handleAcceptableErrors(invalidCount, event, errors) *> Monad[F].pure(Right(()))
       case Invalid(errors) =>
-        Monad[F].pure(buildBadRow(event, rawEvent, processor, errors).asLeft)
+        Monad[F].pure(AtomicFields.errorsToSchemaViolation(errors).asLeft)
       case Valid(()) =>
         Monad[F].pure(Right(()))
     }
 
   private def validateField(
-    event: EnrichedEvent
-  )(
+    event: EnrichedEvent,
     atomicField: LimitedAtomicField
-  ): ValidatedNel[String, Unit] = {
+  ): Either[ValidatorReport, Unit] = {
     val actualValue = atomicField.value.enrichedValueExtractor(event)
     if (actualValue != null && actualValue.length > atomicField.limit)
-      s"Field ${atomicField.value.name} longer than maximum allowed size ${atomicField.limit}".invalidNel
+      ValidatorReport(
+        s"Field is longer than maximum allowed size ${atomicField.limit}",
+        Some(atomicField.value.name),
+        Nil,
+        Some(actualValue)
+      ).asLeft
     else
-      Valid(())
+      Right(())
   }
 
-  private def buildBadRow(
-    event: EnrichedEvent,
-    rawEvent: RawEvent,
-    processor: Processor,
-    errors: NonEmptyList[String]
-  ): BadRow.EnrichmentFailures =
-    EnrichmentManager.buildEnrichmentFailuresBadRow(
-      NonEmptyList(
-        asEnrichmentFailure("Enriched event does not conform to atomic schema field's length restrictions"),
-        errors.toList.map(asEnrichmentFailure)
-      ),
-      EnrichedEvent.toPartiallyEnrichedEvent(event),
-      RawEvent.toRawEvent(rawEvent),
-      processor
-    )
-
-  private def handleAcceptableBadRow[F[_]: Monad](
+  private def handleAcceptableErrors[F[_]: Monad](
     invalidCount: F[Unit],
     event: EnrichedEvent,
-    errors: NonEmptyList[String]
+    errors: NonEmptyList[ValidatorReport]
   ): F[Unit] =
     invalidCount *>
       Monad[F].pure(
         logger.debug(
-          s"Enriched event not valid against atomic schema. Event id: ${event.event_id}. Invalid fields: ${errors.toList.mkString(",")}"
+          s"Enriched event not valid against atomic schema. Event id: ${event.event_id}. Invalid fields: ${errors.map(_.path).toList.flatten.mkString(", ")}"
         )
       )
 
-  private def asEnrichmentFailure(errorMessage: String): EnrichmentFailure =
-    EnrichmentFailure(
-      enrichment = None,
-      FailureDetails.EnrichmentFailureMessage.Simple(errorMessage)
-    )
 }
diff --git a/...in/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/ClientEnrichments.scala b/...in/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/ClientEnrichments.scala
@@ -14,7 +14,7 @@ import java.lang.{Integer => JInteger}
 
 import cats.syntax.either._
 
-import com.snowplowanalytics.snowplow.badrows._
+import com.snowplowanalytics.iglu.client.validator.ValidatorReport
 
 /**
  * Contains enrichments related to the client - where the client is the software which is using the
@@ -36,21 +36,16 @@ object ClientEnrichments {
    * @param res The packed string holding the screen dimensions
    * @return the ResolutionTuple or an error message, boxed in a Scalaz Validation
    */
-  val extractViewDimensions: (String, String) => Either[FailureDetails.EnrichmentFailure, (JInteger, JInteger)] =
+  val extractViewDimensions: (String, String) => Either[ValidatorReport, (JInteger, JInteger)] =
     (field, res) =>
       (res match {
         case ResRegex(width, height) =>
           Either
             .catchNonFatal((width.toInt: JInteger, height.toInt: JInteger))
-            .leftMap(_ => "could not be converted to java.lang.Integer s")
-        case _ => s"does not conform to regex ${ResRegex.toString}".asLeft
+            .leftMap(_ => "Could not be converted to java.lang.Integer s")
+        case _ => s"Does not conform to regex ${ResRegex.toString}".asLeft
       }).leftMap { msg =>
-        val f = FailureDetails.EnrichmentFailureMessage.InputData(
-          field,
-          Option(res),
-          msg
-        )
-        FailureDetails.EnrichmentFailure(None, f)
+        ValidatorReport(msg, Some(field), Nil, Option(res))
       }
 
 }