fix: Missing Ingestion Jobs from WebUI Table

components/clp-py-utils/clp_py_utils/initialize-orchestration-db.py

@@ -55,6 +55,7 @@ def main(argv):
                     `status_msg` VARCHAR(512) NOT NULL DEFAULT '',
                     `creation_time` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
                     `start_time` DATETIME(3) NULL DEFAULT NULL,
+                    `update_time` DATETIME(3) NULL DEFAULT NULL,
                     `duration` FLOAT NULL DEFAULT NULL,
                     `original_size` BIGINT NOT NULL DEFAULT '0',
                     `uncompressed_size` BIGINT NOT NULL DEFAULT '0',

components/job-orchestration/job_orchestration/scheduler/compress/compression_scheduler.py

@@ -209,12 +209,14 @@ def search_and_schedule_new_tasks(db_conn, db_cursor, clp_metadata_db_connection
 
         if len(tasks) == 0:
             logger.warning(f"No tasks were created for job {job_id}")
+            update_time = datetime.datetime.now()
             update_compression_job_metadata(
                 db_cursor,
                 job_id,
                 {
                     "status": CompressionJobStatus.FAILED,
                     "status_msg": "invalid input path",
+                    "update_time": update_time
                 },
             )
             db_conn.commit()
@@ -229,6 +231,7 @@ def search_and_schedule_new_tasks(db_conn, db_cursor, clp_metadata_db_connection
                 "num_tasks": paths_to_compress_buffer.num_tasks,
                 "status": CompressionJobStatus.RUNNING,
                 "start_time": start_time,
+                "update_time": start_time
             },
         )
         db_conn.commit()
@@ -324,6 +327,7 @@ def poll_running_jobs(db_conn, db_cursor):
             logger.error(f"Error while getting results for job {job_id}: {e}")
             job_success = False
 
+        update_time = datetime.datetime.now()
         if job_success:
             logger.info(f"Job {job_id} succeeded.")
             update_compression_job_metadata(
@@ -332,6 +336,7 @@ def poll_running_jobs(db_conn, db_cursor):
                 dict(
                     status=CompressionJobStatus.SUCCEEDED,
                     duration=duration,
+                    update_time=update_time
                 ),
             )
         else:
@@ -342,6 +347,7 @@ def poll_running_jobs(db_conn, db_cursor):
                 dict(
                     status=CompressionJobStatus.FAILED,
                     status_msg=error_message,
+                    update_time=update_time
                 ),
             )
         db_conn.commit()

components/webui/imports/api/ingestion/constants.js

@@ -10,6 +10,7 @@ const COMPRESSION_JOBS_TABLE_COLUMN_NAMES = Object.freeze({
     STATUS_MSG: "status_msg",
     CREATION_TIME: "creation_time",
     START_TIME: "start_time",
+    UPDATE_TIME: "update_time",
     DURATION: "duration",
     ORIGINAL_SIZE: "original_size",
     UNCOMPRESSED_SIZE: "uncompressed_size",

components/webui/imports/api/ingestion/server/CompressionDbManager.js

@@ -23,12 +23,12 @@ class CompressionDbManager {
      * Retrieves the last `limit` number of jobs and the ones with the given
      * job IDs.
      *
-     * @param {number} limit
+     * @param {string} lastUpdateDate
      * @param {number[]} jobIds
      * @return {Promise<object[]>} Job objects with fields with the names in
      * `COMPRESSION_JOBS_TABLE_COLUMN_NAMES`
      */
-    async getCompressionJobs (limit, jobIds) {
+    async getCompressionJobs (lastUpdateDate, jobIds) {
         const queries = [];
 
         queries.push(`
@@ -38,6 +38,7 @@ class CompressionDbManager {
                     ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.STATUS},
                     ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.STATUS_MSG},
                     ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.START_TIME},
+                    ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.UPDATE_TIME},
                     ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.DURATION},
                     ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.UNCOMPRESSED_SIZE},
                     ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.COMPRESSED_SIZE}
@@ -47,7 +48,6 @@ class CompressionDbManager {
                 SELECT *
                 FROM SelectedColumns
                 ORDER BY _id DESC
-                LIMIT ${limit}
             )
         `);
 
@@ -59,7 +59,9 @@ class CompressionDbManager {
                 (
                     SELECT *
                     FROM SelectedColumns
-                    WHERE _id=${jobId}
+                    WHERE 
+                        _id=${jobId} && 
+                        ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.UPDATE_TIME} >= '${lastUpdateDate}'
                 )
             `);
         });

components/webui/imports/api/ingestion/server/publications.js

@@ -18,13 +18,10 @@ import StatsDbManager from "./StatsDbManager";
 
 const COMPRESSION_JOBS_REFRESH_INTERVAL_MILLIS = 1000;
 
-/**
- * The maximum number of compression jobs to retrieve at a time.
- */
-const COMPRESSION_MAX_RETRIEVE_JOBS = 5;
-
 const STATS_REFRESH_INTERVAL_MILLIS = 5000;
 
+const CONST_FOR_DATE_FORMAT = 19;
+
 /**
  * @type {CompressionDbManager|null}
  */
@@ -45,6 +42,13 @@ let compressionJobsRefreshTimeout = null;
  */
 let statsRefreshInterval = null;
 
+/**
+ * @type {string}
+ */
+let lastUpdateDate = new Date().toISOString()
+    .slice(0, CONST_FOR_DATE_FORMAT)
+    .replace("T", " ");
+
 /**
  * Updates the compression statistics in the StatsCollection.
  *
@@ -99,11 +103,18 @@ const refreshCompressionJobs = async () => {
             job._id
         ));
 
+    //  Save timestamp before querying the combressionDBManager
+    const newDate = new Date().toISOString()
+        .slice(0, CONST_FOR_DATE_FORMAT)
+        .replace("T", " ");
+
     const jobs = await compressionDbManager.getCompressionJobs(
-        COMPRESSION_MAX_RETRIEVE_JOBS,
+        lastUpdateDate,
         pendingJobIds
     );
 
+    lastUpdateDate = newDate;
+
     const operations = jobs.map((doc) => ({
         updateOne: {
             filter: {_id: doc._id},

components/webui/imports/ui/IngestView/panels/IngestionJobs/IngestionJobRow.jsx

@@ -106,6 +106,13 @@ const IngestionJobRow = ({job}) => {
                     isAlwaysVisible={isPlaceholderVisible}
                     text={compressedSizeText}/>
             </td>
+            <td className={"text-end"}>
+                <PlaceholderText
+                    isAlwaysVisible={isPlaceholderVisible}
+                    text={(null === job.update_time) ?
+                        "null" :
+                        new Date(job.update_time).toLocaleString()}/>
+            </td>
         </tr>
     );
 };

components/webui/imports/ui/IngestView/panels/IngestionJobs/index.jsx

@@ -47,6 +47,7 @@ const IngestionJobs = () => {
                         <th className={"text-end"}>Speed</th>
                         <th className={"text-end"}>Data Ingested</th>
                         <th className={"text-end"}>Compressed Size</th>
+                        <th className={"text-end"}>Last Updated</th>
                     </tr>
                 </thead>
                 <tbody>