YARN-3087. Made the REST server of per-node aggregator work alone in NM daemon. Conntributed by Li Lu.

(cherry picked from commit 41a08ad404d4278fe598d6c222b2ae0e84bae0df)

hadoop-yarn-project/CHANGES.txt

@@ -17,6 +17,9 @@ Branch YARN-2928: Timeline Server Next Generation: Phase 1
     YARN-3240. Implement client API to put generic entities. (Zhijie Shen via
     junping_du)
 
+    YARN-3087. Made the REST server of per-node aggregator work alone in NM
+    daemon. (Li Lu via zjshen)
+
   IMPROVEMENTS
 
   OPTIMIZATIONS

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/HierarchicalTimelineEntity.java

@@ -33,7 +33,7 @@ import java.util.Set;
 @InterfaceStability.Unstable
 public abstract class HierarchicalTimelineEntity extends TimelineEntity {
   private Identifier parent;
-  private Map<String, Set<String>> children = new HashMap<>();
+  private HashMap<String, Set<String>> children = new HashMap<>();
 
   HierarchicalTimelineEntity(String type) {
     super(type);
@@ -56,14 +56,24 @@ public abstract class HierarchicalTimelineEntity extends TimelineEntity {
     parent.setId(id);
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
   @XmlElement(name = "children")
+  public HashMap<String, Set<String>> getChildrenJAXB() {
+    return children;
+  }
+
   public Map<String, Set<String>> getChildren() {
     return children;
   }
 
   public void setChildren(Map<String, Set<String>> children) {
     validateChildren(children);
-    this.children = children;
+    if (children != null && !(children instanceof HashMap)) {
+      this.children = new HashMap<String, Set<String>>(children);
+    } else {
+      this.children = (HashMap) children;
+    }
   }
 
   public void addChildren(Map<String, Set<String>> children) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntity.java

@@ -65,12 +65,12 @@ public class TimelineEntity {
   }
 
   private Identifier identifier;
-  private Map<String, Object> info = new HashMap<>();
-  private Map<String, Object> configs = new HashMap<>();
+  private HashMap<String, Object> info = new HashMap<>();
+  private HashMap<String, Object> configs = new HashMap<>();
   private Set<TimelineMetric> metrics = new HashSet<>();
   private Set<TimelineEvent> events = new HashSet<>();
-  private Map<String, Set<String>> isRelatedToEntities = new HashMap<>();
-  private Map<String, Set<String>> relatesToEntities = new HashMap<>();
+  private HashMap<String, Set<String>> isRelatedToEntities = new HashMap<>();
+  private HashMap<String, Set<String>> relatesToEntities = new HashMap<>();
   private long createdTime;
   private long modifiedTime;
 
@@ -109,13 +109,23 @@ public class TimelineEntity {
     this.identifier = identifier;
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
   @XmlElement(name = "info")
+  public HashMap<String, Object> getInfoJAXB() {
+    return info;
+  }
+
   public Map<String, Object> getInfo() {
     return info;
   }
 
   public void setInfo(Map<String, Object> info) {
-    this.info = info;
+    if (info != null && !(info instanceof HashMap)) {
+      this.info = new HashMap<String, Object>(info);
+    } else {
+      this.info = (HashMap<String, Object>) info;
+    }
   }
 
   public void addInfo(Map<String, Object> info) {
@@ -126,13 +136,23 @@ public class TimelineEntity {
     info.put(key, value);
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
   @XmlElement(name = "configs")
+  public HashMap<String, Object> getConfigsJAXB() {
+    return configs;
+  }
+
   public Map<String, Object> getConfigs() {
     return configs;
   }
 
   public void setConfigs(Map<String, Object> configs) {
-    this.configs = configs;
+    if (configs != null && !(configs instanceof HashMap)) {
+      this.configs = new HashMap<String, Object>(configs);
+    } else {
+      this.configs = (HashMap<String, Object>) configs;
+    }
   }
 
   public void addConfigs(Map<String, Object> configs) {
@@ -177,14 +197,24 @@ public class TimelineEntity {
     events.add(event);
   }
 
-  @XmlElement(name = "isrelatedto")
   public Map<String, Set<String>> getIsRelatedToEntities() {
     return isRelatedToEntities;
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
+  @XmlElement(name = "isrelatedto")
+  public HashMap<String, Set<String>> getIsRelatedToEntitiesJAXB() {
+    return isRelatedToEntities;
+  }
+
   public void setIsRelatedToEntities(
       Map<String, Set<String>> isRelatedToEntities) {
-    this.isRelatedToEntities = isRelatedToEntities;
+    if (isRelatedToEntities != null && !(isRelatedToEntities instanceof HashMap)) {
+      this.isRelatedToEntities = new HashMap<String, Set<String>>(isRelatedToEntities);
+    } else {
+      this.isRelatedToEntities = (HashMap<String, Set<String>>) isRelatedToEntities;
+    }
   }
 
   public void addIsRelatedToEntities(
@@ -209,7 +239,13 @@ public class TimelineEntity {
     ids.add(id);
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
   @XmlElement(name = "relatesto")
+  public HashMap<String, Set<String>> getRelatesToEntitiesJAXB() {
+    return relatesToEntities;
+  }
+
   public Map<String, Set<String>> getRelatesToEntities() {
     return relatesToEntities;
   }
@@ -235,7 +271,11 @@ public class TimelineEntity {
   }
 
   public void setRelatesToEntities(Map<String, Set<String>> relatesToEntities) {
-    this.relatesToEntities = relatesToEntities;
+    if (relatesToEntities != null && !(relatesToEntities instanceof HashMap)) {
+      this.relatesToEntities = new HashMap<String, Set<String>>(relatesToEntities);
+    } else {
+      this.relatesToEntities = (HashMap<String, Set<String>>) relatesToEntities;
+    }
   }
 
   @XmlElement(name = "createdtime")

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEvent.java

@@ -33,7 +33,7 @@ import java.util.Map;
 @InterfaceStability.Unstable
 public class TimelineEvent {
   private String id;
-  private Map<String, Object> info = new HashMap<>();
+  private HashMap<String, Object> info = new HashMap<>();
   private long timestamp;
 
   public TimelineEvent() {
@@ -49,13 +49,23 @@ public class TimelineEvent {
     this.id = id;
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
   @XmlElement(name = "info")
+  public HashMap<String, Object> getInfoJAXB() {
+    return info;
+  }
+
   public Map<String, Object> getInfo() {
     return info;
   }
 
   public void setInfo(Map<String, Object> info) {
-    this.info = info;
+    if (info != null && !(info instanceof HashMap)) {
+      this.info = new HashMap<String, Object>(info);
+    } else {
+      this.info = (HashMap<String, Object>) info;
+    }
   }
 
   public void addInfo(Map<String, Object> info) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetric.java

@@ -34,9 +34,9 @@ import java.util.Map;
 @InterfaceStability.Unstable
 public class TimelineMetric {
   private String id;
-  private Map<String, Object> info = new HashMap<>();
+  private HashMap<String, Object> info = new HashMap<>();
   private Object singleData;
-  private Map<Long, Object> timeSeries = new LinkedHashMap<>();
+  private HashMap<Long, Object> timeSeries = new LinkedHashMap<>();
   private long startTime;
   private long endTime;
 
@@ -53,13 +53,23 @@ public class TimelineMetric {
     this.id = id;
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
   @XmlElement(name = "info")
+  public HashMap<String, Object> getInfoJAXB() {
+    return info;
+  }
+
   public Map<String, Object> getInfo() {
     return info;
   }
 
   public void setInfo(Map<String, Object> info) {
-    this.info = info;
+    if (info != null && !(info instanceof HashMap)) {
+      this.info = new HashMap<String, Object>(info);
+    } else {
+      this.info = (HashMap<String, Object>) info;
+    }
   }
 
   public void addInfo(Map<String, Object> info) {
@@ -79,13 +89,23 @@ public class TimelineMetric {
     this.singleData = singleData;
   }
 
+  // required by JAXB
+  @InterfaceAudience.Private
   @XmlElement(name = "timeseries")
+  public HashMap<Long, Object> getTimeSeriesJAXB() {
+    return timeSeries;
+  }
+
   public Map<Long, Object> getTimeSeries() {
     return timeSeries;
   }
 
   public void setTimeSeries(Map<Long, Object> timeSeries) {
-    this.timeSeries = timeSeries;
+    if (timeSeries != null && !(timeSeries instanceof LinkedHashMap)) {
+      this.timeSeries = new LinkedHashMap<Long, Object>(timeSeries);
+    } else {
+      this.timeSeries = (LinkedHashMap<Long, Object>) timeSeries;
+    }
   }
 
   public void addTimeSeries(Map<Long, Object> timeSeries) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/webapp/WebServer.java

@@ -124,12 +124,6 @@ public class WebServer extends AbstractService {
       bind(NMWebServices.class);
       bind(GenericExceptionHandler.class);
       bind(JAXBContextResolver.class);
-      // host the timeline service aggregator web service temporarily
-      // (see YARN-3087)
-      bind(PerNodeAggregatorWebService.class);
-      // bind to the global singleton instance
-      bind(AppLevelServiceManager.class).
-          toProvider(AppLevelServiceManagerProvider.class);
       bind(ResourceView.class).toInstance(this.resourceView);
       bind(ApplicationACLsManager.class).toInstance(this.aclsManager);
       bind(LocalDirsHandlerService.class).toInstance(dirsHandler);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/PerNodeAggregatorServer.java

@@ -18,14 +18,16 @@
 
 package org.apache.hadoop.yarn.server.timelineservice.aggregator;
 
+import java.net.URI;
 import java.nio.ByteBuffer;
+import java.util.HashMap;
 
-import com.google.inject.Inject;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.http.lib.StaticUserWebFilter;
 import org.apache.hadoop.util.ExitUtil;
 import org.apache.hadoop.util.ShutdownHookManager;
 import org.apache.hadoop.util.StringUtils;
@@ -40,11 +42,16 @@ import org.apache.hadoop.yarn.server.api.AuxiliaryService;
 import org.apache.hadoop.yarn.server.api.ContainerContext;
 import org.apache.hadoop.yarn.server.api.ContainerInitializationContext;
 import org.apache.hadoop.yarn.server.api.ContainerTerminationContext;
-import org.apache.hadoop.yarn.webapp.*;
+import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
+import org.apache.hadoop.yarn.webapp.YarnJacksonJaxbJsonProvider;
 import org.apache.hadoop.yarn.webapp.util.WebAppUtils;
+import org.apache.hadoop.http.HttpServer2;
 
 import com.google.common.annotations.VisibleForTesting;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeys.DEFAULT_HADOOP_HTTP_STATIC_USER;
+import static org.apache.hadoop.fs.CommonConfigurationKeys.HADOOP_HTTP_STATIC_USER;
+
 /**
  * The top-level server for the per-node timeline aggregator service. Currently
  * it is defined as an auxiliary service to accommodate running within another
@@ -56,9 +63,10 @@ public class PerNodeAggregatorServer extends AuxiliaryService {
   private static final Log LOG =
       LogFactory.getLog(PerNodeAggregatorServer.class);
   private static final int SHUTDOWN_HOOK_PRIORITY = 30;
+  static final String AGGREGATOR_COLLECTION_ATTR_KEY = "aggregator.collection";
 
   private final AppLevelServiceManager serviceManager;
-  private WebApp webApp;
+  private HttpServer2 timelineRestServer;
 
   public PerNodeAggregatorServer() {
     // use the same singleton
@@ -86,8 +94,8 @@ public class PerNodeAggregatorServer extends AuxiliaryService {
 
   @Override
   protected void serviceStop() throws Exception {
-    if (webApp != null) {
-      webApp.stop();
+    if (timelineRestServer != null) {
+      timelineRestServer.stop();
     }
     // stop the service manager
     serviceManager.stop();
@@ -103,11 +111,31 @@ public class PerNodeAggregatorServer extends AuxiliaryService {
                           WebAppUtils.getAHSWebAppURLWithoutScheme(conf));
     LOG.info("Instantiating the per-node aggregator webapp at " + bindAddress);
     try {
-      webApp =
-          WebApps
-            .$for("timeline", null, null, "ws")
-            .with(conf).at(bindAddress).start(
-                new TimelineServiceWebApp());
+      Configuration confForInfoServer = new Configuration(conf);
+      confForInfoServer.setInt(HttpServer2.HTTP_MAX_THREADS, 10);
+      HttpServer2.Builder builder = new HttpServer2.Builder()
+          .setName("timeline")
+          .setConf(conf)
+          .addEndpoint(URI.create("http://" + bindAddress));
+      timelineRestServer = builder.build();
+      // TODO: replace this by an authentification filter in future.
+      HashMap<String, String> options = new HashMap<String, String>();
+      String username = conf.get(HADOOP_HTTP_STATIC_USER,
+          DEFAULT_HADOOP_HTTP_STATIC_USER);
+      options.put(HADOOP_HTTP_STATIC_USER, username);
+      HttpServer2.defineFilter(timelineRestServer.getWebAppContext(),
+          "static_user_filter_timeline",
+          StaticUserWebFilter.StaticUserFilter.class.getName(),
+          options, new String[] {"/*"});
+
+      timelineRestServer.addJerseyResourcePackage(
+          PerNodeAggregatorWebService.class.getPackage().getName() + ";"
+              + GenericExceptionHandler.class.getPackage().getName() + ";"
+              + YarnJacksonJaxbJsonProvider.class.getPackage().getName(),
+          "/*");
+      timelineRestServer.setAttribute(AGGREGATOR_COLLECTION_ATTR_KEY,
+          AppLevelServiceManager.getInstance());
+      timelineRestServer.start();
     } catch (Exception e) {
       String msg = "The per-node aggregator webapp failed to start.";
       LOG.error(msg, e);
@@ -115,19 +143,6 @@ public class PerNodeAggregatorServer extends AuxiliaryService {
     }
   }
 
-  private static class TimelineServiceWebApp
-      extends WebApp implements YarnWebParams {
-    @Override
-    public void setup() {
-      bind(YarnJacksonJaxbJsonProvider.class);
-      bind(GenericExceptionHandler.class);
-      bind(PerNodeAggregatorWebService.class);
-      // bind to the global singleton
-      bind(AppLevelServiceManager.class).
-          toProvider(AppLevelServiceManagerProvider.class);
-    }
-  }
-
   // these methods can be used as the basis for future service methods if the
   // per-node aggregator runs separate from the node manager
   /**

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/aggregator/PerNodeAggregatorWebService.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.yarn.server.timelineservice.aggregator;
 
+import javax.servlet.ServletContext;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 import javax.ws.rs.*;
@@ -41,11 +42,8 @@ import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.apache.hadoop.yarn.webapp.ForbiddenException;
 import org.apache.hadoop.yarn.webapp.NotFoundException;
 
-import com.google.inject.Inject;
 import com.google.inject.Singleton;
 
-import java.net.URI;
-
 /**
  * The main per-node REST end point for timeline service writes. It is
  * essentially a container service that routes requests to the appropriate
@@ -59,12 +57,7 @@ public class PerNodeAggregatorWebService {
   private static final Log LOG =
       LogFactory.getLog(PerNodeAggregatorWebService.class);
 
-  private final AppLevelServiceManager serviceManager;
-
-  @Inject
-  public PerNodeAggregatorWebService(AppLevelServiceManager serviceManager) {
-    this.serviceManager = serviceManager;
-  }
+  private @Context ServletContext context;
 
   @XmlRootElement(name = "about")
   @XmlAccessorType(XmlAccessType.NONE)
@@ -135,7 +128,7 @@ public class PerNodeAggregatorWebService {
       if (appId == null) {
         return Response.status(Response.Status.BAD_REQUEST).build();
       }
-      AppLevelAggregatorService service = serviceManager.getService(appId);
+      AppLevelAggregatorService service = getAggregatorService(req, appId);
       if (service == null) {
         LOG.error("Application not found");
         throw new NotFoundException(); // different exception?
@@ -163,6 +156,15 @@ public class PerNodeAggregatorWebService {
     }
   }
 
+  private AppLevelAggregatorService
+      getAggregatorService(HttpServletRequest req, String appIdToParse) {
+    String appIdString = parseApplicationId(appIdToParse);
+    final AppLevelServiceManager serviceManager =
+        (AppLevelServiceManager) context.getAttribute(
+            PerNodeAggregatorServer.AGGREGATOR_COLLECTION_ATTR_KEY);
+    return serviceManager.getService(appIdString);
+  }
+
   private void init(HttpServletResponse response) {
     response.setContentType(null);
   }