Include metadata in compute jobs. Closes #1830 (#1889)

Author: brollb

src/common/compute/backends/ComputeClient.js

@@ -12,8 +12,8 @@ define([], function() {
             unimplemented(this.logger, 'cancelJob');
         }
 
-        async createJob (/*hash*/) {
-            unimplemented(this.logger, 'createJob');
+        async startJob (/*hash, metadata*/) {
+            unimplemented(this.logger, 'startJob');
         }
 
         async getStatus (/*jobInfo*/) {

src/common/compute/backends/ComputeJob.js

@@ -0,0 +1,32 @@
+/* globals define */
+define([], function() {
+
+    class ComputeJob {
+        constructor (hash, name='DeepForge Job') {
+            this.name = name;
+            this.hash = hash;
+        }
+    }
+
+    class PipelineJob extends ComputeJob {
+        constructor (hash, projectId, branch, core, job) {
+            const execNode = core.getParent(job);
+            const jobInfo = PipelineJob.getNodeMetadata(core, job);
+            const execution = PipelineJob.getNodeMetadata(core, execNode);
+            const name = `DeepForge Job: ${jobInfo.name} (${execution.name}) in ${projectId} (${branch})`;
+            super(hash, name);
+            this.job = jobInfo;
+            this.execution = execution;
+            this.branch = branch;
+            this.projectId = projectId;
+        }
+
+        static getNodeMetadata (core, node) {
+            const id = core.getPath(node);
+            const name = core.getAttribute(node, 'name');
+            return {id, name};
+        }
+    }
+
+    return {ComputeJob, PipelineJob};
+});

src/common/compute/backends/gme/Client.js

@@ -96,9 +96,9 @@ define([
             return this.executor.getInfo(job.hash);
         }
 
-        async createJob (hash) {
+        async startJob (job) {
             await this.checkExecutionEnv();
-
+            const {hash} = job;
             const result = await this.executor.createJob({hash});
 
             this.poll(hash);

src/common/compute/backends/local/Client.js

@@ -108,7 +108,8 @@ define([
             }
         }
 
-        async createJob (hash) {
+        async startJob (job) {
+            const {hash} = job;
             this.jobQueue.push(hash);
             this._processNextJob();
 
@@ -143,15 +144,15 @@ define([
 
             this.currentJob = this.jobQueue.shift();
             if (this.currentJob) {
-                return this._createJob(this.currentJob);
+                return this._startJob(this.currentJob);
             }
         }
 
         _getWorkingDir (hash) {
             return path.join(os.tmpdir(), `deepforge-local-exec-${hash}`);
         }
 
-        async _createJob (hash) {
+        async _startJob (hash) {
             const jobInfo = {hash};
             this.emit('update', jobInfo.hash, this.PENDING);
             const tmpdir = this._getWorkingDir(hash);

src/common/compute/backends/sciserver-compute/Client.js

@@ -32,19 +32,16 @@ define([
             this.consoleOutputLen = {};
         }
 
-        async createJob (hash) {
-            const filesInfo = await this._uploadFiles(hash);
-            const job = await this._createJob(filesInfo);
-            const jobInfo = {
-                id: job.id,
-                hash,
-            };
+        async startJob (job) {
+            const filesInfo = await this._uploadFiles(job.hash);
+            const {id} = await this._startJob(job, filesInfo);
+            const jobInfo = {id, hash: job.hash};
             this._poll(jobInfo);
 
             return jobInfo;
         }
 
-        async _createConfig (filesInfo) {
+        async _createConfig (job, filesInfo) {
             const {dirname, volumePool, volume} = filesInfo;
             const domain = await this._getComputeDomain();
             const userVolumes = domain.userVolumes.map(volume => ({
@@ -58,7 +55,7 @@ define([
                 dockerComputeEndpoint: domain.apiEndpoint,
                 dockerImageName: 'DeepForge',
                 resultsFolderURI: '',
-                submitterDID: 'DeepForge Job',
+                submitterDID: job.name,
                 volumeContainers: [],
                 userVolumes: userVolumes
             };
@@ -89,8 +86,8 @@ define([
             return filesInfo;
         }
 
-        async _createJob (filesInfo) {
-            const config = await this._createConfig(filesInfo);
+        async _startJob (job, filesInfo) {
+            const config = await this._createConfig(job, filesInfo);
             const url = 'https://apps.sciserver.org/racm//jobm/rest/jobs/docker';
 
             const opts = {

src/plugins/ExecuteJob/ExecuteJob.js

@@ -20,6 +20,7 @@ define([
     'q',
     'superagent',
     'underscore',
+    'deepforge/compute/backends/ComputeJob',
 ], function (
     assert,
     pluginMetadata,
@@ -41,9 +42,11 @@ define([
     Q,
     superagent,
     _,
+    ComputeJob,
 ) {
     'use strict';
 
+    const {PipelineJob} = ComputeJob;
     pluginMetadata = JSON.parse(pluginMetadata);
 
     /**
@@ -510,7 +513,14 @@ define([
     ExecuteJob.prototype.createJob = async function (job, hash) {
         // Record the job info for the given hash
         this._execHashToJobNode[hash] = job;
-        const jobInfo = await this.compute.createJob(hash);
+        const computeJob = new PipelineJob(
+            hash,
+            this.projectId,
+            this.branchName,
+            this.core,
+            job
+        );
+        const jobInfo = await this.compute.startJob(computeJob);
         this.core.setAttribute(job, 'jobInfo', JSON.stringify(jobInfo));
         this.core.setAttribute(job, 'execFiles', hash);
 

src/routers/InteractiveCompute/Session.js

@@ -9,6 +9,7 @@ const Channel = require('./Channel');
 const EventEmitter = requireJS('deepforge/EventEmitter');
 const Message = requireJS('deepforge/compute/interactive/message');
 const ComputeClient = requireJS('deepforge/compute/backends/ComputeClient');
+const {ComputeJob} = requireJS('deepforge/compute/backends/ComputeJob');
 
 class Session extends EventEmitter {
     constructor(blobClient, compute, clientSocket) {
@@ -36,7 +37,9 @@ class Session extends EventEmitter {
 
         const files = new JobFiles(blobClient, SERVER_URL, this.id);
         const hash = await files.upload();
-        this.jobInfo = this.compute.createJob(hash);
+        const name = 'DeepForge Interactive Session';
+        const computeJob = new ComputeJob(hash, name);
+        this.jobInfo = this.compute.startJob(computeJob);
         this.compute.on('end', (id, info) => {
             const isError = this.clientSocket.readyState === WebSocket.OPEN &&
                 info.status !== ComputeClient.SUCCESS;