<commit_before>// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

/// <reference path="./definitions/async.d.ts"/>

import async = require('async');
import ctxm = require('./context');
import dm = require('./diagnostics');
import fs = require('fs');
import ifm = require('./api/interfaces');
import lm = require('./logging');
import path = require('path');
import plgm = require('./plugins');
import taskm = require('./taskmanager');
import tm = require('./tracing');
import cm = require('./common');

var shell = require('shelljs');

var trace: tm.Tracing;

//
// TODO: fix this for context API changes
//       get compiling
//       add more tracing
//       discuss ordering on log creation, associating timeline with logid
//
export class JobRunner {
	constructor(agCtx: ctxm.AgentContext, jobCtx: ctxm.JobContext) {
		this.agentContext = agCtx;
		trace = new tm.Tracing(__filename, agCtx);
		trace.enter('JobRunner');
		this.jobContext = jobCtx;
		this.job = jobCtx.job;		
	}

	private agentContext: ctxm.AgentContext;
	private jobContext: ctxm.JobContext;

	private job: ifm.JobRequestMessage;

	private _replaceTaskInputVars() {
		trace.enter('replaceTaskInputVars');

		// replace variables in inputs
		if (this.job.environment.variables) {
			this.job.tasks.forEach((task) => {
				trace.write(task.name);
				for (var key in task.inputs){
					task.inputs[key] = task.inputs[key].replaceVars(this.job.environment.variables);
				}
			});
		}
		trace.state('tasks', this.job.tasks);
	}

	public run(complete: (err:any, result: ifm.TaskResult) => void) {
		trace.enter('run');

		var ag = this.agentContext;
		this._replaceTaskInputVars();

		var _this: JobRunner = this;
		var jobCtx: ctxm.JobContext = this.jobContext;

		ag.info('Setting job to in progress');
		jobCtx.setJobInProgress();
		jobCtx.writeConsoleSection('Preparing tasks');

		jobCtx.status('Donloading required tasks');
		var taskManager = new taskm.TaskManager(ag);
		taskManager.ensureTasksExist(this.job.tasks, function(err) {
			if (err) {
				complete(err, ifm.TaskResult.Failed);
				return;
			}
			// prepare (might download) up to 5 tasks in parallel and then run tasks seuentially
			ag.status('Preparing Tasks');
			async.forEach(_this.job.tasks, 
				function(pTask, callback){
					_this.prepareTask(pTask, callback);
				},
				function(err){
			        if (err) {
			        	trace.write('error preparing tasks');
			            complete(err, ifm.TaskResult.Failed);
			            return;
			        }
			        			
					ag.info('Task preparations complete.');

					// TODO: replace build with sender id once confirm how sent

					ag.info('loading plugins...');
					plgm.load('build', ag, (err:any, plugins:any) => {
						if (err) {
							trace.write('error loading plugins');
							complete(err, ifm.TaskResult.Failed);
							return;
						}

						//
						// Write out plug-ins and tasks we are about to run.
						// Create timeline entries for each in Pending state
						//
						var order = 1;
						ag.info('beforeJob Plugins:')
						plugins['beforeJob'].forEach(function(plugin) {
							ag.info(plugin.pluginName() + ":" + plugin.beforeId);
							jobCtx.registerPendingTask(plugin.beforeId, plugin.pluginName(), order);
							order++;
						});

						ag.info('tasks:')
						jobCtx.job.tasks.forEach(function(task) {
							ag.info(task.name + ":" + task.id);
							jobCtx.registerPendingTask(task.instanceId, task.name, order);
							order++;
						});

						ag.info('afterJob Plugins:')
						plugins['afterJob'].forEach(function(plugin) {
							ag.info(plugin.name + ":" + plugin.afterId);
							jobCtx.registerPendingTask(plugin.afterId, plugin.pluginName(), order);
							order++;
						});

						ag.info('buildDirectory: ' + jobCtx.buildDirectory);
						shell.mkdir('-p', jobCtx.buildDirectory);
						shell.cd(jobCtx.buildDirectory);
						trace.write(process.cwd());

						var jobSuccess = true;
						async.series([
								function(done) {
									ag.info('Running beforeJob Plugins ...');
									plgm.beforeJob(plugins, jobCtx, ag, function(err, success) {
										ag.info('Finished running beforeJob plugins');
										trace.state('variables after plugins:', _this.job.environment.variables);

										// plugins can contribute to vars so replace again
										_this._replaceTaskInputVars();

										jobSuccess = !err && success;
										trace.write('jobSuccess: ' + jobSuccess);

										if (err) {
											ag.error(err.message);
										}

										// we always run afterJob plugins
										done(null);
									})
								},
								function(done) {
									// if prepare plugins fail, we should not run tasks (getting code failed etc...)
									if (!jobSuccess) {
										done(null);
										trace.write('skipping running tasks since prepare plugins failed.');
										return;
									}

									ag.info('Running Tasks ...');
									_this.runTasks((err: any, success: boolean) => {
										ag.info('Finished running tasks');
										jobSuccess = jobSuccess && !err && success;
										trace.write('jobSuccess: ' + jobSuccess);

										if (err) {
											ag.error(err.message);
										}

										done(null);
									});
								},
								function(done) {
									ag.info('Running afterJob Plugins ...');
									plgm.afterJob(plugins, jobCtx, ag, jobSuccess, function(err, success) {
										ag.info('Finished running afterJob plugins');
										jobSuccess = jobSuccess && !err && success;
										trace.write('jobSuccess: ' + jobSuccess);

										if (err) {
											ag.error(err.message);
	                                    }

	                                    done(err);
									});
								}			
							], 
	                        function (err) {
								var jobResult = jobSuccess ? ifm.TaskResult.Succeeded : ifm.TaskResult.Failed;
								trace.write('jobResult: ' + jobResult);

								if (err) {
									jobResult = ifm.TaskResult.Failed; 
								}

								complete(err, jobResult);
							});
					});

			});
		});
	}

	private runTasks(callback: (err:any, success:boolean) => void): void {
		trace.enter('runTasks');

		var job: ifm.JobRequestMessage = this.job;
		var ag = this.agentContext;
		var jobCtx: ctxm.JobContext = this.jobContext;

		var success = true;
		var _this: JobRunner = this;

		trace.state('tasks', job.tasks);
		async.forEachSeries(job.tasks,
			function(item: ifm.TaskInstance, done: (err:any) => void){
				
				jobCtx.writeConsoleSection('Running ' + item.name);
				var taskCtx: ctxm.TaskContext = new ctxm.TaskContext(jobCtx.jobInfo, 
					                                                 item.instanceId,
					                                                 jobCtx.feedback, 
					                                                 ag);

				taskCtx.on('message', function(message) {
					jobCtx.feedback.queueConsoleLine(message);
				});

				shell.cd(taskCtx.buildDirectory);
				jobCtx.setTaskStarted(item.instanceId, item.name);
				_this.runTask(item, taskCtx, function(err) {

					var taskResult: ifm.TaskResult = ifm.TaskResult.Succeeded;
					if (err || taskCtx.hasErrors) {
						success = false;
						taskResult = ifm.TaskResult.Failed;
					}

					trace.write('taskResult: ' + taskResult);
					jobCtx.setTaskResult(item.instanceId, item.name, taskResult);

					taskCtx.end();
					done(err);
				});
			}, function(err) {
				ag.info('Done running tasks.');
				trace.write('jobSucess: ' + success);

				if (err) {
	                ag.error(err.message);
	                callback(err, success);
	                return;		                
				}
				
				callback(null, success);
			});		
	}

	private taskExecution = {};
	private taskMetadata = {};

	private prepareTask(task: ifm.TaskInstance, callback) {
		trace.enter('prepareTask');

		var ag = this.agentContext;

		var taskPath = path.join(this.jobContext.workingDirectory, 'tasks', task.name, task.version);
		trace.write('taskPath: ' + taskPath);

		ag.info('preparing task ' + task.name);

		var taskJsonPath = path.join(taskPath, 'task.json');
		trace.write('taskJsonPath: ' + taskJsonPath);

		fs.readFile(taskJsonPath, 'utf8', (err, data) => {
			if (err) {
				trace.write('error reading: ' + taskJsonPath);
				callback(err);
				return;
			}

			// TODO: task metadata should be typed
			try {
				var taskMetadata = JSON.parse(data);
				trace.state('taskMetadata', taskMetadata);
				this.taskMetadata[task.id] = taskMetadata;

				var execution = taskMetadata.execution;

				var handlers = ['JavaScript', 'Python', 'ShellScript'];
				var instructions;
				handlers.some(function(handler) {
					if (execution.hasOwnProperty(handler)) {
						instructions = execution[handler];
						instructions["handler"] = handler;

						trace.write('handler: ' + handler);

						return true;
					}
				});
				
				if (!instructions) {
					trace.write('no handler for this task');
					callback(new Error('Agent could not find a handler for this task'));
					return;
				}

				instructions['target'] = path.join(taskPath, instructions.target);
				trace.state('instructions', instructions);<commit_msg>Fixing issue where agent could not handle BOM in task.json files.<commit_after>// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

/// <reference path="./definitions/async.d.ts"/>

import async = require('async');
import ctxm = require('./context');
import dm = require('./diagnostics');
import fs = require('fs');
import ifm = require('./api/interfaces');
import lm = require('./logging');
import path = require('path');
import plgm = require('./plugins');
import taskm = require('./taskmanager');
import tm = require('./tracing');
import cm = require('./common');

var shell = require('shelljs');

var trace: tm.Tracing;

//
// TODO: fix this for context API changes
//       get compiling
//       add more tracing
//       discuss ordering on log creation, associating timeline with logid
//
export class JobRunner {
	constructor(agCtx: ctxm.AgentContext, jobCtx: ctxm.JobContext) {
		this.agentContext = agCtx;
		trace = new tm.Tracing(__filename, agCtx);
		trace.enter('JobRunner');
		this.jobContext = jobCtx;
		this.job = jobCtx.job;		
	}

	private agentContext: ctxm.AgentContext;
	private jobContext: ctxm.JobContext;

	private job: ifm.JobRequestMessage;

	private _replaceTaskInputVars() {
		trace.enter('replaceTaskInputVars');

		// replace variables in inputs
		if (this.job.environment.variables) {
			this.job.tasks.forEach((task) => {
				trace.write(task.name);
				for (var key in task.inputs){
					task.inputs[key] = task.inputs[key].replaceVars(this.job.environment.variables);
				}
			});
		}
		trace.state('tasks', this.job.tasks);
	}

	public run(complete: (err:any, result: ifm.TaskResult) => void) {
		trace.enter('run');

		var ag = this.agentContext;
		this._replaceTaskInputVars();

		var _this: JobRunner = this;
		var jobCtx: ctxm.JobContext = this.jobContext;

		ag.info('Setting job to in progress');
		jobCtx.setJobInProgress();
		jobCtx.writeConsoleSection('Preparing tasks');

		jobCtx.status('Donloading required tasks');
		var taskManager = new taskm.TaskManager(ag);
		taskManager.ensureTasksExist(this.job.tasks, function(err) {
			if (err) {
				complete(err, ifm.TaskResult.Failed);
				return;
			}
			// prepare (might download) up to 5 tasks in parallel and then run tasks seuentially
			ag.status('Preparing Tasks');
			async.forEach(_this.job.tasks, 
				function(pTask, callback){
					_this.prepareTask(pTask, callback);
				},
				function(err){
			        if (err) {
			        	trace.write('error preparing tasks');
			            complete(err, ifm.TaskResult.Failed);
			            return;
			        }
			        			
					ag.info('Task preparations complete.');

					// TODO: replace build with sender id once confirm how sent

					ag.info('loading plugins...');
					plgm.load('build', ag, (err:any, plugins:any) => {
						if (err) {
							trace.write('error loading plugins');
							complete(err, ifm.TaskResult.Failed);
							return;
						}

						//
						// Write out plug-ins and tasks we are about to run.
						// Create timeline entries for each in Pending state
						//
						var order = 1;
						ag.info('beforeJob Plugins:')
						plugins['beforeJob'].forEach(function(plugin) {
							ag.info(plugin.pluginName() + ":" + plugin.beforeId);
							jobCtx.registerPendingTask(plugin.beforeId, plugin.pluginName(), order);
							order++;
						});

						ag.info('tasks:')
						jobCtx.job.tasks.forEach(function(task) {
							ag.info(task.name + ":" + task.id);
							jobCtx.registerPendingTask(task.instanceId, task.name, order);
							order++;
						});

						ag.info('afterJob Plugins:')
						plugins['afterJob'].forEach(function(plugin) {
							ag.info(plugin.name + ":" + plugin.afterId);
							jobCtx.registerPendingTask(plugin.afterId, plugin.pluginName(), order);
							order++;
						});

						ag.info('buildDirectory: ' + jobCtx.buildDirectory);
						shell.mkdir('-p', jobCtx.buildDirectory);
						shell.cd(jobCtx.buildDirectory);
						trace.write(process.cwd());

						var jobSuccess = true;
						async.series([
								function(done) {
									ag.info('Running beforeJob Plugins ...');
									plgm.beforeJob(plugins, jobCtx, ag, function(err, success) {
										ag.info('Finished running beforeJob plugins');
										trace.state('variables after plugins:', _this.job.environment.variables);

										// plugins can contribute to vars so replace again
										_this._replaceTaskInputVars();

										jobSuccess = !err && success;
										trace.write('jobSuccess: ' + jobSuccess);

										if (err) {
											ag.error(err.message);
										}

										// we always run afterJob plugins
										done(null);
									})
								},
								function(done) {
									// if prepare plugins fail, we should not run tasks (getting code failed etc...)
									if (!jobSuccess) {
										done(null);
										trace.write('skipping running tasks since prepare plugins failed.');
										return;
									}

									ag.info('Running Tasks ...');
									_this.runTasks((err: any, success: boolean) => {
										ag.info('Finished running tasks');
										jobSuccess = jobSuccess && !err && success;
										trace.write('jobSuccess: ' + jobSuccess);

										if (err) {
											ag.error(err.message);
										}

										done(null);
									});
								},
								function(done) {
									ag.info('Running afterJob Plugins ...');
									plgm.afterJob(plugins, jobCtx, ag, jobSuccess, function(err, success) {
										ag.info('Finished running afterJob plugins');
										jobSuccess = jobSuccess && !err && success;
										trace.write('jobSuccess: ' + jobSuccess);

										if (err) {
											ag.error(err.message);
	                                    }

	                                    done(err);
									});
								}			
							], 
	                        function (err) {
								var jobResult = jobSuccess ? ifm.TaskResult.Succeeded : ifm.TaskResult.Failed;
								trace.write('jobResult: ' + jobResult);

								if (err) {
									jobResult = ifm.TaskResult.Failed; 
								}

								complete(err, jobResult);
							});
					});

			});
		});
	}

	private runTasks(callback: (err:any, success:boolean) => void): void {
		trace.enter('runTasks');

		var job: ifm.JobRequestMessage = this.job;
		var ag = this.agentContext;
		var jobCtx: ctxm.JobContext = this.jobContext;

		var success = true;
		var _this: JobRunner = this;

		trace.state('tasks', job.tasks);
		async.forEachSeries(job.tasks,
			function(item: ifm.TaskInstance, done: (err:any) => void){
				
				jobCtx.writeConsoleSection('Running ' + item.name);
				var taskCtx: ctxm.TaskContext = new ctxm.TaskContext(jobCtx.jobInfo, 
					                                                 item.instanceId,
					                                                 jobCtx.feedback, 
					                                                 ag);

				taskCtx.on('message', function(message) {
					jobCtx.feedback.queueConsoleLine(message);
				});

				shell.cd(taskCtx.buildDirectory);
				jobCtx.setTaskStarted(item.instanceId, item.name);
				_this.runTask(item, taskCtx, function(err) {

					var taskResult: ifm.TaskResult = ifm.TaskResult.Succeeded;
					if (err || taskCtx.hasErrors) {
						success = false;
						taskResult = ifm.TaskResult.Failed;
					}

					trace.write('taskResult: ' + taskResult);
					jobCtx.setTaskResult(item.instanceId, item.name, taskResult);

					taskCtx.end();
					done(err);
				});
			}, function(err) {
				ag.info('Done running tasks.');
				trace.write('jobSucess: ' + success);

				if (err) {
	                ag.error(err.message);
	                callback(err, success);
	                return;		                
				}
				
				callback(null, success);
			});		
	}

	private taskExecution = {};
	private taskMetadata = {};

	private prepareTask(task: ifm.TaskInstance, callback) {
		trace.enter('prepareTask');

		var ag = this.agentContext;

		var taskPath = path.join(this.jobContext.workingDirectory, 'tasks', task.name, task.version);
		trace.write('taskPath: ' + taskPath);

		ag.info('preparing task ' + task.name);

		var taskJsonPath = path.join(taskPath, 'task.json');
		trace.write('taskJsonPath: ' + taskJsonPath);

		fs.exists(taskJsonPath,  (exists) => {
			if (!exists) {
				trace.write('cannot find task: ' + taskJsonPath);
				callback(new Error('Could not find task: ' + taskJsonPath));
				return;
			}

			// TODO: task metadata should be typed
			try {
				// Using require to help strip out BOM information
				// Using JSON.stringify/JSON.parse in order to clone the task.json object
				// Otherwise we modify the cached object for everyone
				var taskMetadata = JSON.parse(JSON.stringify(require(taskJsonPath)));
				trace.state('taskMetadata', taskMetadata);
				this.taskMetadata[task.id] = taskMetadata;

				var execution = taskMetadata.execution;

				var handlers = ['JavaScript', 'Python', 'ShellScript'];
				var instructions;
				handlers.some(function(handler) {
					if (execution.hasOwnProperty(handler)) {
						instructions = execution[handler];
						instructions["handler"] = handler;

						trace.write('handler: ' + handler);

						return true;
					}
				});
				
				if (!instructions) {
					trace.write('no handler for this task');
					callback(new Error('Agent could not find a handler for this task'));
					return;
				}

				instructions['target'] = path.join(taskPath, instructions.target);
				trace.state('instructions', instructions);
{'fraction_non_alphanumeric': 0.13485775059824515, 'fraction_numerical': 0.00026588673225206064, 'mean_word_length': 2.4348858447488584, 'pattern_counts': {'":': 6, '<': 5, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '459961', 'n_tokens_mistral': 7413, 'n_tokens_neox': 5456, 'n_words': 1222}