You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by av...@apache.org on 2016/03/22 20:54:11 UTC
ambari git commit: AMBARI-15476 : [AMS / Grafana] Rate calculation
with sum() is wrong (avijayan)
Repository: ambari
Updated Branches:
refs/heads/branch-2.2 472aab47f -> 800f27534
AMBARI-15476 : [AMS / Grafana] Rate calculation with sum() is wrong (avijayan)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/800f2753
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/800f2753
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/800f2753
Branch: refs/heads/branch-2.2
Commit: 800f275348399d691fc255b0043ac17260aad280
Parents: 472aab4
Author: Aravindan Vijayan <av...@hortonworks.com>
Authored: Mon Mar 21 21:36:31 2016 -0700
Committer: Aravindan Vijayan <av...@hortonworks.com>
Committed: Tue Mar 22 12:41:00 2016 -0700
----------------------------------------------------------------------
.../sink/timeline/PostProcessingUtil.java | 81 ++++++++++++++++++
.../timeline/cache/PostProcessingUtilTest.java | 85 +++++++++++++++++++
.../timeline/TimelineMetricConfiguration.java | 3 +
.../TimelineMetricClusterAggregatorSecond.java | 80 ++++++++++++++++--
...melineMetricClusterAggregatorSecondTest.java | 87 ++++++++++++++++++++
.../server/upgrade/UpgradeCatalog222.java | 13 ++-
.../0.1.0/configuration/ams-hbase-env.xml | 6 ++
.../0.1.0/configuration/ams-site.xml | 9 +-
.../stacks/HDP/2.0.6/services/stack_advisor.py | 8 +-
.../server/upgrade/UpgradeCatalog222Test.java | 3 +-
.../stacks/2.2/common/test_stack_advisor.py | 2 +-
11 files changed, 361 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/PostProcessingUtil.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/PostProcessingUtil.java b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/PostProcessingUtil.java
new file mode 100644
index 0000000..ab890ec
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/PostProcessingUtil.java
@@ -0,0 +1,81 @@
+/**
+ * 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
+ * <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.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.metrics2.sink.timeline;
+
+import java.util.Map;
+import java.util.TreeMap;
+
+public class PostProcessingUtil {
+
+ /*
+ Helper function to interpolate missing data on a series.
+ */
+ public static Map<Long, Double> interpolateMissingData(Map<Long, Double> metricValues, long expectedInterval) {
+
+ if (metricValues == null)
+ return null;
+
+ Long prevTime = null;
+ Double prevVal = null;
+ Map<Long, Double> interpolatedMetricValues = new TreeMap<Long, Double>();
+
+ for (Map.Entry<Long, Double> timeValueEntry : metricValues.entrySet()) {
+ Long currTime = timeValueEntry.getKey();
+ Double currVal = timeValueEntry.getValue();
+
+ if (prevTime != null) {
+ Long stepTime = prevTime;
+ while ((currTime - stepTime) > expectedInterval) {
+ stepTime+=expectedInterval;
+ double interpolatedValue = interpolate(stepTime,
+ prevTime, prevVal,
+ currTime, currVal);
+ interpolatedMetricValues.put(stepTime, interpolatedValue);
+ }
+ }
+
+ interpolatedMetricValues.put(currTime, currVal);
+ prevTime = currTime;
+ prevVal = currVal;
+ }
+ return interpolatedMetricValues;
+ }
+
+ public static Double interpolate(Long t, Long t1, Double m1,
+ Long t2, Double m2) {
+
+
+ //Linear Interpolation : y = y0 + (y1 - y0) * ((x - x0) / (x1 - x0))
+ if (m1 == null && m2 == null) {
+ return null;
+ }
+
+ if (m1 == null)
+ return m2;
+
+ if (m2 == null)
+ return m1;
+
+ if (t1 == null || t2 == null)
+ return null;
+
+ double slope = (m2 - m1) / (t2 - t1);
+ return m1 + slope * (t - t1);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/PostProcessingUtilTest.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/PostProcessingUtilTest.java b/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/PostProcessingUtilTest.java
new file mode 100644
index 0000000..1ec71d0
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/PostProcessingUtilTest.java
@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.metrics2.sink.timeline.cache;
+
+import junit.framework.Assert;
+import org.apache.hadoop.metrics2.sink.timeline.PostProcessingUtil;
+import org.apache.hadoop.metrics2.sink.timeline.SingleValuedTimelineMetric;
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.Map;
+import java.util.TreeMap;
+
+public class PostProcessingUtilTest {
+
+ @Test
+ public void testInterpolateMissinData() throws Exception {
+
+ Map<Long, Double> metricValues = new TreeMap<Long, Double>();
+ long interval = 60*1000;
+
+ long currentTime = System.currentTimeMillis();
+
+ for(int i = 10; i>=1;i--) {
+ if (i%4 != 0 && i != 5) { //Skip time points 4,5,8
+ metricValues.put(currentTime - i*interval, (double)i);
+ }
+ }
+ metricValues = PostProcessingUtil.interpolateMissingData(metricValues, interval);
+ Assert.assertTrue(metricValues.size() == 10);
+
+ Iterator it = metricValues.entrySet().iterator();
+ double sum = 0;
+ while (it.hasNext()) {
+ Map.Entry entry = (Map.Entry)it.next();
+ sum+= (double)entry.getValue();
+ }
+ Assert.assertEquals(sum, 55.0);
+ }
+
+ @Test
+ public void testInterpolate() throws Exception {
+
+ long t2 = System.currentTimeMillis();
+ long t1 = t2 - 60000;
+ double interpolatedValue;
+
+ //Test Equal Values
+ interpolatedValue = PostProcessingUtil.interpolate((t1 + 30000), t1, 10.0, t2, 10.0);
+ Assert.assertEquals(interpolatedValue, 10.0);
+
+ //Test Linear increase Values
+ interpolatedValue = PostProcessingUtil.interpolate((t1 + 30000), t1, 10.0, t2, 20.0);
+ Assert.assertEquals(interpolatedValue, 15.0);
+
+ //Test Linear decrease Values
+ interpolatedValue = PostProcessingUtil.interpolate((t1 + 30000), t1, 20.0, t2, 10.0);
+ Assert.assertEquals(interpolatedValue, 15.0);
+
+ //Test interpolation with non mid point time
+ interpolatedValue = PostProcessingUtil.interpolate((t1 + 20000), t1, 15.0, t2, 30.0); // 1:2 ratio
+ Assert.assertEquals(interpolatedValue, 20.0);
+
+ //Test interpolation with past time
+ interpolatedValue = PostProcessingUtil.interpolate((t1 - 60000), t1, 20.0, t2, 30.0);
+ Assert.assertEquals(interpolatedValue, 10.0);
+
+ }
+
+ }
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java
index aa12735..0ebc248 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricConfiguration.java
@@ -213,6 +213,9 @@ public class TimelineMetricConfiguration {
public static final String METRICS_METADATA_SYNC_SCHEDULE_DELAY =
"timeline.metrics.service.metadata.sync.delay";
+ public static final String TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED =
+ "timeline.metrics.cluster.aggregator.interpolation.enabled";
+
public static final String HOST_APP_ID = "HOST";
private Configuration hbaseConf;
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java
index e8e16a7..9354bc3 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecond.java
@@ -18,9 +18,8 @@
package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators;
-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.PostProcessingUtil;
import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.discovery.TimelineMetricMetadataManager;
@@ -31,10 +30,13 @@ import java.io.IOException;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
+import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+
import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.SERVER_SIDE_TIMESIFT_ADJUSTMENT;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED;
import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.GET_METRIC_SQL;
import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.METRICS_RECORD_TABLE_NAME;
import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.NATIVE_TIME_RANGE_DELTA;
@@ -50,6 +52,7 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
private final TimelineMetricAppAggregator appAggregator;
// 1 minute client side buffering adjustment
private final Long serverTimeShiftAdjustment;
+ private final boolean interpolationEnabled;
public TimelineMetricClusterAggregatorSecond(String aggregatorName,
@@ -71,13 +74,16 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
appAggregator = new TimelineMetricAppAggregator(metadataManager, metricsConf);
this.timeSliceIntervalMillis = timeSliceInterval;
this.serverTimeShiftAdjustment = Long.parseLong(metricsConf.get(SERVER_SIDE_TIMESIFT_ADJUSTMENT, "90000"));
+ this.interpolationEnabled = Boolean.parseBoolean(metricsConf.get(TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED, "true"));
}
@Override
protected void aggregate(ResultSet rs, long startTime, long endTime) throws SQLException, IOException {
// Account for time shift due to client side buffering by shifting the
// timestamps with the difference between server time and series start time
- List<Long[]> timeSlices = getTimeSlices(startTime - serverTimeShiftAdjustment, endTime);
+ // Also, we do not want to look at the shift time period from the end as well since we can interpolate those points
+ // that come earlier than the expected, during the next run.
+ List<Long[]> timeSlices = getTimeSlices(startTime - serverTimeShiftAdjustment, endTime - serverTimeShiftAdjustment);
// Initialize app aggregates for host metrics
appAggregator.init();
Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics =
@@ -100,15 +106,15 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
// Retaining order of the row-key avoids client side merge sort.
condition.addOrderByColumn("METRIC_NAME");
condition.addOrderByColumn("HOSTNAME");
- condition.addOrderByColumn("SERVER_TIME");
condition.addOrderByColumn("APP_ID");
+ condition.addOrderByColumn("SERVER_TIME");
return condition;
}
/**
* Return time slices to normalize the timeseries data.
*/
- private List<Long[]> getTimeSlices(long startTime, long endTime) {
+ protected List<Long[]> getTimeSlices(long startTime, long endTime) {
List<Long[]> timeSlices = new ArrayList<Long[]>();
long sliceStartTime = startTime;
while (sliceStartTime < endTime) {
@@ -119,7 +125,7 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
}
private Map<TimelineClusterMetric, MetricClusterAggregate> aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices)
- throws SQLException, IOException {
+ throws SQLException, IOException {
Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics =
new HashMap<TimelineClusterMetric, MetricClusterAggregate>();
@@ -186,8 +192,8 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
}
}
- private Map<TimelineClusterMetric, Double> sliceFromTimelineMetric(
- TimelineMetric timelineMetric, List<Long[]> timeSlices) {
+ protected Map<TimelineClusterMetric, Double> sliceFromTimelineMetric(
+ TimelineMetric timelineMetric, List<Long[]> timeSlices) {
if (timelineMetric.getMetricValues().isEmpty()) {
return null;
@@ -203,6 +209,7 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
timeShift = 0l;
}
+ Map<Long,Double> timeSliceValueMap = new HashMap<>();
for (Map.Entry<Long, Double> metric : timelineMetric.getMetricValues().entrySet()) {
// TODO: investigate null values - pre filter
if (metric.getValue() == null) {
@@ -229,13 +236,68 @@ public class TimelineMetricClusterAggregatorSecond extends AbstractTimelineAggre
Double oldValue = timelineClusterMetricMap.get(clusterMetric);
sum = oldValue + metric.getValue();
}
- timelineClusterMetricMap.put(clusterMetric, (sum / count));
+ double metricValue = sum / count;
+ timelineClusterMetricMap.put(clusterMetric, metricValue);
+ timeSliceValueMap.put(timestamp, metricValue);
}
}
+ if (interpolationEnabled) {
+ interpolateMissingPeriods(timelineClusterMetricMap, timelineMetric, timeSlices, timeSliceValueMap);
+ }
return timelineClusterMetricMap;
}
+ private void interpolateMissingPeriods(Map<TimelineClusterMetric, Double> timelineClusterMetricMap,
+ TimelineMetric timelineMetric,
+ List<Long[]> timeSlices,
+ Map<Long, Double> timeSliceValueMap) {
+
+
+ for (int sliceNum = 0; sliceNum < timeSlices.size(); sliceNum++) {
+ Long[] timeSlice = timeSlices.get(sliceNum);
+
+ if (!timeSliceValueMap.containsKey(timeSlice[1])) {
+ LOG.debug("Found an empty slice : " + new Date(timeSlice[0]) + ", " + new Date(timeSlice[1]));
+
+ Double lastSeenValue = null;
+ int index = sliceNum - 1;
+ Long[] prevTimeSlice = null;
+ while (lastSeenValue == null && index >= 0) {
+ prevTimeSlice = timeSlices.get(index--);
+ lastSeenValue = timeSliceValueMap.get(prevTimeSlice[1]);
+ }
+
+ Double nextSeenValue = null;
+ index = sliceNum + 1;
+ Long[] nextTimeSlice = null;
+ while ( nextSeenValue == null && index < timeSlices.size()) {
+ nextTimeSlice = timeSlices.get(index++);
+ nextSeenValue = timeSliceValueMap.get(nextTimeSlice[1]);
+ }
+
+ Double interpolatedValue = PostProcessingUtil.interpolate(timeSlice[1],
+ (prevTimeSlice != null ? prevTimeSlice[1] : null), lastSeenValue,
+ (nextTimeSlice != null ? nextTimeSlice[1] : null), nextSeenValue);
+
+ if (interpolatedValue != null) {
+ TimelineClusterMetric clusterMetric = new TimelineClusterMetric(
+ timelineMetric.getMetricName(),
+ timelineMetric.getAppId(),
+ timelineMetric.getInstanceId(),
+ timeSlice[1],
+ timelineMetric.getType());
+
+ LOG.debug("Interpolated value : " + interpolatedValue);
+ timelineClusterMetricMap.put(clusterMetric, interpolatedValue);
+ } else {
+ LOG.debug("Cannot compute interpolated value, hence skipping.");
+ }
+
+ }
+ }
+ }
+
/**
* Return end of the time slice into which the metric fits.
*/
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java b/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java
new file mode 100644
index 0000000..d2d478c
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TimelineMetricClusterAggregatorSecondTest.java
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators;
+
+import junit.framework.Assert;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.discovery.TimelineMetricMetadataManager;
+import org.easymock.EasyMock;
+import org.junit.Test;
+
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+
+public class TimelineMetricClusterAggregatorSecondTest {
+
+ @Test
+ public void testClusterSecondAggregatorWithInterpolation() {
+
+ long aggregatorInterval = 120000l;
+ long sliceInterval = 30000l;
+ long metricInterval = 10000l;
+
+ Configuration configuration = new Configuration();
+ TimelineMetricMetadataManager metricMetadataManagerMock = EasyMock.createNiceMock(TimelineMetricMetadataManager.class);
+
+ TimelineMetricClusterAggregatorSecond secondAggregator = new TimelineMetricClusterAggregatorSecond(
+ "TimelineClusterAggregatorSecond", metricMetadataManagerMock, null, configuration, null,
+ aggregatorInterval, 2, "false", "", "", aggregatorInterval, sliceInterval
+ );
+
+ secondAggregator.timeSliceIntervalMillis = sliceInterval;
+ long roundedEndTime = AbstractTimelineAggregator.getRoundedAggregateTimeMillis(aggregatorInterval);
+ long roundedStartTime = roundedEndTime - aggregatorInterval;
+ List<Long[]> timeSlices = secondAggregator.getTimeSlices(roundedStartTime ,
+ roundedEndTime);
+
+ TreeMap<Long, Double> metricValues = new TreeMap<Long, Double>();
+
+ long startTime = roundedEndTime - aggregatorInterval;
+
+ for (int i=1; startTime < roundedEndTime; i++) {
+ startTime += metricInterval;
+ if (i%6 == 1 || i%6 == 2) {
+ metricValues.put(startTime, (double)i);
+ }
+ }
+
+ TimelineMetric metric = new TimelineMetric();
+ metric.setMetricName("TestMetric");
+ metric.setHostName("TestHost");
+ metric.setAppId("TestAppId");
+ metric.setMetricValues(metricValues);
+
+ Map<TimelineClusterMetric, Double> timelineClusterMetricMap = secondAggregator.sliceFromTimelineMetric(metric, timeSlices);
+
+ TimelineClusterMetric timelineClusterMetric = new TimelineClusterMetric(metric.getMetricName(), metric.getAppId(),
+ metric.getInstanceId(), 0l, null);
+
+ timelineClusterMetric.setTimestamp(roundedStartTime + 2*sliceInterval);
+ Assert.assertTrue(timelineClusterMetricMap.containsKey(timelineClusterMetric));
+ Assert.assertEquals(timelineClusterMetricMap.get(timelineClusterMetric), 4.5);
+
+ timelineClusterMetric.setTimestamp(roundedStartTime + 4*sliceInterval);
+ Assert.assertTrue(timelineClusterMetricMap.containsKey(timelineClusterMetric));
+ Assert.assertEquals(timelineClusterMetricMap.get(timelineClusterMetric), 7.5);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java
index 6c8d153..573e7f9 100644
--- a/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java
+++ b/ambari-server/src/main/java/org/apache/ambari/server/upgrade/UpgradeCatalog222.java
@@ -97,6 +97,8 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
public static final String HBASE_RPC_TIMEOUT_PROPERTY = "hbase.rpc.timeout";
public static final String PHOENIX_QUERY_TIMEOUT_PROPERTY = "phoenix.query.timeoutMs";
public static final String PHOENIX_QUERY_KEEPALIVE_PROPERTY = "phoenix.query.keepAliveMs";
+ public static final String TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED
+ = "timeline.metrics.cluster.aggregator.interpolation.enabled";
public static final String AMS_SERVICE_NAME = "AMBARI_METRICS";
public static final String AMS_COLLECTOR_COMPONENT_NAME = "METRICS_COLLECTOR";
@@ -249,7 +251,7 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
}
- protected void updateHostRoleCommands() throws SQLException{
+ protected void updateHostRoleCommands() throws SQLException {
dbAccessor.createIndex("idx_hrc_status", "host_role_command", "status", "role");
}
@@ -312,7 +314,7 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
String newTtl = oldTtl;
if (isDistributed) {
if ("86400".equals(oldTtl)) {
- newTtl = String.valueOf(7 * 86400); // 7 days
+ newTtl = String.valueOf(3 * 86400); // 3 days
}
}
newProperties.put(PRECISION_TABLE_TTL_PROPERTY, newTtl);
@@ -343,6 +345,11 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
LOG.info("Setting value of " + CLUSTER_MINUTE_TABLE_TTL_PROPERTY + " : " + newTtl);
}
+ if (!amsSiteProperties.containsKey(TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED)) {
+ LOG.info("Add config " + TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED + " = true");
+ newProperties.put(TIMELINE_METRICS_CLUSTER_AGGREGATOR_INTERPOLATION_ENABLED, String.valueOf(true));
+ }
+
updateConfigurationPropertiesForCluster(cluster, AMS_SITE, newProperties, true, true);
}
@@ -514,7 +521,7 @@ public class UpgradeCatalog222 extends AbstractUpgradeCatalog {
*/
protected void updateUpgradeTable() throws AmbariException, SQLException {
dbAccessor.addColumn(UPGRADE_TABLE,
- new DBAccessor.DBColumnInfo(UPGRADE_SUSPENDED_COLUMN, Short.class, 1, 0, false));
+ new DBAccessor.DBColumnInfo(UPGRADE_SUSPENDED_COLUMN, Short.class, 1, 0, false));
}
}
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml
index b99d1f9..5c67368 100644
--- a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml
+++ b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-hbase-env.xml
@@ -25,6 +25,12 @@
<name>hbase_log_dir</name>
<value>/var/log/ambari-metrics-collector</value>
<description>Log Directories for HBase.</description>
+ <depends-on>
+ <property>
+ <type>ams-env</type>
+ <name>metrics_collector_log_dir</name>
+ </property>
+ </depends-on>
</property>
<property>
<name>hbase_pid_dir</name>
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml
index 7a8984a..f8b82fd 100644
--- a/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml
+++ b/ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml
@@ -334,7 +334,7 @@
<name>timeline.metrics.host.aggregator.ttl</name>
<value>86400</value>
<description>
- 1 minute resolution data purge interval in seconds. Default is 1 day for embedded mode and 7 days for Distributed mode.
+ 1 minute resolution data purge interval in seconds. Default is 1 day for embedded mode and 3 days for Distributed mode.
</description>
<depends-on>
<property>
@@ -559,5 +559,12 @@
Enable Initialization check for HBase tables during Metrics service startup.
</description>
</property>
+ <property>
+ <name>timeline.metrics.cluster.aggregator.interpolation.enabled</name>
+ <value>true</value>
+ <description>
+ Enable Linear interpolation for missing slices of data, while aggregating.
+ </description>
+ </property>
</configuration>
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py
index 61aa2ba..b8e2c45 100644
--- a/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py
+++ b/ambari-server/src/main/resources/stacks/HDP/2.0.6/services/stack_advisor.py
@@ -499,6 +499,12 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
putAmsSiteProperty("timeline.metrics.service.webapp.address", str(metric_collector_host) + ":6188")
+ log_dir = "/var/log/ambari-metrics-collector"
+ if "ams-env" in services["configurations"]:
+ if "metrics_collector_log_dir" in services["configurations"]["ams-env"]["properties"]:
+ log_dir = services["configurations"]["ams-env"]["properties"]["metrics_collector_log_dir"]
+ putHbaseEnvProperty("hbase_log_dir", log_dir)
+
defaultFs = 'file:///'
if "core-site" in services["configurations"] and \
"fs.defaultFS" in services["configurations"]["core-site"]["properties"]:
@@ -511,7 +517,7 @@ class HDP206StackAdvisor(DefaultStackAdvisor):
if operatingMode == "distributed":
putAmsSiteProperty("timeline.metrics.service.watcher.disabled", 'true')
- putAmsSiteProperty("timeline.metrics.host.aggregator.ttl", 604800)
+ putAmsSiteProperty("timeline.metrics.host.aggregator.ttl", 259200)
putAmsHbaseSiteProperty("hbase.cluster.distributed", 'true')
else:
putAmsSiteProperty("timeline.metrics.service.watcher.disabled", 'false')
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java b/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java
index 599ac3e..9a0d713 100644
--- a/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java
+++ b/ambari-server/src/test/java/org/apache/ambari/server/upgrade/UpgradeCatalog222Test.java
@@ -301,7 +301,7 @@ public class UpgradeCatalog222Test {
put("timeline.metrics.host.aggregator.daily.checkpointCutOffMultiplier", String.valueOf(2));
put("timeline.metrics.cluster.aggregator.daily.checkpointCutOffMultiplier", String.valueOf(2));
put("timeline.metrics.service.watcher.disabled", String.valueOf(false));
- put("timeline.metrics.host.aggregator.ttl", String.valueOf(7 * 86400));
+ put("timeline.metrics.host.aggregator.ttl", String.valueOf(3 * 86400));
put("timeline.metrics.host.aggregator.minute.ttl", String.valueOf(7 * 86400));
put("timeline.metrics.host.aggregator.hourly.ttl", String.valueOf(30 * 86400));
put("timeline.metrics.host.aggregator.daily.ttl", String.valueOf(365 * 86400));
@@ -311,6 +311,7 @@ public class UpgradeCatalog222Test {
put("timeline.metrics.cluster.aggregator.daily.ttl", String.valueOf(730 * 86400));
put("timeline.metrics.service.operation.mode", "distributed");
put("timeline.metrics.service.webapp.address", "host1:6188");
+ put("timeline.metrics.cluster.aggregator.interpolation.enabled", String.valueOf(true));
}
};
EasyMockSupport easyMockSupport = new EasyMockSupport();
http://git-wip-us.apache.org/repos/asf/ambari/blob/800f2753/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py
----------------------------------------------------------------------
diff --git a/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py b/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py
index 41da868..1176d45 100644
--- a/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py
+++ b/ambari-server/src/test/python/stacks/2.2/common/test_stack_advisor.py
@@ -2210,7 +2210,7 @@ class TestHDP22StackAdvisor(TestCase):
expected['ams-hbase-env']['properties']['hbase_regionserver_heapsize'] = '512'
expected["ams-hbase-env"]['properties']['hbase_master_xmn_size'] = '102'
expected['ams-hbase-env']['properties']['regionserver_xmn_size'] = '384'
- expected['ams-site']['properties']['timeline.metrics.host.aggregator.ttl'] = '604800'
+ expected['ams-site']['properties']['timeline.metrics.host.aggregator.ttl'] = '259200'
expected['ams-site']['properties']['timeline.metrics.service.watcher.disabled'] = 'true'
self.stackAdvisor.recommendAmsConfigurations(configurations, clusterData, services, hosts)
self.assertEquals(configurations, expected)