YARN-1937. Added owner-only ACLs support for Timeline Client and server. Contributed by Zhijie Shen.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1597186 13f79535-47bb-0310-9956-ffa450edef68

hadoop-yarn-project/CHANGES.txt

@@ -96,6 +96,9 @@ Release 2.5.0 - UNRELEASED
     YARN-1936. Added security support for the Timeline Client. (Zhijie Shen via
     vinodkv)
 
+    YARN-1937. Added owner-only ACLs support for Timeline Client and server.
+    (Zhijie Shen via vinodkv)
+
   OPTIMIZATIONS
 
   BUG FIXES 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelinePutResponse.java

@@ -107,6 +107,17 @@ public class TimelinePutResponse {
      */
     public static final int IO_EXCEPTION = 2;
 
+    /**
+     * Error code returned if the user specifies the timeline system reserved
+     * filter key
+     */
+    public static final int SYSTEM_FILTER_CONFLICT = 3;
+
+    /**
+     * Error code returned if the user is denied to access the timeline data
+     */
+    public static final int ACCESS_DENIED = 4;
+
     private String entityId;
     private String entityType;
     private int errorCode;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/ApplicationHistoryServer.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.LeveldbTimelineStore;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security.TimelineACLsManager;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security.TimelineAuthenticationFilterInitializer;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security.TimelineDelegationTokenSecretManagerService;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.webapp.AHSWebApp;
@@ -63,6 +64,7 @@ public class ApplicationHistoryServer extends CompositeService {
   protected ApplicationHistoryManager historyManager;
   protected TimelineStore timelineStore;
   protected TimelineDelegationTokenSecretManagerService secretManagerService;
+  protected TimelineACLsManager timelineACLsManager;
   protected WebApp webApp;
 
   public ApplicationHistoryServer() {
@@ -79,6 +81,7 @@ public class ApplicationHistoryServer extends CompositeService {
     addIfService(timelineStore);
     secretManagerService = createTimelineDelegationTokenSecretManagerService(conf);
     addService(secretManagerService);
+    timelineACLsManager = createTimelineACLsManager(conf);
 
     DefaultMetricsSystem.initialize("ApplicationHistoryServer");
     JvmMetrics.initSingleton("ApplicationHistoryServer", null);
@@ -169,6 +172,10 @@ public class ApplicationHistoryServer extends CompositeService {
     return new TimelineDelegationTokenSecretManagerService();
   }
 
+  protected TimelineACLsManager createTimelineACLsManager(Configuration conf) {
+    return new TimelineACLsManager(conf);
+  }
+
   protected void startWebApp() {
     Configuration conf = getConfig();
     // Play trick to make the customized filter will only be loaded by the
@@ -196,6 +203,7 @@ public class ApplicationHistoryServer extends CompositeService {
       ahsWebApp.setApplicationHistoryManager(historyManager);
       ahsWebApp.setTimelineStore(timelineStore);
       ahsWebApp.setTimelineDelegationTokenSecretManagerService(secretManagerService);
+      ahsWebApp.setTimelineACLsManager(timelineACLsManager);
       webApp =
           WebApps
             .$for("applicationhistory", ApplicationHistoryClientService.class,

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/timeline/MemoryTimelineStore.java

@@ -19,7 +19,6 @@
 package org.apache.hadoop.yarn.server.applicationhistoryservice.timeline;
 
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
@@ -40,8 +39,8 @@ import org.apache.hadoop.yarn.api.records.timeline.TimelineEntities;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEvents;
-import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEvents.EventsOfOneEntity;
+import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
 import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse.TimelinePutError;
 
 /**
@@ -314,15 +313,29 @@ public class MemoryTimelineStore
     entityToReturn.setEntityId(entity.getEntityId());
     entityToReturn.setEntityType(entity.getEntityType());
     entityToReturn.setStartTime(entity.getStartTime());
-    entityToReturn.setEvents(fields.contains(Field.EVENTS) ?
-        entity.getEvents() : fields.contains(Field.LAST_EVENT_ONLY) ?
-            Arrays.asList(entity.getEvents().get(0)) : null);
-    entityToReturn.setRelatedEntities(fields.contains(Field.RELATED_ENTITIES) ?
-        entity.getRelatedEntities() : null);
-    entityToReturn.setPrimaryFilters(fields.contains(Field.PRIMARY_FILTERS) ?
-        entity.getPrimaryFilters() : null);
-    entityToReturn.setOtherInfo(fields.contains(Field.OTHER_INFO) ?
-        entity.getOtherInfo() : null);
+    // Deep copy
+    if (fields.contains(Field.EVENTS)) {
+      entityToReturn.addEvents(entity.getEvents());
+    } else if (fields.contains(Field.LAST_EVENT_ONLY)) {
+      entityToReturn.addEvent(entity.getEvents().get(0));
+    } else {
+      entityToReturn.setEvents(null);
+    }
+    if (fields.contains(Field.RELATED_ENTITIES)) {
+      entityToReturn.addRelatedEntities(entity.getRelatedEntities());
+    } else {
+      entityToReturn.setRelatedEntities(null);
+    }
+    if (fields.contains(Field.PRIMARY_FILTERS)) {
+      entityToReturn.addPrimaryFilters(entity.getPrimaryFilters());
+    } else {
+      entityToReturn.setPrimaryFilters(null);
+    }
+    if (fields.contains(Field.OTHER_INFO)) {
+      entityToReturn.addOtherInfo(entity.getOtherInfo());
+    } else {
+      entityToReturn.setOtherInfo(null);
+    }
     return entityToReturn;
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/timeline/TimelineStore.java

@@ -18,12 +18,25 @@
 
 package org.apache.hadoop.yarn.server.applicationhistoryservice.timeline;
 
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.service.Service;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
 
-@InterfaceAudience.Private
-@InterfaceStability.Unstable
+@Private
+@Unstable
 public interface TimelineStore extends
     Service, TimelineReader, TimelineWriter {
+
+  /**
+   * The system filter which will be automatically added to a
+   * {@link TimelineEntity}'s primary filter section when storing the entity.
+   * The filter key is case sensitive. Users are supposed not to use the key
+   * reserved by the timeline system.
+   */
+  @Private
+  enum SystemFilter {
+    ENTITY_OWNER
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/timeline/security/TimelineACLsManager.java

@@ -0,0 +1,88 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security;
+
+import java.io.IOException;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.EntityIdentifier;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore.SystemFilter;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * <code>TimelineACLsManager</code> check the entity level timeline data access.
+ */
+@Private
+public class TimelineACLsManager {
+
+  private static final Log LOG = LogFactory.getLog(TimelineACLsManager.class);
+
+  private boolean aclsEnabled;
+
+  public TimelineACLsManager(Configuration conf) {
+    aclsEnabled = conf.getBoolean(YarnConfiguration.YARN_ACL_ENABLE,
+        YarnConfiguration.DEFAULT_YARN_ACL_ENABLE);
+  }
+
+  public boolean checkAccess(UserGroupInformation callerUGI,
+      TimelineEntity entity) throws YarnException, IOException {
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Verifying the access of " + callerUGI.getShortUserName()
+          + " on the timeline entity "
+          + new EntityIdentifier(entity.getEntityId(), entity.getEntityType()));
+    }
+
+    if (!aclsEnabled) {
+      return true;
+    }
+
+    Set<Object> values =
+        entity.getPrimaryFilters().get(
+            SystemFilter.ENTITY_OWNER.toString());
+    if (values == null || values.size() != 1) {
+      throw new YarnException("Owner information of the timeline entity "
+          + new EntityIdentifier(entity.getEntityId(), entity.getEntityType())
+          + " is corrupted.");
+    }
+    String owner = values.iterator().next().toString();
+    // TODO: Currently we just check the user is the timeline entity owner. In
+    // the future, we need to check whether the user is admin or is in the
+    // allowed user/group list
+    if (callerUGI != null && callerUGI.getShortUserName().equals(owner)) {
+      return true;
+    }
+    return false;
+  }
+
+  @Private
+  @VisibleForTesting
+  public void setACLsEnabled(boolean aclsEnabled) {
+    this.aclsEnabled = aclsEnabled;
+  }
+
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/AHSWebApp.java

@@ -23,6 +23,7 @@ import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.yarn.server.api.ApplicationContext;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.ApplicationHistoryManager;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security.TimelineACLsManager;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security.TimelineDelegationTokenSecretManagerService;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
 import org.apache.hadoop.yarn.webapp.WebApp;
@@ -36,6 +37,7 @@ public class AHSWebApp extends WebApp implements YarnWebParams {
   private ApplicationHistoryManager applicationHistoryManager;
   private TimelineStore timelineStore;
   private TimelineDelegationTokenSecretManagerService secretManagerService;
+  private TimelineACLsManager timelineACLsManager;
 
   private static AHSWebApp instance = null;
 
@@ -83,6 +85,14 @@ public class AHSWebApp extends WebApp implements YarnWebParams {
     this.secretManagerService = secretManagerService;
   }
 
+  public TimelineACLsManager getTimelineACLsManager() {
+    return timelineACLsManager;
+  }
+
+  public void setTimelineACLsManager(TimelineACLsManager timelineACLsManager) {
+    this.timelineACLsManager = timelineACLsManager;
+  }
+
   @Override
   public void setup() {
     bind(YarnJacksonJaxbJsonProvider.class);
@@ -93,6 +103,7 @@ public class AHSWebApp extends WebApp implements YarnWebParams {
     bind(TimelineStore.class).toInstance(timelineStore);
     bind(TimelineDelegationTokenSecretManagerService.class).toInstance(
         secretManagerService);
+    bind(TimelineACLsManager.class).toInstance(timelineACLsManager);
     route("/", AHSController.class);
     route(pajoin("/apps", APP_STATE), AHSController.class);
     route(pajoin("/app", APPLICATION_ID), AHSController.class, "app");

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TimelineWebServices.java

@@ -25,6 +25,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 import java.util.SortedSet;
@@ -52,17 +53,21 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntities;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEvents;
 import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
+import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.EntityIdentifier;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.GenericObjectMapper;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.NameValuePair;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineReader.Field;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security.TimelineACLsManager;
 import org.apache.hadoop.yarn.util.timeline.TimelineUtils;
 import org.apache.hadoop.yarn.webapp.BadRequestException;
+import org.apache.hadoop.yarn.webapp.NotFoundException;
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
@@ -75,10 +80,13 @@ public class TimelineWebServices {
   private static final Log LOG = LogFactory.getLog(TimelineWebServices.class);
 
   private TimelineStore store;
+  private TimelineACLsManager timelineACLsManager;
 
   @Inject
-  public TimelineWebServices(TimelineStore store) {
+  public TimelineWebServices(TimelineStore store,
+      TimelineACLsManager timelineACLsManager) {
     this.store = store;
+    this.timelineACLsManager = timelineACLsManager;
   }
 
   @XmlRootElement(name = "about")
@@ -141,6 +149,9 @@ public class TimelineWebServices {
     init(res);
     TimelineEntities entities = null;
     try {
+      EnumSet<Field> fieldEnums = parseFieldsStr(fields, ",");
+      boolean modified = extendFields(fieldEnums);
+      UserGroupInformation callerUGI = getUser(req);
       entities = store.getEntities(
           parseStr(entityType),
           parseLongStr(limit),
@@ -150,7 +161,33 @@ public class TimelineWebServices {
           parseLongStr(fromTs),
           parsePairStr(primaryFilter, ":"),
           parsePairsStr(secondaryFilter, ",", ":"),
-          parseFieldsStr(fields, ","));
+          fieldEnums);
+      if (entities != null) {
+        Iterator<TimelineEntity> entitiesItr =
+            entities.getEntities().iterator();
+        while (entitiesItr.hasNext()) {
+          TimelineEntity entity = entitiesItr.next();
+          try {
+            // check ACLs
+            if (!timelineACLsManager.checkAccess(callerUGI, entity)) {
+              entitiesItr.remove();
+            } else {
+              // clean up system data
+              if (modified) {
+                entity.setPrimaryFilters(null);
+              } else {
+                cleanupOwnerInfo(entity);
+              }
+            }
+          } catch (YarnException e) {
+            LOG.error("Error when verifying access for user " + callerUGI
+                + " on the events of the timeline entity "
+                + new EntityIdentifier(entity.getEntityId(),
+                    entity.getEntityType()), e);
+            entitiesItr.remove();
+          }
+        }
+      }
     } catch (NumberFormatException e) {
       throw new BadRequestException(
           "windowStart, windowEnd or limit is not a numeric value.");
@@ -182,9 +219,25 @@ public class TimelineWebServices {
     init(res);
     TimelineEntity entity = null;
     try {
+      EnumSet<Field> fieldEnums = parseFieldsStr(fields, ",");
+      boolean modified = extendFields(fieldEnums);
       entity =
           store.getEntity(parseStr(entityId), parseStr(entityType),
-              parseFieldsStr(fields, ","));
+              fieldEnums);
+      if (entity != null) {
+        // check ACLs
+        UserGroupInformation callerUGI = getUser(req);
+        if (!timelineACLsManager.checkAccess(callerUGI, entity)) {
+          entity = null;
+        } else {
+          // clean up the system data
+          if (modified) {
+            entity.setPrimaryFilters(null);
+          } else {
+            cleanupOwnerInfo(entity);
+          }
+        }
+      }
     } catch (IllegalArgumentException e) {
       throw new BadRequestException(
           "requested invalid field.");
@@ -192,9 +245,15 @@ public class TimelineWebServices {
       LOG.error("Error getting entity", e);
       throw new WebApplicationException(e,
           Response.Status.INTERNAL_SERVER_ERROR);
+    } catch (YarnException e) {
+      LOG.error("Error getting entity", e);
+      throw new WebApplicationException(e,
+          Response.Status.INTERNAL_SERVER_ERROR);
     }
     if (entity == null) {
-      throw new WebApplicationException(Response.Status.NOT_FOUND);
+      throw new NotFoundException("Timeline entity "
+          + new EntityIdentifier(parseStr(entityId), parseStr(entityType))
+          + " is not found");
     }
     return entity;
   }
@@ -217,6 +276,7 @@ public class TimelineWebServices {
     init(res);
     TimelineEvents events = null;
     try {
+      UserGroupInformation callerUGI = getUser(req);
       events = store.getEntityTimelines(
           parseStr(entityType),
           parseArrayStr(entityId, ","),
@@ -224,6 +284,29 @@ public class TimelineWebServices {
           parseLongStr(windowStart),
           parseLongStr(windowEnd),
           parseArrayStr(eventType, ","));
+      if (events != null) {
+        Iterator<TimelineEvents.EventsOfOneEntity> eventsItr =
+            events.getAllEvents().iterator();
+        while (eventsItr.hasNext()) {
+          TimelineEvents.EventsOfOneEntity eventsOfOneEntity = eventsItr.next();
+          try {
+            TimelineEntity entity = store.getEntity(
+                eventsOfOneEntity.getEntityId(),
+                eventsOfOneEntity.getEntityType(),
+                EnumSet.of(Field.PRIMARY_FILTERS));
+            // check ACLs
+            if (!timelineACLsManager.checkAccess(callerUGI, entity)) {
+              eventsItr.remove();
+            }
+          } catch (Exception e) {
+            LOG.error("Error when verifying access for user " + callerUGI
+                + " on the events of the timeline entity "
+                + new EntityIdentifier(eventsOfOneEntity.getEntityId(),
+                    eventsOfOneEntity.getEntityType()), e);
+            eventsItr.remove();
+          }
+        }
+      }
     } catch (NumberFormatException e) {
       throw new BadRequestException(
           "windowStart, windowEnd or limit is not a numeric value.");
@@ -252,12 +335,61 @@ public class TimelineWebServices {
     if (entities == null) {
       return new TimelinePutResponse();
     }
+    UserGroupInformation callerUGI = getUser(req);
     try {
       List<EntityIdentifier> entityIDs = new ArrayList<EntityIdentifier>();
+      TimelineEntities entitiesToPut = new TimelineEntities();
+      List<TimelinePutResponse.TimelinePutError> errors =
+          new ArrayList<TimelinePutResponse.TimelinePutError>();
       for (TimelineEntity entity : entities.getEntities()) {
         EntityIdentifier entityID =
             new EntityIdentifier(entity.getEntityId(), entity.getEntityType());
+
+        // check if there is existing entity
+        try {
+          TimelineEntity existingEntity =
+              store.getEntity(entityID.getId(), entityID.getType(),
+                  EnumSet.of(Field.PRIMARY_FILTERS));
+          if (existingEntity != null
+              && !timelineACLsManager.checkAccess(callerUGI, existingEntity)) {
+            throw new YarnException("The timeline entity " + entityID
+                + " was not put by " + callerUGI + " before");
+          }
+        } catch (Exception e) {
+          // Skip the entity which already exists and was put by others
+          LOG.warn("Skip the timeline entity: " + entityID + ", because "
+              + e.getMessage());
+          TimelinePutResponse.TimelinePutError error =
+              new TimelinePutResponse.TimelinePutError();
+          error.setEntityId(entityID.getId());
+          error.setEntityType(entityID.getType());
+          error.setErrorCode(
+              TimelinePutResponse.TimelinePutError.ACCESS_DENIED);
+          errors.add(error);
+          continue;
+        }
+
+        // inject owner information for the access check
+        try {
+          injectOwnerInfo(entity,
+              callerUGI == null ? "" : callerUGI.getShortUserName());
+        } catch (YarnException e) {
+          // Skip the entity which messes up the primary filter and record the
+          // error
+          LOG.warn("Skip the timeline entity: " + entityID + ", because "
+              + e.getMessage());
+          TimelinePutResponse.TimelinePutError error =
+              new TimelinePutResponse.TimelinePutError();
+          error.setEntityId(entityID.getId());
+          error.setEntityType(entityID.getType());
+          error.setErrorCode(
+              TimelinePutResponse.TimelinePutError.SYSTEM_FILTER_CONFLICT);
+          errors.add(error);
+          continue;
+        }
+
         entityIDs.add(entityID);
+        entitiesToPut.addEntity(entity);
         if (LOG.isDebugEnabled()) {
           LOG.debug("Storing the entity " + entityID + ", JSON-style content: "
               + TimelineUtils.dumpTimelineRecordtoJSON(entity));
@@ -266,7 +398,10 @@ public class TimelineWebServices {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Storing entities: " + CSV_JOINER.join(entityIDs));
       }
-      return store.put(entities);
+      TimelinePutResponse response =  store.put(entitiesToPut);
+      // add the errors of timeline system filter key conflict
+      response.addErrors(errors);
+      return response;
     } catch (IOException e) {
       LOG.error("Error putting entities", e);
       throw new WebApplicationException(e,
@@ -350,6 +485,14 @@ public class TimelineWebServices {
     }
   }
 
+  private static boolean extendFields(EnumSet<Field> fieldEnums) {
+    boolean modified = false;
+    if (fieldEnums != null && !fieldEnums.contains(Field.PRIMARY_FILTERS)) {
+      fieldEnums.add(Field.PRIMARY_FILTERS);
+      modified = true;
+    }
+    return modified;
+  }
   private static Long parseLongStr(String str) {
     return str == null ? null : Long.parseLong(str.trim());
   }
@@ -358,4 +501,34 @@ public class TimelineWebServices {
     return str == null ? null : str.trim();
   }
 
+  private static UserGroupInformation getUser(HttpServletRequest req) {
+    String remoteUser = req.getRemoteUser();
+    UserGroupInformation callerUGI = null;
+    if (remoteUser != null) {
+      callerUGI = UserGroupInformation.createRemoteUser(remoteUser);
+    }
+    return callerUGI;
+  }
+
+  private static void injectOwnerInfo(TimelineEntity timelineEntity,
+      String owner) throws YarnException {
+    if (timelineEntity.getPrimaryFilters() != null &&
+        timelineEntity.getPrimaryFilters().containsKey(
+            TimelineStore.SystemFilter.ENTITY_OWNER)) {
+      throw new YarnException(
+          "User should not use the timeline system filter key: "
+              + TimelineStore.SystemFilter.ENTITY_OWNER);
+    }
+    timelineEntity.addPrimaryFilter(
+        TimelineStore.SystemFilter.ENTITY_OWNER
+            .toString(), owner);
+  }
+
+  private static void cleanupOwnerInfo(TimelineEntity timelineEntity) {
+    if (timelineEntity.getPrimaryFilters() != null) {
+      timelineEntity.getPrimaryFilters().remove(
+          TimelineStore.SystemFilter.ENTITY_OWNER.toString());
+    }
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/timeline/security/TestTimelineACLsManager.java

@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestTimelineACLsManager {
+
+  @Test
+  public void testYarnACLsNotEnabled() throws Exception {
+    Configuration conf = new YarnConfiguration();
+    conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, false);
+    TimelineACLsManager timelineACLsManager =
+        new TimelineACLsManager(conf);
+    TimelineEntity entity = new TimelineEntity();
+    entity.addPrimaryFilter(
+        TimelineStore.SystemFilter.ENTITY_OWNER
+            .toString(), "owner");
+    Assert.assertTrue(
+        "Always true when ACLs are not enabled",
+        timelineACLsManager.checkAccess(
+            UserGroupInformation.createRemoteUser("user"), entity));
+  }
+
+  @Test
+  public void testYarnACLsEnabled() throws Exception {
+    Configuration conf = new YarnConfiguration();
+    conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
+    TimelineACLsManager timelineACLsManager =
+        new TimelineACLsManager(conf);
+    TimelineEntity entity = new TimelineEntity();
+    entity.addPrimaryFilter(
+        TimelineStore.SystemFilter.ENTITY_OWNER
+            .toString(), "owner");
+    Assert.assertTrue(
+        "Owner should be allowed to access",
+        timelineACLsManager.checkAccess(
+            UserGroupInformation.createRemoteUser("owner"), entity));
+    Assert.assertFalse(
+        "Other shouldn't be allowed to access",
+        timelineACLsManager.checkAccess(
+            UserGroupInformation.createRemoteUser("other"), entity));
+  }
+
+  @Test
+  public void testCorruptedOwnerInfo() throws Exception {
+    Configuration conf = new YarnConfiguration();
+    conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
+    TimelineACLsManager timelineACLsManager =
+        new TimelineACLsManager(conf);
+    TimelineEntity entity = new TimelineEntity();
+    try {
+      timelineACLsManager.checkAccess(
+          UserGroupInformation.createRemoteUser("owner"), entity);
+      Assert.fail("Exception is expected");
+    } catch (YarnException e) {
+      Assert.assertTrue("It's not the exact expected exception", e.getMessage()
+          .contains("is corrupted."));
+    }
+  }
+
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-applicationhistoryservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TestTimelineWebServices.java

@@ -20,19 +20,32 @@ package org.apache.hadoop.yarn.server.applicationhistoryservice.webapp;
 
 import static org.junit.Assert.assertEquals;
 
+import java.io.IOException;
+
+import javax.inject.Singleton;
+import javax.servlet.Filter;
+import javax.servlet.FilterChain;
+import javax.servlet.FilterConfig;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletRequestWrapper;
 import javax.ws.rs.core.MediaType;
 
-import org.junit.Assert;
-
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntities;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent;
 import org.apache.hadoop.yarn.api.records.timeline.TimelineEvents;
 import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TestMemoryTimelineStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.TimelineStore;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.security.TimelineACLsManager;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
 import org.apache.hadoop.yarn.webapp.YarnJacksonJaxbJsonProvider;
+import org.junit.Assert;
 import org.junit.Test;
 
 import com.google.inject.Guice;
@@ -50,6 +63,8 @@ import com.sun.jersey.test.framework.WebAppDescriptor;
 public class TestTimelineWebServices extends JerseyTest {
 
   private static TimelineStore store;
+  private static TimelineACLsManager timelineACLsManager;
+  private static String remoteUser;
   private long beforeTime;
 
   private Injector injector = Guice.createInjector(new ServletModule() {
@@ -65,7 +80,12 @@ public class TestTimelineWebServices extends JerseyTest {
         Assert.fail();
       }
       bind(TimelineStore.class).toInstance(store);
+      Configuration conf = new YarnConfiguration();
+      conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, false);
+      timelineACLsManager = new TimelineACLsManager(conf);
+      bind(TimelineACLsManager.class).toInstance(timelineACLsManager);
       serve("/*").with(GuiceContainer.class);
+      filter("/*").through(TestFilter.class);
     }
 
   });
@@ -340,8 +360,8 @@ public class TestTimelineWebServices extends JerseyTest {
   public void testPostEntities() throws Exception {
     TimelineEntities entities = new TimelineEntities();
     TimelineEntity entity = new TimelineEntity();
-    entity.setEntityId("test id");
-    entity.setEntityType("test type");
+    entity.setEntityId("test id 1");
+    entity.setEntityType("test type 1");
     entity.setStartTime(System.currentTimeMillis());
     entities.addEntity(entity);
     WebResource r = resource();
@@ -355,14 +375,248 @@ public class TestTimelineWebServices extends JerseyTest {
     Assert.assertEquals(0, putResposne.getErrors().size());
     // verify the entity exists in the store
     response = r.path("ws").path("v1").path("timeline")
-        .path("test type").path("test id")
+        .path("test type 1").path("test id 1")
         .accept(MediaType.APPLICATION_JSON)
         .get(ClientResponse.class);
     assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
     entity = response.getEntity(TimelineEntity.class);
     Assert.assertNotNull(entity);
-    Assert.assertEquals("test id", entity.getEntityId());
-    Assert.assertEquals("test type", entity.getEntityType());
+    Assert.assertEquals("test id 1", entity.getEntityId());
+    Assert.assertEquals("test type 1", entity.getEntityType());
+  }
+
+  @Test
+  public void testPostEntitiesWithYarnACLsEnabled() throws Exception {
+    timelineACLsManager.setACLsEnabled(true);
+    remoteUser = "tester";
+    try {
+      TimelineEntities entities = new TimelineEntities();
+      TimelineEntity entity = new TimelineEntity();
+      entity.setEntityId("test id 2");
+      entity.setEntityType("test type 2");
+      entity.setStartTime(System.currentTimeMillis());
+      entities.addEntity(entity);
+      WebResource r = resource();
+      ClientResponse response = r.path("ws").path("v1").path("timeline")
+          .accept(MediaType.APPLICATION_JSON)
+          .type(MediaType.APPLICATION_JSON)
+          .post(ClientResponse.class, entities);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      TimelinePutResponse putResponse = response.getEntity(TimelinePutResponse.class);
+      Assert.assertNotNull(putResponse);
+      Assert.assertEquals(0, putResponse.getErrors().size());
+
+      // override/append timeline data in the same entity with different user
+      remoteUser = "other";
+      response = r.path("ws").path("v1").path("timeline")
+          .accept(MediaType.APPLICATION_JSON)
+          .type(MediaType.APPLICATION_JSON)
+          .post(ClientResponse.class, entities);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      putResponse = response.getEntity(TimelinePutResponse.class);
+      Assert.assertNotNull(putResponse);
+      Assert.assertEquals(1, putResponse.getErrors().size());
+      Assert.assertEquals(TimelinePutResponse.TimelinePutError.ACCESS_DENIED,
+          putResponse.getErrors().get(0).getErrorCode());
+    } finally {
+      timelineACLsManager.setACLsEnabled(false);
+      remoteUser = null;
+    }
+  }
+
+  @Test
+  public void testGetEntityWithYarnACLsEnabled() throws Exception {
+    timelineACLsManager.setACLsEnabled(true);
+    remoteUser = "tester";
+    try {
+      TimelineEntities entities = new TimelineEntities();
+      TimelineEntity entity = new TimelineEntity();
+      entity.setEntityId("test id 3");
+      entity.setEntityType("test type 3");
+      entity.setStartTime(System.currentTimeMillis());
+      entities.addEntity(entity);
+      WebResource r = resource();
+      ClientResponse response = r.path("ws").path("v1").path("timeline")
+          .accept(MediaType.APPLICATION_JSON)
+          .type(MediaType.APPLICATION_JSON)
+          .post(ClientResponse.class, entities);
+      // verify the system data will not be exposed
+      // 1. No field specification
+      response = r.path("ws").path("v1").path("timeline")
+          .path("test type 3").path("test id 3")
+          .accept(MediaType.APPLICATION_JSON)
+          .get(ClientResponse.class);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      entity = response.getEntity(TimelineEntity.class);
+      Assert.assertNull(entity.getPrimaryFilters().get(
+          TimelineStore.SystemFilter.ENTITY_OWNER.toString()));
+      // 2. other field
+      response = r.path("ws").path("v1").path("timeline")
+          .path("test type 3").path("test id 3")
+          .queryParam("fields", "relatedentities")
+          .accept(MediaType.APPLICATION_JSON)
+          .get(ClientResponse.class);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      entity = response.getEntity(TimelineEntity.class);
+      Assert.assertNull(entity.getPrimaryFilters().get(
+          TimelineStore.SystemFilter.ENTITY_OWNER.toString()));
+      // 3. primaryfilters field
+      response = r.path("ws").path("v1").path("timeline")
+          .path("test type 3").path("test id 3")
+          .queryParam("fields", "primaryfilters")
+          .accept(MediaType.APPLICATION_JSON)
+          .get(ClientResponse.class);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      entity = response.getEntity(TimelineEntity.class);
+      Assert.assertNull(entity.getPrimaryFilters().get(
+          TimelineStore.SystemFilter.ENTITY_OWNER.toString()));
+
+      // get entity with other user
+      remoteUser = "other";
+      response = r.path("ws").path("v1").path("timeline")
+          .path("test type 3").path("test id 3")
+          .accept(MediaType.APPLICATION_JSON)
+          .get(ClientResponse.class);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      assertEquals(ClientResponse.Status.NOT_FOUND,
+          response.getClientResponseStatus());
+    } finally {
+      timelineACLsManager.setACLsEnabled(false);
+      remoteUser = null;
+    }
+  }
+
+  @Test
+  public void testGetEntitiesWithYarnACLsEnabled() {
+    timelineACLsManager.setACLsEnabled(true);
+    remoteUser = "tester";
+    try {
+      TimelineEntities entities = new TimelineEntities();
+      TimelineEntity entity = new TimelineEntity();
+      entity.setEntityId("test id 4");
+      entity.setEntityType("test type 4");
+      entity.setStartTime(System.currentTimeMillis());
+      entities.addEntity(entity);
+      WebResource r = resource();
+      ClientResponse response = r.path("ws").path("v1").path("timeline")
+          .accept(MediaType.APPLICATION_JSON)
+          .type(MediaType.APPLICATION_JSON)
+          .post(ClientResponse.class, entities);
+
+      remoteUser = "other";
+      entities = new TimelineEntities();
+      entity = new TimelineEntity();
+      entity.setEntityId("test id 5");
+      entity.setEntityType("test type 4");
+      entity.setStartTime(System.currentTimeMillis());
+      entities.addEntity(entity);
+      r = resource();
+      response = r.path("ws").path("v1").path("timeline")
+          .accept(MediaType.APPLICATION_JSON)
+          .type(MediaType.APPLICATION_JSON)
+          .post(ClientResponse.class, entities);
+
+      response = r.path("ws").path("v1").path("timeline")
+          .path("test type 4")
+          .accept(MediaType.APPLICATION_JSON)
+          .get(ClientResponse.class);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      entities = response.getEntity(TimelineEntities.class);
+      assertEquals(1, entities.getEntities().size());
+      assertEquals("test type 4", entities.getEntities().get(0).getEntityType());
+      assertEquals("test id 5", entities.getEntities().get(0).getEntityId());
+    } finally {
+      timelineACLsManager.setACLsEnabled(false);
+      remoteUser = null;
+    }
+  }
+
+  @Test
+  public void testGetEventsWithYarnACLsEnabled() {
+    timelineACLsManager.setACLsEnabled(true);
+    remoteUser = "tester";
+    try {
+      TimelineEntities entities = new TimelineEntities();
+      TimelineEntity entity = new TimelineEntity();
+      entity.setEntityId("test id 5");
+      entity.setEntityType("test type 5");
+      entity.setStartTime(System.currentTimeMillis());
+      TimelineEvent event = new TimelineEvent();
+      event.setEventType("event type 1");
+      event.setTimestamp(System.currentTimeMillis());
+      entity.addEvent(event);
+      entities.addEntity(entity);
+      WebResource r = resource();
+      ClientResponse response = r.path("ws").path("v1").path("timeline")
+          .accept(MediaType.APPLICATION_JSON)
+          .type(MediaType.APPLICATION_JSON)
+          .post(ClientResponse.class, entities);
+
+      remoteUser = "other";
+      entities = new TimelineEntities();
+      entity = new TimelineEntity();
+      entity.setEntityId("test id 6");
+      entity.setEntityType("test type 5");
+      entity.setStartTime(System.currentTimeMillis());
+      event = new TimelineEvent();
+      event.setEventType("event type 2");
+      event.setTimestamp(System.currentTimeMillis());
+      entity.addEvent(event);
+      entities.addEntity(entity);
+      r = resource();
+      response = r.path("ws").path("v1").path("timeline")
+          .accept(MediaType.APPLICATION_JSON)
+          .type(MediaType.APPLICATION_JSON)
+          .post(ClientResponse.class, entities);
+
+      response = r.path("ws").path("v1").path("timeline")
+          .path("test type 5").path("events")
+          .queryParam("entityId", "test id 5,test id 6")
+          .accept(MediaType.APPLICATION_JSON)
+          .get(ClientResponse.class);
+      assertEquals(MediaType.APPLICATION_JSON_TYPE, response.getType());
+      TimelineEvents events = response.getEntity(TimelineEvents.class);
+      assertEquals(1, events.getAllEvents().size());
+      assertEquals("test id 6", events.getAllEvents().get(0).getEntityId());
+    } finally {
+      timelineACLsManager.setACLsEnabled(false);
+      remoteUser = null;
+    }
   }
 
+  @Singleton
+  private static class TestFilter implements Filter {
+
+    @Override
+    public void init(FilterConfig filterConfig) throws ServletException {
+    }
+
+    @Override
+    public void doFilter(ServletRequest request, ServletResponse response,
+        FilterChain chain) throws IOException, ServletException {
+      if (request instanceof HttpServletRequest) {
+        request =
+            new TestHttpServletRequestWrapper((HttpServletRequest) request);
+      }
+      chain.doFilter(request, response);
+    }
+
+    @Override
+    public void destroy() {
+    }
+
+  }
+
+  private static class TestHttpServletRequestWrapper extends HttpServletRequestWrapper {
+
+    public TestHttpServletRequestWrapper(HttpServletRequest request) {
+      super(request);
+    }
+
+    @Override
+    public String getRemoteUser() {
+      return TestTimelineWebServices.remoteUser;
+    }
+
+  }
 }