Fix the sort to use owner ID (#190)

VEuPathDB · Jan 22, 2024 · 059441a · 059441a
1 parent 8b58fd5
commit 059441a
Show file tree

Hide file tree

Showing 15 changed files with 317 additions and 209 deletions.
diff --git a/components/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/AppDBAccessor.kt b/components/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/AppDBAccessor.kt
@@ -3,7 +3,7 @@ package org.veupathdb.vdi.lib.db.app
 import org.veupathdb.vdi.lib.common.field.DatasetID
 import org.veupathdb.vdi.lib.common.field.ProjectID
 import org.veupathdb.vdi.lib.common.field.UserID
-import org.veupathdb.vdi.lib.common.model.VDIDatasetType
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import org.veupathdb.vdi.lib.db.app.model.*
@@ -84,7 +84,7 @@ interface AppDBAccessor {
    * @return Stream of dataset control records sorted by user ID and then dataset ID. The stream
    * must be closed to release the db connection.
    */
-  fun streamAllSyncControlRecords(): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>>
+  fun streamAllSyncControlRecords(): CloseableIterator<VDIReconcilerTargetRecord>
 
   /**
    * Retrieves all datasets that have an install message record for the given

diff --git a/components/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/AppDBAccessorImpl.kt b/components/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/AppDBAccessorImpl.kt
@@ -4,7 +4,7 @@ import org.slf4j.LoggerFactory
 import org.veupathdb.vdi.lib.common.field.DatasetID
 import org.veupathdb.vdi.lib.common.field.ProjectID
 import org.veupathdb.vdi.lib.common.field.UserID
-import org.veupathdb.vdi.lib.common.model.VDIDatasetType
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import org.veupathdb.vdi.lib.db.app.model.*
@@ -55,7 +55,7 @@ internal class AppDBAccessorImpl(
     return con.use { it.selectDatasetProjectLinks(schema, datasetID) }
   }
 
-  override fun streamAllSyncControlRecords(): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+  override fun streamAllSyncControlRecords(): CloseableIterator<VDIReconcilerTargetRecord> {
     log.debug("Streaming all sync control records")
     return con.selectAllSyncControl(schema)
   }

diff --git a/components/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/AppDBTransactionImpl.kt b/components/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/AppDBTransactionImpl.kt
@@ -4,7 +4,7 @@ import org.slf4j.LoggerFactory
 import org.veupathdb.vdi.lib.common.field.DatasetID
 import org.veupathdb.vdi.lib.common.field.ProjectID
 import org.veupathdb.vdi.lib.common.field.UserID
-import org.veupathdb.vdi.lib.common.model.VDIDatasetType
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import org.veupathdb.vdi.lib.db.app.model.*
@@ -198,7 +198,7 @@ class AppDBTransactionImpl(
     return connection.selectDatasetProjectLinks(schema, datasetID)
   }
 
-  override fun streamAllSyncControlRecords(): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+  override fun streamAllSyncControlRecords(): CloseableIterator<VDIReconcilerTargetRecord> {
     return connection.selectAllSyncControl(schema)
   }
 

diff --git a/...onents/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/sql/select-all-sync-control.kt b/...onents/app-db/src/main/kotlin/org/veupathdb/vdi/lib/db/app/sql/select-all-sync-control.kt
@@ -1,8 +1,9 @@
 package org.veupathdb.vdi.lib.db.app.sql
 
 import org.veupathdb.vdi.lib.common.field.DatasetID
-import org.veupathdb.vdi.lib.common.model.VDIDatasetType
+import org.veupathdb.vdi.lib.common.field.UserID
 import org.veupathdb.vdi.lib.common.model.VDIDatasetTypeImpl
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import java.sql.Connection
@@ -20,39 +21,41 @@ SELECT
 , s.dataset_id
 , d.type_name
 , d.type_version
+, d.owner
 FROM
   ${schema}.sync_control s
   INNER JOIN ${schema}.dataset d
     ON d.dataset_id = s.dataset_id
 ORDER BY UPPER(CONCAT(CONCAT(d.owner,'/'), s.dataset_id))
 """
 
-internal fun Connection.selectAllSyncControl(schema: String): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+internal fun Connection.selectAllSyncControl(schema: String): CloseableIterator<VDIReconcilerTargetRecord> {
   val ps = prepareStatement(sql(schema))
   val rs = ps.executeQuery()
   return RecordIterator(rs, this, ps)
 }
 
 class RecordIterator(val rs: ResultSet,
                      private val con: Connection,
-                     private val ps: PreparedStatement): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+                     private val ps: PreparedStatement): CloseableIterator<VDIReconcilerTargetRecord> {
 
   override fun hasNext(): Boolean {
     return rs.next();
   }
 
-  override fun next(): Pair<VDIDatasetType, VDISyncControlRecord> {
-    return Pair(
-      VDIDatasetTypeImpl(
+  override fun next(): VDIReconcilerTargetRecord {
+    return VDIReconcilerTargetRecord(
+      type=VDIDatasetTypeImpl(
         name = rs.getString("type_name"),
         version = rs.getString("type_version")
-      ), VDISyncControlRecord(
+      ),
+      owner = UserID(rs.getLong("owner")),
+      syncControlRecord = VDISyncControlRecord(
         datasetID = DatasetID(rs.getString("dataset_id")),
         sharesUpdated = rs.getObject("shares_update_time", OffsetDateTime::class.java),
         dataUpdated = rs.getObject("data_update_time", OffsetDateTime::class.java),
         metaUpdated = rs.getObject("meta_update_time", OffsetDateTime::class.java)
-      )
-    )
+      ))
   }
 
   override fun close() {

diff --git a/components/cache-db/src/main/kotlin/org/veupathdb/vdi/lib/db/cache/CacheDB.kt b/components/cache-db/src/main/kotlin/org/veupathdb/vdi/lib/db/cache/CacheDB.kt
@@ -7,6 +7,7 @@ import org.veupathdb.vdi.lib.common.env.*
 import org.veupathdb.vdi.lib.common.field.DatasetID
 import org.veupathdb.vdi.lib.common.field.UserID
 import org.veupathdb.vdi.lib.common.model.VDIDatasetType
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import org.veupathdb.vdi.lib.db.cache.model.*
@@ -173,7 +174,7 @@ object CacheDB {
    * @return Stream of dataset control records sorted by user ID and then dataset ID. The stream
    * must be closed to release the db connection.
    */
-  fun selectAllSyncControlRecords(): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+  fun selectAllSyncControlRecords(): CloseableIterator<VDIReconcilerTargetRecord> {
     log.debug("selecting all sync control records")
     return connection.selectAllSyncControl()
   }

diff --git a/...e-db/src/main/kotlin/org/veupathdb/vdi/lib/db/cache/sql/select/select-all-sync-control.kt b/...e-db/src/main/kotlin/org/veupathdb/vdi/lib/db/cache/sql/select/select-all-sync-control.kt
@@ -1,8 +1,10 @@
 package org.veupathdb.vdi.lib.db.cache.sql.select
 
 import org.veupathdb.vdi.lib.common.field.DatasetID
+import org.veupathdb.vdi.lib.common.field.UserID
 import org.veupathdb.vdi.lib.common.model.VDIDatasetType
 import org.veupathdb.vdi.lib.common.model.VDIDatasetTypeImpl
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import java.sql.Connection
@@ -19,31 +21,36 @@ SELECT
 , s.dataset_id
 , d.type_name
 , d.type_version
+, d.owner
 FROM
   vdi.sync_control AS s
   INNER JOIN vdi.datasets AS d
     USING (dataset_id)
 ORDER BY UPPER(CONCAT(d.owner_id,'/',s.dataset_id))
 """
 
-internal fun Connection.selectAllSyncControl(): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+internal fun Connection.selectAllSyncControl(): CloseableIterator<VDIReconcilerTargetRecord> {
   val ps = prepareStatement(SQL)
   val rs = ps.executeQuery()
   return RecordIterator(rs, this, ps)
 }
 
-class RecordIterator(val rs: ResultSet, val connection: Connection, val preparedStatement: PreparedStatement): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+class RecordIterator(val rs: ResultSet,
+                     val connection: Connection,
+                     val preparedStatement: PreparedStatement): CloseableIterator<VDIReconcilerTargetRecord> {
 
   override fun hasNext(): Boolean {
     return rs.next();
   }
 
-  override fun next(): Pair<VDIDatasetType, VDISyncControlRecord> {
-    return Pair(
-      VDIDatasetTypeImpl(
+  override fun next(): VDIReconcilerTargetRecord {
+    return VDIReconcilerTargetRecord(
+      type = VDIDatasetTypeImpl(
         name = rs.getString("type_name"),
         version = rs.getString("type_version")
-      ), VDISyncControlRecord(
+      ),
+      owner = UserID(rs.getLong("owner")),
+      syncControlRecord = VDISyncControlRecord(
         datasetID = DatasetID(rs.getString("dataset_id")),
         sharesUpdated = rs.getObject("shares_update_time", OffsetDateTime::class.java),
         dataUpdated = rs.getObject("data_update_time", OffsetDateTime::class.java),

diff --git a/components/metrics/src/main/kotlin/vdi/component/metrics/Metrics.kt b/components/metrics/src/main/kotlin/vdi/component/metrics/Metrics.kt
@@ -145,10 +145,15 @@ object Metrics {
     .labelNames("target_name")
     .register()
 
+  val missingInTarget: Counter = Counter.build()
+    .name("dataset_reconciler_missing_in_target")
+    .help("Count of datasets the reconciler finds are missing in the target DB.")
+    .labelNames("target_name")
+    .register()
+
   val malformedDatasetFound: Counter = Counter.build()
     .name("malformed_dataset_found")
     .help("A Malformed dataset was found during reconciliation.")
-    .labelNames("target_name")
     .register()
 
   val reconcilerTimes: Histogram = Histogram.build()

diff --git a/components/s3/build.gradle.kts b/components/s3/build.gradle.kts
@@ -5,6 +5,7 @@ plugins {
 dependencies {
   implementation(platform(project(":platform")))
 
+  implementation(project(":components:metrics"))
   implementation(project(":components:constants"))
 
   implementation("org.veupathdb.vdi:vdi-component-common")

diff --git a/components/s3/src/main/kotlin/org/veupathdb/vdi/lib/s3/datasets/DatasetManager.kt b/components/s3/src/main/kotlin/org/veupathdb/vdi/lib/s3/datasets/DatasetManager.kt
@@ -1,13 +1,17 @@
 package org.veupathdb.vdi.lib.s3.datasets
 
+import org.slf4j.LoggerFactory
 import org.veupathdb.lib.s3.s34k.buckets.S3Bucket
 import org.veupathdb.lib.s3.s34k.objects.S3Object
 import org.veupathdb.vdi.lib.common.field.DatasetID
 import org.veupathdb.vdi.lib.common.field.UserID
 import org.veupathdb.vdi.lib.common.field.toUserIDOrNull
+import org.veupathdb.vdi.lib.s3.datasets.exception.MalformedDatasetException
 import org.veupathdb.vdi.lib.s3.datasets.paths.S3DatasetPathFactory
 import org.veupathdb.vdi.lib.s3.datasets.paths.S3PathFactory
 import org.veupathdb.vdi.lib.s3.datasets.paths.S3Paths
+import vdi.component.metrics.Metrics
+
 import java.util.*
 import java.util.stream.Stream
 import java.util.stream.StreamSupport
@@ -75,6 +79,7 @@ class DatasetManager(private val s3Bucket: S3Bucket) {
   private class DatasetDirIterator(
     private val objectStream: Iterator<S3Object>
   ) : Iterator<DatasetDirectory> {
+    private val log = LoggerFactory.getLogger(javaClass)
     private var stagedObjects: List<S3Object> = mutableListOf()
     private var currentDataset: DatasetDirectory? = initFirstDataset()
 
@@ -107,25 +112,35 @@ class DatasetManager(private val s3Bucket: S3Bucket) {
         if (initialDatasetID != currDatasetID) {
           // We've found a new dataset, construct our currentDataset with objects seen so far and reset objects variable
           // to contain the new dataset's object.
-          currentDataset = EagerlyLoadedDatasetDirectory(
-            stagedObjects,
-            initialUserID,
-            initialDatasetID,
-            S3DatasetPathFactory(initialUserID, initialDatasetID)
-          )
+          try {
+            currentDataset = EagerlyLoadedDatasetDirectory(
+              stagedObjects,
+              initialUserID,
+              initialDatasetID,
+              S3DatasetPathFactory(initialUserID, initialDatasetID)
+            )
+          } catch (e: MalformedDatasetException) {
+            Metrics.malformedDatasetFound.inc()
+            log.warn("Found a malformed dataset with ID $initialDatasetID.")
+          }
           // Set staged objects to contain the object belonging to new dataset.
           stagedObjects = mutableListOf(s3Object)
           return currentDataset
         }
         stagedObjects = stagedObjects + s3Object
         if (!objectStream.hasNext()) {
-          // Special handling if there's a single directory in the stream.
-          currentDataset = EagerlyLoadedDatasetDirectory(
-            stagedObjects,
-            initialUserID,
-            initialDatasetID,
-            S3DatasetPathFactory(initialUserID, initialDatasetID)
-          )
+          try {
+            // Special handling if there's a single directory in the stream.
+            currentDataset = EagerlyLoadedDatasetDirectory(
+              stagedObjects,
+              initialUserID,
+              initialDatasetID,
+              S3DatasetPathFactory(initialUserID, initialDatasetID)
+            )
+          } catch (e: MalformedDatasetException) {
+            Metrics.malformedDatasetFound.inc()
+            log.warn("Found a malformed dataset with ID $initialDatasetID.")
+          }
           // Stream is exhausted. Indicate as much.
           stagedObjects = emptyList()
         }

diff --git a/modules/reconciler/src/main/kotlin/org/veupathdb/vdi/lib/reconciler/AppDBTarget.kt b/modules/reconciler/src/main/kotlin/org/veupathdb/vdi/lib/reconciler/AppDBTarget.kt
@@ -2,7 +2,7 @@ package org.veupathdb.vdi.lib.reconciler
 
 import org.veupathdb.vdi.lib.common.field.DatasetID
 import org.veupathdb.vdi.lib.common.model.VDIDatasetType
-import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import org.veupathdb.vdi.lib.db.app.AppDB
 import org.veupathdb.vdi.lib.handler.mapping.PluginHandlers
@@ -15,7 +15,7 @@ class AppDBTarget(
 
     override val type = ReconcilerTargetType.Install
 
-    override fun streamSortedSyncControlRecords(): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+    override fun streamSortedSyncControlRecords(): CloseableIterator<VDIReconcilerTargetRecord> {
         return AppDB.accessor(projectID)!!.streamAllSyncControlRecords()
     }
 

diff --git a/modules/reconciler/src/main/kotlin/org/veupathdb/vdi/lib/reconciler/CacheDBTarget.kt b/modules/reconciler/src/main/kotlin/org/veupathdb/vdi/lib/reconciler/CacheDBTarget.kt
@@ -2,7 +2,7 @@ package org.veupathdb.vdi.lib.reconciler
 
 import org.veupathdb.vdi.lib.common.field.DatasetID
 import org.veupathdb.vdi.lib.common.model.VDIDatasetType
-import org.veupathdb.vdi.lib.common.model.VDISyncControlRecord
+import org.veupathdb.vdi.lib.common.model.VDIReconcilerTargetRecord
 import org.veupathdb.vdi.lib.common.util.CloseableIterator
 import org.veupathdb.vdi.lib.db.cache.CacheDB
 import org.veupathdb.vdi.lib.db.cache.CacheDBTransaction
@@ -12,7 +12,7 @@ class CacheDBTarget : ReconcilerTarget {
 
     override val type = ReconcilerTargetType.Cache
 
-    override fun streamSortedSyncControlRecords(): CloseableIterator<Pair<VDIDatasetType, VDISyncControlRecord>> {
+    override fun streamSortedSyncControlRecords(): CloseableIterator<VDIReconcilerTargetRecord> {
         return CacheDB.selectAllSyncControlRecords()
     }