7_THS_displayEvents

.gitignore

@@ -1,6 +1,7 @@
 **/*.log
 **/.keep
 **/generated/
+**/version-info.properties
 *.avro
 *.iml
 *.ipr

tony-core/build.gradle

@@ -1,3 +1,6 @@
+plugins {
+  id "com.commercehub.gradle.plugin.avro" version "0.9.0"
+}
 apply plugin: 'java'
 apply plugin: 'com.google.protobuf'
 
@@ -6,6 +9,7 @@ dependencies {
 
   compile deps.azkaban.az_core
   compile deps.azkaban.az_hadoop_jobtype_plugin
+  compile deps.external.avro
   compile deps.external.jackson_databind
   compile deps.external.py4j
   compile deps.external.text

tony-core/src/main/avro/ApplicationFinished.avsc

@@ -0,0 +1,11 @@
+{
+  "namespace": "com.linkedin.tony.events",
+  "type": "record",
+  "name": "ApplicationFinished",
+  "fields": [
+    {"name": "applicationId", "type": "string"},
+    {"name": "finishedTasks", "type": "int"},
+    {"name": "failedTasks", "type": "int"},
+    {"name": "metrics", "type": {"type": "array", "items": "Metric"}}
+  ]
+}

tony-core/src/main/avro/ApplicationInited.avsc

@@ -0,0 +1,10 @@
+{
+  "namespace": "com.linkedin.tony.events",
+  "type": "record",
+  "name": "ApplicationInited",
+  "fields": [
+    {"name": "applicationId", "type": "string"},
+    {"name": "numTasks", "type": "int"},
+    {"name": "host", "type": "string"}
+  ]
+}

tony-core/src/main/avro/Event.avsc

@@ -0,0 +1,10 @@
+{
+  "namespace": "com.linkedin.tony.events",
+  "type": "record",
+  "name": "Event",
+  "fields": [
+    {"name": "type", "type": "EventType"},
+    {"name": "event", "type": [ "ApplicationInited", "ApplicationFinished" ]},
+    {"name": "timestamp", "type": "long"}
+  ]
+}

tony-core/src/main/avro/EventType.avsc

@@ -0,0 +1,5 @@
+{
+  "namespace": "com.linkedin.tony.events",
+  "type": "enum", "name": "EventType",
+  "symbols": [ "APPLICATION_INITED", "APPLICATION_FINISHED" ]
+}

tony-core/src/main/avro/Metric.avsc

@@ -0,0 +1,9 @@
+{
+  "namespace": "com.linkedin.tony.events",
+  "type": "record",
+  "name": "Metric",
+  "fields": [
+    {"name": "name", "type": "string"},
+    {"name": "value", "type": "double"}
+  ]
+}

tony-core/src/main/java/com/linkedin/tony/Constants.java

@@ -18,6 +18,8 @@ public class Constants {
   public static final String JOBS_SUFFIX = "jobs";
   public static final String CONFIG_SUFFIX = "config";
 
+  public static final String TMP_AVRO = "tmp.avro";
+
   // TensorFlow constants
   public static final String TB_PORT = "TB_PORT";
   public static final String TASK_INDEX = "TASK_INDEX";

tony-core/src/main/java/com/linkedin/tony/TonyApplicationMaster.java

@@ -7,13 +7,18 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.linkedin.tony.events.ApplicationFinished;
+import com.linkedin.tony.events.ApplicationInited;
+import com.linkedin.tony.events.Event;
+import com.linkedin.tony.events.EventHandler;
+import com.linkedin.tony.events.EventType;
+import com.linkedin.tony.events.Metric;
 import com.linkedin.tony.rpc.ApplicationRpc;
 import com.linkedin.tony.rpc.ApplicationRpcServer;
 import com.linkedin.tony.rpc.TaskUrl;
 import com.linkedin.tony.tensorflow.TensorFlowContainerRequest;
 import com.linkedin.tony.tensorflow.TonySession;
 import com.linkedin.tony.tensorflow.TonySession.TonyTask;
-import com.linkedin.tony.util.HistoryFileUtils;
 import com.linkedin.tony.util.Utils;
 import java.io.BufferedReader;
 import java.io.File;
@@ -31,7 +36,9 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.apache.commons.cli.CommandLine;
@@ -91,8 +98,9 @@ public class TonyApplicationMaster {
   private String amHostPort;
   private FileSystem fs;
   private String tonyHistoryFolder;
-  private Path jobDir;
+  private Path jobDir = null;
   private String user = null;
+  private BlockingQueue<Event> eventQueue = new LinkedBlockingQueue<>();
 
   // Container info
   private int amRetryCount;
@@ -341,6 +349,9 @@ private boolean run(String[] args) {
     }
 
     mainThread = Thread.currentThread();
+    EventHandler eventHandler = new EventHandler(fs, eventQueue);
+    Thread eventHandlerThread = new Thread(eventHandler);
+    eventHandlerThread.start();
     boolean succeeded;
     do {
       // Crash AM on purpose during AM crash tests.
@@ -351,6 +362,7 @@ private boolean run(String[] args) {
       }
 
       try {
+        eventHandler.emitEvent(constructEvent("APPLICATION_INITED"));
         start();
       } catch (Exception e) {
         LOG.error("Exception when we're starting TonyAM", e);
@@ -372,12 +384,36 @@ private boolean run(String[] args) {
     stop();
     long completed = System.currentTimeMillis();
     printTaskUrls();
-    // By this time jobDir should have been set
-    HistoryFileUtils.createHistoryFile(fs,
-        TonyJobMetadata.newInstance(yarnConf, appIdString, started, completed, succeeded), jobDir);
+    eventHandler.emitEvent(constructEvent("APPLICATION_FINISHED"));
+    eventHandler.stop(jobDir, TonyJobMetadata.newInstance(yarnConf, appIdString, started, completed, succeeded, user));
+    // Wait for eventHandlerThread to wrap up before exiting
+    try {
+      eventHandlerThread.join();
+    } catch (InterruptedException e) {
+      LOG.error("Failed to dump leftover events", e);
+    }
     return succeeded;
   }
 
+  private Event constructEvent(String type) {
+    Event event = new Event();
+    event.setTimestamp(System.currentTimeMillis());
+    switch (type) {
+      case "APPLICATION_INITED":
+        event.setType(EventType.APPLICATION_INITED);
+        event.setEvent(new ApplicationInited(appIdString, (int) numTotalWorkerTasks, Utils.getCurrentHostName()));
+        return event;
+      case "APPLICATION_FINISHED":
+        event.setType(EventType.APPLICATION_FINISHED);
+        List<Metric> metrics = new ArrayList<>();
+        int numFailedTasks = (int) numTotalWorkerTasks - numCompletedWorkerTasks.get();
+        event.setEvent(new ApplicationFinished(appIdString, (int) numTotalWorkerTasks, numFailedTasks, metrics));
+        return event;
+      default:
+        return null;
+    }
+  }
+
   /**
    * Prepare the application master. This part is shared across different retries.
    */
@@ -451,8 +487,8 @@ private void setupJobDir(FileSystem fs, String histFolder, String appId) {
     }
 
     jobDir = new Path(interm, appId);
-    // set to `tony` group so THS can move files to finished,
-    // and other users can't delete this job directory
+    // set to `tony` group by default
+    // due to inherited permission from parent folder
     Utils.createDir(fs, jobDir, Constants.PERM770);
   }
 

tony-core/src/main/java/com/linkedin/tony/TonyClient.java

@@ -342,9 +342,6 @@ public boolean init(String[] args) throws ParseException {
    * @param cliParser CommandLine object that has all the command line arguments.
    */
   public static void initTonyConf(Configuration tonyConf, CommandLine cliParser) {
-    if (System.getenv(Constants.TONY_CONF_DIR) != null) {
-      tonyConf.addResource(new Path(System.getenv(Constants.TONY_CONF_DIR) + File.separatorChar + Constants.TONY_SITE_CONF));
-    }
     tonyConf.addResource(Constants.TONY_DEFAULT_XML);
     if (cliParser.hasOption("conf_file")) {
       tonyConf.addResource(new Path(cliParser.getOptionValue("conf_file")));
@@ -357,6 +354,9 @@ public static void initTonyConf(Configuration tonyConf, CommandLine cliParser) {
         tonyConf.set(cliConf.getKey(), cliConf.getValue());
       }
     }
+    if (System.getenv(Constants.TONY_CONF_DIR) != null) {
+      tonyConf.addResource(new Path(System.getenv(Constants.TONY_CONF_DIR) + File.separatorChar + Constants.TONY_SITE_CONF));
+    }
   }
 
   public Configuration getTonyConf() {

tony-core/src/main/java/com/linkedin/tony/TonyConfigurationKeys.java

@@ -35,20 +35,23 @@ private TonyConfigurationKeys() {
   public static final String TONY_HISTORY_LOCATION = TONY_PREFIX + "history.location";
   public static final String DEFAULT_TONY_HISTORY_LOCATION = "/path/to/tony-history";
 
-  // All these variables are here to pass TestTonyConfigurationFields.
-  // Do not remove unless it is also removed in tony-default.xml.
   public static final String TONY_HISTORY_INTERMEDIATE = TONY_PREFIX + "history.intermediate";
   public static final String DEFAULT_TONY_HISTORY_INTERMEDIATE = DEFAULT_TONY_HISTORY_LOCATION + "/intermediate";
 
   public static final String TONY_HISTORY_FINISHED = TONY_PREFIX + "history.finished";
   public static final String DEFAULT_TONY_HISTORY_FINISHED = DEFAULT_TONY_HISTORY_LOCATION + "/finished";
 
+  public static final String TONY_HISTORY_CACHE_MAX_ENTRIES = TONY_PREFIX + "history.cache.max-entries";
+  public static final String DEFAULT_TONY_HISTORY_CACHE_MAX_ENTRIES = "1000";
+
   public static final String TONY_KEYTAB_USER = TONY_PREFIX + "keytab.user";
   public static final String DEFAULT_TONY_KEYTAB_USER = "user";
 
   public static final String TONY_KEYTAB_LOCATION = TONY_PREFIX + "keytab.location";
   public static final String DEFAULT_TONY_KEYTAB_LOCATION = "/path/to/tony.keytab";
 
+  // All these variables are here to pass TestTonyConfigurationFields.
+  // Do not remove unless it is also removed in tony-default.xml.
   public static final String TONY_HTTPS_PORT = TONY_PREFIX + "https.port";
   public static final String DEFAULT_TONY_HTTPS_PORT = "19886";
 
@@ -73,6 +76,9 @@ private TonyConfigurationKeys() {
   public static final String TONY_INIT_MODULE = TONY_PREFIX + "init.module";
   public static final String DEFAULT_TONY_INIT_MODULE = "Startup";
 
+  public static final String TONY_HISTORY_MAX_APPEND = TONY_PREFIX + "history.maxAppends";
+  public static final int DEFAULT_TONY_HISTORY_MAX_APPEND = 3;
+
   // Application configurations
   public static final String YARN_QUEUE_NAME = TONY_PREFIX + "yarn.queue";
   public static final String DEFAULT_YARN_QUEUE_NAME = "default";

tony-core/src/main/java/com/linkedin/tony/TonyJobMetadata.java

@@ -5,11 +5,9 @@
 package com.linkedin.tony;
 
 import com.linkedin.tony.util.Utils;
-import java.io.IOException;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.security.UserGroupInformation;
 
 
 public class TonyJobMetadata {
@@ -21,7 +19,17 @@ public class TonyJobMetadata {
   private String status;
   private String user;
 
-  public TonyJobMetadata(String id, String url, long started, long completed, String status, String user) {
+  // for testing only
+  TonyJobMetadata() {
+    this.id = "";
+    this.url = "";
+    this.started = 0;
+    this.completed = 0;
+    this.status = "";
+    this.user = "";
+  }
+
+  private TonyJobMetadata(String id, String url, long started, long completed, String status, String user) {
     this.id = id;
     this.url = url;
     this.started = started;
@@ -31,14 +39,8 @@ public TonyJobMetadata(String id, String url, long started, long completed, Stri
   }
 
   public static TonyJobMetadata newInstance(Configuration yarnConf, String appId, long started, long completed,
-      boolean status) {
+      boolean status, String user) {
     String jobStatus = status ? "SUCCEEDED" : "FAILED";
-    String user = null;
-    try {
-      user = UserGroupInformation.getCurrentUser().getShortUserName();
-    } catch (IOException e) {
-      LOG.error("Failed reading from disk. Set user to null", e);
-    }
     return new TonyJobMetadata(appId, Utils.buildRMUrl(yarnConf, appId), started, completed, jobStatus, user);
   }