Compute Job - Check container stats, regarding cpu & mem usage

src/components/c2d/compute_engine_base.ts

@@ -111,6 +111,18 @@ export abstract class C2DEngine {
   public getStreamableLogs(jobId: string): Promise<NodeJS.ReadableStream> {
     throw new Error(`Not implemented for this engine type`)
   }
+
+  protected async getJobEnvironment(job: DBComputeJob): Promise<ComputeEnvironment> {
+    const environments: ComputeEnvironment[] = await (
+      await this.getComputeEnvironments()
+    ).filter((env: ComputeEnvironment) => env.id === job.environment)
+    // found it
+    if (environments.length === 1) {
+      const environment = environments[0]
+      return environment
+    }
+    return null
+  }
 }
 
 export class C2DEngineLocal extends C2DEngine {

src/components/c2d/compute_engine_docker.ts

@@ -19,7 +19,12 @@ import { C2DDatabase } from '../database/C2DDatabase.js'
 import { create256Hash } from '../../utils/crypt.js'
 import { Storage } from '../storage/index.js'
 import Dockerode from 'dockerode'
-import type { ContainerCreateOptions, VolumeCreateOptions } from 'dockerode'
+import type {
+  ContainerCreateOptions,
+  HostConfig,
+  // ContainerStats,
+  VolumeCreateOptions
+} from 'dockerode'
 import * as tar from 'tar'
 import {
   createWriteStream,
@@ -41,6 +46,7 @@ import { Service } from '../../@types/DDO/Service.js'
 import { decryptFilesObject, omitDBComputeFieldsFromComputeJob } from './index.js'
 import * as drc from 'docker-registry-client'
 import { ValidateParams } from '../httpRoutes/validateCommands.js'
+import { convertGigabytesToBytes } from '../../utils/util.js'
 
 export class C2DEngineDocker extends C2DEngine {
   private envs: ComputeEnvironment[] = []
@@ -202,7 +208,7 @@ export class C2DEngineDocker extends C2DEngine {
 
     const jobId = generateUniqueID()
 
-    // TO DO C2D - Check image, check arhitecture, etc
+    // C2D - Check image, check arhitecture, etc
     const image = getAlgorithmImage(algorithm)
     // ex: node@sha256:1155995dda741e93afe4b1c6ced2d01734a6ec69865cc0997daf1f4db7259a36
     if (!image) {
@@ -219,6 +225,7 @@ export class C2DEngineDocker extends C2DEngine {
       envIdWithHash ? environment : null,
       environment
     )
+
     const validation = await C2DEngineDocker.checkDockerImage(image, env.platform)
     if (!validation.valid)
       throw new Error(`Unable to validate docker image ${image}: ${validation.reason}`)
@@ -498,9 +505,11 @@ export class C2DEngineDocker extends C2DEngine {
         await this.db.updateJob(job)
         await this.cleanupJob(job)
       }
+      // get env info
+      const environment = await this.getJobEnvironment(job)
       // create the container
       const mountVols: any = { '/data': {} }
-      const hostConfig: any = {
+      const hostConfig: HostConfig = {
         Mounts: [
           {
             Type: 'volume',
@@ -510,6 +519,18 @@ export class C2DEngineDocker extends C2DEngine {
           }
         ]
       }
+      if (environment != null) {
+        // limit container CPU & Memory usage according to env specs
+        hostConfig.CpuCount = environment.cpuNumber || 1
+        // if more than 1 CPU
+        if (hostConfig.CpuCount > 1) {
+          hostConfig.CpusetCpus = `0-${hostConfig.CpuCount - 1}`
+        }
+        hostConfig.Memory = 0 || convertGigabytesToBytes(environment.ramGB)
+        // set swap to same memory value means no swap (otherwise it use like 2X mem)
+        hostConfig.MemorySwap = hostConfig.Memory
+      }
+      // console.log('host config: ', hostConfig)
       const containerInfo: ContainerCreateOptions = {
         name: job.jobId + '-algoritm',
         Image: job.containerImage,
@@ -533,6 +554,7 @@ export class C2DEngineDocker extends C2DEngine {
 
       try {
         const container = await this.docker.createContainer(containerInfo)
+        // this.checkResources(job, container)
         console.log('container: ', container)
         job.status = C2DStatusNumber.Provisioning
         job.statusText = C2DStatusText.Provisioning
@@ -572,6 +594,7 @@ export class C2DEngineDocker extends C2DEngine {
         if (details.State.Running === false) {
           try {
             await container.start()
+            // this.checkResources(job, container)
             job.isStarted = true
             await this.db.updateJob(job)
             return
@@ -589,6 +612,7 @@ export class C2DEngineDocker extends C2DEngine {
           }
         }
       } else {
+        // this.checkResources(job, container)
         // is running, we need to stop it..
         console.log('running, need to stop it?')
         const timeNow = Date.now() / 1000
@@ -646,6 +670,21 @@ export class C2DEngineDocker extends C2DEngine {
     }
   }
 
+  // Seems like monitoring container stats is useles... everything related with cpu/mem is at zeros
+  // private async checkResources(job: DBComputeJob, container: Dockerode.Container) {
+  //   const environment = await this.getJobEnvironment(job)
+  //   try {
+  //     const statsRaw: any = await container.stats({ stream: false })
+  //     const stats = await JSON.parse(
+  //       JSON.stringify(statsRaw) // await streamToString(statsRaw as Readable))
+  //     )
+  //     const memory = stats.memory_stats
+  //     const cpu = stats.cpu_stats
+  //   } catch (e) {
+  //     console.error('error getting stats: ', e)
+  //   }
+  // }
+
   // eslint-disable-next-line require-await
   private async cleanupJob(job: DBComputeJob) {
     // cleaning up

src/utils/util.ts

@@ -154,3 +154,13 @@ export function deleteKeysFromObject(source: any, keys: string[]): any {
   })
   return source
 }
+
+export function convertGigabytesToBytes(gigabytes: number): number {
+  if (gigabytes < 0) {
+    throw new Error('Input must be a non-negative number')
+  }
+
+  const bytesInAGigabyte = 1024 ** 3 // 1 gigabyte = 1024^3 bytes
+  const bytes = gigabytes * bytesInAGigabyte
+  return bytes
+}