You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by jo...@apache.org on 2016/10/21 17:20:10 UTC
[22/30] ambari git commit: AMBARI-18593 : Provide ability to use
downsampling function on certain metrics like client side topN. (avijayan)
AMBARI-18593 : Provide ability to use downsampling function on certain metrics like client side topN. (avijayan)
Project: http://git-wip-us.apache.org/repos/asf/ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/ambari/commit/03f7b909
Tree: http://git-wip-us.apache.org/repos/asf/ambari/tree/03f7b909
Diff: http://git-wip-us.apache.org/repos/asf/ambari/diff/03f7b909
Branch: refs/heads/branch-feature-AMBARI-18456
Commit: 03f7b909517a4939199078911603ff6b2e4c99c8
Parents: 685e926
Author: Aravindan Vijayan <av...@hortonworks.com>
Authored: Thu Oct 20 14:00:14 2016 -0700
Committer: Aravindan Vijayan <av...@hortonworks.com>
Committed: Thu Oct 20 14:00:14 2016 -0700
----------------------------------------------------------------------
.../aggregators/AbstractTimelineAggregator.java | 130 ++++++++++++++++++-
.../timeline/aggregators/CustomDownSampler.java | 44 +++++++
.../timeline/aggregators/DownSamplerUtils.java | 118 +++++++++++++++++
.../timeline/aggregators/TopNDownSampler.java | 112 ++++++++++++++++
.../v2/TimelineMetricClusterAggregator.java | 2 +-
.../v2/TimelineMetricHostAggregator.java | 3 +-
.../timeline/query/PhoenixTransactSQL.java | 27 +++-
.../metrics/timeline/query/TopNCondition.java | 25 ++--
.../timeline/aggregators/DownSamplerTest.java | 95 ++++++++++++++
.../0.1.0/configuration/ams-site.xml | 27 ++++
10 files changed, 561 insertions(+), 22 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/AbstractTimelineAggregator.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/AbstractTimelineAggregator.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/AbstractTimelineAggregator.java
index 30f78e6..1096bc4 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/AbstractTimelineAggregator.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/AbstractTimelineAggregator.java
@@ -17,13 +17,16 @@
*/
package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.availability.AggregationTaskRunner;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.availability.AggregationTaskRunner.AGGREGATOR_NAME;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.availability.MetricCollectorHAController;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.Condition;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.EmptyCondition;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;
@@ -34,6 +37,9 @@ import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.AGGREGATOR_CHECKPOINT_DELAY;
import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.RESULTSET_FETCH_SIZE;
@@ -57,6 +63,8 @@ public abstract class AbstractTimelineAggregator implements TimelineMetricAggreg
protected String outputTableName;
protected Long nativeTimeRangeDelay;
protected AggregationTaskRunner taskRunner;
+ protected List<String> downsampleMetricPatterns;
+ protected List<CustomDownSampler> configuredDownSamplers;
// Explicitly name aggregators for logging needs
private final AGGREGATOR_NAME aggregatorName;
@@ -70,6 +78,8 @@ public abstract class AbstractTimelineAggregator implements TimelineMetricAggreg
this.checkpointDelayMillis = SECONDS.toMillis(metricsConf.getInt(AGGREGATOR_CHECKPOINT_DELAY, 120));
this.resultsetFetchSize = metricsConf.getInt(RESULTSET_FETCH_SIZE, 2000);
this.LOG = LoggerFactory.getLogger(ACTUAL_AGGREGATOR_NAMES.get(aggregatorName));
+ this.configuredDownSamplers = DownSamplerUtils.getDownSamplers(metricsConf);
+ this.downsampleMetricPatterns = DownSamplerUtils.getDownsampleMetricPatterns(metricsConf);
}
public AbstractTimelineAggregator(AGGREGATOR_NAME aggregatorName,
@@ -260,14 +270,15 @@ public abstract class AbstractTimelineAggregator implements TimelineMetricAggreg
if (condition.doUpdate()) {
int rows = stmt.executeUpdate();
conn.commit();
- LOG.info(rows + " row(s) updated.");
+ LOG.info(rows + " row(s) updated in aggregation.");
+
+ downsample(conn, startTime, endTime);
} else {
rs = stmt.executeQuery();
}
LOG.debug("Query returned @: " + new Date());
aggregate(rs, startTime, endTime);
- LOG.info("End aggregation cycle @ " + new Date());
} catch (SQLException | IOException e) {
LOG.error("Exception during aggregating metrics.", e);
@@ -304,6 +315,42 @@ public abstract class AbstractTimelineAggregator implements TimelineMetricAggreg
protected abstract void aggregate(ResultSet rs, long startTime, long endTime) throws IOException, SQLException;
+ protected void downsample(Connection conn, Long startTime, Long endTime) {
+
+ LOG.info("Checking for downsampling requests.");
+ if (CollectionUtils.isEmpty(configuredDownSamplers)) {
+ LOG.info("No downsamplers configured");
+ return;
+ }
+
+ // Generate UPSERT query prefix. UPSERT part of the query is needed on the Aggregator side.
+ // SELECT part of the query is provided by the downsampler.
+ String queryPrefix = PhoenixTransactSQL.DOWNSAMPLE_CLUSTER_METRIC_SQL_UPSERT_PREFIX;
+ if (outputTableName.contains("RECORD")) {
+ queryPrefix = PhoenixTransactSQL.DOWNSAMPLE_HOST_METRIC_SQL_UPSERT_PREFIX;
+ }
+ queryPrefix = String.format(queryPrefix, getQueryHint(startTime), outputTableName);
+
+ for (Iterator<CustomDownSampler> iterator = configuredDownSamplers.iterator(); iterator.hasNext();){
+ CustomDownSampler downSampler = iterator.next();
+
+ if (downSampler.validateConfigs()) {
+ EmptyCondition downSamplingCondition = new EmptyCondition();
+ downSamplingCondition.setDoUpdate(true);
+ List<String> stmts = downSampler.prepareDownSamplingStatement(startTime, endTime, tableName);
+ for (String stmt : stmts) {
+ downSamplingCondition.setStatement(queryPrefix + stmt);
+ runDownSamplerQuery(conn, downSamplingCondition);
+ }
+ } else {
+ LOG.warn("The following downsampler failed config validation : " + downSampler.getClass().getName() + "." +
+ "Removing it from downsamplers list.");
+ iterator.remove();
+ }
+ }
+
+ }
+
public Long getSleepIntervalMillis() {
return sleepIntervalMillis;
}
@@ -351,6 +398,85 @@ public abstract class AbstractTimelineAggregator implements TimelineMetricAggreg
}
/**
+ * Run 1 downsampler query.
+ * @param conn
+ * @param condition
+ */
+ private void runDownSamplerQuery(Connection conn, Condition condition) {
+
+ PreparedStatement stmt = null;
+ ResultSet rs = null;
+ LOG.info("Downsampling query : " + condition.getStatement());
+
+ try {
+ stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition);
+
+ LOG.debug("Downsampler Query issued...");
+ if (condition.doUpdate()) {
+ int rows = stmt.executeUpdate();
+ conn.commit();
+ LOG.info(rows + " row(s) updated in downsampling.");
+ } else {
+ rs = stmt.executeQuery();
+ }
+ LOG.debug("Downsampler Query returned ...");
+ LOG.info("End Downsampling cycle.");
+
+ } catch (SQLException e) {
+ LOG.error("Exception during downsampling metrics.", e);
+ } finally {
+ if (rs != null) {
+ try {
+ rs.close();
+ } catch (SQLException e) {
+ // Ignore
+ }
+ }
+ if (stmt != null) {
+ try {
+ stmt.close();
+ } catch (SQLException e) {
+ // Ignore
+ }
+ }
+ if (conn != null) {
+ try {
+ conn.close();
+ } catch (SQLException sql) {
+ // Ignore
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns the METRIC_NAME NOT LIKE clause if certain metrics or metric patterns are to be skipped
+ * since they will be downsampled.
+ * @return
+ */
+ protected String getDownsampledMetricSkipClause() {
+ if (CollectionUtils.isEmpty(this.downsampleMetricPatterns)) {
+ return StringUtils.EMPTY;
+ }
+
+ StringBuilder sb = new StringBuilder();
+
+ for (int i = 0; i < downsampleMetricPatterns.size(); i++) {
+ sb.append(" METRIC_NAME");
+ sb.append(" NOT");
+ sb.append(" LIKE ");
+ sb.append("'" + downsampleMetricPatterns.get(i) + "'");
+
+ if (i < downsampleMetricPatterns.size() - 1) {
+ sb.append(" AND ");
+ }
+ }
+
+ sb.append(" AND ");
+ return sb.toString();
+ }
+
+ /**
* Get @AGGREGATOR_TYPE based on the output table.
* This is solely used by the HAController to determine which lock to acquire.
*/
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/CustomDownSampler.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/CustomDownSampler.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/CustomDownSampler.java
new file mode 100644
index 0000000..cfd82dd
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/CustomDownSampler.java
@@ -0,0 +1,44 @@
+/**
+ * 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 java.util.List;
+
+/**
+ * Interface to add a custom downsampler.
+ * Each configured downsampler will be executed during an aggregation cycle.
+ */
+public interface CustomDownSampler {
+
+ /**
+ * Gatekeeper to check the configs. If this fails, the downsampling is not done.
+ * @return
+ */
+ public boolean validateConfigs();
+
+ /**
+ * Return the set of statements that needs to be executed for the downsampling.
+ * @param startTime
+ * @param endTime
+ * @param tableName
+ * @return
+ */
+ public List<String> prepareDownSamplingStatement(Long startTime, Long endTime, String tableName);
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerUtils.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerUtils.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerUtils.java
new file mode 100644
index 0000000..b320e47
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerUtils.java
@@ -0,0 +1,118 @@
+/**
+ * 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 org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * DownSampler utility class. Responsible for fetching downsampler configs from Metrics config, and determine if
+ * any downsamplers are configured.
+ */
+
+public class DownSamplerUtils {
+
+ public static final String downSamplerConfigPrefix = "timeline.metrics.downsampler.";
+ public static final String downSamplerMetricPatternsConfig = "metric.patterns";
+ public static final String topNDownSampler = "topn";
+ private static final Log LOG = LogFactory.getLog(DownSamplerUtils.class);
+
+
+
+ /**
+ * Get the list of metrics that are requested to be downsampled.
+ * @param configuration
+ * @return List of metric patterns/names that are to be downsampled.
+ */
+ public static List<String> getDownsampleMetricPatterns(Configuration configuration) {
+ Map<String, String> conf = configuration.getValByRegex(downSamplerConfigPrefix + "*");
+ List<String> metricPatterns = new ArrayList<>();
+ Set<String> keys = conf.keySet();
+ for (String key : keys) {
+ if (key.endsWith(downSamplerMetricPatternsConfig)) {
+ String patternString = conf.get(key);
+ String[] patterns = StringUtils.split(patternString, ",");
+ for (String pattern : patterns) {
+ if (StringUtils.isNotEmpty(pattern)) {
+ String trimmedPattern = pattern.trim();
+ metricPatterns.add(trimmedPattern);
+ }
+ }
+ }
+ }
+ return metricPatterns;
+ }
+
+ /**
+ * Get the list of downsamplers that are configured in ams-site
+ * Sample config
+ <name>timeline.metrics.downsampler.topn.metric.patterns</name>
+ <value>dfs.NNTopUserOpCounts.windowMs=60000.op%,dfs.NNTopUserOpCounts.windowMs=300000.op%</value>
+
+ <name>timeline.metrics.downsampler.topn.value</name>
+ <value>10</value>
+
+ <name>timeline.metrics.downsampler.topn.function</name>
+ <value>max</value>
+ * @param configuration
+ * @return
+ */
+ public static List<CustomDownSampler> getDownSamplers(Configuration configuration) {
+
+ Map<String,String> conf = configuration.getValByRegex(downSamplerConfigPrefix + "*");
+ List<CustomDownSampler> downSamplers = new ArrayList<>();
+ Set<String> keys = conf.keySet();
+
+ try {
+ for (String key : keys) {
+ if (key.startsWith(downSamplerConfigPrefix) && key.endsWith(downSamplerMetricPatternsConfig)) {
+ String type = key.split("\\.")[3];
+ CustomDownSampler downSampler = getDownSamplerByType(type, conf);
+ if (downSampler != null) {
+ downSamplers.add(downSampler);
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("Exception caught while parsing downsampler configs from ams-site : " + e.getMessage());
+ }
+ return downSamplers;
+ }
+
+ public static CustomDownSampler getDownSamplerByType(String type, Map<String, String> conf) {
+ if (type == null) {
+ return null;
+ }
+
+ if (StringUtils.isNotEmpty(type) && type.equalsIgnoreCase(topNDownSampler)) {
+ return TopNDownSampler.fromConfig(conf);
+ }
+
+ LOG.warn("Unknown downsampler requested : " + type);
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TopNDownSampler.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TopNDownSampler.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TopNDownSampler.java
new file mode 100644
index 0000000..59e4453
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/TopNDownSampler.java
@@ -0,0 +1,112 @@
+/**
+ * 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 org.apache.commons.lang3.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.metrics2.sink.timeline.TopNConfig;
+import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.TopNCondition;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.TOPN_DOWNSAMPLER_CLUSTER_METRIC_SELECT_SQL;
+import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.TOPN_DOWNSAMPLER_HOST_METRIC_SELECT_SQL;
+
+public class TopNDownSampler implements CustomDownSampler {
+
+ private TopNConfig topNConfig;
+ private static final Log LOG = LogFactory.getLog(TopNDownSampler.class);
+ protected String metricPatterns;
+
+ public static TopNDownSampler fromConfig(Map<String, String> conf) {
+ String metricPatterns = conf.get(DownSamplerUtils.downSamplerConfigPrefix + "topn." +
+ DownSamplerUtils.downSamplerMetricPatternsConfig);
+
+ String topNString = conf.get(DownSamplerUtils.downSamplerConfigPrefix + "topn.value");
+ Integer topNValue = topNString != null ? Integer.valueOf(topNString) : 10;
+ String topNFunction = conf.get(DownSamplerUtils.downSamplerConfigPrefix + "topn.function");
+
+ return new TopNDownSampler(new TopNConfig(topNValue, topNFunction, false), metricPatterns);
+ }
+
+ public TopNDownSampler(TopNConfig topNConfig, String metricPatterns) {
+ this.topNConfig = topNConfig;
+ this.metricPatterns = metricPatterns;
+ }
+
+ @Override
+ public boolean validateConfigs() {
+ if (topNConfig == null) {
+ return false;
+ }
+
+ if (topNConfig.getTopN() <= 0) {
+ return false;
+ }
+
+ if (StringUtils.isEmpty(metricPatterns)) {
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Prepare downsampling SELECT statement(s) used to determine the data to be written into the Aggregate table.
+ * @param startTime
+ * @param endTime
+ * @param tableName
+ * @return
+ */
+ @Override
+ public List<String> prepareDownSamplingStatement(Long startTime, Long endTime, String tableName) {
+ List<String> stmts = new ArrayList<>();
+
+ Function.ReadFunction readFunction = Function.ReadFunction.getFunction(topNConfig.getTopNFunction());
+ Function function = new Function(readFunction, null);
+ String columnSelect = TopNCondition.getColumnSelect(function);
+
+ List<String> metricPatternList = Arrays.asList(metricPatterns.split(","));
+
+ for (String metricPattern : metricPatternList) {
+ String metricPatternClause = "'" + metricPattern + "'";
+ //TODO : Need a better way to find out what kind of aggregation the current one is.
+ if (tableName.contains("RECORD")) {
+ stmts.add(String.format(TOPN_DOWNSAMPLER_HOST_METRIC_SELECT_SQL,
+ endTime, columnSelect, columnSelect, columnSelect, tableName, metricPatternClause,
+ startTime, endTime, columnSelect, topNConfig.getTopN()));
+ } else {
+ stmts.add(String.format(TOPN_DOWNSAMPLER_CLUSTER_METRIC_SELECT_SQL,
+ endTime, columnSelect, columnSelect, columnSelect, tableName, metricPatternClause,
+ startTime, endTime, columnSelect, topNConfig.getTopN()));
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("DownSampling Stmt: " + stmts.toString());
+ }
+
+ return stmts;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricClusterAggregator.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricClusterAggregator.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricClusterAggregator.java
index 6b49a75..c7b605f 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricClusterAggregator.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricClusterAggregator.java
@@ -75,7 +75,7 @@ public class TimelineMetricClusterAggregator extends AbstractTimelineAggregator
condition.setStatement(String.format(GET_AGGREGATED_APP_METRIC_GROUPBY_SQL,
getQueryHint(startTime), outputTableName, endTime, aggregateColumnName, tableName,
- startTime, endTime));
+ getDownsampledMetricSkipClause(), startTime, endTime));
if (LOG.isDebugEnabled()) {
LOG.debug("Condition: " + condition.toString());
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricHostAggregator.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricHostAggregator.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricHostAggregator.java
index a0d4d87..57a3034 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricHostAggregator.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/v2/TimelineMetricHostAggregator.java
@@ -64,7 +64,8 @@ public class TimelineMetricHostAggregator extends AbstractTimelineAggregator {
condition.setDoUpdate(true);
condition.setStatement(String.format(GET_AGGREGATED_HOST_METRIC_GROUPBY_SQL,
- getQueryHint(startTime), outputTableName, endTime, tableName, startTime, endTime));
+ getQueryHint(startTime), outputTableName, endTime, tableName,
+ getDownsampledMetricSkipClause(), startTime, endTime));
if (LOG.isDebugEnabled()) {
LOG.debug("Condition: " + condition.toString());
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixTransactSQL.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixTransactSQL.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixTransactSQL.java
index 177e444..bc218f2 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixTransactSQL.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixTransactSQL.java
@@ -26,10 +26,7 @@ import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
-import java.util.Arrays;
-import java.util.HashSet;
import java.util.List;
-import java.util.Set;
import java.util.concurrent.TimeUnit;
/**
@@ -317,10 +314,20 @@ public class PhoenixTransactSQL {
"METRIC_SUM, METRIC_COUNT, METRIC_MAX, METRIC_MIN) " +
"SELECT METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, %s AS SERVER_TIME, UNITS, " +
"SUM(METRIC_SUM), SUM(METRIC_COUNT), MAX(METRIC_MAX), MIN(METRIC_MIN) " +
- "FROM %s WHERE SERVER_TIME > %s AND SERVER_TIME <= %s " +
+ "FROM %s WHERE%s SERVER_TIME > %s AND SERVER_TIME <= %s " +
"GROUP BY METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, UNITS";
/**
+ * Downsample host metrics.
+ */
+ public static final String DOWNSAMPLE_HOST_METRIC_SQL_UPSERT_PREFIX = "UPSERT %s INTO %s (METRIC_NAME, HOSTNAME, " +
+ "APP_ID, INSTANCE_ID, SERVER_TIME, UNITS, METRIC_SUM, METRIC_COUNT, METRIC_MAX, METRIC_MIN) ";
+
+ public static final String TOPN_DOWNSAMPLER_HOST_METRIC_SELECT_SQL = "SELECT METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, " +
+ "%s AS SERVER_TIME, UNITS, %s, 1, %s, %s FROM %s WHERE METRIC_NAME LIKE %s AND SERVER_TIME > %s AND SERVER_TIME <= %s " +
+ "GROUP BY METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, UNITS ORDER BY %s DESC LIMIT %s";
+
+ /**
* Aggregate app metrics using a GROUP BY clause to take advantage of
* N - way parallel scan where N = number of regions.
*/
@@ -328,9 +335,19 @@ public class PhoenixTransactSQL {
"INTO %s (METRIC_NAME, APP_ID, INSTANCE_ID, SERVER_TIME, UNITS, " +
"METRIC_SUM, METRIC_COUNT, METRIC_MAX, METRIC_MIN) SELECT METRIC_NAME, APP_ID, " +
"INSTANCE_ID, %s AS SERVER_TIME, UNITS, ROUND(AVG(METRIC_SUM),2), ROUND(AVG(%s)), " +
- "MAX(METRIC_MAX), MIN(METRIC_MIN) FROM %s WHERE SERVER_TIME > %s AND " +
+ "MAX(METRIC_MAX), MIN(METRIC_MIN) FROM %s WHERE%s SERVER_TIME > %s AND " +
"SERVER_TIME <= %s GROUP BY METRIC_NAME, APP_ID, INSTANCE_ID, UNITS";
+ /**
+ * Downsample cluster metrics.
+ */
+ public static final String DOWNSAMPLE_CLUSTER_METRIC_SQL_UPSERT_PREFIX = "UPSERT %s INTO %s (METRIC_NAME, APP_ID, " +
+ "INSTANCE_ID, SERVER_TIME, UNITS, METRIC_SUM, METRIC_COUNT, METRIC_MAX, METRIC_MIN) ";
+
+ public static final String TOPN_DOWNSAMPLER_CLUSTER_METRIC_SELECT_SQL = "SELECT METRIC_NAME, APP_ID, INSTANCE_ID," +
+ " %s AS SERVER_TIME, UNITS, %s, 1, %s, %s FROM %s WHERE METRIC_NAME LIKE %s AND SERVER_TIME > %s AND SERVER_TIME <= %s " +
+ "GROUP BY METRIC_NAME, APP_ID, INSTANCE_ID, UNITS ORDER BY %s DESC LIMIT %s";
+
public static final String METRICS_RECORD_TABLE_NAME = "METRIC_RECORD";
public static final String CONTAINER_METRICS_TABLE_NAME = "CONTAINER_METRICS";
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/TopNCondition.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/TopNCondition.java b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/TopNCondition.java
index f7060e0..0f2a02c 100644
--- a/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/TopNCondition.java
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/TopNCondition.java
@@ -106,31 +106,30 @@ public class TopNCondition extends DefaultCondition{
private String getTopNOrderByClause() {
- String orderByClause = null;
+ String orderByClause = getColumnSelect(this.topNFunction);
+ orderByClause += (isBottomN ? " ASC" : " DESC");
+ return orderByClause;
+ }
+ public static String getColumnSelect(Function topNFunction) {
+ String columnSelect = null;
if (topNFunction != null) {
switch (topNFunction.getReadFunction()) {
case AVG:
- orderByClause = "ROUND(AVG(METRIC_SUM),2)";
+ columnSelect = "ROUND(AVG(METRIC_SUM),2)";
break;
case SUM:
- orderByClause = "SUM(METRIC_SUM)";
+ columnSelect = "SUM(METRIC_SUM)";
break;
default:
- orderByClause = "MAX(METRIC_MAX)";
+ columnSelect = "MAX(METRIC_MAX)";
break;
}
}
-
- if (orderByClause == null) {
- orderByClause = "MAX(METRIC_MAX)";
+ if (columnSelect == null) {
+ columnSelect = "MAX(METRIC_MAX)";
}
-
- if (!isBottomN) {
- orderByClause += " DESC";
- }
-
- return orderByClause;
+ return columnSelect;
}
public boolean isTopNHostCondition() {
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerTest.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerTest.java b/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerTest.java
new file mode 100644
index 0000000..a0fea40
--- /dev/null
+++ b/ambari-metrics/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/aggregators/DownSamplerTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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.junit.Test;
+
+import java.util.List;
+import java.util.Map;
+
+public class DownSamplerTest {
+
+ @Test
+ public void testGetDownsampleMetricPatterns() throws Exception {
+
+ Configuration configuration = new Configuration();
+ configuration.setIfUnset("timeline.metrics.downsampler.topn.metric.patterns", "pattern1,pattern2");
+ configuration.setIfUnset("timeline.metrics.downsampler.lastvalue.metric.patterns", "pattern3");
+
+ List<String> patterns = DownSamplerUtils.getDownsampleMetricPatterns(configuration);
+ Assert.assertEquals(patterns.size(), 3);
+ Assert.assertTrue(patterns.contains("pattern1"));
+ Assert.assertTrue(patterns.contains("pattern2"));
+ Assert.assertTrue(patterns.contains("pattern3"));
+
+ Configuration configuration2 = new Configuration();
+ patterns = DownSamplerUtils.getDownsampleMetricPatterns(configuration2);
+ Assert.assertEquals(patterns.size(), 0);
+ }
+
+ @Test
+ public void testGetDownSamplers() throws Exception {
+
+ Configuration configuration = new Configuration();
+ configuration.setIfUnset("timeline.metrics.downsampler.topn.metric.patterns", "pattern1,pattern2");
+ configuration.setIfUnset("timeline.metrics.downsampler.test.metric.patterns", "pattern3");
+
+ List<CustomDownSampler> downSamplers = DownSamplerUtils.getDownSamplers(configuration);
+ Assert.assertEquals(downSamplers.size(), 1);
+ Assert.assertTrue(downSamplers.get(0) instanceof TopNDownSampler);
+ }
+
+ @Test
+ public void testPrepareTopNDownSamplingStatement() throws Exception {
+ Configuration configuration = new Configuration();
+ configuration.setIfUnset("timeline.metrics.downsampler.topn.metric.patterns", "pattern1,pattern2");
+ configuration.setIfUnset("timeline.metrics.downsampler.topn.value", "3");
+
+ Map<String, String> conf = configuration.getValByRegex(DownSamplerUtils.downSamplerConfigPrefix);
+
+ TopNDownSampler topNDownSampler = TopNDownSampler.fromConfig(conf);
+ List<String> stmts = topNDownSampler.prepareDownSamplingStatement(14000000l, 14100000l, "METRIC_RECORD");
+ Assert.assertEquals(stmts.size(),2);
+ Assert.assertTrue(stmts.contains("SELECT METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, 14100000 AS SERVER_TIME, UNITS, " +
+ "MAX(METRIC_MAX), 1, MAX(METRIC_MAX), MAX(METRIC_MAX) FROM METRIC_RECORD WHERE " +
+ "METRIC_NAME LIKE 'pattern1' AND SERVER_TIME > 14000000 AND SERVER_TIME <= 14100000 " +
+ "GROUP BY METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, UNITS ORDER BY MAX(METRIC_MAX) DESC LIMIT 3"));
+
+ Assert.assertTrue(stmts.contains("SELECT METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, 14100000 AS SERVER_TIME, UNITS, " +
+ "MAX(METRIC_MAX), 1, MAX(METRIC_MAX), MAX(METRIC_MAX) FROM METRIC_RECORD WHERE " +
+ "METRIC_NAME LIKE 'pattern2' AND SERVER_TIME > 14000000 AND SERVER_TIME <= 14100000 " +
+ "GROUP BY METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, UNITS ORDER BY MAX(METRIC_MAX) DESC LIMIT 3"));
+
+ configuration.clear();
+ configuration.setIfUnset("timeline.metrics.downsampler.topn.metric.patterns", "pattern1");
+ configuration.setIfUnset("timeline.metrics.downsampler.topn.value", "4");
+ configuration.setIfUnset("timeline.metrics.downsampler.topn.function", "sum");
+ conf = configuration.getValByRegex(DownSamplerUtils.downSamplerConfigPrefix);
+ topNDownSampler = TopNDownSampler.fromConfig(conf);
+ stmts = topNDownSampler.prepareDownSamplingStatement(14000000l, 14100000l, "METRIC_AGGREGATE_MINUTE");
+ Assert.assertEquals(stmts.size(),1);
+
+ Assert.assertTrue(stmts.contains("SELECT METRIC_NAME, APP_ID, INSTANCE_ID, 14100000 AS SERVER_TIME, UNITS, " +
+ "SUM(METRIC_SUM), 1, SUM(METRIC_SUM), SUM(METRIC_SUM) FROM METRIC_AGGREGATE_MINUTE WHERE " +
+ "METRIC_NAME LIKE 'pattern1' AND SERVER_TIME > 14000000 AND SERVER_TIME <= 14100000 " +
+ "GROUP BY METRIC_NAME, APP_ID, INSTANCE_ID, UNITS ORDER BY SUM(METRIC_SUM) DESC LIMIT 4"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/ambari/blob/03f7b909/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 2f2e114..9235777 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
@@ -756,4 +756,31 @@
<empty-value-valid>true</empty-value-valid>
</value-attributes>
</property>
+ <property>
+ <name>timeline.metrics.downsampler.topn.metric.patterns</name>
+ <value>dfs.NNTopUserOpCounts.windowMs=60000.op=__%.user=%,dfs.NNTopUserOpCounts.windowMs=300000.op=__%.user=%,dfs.NNTopUserOpCounts.windowMs=1500000.op=__%.user=%</value>
+ <description>
+ Commas separated list of metric name regular expressions that are candidates for Top N downsampling.
+ </description>
+ <value-attributes>
+ <empty-value-valid>true</empty-value-valid>
+ </value-attributes>
+ <on-ambari-upgrade add="true"/>
+ </property>
+ <property>
+ <name>timeline.metrics.downsampler.topn.value</name>
+ <value>10</value>
+ <description>
+ Top N value for to be used for top N downsampling. Default is 10.
+ </description>
+ <on-ambari-upgrade add="true"/>
+ </property>
+ <property>
+ <name>timeline.metrics.downsampler.topn.function</name>
+ <value>max</value>
+ <description>
+ Top N function for to be used for top N downsampling (avg/max/sum)
+ </description>
+ <on-ambari-upgrade add="true"/>
+ </property>
</configuration>