Signed-off-by: lishid <[email protected]>

.gitignore

@@ -31,3 +31,6 @@ target
 # Play
 logs
 RUNNING_PID
+
+
+*.jar

app/Global.java

@@ -0,0 +1,50 @@
+import com.linkedin.drelephant.DrElephant;
+import com.sun.security.sasl.util.AbstractSaslImpl;
+import play.Application;
+import play.GlobalSettings;
+import play.Logger;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+import java.util.logging.Level;
+
+public class Global extends GlobalSettings {
+
+ DrElephant drElephant;
+
+ public void onStart(Application app) {
+ Logger.info("Application has started");
+ java.util.logging.Logger logger = java.util.logging.Logger.getLogger("javax.security.sasl");
+ logger.setLevel(Level.OFF);
+ Logger.info("LOGGER LEVEL: " + logger.getLevel());
+ try {
+ setFinalStatic(AbstractSaslImpl.class.getDeclaredField("logger"), logger);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ try {
+ drElephant = new DrElephant();
+ drElephant.start();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+
+ public void onStop(Application app) {
+ Logger.info("Application shutdown...");
+ if (drElephant != null) {
+ drElephant.kill();
+ }
+ }
+
+ static void setFinalStatic(Field field, Object newValue) throws Exception {
+ field.setAccessible(true);
+
+ Field modifiersField = Field.class.getDeclaredField("modifiers");
+ modifiersField.setAccessible(true);
+ modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL);
+
+ field.set(null, newValue);
+ }
+}

src/com/linkedin/drelephant/DrElephant.java → app/com/linkedin/drelephant/DrElephant.java

@@ -7,18 +7,14 @@
 import java.io.File;
 import java.io.IOException;
 
-public class DrElephant {
+public class DrElephant extends Thread {
+ private ElephantRunner elephant;
+
  public static void main(String[] args) throws IOException {
  if (args.length > 0) {
  analyse(args[0]);
  } else {
- File storage = new File("results.txt");
- if (!storage.exists()) {
- storage.createNewFile();
- }
-
- ElephantRunner elephant = new ElephantRunner(storage);
- elephant.run();
+ new DrElephant().start();
  }
  }
 
@@ -30,4 +26,27 @@ public static void analyse(String jobId) throws IOException {
  HeuristicResult result = analyser.analyse(jobData);
  System.out.println(result.getMessage());
  }
+
+ public DrElephant() throws IOException {
+ File storage = new File("results.txt");
+ if (!storage.exists()) {
+ try {
+ storage.createNewFile();
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ }
+ elephant = new ElephantRunner(storage);
+ }
+
+ @Override
+ public void run() {
+ elephant.run();
+ }
+
+ public void kill() {
+ if (elephant != null) {
+ elephant.kill();
+ }
+ }
 }

src/com/linkedin/drelephant/ElephantAnalyser.java → app/com/linkedin/drelephant/ElephantAnalyser.java

@@ -25,7 +25,7 @@ public ElephantAnalyser() {
 
  public HeuristicResult analyse(HadoopJobData data) {
  if (data.getMapperData().length == 0 && data.getReducerData().length == 0) {
- return new HeuristicResult("No mapper/reducer data received", false);
+ return new HeuristicResult("No mapper/reducer data received", true);
  }
 
  HeuristicResult result = null;

src/com/linkedin/drelephant/ElephantFetcher.java → app/com/linkedin/drelephant/ElephantFetcher.java

@@ -20,7 +20,9 @@ public ElephantFetcher() throws IOException {
 
  private void init() throws IOException {
  logger.info("Connecting to the jobtracker");
- jobClient = new JobClient(new JobConf(new Configuration()));
+ Configuration config = new Configuration();
+ logger.info(config);
+ jobClient = new JobClient(new JobConf(config));
  }
 
  public HadoopJobData getJobData(JobID job_id) throws IOException {
@@ -29,6 +31,12 @@ public HadoopJobData getJobData(JobID job_id) throws IOException {
  return null;
  }
 
+ JobStatus status = job.getJobStatus();
+
+ String username = status.getUsername();
+ long startTime = status.getStartTime();
+ String jobUrl = job.getTrackingURL();
+
  TaskReport[] mapperTasks = getMapTaskReports(job_id);
  TaskReport[] reducerTasks = getReduceTaskReports(job_id);
 
@@ -42,7 +50,7 @@ public HadoopJobData getJobData(JobID job_id) throws IOException {
  reducers[i] = new HadoopTaskData(job, reducerTasks[i], true);
  }
 
- return new HadoopJobData(counterHolder, mappers, reducers);
+ return new HadoopJobData(counterHolder, mappers, reducers).setUsername(username).setStartTime(startTime).setUrl(jobUrl);
  }
 
  private RunningJob getJob(JobID job_id) throws IOException {

app/com/linkedin/drelephant/ElephantRunner.java

@@ -0,0 +1,154 @@
+package com.linkedin.drelephant;
+
+import com.linkedin.drelephant.analysis.Constants;
+import com.linkedin.drelephant.analysis.HeuristicResult;
+import com.linkedin.drelephant.hadoop.HadoopJobData;
+import model.AnalysisResult;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.log4j.Logger;
+
+import java.io.File;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class ElephantRunner implements Runnable {
+ private static final long WAIT_INTERVAL = 5 * 60 * 1000;
+ private static final Logger logger = Logger.getLogger(ElephantRunner.class);
+ private AtomicBoolean running = new AtomicBoolean(true);
+ private File storage;
+ private boolean firstRun = true;
+
+ public ElephantRunner(File storage) {
+ this.storage = storage;
+ }
+
+ @Override
+ public void run() {
+ Constants.load();
+ try {
+ ElephantFetcher fetcher = new ElephantFetcher();
+ ElephantAnalyser analyser = new ElephantAnalyser();
+ Set<JobID> previousJobs = new HashSet<JobID>();
+ long lastRun;
+
+ while (running.get()) {
+ lastRun = System.currentTimeMillis();
+
+ try {
+ logger.info("Fetching job list.");
+ JobStatus[] jobs = fetcher.getJobList();
+ if (jobs == null) {
+ throw new IllegalArgumentException("Jobtracker returned 'null' for job list");
+ }
+
+ Set<JobID> successJobs = filterSuccessfulJobs(jobs);
+
+ successJobs = filterPreviousJobs(successJobs, previousJobs);
+
+ logger.info(successJobs.size() + " jobs to analyse.");
+
+ //Analyse all ready jobs
+ for (JobID jobId : successJobs) {
+ try {
+ analyzeJob(fetcher, analyser, jobId);
+ previousJobs.add(jobId);
+ } catch (Exception e) {
+ logger.error("Error analysing job", e);
+ }
+ }
+ logger.info("Finished all jobs. Waiting for refresh.");
+
+ } catch (Exception e) {
+ logger.error("Error getting job list", e);
+ }
+
+ //Wait for long enough
+ long nextRun = lastRun + WAIT_INTERVAL;
+ long waitTime = nextRun - System.currentTimeMillis();
+ while (running.get() && waitTime > 0) {
+ try {
+ Thread.sleep(waitTime);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ waitTime = nextRun - System.currentTimeMillis();
+ }
+ }
+ } catch (Exception e) {
+ logger.error("Error in ElephantRunner", e);
+ }
+ }
+
+ private Set<JobID> filterSuccessfulJobs(JobStatus[] jobs) {
+ Set<JobID> successJobs = new HashSet<JobID>();
+ for (JobStatus job : jobs) {
+ if (job.getRunState() == JobStatus.SUCCEEDED && job.isJobComplete()) {
+ successJobs.add(job.getJobID());
+ }
+ }
+ return successJobs;
+ }
+
+ private Set<JobID> filterPreviousJobs(Set<JobID> jobs, Set<JobID> previousJobs) {
+ logger.info("Cleaning up previous runs.");
+ //On first run, check against DB
+ if (firstRun) {
+ Set<JobID> newJobs = new HashSet<JobID>();
+ for (JobID jobId : jobs) {
+ AnalysisResult prevResult = AnalysisResult.find.byId(jobId.toString());
+ if (prevResult == null) {
+ //Job not found, add to new jobs list
+ newJobs.add(jobId);
+ } else {
+ //Job found, add to old jobs list
+ previousJobs.add(jobId);
+ }
+ }
+ jobs = newJobs;
+ firstRun = false;
+ } else {
+ //Remove untracked jobs
+ previousJobs.retainAll(jobs);
+ //Remove previously analysed jobs
+ jobs.removeAll(previousJobs);
+ }
+
+ return jobs;
+ }
+
+ private void analyzeJob(ElephantFetcher fetcher, ElephantAnalyser analyser, JobID jobId) throws Exception {
+ logger.info("Looking at job " + jobId);
+ HadoopJobData jobData = fetcher.getJobData(jobId);
+
+ //Job wiped from jobtracker already.
+ if (jobData == null) {
+ return;
+ }
+
+ HeuristicResult analysisResult = analyser.analyse(jobData);
+
+ //Save to DB
+ AnalysisResult result = new AnalysisResult();
+ result.job_id = jobId.toString();
+ result.success = analysisResult.succeeded();
+ result.url = jobData.getUrl();
+ result.username = jobData.getUsername();
+ result.message = analysisResult.getMessage();
+ result.data = analysisResult.getDetailsCSV();
+ result.dataColumns = analysisResult.getDetailsColumns();
+ result.startTime = jobData.getStartTime();
+ result.analysisTime = System.currentTimeMillis();
+
+ if(result.dataColumns < 1) {
+ result.dataColumns = 1;
+ }
+
+ result.save();
+ }
+
+ public void kill() {
+ running.set(false);
+ }
+}

src/com/linkedin/drelephant/analysis/Constants.java → app/com/linkedin/drelephant/analysis/Constants.java

@@ -9,22 +9,17 @@
 
 public class Constants {
  private static final Logger logger = Logger.getLogger(Constants.class);
- public static final long HDFS_BLOCK_SIZE;
+ public static long HDFS_BLOCK_SIZE = 64 * 1024 * 1024;
  public static final long DISK_READ_SPEED = 100 * 1024 * 1024;
+ public static final int SHUFFLE_SORT_MAX_SAMPLE_SIZE = 50;
 
- static {
- long block_size = 64 * 1024 * 1024;
-
+ public static void load() {
  try {
- block_size = FileSystem.get(new Configuration()).getDefaultBlockSize(new Path("/"));
+ HDFS_BLOCK_SIZE = FileSystem.get(new Configuration()).getDefaultBlockSize(new Path("/"));
  } catch (IOException e) {
  logger.error("Error getting FS Block Size!", e);
  }
 
- HDFS_BLOCK_SIZE = block_size;
- }
-
- public static void load() {
  logger.info("HDFS BLock size: " + HDFS_BLOCK_SIZE);
  }
 }

app/com/linkedin/drelephant/analysis/HeuristicResult.java

@@ -0,0 +1,63 @@
+package com.linkedin.drelephant.analysis;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class HeuristicResult {
+ public static final HeuristicResult SUCCESS = new HeuristicResult("Everything looks good", true);
+
+ private String message;
+ private List<String> details;
+ private int detailsColumns = 0;
+ private boolean success;
+
+ public HeuristicResult(String message, boolean success) {
+ this.message = message;
+ this.details = new ArrayList<String>();
+ this.success = success;
+ }
+
+ public boolean succeeded() {
+ return success;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public List<String> getDetails() {
+ return details;
+ }
+
+ public String getDetailsCSV() {
+ StringBuilder sb = new StringBuilder();
+ for (String line : details) {
+ sb.append(line).append("\n");
+ }
+ return sb.toString().trim();
+ }
+
+ public int getDetailsColumns() {
+ return detailsColumns;
+ }
+
+ public void addDetail(String... parts) {
+ details.add(createLine(parts));
+ if (parts.length > detailsColumns) {
+ detailsColumns = parts.length;
+ }
+ }
+
+ public static String createLine(String... parts) {
+ StringBuilder sb = new StringBuilder();
+ String quotes = "\"";
+ String comma = ",";
+ for (int i = 0; i < parts.length; i++) {
+ sb.append(quotes).append(parts[i].replaceAll("\"", "\\\"")).append(quotes);
+ if (i != parts.length - 1) {
+ sb.append(comma);
+ }
+ }
+ return sb.toString();
+ }
+}