YARN-4179. [reader implementation] support flow activity queries based on time (Varun Saxena via sjlee)
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/c0910861 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/c0910861 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/c0910861 Branch: refs/heads/YARN-2928-rebase Commit: c09108617689c58a2658794c84714b14613e6d41 Parents: 7f10622 Author: Sangjin Lee <sj...@apache.org> Authored: Thu Oct 22 17:41:40 2015 -0700 Committer: Sangjin Lee <sj...@apache.org> Committed: Mon Nov 9 16:13:16 2015 -0800 ---------------------------------------------------------------------- hadoop-yarn-project/CHANGES.txt | 3 + .../timelineservice/FlowActivityEntity.java | 10 +- .../reader/TimelineReaderWebServices.java | 134 ++++++++++++++++++- .../storage/FlowActivityEntityReader.java | 17 ++- .../storage/flow/FlowActivityRowKey.java | 21 +++ ...stTimelineReaderWebServicesHBaseStorage.java | 59 ++++++++ 6 files changed, 235 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/c0910861/hadoop-yarn-project/CHANGES.txt ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/CHANGES.txt b/hadoop-yarn-project/CHANGES.txt index d9e0968..7959609 100644 --- a/hadoop-yarn-project/CHANGES.txt +++ b/hadoop-yarn-project/CHANGES.txt @@ -121,6 +121,9 @@ Branch YARN-2928: Timeline Server Next Generation: Phase 1 YARN-3864. Implement support for querying single app and all apps for a flow run (Varun Saxena via sjlee) + YARN-4179. [reader implementation] support flow activity queries based on + time (Varun Saxena via sjlee) + IMPROVEMENTS YARN-3276. Code cleanup for timeline service API records. (Junping Du via http://git-wip-us.apache.org/repos/asf/hadoop/blob/c0910861/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/timelineservice/FlowActivityEntity.java ---------------------------------------------------------------------- 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 index 163bd5c..cf19328 100644 --- 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 @@ -140,7 +140,15 @@ public class FlowActivityEntity extends TimelineEntity { } public Date getDate() { - return (Date)getInfo().get(DATE_INFO_KEY); + 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) { http://git-wip-us.apache.org/repos/asf/hadoop/blob/c0910861/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/reader/TimelineReaderWebServices.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/reader/TimelineReaderWebServices.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/reader/TimelineReaderWebServices.java index 83062f3..d82a402 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/reader/TimelineReaderWebServices.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/reader/TimelineReaderWebServices.java @@ -19,12 +19,18 @@ package org.apache.hadoop.yarn.server.timelineservice.reader; import java.io.IOException; +import java.text.DateFormat; +import java.text.ParseException; +import java.text.SimpleDateFormat; import java.util.Collections; +import java.util.Date; import java.util.EnumSet; import java.util.HashMap; import java.util.HashSet; +import java.util.Locale; import java.util.Map; import java.util.Set; +import java.util.TimeZone; import javax.servlet.ServletContext; import javax.servlet.http.HttpServletRequest; @@ -54,6 +60,7 @@ import org.apache.hadoop.yarn.util.timeline.TimelineUtils; import org.apache.hadoop.yarn.webapp.BadRequestException; import org.apache.hadoop.yarn.webapp.NotFoundException; +import com.google.common.annotations.VisibleForTesting; import com.google.inject.Singleton; /** REST end point for Timeline Reader */ @@ -70,11 +77,96 @@ public class TimelineReaderWebServices { private static final String COMMA_DELIMITER = ","; private static final String COLON_DELIMITER = ":"; private static final String QUERY_STRING_SEP = "?"; + private static final String RANGE_DELIMITER = "-"; + private static final String DATE_PATTERN = "yyyyMMdd"; + + @VisibleForTesting + static ThreadLocal<DateFormat> DATE_FORMAT = new ThreadLocal<DateFormat>() { + @Override + protected DateFormat initialValue() { + SimpleDateFormat format = + new SimpleDateFormat(DATE_PATTERN, Locale.ENGLISH); + format.setTimeZone(TimeZone.getTimeZone("GMT")); + format.setLenient(false); + return format; + } + }; private void init(HttpServletResponse response) { response.setContentType(null); } + private static class DateRange { + Long dateStart; + Long dateEnd; + private DateRange(Long start, Long end) { + this.dateStart = start; + this.dateEnd = end; + } + } + + private static long parseDate(String strDate) throws ParseException { + Date date = DATE_FORMAT.get().parse(strDate); + return date.getTime(); + } + + /** + * Parses date range which can be a single date or in the format + * "[startdate]-[enddate]" where either of start or end date may not exist. + * @param dateRange + * @return a {@link DateRange} object. + * @throws IllegalArgumentException + */ + private static DateRange parseDateRange(String dateRange) + throws IllegalArgumentException { + if (dateRange == null || dateRange.isEmpty()) { + return new DateRange(null, null); + } + // Split date range around "-" fetching two components indicating start and + // end date. + String[] dates = dateRange.split(RANGE_DELIMITER, 2); + Long start = null; + Long end = null; + try { + String startDate = dates[0].trim(); + if (!startDate.isEmpty()) { + // Start date is not in yyyyMMdd format. + if (startDate.length() != DATE_PATTERN.length()) { + throw new IllegalArgumentException("Invalid date range " + dateRange); + } + // Parse start date which exists before "-" in date range. + // If "-" does not exist in date range, this effectively + // gives single date. + start = parseDate(startDate); + } + if (dates.length > 1) { + String endDate = dates[1].trim(); + if (!endDate.isEmpty()) { + // End date is not in yyyyMMdd format. + if (endDate.length() != DATE_PATTERN.length()) { + throw new IllegalArgumentException( + "Invalid date range " + dateRange); + } + // Parse end date which exists after "-" in date range. + end = parseDate(endDate); + } + } else { + // Its a single date(without "-" in date range), so set + // end equal to start. + end = start; + } + if (start != null && end != null) { + if (start > end) { + throw new IllegalArgumentException("Invalid date range " + dateRange); + } + } + return new DateRange(start, end); + } catch (ParseException e) { + // Date could not be parsed. + throw new IllegalArgumentException("Invalid date range " + dateRange); + } + } + private static Set<String> parseValuesStr(String str, String delimiter) { if (str == null || str.isEmpty()) { return null; @@ -205,7 +297,8 @@ public class TimelineReaderWebServices { if (e instanceof NumberFormatException) { throw new BadRequestException(invalidNumMsg + " is not a numeric value."); } else if (e instanceof IllegalArgumentException) { - throw new BadRequestException("Requested Invalid Field."); + throw new BadRequestException(e.getMessage() == null ? + "Requested Invalid Field." : e.getMessage()); } else { LOG.error("Error while processing REST request", e); throw new WebApplicationException(e, @@ -514,8 +607,20 @@ public class TimelineReaderWebServices { } /** - * Return a list of flows for a given cluster id. Cluster ID is not - * provided by client so default cluster ID has to be taken. + * Return a list of flows. Cluster ID is not provided by client so default + * cluster ID has to be taken. daterange, if specified is given as + * "[startdate]-[enddate]"(i.e. start and end date separated by -) or + * single date. Dates are interpreted in yyyyMMdd format and are assumed to + * be in GMT. If a single date is specified, all flows active on that date are + * returned. If both startdate and enddate is given, all flows active between + * start and end date will be returned. If only startdate is given, flows + * active on and after startdate are returned. If only enddate is given, flows + * active on and before enddate are returned. + * For example : + * "daterange=20150711" returns flows active on 20150711. + * "daterange=20150711-20150714" returns flows active between these 2 dates. + * "daterange=20150711-" returns flows active on and after 20150711. + * "daterange=-20150711" returns flows active on and before 20150711. */ @GET @Path("/flows/") @@ -524,12 +629,25 @@ public class TimelineReaderWebServices { @Context HttpServletRequest req, @Context HttpServletResponse res, @QueryParam("limit") String limit, + @QueryParam("daterange") String dateRange, @QueryParam("fields") String fields) { - return getFlows(req, res, null, limit, fields); + return getFlows(req, res, null, limit, dateRange, fields); } /** - * Return a list of flows for a given cluster id. + * Return a list of flows for a given cluster id. daterange, if specified is + * given as "[startdate]-[enddate]"(i.e. start and end date separated by -) or + * single date. Dates are interpreted in yyyyMMdd format and are assumed to + * be in GMT. If a single date is specified, all flows active on that date are + * returned. If both startdate and enddate is given, all flows active between + * start and end date will be returned. If only startdate is given, flows + * active on and after startdate are returned. If only enddate is given, flows + * active on and before enddate are returned. + * For example : + * "daterange=20150711" returns flows active on 20150711. + * "daterange=20150711-20150714" returns flows active between these 2 dates. + * "daterange=20150711-" returns flows active on and after 20150711. + * "daterange=-20150711" returns flows active on and before 20150711. */ @GET @Path("/flows/{clusterid}/") @@ -539,6 +657,7 @@ public class TimelineReaderWebServices { @Context HttpServletResponse res, @PathParam("clusterid") String clusterId, @QueryParam("limit") String limit, + @QueryParam("daterange") String dateRange, @QueryParam("fields") String fields) { String url = req.getRequestURI() + (req.getQueryString() == null ? "" : @@ -550,11 +669,12 @@ public class TimelineReaderWebServices { TimelineReaderManager timelineReaderManager = getTimelineReaderManager(); Set<TimelineEntity> entities = null; try { + DateRange range = parseDateRange(dateRange); entities = timelineReaderManager.getEntities( null, parseStr(clusterId), null, null, null, TimelineEntityType.YARN_FLOW_ACTIVITY.toString(), parseLongStr(limit), - null, null, null, null, null, null, null, null, null, null, - parseFieldsStr(fields, COMMA_DELIMITER)); + range.dateStart, range.dateEnd, null, null, null, null, null, null, + null, null, parseFieldsStr(fields, COMMA_DELIMITER)); } catch (Exception e) { handleException(e, url, startTime, "limit"); } http://git-wip-us.apache.org/repos/asf/hadoop/blob/c0910861/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/FlowActivityEntityReader.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/FlowActivityEntityReader.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/FlowActivityEntityReader.java index 70a0915..3e32128 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/FlowActivityEntityReader.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/FlowActivityEntityReader.java @@ -87,6 +87,12 @@ class FlowActivityEntityReader extends TimelineEntityReader { if (limit == null || limit < 0) { limit = TimelineReader.DEFAULT_LIMIT; } + if (createdTimeBegin == null) { + createdTimeBegin = DEFAULT_BEGIN_TIME; + } + if (createdTimeEnd == null) { + createdTimeEnd = DEFAULT_END_TIME; + } } @Override @@ -100,7 +106,16 @@ class FlowActivityEntityReader extends TimelineEntityReader { protected ResultScanner getResults(Configuration hbaseConf, Connection conn) throws IOException { Scan scan = new Scan(); - scan.setRowPrefixFilter(FlowActivityRowKey.getRowKeyPrefix(clusterId)); + if (createdTimeBegin == DEFAULT_BEGIN_TIME && + createdTimeEnd == DEFAULT_END_TIME) { + scan.setRowPrefixFilter(FlowActivityRowKey.getRowKeyPrefix(clusterId)); + } else { + scan.setStartRow( + FlowActivityRowKey.getRowKeyPrefix(clusterId, createdTimeEnd)); + scan.setStopRow( + FlowActivityRowKey.getRowKeyPrefix(clusterId, + (createdTimeBegin <= 0 ? 0: (createdTimeBegin - 1)))); + } // use the page filter to limit the result to the page size // the scanner may still return more than the limit; therefore we need to // read the right number as we iterate http://git-wip-us.apache.org/repos/asf/hadoop/blob/c0910861/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/flow/FlowActivityRowKey.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/flow/FlowActivityRowKey.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/flow/FlowActivityRowKey.java index f7841e0..fc1aa70 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/flow/FlowActivityRowKey.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/main/java/org/apache/hadoop/yarn/server/timelineservice/storage/flow/FlowActivityRowKey.java @@ -55,11 +55,32 @@ public class FlowActivityRowKey { return flowId; } + /** + * Constructs a row key prefix for the flow activity table as follows: + * {@code clusterId!} + * + * @param clusterId + * @return byte array with the row key prefix + */ public static byte[] getRowKeyPrefix(String clusterId) { return Bytes.toBytes(Separator.QUALIFIERS.joinEncoded(clusterId, "")); } /** + * Constructs a row key prefix for the flow activity table as follows: + * {@code clusterId!dayTimestamp!} + * + * @param clusterId + * @param dayTs + * @return byte array with the row key prefix + */ + public static byte[] getRowKeyPrefix(String clusterId, long dayTs) { + return Separator.QUALIFIERS.join( + Bytes.toBytes(Separator.QUALIFIERS.encode(clusterId)), + Bytes.toBytes(TimelineStorageUtils.invertLong(dayTs)), new byte[0]); + } + + /** * Constructs a row key for the flow activity table as follows: * {@code clusterId!dayTimestamp!user!flowId} * http://git-wip-us.apache.org/repos/asf/hadoop/blob/c0910861/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/reader/TestTimelineReaderWebServicesHBaseStorage.java ---------------------------------------------------------------------- diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/reader/TestTimelineReaderWebServicesHBaseStorage.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/reader/TestTimelineReaderWebServicesHBaseStorage.java index f6a5090..4f53fe2 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/reader/TestTimelineReaderWebServicesHBaseStorage.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-timelineservice/src/test/java/org/apache/hadoop/yarn/server/timelineservice/reader/TestTimelineReaderWebServicesHBaseStorage.java @@ -27,6 +27,7 @@ import java.lang.reflect.UndeclaredThrowableException; import java.net.HttpURLConnection; import java.net.URI; import java.net.URL; +import java.text.DateFormat; import java.util.HashMap; import java.util.HashSet; import java.util.Map; @@ -48,6 +49,7 @@ import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.server.metrics.ApplicationMetricsConstants; import org.apache.hadoop.yarn.server.timelineservice.storage.HBaseTimelineWriterImpl; import org.apache.hadoop.yarn.server.timelineservice.storage.TimelineSchemaCreator; +import org.apache.hadoop.yarn.server.timelineservice.storage.common.TimelineStorageUtils; import org.apache.hadoop.yarn.webapp.YarnJacksonJaxbJsonProvider; import org.junit.After; import org.junit.AfterClass; @@ -70,6 +72,8 @@ public class TestTimelineReaderWebServicesHBaseStorage { private TimelineReaderServer server; private static HBaseTestingUtility util; private static long ts = System.currentTimeMillis(); + private static long dayTs = + TimelineStorageUtils.getTopOfTheDayTimestamp(ts); @BeforeClass public static void setup() throws Exception { @@ -509,6 +513,61 @@ public class TestTimelineReaderWebServicesHBaseStorage { entities = resp.getEntity(new GenericType<Set<FlowActivityEntity>>(){}); assertNotNull(entities); assertEquals(1, entities.size()); + + DateFormat fmt = TimelineReaderWebServices.DATE_FORMAT.get(); + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=" + fmt.format(dayTs) + "-" + + fmt.format(dayTs + (2*86400000L))); + resp = getResponse(client, uri); + entities = resp.getEntity(new GenericType<Set<FlowActivityEntity>>(){}); + assertNotNull(entities); + assertEquals(2, entities.size()); + for (FlowActivityEntity entity : entities) { + assertTrue((entity.getId().endsWith("@flow_name") && + entity.getFlowRuns().size() == 2) || + (entity.getId().endsWith("@flow_name2") && + entity.getFlowRuns().size() == 1)); + } + + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=" + + fmt.format(dayTs + (4*86400000L))); + resp = getResponse(client, uri); + entities = resp.getEntity(new GenericType<Set<FlowActivityEntity>>(){}); + assertNotNull(entities); + assertEquals(0, entities.size()); + + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=-" + + fmt.format(dayTs + (2*86400000L))); + resp = getResponse(client, uri); + entities = resp.getEntity(new GenericType<Set<FlowActivityEntity>>(){}); + assertNotNull(entities); + assertEquals(2, entities.size()); + + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=" + + fmt.format(dayTs - (2*86400000L)) + "-"); + resp = getResponse(client, uri); + entities = resp.getEntity(new GenericType<Set<FlowActivityEntity>>(){}); + assertNotNull(entities); + assertEquals(2, entities.size()); + + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=20150711:20150714"); + verifyHttpResponse(client, uri, Status.BAD_REQUEST); + + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=20150714-20150711"); + verifyHttpResponse(client, uri, Status.BAD_REQUEST); + + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=2015071129-20150712"); + verifyHttpResponse(client, uri, Status.BAD_REQUEST); + + uri = URI.create("http://localhost:"; + serverPort + "/ws/v2/" + + "timeline/flows/cluster1?daterange=20150711-2015071243"); + verifyHttpResponse(client, uri, Status.BAD_REQUEST); } finally { client.destroy(); }
