Add reporting and improve Nextflow support

src/main/java/cloudgene/mapred/jobs/workspace/IWorkspace.java

@@ -29,10 +29,14 @@ public interface IWorkspace {
 
 	public String createFile(String name, String name2);
 
+	public String createLogFile(String name);
+
 	public String createTempFolder(String string);
 
 	public List<Download> getDownloads(String url);
 
 	public void cleanup(String job) throws IOException;
 
+	public boolean exists(String path) throws IOException;
+
 }

src/main/java/cloudgene/mapred/jobs/workspace/LocalWorkspace.java

@@ -19,10 +19,14 @@
 
 public class LocalWorkspace implements IWorkspace {
 
+	private static final String OUTPUT_DIRECTORY = "outputs";
+
 	private static final String INPUT_DIRECTORY = "input";
 
 	private static final String TEMP_DIRECTORY = "temp";
 
+	private static final String LOGS_DIRECTORY = "logs";
+
 	private static final Logger log = LoggerFactory.getLogger(LocalWorkspace.class);
 
 	private String location;
@@ -54,7 +58,7 @@ public String upload(String id, File file) throws IOException {
 		FileUtil.copy(file.getAbsolutePath(), target);
 		return target;
 	}
-	
+
 	@Override
 	public String uploadInput(String id, File file) throws IOException {
 		return upload(FileUtil.path(INPUT_DIRECTORY, id), file);
@@ -64,7 +68,7 @@ public String uploadInput(String id, File file) throws IOException {
 	public InputStream download(String path) throws IOException {
 		String absolutePath = path;
 		if (!absolutePath.startsWith("/")) {
-			 absolutePath = FileUtil.path(location, path);
+			absolutePath = FileUtil.path(location, path);
 		}
 		File file = new File(absolutePath);
 		if (file.exists()) {
@@ -74,6 +78,16 @@ public InputStream download(String path) throws IOException {
 		}
 	}
 
+	@Override
+	public boolean exists(String path) throws IOException {
+		String absolutePath = path;
+		if (!absolutePath.startsWith("/")) {
+			absolutePath = FileUtil.path(location, path);
+		}
+		File file = new File(absolutePath);
+		return file.exists();
+	}
+
 	@Override
 	public void delete(String job) throws IOException {
 
@@ -90,12 +104,12 @@ public void delete(String job) throws IOException {
 		}
 
 	}
-	
+
 	@Override
 	public void cleanup(String job) throws IOException {
-		
-		//TODO: add flag to disable cleanup (e.g. debugging)
-		
+
+		// TODO: add flag to disable cleanup (e.g. debugging)
+
 		try {
 			log.debug("Cleanup " + job + " on local workspace...");
 			String temp = FileUtil.path(location, job, TEMP_DIRECTORY);
@@ -125,14 +139,21 @@ public String getParent(String url) {
 
 	@Override
 	public String createFolder(String id) {
-		String folder = FileUtil.path(workspace, id);
+		String folder = FileUtil.path(workspace, OUTPUT_DIRECTORY, id);
 		FileUtil.createDirectory(folder);
 		return folder;
 	}
 
 	@Override
 	public String createFile(String parent, String id) {
-		String folder = FileUtil.path(workspace, parent);
+		String folder = FileUtil.path(workspace, OUTPUT_DIRECTORY, parent);
+		FileUtil.createDirectory(folder);
+		return FileUtil.path(folder, id);
+	}
+
+	@Override
+	public String createLogFile(String id) {
+		String folder = FileUtil.path(workspace, LOGS_DIRECTORY);
 		FileUtil.createDirectory(folder);
 		return FileUtil.path(folder, id);
 	}

src/main/java/cloudgene/mapred/jobs/workspace/S3Workspace.java

@@ -28,12 +28,14 @@ public class S3Workspace implements IWorkspace {
 
 	private static final String OUTPUT_DIRECTORY = "outputs";
 
-	public static long EXPIRATION_MS = 1000 * 60 * 60;
-
 	private static final String INPUT_DIRECTORY = "input";
 
+	private static final String LOGS_DIRECTORY = "logs";
+
 	private static final String TEMP_DIRECTORY = "temp";
 
+	public static long EXPIRATION_MS = 1000 * 60 * 60;
+
 	private static final Logger log = LoggerFactory.getLogger(S3Workspace.class);
 
 	private String location;
@@ -98,6 +100,13 @@ public InputStream download(String url) throws IOException {
 		return s3is;
 	}
 
+	public boolean exists(String url) {
+		String bucket = S3Util.getBucket(url);
+		String key = S3Util.getKey(url);
+		AmazonS3 s3 = S3Util.getAmazonS3();
+		return s3.doesObjectExist(bucket, key);
+	}
+
 	@Override
 	public void delete(String job) throws IOException {
 
@@ -192,6 +201,11 @@ public String createFile(String folder, String id) {
 		return location + "/" + job + "/" + OUTPUT_DIRECTORY + "/" + folder + "/" + id;
 	}
 
+	@Override
+	public String createLogFile(String id) {
+		return location + "/" + job + "/" + LOGS_DIRECTORY + "/" + id;
+	}
+
 	@Override
 	public String createTempFolder(String id) {
 		return location + "/" + job + "/" + TEMP_DIRECTORY + "/" + id;

src/main/java/cloudgene/mapred/plugins/nextflow/NextflowStep.java

@@ -90,7 +90,6 @@ public boolean run(WdlStep step, CloudgeneContext context) {
 		} else {
 			IWorkspace workspace = job.getWorkspace();
 			String workDir = workspace.createTempFolder("nextflow");
-			FileUtil.createDirectory(workDir);
 			nextflowCommand.add("-w");
 			nextflowCommand.add(workDir);
 		}
@@ -135,16 +134,30 @@ public boolean run(WdlStep step, CloudgeneContext context) {
 			return false;
 		}
 
+		IWorkspace workspace = job.getWorkspace();
+
 		nextflowCommand.add("-params-file");
 		nextflowCommand.add(paramsFile.getAbsolutePath());
 
 		nextflowCommand.add("-ansi-log");
 		nextflowCommand.add("false");
-
+		
 		nextflowCommand.add("-with-weblog");
 		nextflowCommand.add(context.getSettings().getServerUrl() + context.getSettings().getUrlPrefix()
 				+ "/api/v2/collect/" + makeSecretJobId(context.getJobId()));
-
+
+		//nextflowCommand.add("-log");
+		//nextflowCommand.add(workspace.createLogFile("nextflow.log"));
+
+		nextflowCommand.add("-with-trace");
+		nextflowCommand.add(workspace.createLogFile("trace.csv"));
+
+		nextflowCommand.add("-with-report");
+		nextflowCommand.add(workspace.createLogFile("report.html"));
+
+		nextflowCommand.add("-with-timeline");
+		nextflowCommand.add(workspace.createLogFile("timeline.html"));
+
 		StringBuilder output = new StringBuilder();
 
 		List<String> command = new Vector<String>();
@@ -260,7 +273,7 @@ private String join(List<String> array) {
 		String result = "";
 		for (int i = 0; i < array.size(); i++) {
 			if (i > 0) {
-				result += " \\\n";
+				result += " ";
 			}
 			result += array.get(i);
 		}

src/main/java/cloudgene/mapred/plugins/nextflow/NextflowTask.java

@@ -1,13 +1,12 @@
 package cloudgene.mapred.plugins.nextflow;
 
 import java.io.DataInputStream;
-import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Map;
 
 import cloudgene.mapred.jobs.CloudgeneContext;
-import cloudgene.sdk.internal.IExternalWorkspace;
+import cloudgene.mapred.jobs.workspace.IWorkspace;
 import genepi.io.FileUtil;
 import genepi.io.text.LineReader;
 
@@ -39,10 +38,13 @@ public void update(Map<String, Object> trace) throws IOException {
 		if (status.equals("COMPLETED") || status.equals("FAILED")) {
 			String workDir = (String) trace.get("workdir");
 			String logFilename = FileUtil.path(workDir, CLOUDGENE_LOG);
-			IExternalWorkspace workspace = context.getExternalWorkspace();
-			InputStream stream = workspace.download(logFilename);
-			log = FileUtil.readFileAsString(stream);
-			parseFile(logFilename);
+			IWorkspace workspace = context.getJob().getWorkspace();
+			if (workspace.exists(logFilename)) {
+				context.log("Load log file from '" + logFilename + "'");
+				InputStream stream = workspace.download(logFilename);
+				log = FileUtil.readFileAsString(stream);
+				parseFile(logFilename);
+			}
 		}
 
 	}

src/main/java/cloudgene/mapred/util/Settings.java

@@ -75,7 +75,7 @@ public class Settings {
 
 	private Map<String, String> externalWorkspace = null;
 
-	private int uploadLimit = 500;
+	private int uploadLimit = 5000;
 
 	private String googleAnalytics = "";