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.

modules/common/src/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,23 @@ object AtomicFields {
 
     AtomicFields(withLimits)
   }
+
+  def atomicErrorsToSchemaViolation(errors: NonEmptyList[AtomicError]): FailureDetails.SchemaViolation = {
+    val messages = errors.map { error =>
+      ValidatorReport(error.message, Some(error.field), Nil, error.value)
+    }
+    val validatorError = ValidatorError.InvalidData(messages)
+    val clientError = ValidationError(validatorError, None)
+
+    FailureDetails.SchemaViolation.IgluError(
+      AtomicFields.atomicSchema,
+      clientError
+    )
+  }
 }
+
+case class AtomicError(
+  field: String,
+  value: Option[String],
+  message: String
+)

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/AtomicFieldsLengthValidator.scala

@@ -14,14 +14,11 @@ 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.snowplow.enrich.common.adapters.RawEvent
+import com.snowplowanalytics.snowplow.badrows.FailureDetails
 import com.snowplowanalytics.snowplow.enrich.common.enrichments.AtomicFields.LimitedAtomicField
 import com.snowplowanalytics.snowplow.enrich.common.outputs.EnrichedEvent
 
@@ -35,66 +32,46 @@ 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.atomicErrorsToSchemaViolation(errors).asLeft)
       case Valid(()) =>
         Monad[F].pure(Right(()))
     }
 
   private def validateField(
-    event: EnrichedEvent
-  )(
+    event: EnrichedEvent,
     atomicField: LimitedAtomicField
-  ): ValidatedNel[String, Unit] = {
+  ): Either[AtomicError, 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
+      AtomicError(
+        atomicField.value.name,
+        Option(actualValue),
+        s"Field is longer than maximum allowed size ${atomicField.limit}"
+      ).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[AtomicError]
   ): 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(_.field).toList.mkString(", ")}"
         )
       )
 
-  private def asEnrichmentFailure(errorMessage: String): EnrichmentFailure =
-    EnrichmentFailure(
-      enrichment = None,
-      FailureDetails.EnrichmentFailureMessage.Simple(errorMessage)
-    )
 }

modules/common/src/main/scala/com.snowplowanalytics.snowplow.enrich/common/enrichments/ClientEnrichments.scala

@@ -14,8 +14,6 @@ import java.lang.{Integer => JInteger}
 
 import cats.syntax.either._
 
-import com.snowplowanalytics.snowplow.badrows._
-
 /**
  * Contains enrichments related to the client - where the client is the software which is using the
  * Snowplow tracker. Enrichments relate to browser resolution.
@@ -36,21 +34,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[AtomicError, (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)
+        AtomicError(field, Option(res), msg)
       }
 
 }