package detective.core.distribute;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.Date;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import detective.core.distribute.collector.JobCollector;
import detective.core.services.DetectiveFactory;

public class SparkDriver {

  private final static Logger logger = LoggerFactory.getLogger(SparkDriver.class);
  
  private final static SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
  
  public static String getDefaultAppName(String packageOrClassName){
    return "Detective-" + packageOrClassName + "-" + format.format(new Date());
  }
  
  /**
   * ./bin/spark-submit --class detective.core.distribute.SparkDriver --master spark://127.0.0.1:7000 your.jar spark://127.0.0.1:7000
   * 
   * @param args classOrPackageName
   * @return
   */
  public static void main(String[] args) {
    if (args == null || args.length < 1){
      System.err.println("Usage: detective.core.distribute.SparkDriver packageOrClassName applicationName(Optional)\n"
          + "  if you'd like to setup master, please change your detective.conf file or -Dspark.master=yourMasterUrl\n"
          + "  -Dspark.driver.host is the host name if you'd like to use");
      System.exit(-1);
    }
    
    long errors = run(args);
    
    if (errors > 0)
      System.exit(-1);
  }

  /**
   * 
   * @param args
   * @return how many jobs failed
   */
  public static long run(String[] args) {
    String master = DetectiveFactory.INSTANCE.getConfig().getString("spark.master");
    logger.info("Detective is running with master " + master);
    String packageName = args[0];
    
    String appName = null;
    if (args.length >= 2){
      appName = args[1];
    }else
      appName = getDefaultAppName(packageName);
    
    SparkConf sparkConf = new SparkConf()    
      .setAppName(appName)
      .setMaster(master)
      //.setMaster("local[8]")
      //.setMaster("spark://127.0.0.1:7000")
      //.set("spark.driver.host", "localhost")
      //.set("spark.driver.port", "5555")
      ;
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);

    int duplicatedtasks = DetectiveFactory.INSTANCE.getConfig().getInt("spark.pressureTest.duplicateTasks");
    logger.info("Detective pressure test, jobs duplication: " + duplicatedtasks);
    List<JobToRun> jobs = JobCollector.collectAll(packageName, duplicatedtasks);
    
    JavaRDD<JobToRun> dataset = jsc.parallelize(jobs, jobs.size());
    
    JavaRDD<JobRunResult> datasetResult = dataset.map(new Function<JobToRun, JobRunResult>(){

      @Override
      public JobRunResult call(JobToRun job) throws Exception {
        logger.info(job.toString());
        
        JobRunner runner = new JobRunnerFilterImpl();
        runner.run(job);
        return job.getJobResult();
      }});
    
    List<JobRunResult> jobsAfterRun = datasetResult.collect();
    Collections.sort(jobsAfterRun);
    long errors = 0;
    for (JobRunResult job : jobsAfterRun){
      if (! job.getSuccessed())
        errors = errors + 1;
      
      logger.info(job.toString());      
    }
    
    logger.info(jobsAfterRun.size() + " tasks ran, " + errors + " failed, " + (jobsAfterRun.size() - errors) + " successed");
    return errors;
  }
  


}

@@ -168,16 +168,54 @@
 il.List;
+%0Aimport java.util.concurrent.TimeUnit;
 %0A%0Aimport
@@ -2741,32 +2741,77 @@
 tedtasks);%0A    %0A
+    Long startTime = System.nanoTime();%0A    %0A
     JavaRDD%3CJobT
@@ -2858,24 +2858,24 @@
 bs.size());%0A
-
     %0A    Jav
@@ -3277,24 +3277,72 @@
 .collect();%0A
+    %0A    Long endTime = System.nanoTime();%0A    %0A
     Collecti
@@ -3386,16 +3386,38 @@
 rs = 0;%0A
+    long skipped = 0;%0A
     for 
@@ -3512,16 +3512,63 @@
 rs + 1;%0A
+      if (job.isIgnored())%0A        skipped ++;%0A
       %0A 
@@ -3614,24 +3614,119 @@
 %0A    %7D%0A    %0A
+    Long timeElapsedSec = TimeUnit.SECONDS.convert(endTime - startTime, TimeUnit.NANOSECONDS);%0A
     logger.i
@@ -3729,16 +3729,30 @@
 er.info(
+%22Tests run: %22+
 jobsAfte
@@ -3766,95 +3766,98 @@
 ze()
- + %22 tasks ran, %22 + errors + %22 failed,
++%22, Errors: %22+errors+%22, Skipped: %22+skipped+%22, Time elapsed:
  %22 + 
-(jobsAfterRun.size() - errors) + %22 successed
+timeElapsedSec + %22Seconds 
 %22);%0A

add more test case