Merge branch 'main' into harmony-1878-1

bin/stop-harmony-and-services

@@ -2,6 +2,10 @@
 # This script will delete all of the kubernetes harmony resources. If you use this script
 # you will also delete all of your local harmony jobs since the database will be destroyed.
 
+# this will only have an effect if the development services are running - otherwise it does
+# nothing
+bin/stop-dev-services
+
 current_context=$(kubectl config current-context)
 if [ "$current_context" != "colima" ] && [ "$current_context" != "docker-desktop" ] && [ "$current_context" != "minikube" ]; then
   echo 'ERROR: Attempting to use a non-local k8s context while deleting harmony resources.'

db/db.sql

@@ -46,6 +46,25 @@ CREATE TABLE `job_errors` (
   FOREIGN KEY(jobID) REFERENCES jobs(jobID)
 );
 
+CREATE TABLE `labels` (
+  `id` integer not null primary key autoincrement,
+  `username` varchar(255) not null,
+  `value` varchar(255) not null,
+  `createdAt` datetime not null,
+  `updatedAt` datetime not null,
+  UNIQUE(username, value)
+);
+
+CREATE TABLE `jobs_labels` (
+  `id` integer not null primary key autoincrement,
+  `job_id` char(36) not null,
+  `label_id` integer not null,
+  `createdAt` datetime not null,
+  `updatedAt` datetime not null,
+  FOREIGN KEY(job_id) REFERENCES jobs(jobID)
+  FOREIGN KEY(label_id) REFERENCES labels(id)
+);
+
 CREATE TABLE `work_items` (
   `id` integer not null primary key autoincrement,
   `jobID` char(36) not null,

db/migrations/20240925182705_add_labels_table.js

@@ -0,0 +1,30 @@
+/**
+ * @param { import('knex').Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema
+    .createTable('labels', (t) => {
+      t.increments('id').primary();
+      t.string('username', 255).notNullable();
+      t.string('value', 255).notNullable();
+      t.timestamp('createdAt').notNullable();
+      t.timestamp('updatedAt').notNullable();
+      t.unique(['username', 'value']);
+      t.index(['username']);
+      t.index(['value']);
+    }).raw(`
+      ALTER TABLE "labels"
+      ADD CONSTRAINT "lower_case_value"
+      CHECK (value = lower(value))
+    `);
+};
+
+/**
+ * @param { import('knex').Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function(knex) {
+  return knex.schema
+    .dropTable('labels');
+};

db/migrations/20240925182722_add_jobs_labels_table.js

@@ -0,0 +1,39 @@
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.up = function (knex) {
+  return knex.schema.createTable('jobs_labels', (t) => {
+    t.increments('id')
+      .primary();
+
+    t.uuid('job_id')
+      .notNullable()
+      .references('jobID')
+      .inTable('jobs')
+      .onDelete('CASCADE');
+
+    t.integer('label_id')
+      .notNullable()
+      .references('id')
+      .inTable('labels');
+
+    t.timestamp('createdAt')
+      .notNullable();
+
+    t.timestamp('updatedAt')
+      .notNullable();
+
+    t.index(['job_id']);
+    t.index(['label_id']);
+  });
+};
+
+/**
+ * @param { import("knex").Knex } knex
+ * @returns { Promise<void> }
+ */
+exports.down = function(knex) {
+  return knex.schema
+    .dropTable('jobs_labels');
+};

services/harmony/app/backends/service-response.ts

@@ -126,7 +126,7 @@ export async function responseHandler(req: Request, res: Response): Promise<void
 
   const trx = await db.transaction();
 
-  const { job } = await Job.byJobID(trx, requestId, true, false, 1, 1);
+  const { job } = await Job.byJobID(trx, requestId, true, false, false, 1, 1);
   if (!job) {
     await trx.rollback();
     res.status(404);

services/harmony/app/backends/workflow-orchestration/work-item-updates.ts

@@ -863,7 +863,7 @@ export async function processWorkItems(
       const { job } = await (await logAsyncExecutionTime(
         Job.byJobID,
         'HWIUWJI.Job.byJobID',
-        logger))(tx, jobID, false, true);
+        logger))(tx, jobID, false, false, true);
 
       const thisStep: WorkflowStep = await (await logAsyncExecutionTime(
         getWorkflowStepByJobIdStepIndex,
@@ -916,7 +916,7 @@ export async function handleWorkItemUpdateWithJobId(
       const { job } = await (await logAsyncExecutionTime(
         Job.byJobID,
         'HWIUWJI.Job.byJobID',
-        logger))(tx, jobID, false, true);
+        logger))(tx, jobID, false, false, true);
 
       await processWorkItem(tx, preprocessResult, job, update, logger, true, undefined);
       await job.save(tx);

services/harmony/app/frontends/jobs.ts

@@ -172,7 +172,7 @@ export async function getJobStatus(
     let errors: JobError[];
 
     await db.transaction(async (tx) => {
-      ({ job, pagination } = await Job.byJobID(tx, jobID, true, false, page, limit));
+      ({ job, pagination } = await Job.byJobID(tx, jobID, true, true, false, page, limit));
       errors = await getErrorsForJob(tx, jobID);
     });
     if (!job) {

services/harmony/app/markdown/apis.md

@@ -66,6 +66,7 @@ As such it accepts parameters in the URL path as well as query parameters.
 | height | number of rows to return in the output coverage |
 | forceAsync | if "true", override the default API behavior and always treat the request as asynchronous |
 | format | the mime-type of the output format to return |
+| label | the label(s) to add for the job that runs the request. Multiple labels can be specified as a comma-separated list. A label can contain any characters up to a 255 character limit, but if a label contains commas the request can only be a POST with with the label field in the body. Labels are always converted to lower case.
 | maxResults | limits the number of input files processed in the request |
 | skipPreview | if "true", override the default API behavior and never auto-pause jobs |
 | ignoreErrors | if "true", continue processing a request to completion even if some items fail. If "false" immediately fail the request. Defaults to true |
@@ -125,6 +126,7 @@ Currently only the `/position`, `/cube`, `/trajectory` and `/area` routes are su
 | grid | the name of the output grid to use for regridding requests. The name must match the UMM  |grid name in the CMR.
 | ignoreErrors | if "true", continue processing a request to completion even if some items fail. If "false" immediately fail the request. Defaults to true |
 | interpolation | specify the interpolation method used during reprojection and scaling |
+| label | the label(s) to add for the job that runs the request. Multiple labels can be specified as a comma-separated list or as an array in the JSON body for POST requests. A label can contain any characters up to a 255 character limit, but if a label contains commas the request can only be a POST. Labels are always converted to lower case.
 | maxResults | limits the number of input files processed in the request |
 | scaleExtent | scale the resulting coverage along one axis to a given extent |
 | scaleSize | scale the resulting coverage along one axis to a given size |

services/harmony/app/middleware/label.ts

@@ -0,0 +1,27 @@
+import { Response, NextFunction } from 'express';
+import HarmonyRequest from '../models/harmony-request';
+import { parseMultiValueParameter } from '../util/parameter-parsing-helpers';
+
+/**
+ * Express.js middleware to convert label parameter to an array (if needed) and add
+ * it to the body of the request
+ *
+ * @param req - The client request
+ * @param _res - The client response (not used)
+ * @param next - The next function in the middleware chain
+ */
+export default async function handleLabelParameter(
+  req: HarmonyRequest, _res: Response, next: NextFunction,
+): Promise<void> {
+  // Check if 'label' exists in the query parameters (GET), form-encoded body, or JSON body
+  let label = req.query.label || req.body.label;
+
+  // If 'label' exists, convert it to an array (if not already) and assign it to 'label' in the body
+  if (label) {
+    label = parseMultiValueParameter(label);
+    req.body.label = label;
+  }
+
+  // Call next to pass control to the next middleware or route handler
+  next();
+}

services/harmony/app/middleware/parameter-validation.ts

@@ -9,6 +9,7 @@ import { getEdrParameters } from '../frontends/ogc-edr/index';
 import env from '../util/env';
 import { getRequestRoot } from '../util/url';
 import { validateNoConflictingGridParameters } from '../util/grids';
+import { checkLabel } from '../models/label';
 
 const { awsDefaultRegion } = env;
 
@@ -154,6 +155,25 @@ function validateBucketPathParameter(req: HarmonyRequest): void {
   }
 }
 
+/**
+ * Verify that the given label is valid. Send an error if it is not.
+ * @param req - The request object
+ * @throws RequestValidationError - if a label is invalid (too short or too long)
+ */
+export function validateLabelParameter(req: HarmonyRequest): void {
+  const labels = req.body.label;
+
+  if (labels) {
+    for (const label of labels) {
+      const errMsg = checkLabel(label);
+
+      if (errMsg) {
+        throw new RequestValidationError(errMsg);
+      }
+    }
+  }
+}
+
 /**
  * Validate that the parameter names are correct.
  *  (Performs case insensitive comparison.)
@@ -225,6 +245,7 @@ export default async function parameterValidation(
     if (req.url.match(EDR_ROUTE_REGEX)) {
       validateEdrParameterNames(req);
     }
+    validateLabelParameter(req);
   } catch (e) {
     return next(e);
   }

services/harmony/app/models/job.ts

@@ -21,6 +21,7 @@ import JobError from './job-error';
 import { setReadyCountToZero } from './user-work';
 import { Knex } from 'knex';
 import { Logger } from 'winston';
+import { getLabelsForJob, setLabelsForJob } from './label';
 const { awsDefaultRegion } = env;
 
 // Lazily load the list of unique provider ids, once, when requested
@@ -103,6 +104,8 @@ export class JobForDisplay {
 
   links: JobLink[];
 
+  labels: string[];
+
   request: string;
 
   numInputGranules: number;
@@ -419,6 +422,8 @@ export class Job extends DBRecord implements JobRecord {
 
   provider_id?: string;
 
+  labels: string[];
+
   /**
    * Get the job message for the current status.
    * @returns the message string describing the job
@@ -504,6 +509,7 @@ export class Job extends DBRecord implements JobRecord {
     tx: Transaction,
     constraints: JobQuery = {},
     includeLinks = false,
+    includeLabels = false,
     lock = false,
     currentPage = 0,
     perPage = env.defaultResultPageSize,
@@ -515,10 +521,15 @@ export class Job extends DBRecord implements JobRecord {
     const result = await query;
     const job = result ? new Job(result) : null;
     let paginationInfo;
-    if (job && includeLinks) {
-      const linkData = await getLinksForJob(tx, job.jobID, currentPage, perPage);
-      job.links = linkData.data;
-      paginationInfo = linkData.pagination;
+    if (job) {
+      if (includeLinks) {
+        const linkData = await getLinksForJob(tx, job.jobID, currentPage, perPage);
+        job.links = linkData.data;
+        paginationInfo = linkData.pagination;
+      }
+      if (includeLabels) {
+        job.labels = await getLabelsForJob(tx, job.jobID);
+      }
     }
     return { job, pagination: paginationInfo };
   }
@@ -544,17 +555,18 @@ export class Job extends DBRecord implements JobRecord {
   * @param transaction - the transaction to use for querying
   * @param jobID - the jobID for the job that should be retrieved
   * @param includeLinks - if true include the job links when returning the job
+  * @param includeLabels - if true include the labels when returning the job
   * @param lock - if true lock the row in the jobs table
   * @param currentPage - the index of the page of job links to show
   * @param perPage - the number of job links to include per page
   * @returns the Job with the given JobID or null if not found
   */
   static async byJobID(
-    tx: Transaction, jobID: string, includeLinks = false, lock = false, currentPage = 0,
+    tx: Transaction, jobID: string, includeLinks = false, includeLabels = false, lock = false, currentPage = 0,
     perPage = env.defaultResultPageSize,
   ): Promise<{ job: Job; pagination: ILengthAwarePagination }> {
     const constraints = { where: { jobID } };
-    return Job.queryForSingleJob(tx, constraints, includeLinks, lock, currentPage, perPage);
+    return Job.queryForSingleJob(tx, constraints, includeLinks, includeLabels, lock, currentPage, perPage);
   }
 
   /**
@@ -580,6 +592,7 @@ export class Job extends DBRecord implements JobRecord {
    * @param username - the username associated with the job
    * @param jobID - the job ID for the request
    * @param includeLinks - if true, load all JobLinks into job.links
+   * @param includeLabels - if true include labels with the job
    * @param lock - if true lock the row in the jobs table
    * @param currentPage - the index of the page of links to show
    * @param perPage - the number of link results per page
@@ -591,12 +604,13 @@ export class Job extends DBRecord implements JobRecord {
     username,
     jobID,
     includeLinks = false,
+    includeLabels = false,
     lock = false,
     currentPage = 0,
     perPage = env.defaultResultPageSize,
   ): Promise<{ job: Job; pagination: ILengthAwarePagination }> {
     const constraints = { where: { username, jobID } };
-    return Job.queryForSingleJob(tx, constraints, includeLinks, lock, currentPage, perPage);
+    return Job.queryForSingleJob(tx, constraints, includeLinks, includeLabels, lock, currentPage, perPage);
   }
 
   /**
@@ -1008,6 +1022,7 @@ export class Job extends DBRecord implements JobRecord {
       }
     }
     await Promise.all(promises);
+    await setLabelsForJob(tx, this.jobID, this.username, this.labels);
   }
 
   /**
@@ -1026,6 +1041,7 @@ export class Job extends DBRecord implements JobRecord {
       updatedAt: new Date(this.updatedAt),
       dataExpiration: this.getDataExpiration(),
       links: this.links,
+      labels: this.labels,
       request: this.request,
       numInputGranules: this.numInputGranules,
       jobID: this.jobID,