Handle all versions of the log stream names after Batch API changes

server/handlers/awsJobs.js

@@ -495,7 +495,8 @@ let handlers = {
                 if (job.attempts && job.attempts.length > 0) {
                     job.attempts.forEach((attempt)=> {
                         let streamObj = {
-                            name: job.jobName + '/' + job.jobId + '/' + attempt.container.taskArn.split('/').pop(),
+                            // Prior to August 21st, 2017 default was job.jobId
+                            name: job.jobName + '/default/' + attempt.container.taskArn.split('/').pop(),
                             environment: job.container.environment,
                             exitCode: job.container.exitCode
                         };

server/handlers/jobs.js

@@ -134,7 +134,11 @@ let handlers = {
             if (err) {return next(err);}
             for (let job of jobs) {
                 if (job.analysis.logstreams) {
-                    job.analysis.logstreams = job.analysis.logstreams.map(aws.cloudwatch.formatLegacyLogStream);
+                    let streamNameVersion = aws.cloudwatch.streamNameVersion(job);
+                    job.analysis.logstreams = job.analysis.logstreams.map((stream) => {
+                        // Fix legacy internal logstream values and adapt for changes in Batch names
+                        return aws.cloudwatch.formatLegacyLogStream(stream, streamNameVersion);
+                    });
                 }
             }
             if (snapshot) {

server/libs/aws/cloudwatchlogs.js

@@ -3,6 +3,7 @@ import mongo from '../../libs/mongo';
 import {ObjectID} from 'mongodb';
 import mapValuesLimit from 'async/mapValuesLimit';
 import config from '../../config';
+import moment from 'moment';
 
 let c = mongo.collections;
 
@@ -48,7 +49,7 @@ export default (aws) => {
                 if(err) {callback(err);}
                 let logStreamNames = job.analysis.logstreams || [];
                 let logStreams = logStreamNames.reduce((streams, ls) => {
-                    let stream = this.formatLegacyLogStream(ls);
+                    let stream = this.formatLegacyLogStream(ls, this.streamNameVersion(job));
                     streams[stream.name] = stream;
                     return streams;
                 }, {});
@@ -70,12 +71,65 @@ export default (aws) => {
             };
         },
 
-        formatLegacyLogStream(stream) {
+        /*
+         * Return a version value for breaking Batch API changes to log stream names
+         */
+        streamNameVersion(job) {
+            // The only way to determin which version a job uses is date started
+            if (job.analysis.created < moment('2017-08-21T14:00-07:00')) {
+                // The original stream name format
+                // appDefName / jobId / ecsTaskId
+                // FreeSurfer/eb251a5f-6314-457f-a36f-d11665451ddb/bf4fc87d-2e87-4309-abd9-998a0de708de
+                return 0;
+            } else {
+                // New stream name format
+                // appDefName / 'default' / ecsTaskId
+                // FreeSurfer/default/bf4fc87d-2e87-4309-abd9-998a0de708de
+                return 1;
+            }
+        },
+
+
+        /*
+         * Relate a name and version to the function to fix it
+         */
+        _repairStreamName(streamName, version) {
+            if (version >= 1) {
+                return this._renameStream(streamName);
+            } else {
+                return streamName;
+            }
+        },
+
+        /*
+         * The cloudwatch stream names used by Batch changed for jobs after
+         * 2017-08-21 14:00 UTC-7 and we need to translate them to support jobs
+         * submitted previously
+         */
+        _renameStream(streamName) {
+            if (streamName.indexOf('/default/') !== -1) {
+                return streamName;
+            } else {
+                let tokens = streamName.split('/');
+                return [tokens[0], 'default', tokens[2]].join('/');
+            }
+        },
+
+        /*
+         * Some jobs were created with a string instead of an object for
+         * log data
+         */
+        formatLegacyLogStream(stream, version=0) {
             if (stream instanceof Object) {
+                stream.name = this._repairStreamName(stream.name, version);
                 return stream;
             } else {
                 // If it's not an object, it should be the old string format
-                return {name: stream, environment: null, exitCode: null};
+                return {
+                    name: this._repairStreamName(stream, version),
+                    environment: null,
+                    exitCode: null
+                };
             }
         }
     };

server/tests/libs/aws.spec.js

@@ -1,4 +1,5 @@
-var assert   = require('assert');
+var assert = require('assert');
+var moment = require('moment');
 var aws = require('../../libs/aws');
 
 const subjectParam = {participant_label: ['01', '02', '03']};
@@ -90,3 +91,54 @@ describe('libs/aws/batch.js', () => {
         });
     });
 });
+
+describe('libs/aws/cloudwatchlogs.js', () => {
+    let oldStream = 'FreeSurfer/eb251a5f-6314-457f-a36f-d11665451ddb/bf4fc87d-2e87-4309-abd9-998a0de708de';
+    let newStream = 'FreeSurfer/default/bf4fc87d-2e87-4309-abd9-998a0de708de';
+    let streamObjOrig = {name: oldStream, exitCode: 0, environment: {'TEST_ENV': true}};
+    let streamObj = {};
+    beforeEach(() => {
+        // Use the copy since some tests modify it in place
+        streamObj = Object.assign({}, streamObjOrig);
+    });
+    describe('streamNameVersion()', () => {
+        let versions = {
+            0: {'analysis':
+                {'created': moment('2017-08-21T13:23-07:00')}
+            },
+            1: {'analysis':
+                {'created': moment('2017-08-21T18:14-07:00')}
+            }
+        };
+        it('should return 0 for jobs created before 2017-08-21 14:00 UTC-7', () => {
+            assert.equal(aws.cloudwatch.streamNameVersion(versions[0]), 0);
+        });
+        it('should return 1 for jobs created after 2017-08-21 14:00 UTC-7', () => {
+            assert.equal(aws.cloudwatch.streamNameVersion(versions[1]), 1);
+        });
+    });
+    describe('_batchRenameStream()', () => {
+        it('should translate old stream names to new ones', () => {
+            assert.equal(aws.cloudwatch._renameStream(oldStream), newStream);
+        });
+        it('should leave new stream names unchanged', () => {
+            assert.equal(aws.cloudwatch._renameStream(newStream), newStream);
+        });
+    });
+    describe('formatLegacyLogStream', () => {
+        it('should return a stream object with string arguments', () => {
+            assert(aws.cloudwatch.formatLegacyLogStream(oldStream) instanceof Object);
+        });
+        it('should return an object for stream object arguments', () => {
+            assert(aws.cloudwatch.formatLegacyLogStream(streamObj) instanceof Object);
+        });
+        it('should not change the format for version 0 stream objects', () => {
+            assert.deepEqual(aws.cloudwatch.formatLegacyLogStream(streamObj, 0), streamObjOrig);
+        });
+        it('should fix stream names for version 1 objects', () => {
+            let formatted = aws.cloudwatch.formatLegacyLogStream(streamObj, 1);
+            assert.notDeepEqual(formatted, streamObjOrig);
+            assert.equal(formatted.name, newStream);
+        });
+    });
+});