Added log-storage for running jobs. Fixes #785 (#800)

WIP #785 Initial router commit

WIP #785 Added the log manager api

WIP #785 Added some comments for needed updates

WIP #785 Fixed typo in export

WIP #785 Added basic fn-ality to JobLogsClient

WIP #785 Added tests for the log router

WIP #785 uri encode project, branch. Updated viz to use the log-storage

WIP #785 Fixed updating on stdout update

WIP #785 Filtered out stdout update msgs from the notification widget

WIP #785 Added stdout save on canceled

WIP #785 Added tests for the joblogsclient

WIP #785 Moved job logs client to src/common

WIP #785 Added forking support to api

WIP #785 Added fork support for the job log client

WIP #785 Fixed flashing on canceled job

WIP #785 Fixed minor code climate issues

WIP #785 Create test-tmp on npm test

config/config.test.js

@@ -5,9 +5,11 @@
 
 'use strict';
 
-var config = require('./config.default');
+var config = require('./config.default'),
+    path = require('path');
 
 config.server.port = 9001;
 config.mongo.uri = 'mongodb://127.0.0.1:27017/webgme_tests';
+config.blob.fsDir = path.join(__dirname, '..', 'test-tmp', 'blob');
 
-module.exports = config;
+module.exports = config;

config/config.webgme.js

@@ -6,33 +6,34 @@
 var config = require('webgme/config/config.default'),
     validateConfig = require('webgme/config/validator');
 
-
 // The paths can be loaded from the webgme-setup.json
-config.plugin.basePaths.push('src/plugins');
-config.plugin.basePaths.push('node_modules/webgme-simple-nodes/src/plugins');
-config.visualization.layout.basePaths.push('node_modules/webgme-chflayout/src/layouts');
-config.visualization.decoratorPaths.push('src/decorators');
-config.visualization.decoratorPaths.push('node_modules/webgme-easydag/src/decorators');
-config.seedProjects.basePaths.push('src/seeds/nn');
-config.seedProjects.basePaths.push('src/seeds/devTests');
-config.seedProjects.basePaths.push('src/seeds/devUtilTests');
-config.seedProjects.basePaths.push('src/seeds/pipeline');
-config.seedProjects.basePaths.push('src/seeds/devPipelineTests');
-config.seedProjects.basePaths.push('src/seeds/project');
-config.seedProjects.basePaths.push('src/seeds/cifar10');
-config.seedProjects.basePaths.push('src/seeds/xor');
+config.plugin.basePaths.push(__dirname + '/../src/plugins');
+config.plugin.basePaths.push(__dirname + '/../node_modules/webgme-simple-nodes/src/plugins');
+config.visualization.layout.basePaths.push(__dirname + '/../node_modules/webgme-chflayout/src/layouts');
+config.visualization.decoratorPaths.push(__dirname + '/../src/decorators');
+config.visualization.decoratorPaths.push(__dirname + '/../node_modules/webgme-easydag/src/decorators');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/nn');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/devTests');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/devUtilTests');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/pipeline');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/devPipelineTests');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/project');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/cifar10');
+config.seedProjects.basePaths.push(__dirname + '/../src/seeds/xor');
 
 
 
-config.visualization.panelPaths.push('node_modules/webgme-fab/src/visualizers/panels');
-config.visualization.panelPaths.push('node_modules/webgme-breadcrumbheader/src/visualizers/panels');
-config.visualization.panelPaths.push('node_modules/webgme-autoviz/src/visualizers/panels');
-config.visualization.panelPaths.push('node_modules/webgme-easydag/src/visualizers/panels');
-config.visualization.panelPaths.push('src/visualizers/panels');
+config.visualization.panelPaths.push(__dirname + '/../node_modules/webgme-fab/src/visualizers/panels');
+config.visualization.panelPaths.push(__dirname + '/../node_modules/webgme-breadcrumbheader/src/visualizers/panels');
+config.visualization.panelPaths.push(__dirname + '/../node_modules/webgme-autoviz/src/visualizers/panels');
+config.visualization.panelPaths.push(__dirname + '/../node_modules/webgme-easydag/src/visualizers/panels');
+config.visualization.panelPaths.push(__dirname + '/../src/visualizers/panels');
 
 
+config.rest.components['execution/logs'] = __dirname + '/../src/routers/JobLogsAPI/JobLogsAPI.js'
+
 // Visualizer descriptors
-config.visualization.visualizerDescriptors.push('./src/visualizers/Visualizers.json');
+config.visualization.visualizerDescriptors.push(__dirname + '/../src/visualizers/Visualizers.json');
 // Add requirejs paths
 config.requirejsPaths = {
   'EllipseDecorator': 'node_modules/webgme-easydag/src/decorators/EllipseDecorator',

package.json

@@ -8,7 +8,7 @@
     "start-dev": "NODE_ENV=dev node app.js",
     "local": "node ./bin/start-local.js",
     "worker": "node ./bin/start-worker.js",
-    "test": "node ./node_modules/mocha/bin/mocha --recursive test",
+    "test": "mkdir ./test-tmp; node ./node_modules/mocha/bin/mocha --recursive test",
     "watch-test": "./node_modules/nodemon/bin/nodemon.js --exec 'node ./node_modules/mocha/bin/mocha --recursive test'",
     "build-nn": "node ./utils/nn-parser.js"
   },

src/common/Constants.js

@@ -18,5 +18,8 @@ define({
     OP: {
         INPUT: 'Input',
         OUTPUT: 'Output'
-    }
+    },
+
+    // Job stdout update
+    STDOUT_UPDATE: 'stdout_update'
 });

src/common/JobLogsClient.js

@@ -0,0 +1,132 @@
+/* globals define */
+define([
+    'q',
+    'superagent'
+], function(
+    Q,
+    superagent
+) {
+    'use strict';
+
+    // Wrap the ability to read, update, and delete logs using the JobLogsAPI
+    var JobLogsClient = function(params) {
+        params = params || {};
+
+        this.logger = params.logger.fork('JobLogsClient');
+
+        // Get the server url
+        this.token = params.token;
+        this.origin = this._getServerUrl(params);
+        this.relativeUrl = '/execution/logs';
+        this.url = this.origin + this.relativeUrl;
+
+        this.logger.debug(`Setting url to ${this.url}`);
+
+        // Get the project, branch name
+        if (!(params.branchName && params.projectId)) {
+            throw Error('"branchName" and "projectId" required');
+        }
+        this.branch = params.branchName;
+        this.project = params.projectId;
+        this._modifiedJobs = [];
+
+        this.logger.debug(`Using <project>:<branch>: "${this.project}"/"${this.branch}"`);
+        this.logger.info('ctor finished');
+    };
+
+    JobLogsClient.prototype._getServerUrl = function(params) {
+        if (typeof window !== 'undefined') {
+            return window.location.origin;
+        }
+
+        // If not in browser, set using the params
+        var server = params.server || '127.0.0.1',
+            port = params.port || '80',
+            protocol = params.httpsecure ? 'https' : 'http';  // default is http
+
+        return params.origin || `${protocol}://${server}:${port}`;
+    };
+
+    // This method could be optimized - it could make a log of requests
+    JobLogsClient.prototype.fork = function(forkName) {
+        var jobIds = this._modifiedJobs,
+            deferred = Q.defer(),
+            url = [
+                this.url,
+                'migrate',
+                encodeURIComponent(this.project),
+                encodeURIComponent(this.branch),
+                encodeURIComponent(forkName)
+            ].join('/'),
+            req = superagent.post(url);
+
+        this.logger.info(`migrating ${jobIds.length} jobs from ${this.branch} to ${forkName} in ${this.project}`);
+        if (this.token) {
+            req.set('Authorization', 'Bearer ' + this.token);
+        }
+
+        req.send({jobs: jobIds})
+            .end((err, res) => {
+                if (err || res.status > 399) {
+                    return deferred.reject(err || res.status);
+                }
+
+                return deferred.resolve(res);
+            });
+
+        this.branch = forkName;
+        return deferred.promise;
+    };
+
+    JobLogsClient.prototype.getUrl = function(jobId) {
+        return [
+            this.url,
+            encodeURIComponent(this.project),
+            encodeURIComponent(this.branch),
+            encodeURIComponent(jobId)
+        ].join('/');
+    };
+
+    JobLogsClient.prototype._logRequest = function(method, jobId, content) {
+        var deferred = Q.defer(),
+            req = superagent[method](this.getUrl(jobId));
+
+        this.logger.info(`sending ${method} request to ${this.getUrl(jobId)}`);
+        if (this.token) {
+            req.set('Authorization', 'Bearer ' + this.token);
+        }
+
+        if (content) {
+            req = req.send(content);
+        }
+
+        req.end((err, res) => {
+            if (err || res.status > 399) {
+                return deferred.reject(err || res.status);
+            }
+
+            return deferred.resolve(res);
+        });
+
+        return deferred.promise;
+    };
+
+    JobLogsClient.prototype.appendTo = function(jobId, text) {
+        this._modifiedJobs.push(jobId);
+        this.logger.info(`Appending logs to ${jobId}`);
+        return this._logRequest('patch', jobId, {patch: text});
+    };
+
+    JobLogsClient.prototype.getLog = function(jobId) {
+        this.logger.info(`Getting logs for ${jobId}`);
+        return this._logRequest('get', jobId)
+            .then(res => res.text);
+    };
+
+    JobLogsClient.prototype.deleteLog = function(jobId) {
+        this.logger.info(`Deleting logs for ${jobId}`);
+        return this._logRequest('delete', jobId);
+    };
+
+    return JobLogsClient;
+});

src/plugins/ExecuteJob/ExecuteJob.js

@@ -8,6 +8,7 @@ define([
     'plugin/PluginBase',
     'deepforge/plugin/LocalExecutor',
     'deepforge/plugin/PtrCodeGen',
+    'deepforge/JobLogsClient',
     'deepforge/Constants',
     './templates/index',
     'q',
@@ -19,6 +20,7 @@ define([
     PluginBase,
     LocalExecutor,  // DeepForge operation primitives
     PtrCodeGen,
+    JobLogsClient,
     CONSTANTS,
     Templates,
     Q,
@@ -49,6 +51,7 @@ define([
         this._oldMetadataByName = {};  // name -> id
         this.lastAppliedCmd = {};
         this.canceled = false;
+        this.logManager = null;
     };
 
     /**
@@ -81,6 +84,13 @@ define([
             return callback(`Cannot execute ${typeName} (expected Job)`, this.result);
         }
 
+        // Get the gmeConfig...
+        this.logManager = new JobLogsClient({
+            logger: this.logger,
+            port: this.gmeConfig.server.port,
+            branchName: this.branchName,
+            projectId: this.projectId
+        });
         this._callback = callback;
         this.currentForkName = null;
         this.prepare()
@@ -114,6 +124,7 @@ define([
                 if (result.status === STORAGE_CONSTANTS.FORKED) {
                     msg = `"${name}" execution has forked to "${result.forkName}"`;
                     this.currentForkName = result.forkName;
+                    this.logManager.fork(result.forkName);
                     this.sendNotification(msg);
                 } else if (result.status === STORAGE_CONSTANTS.MERGED) {
                     this.logger.debug('Merged changes. About to update plugin nodes');
@@ -484,7 +495,8 @@ define([
         this.outputLineCount[jobId] = 0;
         // Set the job status to 'running'
         this.core.setAttribute(job, 'status', 'queued');
-        this.core.setAttribute(job, 'stdout', '');
+        this.core.delAttribute(job, 'stdout');
+        this.logManager.deleteLog(jobId);
         this.logger.info(`Setting ${jobId} status to "queued" (${this.currentHash})`);
         this.logger.debug(`Making a commit from ${this.currentHash}`);
         this.save(`Queued "${name}" operation in ${this.pipelineName}`)
@@ -818,6 +830,13 @@ define([
         });
     };
 
+    ExecuteJob.prototype.notifyStdoutUpdate = function (nodeId) {
+        this.sendNotification({
+            message: `${CONSTANTS.STDOUT_UPDATE}/${nodeId}`,
+            toBranch: true
+        });
+    };
+
     ExecuteJob.prototype.watchOperation = function (executor, hash, op, job) {
         var jobId = this.core.getPath(job),
             opId = this.core.getPath(op),
@@ -861,9 +880,9 @@ define([
                             output = this.processStdout(job, output, true);
 
                             if (output) {
-                                stdout += output;
-                                this.core.setAttribute(job, 'stdout', stdout);
-                                return this.save(`Received stdout for ${name}`);
+                                // Send notification to all clients watching the branch
+                                this.logManager.appendTo(jobId, output)
+                                    .then(() => this.notifyStdoutUpdate(jobId));
                             }
                         });
                 }
@@ -888,7 +907,11 @@ define([
                     // If it was cancelled, the pipeline has been stopped
                     this.logger.debug(`"${name}" has been CANCELED!`);
                     this.canceled = true;
-                    return this.onOperationCanceled(op);
+                    return this.logManager.getLog(jobId)
+                        .then(stdout => {
+                            this.core.setAttribute(job, 'stdout', stdout);
+                            return this.onOperationCanceled(op);
+                        });
                 }
 
                 if (info.status === 'SUCCESS' || info.status === 'FAILED_TO_EXECUTE') {
@@ -902,6 +925,7 @@ define([
                             // Parse the remaining code
                             stdout = this.processStdout(job, stdout);
                             this.core.setAttribute(job, 'stdout', stdout);
+                            this.logManager.deleteLog(jobId);
                             if (info.status !== 'SUCCESS') {
                                 // Download all files
                                 this.result.addArtifact(info.resultHashes[name + '-all-files']);

src/plugins/ExecutePipeline/ExecutePipeline.js

@@ -4,6 +4,7 @@
 define([
     'plugin/CreateExecution/CreateExecution/CreateExecution',
     'plugin/ExecuteJob/ExecuteJob/ExecuteJob',
+    'deepforge/JobLogsClient',
     'common/storage/constants',
     'common/core/constants',
     'q',
@@ -12,6 +13,7 @@ define([
 ], function (
     CreateExecution,
     ExecuteJob,
+    JobLogsClient,
     STORAGE_CONSTANTS,
     CONSTANTS,
     Q,
@@ -111,6 +113,13 @@ define([
             return callback('Current node is not a Pipeline or Execution!', this.result);
         }
 
+        // Get the gmeConfig...
+        this.logManager = new JobLogsClient({
+            logger: this.logger,
+            port: this.gmeConfig.server.port,
+            branchName: this.branchName,
+            projectId: this.projectId
+        });
         this._callback = callback;
         this.currentForkName = null;
 
@@ -144,6 +153,7 @@ define([
                 var msg;
                 if (result.status === STORAGE_CONSTANTS.FORKED) {
                     this.currentForkName = result.forkName;
+                    this.logManager.fork(result.forkName);
                     msg = `"${this.pipelineName}" execution has forked to "${result.forkName}"`;
                     this.sendNotification(msg);
                 } else if (result.status === STORAGE_CONSTANTS.MERGED) {

src/routers/JobLogsAPI/JobLogManager.js

@@ -0,0 +1,125 @@
+var path = require('path'),
+    Q = require('q'),
+    fs = require('fs'),
+    exists = require('exists-file'),
+    NO_LOG_FOUND = '';
+
+var JobLogManager = function(logger, config) {
+    this.rootDir = path.join(config.blob.fsDir, 'log-storage');
+    this.logger = logger.fork('JobLogManager');
+    this._onCopyFinished = {};
+};
+
+JobLogManager.prototype._getFilePath = function(jobInfo) {
+    var jobId = jobInfo.job.replace(/\//g, '_'),
+        filename = `${jobId}.txt`;
+
+    return path.join(this.rootDir, jobInfo.project, jobInfo.branch, filename);
+};
+
+JobLogManager.prototype.exists = function(jobInfo) {
+    var filename = this._getFilePath(jobInfo);
+    return Q.nfcall(exists, filename);
+};
+
+JobLogManager.prototype.mkdirIfNeeded = function(dir) {
+    return Q.nfcall(exists, dir).then(exist => {
+        if (!exist) {
+            this.logger.debug('making dir:', dir);
+            return Q.nfcall(fs.mkdir, dir)
+                .catch(() => this.logger.debug(`dir already created: ${dir}`));
+        }
+    });
+};
+
+JobLogManager.prototype._copyFile = function(src, dst) {
+    return this.mkdirIfNeeded(path.dirname(dst)).then(() => {
+        var deferred = Q.defer(),
+            stream = fs.createReadStream(src).pipe(fs.createWriteStream(dst));
+
+        stream.on('error', deferred.reject);
+        stream.on('finish', deferred.resolve);
+
+        return deferred.promise;
+    });
+};
+
+// Copy one branch info to the next
+// Could optimize this to symlink until data appended...
+JobLogManager.prototype.migrate = function(migrationInfo, jobIds) {
+    // Recursively copy the srcBranch dir to the dstBranch dir
+    // Should probably use streams...
+    // Need to block appends to the given files so they are not written
+    // to until they have finished copying...
+    // TODO
+    var jobs,
+        src,
+        dst,
+        i;
+
+    for (i = jobIds.length; i--;) {
+        this._onCopyFinished[jobIds[i]] = [];
+    }
+
+    // Copy the job files and evaluate each of the finish functions
+    this.logger.info('migrating from ' + migrationInfo.srcBranch + ' to '+ migrationInfo.dstBranch);
+    return Q.all(jobIds.map(jobId => {
+        src = this._getFilePath({
+            project: migrationInfo.project,
+            branch: migrationInfo.srcBranch,
+            job: jobId
+        });
+        dst = this._getFilePath({
+            project: migrationInfo.project,
+            branch: migrationInfo.dstBranch,
+            job: jobId
+        });
+        return this._copyFile(src, dst).then(() => {
+            jobs = this._onCopyFinished[jobId];
+            for (var j = jobs.length; j--;) {
+                jobs[j]();
+            }
+        });
+    }));
+};
+
+JobLogManager.prototype.appendTo = function(jobInfo, logs) {
+    var filename = this._getFilePath(jobInfo),
+        branchDirname = path.dirname(filename),
+        projDirname = path.dirname(branchDirname);
+
+    this.logger.info(`Appending content to ${filename}`);
+    // Make directory if needed
+    return this.mkdirIfNeeded(this.rootDir)
+        .then(() => this.mkdirIfNeeded(projDirname))
+        .then(() => this.mkdirIfNeeded(branchDirname))
+        .then(() => Q.nfcall(fs.appendFile, filename, logs));
+};
+
+JobLogManager.prototype.getLog = function(jobInfo) {
+    var filename = this._getFilePath(jobInfo);
+
+    this.logger.info(`Getting log content to ${filename}`);
+    return this.exists(jobInfo)
+        .then(exists => {
+            if (exists) {
+                return Q.nfcall(fs.readFile, filename);
+            }
+            return NO_LOG_FOUND;
+        });
+};
+
+JobLogManager.prototype.delete = function(jobInfo) {
+    var filename = this._getFilePath(jobInfo);
+
+    return this.exists(jobInfo)
+        .then(exists => {
+            if (exists) {
+                this.logger.debug(`Removing file ${filename}`);
+                return Q.nfcall(fs.unlink, filename);
+            }
+            this.logger.info(`${filename} doesn't exist. No need to delete...`);
+        });
+};
+
+module.exports = JobLogManager;

src/routers/JobLogsAPI/JobLogsAPI.js

@@ -0,0 +1,93 @@
+/*jshint node:true*/
+
+'use strict';
+
+var express = require('express'),
+    JobLogManager = require('./JobLogManager'),
+    router = express.Router();
+
+/**
+ * Called when the server is created but before it starts to listening to incoming requests.
+ * N.B. gmeAuth, safeStorage and workerManager are not ready to use until the start function is called.
+ * (However inside an incoming request they are all ensured to have been initialized.)
+ *
+ * @param {object} middlewareOpts - Passed by the webgme server.
+ * @param {GmeConfig} middlewareOpts.gmeConfig - GME config parameters.
+ * @param {GmeLogger} middlewareOpts.logger - logger
+ * @param {function} middlewareOpts.ensureAuthenticated - Ensures the user is authenticated.
+ * @param {function} middlewareOpts.getUserId - If authenticated retrieves the userId from the request.
+ * @param {object} middlewareOpts.gmeAuth - Authorization module.
+ * @param {object} middlewareOpts.safeStorage - Accesses the storage and emits events (PROJECT_CREATED, COMMIT..).
+ * @param {object} middlewareOpts.workerManager - Spawns and keeps track of "worker" sub-processes.
+ */
+function initialize(middlewareOpts) {
+    var logger = middlewareOpts.logger.fork('JobLogsAPI'),
+        ensureAuthenticated = middlewareOpts.ensureAuthenticated,
+        gmeConfig = middlewareOpts.gmeConfig,
+        logManager = new JobLogManager(logger, gmeConfig);
+
+    logger.debug('initializing ...');
+
+    // Ensure authenticated can be used only after this rule.
+    router.use('*', function (req, res, next) {
+        // This header ensures that any failures with authentication won't redirect.
+        res.setHeader('X-WebGME-Media-Type', 'webgme.v1');
+        next();
+    });
+
+    // Use ensureAuthenticated if the routes require authentication. (Can be set explicitly for each route.)
+    router.use('*', ensureAuthenticated);
+
+    router.get('/:project/:branch/:job', function (req, res/*, next*/) {
+        // Retrieve the job logs for the given job
+        logManager.getLog(req.params).then(log => {
+            res.set('Content-Type', 'text/plain');
+            res.send(log);
+        });
+    });
+
+    router.patch('/:project/:branch/:job', function (req, res/*, next*/) {
+        var logs = req.body.patch;
+        logger.info(`Received append request for ${req.params.job} in ${req.params.project}`);
+        logManager.appendTo(req.params, logs)
+            .then(() => res.send('Append successful'))
+            .catch(err => logger.error(`Append failed: ${err}`));
+    });
+
+    router.delete('/:project/:branch/:job', function (req, res/*, next*/) {
+        logManager.delete(req.params).then(() => res.send('delete successful'));
+    });
+
+    router.post('/migrate/:project/:srcBranch/:dstBranch', function (req, res/*, next*/) {
+        var jobs = req.body.jobs;
+        logManager.migrate(req.params, jobs)
+            .then(() => res.send('migration successful'))
+            .fail(err => logger.error(err));
+    });
+
+    logger.debug('ready');
+}
+
+/**
+ * Called before the server starts listening.
+ * @param {function} callback
+ */
+function start(callback) {
+    callback();
+}
+
+/**
+ * Called after the server stopped listening.
+ * @param {function} callback
+ */
+function stop(callback) {
+    callback();
+}
+
+
+module.exports = {
+    initialize: initialize,
+    router: router,
+    start: start,
+    stop: stop
+};

src/visualizers/panels/Footer/FilteredNotificationWidget.js

@@ -0,0 +1,28 @@
+/* globals define */
+// A notification widget which filters out stdout update notifications
+define([
+    'deepforge/Constants',
+    'js/Widgets/Notification/NotificationWidget'
+], function(
+    CONSTANTS,
+    GmeNotificationWidget
+) {
+    var NotificationWidget = function() {
+        GmeNotificationWidget.apply(this, arguments);
+    };
+
+    NotificationWidget.prototype = Object.create(GmeNotificationWidget.prototype);
+
+    NotificationWidget.prototype.isUserNotication = function(data) {
+        return data.message.indexOf(CONSTANTS.STDOUT_UPDATE) === -1;
+    };
+
+    NotificationWidget.prototype._refreshNotifications = function(eventData) {
+
+        if (this.isUserNotication(eventData)) {
+            GmeNotificationWidget.prototype._refreshNotifications.call(this, eventData);
+        }
+    };
+
+    return NotificationWidget;
+});

src/visualizers/panels/Footer/FooterPanel.js

@@ -1,8 +1,5 @@
 /*globals define, _, WebGMEGlobal, $ */
 /*jshint browser: true*/
-/**
- * @author rkereskenyi / https://github.com/rkereskenyi
- */
 
 define([
     'js/PanelBase/PanelBase',
@@ -10,13 +7,15 @@ define([
     'js/Widgets/BranchStatus/BranchStatusWidget',
     'js/Widgets/BranchSelector/BranchSelectorWidget',
     'js/Widgets/KeyboardManager/KeyboardManagerWidget',
-    'js/Widgets/Notification/NotificationWidget'
-], function (PanelBase,
-             NetworkStatusWidget,
-             BranchStatusWidget,
-             BranchSelectorWidget,
-             KeyboardManagerWidget,
-             NotificationWidget) {
+    './FilteredNotificationWidget'
+], function (
+    PanelBase,
+    NetworkStatusWidget,
+    BranchStatusWidget,
+    BranchSelectorWidget,
+    KeyboardManagerWidget,
+    NotificationWidget
+) {
 
     'use strict';
 

src/visualizers/panels/LogViewer/LogViewerControl.js

@@ -2,9 +2,18 @@
 /*jshint browser: true*/
 
 // This is a read-only view of the 'stdout' attribute for a Job node
+// if the job is running, get the logs from the log-storage
 define([
+    'q',
+    'deepforge/JobLogsClient',
+    'js/Constants',
+    'deepforge/Constants',
     'panels/TextEditor/TextEditorControl'
 ], function (
+    Q,
+    JobLogsClient,
+    GME_CONSTANTS,
+    CONSTANTS,
     TextEditorControl
 ) {
 
@@ -19,11 +28,68 @@ define([
 
     _.extend(LogViewerControl.prototype, TextEditorControl.prototype);
 
+    LogViewerControl.prototype.getFullDescriptor = function (id) {
+        var desc = LogViewerControl.prototype._getObjectDescriptor.call(this, id);
+
+        return this._getRunningLogs(id).then(text => {
+            // Use attribute or running log if none
+            desc.text = desc.text || text;
+            return desc;
+        });
+    };
+
+    LogViewerControl.prototype.getUpdatedJobId = function (msg) {
+        // verify that it is the given notification type
+        if (msg.indexOf(CONSTANTS.STDOUT_UPDATE) !== -1) {
+            return msg.replace(/^[^\/]*\//, '');
+        }
+    };
+
+    LogViewerControl.prototype.selectedObjectChanged = function (id) {
+        TextEditorControl.prototype.selectedObjectChanged.call(this, id);
+        // Listen for notifications about updated logs
+        this.removeNotificationHandler();
+        this.notificationHandler = (sender, data) => {
+            var nodeId = this.getUpdatedJobId(data.message);
+            if (nodeId === id) {
+                this._onUpdate(id);
+            }
+        };
+        this._client.addEventListener(GME_CONSTANTS.CLIENT.NOTIFICATION, this.notificationHandler);
+    };
+
+    LogViewerControl.prototype.removeNotificationHandler = function () {
+        // Remove the notifications listener
+        if (this.notificationHandler) {
+            this._client.removeEventListener();
+            this.notificationHandler = null;
+        }
+    };
+
+    LogViewerControl.prototype.destroy = function () {
+        TextEditorControl.prototype.destroy.call(this);
+        this.removeNotificationHandler();
+    };
+
+    LogViewerControl.prototype._onLoad = function (id) {
+        this.getFullDescriptor(id).then(desc => this._widget.addNode(desc));
+    };
+
     LogViewerControl.prototype._onUpdate = function (id) {
         if (id === this._currentNodeId) {
-            TextEditorControl.prototype._onUpdate.call(this, id);
+            this.getFullDescriptor(id).then(desc => this._widget.updateNode(desc));
         }
     };
 
+    LogViewerControl.prototype._getRunningLogs = function (id) {
+        var logManager = new JobLogsClient({
+            logger: this._logger,
+            projectId: this._client.getActiveProjectId(),
+            branchName: this._client.getActiveBranchName()
+        });
+
+        return logManager.getLog(id);
+    };
+
     return LogViewerControl;
 });

test/common/JobLogsClient.spec.js

@@ -0,0 +1,138 @@
+var testFixture = require('../globals'),
+    expect = testFixture.expect,
+    assert = require('assert'),
+    path = testFixture.path,
+    gmeConfig = testFixture.getGmeConfig(),
+    blobDir = gmeConfig.blob.fsDir,
+    server = testFixture.WebGME.standaloneServer(gmeConfig),
+    Logger = require('webgme/src/server/logger'),
+    logger = Logger.createWithGmeConfig('gme', gmeConfig, true),
+    JobLogsClient = testFixture.requirejs('deepforge/JobLogsClient'),
+    rm_rf = require('rimraf'),
+    exists = require('exists-file');
+
+describe('JobLogsClient', function() {
+    var logClient = new JobLogsClient({
+            logger: logger,
+            origin: server.getUrl(),
+            projectId: 'testProject',
+            branchName: 'master'
+        }),
+        jobId = '/4/q/l',
+        firstLog = 'hello world';
+
+    before(function(done) {
+        testFixture.mkdir(blobDir);
+        server.start(done);
+    });
+
+    after(function(done) {
+        rm_rf.sync(blobDir);
+        server.stop(done);
+    });
+
+    describe('appendTo', function() {
+
+        before(function(done) {
+            logClient.appendTo(jobId, firstLog)
+                .then(() => done())
+                .catch(err => done(err));
+        });
+
+        it('should create job-logs directory', function() {
+            assert(exists.sync(path.join(blobDir, 'log-storage')));
+        });
+
+        describe('getLog', function() {
+            it('should return the logs from the job', function(done) {
+                logClient.getLog(jobId).then(log => {
+                    expect(log).to.contain(firstLog);
+                    done();
+                });
+            });
+
+            it('should append additional logs to the file', function(done) {
+                var secondLog = 'goodbye world';
+                logClient.appendTo(jobId, secondLog)
+                    .then(() => logClient.getLog(jobId))
+                    .then(log => {
+                        expect(log).to.contain(secondLog);
+                        done();
+                    });
+            });
+
+        });
+    });
+
+    describe('delete', function() {
+        var delJobId = '/4/8/l';
+
+        before(function(done) {
+            logClient.appendTo(delJobId, firstLog).then(() => done());
+        });
+
+        it('should delete the file from job-logs directory', function(done) {
+            logClient.deleteLog(delJobId)
+                .then(() => logClient.getLog(delJobId))
+                .then(log => {
+                    expect(log).to.equal('');
+                    done();
+                });
+        });
+    });
+
+    describe('migration', function() {
+        var client,
+            jId = '/asd/4/q',
+            logs = 'asdfasde',
+            newBranch = 'otherBranch';
+
+        before(function(done) {
+            client = new JobLogsClient({
+                logger: logger,
+                origin: server.getUrl(),
+                projectId: 'migTest',
+                branchName: 'master'
+            });
+            // Write logs to job
+            client.appendTo(jId, logs)
+                .then(() => client.fork(newBranch))
+                .catch(err => done(err))
+                .then(() => done());
+        });
+
+        it('should migrate the edited nodes to the new branch', function(done) {
+            client.getLog(jId).then(log => {
+                expect(log).to.equal(logs);
+                done();
+            });
+        });
+
+        describe('new logs', function() {
+            it('should not edit old job logs', function() {
+                var c2 = new JobLogsClient({
+                    logger: logger,
+                    origin: server.getUrl(),
+                    projectId: 'migTest',
+                    branchName: 'master'
+                });
+                c2.getLog(jId).then(log => {
+                    expect(log).to.equal(logs);
+                    done();
+                });
+            });
+
+            it('should write new logs', function() {
+                client.appendTo(jId, 'moreStuff')
+                    .then(() => client.getLog(jId))
+                    .then(log => {
+                        expect(log).to.contain('moreStuff');
+                        done();
+                    });
+            });
+        });
+
+        // TODO: Test that, on a fork, the logClient will copy all the current jobs to 
+        // the new fork name
+    });
+});

test/globals.js

@@ -7,6 +7,9 @@
 'use strict';
 
 var testFixture = require('webgme/test/_globals'),
+    path = require('path'),
+    fs = require('fs'),
+    exists = require('exists-file'),
     WEBGME_CONFIG_PATH = '../config';
 
 // This flag will make sure the config.test.js is being used
@@ -34,4 +37,18 @@ testFixture.requirejs.config({
 testFixture.getGmeConfig = getGmeConfig;
 
 testFixture.DF_SEED_DIR = testFixture.path.join(__dirname, '..', 'src', 'seeds');
+
+testFixture.mkdir = function(dir) {
+    var dirs = path.resolve(dir).split(path.sep),
+        shortDir,
+        i = 1;
+
+    while (i++ < dirs.length) {
+        shortDir = dirs.slice(0,i).join(path.sep);
+        if (!exists.sync(shortDir)) {
+            fs.mkdirSync(shortDir);
+        }
+    }
+};
+
 module.exports = testFixture;

test/routers/JobLogsAPI.spec.js

@@ -0,0 +1,198 @@
+var testFixture = require('../globals'),
+    superagent = testFixture.superagent,
+    Q = testFixture.Q,
+    expect = testFixture.expect,
+    assert = require('assert'),
+    path = testFixture.path,
+    gmeConfig = testFixture.getGmeConfig(),
+    blobDir = gmeConfig.blob.fsDir,
+    server = testFixture.WebGME.standaloneServer(gmeConfig),
+    mntPt = 'execution/logs',
+    rm_rf = require('rimraf'),
+    exists = require('exists-file');
+
+describe('JobLogsAPI', function() {
+    var project = 'testProject',
+        branch = 'master',
+        jobId = encodeURIComponent('/4/q/l'),
+        firstLog = 'hello world',
+        url = [
+            server.getUrl(),
+            mntPt,
+            project,
+            branch,
+            jobId
+        ].join('/');
+
+    before(function(done) {
+        testFixture.mkdir(blobDir);
+        server.start(done);
+    });
+
+    after(function(done) {
+        rm_rf.sync(blobDir);
+        server.stop(done);
+    });
+
+    describe('appendTo', function() {
+
+        before(function(done) {
+            superagent.patch(url)
+                .send({patch: firstLog})
+                .end(function (err, res) {
+                    expect(res.status).equal(200, err);
+                    done();
+                });
+        });
+
+        it('should create job-logs directory', function() {
+            assert(exists.sync(path.join(blobDir, 'log-storage')));
+        });
+
+        describe('getLog', function() {
+            it('should return the logs from the job', function(done) {
+                superagent.get(url)
+                    .end(function (err, res) {
+                        expect(res.status).equal(200, err);
+                        expect(res.text).to.contain(firstLog);
+                        done();
+                    });
+            });
+
+            it('should append additional logs to the file', function(done) {
+                var secondLog = 'goodbye world';
+                superagent.patch(url)
+                    .send({patch: secondLog})
+                    .end(function (err, res) {
+                        expect(res.status).equal(200, err);
+                        superagent.get(url)
+                            .end(function (err, res) {
+                                expect(res.status).equal(200, err);
+                                expect(res.text).to.contain(secondLog);
+                                done();
+                            });
+                    });
+            });
+
+        });
+    });
+
+    describe('delete', function() {
+        var delUrl = [
+            server.getUrl(),
+            mntPt,
+            'testProject',
+            'other',
+            encodeURIComponent('/4/8/l')
+        ].join('/');
+
+        before(function(done) {
+            superagent.patch(delUrl)
+                .send({patch: firstLog})
+                .end(function (err, res) {
+                    expect(res.status).equal(200, err);
+                    done();
+                });
+        });
+
+        it('should delete the file from job-logs directory', function(done) {
+            superagent.delete(delUrl)
+                .end(function (err, res) {
+                    expect(res.status).equal(200, err);
+                    superagent.get(delUrl)
+                        .end(function (err, res) {
+                            expect(res.status).equal(200, err);
+                            expect(res.text).to.equal('');
+                            done();
+                        });
+                });
+        });
+    });
+
+    function getUrl(project, branch, job) {
+        return [
+            server.getUrl(),
+            mntPt,
+            encodeURIComponent(project),
+            encodeURIComponent(branch),
+            encodeURIComponent(job)
+        ].join('/');
+    }
+
+    function addLog(project, branch, job, log) {
+        var deferred = Q.defer();
+
+        console.log('adding log', log);
+        superagent.patch(getUrl(project, branch, job))
+            .send({patch: log})
+            .end(function (err, res) {
+                if (err) {
+                    return deferred.reject(err + ' (' + job + ')');
+                }
+                return deferred.resolve(res);
+            });
+            
+        return deferred.promise;
+    }
+
+    describe('migrate', function() {
+        var j1 = '/s/p/3',
+            j2 = '/1/d/4',
+            b1 = 'asdfmaster',
+            b2 = 'not-master',
+            proj = 'someProject',
+            j1log = 'I am ' + j1,
+            j2log = 'I am ' + j2,
+            url;
+
+        before(function(done) {
+            Q.all([
+                addLog(proj, b1, j1, j1log),
+                addLog(proj, b1, j2, j2log),
+                addLog(proj, b2, j2, 'otherStuff')
+            ]).then(() => {
+                url = [
+                    server.getUrl(),
+                    mntPt + '/migrate',
+                    encodeURIComponent(proj),
+                    encodeURIComponent(b1),
+                    encodeURIComponent(b2)
+                ].join('/');
+                superagent.post(url)
+                    .send({jobs: [j1]})
+                    .end(err => done(err));
+            })
+            .catch(err => done(err));
+
+        });
+
+        it('should copy the log content to the new branch', function(done) {
+            url = getUrl(proj, b2, j1);
+            superagent.get(url)
+                .end((err, res) => {
+                    expect(res.text).to.equal(j1log);
+                    done(err);
+                });
+        });
+
+        it('should not copy other jobs', function(done) {
+            url = getUrl(proj, b2, j2);
+            superagent.get(url)
+                .end((err, res) => {
+                    // This log shouldn't be updated
+                    expect(res.text).to.not.equal(j2log);
+                    done(err);
+                });
+        });
+
+        it('should not change original log', function(done) {
+            url = getUrl(proj, b1, j1);
+            superagent.get(url)
+                .end((err, res) => {
+                    expect(res.text).to.equal(j1log);
+                    done(err);
+                });
+        });
+    });
+
+});

webgme-setup.json

@@ -281,6 +281,12 @@
       "xor": {
         "src": "src/seeds/xor"
       }
+    },
+    "routers": {
+      "JobLogsAPI": {
+        "src": "src/routers/JobLogsAPI",
+        "mount": "execution/logs"
+      }
     }
   },
   "dependencies": {
@@ -337,6 +343,7 @@
         "path": "node_modules/webgme-easydag/src/decorators/EllipseDecorator"
       }
     },
-    "seeds": {}
+    "seeds": {},
+    "routers": {}
   }
 }