Metadata in compute jobs (#976)

* metadata

* remove editor spaces

* lint fix

* test and readme

Author: giurgiur99

docs/API.md

@@ -1275,17 +1275,18 @@ starts a free compute job and returns jobId if succesfull
 
 #### Parameters
 
-| name            | type   | required | description                               |
-| --------------- | ------ | -------- | ----------------------------------------- |
-| command         | string | v        | command name                              |
-| node            | string |          | if not present it means current node      |
-| consumerAddress | string | v        | consumer address                          |
-| signature       | string | v        | signature (msg=String(nonce) )            |
-| nonce           | string | v        | nonce for the request                     |
-| datasets        | object |          | list of ComputeAsset to be used as inputs |
-| algorithm       | object |          | ComputeAlgorithm definition               |
-| environment     | string | v        | compute environment to use                |
-| resources       | object |          | optional list of required resources       |
+| name            | type   | required | description                                              |
+| --------------- | ------ | -------- | -------------------------------------------------------- |
+| command         | string | v        | command name                                             |
+| node            | string |          | if not present it means current node                     |
+| consumerAddress | string | v        | consumer address                                         |
+| signature       | string | v        | signature (msg=String(nonce) )                           |
+| nonce           | string | v        | nonce for the request                                    |
+| datasets        | object |          | list of ComputeAsset to be used as inputs                |
+| algorithm       | object |          | ComputeAlgorithm definition                              |
+| environment     | string | v        | compute environment to use                               |
+| resources       | object |          | optional list of required resources                      |
+| metadata        | object |          | optional metadata for the job, data provided by the user |
 
 #### Request
 
@@ -1300,7 +1301,8 @@ starts a free compute job and returns jobId if succesfull
   "signature": "123",
   "nonce": 1,
   "environment": "0x7d187e4c751367be694497ead35e2937ece3c7f3b325dcb4f7571e5972d092bd-0xbeaf12703d708f39ef98c3d8939ce458553254176dbb69fe83d535883c4cee38",
-  "resources": [{ "id": "cpu", "amount": 1 }]
+  "resources": [{ "id": "cpu", "amount": 1 }],
+  "metadata": { "key": "value" }
 }
 ```
 
@@ -1324,7 +1326,8 @@ starts a free compute job and returns jobId if succesfull
       { "id": "ram", "amount": 1000000000 },
       { "id": "disk", "amount": 0 }
     ],
-    "isFree": true
+    "isFree": true,
+    "metadata": { "key": "value" }
   }
 ]
 ```
@@ -1338,6 +1341,7 @@ starts a free compute job and returns jobId if succesfull
 returns job status
 
 #### Parameters
+
 Required at least one of the following parameters:
 
 | name            | type   | required | description                          |
@@ -1378,7 +1382,8 @@ Required at least one of the following parameters:
         "amount": 1000000000
       }
     ],
-    "isFree": true
+    "isFree": true,
+    "metadata": { "key": "value" }
   }
 ]
 ```

src/@types/C2D/C2D.ts

@@ -141,6 +141,9 @@ export interface ComputeResult {
   index?: number
 }
 
+export type DBComputeJobMetadata = {
+  [key: string]: string | number | boolean
+}
 export interface ComputeJob {
   owner: string
   did?: string
@@ -155,6 +158,7 @@ export interface ComputeJob {
   maxJobDuration?: number
   agreementId?: string
   environment?: string
+  metadata?: DBComputeJobMetadata
 }
 
 export interface ComputeOutput {
@@ -198,6 +202,7 @@ export interface DBComputeJobPayment {
   lockTx: string
   claimTx: string
 }
+
 // this is the internal structure
 export interface DBComputeJob extends ComputeJob {
   clusterHash: string
@@ -216,6 +221,7 @@ export interface DBComputeJob extends ComputeJob {
   algoStopTimestamp: string
   resources: ComputeResourceRequestWithPrice[]
   payment?: DBComputeJobPayment
+  metadata?: DBComputeJobMetadata
 }
 
 // make sure we keep them both in sync

src/@types/commands.ts

@@ -5,7 +5,8 @@ import type {
   ComputeAsset,
   ComputeAlgorithm,
   ComputeOutput,
-  ComputeResourceRequest
+  ComputeResourceRequest,
+  DBComputeJobMetadata
 } from './C2D/C2D.js'
 import {
   ArweaveFileObject,
@@ -200,6 +201,7 @@ export interface FreeComputeStartCommand extends Command {
   output?: ComputeOutput
   resources?: ComputeResourceRequest[]
   maxJobDuration?: number
+  metadata?: DBComputeJobMetadata
 }
 export interface PaidComputeStartCommand extends FreeComputeStartCommand {
   payment: ComputePayment

src/components/c2d/compute_engine_base.ts

@@ -13,7 +13,8 @@ import type {
   ComputeResourcesPricingInfo,
   DBComputeJobPayment,
   DBComputeJob,
-  dockerDeviceRequest
+  dockerDeviceRequest,
+  DBComputeJobMetadata
 } from '../../@types/C2D/C2D.js'
 import { C2DClusterType } from '../../@types/C2D/C2D.js'
 import { C2DDatabase } from '../database/C2DDatabase.js'
@@ -61,7 +62,8 @@ export abstract class C2DEngine {
     maxJobDuration: number,
     resources: ComputeResourceRequest[],
     payment: DBComputeJobPayment,
-    jobId: string
+    jobId: string,
+    metadata?: DBComputeJobMetadata
   ): Promise<ComputeJob[]>
 
   public abstract stopComputeJob(

src/components/c2d/compute_engine_docker.ts

@@ -1,6 +1,10 @@
 /* eslint-disable security/detect-non-literal-fs-filename */
 import { Readable } from 'stream'
-import { C2DStatusNumber, C2DStatusText } from '../../@types/C2D/C2D.js'
+import {
+  C2DStatusNumber,
+  C2DStatusText,
+  DBComputeJobMetadata
+} from '../../@types/C2D/C2D.js'
 import type {
   C2DClusterInfo,
   ComputeEnvironment,
@@ -354,7 +358,8 @@ export class C2DEngineDocker extends C2DEngine {
     maxJobDuration: number,
     resources: ComputeResourceRequest[],
     payment: DBComputeJobPayment,
-    jobId: string
+    jobId: string,
+    metadata?: DBComputeJobMetadata
   ): Promise<ComputeJob[]> {
     if (!this.docker) return []
     // TO DO - iterate over resources and get default runtime
@@ -371,6 +376,14 @@ export class C2DEngineDocker extends C2DEngine {
         )}`
       )
     }
+
+    if (metadata && Object.keys(metadata).length > 0) {
+      const metadataSize = JSON.stringify(metadata).length
+      if (metadataSize > 1024) {
+        throw new Error('Metadata size is too large')
+      }
+    }
+
     const envIdWithHash = environment && environment.indexOf('-') > -1
     const env = await this.getComputeEnvironment(
       payment && payment.chainId ? payment.chainId : null,
@@ -408,7 +421,8 @@ export class C2DEngineDocker extends C2DEngine {
       isFree,
       algoStartTimestamp: '0',
       algoStopTimestamp: '0',
-      payment
+      payment,
+      metadata
     }
     await this.makeJobFolders(job)
     // make sure we actually were able to insert on DB

src/components/core/compute/startCompute.ts

@@ -366,7 +366,8 @@ export class PaidComputeStartHandler extends CommandHandler {
         maxJobDuration: task.maxJobDuration,
         chainId: task.payment.chainId,
         agreementId: '',
-        resources
+        resources,
+        metadata: task.metadata
       }
       // job ID unicity
       const jobId = generateUniqueID(s)
@@ -412,7 +413,8 @@ export class PaidComputeStartHandler extends CommandHandler {
             lockTx: agreementId,
             claimTx: null
           },
-          jobId
+          jobId,
+          task.metadata
         )
         CORE_LOGGER.logMessage(
           'ComputeStartCommand Response: ' + JSON.stringify(response, null, 2),
@@ -566,7 +568,8 @@ export class FreeComputeStartHandler extends CommandHandler {
         environment: task.environment,
         owner: task.consumerAddress,
         maxJobDuration: task.maxJobDuration,
-        resources: task.resources
+        resources: task.resources,
+        metadata: task.metadata
       }
       const jobId = generateUniqueID(s)
       const response = await engine.startComputeJob(
@@ -578,7 +581,8 @@ export class FreeComputeStartHandler extends CommandHandler {
         task.maxJobDuration,
         task.resources,
         null,
-        jobId
+        jobId,
+        task.metadata
       )
 
       CORE_LOGGER.logMessage(

src/components/database/sqliteCompute.ts

@@ -44,7 +44,8 @@ function getInternalStructure(job: DBComputeJob): any {
     resources: job.resources,
     isFree: job.isFree,
     algoStartTimestamp: job.algoStartTimestamp,
-    algoStopTimestamp: job.algoStopTimestamp
+    algoStopTimestamp: job.algoStopTimestamp,
+    metadata: job.metadata
   }
   return internalBlob
 }
@@ -155,7 +156,8 @@ export class SQLiteCompute implements ComputeDatabaseProvider {
         maxJobDuration: job.maxJobDuration,
         chainId: job.payment?.chainId || null,
         agreementId: job.agreementId,
-        resources: job.resources
+        resources: job.resources,
+        metadata: job.metadata
       }
       jobId = generateUniqueID(jobStructure)
       job.jobId = jobId

src/components/httpRoutes/compute.ts

@@ -77,6 +77,7 @@ computeRoutes.post(`${SERVICES_API_BASE_PATH}/compute`, async (req, res) => {
       datasets: (req.body.datasets as unknown as ComputeAsset[]) || null,
       payment: (req.body.payment as unknown as ComputePayment) || null,
       resources: (req.body.resources as unknown as ComputeResourceRequest[]) || null,
+      metadata: req.body.metadata || null,
       authorization: req.headers?.authorization
     }
     if (req.body.output) {
@@ -119,6 +120,7 @@ computeRoutes.post(`${SERVICES_API_BASE_PATH}/freeCompute`, async (req, res) =>
       datasets: (req.body.datasets as unknown as ComputeAsset[]) || null,
       resources: (req.body.resources as unknown as ComputeResourceRequest[]) || null,
       maxJobDuration: req.body.maxJobDuration || null,
+      metadata: req.body.metadata || null,
       authorization: req.headers?.authorization
     }
     if (req.body.output) {

src/test/integration/compute.test.ts

@@ -884,7 +884,10 @@ describe('Compute', () => {
         transferTxId: algoOrderTxId,
         meta: publishedAlgoDataset.ddo.metadata.algorithm
       },
-      output: {}
+      output: {},
+      metadata: {
+        key: 'value'
+      }
       // additionalDatasets?: ComputeAsset[]
       // output?: ComputeOutput
     }
@@ -916,6 +919,7 @@ describe('Compute', () => {
     assert(response.stream, 'Failed to get stream')
     expect(response.stream).to.be.instanceOf(Readable)
     const jobs = await streamToObject(response.stream as Readable)
+    expect(jobs[0].metadata).to.deep.equal({ key: 'value' })
     console.log(jobs)
   })