AMBARI-13411. Problem in precision handling of metrics returned by AMS. (Aravindan Vijayan via swagle)

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/Precision.java → ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/Precision.java

@@ -15,7 +15,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
+package org.apache.hadoop.metrics2.sink.timeline;
 
 /**
  * Is used to determine metrics aggregate table.
@@ -45,4 +45,19 @@ public enum Precision {
         "minutes, hours or days", e);
     }
   }
+
+  public static Precision getPrecision(long startTime, long endTime) {
+    long HOUR = 3600000; // 1 hour
+    long DAY = 86400000; // 1 day
+    long timeRange = endTime - startTime;
+    if (timeRange > 30 * DAY) {
+      return Precision.DAYS;
+    } else if (timeRange > 1 * DAY) {
+      return Precision.HOURS;
+    } else if (timeRange > 2 * HOUR) {
+      return Precision.MINUTES;
+    } else {
+      return Precision.SECONDS;
+    }
+  }
 }

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/HBaseTimelineMetricStore.java

@@ -21,6 +21,7 @@ package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.service.AbstractService;

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java

@@ -23,6 +23,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.util.RetryCounter;
 import org.apache.hadoop.hbase.util.RetryCounterFactory;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.SingleValuedTimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
@@ -102,10 +103,10 @@ public class PhoenixHBaseAccessor {
   // cluster and host levels.
   static final long DEFAULT_OUT_OF_BAND_TIME_ALLOWANCE = 300000;
   /**
-   * 4 metrics/min * 60 * 24: Retrieve data for 1 day.
+   * 8 metrics * 60minutes * 24hours => Reasonable upper bound on the limit such that our Precision calculation for a given time range makes sense.
    */
-  private static final int METRICS_PER_MINUTE = 4;
-  public static int RESULTSET_LIMIT = (int)TimeUnit.DAYS.toMinutes(1) * METRICS_PER_MINUTE;
+  private static final int METRICS_PER_MINUTE = 8;
+  public static int RESULTSET_LIMIT = (int)TimeUnit.HOURS.toMinutes(24) * METRICS_PER_MINUTE;
 
   private static final TimelineMetricReadHelper TIMELINE_METRIC_READ_HELPER = new TimelineMetricReadHelper();
   private static ObjectMapper mapper = new ObjectMapper();
@@ -127,7 +128,7 @@ public class PhoenixHBaseAccessor {
                               ConnectionProvider dataSource) {
     this.hbaseConf = hbaseConf;
     this.metricsConf = metricsConf;
-    RESULTSET_LIMIT = metricsConf.getInt(GLOBAL_RESULT_LIMIT, 5760);
+    RESULTSET_LIMIT = metricsConf.getInt(GLOBAL_RESULT_LIMIT, RESULTSET_LIMIT);
     try {
       Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
     } catch (ClassNotFoundException e) {
@@ -515,7 +516,8 @@ public class PhoenixHBaseAccessor {
       List<Function>> metricFunctions, ResultSet rs)
       throws SQLException, IOException {
     if (condition.getPrecision() == Precision.HOURS
-      || condition.getPrecision() == Precision.MINUTES) {
+      || condition.getPrecision() == Precision.MINUTES
+      || condition.getPrecision() == Precision.DAYS) {
 
       String metricName = rs.getString("METRIC_NAME");
       List<Function> functions = metricFunctions.get(metricName);

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricStore.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
 
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/Condition.java

@@ -1,6 +1,6 @@
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query;
 
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.Precision;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 
 import java.util.List;
 

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/DefaultCondition.java

@@ -17,7 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.Precision;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 
 import java.util.LinkedHashSet;
 import java.util.List;

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/EmptyCondition.java

@@ -17,7 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query;
 
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.Precision;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 
 import java.util.List;
 

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixTransactSQL.java

@@ -6,9 +6,9 @@
  * 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
- *
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
  * 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.
@@ -20,7 +20,7 @@ package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.Precision;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 
 import java.sql.Connection;
 import java.sql.PreparedStatement;
@@ -56,50 +56,50 @@ public class PhoenixTransactSQL {
 
   public static final String CREATE_METRICS_AGGREGATE_TABLE_SQL =
     "CREATE TABLE IF NOT EXISTS %s " +
-    "(METRIC_NAME VARCHAR, " +
-    "HOSTNAME VARCHAR, " +
-    "APP_ID VARCHAR, " +
-    "INSTANCE_ID VARCHAR, " +
-    "SERVER_TIME UNSIGNED_LONG NOT NULL, " +
-    "UNITS CHAR(20), " +
-    "METRIC_SUM DOUBLE," +
-    "METRIC_COUNT UNSIGNED_INT, " +
-    "METRIC_MAX DOUBLE," +
-    "METRIC_MIN DOUBLE CONSTRAINT pk " +
-    "PRIMARY KEY (METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, " +
-    "SERVER_TIME)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, TTL=%s," +
-    " COMPRESSION='%s'";
+      "(METRIC_NAME VARCHAR, " +
+      "HOSTNAME VARCHAR, " +
+      "APP_ID VARCHAR, " +
+      "INSTANCE_ID VARCHAR, " +
+      "SERVER_TIME UNSIGNED_LONG NOT NULL, " +
+      "UNITS CHAR(20), " +
+      "METRIC_SUM DOUBLE," +
+      "METRIC_COUNT UNSIGNED_INT, " +
+      "METRIC_MAX DOUBLE," +
+      "METRIC_MIN DOUBLE CONSTRAINT pk " +
+      "PRIMARY KEY (METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, " +
+      "SERVER_TIME)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, TTL=%s," +
+      " COMPRESSION='%s'";
 
   public static final String CREATE_METRICS_CLUSTER_AGGREGATE_TABLE_SQL =
     "CREATE TABLE IF NOT EXISTS %s " +
-    "(METRIC_NAME VARCHAR, " +
-    "APP_ID VARCHAR, " +
-    "INSTANCE_ID VARCHAR, " +
-    "SERVER_TIME UNSIGNED_LONG NOT NULL, " +
-    "UNITS CHAR(20), " +
-    "METRIC_SUM DOUBLE, " +
-    "HOSTS_COUNT UNSIGNED_INT, " +
-    "METRIC_MAX DOUBLE, " +
-    "METRIC_MIN DOUBLE " +
-    "CONSTRAINT pk PRIMARY KEY (METRIC_NAME, APP_ID, INSTANCE_ID, " +
-    "SERVER_TIME)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, " +
-    "TTL=%s, COMPRESSION='%s'";
+      "(METRIC_NAME VARCHAR, " +
+      "APP_ID VARCHAR, " +
+      "INSTANCE_ID VARCHAR, " +
+      "SERVER_TIME UNSIGNED_LONG NOT NULL, " +
+      "UNITS CHAR(20), " +
+      "METRIC_SUM DOUBLE, " +
+      "HOSTS_COUNT UNSIGNED_INT, " +
+      "METRIC_MAX DOUBLE, " +
+      "METRIC_MIN DOUBLE " +
+      "CONSTRAINT pk PRIMARY KEY (METRIC_NAME, APP_ID, INSTANCE_ID, " +
+      "SERVER_TIME)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, " +
+      "TTL=%s, COMPRESSION='%s'";
 
   // HOSTS_COUNT vs METRIC_COUNT
   public static final String CREATE_METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_SQL =
     "CREATE TABLE IF NOT EXISTS %s " +
-    "(METRIC_NAME VARCHAR, " +
-    "APP_ID VARCHAR, " +
-    "INSTANCE_ID VARCHAR, " +
-    "SERVER_TIME UNSIGNED_LONG NOT NULL, " +
-    "UNITS CHAR(20), " +
-    "METRIC_SUM DOUBLE, " +
-    "METRIC_COUNT UNSIGNED_INT, " +
-    "METRIC_MAX DOUBLE, " +
-    "METRIC_MIN DOUBLE " +
-    "CONSTRAINT pk PRIMARY KEY (METRIC_NAME, APP_ID, INSTANCE_ID, " +
-    "SERVER_TIME)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, " +
-    "TTL=%s, COMPRESSION='%s'";
+      "(METRIC_NAME VARCHAR, " +
+      "APP_ID VARCHAR, " +
+      "INSTANCE_ID VARCHAR, " +
+      "SERVER_TIME UNSIGNED_LONG NOT NULL, " +
+      "UNITS CHAR(20), " +
+      "METRIC_SUM DOUBLE, " +
+      "METRIC_COUNT UNSIGNED_INT, " +
+      "METRIC_MAX DOUBLE, " +
+      "METRIC_MIN DOUBLE " +
+      "CONSTRAINT pk PRIMARY KEY (METRIC_NAME, APP_ID, INSTANCE_ID, " +
+      "SERVER_TIME)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, " +
+      "TTL=%s, COMPRESSION='%s'";
 
   /**
    * ALTER table to set new options
@@ -269,7 +269,7 @@ public class PhoenixTransactSQL {
   }
 
   public static PreparedStatement prepareGetMetricsSqlStmt(Connection connection,
-      Condition condition) throws SQLException {
+                                                           Condition condition) throws SQLException {
 
     validateConditionIsNotEmpty(condition);
     validateRowCountLimit(condition);
@@ -284,42 +284,25 @@ public class PhoenixTransactSQL {
       if (condition.getPrecision() == null) {
         long endTime = condition.getEndTime() == null ? System.currentTimeMillis() : condition.getEndTime();
         long startTime = condition.getStartTime() == null ? 0 : condition.getStartTime();
-        Long timeRange = endTime - startTime;
-        if (timeRange > 30 * DAY) {
+        Precision precision = Precision.getPrecision(startTime, endTime);
+        condition.setPrecision(precision);
+      }
+      switch (condition.getPrecision()) {
+        case DAYS:
           metricsTable = METRICS_AGGREGATE_DAILY_TABLE_NAME;
           query = GET_METRIC_AGGREGATE_ONLY_SQL;
-          condition.setPrecision(Precision.DAYS);
-        } else if (timeRange > DAY) {
+          break;
+        case HOURS:
           metricsTable = METRICS_AGGREGATE_HOURLY_TABLE_NAME;
           query = GET_METRIC_AGGREGATE_ONLY_SQL;
-          condition.setPrecision(Precision.HOURS);
-        } else if (timeRange > 10 * HOUR) {
+          break;
+        case MINUTES:
           metricsTable = METRICS_AGGREGATE_MINUTE_TABLE_NAME;
           query = GET_METRIC_AGGREGATE_ONLY_SQL;
-          condition.setPrecision(Precision.MINUTES);
-        } else {
+          break;
+        default:
           metricsTable = METRICS_RECORD_TABLE_NAME;
           query = GET_METRIC_SQL;
-          condition.setPrecision(Precision.SECONDS);
-        }
-      } else {
-        switch (condition.getPrecision()) {
-          case DAYS:
-            metricsTable = METRICS_AGGREGATE_DAILY_TABLE_NAME;
-            query = GET_METRIC_AGGREGATE_ONLY_SQL;
-            break;
-          case HOURS:
-            metricsTable = METRICS_AGGREGATE_HOURLY_TABLE_NAME;
-            query = GET_METRIC_AGGREGATE_ONLY_SQL;
-            break;
-          case MINUTES:
-            metricsTable = METRICS_AGGREGATE_MINUTE_TABLE_NAME;
-            query = GET_METRIC_AGGREGATE_ONLY_SQL;
-            break;
-          default:
-            metricsTable = METRICS_RECORD_TABLE_NAME;
-            query = GET_METRIC_SQL;
-        }
       }
 
       stmtStr = String.format(query,
@@ -413,19 +396,32 @@ public class PhoenixTransactSQL {
 
   private static void validateRowCountLimit(Condition condition) {
     if (condition.getMetricNames() == null
-        || condition.getMetricNames().isEmpty() ) {
+      || condition.getMetricNames().isEmpty()) {
       //aggregator can use empty metrics query
       return;
     }
 
     long range = condition.getEndTime() - condition.getStartTime();
-    long rowsPerMetric = TimeUnit.MILLISECONDS.toHours(range) + 1;
+    long rowsPerMetric;
 
+    //Get Precision (passed in or computed) and estimate values returned based on that.
     Precision precision = condition.getPrecision();
-    // for minutes and seconds we can use the rowsPerMetric computed based on
-    // minutes
-    if (precision != null && precision == Precision.HOURS) {
-      rowsPerMetric = TimeUnit.MILLISECONDS.toHours(range) + 1;
+    if (precision == null) {
+      precision = Precision.getPrecision(condition.getStartTime(), condition.getEndTime());
+    }
+
+    switch (precision) {
+      case DAYS:
+        rowsPerMetric = TimeUnit.MILLISECONDS.toDays(range);
+        break;
+      case HOURS:
+        rowsPerMetric = TimeUnit.MILLISECONDS.toHours(range);
+        break;
+      case MINUTES:
+        rowsPerMetric = TimeUnit.MILLISECONDS.toMinutes(range);
+        break;
+      default:
+        rowsPerMetric = TimeUnit.MILLISECONDS.toSeconds(range)/10; //10 second data in METRIC_AGGREGATE table
     }
 
     long totalRowsRequested = rowsPerMetric * condition.getMetricNames().size();
@@ -437,7 +433,7 @@ public class PhoenixTransactSQL {
   }
 
   public static PreparedStatement prepareGetLatestMetricSqlStmt(
-          Connection connection, Condition condition) throws SQLException {
+    Connection connection, Condition condition) throws SQLException {
 
     validateConditionIsNotEmpty(condition);
 
@@ -473,6 +469,7 @@ public class PhoenixTransactSQL {
 
     return stmt;
   }
+
   private static PreparedStatement setQueryParameters(PreparedStatement stmt,
                                                       Condition condition)
     throws SQLException {
@@ -480,7 +477,7 @@ public class PhoenixTransactSQL {
     //For GET_LATEST_METRIC_SQL_SINGLE_HOST parameters should be set 2 times
     do {
       if (condition.getMetricNames() != null) {
-        for (String metricName: condition.getMetricNames()) {
+        for (String metricName : condition.getMetricNames()) {
           if (LOG.isDebugEnabled()) {
             LOG.debug("Setting pos: " + pos + ", value = " + metricName);
           }
@@ -519,48 +516,36 @@ public class PhoenixTransactSQL {
   }
 
   public static PreparedStatement prepareGetAggregateSqlStmt(
-          Connection connection, Condition condition) throws SQLException {
+    Connection connection, Condition condition) throws SQLException {
 
     validateConditionIsNotEmpty(condition);
+    validateRowCountLimit(condition);
 
     String metricsAggregateTable;
     String queryStmt;
     if (condition.getPrecision() == null) {
       long endTime = condition.getEndTime() == null ? System.currentTimeMillis() : condition.getEndTime();
       long startTime = condition.getStartTime() == null ? 0 : condition.getStartTime();
-      Long timeRange = endTime - startTime;
-      if (timeRange > 30 * DAY) {
+      condition.setPrecision(Precision.getPrecision(startTime, endTime));
+    }
+    switch (condition.getPrecision()) {
+      case DAYS:
         metricsAggregateTable = METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME;
         queryStmt = GET_CLUSTER_AGGREGATE_TIME_SQL;
-        condition.setPrecision(Precision.DAYS);
-      } else if (timeRange > DAY) {
+        break;
+      case HOURS:
         metricsAggregateTable = METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME;
         queryStmt = GET_CLUSTER_AGGREGATE_TIME_SQL;
-        condition.setPrecision(Precision.HOURS);
-      } else {
+        break;
+      //TODO : Include MINUTE case after introducing CLUSTER_AGGREGATOR_MINUTE
+      default:
         metricsAggregateTable = METRICS_CLUSTER_AGGREGATE_TABLE_NAME;
         queryStmt = GET_CLUSTER_AGGREGATE_SQL;
-        condition.setPrecision(Precision.SECONDS);
-      }
-    } else {
-      switch (condition.getPrecision()) {
-        case DAYS:
-          metricsAggregateTable = METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME;
-          queryStmt = GET_CLUSTER_AGGREGATE_TIME_SQL;
-          break;
-        case HOURS:
-          metricsAggregateTable = METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME;
-          queryStmt = GET_CLUSTER_AGGREGATE_TIME_SQL;
-          break;
-        default:
-          metricsAggregateTable = METRICS_CLUSTER_AGGREGATE_TABLE_NAME;
-          queryStmt = GET_CLUSTER_AGGREGATE_SQL;
-      }
     }
 
     queryStmt = String.format(queryStmt,
-            getNaiveTimeRangeHint(condition.getStartTime(), NATIVE_TIME_RANGE_DELTA),
-            metricsAggregateTable);
+      getNaiveTimeRangeHint(condition.getStartTime(), NATIVE_TIME_RANGE_DELTA),
+      metricsAggregateTable);
 
     StringBuilder sb = new StringBuilder(queryStmt);
     sb.append(" WHERE ");
@@ -571,6 +556,7 @@ public class PhoenixTransactSQL {
     }
 
     String query = sb.toString();
+
     if (LOG.isDebugEnabled()) {
       LOG.debug("SQL => " + query + ", condition => " + condition);
     }
@@ -617,7 +603,7 @@ public class PhoenixTransactSQL {
       stmtStr = condition.getStatement();
     } else {
       stmtStr = String.format(GET_CLUSTER_AGGREGATE_SQL, "",
-          METRICS_CLUSTER_AGGREGATE_TABLE_NAME);
+        METRICS_CLUSTER_AGGREGATE_TABLE_NAME);
     }
 
     StringBuilder sb = new StringBuilder(stmtStr);
@@ -639,25 +625,25 @@ public class PhoenixTransactSQL {
 
     PreparedStatement stmt = null;
     try {
-    stmt = connection.prepareStatement(query);
-    int pos = 1;
-    if (condition.getMetricNames() != null) {
-      for (; pos <= condition.getMetricNames().size(); pos++) {
-        stmt.setString(pos, condition.getMetricNames().get(pos - 1));
+      stmt = connection.prepareStatement(query);
+      int pos = 1;
+      if (condition.getMetricNames() != null) {
+        for (; pos <= condition.getMetricNames().size(); pos++) {
+          stmt.setString(pos, condition.getMetricNames().get(pos - 1));
+        }
       }
-    }
-    if (condition.getAppId() != null) {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Setting pos: " + pos + ", value: " + condition.getAppId());
+      if (condition.getAppId() != null) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Setting pos: " + pos + ", value: " + condition.getAppId());
+        }
+        stmt.setString(pos++, condition.getAppId());
+      }
+      if (condition.getInstanceId() != null) {
+        stmt.setString(pos, condition.getInstanceId());
       }
-      stmt.setString(pos++, condition.getAppId());
-    }
-    if (condition.getInstanceId() != null) {
-      stmt.setString(pos, condition.getInstanceId());
-    }
     } catch (SQLException e) {
       if (stmt != null) {
-        
+
       }
       throw e;
     }

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/SplitByMetricNamesCondition.java

@@ -17,7 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query;
 
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.Precision;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 
 import java.util.Collections;
 import java.util.List;

ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/webapp/TimelineWebServices.java

@@ -30,7 +30,7 @@ import org.apache.hadoop.yarn.api.records.timeline.TimelineEvents;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.Precision;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricStore;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.EntityIdentifier;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.timeline.GenericObjectMapper;

ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/ITPhoenixHBaseAccessor.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.Function;

ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessorTest.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.Function;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.Condition;

ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TestPhoenixTransactSQL.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
 
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.Condition;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.DefaultCondition;
 import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL;
@@ -172,7 +173,7 @@ public class TestPhoenixTransactSQL {
     Assert.assertEquals(Precision.SECONDS, condition.getPrecision());
     verify(connection, preparedStatement);
 
-    // SECONDS precision
+    // MINUTES precision
     startTime = endTime-PhoenixTransactSQL.DAY/1000;
     condition = new DefaultCondition(
       Arrays.asList("cpu_user", "mem_free"), Collections.singletonList("h1"),
@@ -187,7 +188,7 @@ public class TestPhoenixTransactSQL {
     PhoenixTransactSQL.prepareGetAggregateSqlStmt(connection, condition);
     stmt = stmtCapture.getValue();
     Assert.assertTrue(stmt.contains("FROM METRIC_AGGREGATE"));
-    Assert.assertEquals(Precision.SECONDS, condition.getPrecision());
+    Assert.assertEquals(Precision.MINUTES, condition.getPrecision());
     verify(connection, preparedStatement);
 
     // HOURS precision
@@ -286,7 +287,7 @@ public class TestPhoenixTransactSQL {
     reset(connection, preparedStatement);
 
     // SECONDS precision
-    startTime = endTime-PhoenixTransactSQL.HOUR*10/1000;
+    startTime = endTime-PhoenixTransactSQL.HOUR*2/1000;
     condition = new DefaultCondition(
       Arrays.asList("cpu_user", "mem_free"), Collections.singletonList("h1"),
       "a1", "i1", startTime, endTime, null, null, false);

ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TestTimelineMetricStore.java

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline;
 
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricCacheEntryFactory.java

@@ -26,6 +26,7 @@ import org.apache.ambari.server.controller.internal.URLStreamProvider;
 import org.apache.ambari.server.controller.metrics.timeline.MetricsRequestHelper;
 import org.apache.ambari.server.controller.spi.TemporalInfo;
 import org.apache.ambari.server.controller.utilities.StreamProvider;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
 import org.apache.http.client.utils.URIBuilder;
@@ -49,6 +50,8 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
   private MetricsRequestHelper requestHelperForGets;
   private MetricsRequestHelper requestHelperForUpdates;
   private final Long BUFFER_TIME_DIFF_CATCHUP_INTERVAL;
+  public static final long HOUR = 3600000; // 1 hour
+  public static final long DAY = 86400000; // 1 day
 
   @Inject
   public TimelineMetricCacheEntryFactory(Configuration configuration) {
@@ -100,7 +103,9 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
       value = new TimelineMetricsCacheValue(
         metricCacheKey.getTemporalInfo().getStartTime(),
         metricCacheKey.getTemporalInfo().getEndTime(),
-        cacheValue // Null or empty should prompt a refresh
+        cacheValue, // Null or empty should prompt a refresh
+        Precision.getPrecision(metricCacheKey.getTemporalInfo().getStartTimeMillis(),
+          metricCacheKey.getTemporalInfo().getEndTimeMillis()) //Initial Precision
       );
 
       LOG.debug("Created cache entry: " + value);
@@ -129,33 +134,49 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
     Long existingSeriesEndTime = existingMetrics.getEndTime();
 
     TemporalInfo newTemporalInfo = metricCacheKey.getTemporalInfo();
-    Long requestedStartTime = newTemporalInfo.getStartTime();
-    Long requestedEndTime = newTemporalInfo.getEndTime();
+    Long requestedStartTime = newTemporalInfo.getStartTimeMillis();
+    Long requestedEndTime = newTemporalInfo.getEndTimeMillis();
 
     // Calculate new start and end times
     URIBuilder uriBuilder = new URIBuilder(metricCacheKey.getSpec());
-    Long newStartTime = getRefreshRequestStartTime(existingSeriesStartTime,
-      existingSeriesEndTime, requestedStartTime);
-    Long newEndTime = getRefreshRequestEndTime(existingSeriesStartTime,
-      existingSeriesEndTime, requestedEndTime);
+
+    Precision requestedPrecision = Precision.getPrecision(requestedStartTime, requestedEndTime);
+    Precision currentPrecision = existingMetrics.getPrecision();
+
+    Long newStartTime = null;
+    Long newEndTime = null;
+    if(!requestedPrecision.equals(currentPrecision)) {
+      // Ignore cache entry. Get the entire data from the AMS and update the cache.
+      LOG.debug("Precision changed from " + currentPrecision + " to " + requestedPrecision);
+      newStartTime = requestedStartTime;
+      newEndTime = requestedEndTime;
+    } else {
+      //Get only the metric values for the delta period from the cache.
+      LOG.debug("No change in precision " + currentPrecision);
+      newStartTime = getRefreshRequestStartTime(existingSeriesStartTime,
+          existingSeriesEndTime, requestedStartTime);
+      newEndTime = getRefreshRequestEndTime(existingSeriesStartTime,
+          existingSeriesEndTime, requestedEndTime);
+    }
 
     // Cover complete overlap scenario
-    // time axis: |-------- exSt ----- reqSt ------ exEnd ----- extEnd ---------|
+    // time axis: |-------- exSt ----- reqSt ------ reqEnd ----- exEnd ---------|
     if (newEndTime > newStartTime &&
-       !(newStartTime.equals(existingSeriesStartTime) &&
-       newEndTime.equals(existingSeriesEndTime))) {
+       !((newStartTime.equals(existingSeriesStartTime) &&
+       newEndTime.equals(existingSeriesEndTime)) && requestedPrecision.equals(currentPrecision)) ) {
 
       LOG.debug("Existing cached timeseries startTime = " +
-        new Date(getMillisecondsTime(existingSeriesStartTime)) + ", endTime = " +
-        new Date(getMillisecondsTime(existingSeriesEndTime)));
+          new Date(getMillisecondsTime(existingSeriesStartTime)) + ", endTime = " +
+          new Date(getMillisecondsTime(existingSeriesEndTime)));
 
       LOG.debug("Requested timeseries startTime = " +
-        new Date(getMillisecondsTime(newStartTime)) + ", endTime = " +
-        new Date(getMillisecondsTime(newEndTime)));
+          new Date(getMillisecondsTime(newStartTime)) + ", endTime = " +
+          new Date(getMillisecondsTime(newEndTime)));
 
       // Update spec with new start and end time
       uriBuilder.setParameter("startTime", String.valueOf(newStartTime));
       uriBuilder.setParameter("endTime", String.valueOf(newEndTime));
+      uriBuilder.setParameter("precision",requestedPrecision.toString());
 
       try {
         TimelineMetrics newTimeSeries = requestHelperForUpdates.fetchTimelineMetrics(uriBuilder.toString());
@@ -163,11 +184,12 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
         // Update existing time series with new values
         updateTimelineMetricsInCache(newTimeSeries, existingMetrics,
           getMillisecondsTime(requestedStartTime),
-          getMillisecondsTime(requestedEndTime));
+          getMillisecondsTime(requestedEndTime), !currentPrecision.equals(requestedPrecision));
 
         // Replace old boundary values
         existingMetrics.setStartTime(requestedStartTime);
         existingMetrics.setEndTime(requestedEndTime);
+        existingMetrics.setPrecision(requestedPrecision);
 
       } catch (IOException io) {
         if (LOG.isDebugEnabled()) {
@@ -186,7 +208,7 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
    */
   protected void updateTimelineMetricsInCache(TimelineMetrics newMetrics,
       TimelineMetricsCacheValue timelineMetricsCacheValue,
-      Long requestedStartTime, Long requestedEndTime) {
+      Long requestedStartTime, Long requestedEndTime, boolean removeAll) {
 
     Map<String, TimelineMetric> existingTimelineMetricMap = timelineMetricsCacheValue.getTimelineMetrics();
 
@@ -205,6 +227,11 @@ public class TimelineMetricCacheEntryFactory implements UpdatingCacheEntryFactor
       TimelineMetric existingMetric = existingTimelineMetricMap.get(timelineMetric.getMetricName());
 
       if (existingMetric != null) {
+
+        if(removeAll) {
+          existingMetric.setMetricValues(new TreeMap<Long, Double>());
+        }
+
         Map<Long, Double> existingMetricValues = existingMetric.getMetricValues();
         LOG.trace("Existing metric: " + timelineMetric.getMetricName() +
           " # " + existingMetricValues.size());

ambari-server/src/main/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricsCacheValue.java

@@ -17,6 +17,7 @@
  */
 package org.apache.ambari.server.controller.metrics.timeline.cache;
 
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 
 import java.util.Date;
@@ -30,11 +31,13 @@ public class TimelineMetricsCacheValue {
   private Long startTime;
   private Long endTime;
   private Map<String, TimelineMetric> timelineMetrics;
+  private Precision precision;
 
-  public TimelineMetricsCacheValue(Long startTime, Long endTime, Map<String, TimelineMetric> timelineMetrics) {
+  public TimelineMetricsCacheValue(Long startTime, Long endTime, Map<String, TimelineMetric> timelineMetrics, Precision precision) {
     this.startTime = startTime;
     this.endTime = endTime;
     this.timelineMetrics = timelineMetrics;
+    this.precision = precision;
   }
 
   public Map<String, TimelineMetric> getTimelineMetrics() {
@@ -79,6 +82,7 @@ public class TimelineMetricsCacheValue {
       "metricNames = " + timelineMetrics.keySet() +
       ", startTime = " + new Date(getMillisecondsTime(startTime)) +
       ", endTime = " + new Date(getMillisecondsTime(endTime)) +
+      ", precision = " + precision +
       ", timelineMetrics =");
 
     for (TimelineMetric metric : timelineMetrics.values()) {
@@ -91,4 +95,12 @@ public class TimelineMetricsCacheValue {
     sb.append("}");
     return sb.toString();
   }
+
+  public Precision getPrecision() {
+    return precision;
+  }
+
+  public void setPrecision(Precision precision) {
+    this.precision = precision;
+  }
 }

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml

@@ -317,12 +317,12 @@
   </property>
   <property>
     <name>timeline.metrics.service.default.result.limit</name>
-    <value>5760</value>
-    <display-name>Metrics service default result limit</display-name>
+    <value>11520</value>
     <description>
       Max result limit on number of rows returned. Calculated as follows:
-      4 aggregate metrics/min * 60 * 24: Retrieve aggregate data for 1 day.
+      8 aggregate metrics/min * 60 * 24 : Retrieve MINUTE data for 24 hours.
     </description>
+    <display-name>Metrics service default result limit</display-name>
     <value-attributes>
       <type>int</type>
     </value-attributes>

ambari-server/src/test/java/org/apache/ambari/server/controller/metrics/timeline/cache/TimelineMetricCacheTest.java

@@ -28,14 +28,20 @@ import net.sf.ehcache.constructs.blocking.UpdatingCacheEntryFactory;
 import net.sf.ehcache.constructs.blocking.UpdatingSelfPopulatingCache;
 import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.controller.internal.TemporalInfoImpl;
+import org.apache.ambari.server.controller.metrics.timeline.MetricsRequestHelper;
 import org.apache.ambari.server.controller.spi.TemporalInfo;
 import org.apache.ambari.server.state.stack.OsFamily;
+import org.apache.hadoop.metrics2.sink.timeline.Precision;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
 import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
+import org.apache.http.client.utils.URIBuilder;
+import org.easymock.EasyMock;
 import org.easymock.IAnswer;
 import org.junit.After;
 import org.junit.Test;
 
+import java.lang.reflect.Field;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -146,7 +152,7 @@ public class TimelineMetricCacheTest {
       "app1",
       new TemporalInfoImpl(now, now + 1000, 1)
     );
-    TimelineMetricsCacheValue value = new TimelineMetricsCacheValue(now, now + 1000, valueMap);
+    TimelineMetricsCacheValue value = new TimelineMetricsCacheValue(now, now + 1000, valueMap, null);
     TimelineAppMetricCacheKey testKey = new TimelineAppMetricCacheKey(
       Collections.singleton("cpu_user"),
       "app1",
@@ -316,7 +322,7 @@ public class TimelineMetricCacheTest {
       new HashMap<String, TimelineMetric>() {{
         put("cpu_user", timelineMetric1);
         put("cpu_nice", timelineMetric2);
-      }});
+      }}, null);
 
     // New values
     TimelineMetrics newMetrics = new TimelineMetrics();
@@ -331,7 +337,7 @@ public class TimelineMetricCacheTest {
     newMetrics.getMetrics().add(timelineMetric3);
 
     factory.updateTimelineMetricsInCache(newMetrics, existingMetricValue,
-      now, now + 2000);
+      now, now + 2000,true);
 
     Assert.assertEquals(2, existingMetricValue.getTimelineMetrics().size());
     Assert.assertEquals(3, existingMetricValue.getTimelineMetrics().get("cpu_user").getMetricValues().size());
@@ -369,4 +375,113 @@ public class TimelineMetricCacheTest {
 
     Assert.assertTrue(key1.equals(key2));
   }
+
+  @Test
+  public void testTimelineMetricCachePrecisionUpdates () throws Exception {
+
+    Configuration configuration = createNiceMock(Configuration.class);
+    expect(configuration.getMetricCacheTTLSeconds()).andReturn(3600);
+    expect(configuration.getMetricCacheIdleSeconds()).andReturn(100);
+    expect(configuration.getMetricsCacheManagerHeapPercent()).andReturn("10%");
+    expect(configuration.getMetricRequestBufferTimeCatchupInterval()).andReturn(1000l).anyTimes();
+    replay(configuration);
+
+    final long now = System.currentTimeMillis();
+    long second = 1000;
+    long min = 60*second;
+    long hour = 60*min;
+    long day = 24*hour;
+    long year = 365*day;
+
+    //Original Values
+    Map<String, TimelineMetric> valueMap = new HashMap<String, TimelineMetric>();
+    TimelineMetric timelineMetric = new TimelineMetric();
+    timelineMetric.setMetricName("cpu_user");
+    timelineMetric.setAppId("app1");
+
+    TreeMap<Long, Double> metricValues = new TreeMap<Long, Double>();
+    for(long i=1*year-1*day;i>=0;i-=1*day) {
+      metricValues.put(now-i, 1.0);
+    }
+
+    timelineMetric.setMetricValues(metricValues);
+    valueMap.put("cpu_user", timelineMetric);
+
+    List<TimelineMetric> timelineMetricList = new ArrayList<>();
+    timelineMetricList.add(timelineMetric);
+    TimelineMetrics metrics = new TimelineMetrics();
+    metrics.setMetrics(timelineMetricList);
+
+    TimelineAppMetricCacheKey key = new TimelineAppMetricCacheKey(
+        Collections.singleton("cpu_user"),
+        "app1",
+        new TemporalInfoImpl(now-1*year, now, 1)
+    );
+    key.setSpec("");
+
+    //Updated values
+    Map<String, TimelineMetric> newValueMap = new HashMap<String, TimelineMetric>();
+    TimelineMetric newTimelineMetric = new TimelineMetric();
+    newTimelineMetric.setMetricName("cpu_user");
+    newTimelineMetric.setAppId("app1");
+
+    TreeMap<Long, Double> newMetricValues = new TreeMap<Long, Double>();
+    for(long i=1*hour;i<=2*day;i+=hour) {
+      newMetricValues.put(now-1*day+i, 2.0);
+    }
+
+    newTimelineMetric.setMetricValues(newMetricValues);
+    newValueMap.put("cpu_user", newTimelineMetric);
+
+    List<TimelineMetric> newTimelineMetricList = new ArrayList<>();
+    newTimelineMetricList.add(newTimelineMetric);
+    TimelineMetrics newMetrics = new TimelineMetrics();
+    newMetrics.setMetrics(newTimelineMetricList);
+
+    TimelineAppMetricCacheKey newKey = new TimelineAppMetricCacheKey(
+        Collections.singleton("cpu_user"),
+        "app1",
+        new TemporalInfoImpl(now-1*day, now+2*day, 1)
+    );
+    newKey.setSpec("");
+
+    MetricsRequestHelper metricsRequestHelperForGets = createMock(MetricsRequestHelper.class);
+    expect(metricsRequestHelperForGets.fetchTimelineMetrics(EasyMock.isA(String.class)))
+      .andReturn(metrics).andReturn(newMetrics);
+    replay(metricsRequestHelperForGets);
+
+    TimelineMetricCacheEntryFactory cacheEntryFactory = createMockBuilder(TimelineMetricCacheEntryFactory.class)
+     .withConstructor(Configuration.class).withArgs(configuration).createMock();
+
+    Field requestHelperField = TimelineMetricCacheEntryFactory.class.getDeclaredField("requestHelperForGets");
+    requestHelperField.setAccessible(true);
+    requestHelperField.set(cacheEntryFactory, metricsRequestHelperForGets);
+
+    requestHelperField = TimelineMetricCacheEntryFactory.class.getDeclaredField("requestHelperForUpdates");
+    requestHelperField.setAccessible(true);
+    requestHelperField.set(cacheEntryFactory, metricsRequestHelperForGets);
+
+    replay(cacheEntryFactory);
+
+    TimelineMetricCacheProvider cacheProvider = getMetricCacheProvider(configuration, cacheEntryFactory);
+    TimelineMetricCache cache = cacheProvider.getTimelineMetricsCache();
+
+    // call to get
+    metrics = cache.getAppTimelineMetricsFromCache(key);
+    List<TimelineMetric> metricsList = metrics.getMetrics();
+    Assert.assertEquals(1, metricsList.size());
+    TimelineMetric metric = metricsList.iterator().next();
+    Assert.assertEquals("cpu_user", metric.getMetricName());
+    Assert.assertEquals("app1", metric.getAppId());
+    Assert.assertEquals(metricValues, metric.getMetricValues());
+
+    // call to update with new key
+    metrics = cache.getAppTimelineMetricsFromCache(newKey);
+    metricsList = metrics.getMetrics();
+    Assert.assertEquals(1, metricsList.size());
+    Assert.assertEquals("cpu_user", metric.getMetricName());
+    Assert.assertEquals("app1", metric.getAppId());
+    Assert.assertEquals(newMetricValues,metric.getMetricValues());
+
+  }
 }