You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by sw...@apache.org on 2014/12/01 21:03:44 UTC
[16/22] ambari git commit: AMBARI-5707. Renaming a module. (swagle)
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricHostAggregate.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricHostAggregate.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricHostAggregate.java
deleted file mode 100644
index 02cc207..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricHostAggregate.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/**
- * 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;
-
-
-import org.codehaus.jackson.annotate.JsonCreator;
-import org.codehaus.jackson.annotate.JsonProperty;
-
-/**
- * Represents a collection of minute based aggregation of values for
- * resolution greater than a minute.
- */
-public class MetricHostAggregate extends MetricAggregate {
-
- private long numberOfSamples = 0;
-
- @JsonCreator
- public MetricHostAggregate() {
- super(0.0, 0.0, Double.MIN_VALUE, Double.MAX_VALUE);
- }
-
- public MetricHostAggregate(Double sum, int numberOfSamples,
- Double deviation,
- Double max, Double min) {
- super(sum, deviation, max, min);
- this.numberOfSamples = numberOfSamples;
- }
-
- @JsonProperty("numberOfSamples")
- long getNumberOfSamples() {
- return numberOfSamples == 0 ? 1 : numberOfSamples;
- }
-
- void updateNumberOfSamples(long count) {
- this.numberOfSamples += count;
- }
-
- public void setNumberOfSamples(long numberOfSamples) {
- this.numberOfSamples = numberOfSamples;
- }
-
- public double getAvg() {
- return sum / numberOfSamples;
- }
-
- /**
- * Find and update min, max and avg for a minute
- */
- void updateAggregates(MetricHostAggregate hostAggregate) {
- updateMax(hostAggregate.getMax());
- updateMin(hostAggregate.getMin());
- updateSum(hostAggregate.getSum());
- updateNumberOfSamples(hostAggregate.getNumberOfSamples());
- }
-
- @Override
- public String toString() {
- return "MetricHostAggregate{" +
- "sum=" + sum +
- ", numberOfSamples=" + numberOfSamples +
- ", deviation=" + deviation +
- ", max=" + max +
- ", min=" + min +
- '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricsInitializationException.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricsInitializationException.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricsInitializationException.java
deleted file mode 100644
index 88a427a..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/MetricsInitializationException.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/**
- * 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;
-
-/**
- * RuntimeException for initialization of metrics schema. It is RuntimeException
- * since this is a not recoverable situation, and should be handled by main or
- * service method followed by shutdown.
- */
-public class MetricsInitializationException extends RuntimeException {
- public MetricsInitializationException() {
- }
-
- public MetricsInitializationException(String msg) {
- super(msg);
- }
-
- public MetricsInitializationException(Throwable t) {
- super(t);
- }
-
- public MetricsInitializationException(String msg, Throwable t) {
- super(msg, t);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java
deleted file mode 100644
index 4f248b7..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java
+++ /dev/null
@@ -1,678 +0,0 @@
-/**
- * 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;
-
-
-import org.apache.commons.logging.Log;
-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.TimelineMetric;
-import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
-import org.apache.hadoop.yarn.util.timeline.TimelineUtils;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.codehaus.jackson.type.TypeReference;
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.CREATE_METRICS_AGGREGATE_HOURLY_TABLE_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.CREATE_METRICS_AGGREGATE_MINUTE_TABLE_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.CREATE_METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.CREATE_METRICS_CLUSTER_AGGREGATE_TABLE_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.CREATE_METRICS_TABLE_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.Condition;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.DEFAULT_ENCODING;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.DEFAULT_TABLE_COMPRESSION;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.METRICS_RECORD_TABLE_NAME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.UPSERT_AGGREGATE_RECORD_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.UPSERT_CLUSTER_AGGREGATE_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.UPSERT_CLUSTER_AGGREGATE_TIME_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.UPSERT_METRICS_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_HOUR_TABLE_TTL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_MINUTE_TABLE_TTL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.GLOBAL_MAX_RETRIES;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.GLOBAL_RESULT_LIMIT;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.GLOBAL_RETRY_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HBASE_COMPRESSION_SCHEME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HBASE_ENCODING_SCHEME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_HOUR_TABLE_TTL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_MINUTE_TABLE_TTL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.PRECISION_TABLE_TTL;
-
-/**
- * Provides a facade over the Phoenix API to access HBase schema
- */
-public class PhoenixHBaseAccessor {
-
- private static final Log LOG = LogFactory.getLog(PhoenixHBaseAccessor.class);
- private final Configuration hbaseConf;
- private final Configuration metricsConf;
- private final RetryCounterFactory retryCounterFactory;
-
- static final int PHOENIX_MAX_MUTATION_STATE_SIZE = 50000;
- /**
- * 4 metrics/min * 60 * 24: Retrieve data for 1 day.
- */
- private static final int METRICS_PER_MINUTE = 4;
- public static int RESULTSET_LIMIT = (int)TimeUnit.DAYS.toMinutes(1) *
- METRICS_PER_MINUTE;
- private static ObjectMapper mapper = new ObjectMapper();
-
- private static TypeReference<Map<Long, Double>> metricValuesTypeRef =
- new TypeReference<Map<Long, Double>>() {};
- private final ConnectionProvider dataSource;
-
- public PhoenixHBaseAccessor(Configuration hbaseConf,
- Configuration metricsConf){
- this(hbaseConf, metricsConf, new DefaultPhoenixDataSource(hbaseConf));
- }
-
- public PhoenixHBaseAccessor(Configuration hbaseConf,
- Configuration metricsConf,
- ConnectionProvider dataSource) {
- this.hbaseConf = hbaseConf;
- this.metricsConf = metricsConf;
- RESULTSET_LIMIT = metricsConf.getInt(GLOBAL_RESULT_LIMIT, 5760);
- try {
- Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
- } catch (ClassNotFoundException e) {
- LOG.error("Phoenix client jar not found in the classpath.", e);
- throw new IllegalStateException(e);
- }
- this.dataSource = dataSource;
- this.retryCounterFactory = new RetryCounterFactory(
- metricsConf.getInt(GLOBAL_MAX_RETRIES, 10),
- (int) SECONDS.toMillis(metricsConf.getInt(GLOBAL_RETRY_INTERVAL, 5)));
- }
-
-
- private Connection getConnectionRetryingOnException()
- throws SQLException, InterruptedException {
- RetryCounter retryCounter = retryCounterFactory.create();
- while (true) {
- try{
- return getConnection();
- } catch (SQLException e) {
- if(!retryCounter.shouldRetry()){
- LOG.error("HBaseAccessor getConnection failed after "
- + retryCounter.getMaxAttempts() + " attempts");
- throw e;
- }
- }
- retryCounter.sleepUntilNextRetry();
- }
- }
-
-
- /**
- * Get JDBC connection to HBase store. Assumption is that the hbase
- * configuration is present on the classpath and loaded by the caller into
- * the Configuration object.
- * Phoenix already caches the HConnection between the client and HBase
- * cluster.
- *
- * @return @java.sql.Connection
- */
- public Connection getConnection() throws SQLException {
- return dataSource.getConnection();
- }
-
- public static Map readMetricFromJSON(String json) throws IOException {
- return mapper.readValue(json, metricValuesTypeRef);
- }
-
- @SuppressWarnings("unchecked")
- static TimelineMetric getTimelineMetricFromResultSet(ResultSet rs)
- throws SQLException, IOException {
- TimelineMetric metric = new TimelineMetric();
- metric.setMetricName(rs.getString("METRIC_NAME"));
- metric.setAppId(rs.getString("APP_ID"));
- metric.setInstanceId(rs.getString("INSTANCE_ID"));
- metric.setHostName(rs.getString("HOSTNAME"));
- metric.setTimestamp(rs.getLong("SERVER_TIME"));
- metric.setStartTime(rs.getLong("START_TIME"));
- metric.setType(rs.getString("UNITS"));
- metric.setMetricValues(
- (Map<Long, Double>) readMetricFromJSON(rs.getString("METRICS")));
- return metric;
- }
-
- static TimelineMetric getTimelineMetricKeyFromResultSet(ResultSet rs)
- throws SQLException, IOException {
- TimelineMetric metric = new TimelineMetric();
- metric.setMetricName(rs.getString("METRIC_NAME"));
- metric.setAppId(rs.getString("APP_ID"));
- metric.setInstanceId(rs.getString("INSTANCE_ID"));
- metric.setHostName(rs.getString("HOSTNAME"));
- metric.setTimestamp(rs.getLong("SERVER_TIME"));
- metric.setType(rs.getString("UNITS"));
- return metric;
- }
-
- static MetricHostAggregate getMetricHostAggregateFromResultSet(ResultSet rs)
- throws SQLException {
- MetricHostAggregate metricHostAggregate = new MetricHostAggregate();
- metricHostAggregate.setSum(rs.getDouble("METRIC_SUM"));
- metricHostAggregate.setMax(rs.getDouble("METRIC_MAX"));
- metricHostAggregate.setMin(rs.getDouble("METRIC_MIN"));
- metricHostAggregate.setNumberOfSamples(rs.getLong("METRIC_COUNT"));
-
- metricHostAggregate.setDeviation(0.0);
- return metricHostAggregate;
- }
-
- static TimelineClusterMetric
- getTimelineMetricClusterKeyFromResultSet(ResultSet rs)
- throws SQLException, IOException {
- TimelineClusterMetric metric = new TimelineClusterMetric(
- rs.getString("METRIC_NAME"),
- rs.getString("APP_ID"),
- rs.getString("INSTANCE_ID"),
- rs.getLong("SERVER_TIME"),
- rs.getString("UNITS"));
-
- return metric;
- }
-
- static MetricClusterAggregate
- getMetricClusterAggregateFromResultSet(ResultSet rs)
- throws SQLException {
- MetricClusterAggregate agg = new MetricClusterAggregate();
- agg.setSum(rs.getDouble("METRIC_SUM"));
- agg.setMax(rs.getDouble("METRIC_MAX"));
- agg.setMin(rs.getDouble("METRIC_MIN"));
- agg.setNumberOfHosts(rs.getInt("HOSTS_COUNT"));
-
- agg.setDeviation(0.0);
-
- return agg;
- }
-
- protected void initMetricSchema() {
- Connection conn = null;
- Statement stmt = null;
-
- String encoding = metricsConf.get(HBASE_ENCODING_SCHEME, DEFAULT_ENCODING);
- String compression = metricsConf.get(HBASE_COMPRESSION_SCHEME, DEFAULT_TABLE_COMPRESSION);
- String precisionTtl = metricsConf.get(PRECISION_TABLE_TTL, "86400");
- String hostMinTtl = metricsConf.get(HOST_MINUTE_TABLE_TTL, "604800");
- String hostHourTtl = metricsConf.get(HOST_HOUR_TABLE_TTL, "2592000");
- String clusterMinTtl = metricsConf.get(CLUSTER_MINUTE_TABLE_TTL, "2592000");
- String clusterHourTtl = metricsConf.get(CLUSTER_HOUR_TABLE_TTL, "31536000");
-
- try {
- LOG.info("Initializing metrics schema...");
- conn = getConnectionRetryingOnException();
- stmt = conn.createStatement();
-
- stmt.executeUpdate(String.format(CREATE_METRICS_TABLE_SQL,
- encoding, precisionTtl, compression));
- stmt.executeUpdate(String.format(CREATE_METRICS_AGGREGATE_HOURLY_TABLE_SQL,
- encoding, hostHourTtl, compression));
- stmt.executeUpdate(String.format(CREATE_METRICS_AGGREGATE_MINUTE_TABLE_SQL,
- encoding, hostMinTtl, compression));
- stmt.executeUpdate(String.format(CREATE_METRICS_CLUSTER_AGGREGATE_TABLE_SQL,
- encoding, clusterMinTtl, compression));
- stmt.executeUpdate(String.format(CREATE_METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_SQL,
- encoding, clusterHourTtl, compression));
- conn.commit();
- } catch (SQLException sql) {
- LOG.warn("Error creating Metrics Schema in HBase using Phoenix.", sql);
- throw new MetricsInitializationException(
- "Error creating Metrics Schema in HBase using Phoenix.", sql);
- } catch (InterruptedException e) {
- LOG.warn("Error creating Metrics Schema in HBase using Phoenix.", e);
- throw new MetricsInitializationException(
- "Error creating Metrics Schema in HBase using Phoenix.", e);
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- }
- }
-
- public void insertMetricRecords(TimelineMetrics metrics)
- throws SQLException, IOException {
-
- List<TimelineMetric> timelineMetrics = metrics.getMetrics();
- if (timelineMetrics == null || timelineMetrics.isEmpty()) {
- LOG.debug("Empty metrics insert request.");
- return;
- }
-
- Connection conn = getConnection();
- PreparedStatement metricRecordStmt = null;
- long currentTime = System.currentTimeMillis();
-
- try {
- metricRecordStmt = conn.prepareStatement(String.format(
- UPSERT_METRICS_SQL, METRICS_RECORD_TABLE_NAME));
-
- for (TimelineMetric metric : timelineMetrics) {
- metricRecordStmt.clearParameters();
-
- LOG.trace("host: " + metric.getHostName() + ", " +
- "metricName = " + metric.getMetricName() + ", " +
- "values: " + metric.getMetricValues());
- Aggregator agg = new Aggregator();
- double[] aggregates = agg.calculateAggregates(
- metric.getMetricValues());
-
- metricRecordStmt.setString(1, metric.getMetricName());
- metricRecordStmt.setString(2, metric.getHostName());
- metricRecordStmt.setString(3, metric.getAppId());
- metricRecordStmt.setString(4, metric.getInstanceId());
- metricRecordStmt.setLong(5, currentTime);
- metricRecordStmt.setLong(6, metric.getStartTime());
- metricRecordStmt.setString(7, metric.getType());
- metricRecordStmt.setDouble(8, aggregates[0]);
- metricRecordStmt.setDouble(9, aggregates[1]);
- metricRecordStmt.setDouble(10, aggregates[2]);
- metricRecordStmt.setLong(11, (long)aggregates[3]);
- String json =
- TimelineUtils.dumpTimelineRecordtoJSON(metric.getMetricValues());
- metricRecordStmt.setString(12, json);
-
- try {
- metricRecordStmt.executeUpdate();
- } catch (SQLException sql) {
- LOG.error(sql);
- }
- }
-
- conn.commit();
-
- } finally {
- if (metricRecordStmt != null) {
- try {
- metricRecordStmt.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException sql) {
- // Ignore
- }
- }
- }
- }
-
-
- @SuppressWarnings("unchecked")
- public TimelineMetrics getMetricRecords(final Condition condition)
- throws SQLException, IOException {
-
- if (condition.isEmpty()) {
- throw new SQLException("No filter criteria specified.");
- }
-
- Connection conn = getConnection();
- PreparedStatement stmt = null;
- TimelineMetrics metrics = new TimelineMetrics();
-
- try {
- stmt = PhoenixTransactSQL.prepareGetMetricsSqlStmt(conn, condition);
-
- ResultSet rs = stmt.executeQuery();
-
- while (rs.next()) {
- TimelineMetric metric = getTimelineMetricFromResultSet(rs);
-
- if (condition.isGrouped()) {
- metrics.addOrMergeTimelineMetric(metric);
- } else {
- metrics.getMetrics().add(metric);
- }
- }
-
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException sql) {
- // Ignore
- }
- }
- }
- return metrics;
- }
-
- public void saveHostAggregateRecords(Map<TimelineMetric,
- MetricHostAggregate> hostAggregateMap, String phoenixTableName)
- throws SQLException {
-
- if (hostAggregateMap != null && !hostAggregateMap.isEmpty()) {
- Connection conn = getConnection();
- PreparedStatement stmt = null;
-
- long start = System.currentTimeMillis();
- int rowCount = 0;
-
- try {
- stmt = conn.prepareStatement(
- String.format(UPSERT_AGGREGATE_RECORD_SQL, phoenixTableName));
-
- for (Map.Entry<TimelineMetric, MetricHostAggregate> metricAggregate :
- hostAggregateMap.entrySet()) {
-
- TimelineMetric metric = metricAggregate.getKey();
- MetricHostAggregate hostAggregate = metricAggregate.getValue();
-
- rowCount++;
- stmt.clearParameters();
- stmt.setString(1, metric.getMetricName());
- stmt.setString(2, metric.getHostName());
- stmt.setString(3, metric.getAppId());
- stmt.setString(4, metric.getInstanceId());
- stmt.setLong(5, metric.getTimestamp());
- stmt.setString(6, metric.getType());
- stmt.setDouble(7, hostAggregate.getSum());
- stmt.setDouble(8, hostAggregate.getMax());
- stmt.setDouble(9, hostAggregate.getMin());
- stmt.setDouble(10, hostAggregate.getNumberOfSamples());
-
- try {
- // TODO: Why this exception is swallowed
- stmt.executeUpdate();
- } catch (SQLException sql) {
- LOG.error(sql);
- }
-
- if (rowCount >= PHOENIX_MAX_MUTATION_STATE_SIZE - 1) {
- conn.commit();
- rowCount = 0;
- }
-
- }
-
- conn.commit();
-
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException sql) {
- // Ignore
- }
- }
- }
-
- long end = System.currentTimeMillis();
-
- if ((end - start) > 60000l) {
- LOG.info("Time to save map: " + (end - start) + ", " +
- "thread = " + Thread.currentThread().getClass());
- }
- }
- }
-
- /**
- * Save Metric aggregate records.
- *
- * @throws SQLException
- */
- public void saveClusterAggregateRecords(
- Map<TimelineClusterMetric, MetricClusterAggregate> records)
- throws SQLException {
-
- if (records == null || records.isEmpty()) {
- LOG.debug("Empty aggregate records.");
- return;
- }
-
- long start = System.currentTimeMillis();
-
- Connection conn = getConnection();
- PreparedStatement stmt = null;
- try {
- stmt = conn.prepareStatement(UPSERT_CLUSTER_AGGREGATE_SQL);
- int rowCount = 0;
-
- for (Map.Entry<TimelineClusterMetric, MetricClusterAggregate>
- aggregateEntry : records.entrySet()) {
- TimelineClusterMetric clusterMetric = aggregateEntry.getKey();
- MetricClusterAggregate aggregate = aggregateEntry.getValue();
-
- LOG.trace("clusterMetric = " + clusterMetric + ", " +
- "aggregate = " + aggregate);
-
- rowCount++;
- stmt.clearParameters();
- stmt.setString(1, clusterMetric.getMetricName());
- stmt.setString(2, clusterMetric.getAppId());
- stmt.setString(3, clusterMetric.getInstanceId());
- stmt.setLong(4, clusterMetric.getTimestamp());
- stmt.setString(5, clusterMetric.getType());
- stmt.setDouble(6, aggregate.getSum());
- stmt.setInt(7, aggregate.getNumberOfHosts());
- stmt.setDouble(8, aggregate.getMax());
- stmt.setDouble(9, aggregate.getMin());
-
- try {
- stmt.executeUpdate();
- } catch (SQLException sql) {
- // TODO: Why this exception is swallowed
- LOG.error(sql);
- }
-
- if (rowCount >= PHOENIX_MAX_MUTATION_STATE_SIZE - 1) {
- conn.commit();
- rowCount = 0;
- }
- }
-
- conn.commit();
-
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException sql) {
- // Ignore
- }
- }
- }
- long end = System.currentTimeMillis();
- if ((end - start) > 60000l) {
- LOG.info("Time to save: " + (end - start) + ", " +
- "thread = " + Thread.currentThread().getName());
- }
- }
-
- /**
- * Save Metric aggregate records.
- *
- * @throws SQLException
- */
- public void saveClusterAggregateHourlyRecords(
- Map<TimelineClusterMetric, MetricHostAggregate> records,
- String tableName)
- throws SQLException {
- if (records == null || records.isEmpty()) {
- LOG.debug("Empty aggregate records.");
- return;
- }
-
- long start = System.currentTimeMillis();
-
- Connection conn = getConnection();
- PreparedStatement stmt = null;
- try {
- stmt = conn.prepareStatement(String.format
- (UPSERT_CLUSTER_AGGREGATE_TIME_SQL, tableName));
- int rowCount = 0;
-
- for (Map.Entry<TimelineClusterMetric, MetricHostAggregate>
- aggregateEntry : records.entrySet()) {
- TimelineClusterMetric clusterMetric = aggregateEntry.getKey();
- MetricHostAggregate aggregate = aggregateEntry.getValue();
-
- LOG.trace("clusterMetric = " + clusterMetric + ", " +
- "aggregate = " + aggregate);
-
- rowCount++;
- stmt.clearParameters();
- stmt.setString(1, clusterMetric.getMetricName());
- stmt.setString(2, clusterMetric.getAppId());
- stmt.setString(3, clusterMetric.getInstanceId());
- stmt.setLong(4, clusterMetric.getTimestamp());
- stmt.setString(5, clusterMetric.getType());
- stmt.setDouble(6, aggregate.getSum());
-// stmt.setInt(7, aggregate.getNumberOfHosts());
- stmt.setLong(7, aggregate.getNumberOfSamples());
- stmt.setDouble(8, aggregate.getMax());
- stmt.setDouble(9, aggregate.getMin());
-
- try {
- stmt.executeUpdate();
- } catch (SQLException sql) {
- // we have no way to verify it works!!!
- LOG.error(sql);
- }
-
- if (rowCount >= PHOENIX_MAX_MUTATION_STATE_SIZE - 1) {
- conn.commit();
- rowCount = 0;
- }
- }
-
- conn.commit();
-
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException sql) {
- // Ignore
- }
- }
- }
- long end = System.currentTimeMillis();
- if ((end - start) > 60000l) {
- LOG.info("Time to save: " + (end - start) + ", " +
- "thread = " + Thread.currentThread().getName());
- }
- }
-
-
- public TimelineMetrics getAggregateMetricRecords(final Condition condition)
- throws SQLException {
-
- if (condition.isEmpty()) {
- throw new SQLException("No filter criteria specified.");
- }
-
- Connection conn = getConnection();
- PreparedStatement stmt = null;
- TimelineMetrics metrics = new TimelineMetrics();
-
- try {
- stmt = PhoenixTransactSQL.prepareGetAggregateSqlStmt(conn, condition);
-
- ResultSet rs = stmt.executeQuery();
-
- while (rs.next()) {
- TimelineMetric metric = new TimelineMetric();
- metric.setMetricName(rs.getString("METRIC_NAME"));
- metric.setAppId(rs.getString("APP_ID"));
- metric.setInstanceId(rs.getString("INSTANCE_ID"));
- metric.setTimestamp(rs.getLong("SERVER_TIME"));
- metric.setStartTime(rs.getLong("SERVER_TIME"));
- Map<Long, Double> valueMap = new HashMap<Long, Double>();
- valueMap.put(rs.getLong("SERVER_TIME"),
- rs.getDouble("METRIC_SUM") / rs.getInt("HOSTS_COUNT"));
- metric.setMetricValues(valueMap);
-
- if (condition.isGrouped()) {
- metrics.addOrMergeTimelineMetric(metric);
- } else {
- metrics.getMetrics().add(metric);
- }
- }
-
- } finally {
- if (stmt != null) {
- try {
- stmt.close();
- } catch (SQLException e) {
- // Ignore
- }
- }
- if (conn != null) {
- try {
- conn.close();
- } catch (SQLException sql) {
- // Ignore
- }
- }
- }
- LOG.info("Aggregate records size: " + metrics.getMetrics().size());
- return metrics;
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixTransactSQL.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixTransactSQL.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixTransactSQL.java
deleted file mode 100644
index 0d53f5f..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixTransactSQL.java
+++ /dev/null
@@ -1,528 +0,0 @@
-/**
- * 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;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Set;
-
-/**
- * Encapsulate all metrics related SQL queries.
- */
-public class PhoenixTransactSQL {
-
- static final Log LOG = LogFactory.getLog(PhoenixTransactSQL.class);
- // TODO: Configurable TTL values
- /**
- * Create table to store individual metric records.
- */
- public static final String CREATE_METRICS_TABLE_SQL = "CREATE TABLE IF NOT " +
- "EXISTS METRIC_RECORD (METRIC_NAME VARCHAR, " +
- "HOSTNAME VARCHAR, " +
- "SERVER_TIME UNSIGNED_LONG NOT NULL, " +
- "APP_ID VARCHAR, " +
- "INSTANCE_ID VARCHAR, " +
- "START_TIME UNSIGNED_LONG, " +
- "UNITS CHAR(20), " +
- "METRIC_SUM DOUBLE, " +
- "METRIC_COUNT UNSIGNED_INT, " +
- "METRIC_MAX DOUBLE, " +
- "METRIC_MIN DOUBLE, " +
- "METRICS VARCHAR CONSTRAINT pk " +
- "PRIMARY KEY (METRIC_NAME, HOSTNAME, SERVER_TIME, APP_ID, " +
- "INSTANCE_ID)) DATA_BLOCK_ENCODING='%s', IMMUTABLE_ROWS=true, " +
- "TTL=%s, COMPRESSION='%s'";
-
- public static final String CREATE_METRICS_AGGREGATE_HOURLY_TABLE_SQL =
- "CREATE TABLE IF NOT EXISTS METRIC_RECORD_HOURLY " +
- "(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_AGGREGATE_MINUTE_TABLE_SQL =
- "CREATE TABLE IF NOT EXISTS METRIC_RECORD_MINUTE " +
- "(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 METRIC_AGGREGATE " +
- "(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'";
-
- public static final String CREATE_METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_SQL =
- "CREATE TABLE IF NOT EXISTS METRIC_AGGREGATE_HOURLY " +
- "(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'";
-
- /**
- * Insert into metric records table.
- */
- public static final String UPSERT_METRICS_SQL = "UPSERT INTO %s " +
- "(METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, SERVER_TIME, START_TIME, " +
- "UNITS, " +
- "METRIC_SUM, " +
- "METRIC_MAX, " +
- "METRIC_MIN, " +
- "METRIC_COUNT, " +
- "METRICS) VALUES " +
- "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
-
- public static final String UPSERT_CLUSTER_AGGREGATE_SQL = "UPSERT INTO " +
- "METRIC_AGGREGATE (METRIC_NAME, APP_ID, INSTANCE_ID, SERVER_TIME, " +
- "UNITS, " +
- "METRIC_SUM, " +
- "HOSTS_COUNT, " +
- "METRIC_MAX, " +
- "METRIC_MIN) " +
- "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
-
- public static final String UPSERT_CLUSTER_AGGREGATE_TIME_SQL = "UPSERT INTO" +
- " %s (METRIC_NAME, APP_ID, INSTANCE_ID, SERVER_TIME, " +
- "UNITS, " +
- "METRIC_SUM, " +
- "METRIC_COUNT, " +
- "METRIC_MAX, " +
- "METRIC_MIN) " +
- "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)";
-
-
- public static final String UPSERT_AGGREGATE_RECORD_SQL = "UPSERT INTO " +
- "%s (METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, " +
- "SERVER_TIME, " +
- "UNITS, " +
- "METRIC_SUM, " +
- "METRIC_MAX, " +
- "METRIC_MIN," +
- "METRIC_COUNT) " +
- "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
-
- /**
- * Retrieve a set of rows from metrics records table.
- */
- public static final String GET_METRIC_SQL = "SELECT %s METRIC_NAME, " +
- "HOSTNAME, APP_ID, INSTANCE_ID, SERVER_TIME, START_TIME, UNITS, " +
- "METRIC_SUM, " +
- "METRIC_MAX, " +
- "METRIC_MIN, " +
- "METRIC_COUNT, " +
- "METRICS " +
- "FROM %s";
-
- public static final String GET_METRIC_AGGREGATE_ONLY_SQL = "SELECT %s " +
- "METRIC_NAME, HOSTNAME, APP_ID, INSTANCE_ID, SERVER_TIME, " +
- "UNITS, " +
- "METRIC_SUM, " +
- "METRIC_MAX, " +
- "METRIC_MIN, " +
- "METRIC_COUNT " +
- "FROM %s";
-
- public static final String GET_CLUSTER_AGGREGATE_SQL = "SELECT %s " +
- "METRIC_NAME, APP_ID, " +
- "INSTANCE_ID, SERVER_TIME, " +
- "UNITS, " +
- "METRIC_SUM, " +
- "HOSTS_COUNT, " +
- "METRIC_MAX, " +
- "METRIC_MIN " +
- "FROM METRIC_AGGREGATE";
-
- public static final String METRICS_RECORD_TABLE_NAME = "METRIC_RECORD";
- public static final String METRICS_AGGREGATE_MINUTE_TABLE_NAME =
- "METRIC_RECORD_MINUTE";
- public static final String METRICS_AGGREGATE_HOURLY_TABLE_NAME =
- "METRIC_RECORD_HOURLY";
- public static final String METRICS_CLUSTER_AGGREGATE_TABLE_NAME =
- "METRIC_AGGREGATE";
- public static final String METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME =
- "METRIC_AGGREGATE_HOURLY";
- public static final String DEFAULT_TABLE_COMPRESSION = "SNAPPY";
- public static final String DEFAULT_ENCODING = "FAST_DIFF";
- public static final long NATIVE_TIME_RANGE_DELTA = 120000; // 2 minutes
-
- /** Filter to optimize HBase scan by using file timestamps. This prevents
- * a full table scan of metric records.
- * @return Phoenix Hint String
- */
- public static String getNaiveTimeRangeHint(Long startTime, Long delta) {
- return String.format("/*+ NATIVE_TIME_RANGE(%s) */", (startTime - delta));
- }
-
- public static PreparedStatement prepareGetMetricsSqlStmt(
- Connection connection, Condition condition) throws SQLException {
-
- if (condition.isEmpty()) {
- throw new IllegalArgumentException("Condition is empty.");
- }
- String stmtStr;
- if (condition.getStatement() != null) {
- stmtStr = condition.getStatement();
- } else {
- stmtStr = String.format(GET_METRIC_SQL,
- getNaiveTimeRangeHint(condition.getStartTime(), NATIVE_TIME_RANGE_DELTA),
- METRICS_RECORD_TABLE_NAME);
- }
-
- StringBuilder sb = new StringBuilder(stmtStr);
- sb.append(" WHERE ");
- sb.append(condition.getConditionClause());
- String orderByClause = condition.getOrderByClause();
-
- if (orderByClause != null) {
- sb.append(orderByClause);
- } else {
- sb.append(" ORDER BY METRIC_NAME, SERVER_TIME ");
- }
- if (condition.getLimit() != null) {
- sb.append(" LIMIT ").append(condition.getLimit());
- }
-
- LOG.debug("SQL: " + sb.toString() + ", condition: " + condition);
- PreparedStatement stmt = connection.prepareStatement(sb.toString());
- int pos = 1;
- if (condition.getMetricNames() != null) {
- for (; pos <= condition.getMetricNames().size(); pos++) {
- LOG.debug("Setting pos: " + pos + ", value = " + condition.getMetricNames().get(pos - 1));
- stmt.setString(pos, condition.getMetricNames().get(pos - 1));
- }
- }
- if (condition.getHostname() != null) {
- LOG.debug("Setting pos: " + pos + ", value: " + condition.getHostname());
- stmt.setString(pos++, condition.getHostname());
- }
- // TODO: Upper case all strings on POST
- if (condition.getAppId() != null) {
- // TODO: fix case of appId coming from host metrics
- String appId = condition.getAppId();
- if (!condition.getAppId().equals("HOST")) {
- appId = appId.toLowerCase();
- }
- LOG.debug("Setting pos: " + pos + ", value: " + appId);
- stmt.setString(pos++, appId);
- }
- if (condition.getInstanceId() != null) {
- LOG.debug("Setting pos: " + pos + ", value: " + condition.getInstanceId());
- stmt.setString(pos++, condition.getInstanceId());
- }
- if (condition.getStartTime() != null) {
- LOG.debug("Setting pos: " + pos + ", value: " + condition.getStartTime());
- stmt.setLong(pos++, condition.getStartTime());
- }
- if (condition.getEndTime() != null) {
- LOG.debug("Setting pos: " + pos + ", value: " + condition.getEndTime());
- stmt.setLong(pos, condition.getEndTime());
- }
- if (condition.getFetchSize() != null) {
- stmt.setFetchSize(condition.getFetchSize());
- }
-
- return stmt;
- }
-
-
- public static PreparedStatement prepareGetAggregateSqlStmt(
- Connection connection, Condition condition) throws SQLException {
-
- if (condition.isEmpty()) {
- throw new IllegalArgumentException("Condition is empty.");
- }
-
- StringBuilder sb = new StringBuilder(GET_CLUSTER_AGGREGATE_SQL);
- sb.append(" WHERE ");
- sb.append(condition.getConditionClause());
- sb.append(" ORDER BY METRIC_NAME, SERVER_TIME");
- if (condition.getLimit() != null) {
- sb.append(" LIMIT ").append(condition.getLimit());
- }
-
- LOG.debug("SQL => " + sb.toString() + ", condition => " + condition);
- PreparedStatement stmt = connection.prepareStatement(sb.toString());
- int pos = 1;
- if (condition.getMetricNames() != null) {
- for (; pos <= condition.getMetricNames().size(); pos++) {
- stmt.setString(pos, condition.getMetricNames().get(pos - 1));
- }
- }
- // TODO: Upper case all strings on POST
- if (condition.getAppId() != null) {
- stmt.setString(pos++, condition.getAppId().toLowerCase());
- }
- if (condition.getInstanceId() != null) {
- stmt.setString(pos++, condition.getInstanceId());
- }
- if (condition.getStartTime() != null) {
- stmt.setLong(pos++, condition.getStartTime());
- }
- if (condition.getEndTime() != null) {
- stmt.setLong(pos, condition.getEndTime());
- }
-
- return stmt;
- }
-
- static class Condition {
- List<String> metricNames;
- String hostname;
- String appId;
- String instanceId;
- Long startTime;
- Long endTime;
- Integer limit;
- boolean grouped;
- boolean noLimit = false;
- Integer fetchSize;
- String statement;
- Set<String> orderByColumns = new LinkedHashSet<String>();
-
- Condition(List<String> metricNames, String hostname, String appId,
- String instanceId, Long startTime, Long endTime, Integer limit,
- boolean grouped) {
- this.metricNames = metricNames;
- this.hostname = hostname;
- this.appId = appId;
- this.instanceId = instanceId;
- this.startTime = startTime;
- this.endTime = endTime;
- this.limit = limit;
- this.grouped = grouped;
- }
-
- String getStatement() {
- return statement;
- }
-
- void setStatement(String statement) {
- this.statement = statement;
- }
-
- List<String> getMetricNames() {
- return metricNames == null || metricNames.isEmpty() ? null : metricNames;
- }
-
- String getMetricsClause() {
- StringBuilder sb = new StringBuilder("(");
- if (metricNames != null) {
- for (String name : metricNames) {
- if (sb.length() != 1) {
- sb.append(", ");
- }
- sb.append("?");
- }
- sb.append(")");
- return sb.toString();
- } else {
- return null;
- }
- }
-
- String getConditionClause() {
- StringBuilder sb = new StringBuilder();
- boolean appendConjunction = false;
-
- if (getMetricNames() != null) {
- sb.append("METRIC_NAME IN ");
- sb.append(getMetricsClause());
- appendConjunction = true;
- }
- if (appendConjunction) {
- sb.append(" AND");
- }
- appendConjunction = false;
- if (getHostname() != null) {
- sb.append(" HOSTNAME = ?");
- appendConjunction = true;
- }
- if (appendConjunction) {
- sb.append(" AND");
- }
- appendConjunction = false;
- if (getAppId() != null) {
- sb.append(" APP_ID = ?");
- appendConjunction = true;
- }
- if (appendConjunction) {
- sb.append(" AND");
- }
- appendConjunction = false;
- if (getInstanceId() != null) {
- sb.append(" INSTANCE_ID = ?");
- appendConjunction = true;
- }
- if (appendConjunction) {
- sb.append(" AND");
- }
- appendConjunction = false;
- if (getStartTime() != null) {
- sb.append(" SERVER_TIME >= ?");
- appendConjunction = true;
- }
- if (appendConjunction) {
- sb.append(" AND");
- }
- if (getEndTime() != null) {
- sb.append(" SERVER_TIME < ?");
- }
- return sb.toString();
- }
-
- String getHostname() {
- return hostname == null || hostname.isEmpty() ? null : hostname;
- }
-
- String getAppId() {
- return appId == null || appId.isEmpty() ? null : appId;
- }
-
- String getInstanceId() {
- return instanceId == null || instanceId.isEmpty() ? null : instanceId;
- }
-
- /**
- * Convert to millis.
- */
- Long getStartTime() {
- if (startTime < 9999999999l) {
- return startTime * 1000;
- } else {
- return startTime;
- }
- }
-
- Long getEndTime() {
- if (endTime < 9999999999l) {
- return endTime * 1000;
- } else {
- return endTime;
- }
- }
-
- void setNoLimit() {
- this.noLimit = true;
- }
-
- Integer getLimit() {
- if (noLimit) {
- return null;
- }
- return limit == null ? PhoenixHBaseAccessor.RESULTSET_LIMIT : limit;
- }
-
- boolean isGrouped() {
- return grouped;
- }
-
- boolean isEmpty() {
- return (metricNames == null || metricNames.isEmpty())
- && (hostname == null || hostname.isEmpty())
- && (appId == null || appId.isEmpty())
- && (instanceId == null || instanceId.isEmpty())
- && startTime == null
- && endTime == null;
- }
-
- Integer getFetchSize() {
- return fetchSize;
- }
-
- void setFetchSize(Integer fetchSize) {
- this.fetchSize = fetchSize;
- }
-
- void addOrderByColumn(String column) {
- orderByColumns.add(column);
- }
-
- String getOrderByClause() {
- String orderByStr = " ORDER BY ";
- if (!orderByColumns.isEmpty()) {
- StringBuilder sb = new StringBuilder(orderByStr);
- for (String orderByColumn : orderByColumns) {
- if (sb.length() != orderByStr.length()) {
- sb.append(", ");
- }
- sb.append(orderByColumn);
- }
- sb.append(" ");
- return sb.toString();
- }
- return null;
- }
-
- @Override
- public String toString() {
- return "Condition{" +
- "metricNames=" + metricNames +
- ", hostname='" + hostname + '\'' +
- ", appId='" + appId + '\'' +
- ", instanceId='" + instanceId + '\'' +
- ", startTime=" + startTime +
- ", endTime=" + endTime +
- ", limit=" + limit +
- ", grouped=" + grouped +
- ", orderBy=" + orderByColumns +
- ", noLimit=" + noLimit +
- '}';
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineClusterMetric.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineClusterMetric.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineClusterMetric.java
deleted file mode 100644
index d227993..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineClusterMetric.java
+++ /dev/null
@@ -1,97 +0,0 @@
-/**
- * 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;
-
-public class TimelineClusterMetric {
- private String metricName;
- private String appId;
- private String instanceId;
- private long timestamp;
- private String type;
-
- TimelineClusterMetric(String metricName, String appId, String instanceId,
- long timestamp, String type) {
- this.metricName = metricName;
- this.appId = appId;
- this.instanceId = instanceId;
- this.timestamp = timestamp;
- this.type = type;
- }
-
- String getMetricName() {
- return metricName;
- }
-
- String getAppId() {
- return appId;
- }
-
- String getInstanceId() {
- return instanceId;
- }
-
- long getTimestamp() {
- return timestamp;
- }
-
- String getType() { return type; }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (o == null || getClass() != o.getClass()) return false;
-
- TimelineClusterMetric that = (TimelineClusterMetric) o;
-
- if (timestamp != that.timestamp) return false;
- if (appId != null ? !appId.equals(that.appId) : that.appId != null)
- return false;
- if (instanceId != null ? !instanceId.equals(that.instanceId) : that.instanceId != null)
- return false;
- if (!metricName.equals(that.metricName)) return false;
-
- return true;
- }
-
- public boolean equalsExceptTime(TimelineClusterMetric metric) {
- if (!metricName.equals(metric.metricName)) return false;
- if (!appId.equals(metric.appId)) return false;
- if (instanceId != null ? !instanceId.equals(metric.instanceId) : metric.instanceId != null)
- return false;
-
- return true;
- }
- @Override
- public int hashCode() {
- int result = metricName.hashCode();
- result = 31 * result + (appId != null ? appId.hashCode() : 0);
- result = 31 * result + (instanceId != null ? instanceId.hashCode() : 0);
- result = 31 * result + (int) (timestamp ^ (timestamp >>> 32));
- return result;
- }
-
- @Override
- public String toString() {
- return "TimelineClusterMetric{" +
- "metricName='" + metricName + '\'' +
- ", appId='" + appId + '\'' +
- ", instanceId='" + instanceId + '\'' +
- ", timestamp=" + timestamp +
- '}';
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregator.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregator.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregator.java
deleted file mode 100644
index cab154b..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregator.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/**
- * 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;
-
-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.TimelineMetric;
-import java.io.IOException;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.HashMap;
-import java.util.Map;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.Condition;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.GET_METRIC_AGGREGATE_ONLY_SQL;
-
-public class TimelineMetricAggregator extends AbstractTimelineAggregator {
- private static final Log LOG = LogFactory.getLog
- (TimelineMetricAggregator.class);
-
- private final String checkpointLocation;
- private final Long sleepIntervalMillis;
- private final Integer checkpointCutOffMultiplier;
- private final String hostAggregatorDisabledParam;
- private final String tableName;
- private final String outputTableName;
- private final Long nativeTimeRangeDelay;
-
- public TimelineMetricAggregator(PhoenixHBaseAccessor hBaseAccessor,
- Configuration metricsConf,
- String checkpointLocation,
- Long sleepIntervalMillis,
- Integer checkpointCutOffMultiplier,
- String hostAggregatorDisabledParam,
- String tableName,
- String outputTableName,
- Long nativeTimeRangeDelay) {
- super(hBaseAccessor, metricsConf);
- this.checkpointLocation = checkpointLocation;
- this.sleepIntervalMillis = sleepIntervalMillis;
- this.checkpointCutOffMultiplier = checkpointCutOffMultiplier;
- this.hostAggregatorDisabledParam = hostAggregatorDisabledParam;
- this.tableName = tableName;
- this.outputTableName = outputTableName;
- this.nativeTimeRangeDelay = nativeTimeRangeDelay;
- }
-
- @Override
- protected String getCheckpointLocation() {
- return checkpointLocation;
- }
-
- @Override
- protected void aggregate(ResultSet rs, long startTime, long endTime)
- throws IOException, SQLException {
- Map<TimelineMetric, MetricHostAggregate> hostAggregateMap =
- aggregateMetricsFromResultSet(rs);
-
- LOG.info("Saving " + hostAggregateMap.size() + " metric aggregates.");
- hBaseAccessor.saveHostAggregateRecords(hostAggregateMap,
- outputTableName);
- }
-
- @Override
- protected Condition prepareMetricQueryCondition(long startTime, long endTime) {
- Condition condition = new Condition(null, null, null, null, startTime,
- endTime, null, true);
- condition.setNoLimit();
- condition.setFetchSize(resultsetFetchSize);
- condition.setStatement(String.format(GET_METRIC_AGGREGATE_ONLY_SQL,
- PhoenixTransactSQL.getNaiveTimeRangeHint(startTime, nativeTimeRangeDelay),
- tableName));
- condition.addOrderByColumn("METRIC_NAME");
- condition.addOrderByColumn("HOSTNAME");
- condition.addOrderByColumn("APP_ID");
- condition.addOrderByColumn("INSTANCE_ID");
- condition.addOrderByColumn("SERVER_TIME");
- return condition;
- }
-
- private Map<TimelineMetric, MetricHostAggregate> aggregateMetricsFromResultSet
- (ResultSet rs) throws IOException, SQLException {
- TimelineMetric existingMetric = null;
- MetricHostAggregate hostAggregate = null;
- Map<TimelineMetric, MetricHostAggregate> hostAggregateMap =
- new HashMap<TimelineMetric, MetricHostAggregate>();
-
- while (rs.next()) {
- TimelineMetric currentMetric =
- PhoenixHBaseAccessor.getTimelineMetricKeyFromResultSet(rs);
- MetricHostAggregate currentHostAggregate =
- PhoenixHBaseAccessor.getMetricHostAggregateFromResultSet(rs);
-
- if (existingMetric == null) {
- // First row
- existingMetric = currentMetric;
- hostAggregate = new MetricHostAggregate();
- hostAggregateMap.put(currentMetric, hostAggregate);
- }
-
- if (existingMetric.equalsExceptTime(currentMetric)) {
- // Recalculate totals with current metric
- hostAggregate.updateAggregates(currentHostAggregate);
- } else {
- // Switched over to a new metric - save existing - create new aggregate
- hostAggregate = new MetricHostAggregate();
- hostAggregate.updateAggregates(currentHostAggregate);
- hostAggregateMap.put(currentMetric, hostAggregate);
- existingMetric = currentMetric;
- }
- }
- return hostAggregateMap;
- }
-
- @Override
- protected Long getSleepIntervalMillis() {
- return sleepIntervalMillis;
- }
-
- @Override
- protected Integer getCheckpointCutOffMultiplier() {
- return checkpointCutOffMultiplier;
- }
-
- @Override
- protected boolean isDisabled() {
- return metricsConf.getBoolean(hostAggregatorDisabledParam, false);
- }
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregatorFactory.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregatorFactory.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregatorFactory.java
deleted file mode 100644
index 8b10079..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricAggregatorFactory.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/**
- * 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;
-
-import static java.util.concurrent.TimeUnit.SECONDS;
-import org.apache.commons.io.FilenameUtils;
-import org.apache.hadoop.conf.Configuration;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.METRICS_AGGREGATE_HOURLY_TABLE_NAME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.METRICS_AGGREGATE_MINUTE_TABLE_NAME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.METRICS_RECORD_TABLE_NAME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.DEFAULT_CHECKPOINT_LOCATION;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_HOUR_CHECKPOINT_CUTOFF_MULTIPLIER;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_HOUR_DISABLED;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_HOUR_SLEEP_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_MINUTE_CHECKPOINT_CUTOFF_MULTIPLIER;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_MINUTE_DISABLED;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.HOST_AGGREGATOR_MINUTE_SLEEP_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_AGGREGATOR_CHECKPOINT_DIR;
-
-/**
- *
- */
-public class TimelineMetricAggregatorFactory {
- private static final String MINUTE_AGGREGATE_CHECKPOINT_FILE =
- "timeline-metrics-host-aggregator-checkpoint";
- private static final String MINUTE_AGGREGATE_HOURLY_CHECKPOINT_FILE =
- "timeline-metrics-host-aggregator-hourly-checkpoint";
-
- public static TimelineMetricAggregator createTimelineMetricAggregatorMinute
- (PhoenixHBaseAccessor hBaseAccessor, Configuration metricsConf) {
-
- String checkpointDir = metricsConf.get(
- TIMELINE_METRICS_AGGREGATOR_CHECKPOINT_DIR, DEFAULT_CHECKPOINT_LOCATION);
- String checkpointLocation = FilenameUtils.concat(checkpointDir,
- MINUTE_AGGREGATE_CHECKPOINT_FILE);
- long sleepIntervalMillis = SECONDS.toMillis(metricsConf.getLong
- (HOST_AGGREGATOR_MINUTE_SLEEP_INTERVAL, 300l)); // 5 mins
-
- int checkpointCutOffMultiplier = metricsConf.getInt
- (HOST_AGGREGATOR_MINUTE_CHECKPOINT_CUTOFF_MULTIPLIER, 3);
- String hostAggregatorDisabledParam = HOST_AGGREGATOR_MINUTE_DISABLED;
-
- String inputTableName = METRICS_RECORD_TABLE_NAME;
- String outputTableName = METRICS_AGGREGATE_MINUTE_TABLE_NAME;
-
- return new TimelineMetricAggregator(hBaseAccessor, metricsConf,
- checkpointLocation,
- sleepIntervalMillis,
- checkpointCutOffMultiplier,
- hostAggregatorDisabledParam,
- inputTableName,
- outputTableName,
- 120000l);
- }
-
- public static TimelineMetricAggregator createTimelineMetricAggregatorHourly
- (PhoenixHBaseAccessor hBaseAccessor, Configuration metricsConf) {
-
- String checkpointDir = metricsConf.get(
- TIMELINE_METRICS_AGGREGATOR_CHECKPOINT_DIR, DEFAULT_CHECKPOINT_LOCATION);
- String checkpointLocation = FilenameUtils.concat(checkpointDir,
- MINUTE_AGGREGATE_HOURLY_CHECKPOINT_FILE);
- long sleepIntervalMillis = SECONDS.toMillis(metricsConf.getLong
- (HOST_AGGREGATOR_HOUR_SLEEP_INTERVAL, 3600l));
-
- int checkpointCutOffMultiplier = metricsConf.getInt
- (HOST_AGGREGATOR_HOUR_CHECKPOINT_CUTOFF_MULTIPLIER, 2);
- String hostAggregatorDisabledParam = HOST_AGGREGATOR_HOUR_DISABLED;
-
- String inputTableName = METRICS_AGGREGATE_MINUTE_TABLE_NAME;
- String outputTableName = METRICS_AGGREGATE_HOURLY_TABLE_NAME;
-
- return new TimelineMetricAggregator(hBaseAccessor, metricsConf,
- checkpointLocation,
- sleepIntervalMillis,
- checkpointCutOffMultiplier,
- hostAggregatorDisabledParam,
- inputTableName,
- outputTableName,
- 3600000l);
- }
-
-
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregator.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregator.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregator.java
deleted file mode 100644
index 654c188..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregator.java
+++ /dev/null
@@ -1,220 +0,0 @@
-/**
- * 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;
-
-
-import org.apache.commons.io.FilenameUtils;
-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.TimelineMetric;
-import java.io.IOException;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.Condition;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.GET_METRIC_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.METRICS_RECORD_TABLE_NAME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.NATIVE_TIME_RANGE_DELTA;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_MINUTE_CHECKPOINT_CUTOFF_MULTIPLIER;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_MINUTE_DISABLED;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_MINUTE_SLEEP_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.DEFAULT_CHECKPOINT_LOCATION;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_AGGREGATOR_CHECKPOINT_DIR;
-
-/**
- * Aggregates a metric across all hosts in the cluster. Reads metrics from
- * the precision table and saves into the aggregate.
- */
-public class TimelineMetricClusterAggregator extends AbstractTimelineAggregator {
- private static final Log LOG = LogFactory.getLog(TimelineMetricClusterAggregator.class);
- private static final String CLUSTER_AGGREGATOR_CHECKPOINT_FILE =
- "timeline-metrics-cluster-aggregator-checkpoint";
- private final String checkpointLocation;
- private final Long sleepIntervalMillis;
- public final int timeSliceIntervalMillis;
- private final Integer checkpointCutOffMultiplier;
-
- public TimelineMetricClusterAggregator(PhoenixHBaseAccessor hBaseAccessor,
- Configuration metricsConf) {
- super(hBaseAccessor, metricsConf);
-
- String checkpointDir = metricsConf.get(
- TIMELINE_METRICS_AGGREGATOR_CHECKPOINT_DIR, DEFAULT_CHECKPOINT_LOCATION);
-
- checkpointLocation = FilenameUtils.concat(checkpointDir,
- CLUSTER_AGGREGATOR_CHECKPOINT_FILE);
-
- sleepIntervalMillis = SECONDS.toMillis(metricsConf.getLong
- (CLUSTER_AGGREGATOR_MINUTE_SLEEP_INTERVAL, 120l));
- timeSliceIntervalMillis = (int)SECONDS.toMillis(metricsConf.getInt
- (CLUSTER_AGGREGATOR_TIMESLICE_INTERVAL, 15));
- checkpointCutOffMultiplier =
- metricsConf.getInt(CLUSTER_AGGREGATOR_MINUTE_CHECKPOINT_CUTOFF_MULTIPLIER, 2);
- }
-
- @Override
- protected String getCheckpointLocation() {
- return checkpointLocation;
- }
-
- @Override
- protected void aggregate(ResultSet rs, long startTime, long endTime)
- throws SQLException, IOException {
- List<Long[]> timeSlices = getTimeSlices(startTime, endTime);
- Map<TimelineClusterMetric, MetricClusterAggregate>
- aggregateClusterMetrics = aggregateMetricsFromResultSet(rs, timeSlices);
-
- LOG.info("Saving " + aggregateClusterMetrics.size() + " metric aggregates.");
- hBaseAccessor.saveClusterAggregateRecords(aggregateClusterMetrics);
- }
-
- @Override
- protected Condition prepareMetricQueryCondition(long startTime, long endTime) {
- Condition condition = new Condition(null, null, null, null, startTime,
- endTime, null, true);
- condition.setNoLimit();
- condition.setFetchSize(resultsetFetchSize);
- condition.setStatement(String.format(GET_METRIC_SQL,
- PhoenixTransactSQL.getNaiveTimeRangeHint(startTime, NATIVE_TIME_RANGE_DELTA),
- METRICS_RECORD_TABLE_NAME));
- condition.addOrderByColumn("METRIC_NAME");
- condition.addOrderByColumn("APP_ID");
- condition.addOrderByColumn("INSTANCE_ID");
- condition.addOrderByColumn("SERVER_TIME");
- return condition;
- }
-
- private List<Long[]> getTimeSlices(long startTime, long endTime) {
- List<Long[]> timeSlices = new ArrayList<Long[]>();
- long sliceStartTime = startTime;
- while (sliceStartTime < endTime) {
- timeSlices.add(new Long[] { sliceStartTime, sliceStartTime + timeSliceIntervalMillis});
- sliceStartTime += timeSliceIntervalMillis;
- }
- return timeSlices;
- }
-
- private Map<TimelineClusterMetric, MetricClusterAggregate>
- aggregateMetricsFromResultSet(ResultSet rs, List<Long[]> timeSlices)
- throws SQLException, IOException {
- Map<TimelineClusterMetric, MetricClusterAggregate> aggregateClusterMetrics =
- new HashMap<TimelineClusterMetric, MetricClusterAggregate>();
- // Create time slices
-
- while (rs.next()) {
- TimelineMetric metric =
- PhoenixHBaseAccessor.getTimelineMetricFromResultSet(rs);
-
- Map<TimelineClusterMetric, Double> clusterMetrics =
- sliceFromTimelineMetric(metric, timeSlices);
-
- if (clusterMetrics != null && !clusterMetrics.isEmpty()) {
- for (Map.Entry<TimelineClusterMetric, Double> clusterMetricEntry :
- clusterMetrics.entrySet()) {
- TimelineClusterMetric clusterMetric = clusterMetricEntry.getKey();
- MetricClusterAggregate aggregate = aggregateClusterMetrics.get(clusterMetric);
- Double avgValue = clusterMetricEntry.getValue();
-
- if (aggregate == null) {
- aggregate = new MetricClusterAggregate(avgValue, 1, null,
- avgValue, avgValue);
- aggregateClusterMetrics.put(clusterMetric, aggregate);
- } else {
- aggregate.updateSum(avgValue);
- aggregate.updateNumberOfHosts(1);
- aggregate.updateMax(avgValue);
- aggregate.updateMin(avgValue);
- }
- }
- }
- }
- return aggregateClusterMetrics;
- }
-
- @Override
- protected Long getSleepIntervalMillis() {
- return sleepIntervalMillis;
- }
-
- @Override
- protected Integer getCheckpointCutOffMultiplier() {
- return checkpointCutOffMultiplier;
- }
-
- @Override
- protected boolean isDisabled() {
- return metricsConf.getBoolean(CLUSTER_AGGREGATOR_MINUTE_DISABLED, false);
- }
-
- private Map<TimelineClusterMetric, Double> sliceFromTimelineMetric(
- TimelineMetric timelineMetric, List<Long[]> timeSlices) {
-
- if (timelineMetric.getMetricValues().isEmpty()) {
- return null;
- }
-
- Map<TimelineClusterMetric, Double> timelineClusterMetricMap =
- new HashMap<TimelineClusterMetric, Double>();
-
- for (Map.Entry<Long, Double> metric : timelineMetric.getMetricValues().entrySet()) {
- // TODO: investigate null values - pre filter
- if (metric.getValue() == null) {
- continue;
- }
- Long timestamp = getSliceTimeForMetric(timeSlices,
- Long.parseLong(metric.getKey().toString()));
- if (timestamp != -1) {
- // Metric is within desired time range
- TimelineClusterMetric clusterMetric = new TimelineClusterMetric(
- timelineMetric.getMetricName(),
- timelineMetric.getAppId(),
- timelineMetric.getInstanceId(),
- timestamp,
- timelineMetric.getType());
- if (!timelineClusterMetricMap.containsKey(clusterMetric)) {
- timelineClusterMetricMap.put(clusterMetric, metric.getValue());
- } else {
- Double oldValue = timelineClusterMetricMap.get(clusterMetric);
- Double newValue = (oldValue + metric.getValue()) / 2;
- timelineClusterMetricMap.put(clusterMetric, newValue);
- }
- }
- }
-
- return timelineClusterMetricMap;
- }
-
- /**
- * Return beginning of the time slice into which the metric fits.
- */
- private Long getSliceTimeForMetric(List<Long[]> timeSlices, Long timestamp) {
- for (Long[] timeSlice : timeSlices) {
- if (timestamp >= timeSlice[0] && timestamp < timeSlice[1]) {
- return timeSlice[0];
- }
- }
- return -1l;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/ambari/blob/ba3d6926/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregatorHourly.java
----------------------------------------------------------------------
diff --git a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregatorHourly.java b/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregatorHourly.java
deleted file mode 100644
index 7764ea3..0000000
--- a/ambari-metrics/ambari-metrics-hadoop-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/TimelineMetricClusterAggregatorHourly.java
+++ /dev/null
@@ -1,173 +0,0 @@
-/**
- * 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;
-
-import org.apache.commons.io.FilenameUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import java.io.IOException;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.util.HashMap;
-import java.util.Map;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor.getMetricClusterAggregateFromResultSet;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor.getTimelineMetricClusterKeyFromResultSet;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.Condition;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.GET_CLUSTER_AGGREGATE_SQL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_HOUR_CHECKPOINT_CUTOFF_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_HOUR_CHECKPOINT_CUTOFF_MULTIPLIER;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_HOUR_DISABLED;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.CLUSTER_AGGREGATOR_HOUR_SLEEP_INTERVAL;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.DEFAULT_CHECKPOINT_LOCATION;
-import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration.TIMELINE_METRICS_AGGREGATOR_CHECKPOINT_DIR;
-
-public class TimelineMetricClusterAggregatorHourly extends
- AbstractTimelineAggregator {
- private static final Log LOG = LogFactory.getLog
- (TimelineMetricClusterAggregatorHourly.class);
- private static final String CLUSTER_AGGREGATOR_HOURLY_CHECKPOINT_FILE =
- "timeline-metrics-cluster-aggregator-hourly-checkpoint";
- private final String checkpointLocation;
- private final long sleepIntervalMillis;
- private final Integer checkpointCutOffMultiplier;
- private long checkpointCutOffIntervalMillis;
- private static final Long NATIVE_TIME_RANGE_DELTA = 3600000l; // 1 hour
-
- public TimelineMetricClusterAggregatorHourly(
- PhoenixHBaseAccessor hBaseAccessor, Configuration metricsConf) {
- super(hBaseAccessor, metricsConf);
-
- String checkpointDir = metricsConf.get(
- TIMELINE_METRICS_AGGREGATOR_CHECKPOINT_DIR, DEFAULT_CHECKPOINT_LOCATION);
-
- checkpointLocation = FilenameUtils.concat(checkpointDir,
- CLUSTER_AGGREGATOR_HOURLY_CHECKPOINT_FILE);
-
- sleepIntervalMillis = SECONDS.toMillis(metricsConf.getLong
- (CLUSTER_AGGREGATOR_HOUR_SLEEP_INTERVAL, 3600l));
- checkpointCutOffIntervalMillis = SECONDS.toMillis(metricsConf.getLong
- (CLUSTER_AGGREGATOR_HOUR_CHECKPOINT_CUTOFF_INTERVAL, 7200l));
- checkpointCutOffMultiplier = metricsConf.getInt
- (CLUSTER_AGGREGATOR_HOUR_CHECKPOINT_CUTOFF_MULTIPLIER, 2);
- }
-
- @Override
- protected String getCheckpointLocation() {
- return checkpointLocation;
- }
-
- @Override
- protected void aggregate(ResultSet rs, long startTime, long endTime)
- throws SQLException, IOException {
- Map<TimelineClusterMetric, MetricHostAggregate> hostAggregateMap =
- aggregateMetricsFromResultSet(rs);
-
- LOG.info("Saving " + hostAggregateMap.size() + " metric aggregates.");
- hBaseAccessor.saveClusterAggregateHourlyRecords(hostAggregateMap,
- METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME);
- }
-
- @Override
- protected Condition prepareMetricQueryCondition(long startTime,
- long endTime) {
- Condition condition = new Condition(null, null, null, null, startTime,
- endTime, null, true);
- condition.setNoLimit();
- condition.setFetchSize(resultsetFetchSize);
- condition.setStatement(String.format(GET_CLUSTER_AGGREGATE_SQL,
- PhoenixTransactSQL.getNaiveTimeRangeHint(startTime, NATIVE_TIME_RANGE_DELTA)));
- condition.addOrderByColumn("METRIC_NAME");
- condition.addOrderByColumn("APP_ID");
- condition.addOrderByColumn("INSTANCE_ID");
- condition.addOrderByColumn("SERVER_TIME");
- return condition;
- }
-
- private Map<TimelineClusterMetric, MetricHostAggregate>
- aggregateMetricsFromResultSet(ResultSet rs) throws IOException, SQLException {
-
- TimelineClusterMetric existingMetric = null;
- MetricHostAggregate hostAggregate = null;
- Map<TimelineClusterMetric, MetricHostAggregate> hostAggregateMap =
- new HashMap<TimelineClusterMetric, MetricHostAggregate>();
-
- while (rs.next()) {
- TimelineClusterMetric currentMetric =
- getTimelineMetricClusterKeyFromResultSet(rs);
- MetricClusterAggregate currentHostAggregate =
- getMetricClusterAggregateFromResultSet(rs);
-
- if (existingMetric == null) {
- // First row
- existingMetric = currentMetric;
- hostAggregate = new MetricHostAggregate();
- hostAggregateMap.put(currentMetric, hostAggregate);
- }
-
- if (existingMetric.equalsExceptTime(currentMetric)) {
- // Recalculate totals with current metric
- updateAggregatesFromHost(hostAggregate, currentHostAggregate);
-
- } else {
- // Switched over to a new metric - save existing
- hostAggregate = new MetricHostAggregate();
- updateAggregatesFromHost(hostAggregate, currentHostAggregate);
- hostAggregateMap.put(currentMetric, hostAggregate);
- existingMetric = currentMetric;
- }
-
- }
-
- return hostAggregateMap;
- }
-
- private void updateAggregatesFromHost(
- MetricHostAggregate agg,
- MetricClusterAggregate currentClusterAggregate) {
- agg.updateMax(currentClusterAggregate.getMax());
- agg.updateMin(currentClusterAggregate.getMin());
- agg.updateSum(currentClusterAggregate.getSum());
- agg.updateNumberOfSamples(currentClusterAggregate.getNumberOfHosts());
- }
-
- @Override
- protected Long getSleepIntervalMillis() {
- return sleepIntervalMillis;
- }
-
- @Override
- protected Integer getCheckpointCutOffMultiplier() {
- return checkpointCutOffMultiplier;
- }
-
- @Override
- protected Long getCheckpointCutOffIntervalMillis() {
- return checkpointCutOffIntervalMillis;
- }
-
- @Override
- protected boolean isDisabled() {
- return metricsConf.getBoolean(CLUSTER_AGGREGATOR_HOUR_DISABLED, false);
- }
-
-
-}