webui: Add table to display the status of recent ingestion jobs.

components/clp-package-utils/clp_package_utils/scripts/start_clp.py

@@ -16,6 +16,7 @@
     ALL_TARGET_NAME,
     CLP_METADATA_TABLE_PREFIX,
     CLPConfig,
+    COMPRESSION_JOBS_TABLE_NAME,
     COMPRESSION_SCHEDULER_COMPONENT_NAME,
     COMPRESSION_WORKER_COMPONENT_NAME,
     CONTROLLER_TARGET_NAME,
@@ -685,9 +686,10 @@ def start_webui(instance_id: str, clp_config: CLPConfig, mounts: CLPDockerMounts
             "SqlDbHost": clp_config.database.host,
             "SqlDbPort": clp_config.database.port,
             "SqlDbName": clp_config.database.name,
-            "SqlDbSearchJobsTableName": SEARCH_JOBS_TABLE_NAME,
             "SqlDbClpArchivesTableName": f"{CLP_METADATA_TABLE_PREFIX}archives",
             "SqlDbClpFilesTableName": f"{CLP_METADATA_TABLE_PREFIX}files",
+            "SqlDbCompressionJobsTableName": COMPRESSION_JOBS_TABLE_NAME,
+            "SqlDbSearchJobsTableName": SEARCH_JOBS_TABLE_NAME,
         },
         "public": {
             "ClpStorageEngine": clp_config.package.storage_engine,

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

@@ -12,7 +12,12 @@ const STATS_COLLECTION_ID = Object.freeze({
 
 const StatsCollection = new Mongo.Collection(Meteor.settings.public.StatsCollectionName);
 
+const CompressionJobsCollection = new Mongo.Collection(
+    Meteor.settings.public.CompressionJobsCollectionName
+);
+
 export {
+    CompressionJobsCollection,
     STATS_COLLECTION_ID,
     StatsCollection,
 };

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

@@ -0,0 +1,71 @@
+/* eslint-disable sort-keys */
+/**
+ * Enum of the column names for the `compression_jobs` table.
+ *
+ * @enum {string}
+ */
+const COMPRESSION_JOBS_TABLE_COLUMN_NAMES = Object.freeze({
+    ID: "id",
+    STATUS: "status",
+    STATUS_MSG: "status_msg",
+    CREATION_TIME: "creation_time",
+    START_TIME: "start_time",
+    DURATION: "duration",
+    ORIGINAL_SIZE: "original_size",
+    UNCOMPRESSED_SIZE: "uncompressed_size",
+    COMPRESSED_SIZE: "compressed_size",
+    NUM_TASKS: "num_tasks",
+    NUM_TASKS_COMPLETED: "num_tasks_completed",
+    CLP_BINARY_VERSION: "clp_binary_version",
+    CLP_CONFIG: "clp_config",
+});
+/* eslint-enable sort-keys */
+
+let enumCompressionJobStatus;
+/**
+ * Enum of compression job statuses, matching the `CompressionJobStatus` class in
+ * `job_orchestration.scheduler.constants`.
+ *
+ * @enum {string}
+ */
+const COMPRESSION_JOB_STATUS = Object.freeze({
+    PENDING: (enumCompressionJobStatus = 0),
+    RUNNING: ++enumCompressionJobStatus,
+    SUCCEEDED: ++enumCompressionJobStatus,
+    FAILED: ++enumCompressionJobStatus,
+});
+
+/**
+ * List of waiting states for a compression job.
+ *
+ * @see COMPRESSION_JOB_STATUS
+ */
+const COMPRESSION_JOB_WAITING_STATES = Object.freeze([
+    COMPRESSION_JOB_STATUS.PENDING,
+    COMPRESSION_JOB_STATUS.RUNNING,
+]);
+
+/**
+ * Represents the possible status names for a compression job.
+ *
+ * @type {ReadonlyArray<string>}
+ */
+const COMPRESSION_JOB_STATUS_NAMES = Object.freeze([
+    "PENDING",
+    "RUNNING",
+    "SUCCEEDED",
+    "FAILED",
+]);
+
+/**
+ * The maximum number of compression jobs that is retrieved at a time.
+ */
+const COMPRESSION_MAX_RETRIEVE_JOBS = 5;
+
+export {
+    COMPRESSION_JOB_STATUS,
+    COMPRESSION_JOB_STATUS_NAMES,
+    COMPRESSION_JOB_WAITING_STATES,
+    COMPRESSION_JOBS_TABLE_COLUMN_NAMES,
+    COMPRESSION_MAX_RETRIEVE_JOBS,
+};

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

@@ -0,0 +1,72 @@
+import {COMPRESSION_JOBS_TABLE_COLUMN_NAMES} from "../constants";
+
+
+/**
+ * Class for retrieving compression jobs from the database.
+ */
+class CompressionDbManager {
+    #sqlDbConnPool;
+
+    #compressionJobsTableName;
+
+    /**
+     * @param {import("mysql2/promise").Pool} sqlDbConnPool
+     * @param {object} tableNames
+     * @param {string} tableNames.compressionJobsTableName
+     */
+    constructor (sqlDbConnPool, {
+        compressionJobsTableName,
+    }) {
+        this.#sqlDbConnPool = sqlDbConnPool;
+        this.#compressionJobsTableName = compressionJobsTableName;
+    }
+
+    /**
+     * Retrieve the last `limit` number of jobs. Also includes the ones in
+     * `jobIdList`.
+     *
+     * @param {number} limit
+     * @param {number[]} jobIdList
+     * @return {Promise<object[]>}
+     */
+    async getCompressionJobs (limit, jobIdList) {
+        let queryString = `
+            WITH SelectedColumns AS (
+                SELECT 
+                    id as _id, 
+                    ${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.DURATION},
+                    ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.UNCOMPRESSED_SIZE},
+                    ${COMPRESSION_JOBS_TABLE_COLUMN_NAMES.COMPRESSED_SIZE}
+                FROM ${this.#compressionJobsTableName}
+            )
+            (
+                SELECT *
+                FROM SelectedColumns
+                ORDER BY _id DESC
+                LIMIT ${limit}
+            )
+        `;
+
+        jobIdList.forEach((jobId) => {
+            queryString += `
+                UNION DISTINCT
+                (
+                    SELECT *
+                    FROM SelectedColumns
+                    WHERE _id=${jobId}
+                )
+            `;
+        });
+
+        queryString += "ORDER BY _id DESC;";
+
+        const results = await this.#sqlDbConnPool.query(queryString);
+
+        return results[0];
+    }
+}
+
+export default CompressionDbManager;

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

@@ -1,16 +1,30 @@
 import {Meteor} from "meteor/meteor";
 
 import {logger} from "/imports/utils/logger";
+import {MONGO_SORT_BY_ID} from "/imports/utils/mongo";
 
 import {
+    CompressionJobsCollection,
     STATS_COLLECTION_ID,
     StatsCollection,
 } from "../collections";
+import {
+    COMPRESSION_JOB_WAITING_STATES,
+    COMPRESSION_JOBS_TABLE_COLUMN_NAMES,
+    COMPRESSION_MAX_RETRIEVE_JOBS,
+} from "../constants";
+import CompressionDbManager from "./CompressionDbManager";
 import StatsDbManager from "./StatsDbManager";
 
 
+const COMPRESSION_JOBS_REFRESH_INTERVAL_MILLIS = 1000;
 const STATS_REFRESH_INTERVAL_MILLIS = 5000;
 
+/**
+ * @type {CompressionDbManager|null}
+ */
+let compressionDbManager = null;
+
 /**
  * @type {StatsDbManager|null}
  */
@@ -19,7 +33,12 @@ let statsDbManager = null;
 /**
  * @type {number|null}
  */
-let refreshMeteorInterval = null;
+let compressionJobsRefreshTimeout = null;
+
+/**
+ * @type {number|null}
+ */
+let statsRefreshInterval = null;
 
 /**
  * Updates the compression statistics in the StatsCollection.
@@ -45,6 +64,94 @@ const refreshCompressionStats = async () => {
     await StatsCollection.updateAsync(filter, modifier, options);
 };
 
+/**
+ * Updates the compression jobs in the CompressionJobsCollection.
+ *
+ * @return {Promise<void>}
+ */
+const refreshCompressionJobs = async () => {
+    if (null !== compressionJobsRefreshTimeout) {
+        Meteor.clearTimeout(compressionJobsRefreshTimeout);
+        compressionJobsRefreshTimeout = null;
+    }
+    if (0 === Meteor.server.stream_server.all_sockets().length) {
+        compressionJobsRefreshTimeout = Meteor.setTimeout(
+            refreshCompressionJobs,
+            COMPRESSION_JOBS_REFRESH_INTERVAL_MILLIS
+        );
+
+        return;
+    }
+
+    const pendingJobIdList = await CompressionJobsCollection.find({
+        [COMPRESSION_JOBS_TABLE_COLUMN_NAMES.STATUS]: {
+            $in: COMPRESSION_JOB_WAITING_STATES,
+        },
+    })
+        .fetch()
+        .map((job) => (
+            job._id
+        ));
+
+    const jobs = await compressionDbManager.getCompressionJobs(
+        COMPRESSION_MAX_RETRIEVE_JOBS,
+        pendingJobIdList
+    );
+
+    const operations = jobs.map((doc) => ({
+        updateOne: {
+            filter: {_id: doc._id},
+            update: {$set: doc},
+            upsert: true,
+        },
+    }));
+
+    if (0 !== operations.length) {
+        await CompressionJobsCollection.rawCollection().bulkWrite(operations);
+    }
+
+    // `refreshCompressionJobs()` shall not be run concurrently and therefore incurs no race
+    // condition.
+    // eslint-disable-next-line require-atomic-updates
+    compressionJobsRefreshTimeout = Meteor.setTimeout(
+        refreshCompressionJobs,
+        COMPRESSION_JOBS_REFRESH_INTERVAL_MILLIS
+    );
+};
+
+/**
+ * Initializes the CompressionDbManager and starts a timeout timer (`compressionJobsRefreshTimeout`)
+ * for compression job updates.
+ *
+ * @param {import("mysql2/promise").Pool} sqlDbConnPool
+ * @param {object} tableNames
+ * @param {string} tableNames.compressionJobsTableName
+ * @throws {Error} on error.
+ */
+const initCompressionDbManager = (sqlDbConnPool, {
+    compressionJobsTableName,
+}) => {
+    compressionDbManager = new CompressionDbManager(sqlDbConnPool, {
+        compressionJobsTableName,
+    });
+
+    compressionJobsRefreshTimeout = Meteor.setTimeout(
+        refreshCompressionJobs,
+        COMPRESSION_JOBS_REFRESH_INTERVAL_MILLIS
+    );
+};
+
+/**
+ * De-initializes the CompressionDbManager by clearing the timeout timer for compression job
+ * updates (`refreshCompressionJobs`).
+ */
+const deinitCompressionDbManager = () => {
+    if (null !== compressionJobsRefreshTimeout) {
+        Meteor.clearTimeout(compressionJobsRefreshTimeout);
+        compressionJobsRefreshTimeout = null;
+    }
+};
+
 /**
  * Initializes the StatsDbManager and starts an interval timer (`refreshMeteorInterval`) for
  * compression stats updates.
@@ -64,7 +171,7 @@ const initStatsDbManager = (sqlDbConnPool, {
         clpFilesTableName,
     });
 
-    refreshMeteorInterval = Meteor.setInterval(
+    statsRefreshInterval = Meteor.setInterval(
         refreshCompressionStats,
         STATS_REFRESH_INTERVAL_MILLIS
     );
@@ -75,12 +182,30 @@ const initStatsDbManager = (sqlDbConnPool, {
  * (`refreshMeteorInterval`).
  */
 const deinitStatsDbManager = () => {
-    if (null !== refreshMeteorInterval) {
-        Meteor.clearInterval(refreshMeteorInterval);
-        refreshMeteorInterval = null;
+    if (null !== statsRefreshInterval) {
+        Meteor.clearInterval(statsRefreshInterval);
+        statsRefreshInterval = null;
     }
 };
 
+/**
+ * Updates and publishes compression job statuses.
+ *
+ * @param {string} publicationName
+ * @return {Mongo.Cursor}
+ */
+Meteor.publish(Meteor.settings.public.CompressionJobsCollectionName, async () => {
+    logger.debug(`Subscription '${Meteor.settings.public.CompressionJobsCollectionName}'`);
+
+    await refreshCompressionJobs();
+
+    const findOptions = {
+        sort: [MONGO_SORT_BY_ID],
+    };
+
+    return CompressionJobsCollection.find({}, findOptions);
+});
+
 /**
  * Updates and publishes compression statistics.
  *
@@ -100,6 +225,8 @@ Meteor.publish(Meteor.settings.public.StatsCollectionName, async () => {
 });
 
 export {
+    deinitCompressionDbManager,
     deinitStatsDbManager,
+    initCompressionDbManager,
     initStatsDbManager,
 };

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

@@ -1,10 +1,13 @@
 import {Meteor} from "meteor/meteor";
 
 import {logger} from "/imports/utils/logger";
+import {
+    MONGO_SORT_BY_ID,
+    MONGO_SORT_ORDER,
+} from "/imports/utils/mongo";
 
 import {SearchResultsMetadataCollection} from "../collections";
 import {
-    MONGO_SORT_ORDER,
     SEARCH_MAX_NUM_RESULTS,
     SEARCH_RESULTS_FIELDS,
 } from "../constants";
@@ -58,7 +61,7 @@ Meteor.publish(Meteor.settings.public.SearchResultsCollectionName, ({
         sort: [
             /* eslint-disable @stylistic/js/array-element-newline */
             [SEARCH_RESULTS_FIELDS.TIMESTAMP, MONGO_SORT_ORDER.DESCENDING],
-            [SEARCH_RESULTS_FIELDS.ID, MONGO_SORT_ORDER.DESCENDING],
+            MONGO_SORT_BY_ID,
             /* eslint-enable @stylistic/js/array-element-newline */
         ],
         limit: SEARCH_MAX_NUM_RESULTS,