Feature/spline 669 filter failed executions

consumer-services/src/main/scala/za/co/absa/spline/consumer/service/model/WriteEventInfo.scala

@@ -32,8 +32,10 @@ case class WriteEventInfo
   applicationId: String,
   @ApiModelProperty(value = "When the execution was triggered")
   timestamp: WriteEventInfo.Timestamp,
-  @ApiModelProperty(value = "When the execution was triggered")
-  durationNs: WriteEventInfo.DurationNs,
+  @ApiModelProperty(value = "Duration of execution in nanoseconds (for successful executions)")
+  durationNs: Option[WriteEventInfo.DurationNs],
+  @ApiModelProperty(value = "Error (for failed executions)")
+  error: Option[WriteEventInfo.Error],
   @ApiModelProperty(value = "Output data source name")
   dataSourceName: String,
   @ApiModelProperty(value = "Output data source URI")
@@ -43,13 +45,14 @@ case class WriteEventInfo
   @ApiModelProperty(value = "Write mode - (true=Append; false=Override)")
   append: WriteEventInfo.Append
 ) {
-  def this() = this(null, null, null, null, null, null, null, null, null, null, null)
+  def this() = this(null, null, null, null, null, null, null, null, null, null, null, null)
 }
 
 object WriteEventInfo {
   type Id = String
   type Timestamp = java.lang.Long
-  type DurationNs = Option[Progress.JobDurationInNanos]
+  type DurationNs = Progress.JobDurationInNanos
+  type Error = Any
   type Append = java.lang.Boolean
 }
 

consumer-services/src/main/scala/za/co/absa/spline/consumer/service/repo/DataSourceRepositoryImpl.scala

@@ -156,7 +156,8 @@ class DataSourceRepositoryImpl @Autowired()(db: ArangoDatabaseAsync) extends Dat
         |        "dataSourceUri"    : ds.uri,
         |        "dataSourceType"   : lwe.execPlanDetails.dataSourceType,
         |        "append"           : lwe.execPlanDetails.append,
-        |        "durationNs"       : lwe.durationNs
+        |        "durationNs"       : lwe.durationNs,
+        |        "error"            : lwe.error
         |    }
         |
         |    SORT resItem.@sortField @sortOrder

consumer-services/src/main/scala/za/co/absa/spline/consumer/service/repo/ExecutionEventRepositoryImpl.scala

@@ -115,7 +115,8 @@ class ExecutionEventRepositoryImpl @Autowired()(db: ArangoDatabaseAsync) extends
         |        "dataSourceUri"    : ee.execPlanDetails.dataSourceUri,
         |        "dataSourceType"   : ee.execPlanDetails.dataSourceType,
         |        "append"           : ee.execPlanDetails.append,
-        |        "durationNs"       : ee.durationNs
+        |        "durationNs"       : ee.durationNs,
+        |        "error"            : ee.error
         |    }
         |
         |    SORT resItem.@sortField @sortOrder

persistence/src/main/resources/foxx/spline/services/observed-writes-by-read.js

@@ -33,6 +33,7 @@ function observedWritesByRead(readEvent) {
                     FILTER !wo.append
                     FOR e IN 2 INBOUND wo executes, progressOf
                         FILTER e.timestamp < readTime
+                           AND e.error == null
                         SORT e.timestamp DESC
                         LIMIT 1
                         RETURN e
@@ -43,6 +44,7 @@ function observedWritesByRead(readEvent) {
                     FOR e IN 2 INBOUND wo executes, progressOf
                         FILTER e.timestamp > maybeObservedOverwrite[0].timestamp
                            AND e.timestamp < readTime
+                           AND e.error == null
                         SORT e.timestamp ASC
                         RETURN e
                 )

persistence/src/main/resources/migration-scripts/0.7.0-1.0.0.js

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2021 ABSA Group Limited
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+const VER = "1.0.0"
+
+const {db, aql} = require("@arangodb");
+
+console.log(`[Spline] Start migration to ${VER}`);
+
+console.log("[Spline] Create index 'progress.durationNs'");
+db.progress.ensureIndex({type: "persistent", fields: ["durationNs"]});
+
+console.log(`[Spline] Migration done. Version ${VER}`);

persistence/src/main/scala/za/co/absa/spline/persistence/model/persistentDefs.scala

@@ -177,6 +177,7 @@ object NodeDef {
   object Progress extends NodeDef("progress") with CollectionDef {
     override def indexDefs: Seq[IndexDef] = Seq(
       IndexDef(Seq("timestamp"), new PersistentIndexOptions),
+      IndexDef(Seq("durationNs"), new PersistentIndexOptions),
       IndexDef(Seq("_created"), new PersistentIndexOptions),
       IndexDef(Seq("extra.appId"), new PersistentIndexOptions().sparse(true)),
       IndexDef(Seq("execPlanDetails.executionPlanKey"), new PersistentIndexOptions),

producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/ProducerAPI.scala

@@ -20,10 +20,11 @@ import za.co.absa.commons.version.Version
 import za.co.absa.commons.version.Version._
 
 object ProducerAPI {
-  val CurrentVersion: Version = ver"1.1"
+  val CurrentVersion: Version = ver"1.2"
   val DeprecatedVersions: Seq[Version] = Seq(ver"1" /*, ...*/)
-  val LTSVersions: Seq[Version] = Seq(CurrentVersion /*, ...*/)
+  val LTSVersions: Seq[Version] = Seq(CurrentVersion, ver"1.1" /*, ...*/)
   val SupportedVersions: Seq[Version] = LTSVersions ++ DeprecatedVersions
 
   final val MimeTypeV1_1 = "application/vnd.absa.spline.producer.v1.1+json"
+  final val MimeTypeV1_2 = "application/vnd.absa.spline.producer.v1.2+json"
 }

producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/controller/ExecutionEventsController.scala

@@ -28,7 +28,10 @@ import za.co.absa.spline.producer.service.repo.ExecutionProducerRepository
 import scala.concurrent.{ExecutionContext, Future}
 
 @Controller
-@RequestMapping(consumes = Array(ProducerAPI.MimeTypeV1_1))
+@RequestMapping(consumes = Array(
+  ProducerAPI.MimeTypeV1_1,
+  ProducerAPI.MimeTypeV1_2,
+))
 @Api(tags = Array("execution"))
 class ExecutionEventsController @Autowired()(
   val repo: ExecutionProducerRepository) {

producer-rest-core/src/main/scala/za/co/absa/spline/producer/rest/controller/ExecutionPlansController.scala

@@ -29,7 +29,10 @@ import za.co.absa.spline.producer.service.repo.ExecutionProducerRepository
 import scala.concurrent.{ExecutionContext, Future}
 
 @RestController
-@RequestMapping(consumes = Array(ProducerAPI.MimeTypeV1_1))
+@RequestMapping(consumes = Array(
+  ProducerAPI.MimeTypeV1_1,
+  ProducerAPI.MimeTypeV1_2,
+))
 @Api(tags = Array("execution"))
 class ExecutionPlansController @Autowired()(
   val repo: ExecutionProducerRepository) {