getDecreasedContainers();
+
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/CollectorInfo.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/CollectorInfo.java
new file mode 100644
index 00000000000..30450d655f8
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/CollectorInfo.java
@@ -0,0 +1,59 @@
+/**
+ * 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.api.records;
+
+import org.apache.hadoop.classification.InterfaceStability.Evolving;
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.yarn.util.Records;
+
+/**
+ * Collector info containing collector address and collector token passed from
+ * RM to AM in Allocate Response.
+ */
+@Public
+@Evolving
+public abstract class CollectorInfo {
+
+ protected static final long DEFAULT_TIMESTAMP_VALUE = -1;
+
+ public static CollectorInfo newInstance(String collectorAddr) {
+ return newInstance(collectorAddr, null);
+ }
+
+ public static CollectorInfo newInstance(String collectorAddr, Token token) {
+ CollectorInfo amCollectorInfo =
+ Records.newRecord(CollectorInfo.class);
+ amCollectorInfo.setCollectorAddr(collectorAddr);
+ amCollectorInfo.setCollectorToken(token);
+ return amCollectorInfo;
+ }
+
+ public abstract String getCollectorAddr();
+
+ public abstract void setCollectorAddr(String addr);
+
+ /**
+ * Get delegation token for app collector which AM will use to publish
+ * entities.
+ * @return the delegation token for app collector.
+ */
+ public abstract Token getCollectorToken();
+
+ public abstract void setCollectorToken(Token token);
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEntity.java
index a43259b5ff1..e695050e2b9 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEntity.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEntity.java
@@ -34,6 +34,7 @@ import javax.xml.bind.annotation.XmlRootElement;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Evolving;
+import org.apache.hadoop.yarn.util.TimelineServiceHelper;
/**
*
@@ -231,11 +232,8 @@ public class TimelineEntity implements Comparable {
*/
public void setRelatedEntities(
Map> relatedEntities) {
- if (relatedEntities != null && !(relatedEntities instanceof HashMap)) {
- this.relatedEntities = new HashMap>(relatedEntities);
- } else {
- this.relatedEntities = (HashMap>) relatedEntities;
- }
+ this.relatedEntities = TimelineServiceHelper.mapCastToHashMap(
+ relatedEntities);
}
/**
@@ -297,11 +295,8 @@ public class TimelineEntity implements Comparable {
* a map of primary filters
*/
public void setPrimaryFilters(Map> primaryFilters) {
- if (primaryFilters != null && !(primaryFilters instanceof HashMap)) {
- this.primaryFilters = new HashMap>(primaryFilters);
- } else {
- this.primaryFilters = (HashMap>) primaryFilters;
- }
+ this.primaryFilters =
+ TimelineServiceHelper.mapCastToHashMap(primaryFilters);
}
/**
@@ -350,11 +345,7 @@ public class TimelineEntity implements Comparable {
* a map of other information
*/
public void setOtherInfo(Map otherInfo) {
- if (otherInfo != null && !(otherInfo instanceof HashMap)) {
- this.otherInfo = new HashMap(otherInfo);
- } else {
- this.otherInfo = (HashMap) otherInfo;
- }
+ this.otherInfo = TimelineServiceHelper.mapCastToHashMap(otherInfo);
}
/**
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEvent.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEvent.java
index 73b2e729c1e..d5611f8da99 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEvent.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timeline/TimelineEvent.java
@@ -29,6 +29,7 @@ import javax.xml.bind.annotation.XmlRootElement;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Evolving;
+import org.apache.hadoop.yarn.util.TimelineServiceHelper;
/**
* The class that contains the information of an event that is related to some
@@ -135,11 +136,8 @@ public class TimelineEvent implements Comparable {
* a map of of the information of the event
*/
public void setEventInfo(Map eventInfo) {
- if (eventInfo != null && !(eventInfo instanceof HashMap)) {
- this.eventInfo = new HashMap(eventInfo);
- } else {
- this.eventInfo = (HashMap) eventInfo;
- }
+ this.eventInfo = TimelineServiceHelper.mapCastToHashMap(
+ eventInfo);
}
@Override
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ApplicationAttemptEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ApplicationAttemptEntity.java
new file mode 100644
index 00000000000..053d84edb55
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ApplicationAttemptEntity.java
@@ -0,0 +1,41 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This entity represents an application attempt.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class ApplicationAttemptEntity extends HierarchicalTimelineEntity {
+ public ApplicationAttemptEntity() {
+ super(TimelineEntityType.YARN_APPLICATION_ATTEMPT.toString());
+ }
+
+ public ApplicationAttemptEntity(TimelineEntity entity) {
+ super(entity);
+ if (!entity.getType().equals(
+ TimelineEntityType.YARN_APPLICATION_ATTEMPT.toString())) {
+ throw new IllegalArgumentException("Incompatible entity type: "
+ + getId());
+ }
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ApplicationEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ApplicationEntity.java
new file mode 100644
index 00000000000..20226aa8e9c
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ApplicationEntity.java
@@ -0,0 +1,80 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This entity represents an application.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class ApplicationEntity extends HierarchicalTimelineEntity {
+ public static final String QUEUE_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "QUEUE";
+
+ public ApplicationEntity() {
+ super(TimelineEntityType.YARN_APPLICATION.toString());
+ }
+
+ public ApplicationEntity(TimelineEntity entity) {
+ super(entity);
+ if (!entity.getType().equals(
+ TimelineEntityType.YARN_APPLICATION.toString())) {
+ throw new IllegalArgumentException("Incompatible entity type: "
+ + getId());
+ }
+ }
+
+ public String getQueue() {
+ return getInfo().get(QUEUE_INFO_KEY).toString();
+ }
+
+ public void setQueue(String queue) {
+ addInfo(QUEUE_INFO_KEY, queue);
+ }
+
+ /**
+ * Checks if the input TimelineEntity object is an ApplicationEntity.
+ *
+ * @param te TimelineEntity object.
+ * @return true if input is an ApplicationEntity, false otherwise
+ */
+ public static boolean isApplicationEntity(TimelineEntity te) {
+ return (te == null ? false
+ : te.getType().equals(TimelineEntityType.YARN_APPLICATION.toString()));
+ }
+
+ /**
+ * @param te TimelineEntity object.
+ * @param eventId event with this id needs to be fetched
+ * @return TimelineEvent if TimelineEntity contains the desired event.
+ */
+ public static TimelineEvent getApplicationEvent(TimelineEntity te,
+ String eventId) {
+ if (isApplicationEntity(te)) {
+ for (TimelineEvent event : te.getEvents()) {
+ if (event.getId().equals(eventId)) {
+ return event;
+ }
+ }
+ }
+ return null;
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ClusterEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ClusterEntity.java
new file mode 100644
index 00000000000..1f96505f98c
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ClusterEntity.java
@@ -0,0 +1,40 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This entity represents a YARN cluster.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class ClusterEntity extends HierarchicalTimelineEntity {
+ public ClusterEntity() {
+ super(TimelineEntityType.YARN_CLUSTER.toString());
+ }
+
+ public ClusterEntity(TimelineEntity entity) {
+ super(entity);
+ if (!entity.getType().equals(TimelineEntityType.YARN_CLUSTER.toString())) {
+ throw new IllegalArgumentException("Incompatible entity type: "
+ + getId());
+ }
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ContainerEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ContainerEntity.java
new file mode 100644
index 00000000000..f61920f4950
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/ContainerEntity.java
@@ -0,0 +1,41 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This entity represents a container belonging to an application attempt.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class ContainerEntity extends HierarchicalTimelineEntity {
+ public ContainerEntity() {
+ super(TimelineEntityType.YARN_CONTAINER.toString());
+ }
+
+ public ContainerEntity(TimelineEntity entity) {
+ super(entity);
+ if (!entity.getType().equals(
+ TimelineEntityType.YARN_CONTAINER.toString())) {
+ throw new IllegalArgumentException("Incompatible entity type: "
+ + getId());
+ }
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/FlowActivityEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/FlowActivityEntity.java
new file mode 100644
index 00000000000..cf19328f675
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/FlowActivityEntity.java
@@ -0,0 +1,191 @@
+/*
+ * 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.api.records.timelineservice;
+
+import java.util.Collection;
+import java.util.Date;
+import java.util.NavigableSet;
+import java.util.TreeSet;
+
+import javax.xml.bind.annotation.XmlElement;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+
+/**
+ * Entity that represents a record for flow activity. It's essentially a
+ * container entity for flow runs with limited information.
+ */
+@Public
+@Unstable
+public class FlowActivityEntity extends TimelineEntity {
+ public static final String CLUSTER_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "CLUSTER";
+ public static final String DATE_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "DATE";
+ public static final String USER_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "USER";
+ public static final String FLOW_NAME_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "FLOW_NAME";
+
+ private final NavigableSet flowRuns = new TreeSet<>();
+
+ public FlowActivityEntity() {
+ super(TimelineEntityType.YARN_FLOW_ACTIVITY.toString());
+ // set config to null
+ setConfigs(null);
+ }
+
+ public FlowActivityEntity(String cluster, long time, String user,
+ String flowName) {
+ this();
+ setCluster(cluster);
+ setDate(time);
+ setUser(user);
+ setFlowName(flowName);
+ }
+
+ public FlowActivityEntity(TimelineEntity entity) {
+ super(entity);
+ if (!TimelineEntityType.YARN_FLOW_ACTIVITY.matches(entity.getType())) {
+ throw new IllegalArgumentException("Incompatible entity type: " +
+ getId());
+ }
+ // set config to null
+ setConfigs(null);
+ }
+
+ @XmlElement(name = "id")
+ @Override
+ public String getId() {
+ // flow activity: cluster/day/user@flow_name
+ String id = super.getId();
+ if (id == null) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(getCluster());
+ sb.append('/');
+ sb.append(getDate().getTime());
+ sb.append('/');
+ sb.append(getUser());
+ sb.append('@');
+ sb.append(getFlowName());
+ id = sb.toString();
+ setId(id);
+ }
+ return id;
+ }
+
+ @Override
+ public int compareTo(TimelineEntity entity) {
+ int comparison = getType().compareTo(entity.getType());
+ if (comparison == 0) {
+ // order by cluster, date (descending), user, and flow name
+ FlowActivityEntity other = (FlowActivityEntity)entity;
+ int clusterComparison = getCluster().compareTo(other.getCluster());
+ if (clusterComparison != 0) {
+ return clusterComparison;
+ }
+ int dateComparisonDescending =
+ (int)(other.getDate().getTime() - getDate().getTime()); // descending
+ if (dateComparisonDescending != 0) {
+ return dateComparisonDescending; // descending
+ }
+ int userComparison = getUser().compareTo(other.getUser());
+ if (userComparison != 0) {
+ return userComparison;
+ }
+ return getFlowName().compareTo(other.getFlowName());
+ } else {
+ return comparison;
+ }
+ }
+
+ /**
+ * Reuse the base class equals method.
+ */
+ @Override
+ public boolean equals(Object obj) {
+ return super.equals(obj);
+ }
+
+ /**
+ * Reuse the base class hashCode method.
+ */
+ @Override
+ public int hashCode() {
+ return super.hashCode();
+ }
+
+ public String getCluster() {
+ return (String)getInfo().get(CLUSTER_INFO_KEY);
+ }
+
+ public void setCluster(String cluster) {
+ addInfo(CLUSTER_INFO_KEY, cluster);
+ }
+
+ public Date getDate() {
+ Object date = getInfo().get(DATE_INFO_KEY);
+ if (date != null) {
+ if (date instanceof Long) {
+ return new Date((Long)date);
+ } else if (date instanceof Date) {
+ return (Date)date;
+ }
+ }
+ return null;
+ }
+
+ public void setDate(long time) {
+ Date date = new Date(time);
+ addInfo(DATE_INFO_KEY, date);
+ }
+
+ public String getUser() {
+ return (String)getInfo().get(USER_INFO_KEY);
+ }
+
+ public void setUser(String user) {
+ addInfo(USER_INFO_KEY, user);
+ }
+
+ public String getFlowName() {
+ return (String)getInfo().get(FLOW_NAME_INFO_KEY);
+ }
+
+ public void setFlowName(String flowName) {
+ addInfo(FLOW_NAME_INFO_KEY, flowName);
+ }
+
+ public void addFlowRun(FlowRunEntity run) {
+ flowRuns.add(run);
+ }
+
+ public void addFlowRuns(Collection runs) {
+ flowRuns.addAll(runs);
+ }
+
+ @XmlElement(name = "flowruns")
+ public NavigableSet getFlowRuns() {
+ return flowRuns;
+ }
+
+ public int getNumberOfRuns() {
+ return flowRuns.size();
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/FlowRunEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/FlowRunEntity.java
new file mode 100644
index 00000000000..5c666cdeb7b
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/FlowRunEntity.java
@@ -0,0 +1,126 @@
+/*
+ * 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.api.records.timelineservice;
+
+import javax.xml.bind.annotation.XmlElement;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This entity represents a flow run.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class FlowRunEntity extends HierarchicalTimelineEntity {
+ public static final String USER_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "USER";
+ public static final String FLOW_NAME_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "FLOW_NAME";
+ public static final String FLOW_VERSION_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "FLOW_VERSION";
+ public static final String FLOW_RUN_ID_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "FLOW_RUN_ID";
+ public static final String FLOW_RUN_END_TIME =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "FLOW_RUN_END_TIME";
+
+ public FlowRunEntity() {
+ super(TimelineEntityType.YARN_FLOW_RUN.toString());
+ // set config to null
+ setConfigs(null);
+ }
+
+ public FlowRunEntity(TimelineEntity entity) {
+ super(entity);
+ if (!entity.getType().equals(
+ TimelineEntityType.YARN_FLOW_RUN.toString())) {
+ throw new IllegalArgumentException("Incompatible entity type: "
+ + getId());
+ }
+ // set config to null
+ setConfigs(null);
+ }
+
+ @XmlElement(name = "id")
+ @Override
+ public String getId() {
+ //Flow id schema: user@flow_name(or id)/run_id
+ String id = super.getId();
+ if (id == null) {
+ StringBuilder sb = new StringBuilder();
+ sb.append(getInfo().get(USER_INFO_KEY).toString());
+ sb.append('@');
+ sb.append(getInfo().get(FLOW_NAME_INFO_KEY).toString());
+ sb.append('/');
+ sb.append(getInfo().get(FLOW_RUN_ID_INFO_KEY).toString());
+ id = sb.toString();
+ setId(id);
+ }
+ return id;
+ }
+
+ public String getUser() {
+ return (String)getInfo().get(USER_INFO_KEY);
+ }
+
+ public void setUser(String user) {
+ addInfo(USER_INFO_KEY, user);
+ }
+
+ public String getName() {
+ return (String)getInfo().get(FLOW_NAME_INFO_KEY);
+ }
+
+ public void setName(String name) {
+ addInfo(FLOW_NAME_INFO_KEY, name);
+ }
+
+ public String getVersion() {
+ return (String)getInfo().get(FLOW_VERSION_INFO_KEY);
+ }
+
+ public void setVersion(String version) {
+ addInfo(FLOW_VERSION_INFO_KEY, version);
+ }
+
+ public long getRunId() {
+ Object runId = getInfo().get(FLOW_RUN_ID_INFO_KEY);
+ return runId == null ? 0L : ((Number) runId).longValue();
+ }
+
+ public void setRunId(long runId) {
+ addInfo(FLOW_RUN_ID_INFO_KEY, runId);
+ }
+
+ public long getStartTime() {
+ return getCreatedTime();
+ }
+
+ public void setStartTime(long startTime) {
+ setCreatedTime(startTime);
+ }
+
+ public long getMaxEndTime() {
+ Object time = getInfo().get(FLOW_RUN_END_TIME);
+ return time == null ? 0L : ((Number) time).longValue();
+ }
+
+ public void setMaxEndTime(long endTime) {
+ addInfo(FLOW_RUN_END_TIME, endTime);
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/HierarchicalTimelineEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/HierarchicalTimelineEntity.java
new file mode 100644
index 00000000000..4744e39c1a1
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/HierarchicalTimelineEntity.java
@@ -0,0 +1,133 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * This class extends timeline entity and defines parent-child relationships
+ * with other entities.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public abstract class HierarchicalTimelineEntity extends TimelineEntity {
+ public static final String PARENT_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "PARENT_ENTITY";
+ public static final String CHILDREN_INFO_KEY =
+ TimelineEntity.SYSTEM_INFO_KEY_PREFIX + "CHILDREN_ENTITY";
+
+ HierarchicalTimelineEntity(TimelineEntity entity) {
+ super(entity);
+ }
+
+ HierarchicalTimelineEntity(String type) {
+ super(type);
+ }
+
+ public Identifier getParent() {
+ Object obj = getInfo().get(PARENT_INFO_KEY);
+ if (obj != null) {
+ if (obj instanceof Identifier) {
+ return (Identifier) obj;
+ } else {
+ throw new YarnRuntimeException(
+ "Parent info is invalid identifier object");
+ }
+ }
+ return null;
+ }
+
+ public void setParent(Identifier parent) {
+ validateParent(parent.getType());
+ addInfo(PARENT_INFO_KEY, parent);
+ }
+
+ public void setParent(String type, String id) {
+ setParent(new Identifier(type, id));
+ }
+
+ @SuppressWarnings("unchecked")
+ public Set getChildren() {
+ Object identifiers = getInfo().get(CHILDREN_INFO_KEY);
+ if (identifiers == null) {
+ return new HashSet<>();
+ }
+ TimelineEntityType thisType = TimelineEntityType.valueOf(getType());
+ if (identifiers instanceof Set) {
+ for (Object identifier : (Set) identifiers) {
+ if (!(identifier instanceof Identifier)) {
+ throw new YarnRuntimeException(
+ "Children info contains invalid identifier object");
+ } else {
+ validateChild((Identifier) identifier, thisType);
+ }
+ }
+ } else {
+ throw new YarnRuntimeException(
+ "Children info is invalid identifier set");
+ }
+ Set children = (Set) identifiers;
+ return children;
+ }
+
+ public void setChildren(Set children) {
+ addInfo(CHILDREN_INFO_KEY, children);
+ }
+
+ public void addChildren(Set children) {
+ TimelineEntityType thisType = TimelineEntityType.valueOf(getType());
+ for (Identifier child : children) {
+ validateChild(child, thisType);
+ }
+ Set existingChildren = getChildren();
+ existingChildren.addAll(children);
+ setChildren(existingChildren);
+ }
+
+ public void addChild(Identifier child) {
+ addChildren(Collections.singleton(child));
+ }
+
+ public void addChild(String type, String id) {
+ addChild(new Identifier(type, id));
+ }
+
+ private void validateParent(String type) {
+ TimelineEntityType parentType = TimelineEntityType.valueOf(type);
+ TimelineEntityType thisType = TimelineEntityType.valueOf(getType());
+ if (!thisType.isParent(parentType)) {
+ throw new IllegalArgumentException(
+ type + " is not the acceptable parent of " + this.getType());
+ }
+ }
+
+ private void validateChild(Identifier child, TimelineEntityType thisType) {
+ TimelineEntityType childType = TimelineEntityType.valueOf(child.getType());
+ if (!thisType.isChild(childType)) {
+ throw new IllegalArgumentException(
+ child.getType() + " is not the acceptable child of " +
+ this.getType());
+ }
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/QueueEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/QueueEntity.java
new file mode 100644
index 00000000000..b654450fc2c
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/QueueEntity.java
@@ -0,0 +1,40 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This entity represents a queue.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class QueueEntity extends HierarchicalTimelineEntity {
+ public QueueEntity() {
+ super(TimelineEntityType.YARN_QUEUE.toString());
+ }
+
+ public QueueEntity(TimelineEntity entity) {
+ super(entity);
+ if (!entity.getType().equals(TimelineEntityType.YARN_QUEUE.toString())) {
+ throw new IllegalArgumentException("Incompatible entity type: "
+ + getId());
+ }
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntities.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntities.java
new file mode 100644
index 00000000000..63989e682eb
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntities.java
@@ -0,0 +1,62 @@
+/*
+ * 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.api.records.timelineservice;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This class hosts a set of timeline entities.
+ */
+@XmlRootElement(name = "entities")
+@XmlAccessorType(XmlAccessType.NONE)
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class TimelineEntities {
+
+ private List entities = new ArrayList<>();
+
+ public TimelineEntities() {
+
+ }
+
+ @XmlElement(name = "entities")
+ public List getEntities() {
+ return entities;
+ }
+
+ public void setEntities(List timelineEntities) {
+ this.entities = timelineEntities;
+ }
+
+ public void addEntities(List timelineEntities) {
+ this.entities.addAll(timelineEntities);
+ }
+
+ public void addEntity(TimelineEntity entity) {
+ entities.add(entity);
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntity.java
new file mode 100644
index 00000000000..845e2cc547b
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntity.java
@@ -0,0 +1,610 @@
+/*
+ * 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.api.records.timelineservice;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.NavigableSet;
+import java.util.Set;
+import java.util.TreeSet;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.yarn.util.TimelineServiceHelper;
+import org.codehaus.jackson.annotate.JsonSetter;
+
+/**
+ * The basic timeline entity data structure for timeline service v2. Timeline
+ * entity objects are not thread safe and should not be accessed concurrently.
+ * All collection members will be initialized into empty collections. Two
+ * timeline entities are equal iff. their type and id are identical.
+ *
+ * All non-primitive type, non-collection members will be initialized into null.
+ * User should set the type and id of a timeline entity to make it valid (can be
+ * checked by using the {@link #isValid()} method). Callers to the getters
+ * should perform null checks for non-primitive type, non-collection members.
+ *
+ * Callers are recommended not to alter the returned collection objects from the
+ * getters.
+ */
+@XmlRootElement(name = "entity")
+@XmlAccessorType(XmlAccessType.NONE)
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class TimelineEntity implements Comparable {
+ protected final static String SYSTEM_INFO_KEY_PREFIX = "SYSTEM_INFO_";
+ public final static long DEFAULT_ENTITY_PREFIX = 0L;
+
+ /**
+ * Identifier of timeline entity(entity id + entity type).
+ */
+ @XmlRootElement(name = "identifier")
+ @XmlAccessorType(XmlAccessType.NONE)
+ public static class Identifier {
+ private String type;
+ private String id;
+
+ public Identifier(String type, String id) {
+ this.type = type;
+ this.id = id;
+ }
+
+ public Identifier() {
+
+ }
+
+ @XmlElement(name = "type")
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String entityType) {
+ this.type = entityType;
+ }
+
+ @XmlElement(name = "id")
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String entityId) {
+ this.id = entityId;
+ }
+
+ @Override
+ public String toString() {
+ return "TimelineEntity[" +
+ "type='" + type + '\'' +
+ ", id='" + id + '\'' + "]";
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((id == null) ? 0 : id.hashCode());
+ result =
+ prime * result + ((type == null) ? 0 : type.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof Identifier)) {
+ return false;
+ }
+ Identifier other = (Identifier) obj;
+ if (id == null) {
+ if (other.getId() != null) {
+ return false;
+ }
+ } else if (!id.equals(other.getId())) {
+ return false;
+ }
+ if (type == null) {
+ if (other.getType() != null) {
+ return false;
+ }
+ } else if (!type.equals(other.getType())) {
+ return false;
+ }
+ return true;
+ }
+ }
+
+ private TimelineEntity real;
+ private Identifier identifier;
+ private HashMap info = new HashMap<>();
+ private HashMap configs = new HashMap<>();
+ private Set metrics = new HashSet<>();
+ // events should be sorted by timestamp in descending order
+ private NavigableSet events = new TreeSet<>();
+ private HashMap> isRelatedToEntities = new HashMap<>();
+ private HashMap> relatesToEntities = new HashMap<>();
+ private Long createdTime;
+ private long idPrefix;
+
+ public TimelineEntity() {
+ identifier = new Identifier();
+ }
+
+ /**
+ *
+ * The constuctor is used to construct a proxy {@link TimelineEntity} or its
+ * subclass object from the real entity object that carries information.
+ *
+ *
+ *
+ * It is usually used in the case where we want to recover class polymorphism
+ * after deserializing the entity from its JSON form.
+ *
+ * @param entity the real entity that carries information
+ */
+ public TimelineEntity(TimelineEntity entity) {
+ real = entity.getReal();
+ }
+
+ protected TimelineEntity(String type) {
+ this();
+ identifier.type = type;
+ }
+
+ @XmlElement(name = "type")
+ public String getType() {
+ if (real == null) {
+ return identifier.type;
+ } else {
+ return real.getType();
+ }
+ }
+
+ public void setType(String type) {
+ if (real == null) {
+ identifier.type = type;
+ } else {
+ real.setType(type);
+ }
+ }
+
+ @XmlElement(name = "id")
+ public String getId() {
+ if (real == null) {
+ return identifier.id;
+ } else {
+ return real.getId();
+ }
+ }
+
+ public void setId(String id) {
+ if (real == null) {
+ identifier.id = id;
+ } else {
+ real.setId(id);
+ }
+ }
+
+ public Identifier getIdentifier() {
+ if (real == null) {
+ return identifier;
+ } else {
+ return real.getIdentifier();
+ }
+ }
+
+ public void setIdentifier(Identifier entityIdentifier) {
+ if (real == null) {
+ this.identifier = entityIdentifier;
+ } else {
+ real.setIdentifier(entityIdentifier);
+ }
+ }
+
+ // required by JAXB
+ @InterfaceAudience.Private
+ @XmlElement(name = "info")
+ public HashMap getInfoJAXB() {
+ if (real == null) {
+ return info;
+ } else {
+ return real.getInfoJAXB();
+ }
+ }
+
+ public Map getInfo() {
+ if (real == null) {
+ return info;
+ } else {
+ return real.getInfo();
+ }
+ }
+
+ public void setInfo(Map entityInfos) {
+ if (real == null) {
+ this.info = TimelineServiceHelper.mapCastToHashMap(entityInfos);
+ } else {
+ real.setInfo(entityInfos);
+ }
+ }
+
+ public void addInfo(Map entityInfos) {
+ if (real == null) {
+ this.info.putAll(entityInfos);
+ } else {
+ real.addInfo(entityInfos);
+ }
+ }
+
+ public void addInfo(String key, Object value) {
+ if (real == null) {
+ info.put(key, value);
+ } else {
+ real.addInfo(key, value);
+ }
+ }
+
+ // required by JAXB
+ @InterfaceAudience.Private
+ @XmlElement(name = "configs")
+ public HashMap getConfigsJAXB() {
+ if (real == null) {
+ return configs;
+ } else {
+ return real.getConfigsJAXB();
+ }
+ }
+
+ public Map getConfigs() {
+ if (real == null) {
+ return configs;
+ } else {
+ return real.getConfigs();
+ }
+ }
+
+ public void setConfigs(Map entityConfigs) {
+ if (real == null) {
+ this.configs = TimelineServiceHelper.mapCastToHashMap(entityConfigs);
+ } else {
+ real.setConfigs(entityConfigs);
+ }
+ }
+
+ public void addConfigs(Map entityConfigs) {
+ if (real == null) {
+ this.configs.putAll(entityConfigs);
+ } else {
+ real.addConfigs(entityConfigs);
+ }
+ }
+
+ public void addConfig(String key, String value) {
+ if (real == null) {
+ configs.put(key, value);
+ } else {
+ real.addConfig(key, value);
+ }
+ }
+
+ @XmlElement(name = "metrics")
+ public Set getMetrics() {
+ if (real == null) {
+ return metrics;
+ } else {
+ return real.getMetrics();
+ }
+ }
+
+ public void setMetrics(Set entityMetrics) {
+ if (real == null) {
+ this.metrics = entityMetrics;
+ } else {
+ real.setMetrics(entityMetrics);
+ }
+ }
+
+ public void addMetrics(Set entityMetrics) {
+ if (real == null) {
+ this.metrics.addAll(entityMetrics);
+ } else {
+ real.addMetrics(entityMetrics);
+ }
+ }
+
+ public void addMetric(TimelineMetric metric) {
+ if (real == null) {
+ metrics.add(metric);
+ } else {
+ real.addMetric(metric);
+ }
+ }
+
+ @XmlElement(name = "events")
+ public NavigableSet getEvents() {
+ if (real == null) {
+ return events;
+ } else {
+ return real.getEvents();
+ }
+ }
+
+ public void setEvents(NavigableSet entityEvents) {
+ if (real == null) {
+ this.events = entityEvents;
+ } else {
+ real.setEvents(entityEvents);
+ }
+ }
+
+ public void addEvents(Set entityEvents) {
+ if (real == null) {
+ this.events.addAll(entityEvents);
+ } else {
+ real.addEvents(entityEvents);
+ }
+ }
+
+ public void addEvent(TimelineEvent event) {
+ if (real == null) {
+ events.add(event);
+ } else {
+ real.addEvent(event);
+ }
+ }
+
+ public Map> getIsRelatedToEntities() {
+ if (real == null) {
+ return isRelatedToEntities;
+ } else {
+ return real.getIsRelatedToEntities();
+ }
+ }
+
+ // required by JAXB
+ @InterfaceAudience.Private
+ @XmlElement(name = "isrelatedto")
+ public HashMap> getIsRelatedToEntitiesJAXB() {
+ if (real == null) {
+ return isRelatedToEntities;
+ } else {
+ return real.getIsRelatedToEntitiesJAXB();
+ }
+ }
+
+ @JsonSetter("isrelatedto")
+ public void setIsRelatedToEntities(
+ Map> isRelatedTo) {
+ if (real == null) {
+ this.isRelatedToEntities =
+ TimelineServiceHelper.mapCastToHashMap(isRelatedTo);
+ } else {
+ real.setIsRelatedToEntities(isRelatedTo);
+ }
+ }
+
+ public void addIsRelatedToEntities(
+ Map> isRelatedTo) {
+ if (real == null) {
+ for (Map.Entry> entry : isRelatedTo.entrySet()) {
+ Set ids = this.isRelatedToEntities.get(entry.getKey());
+ if (ids == null) {
+ ids = new HashSet<>();
+ this.isRelatedToEntities.put(entry.getKey(), ids);
+ }
+ ids.addAll(entry.getValue());
+ }
+ } else {
+ real.addIsRelatedToEntities(isRelatedTo);
+ }
+ }
+
+ public void addIsRelatedToEntity(String type, String id) {
+ if (real == null) {
+ Set ids = isRelatedToEntities.get(type);
+ if (ids == null) {
+ ids = new HashSet<>();
+ isRelatedToEntities.put(type, ids);
+ }
+ ids.add(id);
+ } else {
+ real.addIsRelatedToEntity(type, id);
+ }
+ }
+
+ // required by JAXB
+ @InterfaceAudience.Private
+ @XmlElement(name = "relatesto")
+ public HashMap> getRelatesToEntitiesJAXB() {
+ if (real == null) {
+ return relatesToEntities;
+ } else {
+ return real.getRelatesToEntitiesJAXB();
+ }
+ }
+
+ public Map> getRelatesToEntities() {
+ if (real == null) {
+ return relatesToEntities;
+ } else {
+ return real.getRelatesToEntities();
+ }
+ }
+
+ public void addRelatesToEntities(Map> relatesTo) {
+ if (real == null) {
+ for (Map.Entry> entry : relatesTo.entrySet()) {
+ Set ids = this.relatesToEntities.get(entry.getKey());
+ if (ids == null) {
+ ids = new HashSet<>();
+ this.relatesToEntities.put(entry.getKey(), ids);
+ }
+ ids.addAll(entry.getValue());
+ }
+ } else {
+ real.addRelatesToEntities(relatesTo);
+ }
+ }
+
+ public void addRelatesToEntity(String type, String id) {
+ if (real == null) {
+ Set ids = relatesToEntities.get(type);
+ if (ids == null) {
+ ids = new HashSet<>();
+ relatesToEntities.put(type, ids);
+ }
+ ids.add(id);
+ } else {
+ real.addRelatesToEntity(type, id);
+ }
+ }
+
+ @JsonSetter("relatesto")
+ public void setRelatesToEntities(Map> relatesTo) {
+ if (real == null) {
+ this.relatesToEntities =
+ TimelineServiceHelper.mapCastToHashMap(relatesTo);
+ } else {
+ real.setRelatesToEntities(relatesTo);
+ }
+ }
+
+ @XmlElement(name = "createdtime")
+ public Long getCreatedTime() {
+ if (real == null) {
+ return createdTime;
+ } else {
+ return real.getCreatedTime();
+ }
+ }
+
+ @JsonSetter("createdtime")
+ public void setCreatedTime(Long createdTs) {
+ if (real == null) {
+ this.createdTime = createdTs;
+ } else {
+ real.setCreatedTime(createdTs);
+ }
+ }
+
+ /**
+ * Set UID in info which will be then used for query by UI.
+ * @param uidKey key for UID in info.
+ * @param uId UID to be set for the key.
+ */
+ public void setUID(String uidKey, String uId) {
+ if (real == null) {
+ info.put(uidKey, uId);
+ } else {
+ real.addInfo(uidKey, uId);
+ }
+ }
+
+ public boolean isValid() {
+ return (getId() != null && getType() != null);
+ }
+
+ // When get hashCode for a timeline entity, or check if two timeline entities
+ // are equal, we only compare their identifiers (id and type)
+ @Override
+ public int hashCode() {
+ return getIdentifier().hashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (!(obj instanceof TimelineEntity)) {
+ return false;
+ }
+ TimelineEntity other = (TimelineEntity) obj;
+ return getIdentifier().equals(other.getIdentifier());
+ }
+
+ @Override
+ public int compareTo(TimelineEntity other) {
+ int comparison = getType().compareTo(other.getType());
+ if (comparison == 0) {
+ if (getIdPrefix() > other.getIdPrefix()) {
+ // Descending order by entity id prefix
+ return -1;
+ } else if (getIdPrefix() < other.getIdPrefix()) {
+ return 1;
+ } else {
+ return getId().compareTo(other.getId());
+ }
+ } else {
+ return comparison;
+ }
+ }
+
+ protected TimelineEntity getReal() {
+ return real == null ? this : real;
+ }
+
+ public String toString() {
+ if (real == null) {
+ return identifier.toString();
+ } else {
+ return real.toString();
+ }
+ }
+
+ @XmlElement(name = "idprefix")
+ public long getIdPrefix() {
+ if (real == null) {
+ return idPrefix;
+ } else {
+ return real.getIdPrefix();
+ }
+ }
+
+ /**
+ * Sets idPrefix for an entity.
+ *
+ * Note: Entities will be stored in the order of idPrefix specified.
+ * If users decide to set idPrefix for an entity, they MUST provide
+ * the same prefix for every update of this entity.
+ *
+ * Example:
+ * TimelineEntity entity = new TimelineEntity();
+ * entity.setIdPrefix(value);
+ *
+ * Users can use {@link TimelineServiceHelper#invertLong(long)} to invert
+ * the prefix if necessary.
+ *
+ * @param entityIdPrefix prefix for an entity.
+ */
+ @JsonSetter("idprefix")
+ public void setIdPrefix(long entityIdPrefix) {
+ if (real == null) {
+ this.idPrefix = entityIdPrefix;
+ } else {
+ real.setIdPrefix(entityIdPrefix);
+ }
+ }
+}
\ No newline at end of file
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntityType.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntityType.java
new file mode 100644
index 00000000000..8fcc2ae08d4
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEntityType.java
@@ -0,0 +1,101 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Defines type of entity.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public enum TimelineEntityType {
+ YARN_CLUSTER,
+ YARN_FLOW_RUN,
+ YARN_APPLICATION,
+ YARN_APPLICATION_ATTEMPT,
+ YARN_CONTAINER,
+ YARN_USER,
+ YARN_QUEUE,
+ YARN_FLOW_ACTIVITY;
+
+ /**
+ * Whether the input type can be a parent of this entity.
+ *
+ * @param type entity type.
+ * @return true, if this entity type is parent of passed entity type, false
+ * otherwise.
+ */
+ public boolean isParent(TimelineEntityType type) {
+ switch (this) {
+ case YARN_CLUSTER:
+ return false;
+ case YARN_FLOW_RUN:
+ return YARN_FLOW_RUN == type || YARN_CLUSTER == type;
+ case YARN_APPLICATION:
+ return YARN_FLOW_RUN == type || YARN_CLUSTER == type;
+ case YARN_APPLICATION_ATTEMPT:
+ return YARN_APPLICATION == type;
+ case YARN_CONTAINER:
+ return YARN_APPLICATION_ATTEMPT == type;
+ case YARN_QUEUE:
+ return YARN_QUEUE == type;
+ default:
+ return false;
+ }
+ }
+
+ /**
+ * Whether the input type can be a child of this entity.
+ *
+ * @param type entity type.
+ * @return true, if this entity type is child of passed entity type, false
+ * otherwise.
+ */
+ public boolean isChild(TimelineEntityType type) {
+ switch (this) {
+ case YARN_CLUSTER:
+ return YARN_FLOW_RUN == type || YARN_APPLICATION == type;
+ case YARN_FLOW_RUN:
+ return YARN_FLOW_RUN == type || YARN_APPLICATION == type;
+ case YARN_APPLICATION:
+ return YARN_APPLICATION_ATTEMPT == type;
+ case YARN_APPLICATION_ATTEMPT:
+ return YARN_CONTAINER == type;
+ case YARN_CONTAINER:
+ return false;
+ case YARN_QUEUE:
+ return YARN_QUEUE == type;
+ default:
+ return false;
+ }
+ }
+
+ /**
+ * Whether the type of this entity matches the type indicated by the input
+ * argument.
+ *
+ * @param typeString entity type represented as a string.
+ * @return true, if string representation of this entity type matches the
+ * entity type passed.
+ */
+ public boolean matches(String typeString) {
+ return toString().equals(typeString);
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEvent.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEvent.java
new file mode 100644
index 00000000000..87fc291fc1d
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineEvent.java
@@ -0,0 +1,133 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.yarn.util.TimelineServiceHelper;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * This class contains the information of an event that belongs to an entity.
+ * Users are free to define what the event means, such as starting an
+ * application, container being allocated, etc.
+ */
+@XmlRootElement(name = "event")
+@XmlAccessorType(XmlAccessType.NONE)
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class TimelineEvent implements Comparable {
+ public static final long INVALID_TIMESTAMP = 0L;
+
+ private String id;
+ private HashMap info = new HashMap<>();
+ private long timestamp;
+
+ public TimelineEvent() {
+
+ }
+
+ @XmlElement(name = "id")
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String eventId) {
+ this.id = eventId;
+ }
+
+ // required by JAXB
+ @InterfaceAudience.Private
+ @XmlElement(name = "info")
+ public HashMap getInfoJAXB() {
+ return info;
+ }
+
+ public Map getInfo() {
+ return info;
+ }
+
+ public void setInfo(Map infos) {
+ this.info = TimelineServiceHelper.mapCastToHashMap(infos);
+ }
+
+ public void addInfo(Map infos) {
+ this.info.putAll(infos);
+ }
+
+ public void addInfo(String key, Object value) {
+ info.put(key, value);
+ }
+
+ @XmlElement(name = "timestamp")
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(long ts) {
+ this.timestamp = ts;
+ }
+
+ public boolean isValid() {
+ return (id != null && timestamp != INVALID_TIMESTAMP);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = (int) (timestamp ^ (timestamp >>> 32));
+ result = 31 * result + id.hashCode();
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof TimelineEvent)) {
+ return false;
+ }
+
+ TimelineEvent event = (TimelineEvent) o;
+
+ if (timestamp != event.timestamp) {
+ return false;
+ }
+ if (!id.equals(event.id)) {
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public int compareTo(TimelineEvent other) {
+ if (timestamp > other.timestamp) {
+ return -1;
+ } else if (timestamp < other.timestamp) {
+ return 1;
+ } else {
+ return id.compareTo(other.id);
+ }
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetric.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetric.java
new file mode 100644
index 00000000000..5c908d69a6a
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetric.java
@@ -0,0 +1,289 @@
+/*
+ * 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.api.records.timelineservice;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.TreeMap;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+
+/**
+ * This class contains the information of a metric that is related to some
+ * entity. Metric can either be a time series or single value.
+ */
+@XmlRootElement(name = "metric")
+@XmlAccessorType(XmlAccessType.NONE)
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class TimelineMetric {
+
+ /**
+ * Type of metric.
+ */
+ public static enum Type {
+ SINGLE_VALUE,
+ TIME_SERIES
+ }
+
+ private Type type;
+ private String id;
+ // By default, not to do any aggregation operations. This field will NOT be
+ // persisted (like a "transient" member).
+ private TimelineMetricOperation realtimeAggregationOp
+ = TimelineMetricOperation.NOP;
+
+ private TreeMap values
+ = new TreeMap<>(Collections.reverseOrder());
+
+ public TimelineMetric() {
+ this(Type.SINGLE_VALUE);
+ }
+
+ public TimelineMetric(Type type) {
+ this.type = type;
+ }
+
+
+ @XmlElement(name = "type")
+ public Type getType() {
+ return type;
+ }
+
+ public void setType(Type metricType) {
+ this.type = metricType;
+ }
+
+ @XmlElement(name = "id")
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String metricId) {
+ this.id = metricId;
+ }
+
+ /**
+ * Get the real time aggregation operation of this metric.
+ *
+ * @return Real time aggregation operation
+ */
+ // required by JAXB
+ @XmlElement(name = "aggregationOp")
+ public TimelineMetricOperation getRealtimeAggregationOp() {
+ return realtimeAggregationOp;
+ }
+
+ /**
+ * Set the real time aggregation operation of this metric.
+ *
+ * @param op A timeline metric operation that the metric should perform on
+ * real time aggregations
+ */
+ public void setRealtimeAggregationOp(
+ final TimelineMetricOperation op) {
+ this.realtimeAggregationOp = op;
+ }
+
+ // required by JAXB
+ @InterfaceAudience.Private
+ @XmlElement(name = "values")
+ public TreeMap getValuesJAXB() {
+ return values;
+ }
+
+ public Map getValues() {
+ return values;
+ }
+
+ public void setValues(Map vals) {
+ if (type == Type.SINGLE_VALUE) {
+ overwrite(vals);
+ } else {
+ if (vals != null) {
+ this.values = new TreeMap<>(Collections.reverseOrder());
+ this.values.putAll(vals);
+ } else {
+ this.values = null;
+ }
+ }
+ }
+
+ public void addValues(Map vals) {
+ if (type == Type.SINGLE_VALUE) {
+ overwrite(vals);
+ } else {
+ this.values.putAll(vals);
+ }
+ }
+
+ public void addValue(long timestamp, Number value) {
+ if (type == Type.SINGLE_VALUE) {
+ values.clear();
+ }
+ values.put(timestamp, value);
+ }
+
+ private void overwrite(Map vals) {
+ if (vals.size() > 1) {
+ throw new IllegalArgumentException(
+ "Values cannot contain more than one point in " +
+ Type.SINGLE_VALUE + " mode");
+ }
+ this.values.clear();
+ this.values.putAll(vals);
+ }
+
+ public boolean isValid() {
+ return (id != null);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = id.hashCode();
+ result = 31 * result + type.hashCode();
+ return result;
+ }
+
+ // Only check if type and id are equal
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof TimelineMetric)) {
+ return false;
+ }
+
+ TimelineMetric m = (TimelineMetric) o;
+
+ if (!id.equals(m.id)) {
+ return false;
+ }
+ if (type != m.type) {
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "{id: " + id + ", type: " + type +
+ ", realtimeAggregationOp: " +
+ realtimeAggregationOp + "; " + values.toString() +
+ "}";
+ }
+
+ /**
+ * Get the latest timeline metric as single value type.
+ *
+ * @param metric Incoming timeline metric
+ * @return The latest metric in the incoming metric
+ */
+ public static TimelineMetric getLatestSingleValueMetric(
+ TimelineMetric metric) {
+ if (metric.getType() == Type.SINGLE_VALUE) {
+ return metric;
+ } else {
+ TimelineMetric singleValueMetric = new TimelineMetric(Type.SINGLE_VALUE);
+ Long firstKey = metric.values.firstKey();
+ if (firstKey != null) {
+ Number firstValue = metric.values.get(firstKey);
+ singleValueMetric.addValue(firstKey, firstValue);
+ }
+ return singleValueMetric;
+ }
+ }
+
+ /**
+ * Get single data timestamp of the metric.
+ *
+ * @return the single data timestamp
+ */
+ public long getSingleDataTimestamp() {
+ if (this.type == Type.SINGLE_VALUE) {
+ if (values.size() == 0) {
+ throw new YarnRuntimeException("Values for this timeline metric is " +
+ "empty.");
+ } else {
+ return values.firstKey();
+ }
+ } else {
+ throw new YarnRuntimeException("Type for this timeline metric is not " +
+ "SINGLE_VALUE.");
+ }
+ }
+
+ /**
+ * Get single data value of the metric.
+ *
+ * @return the single data value
+ */
+ public Number getSingleDataValue() {
+ if (this.type == Type.SINGLE_VALUE) {
+ if (values.size() == 0) {
+ return null;
+ } else {
+ return values.get(values.firstKey());
+ }
+ } else {
+ throw new YarnRuntimeException("Type for this timeline metric is not " +
+ "SINGLE_VALUE.");
+ }
+ }
+
+ /**
+ * Aggregate an incoming metric to the base aggregated metric with the given
+ * operation state in a stateless fashion. The assumption here is
+ * baseAggregatedMetric and latestMetric should be single value data if not
+ * null.
+ *
+ * @param incomingMetric Incoming timeline metric to aggregate
+ * @param baseAggregatedMetric Base timeline metric
+ * @return Result metric after aggregation
+ */
+ public static TimelineMetric aggregateTo(TimelineMetric incomingMetric,
+ TimelineMetric baseAggregatedMetric) {
+ return aggregateTo(incomingMetric, baseAggregatedMetric, null);
+ }
+
+ /**
+ * Aggregate an incoming metric to the base aggregated metric with the given
+ * operation state. The assumption here is baseAggregatedMetric and
+ * latestMetric should be single value data if not null.
+ *
+ * @param incomingMetric Incoming timeline metric to aggregate
+ * @param baseAggregatedMetric Base timeline metric
+ * @param state Operation state
+ * @return Result metric after aggregation
+ */
+ public static TimelineMetric aggregateTo(TimelineMetric incomingMetric,
+ TimelineMetric baseAggregatedMetric, Map state) {
+ TimelineMetricOperation operation
+ = incomingMetric.getRealtimeAggregationOp();
+ return operation.aggregate(incomingMetric, baseAggregatedMetric, state);
+ }
+
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetricCalculator.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetricCalculator.java
new file mode 100644
index 00000000000..4c9045ff044
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetricCalculator.java
@@ -0,0 +1,115 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+
+/**
+ * A calculator for timeline metrics.
+ */
+public final class TimelineMetricCalculator {
+
+ private TimelineMetricCalculator() {
+ // do nothing.
+ }
+
+ /**
+ * Compare two not-null numbers.
+ * @param n1 Number n1
+ * @param n2 Number n2
+ * @return 0 if n1 equals n2, a negative int if n1 is less than n2, a
+ * positive int otherwise.
+ */
+ public static int compare(Number n1, Number n2) {
+ if (n1 == null || n2 == null) {
+ throw new YarnRuntimeException(
+ "Number to be compared shouldn't be null.");
+ }
+
+ if (n1 instanceof Integer || n1 instanceof Long) {
+ if (n1.longValue() == n2.longValue()) {
+ return 0;
+ } else {
+ return (n1.longValue() < n2.longValue()) ? -1 : 1;
+ }
+ }
+
+ if (n1 instanceof Float || n1 instanceof Double) {
+ if (n1.doubleValue() == n2.doubleValue()) {
+ return 0;
+ } else {
+ return (n1.doubleValue() < n2.doubleValue()) ? -1 : 1;
+ }
+ }
+
+ // TODO throw warnings/exceptions for other types of number.
+ throw new YarnRuntimeException("Unsupported types for number comparison: "
+ + n1.getClass().getName() + ", " + n2.getClass().getName());
+ }
+
+ /**
+ * Subtract operation between two Numbers.
+ * @param n1 Number n1
+ * @param n2 Number n2
+ * @return Number represent to (n1 - n2).
+ */
+ public static Number sub(Number n1, Number n2) {
+ if (n1 == null) {
+ throw new YarnRuntimeException(
+ "Number to be subtracted shouldn't be null.");
+ } else if (n2 == null) {
+ return n1;
+ }
+
+ if (n1 instanceof Integer || n1 instanceof Long) {
+ return n1.longValue() - n2.longValue();
+ }
+
+ if (n1 instanceof Float || n1 instanceof Double) {
+ return n1.doubleValue() - n2.doubleValue();
+ }
+
+ // TODO throw warnings/exceptions for other types of number.
+ return null;
+ }
+
+ /**
+ * Sum up two Numbers.
+ * @param n1 Number n1
+ * @param n2 Number n2
+ * @return Number represent to (n1 + n2).
+ */
+ public static Number sum(Number n1, Number n2) {
+ if (n1 == null) {
+ return n2;
+ } else if (n2 == null) {
+ return n1;
+ }
+
+ if (n1 instanceof Integer || n1 instanceof Long) {
+ return n1.longValue() + n2.longValue();
+ }
+
+ if (n1 instanceof Float || n1 instanceof Double) {
+ return n1.doubleValue() + n2.doubleValue();
+ }
+
+ // TODO throw warnings/exceptions for other types of number.
+ return null;
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetricOperation.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetricOperation.java
new file mode 100644
index 00000000000..58e5c3811cf
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineMetricOperation.java
@@ -0,0 +1,167 @@
+/*
+ * 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.api.records.timelineservice;
+
+import java.util.Map;
+
+/**
+ * Aggregation operations.
+ */
+public enum TimelineMetricOperation {
+ NOP("NOP") {
+ /**
+ * Do nothing on the base metric.
+ *
+ * @param incoming Metric a
+ * @param base Metric b
+ * @param state Operation state (not used)
+ * @return Metric b
+ */
+ @Override
+ public TimelineMetric exec(TimelineMetric incoming,
+ TimelineMetric base, Map state) {
+ return base;
+ }
+ },
+ MAX("MAX") {
+ /**
+ * Keep the greater value of incoming and base. Stateless operation.
+ *
+ * @param incoming Metric a
+ * @param base Metric b
+ * @param state Operation state (not used)
+ * @return the greater value of a and b
+ */
+ @Override
+ public TimelineMetric exec(TimelineMetric incoming,
+ TimelineMetric base, Map state) {
+ if (base == null) {
+ return incoming;
+ }
+ Number incomingValue = incoming.getSingleDataValue();
+ Number aggregateValue = base.getSingleDataValue();
+ if (aggregateValue == null) {
+ aggregateValue = Long.MIN_VALUE;
+ }
+ if (TimelineMetricCalculator.compare(incomingValue, aggregateValue) > 0) {
+ base.addValue(incoming.getSingleDataTimestamp(), incomingValue);
+ }
+ return base;
+ }
+ },
+ REPLACE("REPLACE") {
+ /**
+ * Replace the base metric with the incoming value. Stateless operation.
+ *
+ * @param incoming Metric a
+ * @param base Metric b
+ * @param state Operation state (not used)
+ * @return Metric a
+ */
+ @Override
+ public TimelineMetric exec(TimelineMetric incoming,
+ TimelineMetric base,
+ Map state) {
+ return incoming;
+ }
+ },
+ SUM("SUM") {
+ /**
+ * Return the sum of the incoming metric and the base metric if the
+ * operation is stateless. For stateful operations, also subtract the
+ * value of the timeline metric mapped to the PREV_METRIC_STATE_KEY
+ * in the state object.
+ *
+ * @param incoming Metric a
+ * @param base Metric b
+ * @param state Operation state (PREV_METRIC_STATE_KEY's value as Metric p)
+ * @return A metric with value a + b - p
+ */
+ @Override
+ public TimelineMetric exec(TimelineMetric incoming, TimelineMetric base,
+ Map state) {
+ if (base == null) {
+ return incoming;
+ }
+ Number incomingValue = incoming.getSingleDataValue();
+ Number aggregateValue = base.getSingleDataValue();
+ Number result
+ = TimelineMetricCalculator.sum(incomingValue, aggregateValue);
+
+ // If there are previous value in the state, we will take it off from the
+ // sum
+ if (state != null) {
+ Object prevMetric = state.get(PREV_METRIC_STATE_KEY);
+ if (prevMetric instanceof TimelineMetric) {
+ result = TimelineMetricCalculator.sub(result,
+ ((TimelineMetric) prevMetric).getSingleDataValue());
+ }
+ }
+ base.addValue(incoming.getSingleDataTimestamp(), result);
+ return base;
+ }
+ },
+ AVG("AVERAGE") {
+ /**
+ * Return the average value of the incoming metric and the base metric,
+ * with a given state. Not supported yet.
+ *
+ * @param incoming Metric a
+ * @param base Metric b
+ * @param state Operation state
+ * @return Not finished yet
+ */
+ @Override
+ public TimelineMetric exec(TimelineMetric incoming, TimelineMetric base,
+ Map state) {
+ // Not supported yet
+ throw new UnsupportedOperationException(
+ "Unsupported aggregation operation: AVERAGE");
+ }
+ };
+
+ public static final String PREV_METRIC_STATE_KEY = "PREV_METRIC";
+
+ /**
+ * Perform the aggregation operation.
+ *
+ * @param incoming Incoming metric
+ * @param aggregate Base aggregation metric
+ * @param state Operation state
+ * @return Result metric for this aggregation operation
+ */
+ public TimelineMetric aggregate(TimelineMetric incoming,
+ TimelineMetric aggregate, Map state) {
+ return exec(incoming, aggregate, state);
+ }
+
+ private final String opName;
+
+ TimelineMetricOperation(String opString) {
+ opName = opString;
+ }
+
+ @Override
+ public String toString() {
+ return this.opName;
+ }
+
+ abstract TimelineMetric exec(TimelineMetric incoming, TimelineMetric base,
+ Map state);
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java
new file mode 100644
index 00000000000..eda1ee2ef7e
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/TimelineWriteResponse.java
@@ -0,0 +1,167 @@
+/**
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience.Public;
+import org.apache.hadoop.classification.InterfaceStability.Unstable;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlRootElement;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * A class that holds a list of put errors. This is the response returned when a
+ * list of {@link TimelineEntity} objects is added to the timeline. If there are
+ * errors in storing individual entity objects, they will be indicated in the
+ * list of errors.
+ */
+@XmlRootElement(name = "response")
+@XmlAccessorType(XmlAccessType.NONE)
+@Public
+@Unstable
+public class TimelineWriteResponse {
+
+ private List errors = new ArrayList();
+
+ public TimelineWriteResponse() {
+
+ }
+
+ /**
+ * Get a list of {@link TimelineWriteError} instances.
+ *
+ * @return a list of {@link TimelineWriteError} instances
+ */
+ @XmlElement(name = "errors")
+ public List getErrors() {
+ return errors;
+ }
+
+ /**
+ * Add a single {@link TimelineWriteError} instance into the existing list.
+ *
+ * @param error
+ * a single {@link TimelineWriteError} instance
+ */
+ public void addError(TimelineWriteError error) {
+ errors.add(error);
+ }
+
+ /**
+ * Add a list of {@link TimelineWriteError} instances into the existing list.
+ *
+ * @param writeErrors
+ * a list of {@link TimelineWriteError} instances
+ */
+ public void addErrors(List writeErrors) {
+ this.errors.addAll(writeErrors);
+ }
+
+ /**
+ * Set the list to the given list of {@link TimelineWriteError} instances.
+ *
+ * @param writeErrors
+ * a list of {@link TimelineWriteError} instances
+ */
+ public void setErrors(List writeErrors) {
+ this.errors.clear();
+ this.errors.addAll(writeErrors);
+ }
+
+ /**
+ * A class that holds the error code for one entity.
+ */
+ @XmlRootElement(name = "error")
+ @XmlAccessorType(XmlAccessType.NONE)
+ @Public
+ @Unstable
+ public static class TimelineWriteError {
+
+ /**
+ * Error code returned if an IOException is encountered when storing an
+ * entity.
+ */
+ public static final int IO_EXCEPTION = 1;
+
+ private String entityId;
+ private String entityType;
+ private int errorCode;
+
+ /**
+ * Get the entity Id.
+ *
+ * @return the entity Id
+ */
+ @XmlElement(name = "entity")
+ public String getEntityId() {
+ return entityId;
+ }
+
+ /**
+ * Set the entity Id.
+ *
+ * @param id the entity Id.
+ */
+ public void setEntityId(String id) {
+ this.entityId = id;
+ }
+
+ /**
+ * Get the entity type.
+ *
+ * @return the entity type
+ */
+ @XmlElement(name = "entitytype")
+ public String getEntityType() {
+ return entityType;
+ }
+
+ /**
+ * Set the entity type.
+ *
+ * @param type the entity type.
+ */
+ public void setEntityType(String type) {
+ this.entityType = type;
+ }
+
+ /**
+ * Get the error code.
+ *
+ * @return an error code
+ */
+ @XmlElement(name = "errorcode")
+ public int getErrorCode() {
+ return errorCode;
+ }
+
+ /**
+ * Set the error code to the given error code.
+ *
+ * @param code an error code.
+ */
+ public void setErrorCode(int code) {
+ this.errorCode = code;
+ }
+
+ }
+
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/UserEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/UserEntity.java
new file mode 100644
index 00000000000..ced57c67cd6
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/UserEntity.java
@@ -0,0 +1,40 @@
+/*
+ * 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.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * This entity represents a user.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class UserEntity extends TimelineEntity {
+ public UserEntity() {
+ super(TimelineEntityType.YARN_USER.toString());
+ }
+
+ public UserEntity(TimelineEntity entity) {
+ super(entity);
+ if (!entity.getType().equals(TimelineEntityType.YARN_USER.toString())) {
+ throw new IllegalArgumentException("Incompatible entity type: "
+ + getId());
+ }
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/package-info.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/package-info.java
new file mode 100644
index 00000000000..43805c8c98a
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/package-info.java
@@ -0,0 +1,26 @@
+/*
+ * 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.api.records.timelineservice contains classes
+ * which define the data model for ATSv2.
+ */
+@InterfaceAudience.Public
+package org.apache.hadoop.yarn.api.records.timelineservice;
+
+import org.apache.hadoop.classification.InterfaceAudience;
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
index 817ada1c57e..fb9fb534fff 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java
@@ -96,6 +96,10 @@ public class YarnConfiguration extends Configuration {
new DeprecationDelta(RM_ZK_RETRY_INTERVAL_MS,
CommonConfigurationKeys.ZK_RETRY_INTERVAL_MS),
});
+ Configuration.addDeprecations(new DeprecationDelta[] {
+ new DeprecationDelta(RM_SYSTEM_METRICS_PUBLISHER_ENABLED,
+ SYSTEM_METRICS_PUBLISHER_ENABLED)
+ });
}
//Configurations
@@ -146,6 +150,7 @@ public class YarnConfiguration extends Configuration {
public static final String RM_PREFIX = "yarn.resourcemanager.";
public static final String RM_CLUSTER_ID = RM_PREFIX + "cluster-id";
+ public static final String DEFAULT_RM_CLUSTER_ID = "yarn_cluster";
public static final String RM_HOSTNAME = RM_PREFIX + "hostname";
@@ -500,16 +505,37 @@ public class YarnConfiguration extends Configuration {
/**
* The setting that controls whether yarn system metrics is published on the
- * timeline server or not by RM.
+ * timeline server or not by RM. This configuration setting is for ATS V1.
+ * This is now deprecated in favor of SYSTEM_METRICS_PUBLISHER_ENABLED.
*/
- public static final String RM_SYSTEM_METRICS_PUBLISHER_ENABLED =
- RM_PREFIX + "system-metrics-publisher.enabled";
- public static final boolean DEFAULT_RM_SYSTEM_METRICS_PUBLISHER_ENABLED = false;
+ public static final String RM_SYSTEM_METRICS_PUBLISHER_ENABLED = RM_PREFIX
+ + "system-metrics-publisher.enabled";
+ public static final boolean DEFAULT_RM_SYSTEM_METRICS_PUBLISHER_ENABLED =
+ false;
+
+ /**
+ * The setting that controls whether yarn system metrics is published on the
+ * timeline server or not by RM and NM. This configuration setting is for
+ * ATS v2.
+ */
+ public static final String SYSTEM_METRICS_PUBLISHER_ENABLED = YARN_PREFIX
+ + "system-metrics-publisher.enabled";
+ public static final boolean DEFAULT_SYSTEM_METRICS_PUBLISHER_ENABLED = false;
+
+ /**
+ * The setting that controls whether yarn container events are published to
+ * the timeline service or not by RM. This configuration setting is for ATS
+ * V2
+ */
+ public static final String RM_PUBLISH_CONTAINER_EVENTS_ENABLED = YARN_PREFIX
+ + "rm.system-metrics-publisher.emit-container-events";
+ public static final boolean DEFAULT_RM_PUBLISH_CONTAINER_EVENTS_ENABLED =
+ false;
public static final String RM_SYSTEM_METRICS_PUBLISHER_DISPATCHER_POOL_SIZE =
RM_PREFIX + "system-metrics-publisher.dispatcher.pool-size";
- public static final int DEFAULT_RM_SYSTEM_METRICS_PUBLISHER_DISPATCHER_POOL_SIZE =
- 10;
+ public static final int
+ DEFAULT_RM_SYSTEM_METRICS_PUBLISHER_DISPATCHER_POOL_SIZE = 10;
//RM delegation token related keys
public static final String RM_DELEGATION_KEY_UPDATE_INTERVAL_KEY =
@@ -956,6 +982,11 @@ public class YarnConfiguration extends Configuration {
NM_PREFIX + "container-manager.thread-count";
public static final int DEFAULT_NM_CONTAINER_MGR_THREAD_COUNT = 20;
+ /** Number of threads container manager uses.*/
+ public static final String NM_COLLECTOR_SERVICE_THREAD_COUNT =
+ NM_PREFIX + "collector-service.thread-count";
+ public static final int DEFAULT_NM_COLLECTOR_SERVICE_THREAD_COUNT = 5;
+
/** Number of threads used in cleanup.*/
public static final String NM_DELETE_THREAD_COUNT =
NM_PREFIX + "delete.thread-count";
@@ -983,6 +1014,13 @@ public class YarnConfiguration extends Configuration {
public static final String DEFAULT_NM_LOCALIZER_ADDRESS = "0.0.0.0:" +
DEFAULT_NM_LOCALIZER_PORT;
+ /** Address where the collector service IPC is.*/
+ public static final String NM_COLLECTOR_SERVICE_ADDRESS =
+ NM_PREFIX + "collector-service.address";
+ public static final int DEFAULT_NM_COLLECTOR_SERVICE_PORT = 8048;
+ public static final String DEFAULT_NM_COLLECTOR_SERVICE_ADDRESS =
+ "0.0.0.0:" + DEFAULT_NM_LOCALIZER_PORT;
+
/** Interval in between cache cleanups.*/
public static final String NM_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS =
NM_PREFIX + "localizer.cache.cleanup.interval-ms";
@@ -1831,6 +1869,10 @@ public class YarnConfiguration extends Configuration {
YARN_SECURITY_SERVICE_AUTHORIZATION_APPLICATIONHISTORY_PROTOCOL =
"security.applicationhistory.protocol.acl";
+ public static final String
+ YARN_SECURITY_SERVICE_AUTHORIZATION_COLLECTOR_NODEMANAGER_PROTOCOL =
+ "security.collector-nodemanager.protocol.acl";
+
/** No. of milliseconds to wait between sending a SIGTERM and SIGKILL
* to a running container */
public static final String NM_SLEEP_DELAY_BEFORE_SIGKILL_MS =
@@ -2011,7 +2053,7 @@ public class YarnConfiguration extends Configuration {
public static final String TIMELINE_SERVICE_UI_WEB_PATH_PREFIX =
TIMELINE_SERVICE_PREFIX + "ui-web-path.";
- /** Timeline client settings */
+ /** Timeline client settings. */
public static final String TIMELINE_SERVICE_CLIENT_PREFIX =
TIMELINE_SERVICE_PREFIX + "client.";
@@ -2150,6 +2192,138 @@ public class YarnConfiguration extends Configuration {
= TIMELINE_SERVICE_PREFIX
+ "entity-file.fs-support-append";
+ /**
+ * Settings for timeline service v2.0.
+ */
+ public static final String TIMELINE_SERVICE_WRITER_CLASS =
+ TIMELINE_SERVICE_PREFIX + "writer.class";
+
+ public static final String DEFAULT_TIMELINE_SERVICE_WRITER_CLASS =
+ "org.apache.hadoop.yarn.server.timelineservice"
+ + ".storage.HBaseTimelineWriterImpl";
+
+ public static final String TIMELINE_SERVICE_READER_CLASS =
+ TIMELINE_SERVICE_PREFIX + "reader.class";
+
+ public static final String DEFAULT_TIMELINE_SERVICE_READER_CLASS =
+ "org.apache.hadoop.yarn.server.timelineservice" +
+ ".storage.HBaseTimelineReaderImpl";
+
+
+ /**
+ * default schema prefix for hbase tables.
+ */
+ public static final String DEFAULT_TIMELINE_SERVICE_HBASE_SCHEMA_PREFIX =
+ "prod.";
+
+ /**
+ * config param name to override schema prefix.
+ */
+ public static final String TIMELINE_SERVICE_HBASE_SCHEMA_PREFIX_NAME =
+ TIMELINE_SERVICE_PREFIX + "hbase-schema.prefix";
+
+ /** The setting that controls how often the timeline collector flushes the
+ * timeline writer.
+ */
+ public static final String TIMELINE_SERVICE_WRITER_FLUSH_INTERVAL_SECONDS =
+ TIMELINE_SERVICE_PREFIX + "writer.flush-interval-seconds";
+
+ public static final int
+ DEFAULT_TIMELINE_SERVICE_WRITER_FLUSH_INTERVAL_SECONDS = 60;
+
+ /**
+ * The name for setting that controls how long the final value of
+ * a metric of a completed app is retained before merging
+ * into the flow sum.
+ */
+ public static final String APP_FINAL_VALUE_RETENTION_THRESHOLD =
+ TIMELINE_SERVICE_PREFIX
+ + "hbase.coprocessor.app-final-value-retention-milliseconds";
+
+ /**
+ * The name of the setting for the location of the coprocessor
+ * jar on hdfs.
+ */
+ public static final String FLOW_RUN_COPROCESSOR_JAR_HDFS_LOCATION =
+ TIMELINE_SERVICE_PREFIX
+ + "hbase.coprocessor.jar.hdfs.location";
+
+ /** default hdfs location for flowrun coprocessor jar. */
+ public static final String DEFAULT_HDFS_LOCATION_FLOW_RUN_COPROCESSOR_JAR =
+ "/hbase/coprocessor/hadoop-yarn-server-timelineservice.jar";
+
+ /**
+ * The name for setting that points to an optional HBase configuration
+ * (hbase-site.xml file) with settings that will override the ones found on
+ * the classpath.
+ */
+ public static final String TIMELINE_SERVICE_HBASE_CONFIGURATION_FILE =
+ TIMELINE_SERVICE_PREFIX
+ + "hbase.configuration.file";
+
+ /**
+ * The name for setting that enables or disables authentication checks
+ * for reading timeline service v2 data.
+ */
+ public static final String TIMELINE_SERVICE_READ_AUTH_ENABLED =
+ TIMELINE_SERVICE_PREFIX + "read.authentication.enabled";
+
+ /**
+ * The default setting for authentication checks for reading timeline
+ * service v2 data.
+ */
+ public static final Boolean DEFAULT_TIMELINE_SERVICE_READ_AUTH_ENABLED =
+ false;
+
+ /**
+ * The name for setting that lists the users and groups who are allowed
+ * to read timeline service v2 data. It is a comma separated list of
+ * user, followed by space, then comma separated list of groups.
+ * It will allow this list of users and groups to read the data
+ * and reject everyone else.
+ */
+ public static final String TIMELINE_SERVICE_READ_ALLOWED_USERS =
+ TIMELINE_SERVICE_PREFIX + "read.allowed.users";
+
+ /**
+ * The default value for list of the users who are allowed to read
+ * timeline service v2 data.
+ */
+ public static final String DEFAULT_TIMELINE_SERVICE_READ_ALLOWED_USERS =
+ "";
+
+ /**
+ * The setting that controls how long the final value of a metric of a
+ * completed app is retained before merging into the flow sum. Up to this time
+ * after an application is completed out-of-order values that arrive can be
+ * recognized and discarded at the cost of increased storage.
+ */
+ public static final long DEFAULT_APP_FINAL_VALUE_RETENTION_THRESHOLD = 3 * 24
+ * 60 * 60 * 1000L;
+
+ public static final String ATS_APP_COLLECTOR_LINGER_PERIOD_IN_MS =
+ TIMELINE_SERVICE_PREFIX + "app-collector.linger-period.ms";
+
+ public static final int DEFAULT_ATS_APP_COLLECTOR_LINGER_PERIOD_IN_MS = 1000;
+
+ public static final String NUMBER_OF_ASYNC_ENTITIES_TO_MERGE =
+ TIMELINE_SERVICE_PREFIX
+ + "timeline-client.number-of-async-entities-to-merge";
+
+ public static final int DEFAULT_NUMBER_OF_ASYNC_ENTITIES_TO_MERGE = 10;
+
+ /** default version for any flow. */
+ public static final String DEFAULT_FLOW_VERSION = "1";
+
+ /**
+ * The time period for which timeline v2 client will wait for draining
+ * leftover entities after stop.
+ */
+ public static final String TIMELINE_V2_CLIENT_DRAIN_TIME_MILLIS =
+ TIMELINE_SERVICE_CLIENT_PREFIX + "drain-entities.timeout.ms";
+ public static final long DEFAULT_TIMELINE_V2_CLIENT_DRAIN_TIME_MILLIS
+ = 2000L;
+
// mark app-history related configs @Private as application history is going
// to be integrated into the timeline service
@Private
@@ -2208,6 +2382,7 @@ public class YarnConfiguration extends Configuration {
/** The listening endpoint for the timeline service application.*/
public static final String TIMELINE_SERVICE_BIND_HOST =
TIMELINE_SERVICE_PREFIX + "bind-host";
+ public static final String DEFAULT_TIMELINE_SERVICE_BIND_HOST = "0.0.0.0";
/** The number of threads to handle client RPC API requests. */
public static final String TIMELINE_SERVICE_HANDLER_THREAD_COUNT =
@@ -2407,6 +2582,16 @@ public class YarnConfiguration extends Configuration {
public static final long DEFAULT_TIMELINE_DELEGATION_TOKEN_MAX_LIFETIME =
7*24*60*60*1000; // 7 days
+ // Timeline service v2 offlien aggregation related keys
+ public static final String TIMELINE_OFFLINE_AGGREGATION_PREFIX =
+ YarnConfiguration.TIMELINE_SERVICE_PREFIX + "aggregation.offline.";
+ public static final String PHOENIX_OFFLINE_STORAGE_CONN_STR
+ = TIMELINE_OFFLINE_AGGREGATION_PREFIX
+ + "phoenix.connectionString";
+
+ public static final String PHOENIX_OFFLINE_STORAGE_CONN_STR_DEFAULT
+ = "jdbc:phoenix:localhost:2181:/hbase";
+
// ///////////////////////////////
// Shared Cache Configs
// ///////////////////////////////
@@ -3148,6 +3333,53 @@ public class YarnConfiguration extends Configuration {
YarnConfiguration.DEFAULT_OPPORTUNISTIC_CONTAINER_ALLOCATION_ENABLED);
}
+ // helper methods for timeline service configuration
+ /**
+ * Returns whether the timeline service is enabled via configuration.
+ *
+ * @param conf the configuration
+ * @return whether the timeline service is enabled.
+ */
+ public static boolean timelineServiceEnabled(Configuration conf) {
+ return conf.getBoolean(YarnConfiguration.TIMELINE_SERVICE_ENABLED,
+ YarnConfiguration.DEFAULT_TIMELINE_SERVICE_ENABLED);
+ }
+
+ /**
+ * Returns the timeline service version. It does not check whether the
+ * timeline service itself is enabled.
+ *
+ * @param conf the configuration
+ * @return the timeline service version as a float.
+ */
+ public static float getTimelineServiceVersion(Configuration conf) {
+ return conf.getFloat(TIMELINE_SERVICE_VERSION,
+ DEFAULT_TIMELINE_SERVICE_VERSION);
+ }
+
+ /**
+ * Returns whether the timeline service v.2 is enabled via configuration.
+ *
+ * @param conf the configuration
+ * @return whether the timeline service v.2 is enabled. V.2 refers to a
+ * version greater than equal to 2 but smaller than 3.
+ */
+ public static boolean timelineServiceV2Enabled(Configuration conf) {
+ return timelineServiceEnabled(conf) &&
+ (int)getTimelineServiceVersion(conf) == 2;
+ }
+
+ /**
+ * Returns whether the system publisher is enabled.
+ *
+ * @param conf the configuration
+ * @return whether the system publisher is enabled.
+ */
+ public static boolean systemMetricsPublisherEnabled(Configuration conf) {
+ return conf.getBoolean(YarnConfiguration.SYSTEM_METRICS_PUBLISHER_ENABLED,
+ YarnConfiguration.DEFAULT_SYSTEM_METRICS_PUBLISHER_ENABLED);
+ }
+
/* For debugging. mp configurations to system output as XML format. */
public static void main(String[] args) throws Exception {
new YarnConfiguration(new Configuration()).writeXml(System.out);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/util/TimelineServiceHelper.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/util/TimelineServiceHelper.java
new file mode 100644
index 00000000000..65ed18a7a1e
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/util/TimelineServiceHelper.java
@@ -0,0 +1,57 @@
+/*
+ * 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.util;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
+
+/**
+ * Helper class for Timeline service.
+ */
+@LimitedPrivate({ "MapReduce", "YARN" })
+public final class TimelineServiceHelper {
+
+ private TimelineServiceHelper() {
+ // Utility classes should not have a public or default constructor.
+ }
+
+ /**
+ * Cast map to HashMap for generic type.
+ * @param originalMap the map need to be casted
+ * @param key type
+ * @param value type
+ * @return casted HashMap object
+ */
+ public static HashMap mapCastToHashMap(
+ Map originalMap) {
+ return originalMap == null ? null : originalMap instanceof HashMap ?
+ (HashMap) originalMap : new HashMap(originalMap);
+ }
+
+ /**
+ * Inverts the given key.
+ * @param key value to be inverted .
+ * @return inverted long
+ */
+ public static long invertLong(long key) {
+ return Long.MAX_VALUE - key;
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
index a3ad1e32c53..07b8335d0aa 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_protos.proto
@@ -667,3 +667,8 @@ message ContainerResourceDecreaseProto {
optional ContainerIdProto container_id = 1;
optional ResourceProto capability = 2;
}
+
+message CollectorInfoProto {
+ optional string collector_addr = 1;
+ optional hadoop.common.TokenProto collector_token = 2;
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
index b87c6f0592a..8301971f6d2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/proto/yarn_service_protos.proto
@@ -115,6 +115,7 @@ message AllocateResponseProto {
repeated ContainerResourceDecreaseProto decreased_containers = 11;
optional hadoop.common.TokenProto am_rm_token = 12;
optional PriorityProto application_priority = 13;
+ optional CollectorInfoProto collector_info = 14;
repeated UpdateContainerErrorProto update_errors = 15;
repeated UpdatedContainerProto updated_containers = 16;
}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/api/records/timelineservice/TestApplicationEntity.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/api/records/timelineservice/TestApplicationEntity.java
new file mode 100644
index 00000000000..c3f277718e7
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/api/records/timelineservice/TestApplicationEntity.java
@@ -0,0 +1,71 @@
+/**
+ * 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.api.records.timelineservice;
+
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+
+/**
+ * Various tests for the ApplicationEntity class.
+ *
+ */
+public class TestApplicationEntity {
+
+ @Test
+ public void testIsApplicationEntity() {
+ TimelineEntity te = new TimelineEntity();
+ te.setType(TimelineEntityType.YARN_APPLICATION.toString());
+ assertTrue(ApplicationEntity.isApplicationEntity(te));
+
+ te = null;
+ assertEquals(false, ApplicationEntity.isApplicationEntity(te));
+
+ te = new TimelineEntity();
+ te.setType(TimelineEntityType.YARN_CLUSTER.toString());
+ assertEquals(false, ApplicationEntity.isApplicationEntity(te));
+ }
+
+ @Test
+ public void testGetApplicationEvent() {
+ TimelineEntity te = null;
+ TimelineEvent tEvent = ApplicationEntity.getApplicationEvent(te,
+ "no event");
+ assertEquals(null, tEvent);
+
+ te = new TimelineEntity();
+ te.setType(TimelineEntityType.YARN_APPLICATION.toString());
+ TimelineEvent event = new TimelineEvent();
+ event.setId("start_event");
+ event.setTimestamp(System.currentTimeMillis());
+ te.addEvent(event);
+ tEvent = ApplicationEntity.getApplicationEvent(te, "start_event");
+ assertEquals(event, tEvent);
+
+ te = new TimelineEntity();
+ te.setType(TimelineEntityType.YARN_CLUSTER.toString());
+ event = new TimelineEvent();
+ event.setId("start_event_cluster");
+ event.setTimestamp(System.currentTimeMillis());
+ te.addEvent(event);
+ tEvent = ApplicationEntity.getApplicationEvent(te, "start_event_cluster");
+ assertEquals(null, tEvent);
+
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/api/records/timelineservice/TestTimelineMetric.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/api/records/timelineservice/TestTimelineMetric.java
new file mode 100644
index 00000000000..a6a3fdaf2a0
--- /dev/null
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/api/records/timelineservice/TestTimelineMetric.java
@@ -0,0 +1,92 @@
+/**
+ * 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.api.records.timelineservice;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.yarn.api.records.timelineservice.TimelineMetric.Type;
+
+import org.junit.Test;
+
+/**
+ * Tests {@link TimelineMetric} operations such as aggregation.
+ */
+public class TestTimelineMetric {
+
+ @Test
+ public void testTimelineMetricAggregation() {
+ long ts = System.currentTimeMillis();
+ // single_value metric add against null metric
+ TimelineMetric m1 = getSingleValueMetric("MEGA_BYTES_MILLIS",
+ TimelineMetricOperation.SUM, ts, 10000L);
+ TimelineMetric aggregatedMetric = TimelineMetric.aggregateTo(m1, null);
+ assertEquals(10000L, aggregatedMetric.getSingleDataValue());
+
+ TimelineMetric m2 = getSingleValueMetric("MEGA_BYTES_MILLIS",
+ TimelineMetricOperation.SUM, ts, 20000L);
+ aggregatedMetric = TimelineMetric.aggregateTo(m2, aggregatedMetric);
+ assertEquals(30000L, aggregatedMetric.getSingleDataValue());
+
+ // stateful sum test
+ Map state = new HashMap<>();
+ state.put(TimelineMetricOperation.PREV_METRIC_STATE_KEY, m2);
+ TimelineMetric m2New = getSingleValueMetric("MEGA_BYTES_MILLIS",
+ TimelineMetricOperation.SUM, ts, 10000L);
+ aggregatedMetric = TimelineMetric.aggregateTo(m2New, aggregatedMetric,
+ state);
+ assertEquals(20000L, aggregatedMetric.getSingleDataValue());
+
+ // single_value metric max against single_value metric
+ TimelineMetric m3 = getSingleValueMetric("TRANSFER_RATE",
+ TimelineMetricOperation.MAX, ts, 150L);
+ TimelineMetric aggregatedMax = TimelineMetric.aggregateTo(m3, null);
+ assertEquals(150L, aggregatedMax.getSingleDataValue());
+
+ TimelineMetric m4 = getSingleValueMetric("TRANSFER_RATE",
+ TimelineMetricOperation.MAX, ts, 170L);
+ aggregatedMax = TimelineMetric.aggregateTo(m4, aggregatedMax);
+ assertEquals(170L, aggregatedMax.getSingleDataValue());
+
+ // single_value metric avg against single_value metric
+ TimelineMetric m5 = getSingleValueMetric("TRANSFER_RATE",
+ TimelineMetricOperation.AVG, ts, 150L);
+ try {
+ TimelineMetric.aggregateTo(m5, null);
+ fail("Taking average among metrics is not supported! ");
+ } catch (UnsupportedOperationException e) {
+ // Expected
+ }
+
+ }
+
+ private static TimelineMetric getSingleValueMetric(String id,
+ TimelineMetricOperation op, long timestamp, long value) {
+ TimelineMetric m = new TimelineMetric();
+ m.setId(id);
+ m.setType(Type.SINGLE_VALUE);
+ m.setRealtimeAggregationOp(op);
+ Map metricValues = new HashMap();
+ metricValues.put(timestamp, value);
+ m.setValues(metricValues);
+ return m;
+ }
+}
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/conf/TestYarnConfigurationFields.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/conf/TestYarnConfigurationFields.java
index 6cb92f03829..1d3111ce8b0 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/conf/TestYarnConfigurationFields.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/test/java/org/apache/hadoop/yarn/conf/TestYarnConfigurationFields.java
@@ -66,6 +66,8 @@ public class TestYarnConfigurationFields extends TestConfigurationFieldsBase {
configurationPropsToSkipCompare
.add(YarnConfiguration
.YARN_SECURITY_SERVICE_AUTHORIZATION_RESOURCETRACKER_PROTOCOL);
+ configurationPropsToSkipCompare.add(YarnConfiguration
+ .YARN_SECURITY_SERVICE_AUTHORIZATION_COLLECTOR_NODEMANAGER_PROTOCOL);
configurationPropsToSkipCompare.add(YarnConfiguration.CURATOR_LEADER_ELECTOR);
configurationPropsToSkipCompare
.add(YarnConfiguration.RM_RESERVATION_SYSTEM_MAX_PERIODICITY);
@@ -125,6 +127,9 @@ public class TestYarnConfigurationFields extends TestConfigurationFieldsBase {
// Ignore all YARN Application Timeline Service (version 1) properties
configurationPrefixToSkipCompare.add("yarn.timeline-service.");
+ // skip deprecated RM_SYSTEM_METRICS_PUBLISHER_ENABLED
+ configurationPropsToSkipCompare
+ .add(YarnConfiguration.RM_SYSTEM_METRICS_PUBLISHER_ENABLED);
// skip deprecated ZooKeeper settings
configurationPropsToSkipCompare.add(YarnConfiguration.RM_ZK_ADDRESS);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
index 2c1e36dc1d5..9dde71d98c2 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-applications-distributedshell/pom.xml
@@ -70,6 +70,16 @@
commons-io
commons-io