Merge pull request #50 from ZihengSun/master

Fix the process stop, python package copy, and call between python

WebContent/geoweaver/js/edu.gmu.csiss.geoweaver.js

@@ -18,7 +18,7 @@ edu = {
 
 					sponsor: "ESIPLab incubator project",
 
-					version: "0.7.7",
+					version: "0.7.8",
 
 					author: "Ziheng Sun",
 

WebContent/geoweaver/js/edu.gmu.csiss.geoweaver.process.js

@@ -564,6 +564,10 @@ edu.gmu.csiss.geoweaver.process = {
 
 				status_col = "      <td id=\"status_"+hid+"\"><span class=\"label label-warning\">Running</span></td> ";
 
+			}else if(status == "Stopped"){
+
+				status_col = "      <td id=\"status_"+hid+"\"><span class=\"label label-default\">Stopped</span></td> ";
+
 			}else{
 
 				status_col = "      <td id=\"status_"+hid+"\"><span class=\"label label-primary\">Unknown</span></td> ";
@@ -737,7 +741,7 @@ edu.gmu.csiss.geoweaver.process = {
 
 //					<span id=\"status_"+msg[i].id+"\" class=\"label label-warning\">Pending</span>
 
-					$("#status_" + history_id).html("<span class=\"label label-success\">Stopped</span>");
+					$("#status_" + history_id).html("<span class=\"label label-default\">Stopped</span>");
 
 				}else{
 

pom.xml

@@ -1,7 +1,7 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <artifactId>Geoweaver</artifactId>
-    <version>0.7.7</version>
+    <version>0.7.8</version>
     <packaging>war</packaging>
     <name>Geoweaver</name>
     <description>ESIPLab incubator project</description>
@@ -54,6 +54,11 @@
 	        <version>4.12</version>
 	        <scope>test</scope>
 	    </dependency>
+	    <dependency>
+		    <groupId>org.kamranzafar</groupId>
+		    <artifactId>jtar</artifactId>
+		    <version>2.3</version>
+		</dependency>
         <dependency>
             <groupId>aopalliance</groupId>
             <artifactId>aopalliance</artifactId>

src/edu/gmu/csiss/earthcube/cyberconnector/ssh/FileTool.java

@@ -15,6 +15,7 @@
 import net.schmizz.sshj.sftp.RemoteResourceInfo;
 import net.schmizz.sshj.sftp.SFTPClient;
 import net.schmizz.sshj.xfer.FilePermission;
+import net.schmizz.sshj.xfer.scp.SCPFileTransfer;
 
 public class FileTool {
 
@@ -24,6 +25,79 @@ public class FileTool {
 
 	static Map<String, SFTPClient> token2ftpclient = new HashMap();
 
+	/**
+	 * Upload a file from local to a specific location on a remote host
+	 * @param hid
+	 * @param passwd
+	 * @param localPath
+	 * @param remoteLoc
+	 * @return
+	 */
+	public static String scp_upload(String hid, String passwd, String localPath, String remoteLoc, boolean removelocal) {
+
+		String resp = null;
+
+		SSHSession session = new SSHSessionImpl();
+
+		try {
+
+			//get host ip, port, user name and password
+
+//			String[] hostdetails = HostTool.getHostDetailsById(hid);
+
+			//establish SSH session and generate a token for it
+
+			session.login(hid, passwd, null, false);
+
+			File localfile = new File(localPath);
+
+			String filename = localfile.getName();
+
+			String fileloc = remoteLoc + "/" + filename; //upload file to temporary folder
+
+			log.info("upload " + localPath + " to " + remoteLoc);
+
+//			session.getSsh().newSCPFileTransfer().download(file_path, fileloc);
+
+			session.getSsh().newSCPFileTransfer().upload(localPath, remoteLoc);
+
+			//remove the local temporal files
+
+			if(removelocal) localfile.delete();
+
+			session.getSsh().close();
+
+//			session.getSSHJSession().newSCPFileTransfer().download("test_file", new FileSystemFile("/tmp/"));
+
+//			session.runBash(code, id, false); 
+
+			String file_url = null;
+
+			resp = "{\"filename\": \"" + fileloc + "\", \"ret\": \"success\"}";
+
+		} catch (Exception e) {
+
+			e.printStackTrace();
+
+			throw new RuntimeException(e.getLocalizedMessage());
+
+		}  finally {
+
+
+
+		}
+
+		return resp;
+
+	}
+
+	/**
+	 * Upload a file from local to the home directory of remote host
+	 * @param hid
+	 * @param passwd
+	 * @param localPath
+	 * @return
+	 */
 	public static String scp_upload(String hid, String passwd, String localPath) {
 
 		String resp = null;
@@ -50,7 +124,9 @@ public static String scp_upload(String hid, String passwd, String localPath) {
 
 //			session.getSsh().newSCPFileTransfer().download(file_path, fileloc);
 
-			session.getSsh().newSCPFileTransfer().upload(localPath, fileloc);
+			SCPFileTransfer transfer = session.getSsh().newSCPFileTransfer();
+
+			transfer.upload(localPath, fileloc);
 
 			//remove the local temporal files
 

src/edu/gmu/csiss/earthcube/cyberconnector/ssh/ProcessTool.java

@@ -1,7 +1,10 @@
 package edu.gmu.csiss.earthcube.cyberconnector.ssh;
 
+import java.io.File;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.lang3.StringUtils;
@@ -367,14 +370,47 @@ public static String del(String id) {
 
 	}
 
+	/**
+	 * Get process name by id
+	 * @param pid
+	 * @return
+	 */
+	public static String getNameById(String pid) {
+
+		StringBuffer sql = new StringBuffer("select name from process_type where id = '").append(pid).append("';");
+
+		ResultSet rs = DataBaseOperation.query(sql.toString());
+
+		String name = null;
+
+		try {
+
+			if(rs.next()) {
+
+				name = rs.getString("name");
+
+			}
+
+			DataBaseOperation.closeConnection();
+
+		} catch (SQLException e) {
+
+			e.printStackTrace();
+
+		}
+
+		return name;
+
+	}
+
 	/**
 	 * Get code by Id
 	 * @param pid
 	 * @return
 	 */
 	public static String getCodeById(String pid) {
 
-		StringBuffer sql = new StringBuffer("select code from process_type where id = '").append(pid).append("';");
+		StringBuffer sql = new StringBuffer("select name,code from process_type where id = '").append(pid).append("';");
 
 		ResultSet rs = DataBaseOperation.query(sql.toString());
 
@@ -538,12 +574,12 @@ public static String executeShell(String id, String hid, String pswd, String tok
 
 			session.login(hid, pswd, token, false);
 
-			GeoweaverController.sshSessionManager.sessionsByToken.put(token, session);
-
-			session.runBash(code, id, isjoin); 
+			session.runBash(code, id, isjoin, token); 
 
 			String historyid = session.getHistory_id();
 
+			GeoweaverController.sshSessionManager.sessionsByToken.put(token, session);
+
 			resp = "{\"history_id\": \""+historyid+
 
 					"\", \"token\": \""+token+
@@ -610,7 +646,7 @@ public static String executeJupyterProcess(String id, String hid, String pswd, S
 
 			GeoweaverController.sshSessionManager.sessionsByToken.put(token, session);
 
-			session.runJupyter(code, id, isjoin, bin, pyenv, basedir); 
+			session.runJupyter(code, id, isjoin, bin, pyenv, basedir, token); 
 
 			String historyid = session.getHistory_id();
 
@@ -732,12 +768,82 @@ public static String executeBuiltInProcess(String id, String hid, String pswd, S
 
 	}
 
+	/**
+	 * Package all python files into one zip file
+	 */
+	public static String packageAllPython(String hid) {
+
+		StringBuffer sql = new StringBuffer("select name,code from process_type where description = 'python';");
+
+		logger.info(sql.toString());
+
+		ResultSet rs = DataBaseOperation.query(sql.toString());
+
+		String resp = null, code = null, name = null;
+
+		try {
+
+			String folderpath = BaseTool.getCyberConnectorRootPath() + SysDir.temp_file_path + "/" + hid + "/";
+
+			resp = BaseTool.getCyberConnectorRootPath() + SysDir.temp_file_path + "/" + hid + ".tar";
+
+			new File(folderpath).mkdirs(); //make a temporary folder
+
+			List<String> files = new ArrayList();
+
+			while(rs.next()) {
+
+				code = rs.getString("code");
+
+				name = rs.getString("name");
+
+				String filepath = folderpath;
+
+				if(name.endsWith(".py")) {
+
+					filepath += name;
+
+				}else{
+
+					filepath += name + ".py";
+
+				}
+
+				logger.info(filepath);
+
+				BaseTool.writeString2File(code, filepath);
+
+				files.add(filepath);
+
+			}
+
+			if(files.size()==0) {
+
+				throw new RuntimeException("No python is found in the database");
+
+			}
+			//zip the files into a tar file
+			BaseTool.tar(files, resp);
+
+			DataBaseOperation.closeConnection();
+
+		} catch (SQLException e) {
+
+			e.printStackTrace();
+
+		}
+
+		return resp;
+
+	}
+
 	/**
 	 * Execute Python process
 	 * @param id
 	 * @param hid
 	 * @param pswd
 	 * @param token
+	 * for security reasons
 	 * @param isjoin
 	 * @return
 	 */
@@ -748,31 +854,44 @@ public static String executePythonProcess(String id, String hid, String pswd,
 
 		try {
 
+			if(token == null) {
+
+				token = new RandomString(12).nextString();
+
+			}
+
+			//package all the python files into a tar
+			String packagefile = ProcessTool.packageAllPython(token);
+
+			if(basedir!=null) {
+
+				FileTool.scp_upload(hid, pswd, packagefile, basedir, true);
+
+			}else {
+
+				FileTool.scp_upload(hid, pswd, packagefile);
+
+			}
+
 			//get code of the process
 
 			String code = getCodeById(id);
 
-			logger.info(code);
+//			logger.info(code);
 
 			//get host ip, port, user name and password
 
 //			String[] hostdetails = HostTool.getHostDetailsById(hid);
 
 			//establish SSH session and generate a token for it
 
-			if(token == null) {
-
-				token = new RandomString(12).nextString();
-
-			}
-
 			SSHSession session = new SSHSessionImpl();
 
 			session.login(hid, pswd, token, false);
 
 			GeoweaverController.sshSessionManager.sessionsByToken.put(token, session);
 
-			session.runPython(code, id, isjoin, bin, pyenv, basedir); 
+			session.runPython(code, id, isjoin, bin, pyenv, basedir, token); 
 
 			String historyid = session.getHistory_id();
 

src/edu/gmu/csiss/earthcube/cyberconnector/ssh/SSHSession.java

@@ -73,11 +73,11 @@ public interface SSHSession {
 
 	public void setWebSocketSession(WebSocketSession session);
 
-	public void runBash(String script, String processid, boolean isjoin);
+	public void runBash(String script, String processid, boolean isjoin, String token);
 
-	public void runJupyter(String script, String processid, boolean isjoin, String bin, String env, String basedir);
+	public void runJupyter(String script, String processid, boolean isjoin, String bin, String env, String basedir, String token);
 
-	public void runPython(String script, String processid, boolean isjoin, String bin, String pyenv, String basedir);
+	public void runPython(String script, String processid, boolean isjoin, String bin, String pyenv, String basedir, String token);
 
 	public void runMultipleBashes(String[] script, String processid);