001/* 002 * Copyright (c) 2004-2010 The Regents of the University of California. 003 * All rights reserved. 004 * 005 * '$Author: crawl $' 006 * '$Date: 2012-07-27 18:35:29 +0000 (Fri, 27 Jul 2012) $' 007 * '$Revision: 30295 $' 008 * 009 * Permission is hereby granted, without written agreement and without 010 * license or royalty fees, to use, copy, modify, and distribute this 011 * software and its documentation for any purpose, provided that the above 012 * copyright notice and the following two paragraphs appear in all copies 013 * of this software. 014 * 015 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY 016 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES 017 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF 018 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF 019 * SUCH DAMAGE. 020 * 021 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES, 022 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF 023 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE 024 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF 025 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, 026 * ENHANCEMENTS, OR MODIFICATIONS. 027 * 028 */ 029 030package org.kepler.job; 031 032import org.apache.commons.logging.Log; 033import org.apache.commons.logging.LogFactory; 034 035/** 036 * Support class for NCCS job manager support Class Job uses the methods of a 037 * supporter class to submit jobs and check status 038 */ 039public class JobSupportNCCS implements JobSupport { 040 041 private static final Log log = LogFactory.getLog(JobSupportNCCS.class 042 .getName()); 043 private static final boolean isDebugging = log.isDebugEnabled(); 044 045 public JobSupportNCCS() { 046 } 047 048 public void init(String nccsBinPath) { 049 if (nccsBinPath != null && !nccsBinPath.trim().equals("")) { 050 String binPath = new String(nccsBinPath); 051 if (!nccsBinPath.endsWith("/")) 052 binPath += "/"; 053 _nccsSubmitCmd = binPath + _nccsSubmitCmd; 054 _nccsStatusCmd = binPath + _nccsStatusCmd; 055 _nccsDeleteCmd = binPath + _nccsDeleteCmd; 056 } 057 } 058 059 /** 060 * Create a submission file for the specific job manager, based on the 061 * information available in Job: - executable name - input files - output 062 * files - arguments for the job 063 */ 064 public boolean createSubmitFile(String filename, Job job) { 065 066 return false; 067 } 068 069 /** 070 * Submit command for NCCS return: the command for submission 071 */ 072 public String getSubmitCmd(String submitFile, String options, Job job) throws JobException { 073 074 if(job.getDependentJobs() != null) { 075 throw new JobException("Support for job dependencies with NCCS has not been implemented."); 076 } 077 078 String _commandStr; 079 if (options != null) 080 _commandStr = _nccsSubmitCmd + " " + options + " " + submitFile; 081 else 082 _commandStr = _nccsSubmitCmd + " " + submitFile; 083 084 return _commandStr; 085 } 086 087 /** 088 * Parse output of submission and get information: jobID return String jobID 089 * on success throws JobException at failure (will contain the error stream 090 * or output stream) 091 */ 092 public String parseSubmitOutput(String output, String error) 093 throws JobException { 094 095 // System.out.println("====NCCS parse: picking the jobid from output..."); 096 /* 097 * NCCS qsub output is simple: on success, it is the jobID in one single 098 * line. if submitfile does not exists or other error, messages are 099 * printed on stdout stderr is empty 100 */ 101 String jobID = null; 102 int idx = output.indexOf("\n"); 103 104 if (idx > -1) { 105 String firstrow = output.substring(0, idx); 106 if (firstrow.matches("[0-9]*.*")) { 107 jobID = firstrow; 108 } 109 if (isDebugging) 110 log.debug("NCCS parse: jobID = " + jobID + " firstrow = " 111 + firstrow); 112 } 113 114 if (jobID == null) { 115 if (error != null && error.length() > 0) 116 throw new JobException("Error at submission of NCCS job: " 117 + error); 118 else 119 throw new JobException("Error at submission of NCCS job: " 120 + output); 121 } 122 return jobID; 123 } // end-of-submit 124 125 /** 126 * Get the command to ask the status of the job return: the String of 127 * command 128 */ 129 public String getStatusCmd(String jobID) { 130 String _commandStr = _nccsStatusCmd + jobID; 131 return _commandStr; 132 } 133 134 /** 135 * Parse output of status check command and get status info return: a 136 * JobStatusInfo object, or throws an JobException with the error output 137 */ 138 public JobStatusInfo parseStatusOutput(String jobID, int exitCode, 139 String output, String error) throws JobException { 140 141 // NCCS status does not use exitCode. It can show error, but in real it 142 // can mean only that 143 // job is not in the queue anymore, which is good... 144 145 // System.out.println("+++++ status: picking the status from output" ); 146 JobStatusInfo stat = new JobStatusInfo(); 147 stat.statusCode = JobStatusCode.NotInQueue; 148 149 boolean foundStatus = false; 150 151 String sa[] = output.split("\n"); 152 int idx; 153 for (int i = 0; i < sa.length; i++) { 154 // System.out.println("NCCS status string " + i + " = "+ sa[i]); 155 String vals[] = sa[i].trim().split("( )+", 9); 156 if (jobID.startsWith(vals[0].trim())) { // jobID may be longer than 157 // the first field which is 158 // limited in length 159 if (vals.length >= 5) { 160 stat.jobID = jobID; 161 String jobName = vals[1].trim(); 162 stat.owner = vals[2].trim(); 163 stat.runTime = vals[3].trim(); 164 String sts = vals[4].trim(); 165 switch (sts.charAt(0)) { 166 case 'R': 167 case 'E': 168 stat.statusCode = JobStatusCode.Running; 169 break; 170 case 'Q': 171 case 'H': 172 case 'T': 173 case 'W': 174 case 'S': 175 stat.statusCode = JobStatusCode.Wait; 176 break; 177 default: 178 stat.statusCode = JobStatusCode.Wait; 179 } 180 foundStatus = true; 181 if (isDebugging) 182 log.debug("NCCS status Values: jobid=" + stat.jobID 183 + " owner=" + stat.owner + " runTime=" 184 + stat.runTime + " status=[" + sts + "]"); 185 } 186 } 187 } 188 // System.out.println("NCCS status = " + stat.statusCode); 189 190 if (!foundStatus) { 191 if (error != null && error.length() > 0) { 192 // it can be the message: qstat: Unknown Job Id ... 193 if (error.startsWith("qstat: Unknown Job Id")) { 194 stat.jobID = jobID; 195 stat.statusCode = JobStatusCode.NotInQueue; 196 } else { 197 log.warn("Error string = [" + error + "] len=" 198 + error.length()); 199 stat.jobID = jobID; 200 stat.statusCode = JobStatusCode.Error; 201 } 202 } else { // not an error, just job is not in the job queue now 203 stat.jobID = jobID; 204 stat.statusCode = JobStatusCode.NotInQueue; 205 } 206 } 207 208 return stat; 209 } 210 211 /** 212 * Get the command to remove a job from queue (either running or waiting 213 * jobs). return: the String of command 214 */ 215 public String getDeleteCmd(String jobID) { 216 String _commandStr = _nccsDeleteCmd + jobID; 217 return _commandStr; 218 } 219 220 /** 221 * Parse output of delete command. return: true or false indicating that the 222 * command was successful or not 223 */ 224 public boolean parseDeleteOutput(String jobID, int exitCode, String output, 225 String error) throws JobException { 226 if (exitCode == 0) 227 return true; 228 else 229 return false; 230 } 231 232 // //////////////////////////////////////////////////////////////////// 233 // // private variables //// 234 235 // The combined command to execute. 236 private String _nccsSubmitCmd = "qsub "; 237 private String _nccsStatusCmd = "qstat "; 238 private String _nccsDeleteCmd = "qdel "; 239 240 public String getTaskStatusCmd(String jobID) throws NotSupportedException { 241 throw new NotSupportedException("Task parallel jobs are not supported"); 242 } 243 244 public TaskParallelJobStatusInfo parseTaskStatusOutput(String jobID, 245 int numTasks, int exitCode, String output, String error) 246 throws JobException, NotSupportedException { 247 throw new NotSupportedException("Task parallel jobs are not supported"); 248 } 249 250} // end-of-class-JobSupportNCCS