webservices/compbio/stat/collector/ExecutionStatCollector.java

   1 package compbio.stat.collector;\r
   2 \r
   3 import java.io.File;\r
   4 import java.io.FileFilter;\r
   5 import java.io.IOException;\r
   6 import java.sql.SQLException;\r
   7 import java.text.SimpleDateFormat;\r
   8 import java.util.ArrayList;\r
   9 import java.util.Date;\r
  10 import java.util.HashMap;\r
  11 import java.util.HashSet;\r
  12 import java.util.List;\r
  13 import java.util.Map;\r
  14 import java.util.Set;\r
  15 \r
  16 import org.apache.log4j.Logger;\r
  17 \r
  18 import compbio.engine.client.Executable;\r
  19 import compbio.engine.client.PathValidator;\r
  20 import compbio.engine.client.SkeletalExecutable;\r
  21 import compbio.metadata.JobStatus;\r
  22 import compbio.util.FileUtil;\r
  23 import compbio.ws.client.Services;\r
  24 \r
  25 /**\r
  26  * Number of runs of each WS = number of folders with name\r
  27  * \r
  28  * Number of successful runs = all runs with no result file\r
  29  * \r
  30  * Per period of time = limit per file creating time Runtime (avg/max) =\r
  31  * \r
  32  * started time - finished time\r
  33  * \r
  34  * Task & result size = result.size\r
  35  * \r
  36  * Abandoned runs - not collected runs\r
  37  * \r
  38  * Cancelled runs - cancelled\r
  39  * \r
  40  * Cluster vs local runs\r
  41  * \r
  42  * Reasons for failure = look in the err out?\r
  43  * \r
  44  * \r
  45  * Metadata required:\r
  46  * \r
  47  * work directory for local and cluster tasks = from Helper or cmd parameter. WS\r
  48  * names - enumeration. Status file names and content.\r
  49  * \r
  50  * @author pvtroshin\r
  51  * \r
  52  */\r
  53 public class ExecutionStatCollector implements Runnable {\r
  54 \r
  55         static final int UNDEFINED = -1;\r
  56 \r
  57         private static final Logger log = Logger\r
  58                         .getLogger(ExecutionStatCollector.class);\r
  59 \r
  60         static SimpleDateFormat DF = new SimpleDateFormat("dd/MM/yyyy hh:mm:ss");\r
  61 \r
  62         final private File workDirectory;\r
  63         final private List<JobStat> stats;\r
  64         /**\r
  65          * Consider the job that has been working for longer than timeOutInHours\r
  66          * completed, whatever the outcome\r
  67          */\r
  68         final private int timeOutInHours;\r
  69 \r
  70         /**\r
  71          * List subdirectories in the job directory\r
  72          * \r
  73          * @param workDirectory\r
  74          * @param timeOutInHours\r
  75          */\r
  76         public ExecutionStatCollector(String workDirectory, int timeOutInHours) {\r
  77                 log.info("Starting stat collector for directory: " + workDirectory);\r
  78                 log.info("Maximum allowed runtime(h): " + timeOutInHours);\r
  79                 if (!PathValidator.isValidDirectory(workDirectory)) {\r
  80                         throw new IllegalArgumentException("workDirectory '"\r
  81                                         + workDirectory + "' does not exist!");\r
  82                 }\r
  83                 this.workDirectory = new File(workDirectory);\r
  84                 stats = new ArrayList<JobStat>();\r
  85                 if (timeOutInHours <= 0) {\r
  86                         throw new IllegalArgumentException(\r
  87                                         "Timeout value must be greater than 0! Given value: "\r
  88                                                         + timeOutInHours);\r
  89                 }\r
  90                 this.timeOutInHours = timeOutInHours;\r
  91         }\r
  92 \r
  93         boolean hasCompleted(JobDirectory jd) {\r
  94                 JobStat jstat = jd.getJobStat();\r
  95                 if (jstat.hasResult() || jstat.getIsCancelled()\r
  96                                 || jstat.getIsFinished() || hasTimedOut(jd)) {\r
  97                         return true;\r
  98                 }\r
  99                 return false;\r
 100         }\r
 101 \r
 102         boolean hasTimedOut(JobDirectory jd) {\r
 103                 return ((System.currentTimeMillis() - jd.jobdir.lastModified()) / (1000 * 60 * 60)) > timeOutInHours;\r
 104         }\r
 105 \r
 106         /*\r
 107          * Make sure that collectStatistics methods was called prior to calling\r
 108          * this! TODO consider running collectStatistics from here on the first call\r
 109          */\r
 110         StatProcessor getStats() {\r
 111                 if (stats.isEmpty()) {\r
 112                         log.info("Please make sure collectStatistics method was called prior to calling getStats()!");\r
 113                 }\r
 114                 return new StatProcessor(stats);\r
 115         }\r
 116 \r
 117         void writeStatToDB() throws SQLException {\r
 118                 Set<JobStat> rjobs = new HashSet<JobStat>(stats);\r
 119                 StatDB statdb = new StatDB();\r
 120                 log.debug("Removing records that has already been recorded");\r
 121 \r
 122                 statdb.removeRecordedJobs(rjobs);\r
 123                 log.debug("New records left: " + rjobs.size());\r
 124                 statdb.insertData(rjobs);\r
 125         }\r
 126 \r
 127         /*\r
 128          * static void updateTime(File statFile) throws IOException { long lastMod =\r
 129          * statFile.lastModified(); FileWriter fw = new FileWriter(statFile);\r
 130          * fw.write(new Long(lastMod).toString()); fw.close(); }\r
 131          */\r
 132 \r
 133         /**\r
 134          * Not in use\r
 135          */\r
 136         public static void main(String[] args) throws IOException, SQLException {\r
 137 \r
 138                 // updateTime(new File(\r
 139                 // "D:\\workspace\\JABA2\\jobsout\\AACon#170462904473672\\STARTED"));\r
 140 \r
 141                 File[] files = FileUtil.getFiles("Y:\\fc\\www-jws2\\jaba\\jobsout",\r
 142                                 directories);\r
 143                 List<JobStat> stats = new ArrayList<JobStat>();\r
 144                 for (File file : files) {\r
 145                         JobDirectory jd = new JobDirectory(file);\r
 146                         stats.add(jd.getJobStat());\r
 147                         // System.out.println(jd.getJobStat().getJobReportTabulated());\r
 148                 }\r
 149                 StatProcessor sp = new StatProcessor(stats);\r
 150                 System.out.println(sp.reportStat());\r
 151                 System.out.println();\r
 152                 System.out.println("!!!!!!!!!!!!!!!!!!");\r
 153                 System.out.println();\r
 154 \r
 155                 Set<JobStat> rjobs = new HashSet<JobStat>(sp.stats);\r
 156                 StatDB statdb = new StatDB();\r
 157                 statdb.removeRecordedJobs(rjobs);\r
 158                 statdb.insertData(rjobs);\r
 159         }\r
 160 \r
 161         static FileFilter directories = new FileFilter() {\r
 162                 @Override\r
 163                 public boolean accept(File pathname) {\r
 164                         return pathname.isDirectory()\r
 165                                         && !pathname.getName().startsWith(".");\r
 166                 }\r
 167         };\r
 168 \r
 169         static class JobDirectory {\r
 170 \r
 171                 File jobdir;\r
 172                 Map<String, File> files = new HashMap<String, File>();\r
 173 \r
 174                 JobDirectory(File directory) {\r
 175                         this.jobdir = directory;\r
 176                         for (File f : jobdir.listFiles()) {\r
 177                                 files.put(f.getName(), f);\r
 178                         }\r
 179                 }\r
 180 \r
 181                 boolean hasStatus(JobStatus status) {\r
 182                         return files.containsKey(status.toString());\r
 183                 }\r
 184 \r
 185                 boolean isCollected() {\r
 186                         return hasStatus(JobStatus.COLLECTED);\r
 187                 }\r
 188 \r
 189                 boolean isCancelled() {\r
 190                         return hasStatus(JobStatus.CANCELLED);\r
 191                 }\r
 192 \r
 193                 long getStartTime() {\r
 194                         long starttime = UNDEFINED;\r
 195                         File startfile = files.get(JobStatus.STARTED.toString());\r
 196                         if (startfile == null) {\r
 197                                 startfile = files.get(JobStatus.SUBMITTED.toString());\r
 198                         }\r
 199                         try {\r
 200                                 if (startfile != null) {\r
 201                                         String start = FileUtil.readFileToString(startfile);\r
 202                                         starttime = Long.parseLong(start.trim());\r
 203                                 }\r
 204                         } catch (IOException ignore) {\r
 205                                 log.warn(\r
 206                                                 "IOException while reading STARTED status file! Ignoring...",\r
 207                                                 ignore);\r
 208                                 // fall back\r
 209                                 starttime = startfile.lastModified();\r
 210                         } catch (NumberFormatException ignore) {\r
 211                                 log.warn(\r
 212                                                 "NumberFormatException while reading STARTED status file! Ignoring...",\r
 213                                                 ignore);\r
 214                                 // fall back\r
 215                                 starttime = startfile.lastModified();\r
 216                         }\r
 217 \r
 218                         return starttime;\r
 219                 }\r
 220 \r
 221                 String getClusterJobID() {\r
 222                         String clustjobId = "";\r
 223                         File jobid = files.get("JOBID");\r
 224                         try {\r
 225                                 if (jobid != null) {\r
 226                                         clustjobId = FileUtil.readFileToString(jobid);\r
 227                                 }\r
 228                         } catch (IOException ioe) {\r
 229                                 log.error(\r
 230                                                 "IO Exception while reading the content of JOBID file for job "\r
 231                                                                 + jobid, ioe);\r
 232                         }\r
 233                         return clustjobId.trim();\r
 234                 }\r
 235 \r
 236                 long getFinishedTime() {\r
 237                         long ftime = UNDEFINED;\r
 238                         File finished = files.get(JobStatus.FINISHED.toString());\r
 239                         if (finished != null) {\r
 240                                 try {\r
 241                                         if (finished != null) {\r
 242                                                 String start = FileUtil.readFileToString(finished);\r
 243                                                 ftime = Long.parseLong(start.trim());\r
 244                                         }\r
 245                                 } catch (IOException ignore) {\r
 246                                         log.warn(\r
 247                                                         "IOException while reading FINISHED status file! Ignoring...",\r
 248                                                         ignore);\r
 249                                         // fall back\r
 250                                         ftime = finished.lastModified();\r
 251                                 } catch (NumberFormatException ignore) {\r
 252                                         log.warn(\r
 253                                                         "NumberFormatException while reading FINISHED status file! Ignoring...",\r
 254                                                         ignore);\r
 255                                         // fall back\r
 256                                         ftime = finished.lastModified();\r
 257                                 }\r
 258                         }\r
 259                         return ftime;\r
 260                 }\r
 261 \r
 262                 private Services getService() {\r
 263                         return Services.getServiceByJobDirectory(jobdir);\r
 264                 }\r
 265 \r
 266                 long getResultSize() {\r
 267                         Class<? extends Executable<?>> name = Services\r
 268                                         .getRunnerByJobDirectory(jobdir);\r
 269 \r
 270                         File f = null;\r
 271                         if (name.getSimpleName().equalsIgnoreCase("IUPred")) {\r
 272                                 f = files.get("out.glob");\r
 273                                 if (f == null)\r
 274                                         f = files.get("out.short");\r
 275                                 if (f == null)\r
 276                                         f = files.get("out.long");\r
 277                         } else {\r
 278                                 f = files.get(SkeletalExecutable.OUTPUT);\r
 279                         }\r
 280                         if (f != null) {\r
 281                                 return f.length();\r
 282                         }\r
 283                         return UNDEFINED;\r
 284                 }\r
 285 \r
 286                 long getInputSize() {\r
 287                         Class<? extends Executable<?>> name = Services\r
 288                                         .getRunnerByJobDirectory(jobdir);\r
 289 \r
 290                         File input = files.get(SkeletalExecutable.INPUT);\r
 291                         if (input != null) {\r
 292                                 return input.length();\r
 293                         }\r
 294                         return UNDEFINED;\r
 295                 }\r
 296 \r
 297                 JobStat getJobStat() {\r
 298                         return JobStat.newInstance(getService(), getClusterJobID(),\r
 299                                         jobdir.getName(), getStartTime(), getFinishedTime(),\r
 300                                         getInputSize(), getResultSize(), isCancelled(),\r
 301                                         isCollected());\r
 302                 }\r
 303 \r
 304                 @Override\r
 305                 public int hashCode() {\r
 306                         final int prime = 31;\r
 307                         int result = 1;\r
 308                         result = prime * result\r
 309                                         + ((jobdir == null) ? 0 : jobdir.hashCode());\r
 310                         return result;\r
 311                 }\r
 312 \r
 313                 @Override\r
 314                 public boolean equals(Object obj) {\r
 315                         if (this == obj)\r
 316                                 return true;\r
 317                         if (obj == null)\r
 318                                 return false;\r
 319                         if (getClass() != obj.getClass())\r
 320                                 return false;\r
 321                         JobDirectory other = (JobDirectory) obj;\r
 322                         if (jobdir == null) {\r
 323                                 if (other.jobdir != null)\r
 324                                         return false;\r
 325                         } else if (!jobdir.equals(other.jobdir))\r
 326                                 return false;\r
 327                         return true;\r
 328                 }\r
 329         }\r
 330 \r
 331         void collectStatistics() {\r
 332                 File[] files = workDirectory.listFiles(directories);\r
 333                 for (File file : files) {\r
 334                         JobDirectory jd = new JobDirectory(file);\r
 335                         JobStat jstat = jd.getJobStat();\r
 336                         // Do not record stats on the job that has not completed yet\r
 337                         if (hasCompleted(jd)) {\r
 338                                 stats.add(jstat);\r
 339                         } else {\r
 340                                 log.debug("Skipping the job: " + jstat);\r
 341                                 log.debug("As it has not completed yet");\r
 342                         }\r
 343                         // System.out.println(jd.getJobStat().getJobReportTabulated());\r
 344                 }\r
 345         }\r
 346 \r
 347         @Override\r
 348         public void run() {\r
 349                 log.info("Started updating statistics at " + new Date());\r
 350                 log.info("For directory: " + workDirectory.getAbsolutePath());\r
 351 \r
 352                 collectStatistics();\r
 353 \r
 354                 StatProcessor local_stats = getStats();\r
 355                 log.info("Found " + local_stats.getJobNumber() + " jobs!");\r
 356                 try {\r
 357                         writeStatToDB();\r
 358                 } catch (SQLException e) {\r
 359                         log.error("Fails to update jobs statistics database!");\r
 360                         log.error(e.getLocalizedMessage(), e);\r
 361                 }\r
 362                 log.info("Finished updating statistics at " + new Date());\r
 363         }\r
 364 }\r