AMBARI-22163 : Anomaly Storage: Design Metric anomalies schema. (avijayan)
diff --git a/ambari-metrics-anomaly-detection-service/pom.xml b/ambari-metrics-anomaly-detection-service/pom.xml
index 1a10f86..6f8f8c1 100644
--- a/ambari-metrics-anomaly-detection-service/pom.xml
+++ b/ambari-metrics-anomaly-detection-service/pom.xml
@@ -29,8 +29,9 @@
<artifactId>ambari-metrics-anomaly-detection-service</artifactId>
<version>2.0.0.0-SNAPSHOT</version>
<properties>
- <scala.version>2.10.4</scala.version>
+ <scala.version>2.11.1</scala.version>
<scala.binary.version>2.11</scala.binary.version>
+ <hadoop.version>2.7.3.2.6.0.3-8</hadoop.version>
</properties>
<repositories>
@@ -201,5 +202,35 @@
<version>2.1.1</version>
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-el</groupId>
+ <artifactId>commons-el</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-runtime</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-compiler</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.mortbay.jetty</groupId>
+ <artifactId>jsp-2.1-jetty</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.scalatest</groupId>
+ <artifactId>scalatest_2.11</artifactId>
+ <version>3.0.1</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git a/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/common/ADServiceConfiguration.scala b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/common/ADServiceConfiguration.scala
new file mode 100644
index 0000000..248c74e
--- /dev/null
+++ b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/common/ADServiceConfiguration.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.ambari.metrics.adservice.common
+
+import java.net.{MalformedURLException, URISyntaxException}
+
+import org.apache.hadoop.conf.Configuration
+
+object ADServiceConfiguration {
+
+ private val AMS_AD_SITE_CONFIGURATION_FILE = "ams-ad-site.xml"
+ private val HBASE_SITE_CONFIGURATION_FILE = "hbase-site.xml"
+
+ val ANOMALY_METRICS_TTL = "timeline.metrics.anomaly.data.ttl"
+
+ private var hbaseConf: org.apache.hadoop.conf.Configuration = _
+ private var adConf: org.apache.hadoop.conf.Configuration = _
+
+ def initConfigs(): Unit = {
+
+ var classLoader: ClassLoader = Thread.currentThread.getContextClassLoader
+ if (classLoader == null) classLoader = getClass.getClassLoader
+
+ try {
+ val hbaseResUrl = classLoader.getResource(HBASE_SITE_CONFIGURATION_FILE)
+ if (hbaseResUrl == null) throw new IllegalStateException("Unable to initialize the AD subsystem. No hbase-site present in the classpath.")
+
+ hbaseConf = new Configuration(true)
+ hbaseConf.addResource(hbaseResUrl.toURI.toURL)
+
+ val adSystemConfigUrl = classLoader.getResource(AMS_AD_SITE_CONFIGURATION_FILE)
+ if (adSystemConfigUrl == null) throw new IllegalStateException("Unable to initialize the AD subsystem. No ams-ad-site present in the classpath")
+
+ adConf = new Configuration(true)
+ adConf.addResource(adSystemConfigUrl.toURI.toURL)
+
+ } catch {
+ case me : MalformedURLException => println("MalformedURLException")
+ case ue : URISyntaxException => println("URISyntaxException")
+ }
+ }
+
+ def getHBaseConf: org.apache.hadoop.conf.Configuration = {
+ hbaseConf
+ }
+
+ def getAdConf: org.apache.hadoop.conf.Configuration = {
+ adConf
+ }
+
+ def getAnomalyDataTtl: Int = {
+ if (adConf != null) return adConf.get(ANOMALY_METRICS_TTL, "604800").toInt
+ 604800
+ }
+
+ /**
+ * ttl
+ *
+ */
+}
diff --git a/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/common/PhoenixQueryConstants.scala b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/common/PhoenixQueryConstants.scala
new file mode 100644
index 0000000..5e90d2b
--- /dev/null
+++ b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/common/PhoenixQueryConstants.scala
@@ -0,0 +1,109 @@
+/*
+ * 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.ambari.metrics.adservice.common
+
+object PhoenixQueryConstants {
+
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+ /* Table Name constants */
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+ val METRIC_PROFILE_TABLE_NAME = "METRIC_PROFILE"
+ val METHOD_PARAMETERS_TABLE_NAME = "METHOD_PARAMETERS"
+ val PIT_ANOMALY_METRICS_TABLE_NAME = "PIT_METRIC_ANOMALIES"
+ val TREND_ANOMALY_METRICS_TABLE_NAME = "TREND_METRIC_ANOMALIES"
+ val MODEL_SNAPSHOT = "MODEL_SNAPSHOT"
+
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+ /* CREATE statement constants */
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+ val CREATE_METRIC_PROFILE_TABLE = ""
+
+ val CREATE_METHOD_PARAMETERS_TABLE: String = "CREATE TABLE IF NOT EXISTS %s (" +
+ "METHOD_NAME VARCHAR, " +
+ "METHOD_TYPE VARCHAR, " +
+ "PARAMETERS VARCHAR " +
+ "CONSTRAINT pk PRIMARY KEY (METHOD_NAME)) " +
+ "DATA_BLOCK_ENCODING='FAST_DIFF', IMMUTABLE_ROWS=true, COMPRESSION='SNAPPY'"
+
+ val CREATE_PIT_ANOMALY_METRICS_TABLE_SQL: String = "CREATE TABLE IF NOT EXISTS %s (" +
+ "METRIC_UUID BINARY(20) NOT NULL, " +
+ "METHOD_NAME VARCHAR, " +
+ "ANOMALY_TIMESTAMP UNSIGNED_LONG NOT NULL, " +
+ "METRIC_VALUE DOUBLE, " +
+ "SEASONAL_INFO VARCHAR, " +
+ "ANOMALY_SCORE DOUBLE, " +
+ "MODEL_SNAPSHOT VARCHAR, " +
+ "DETECTION_TIME UNSIGNED_LONG " +
+ "CONSTRAINT pk PRIMARY KEY (METRIC_UUID, METHOD_NAME, ANOMALY_TIMESTAMP)) " +
+ "DATA_BLOCK_ENCODING='FAST_DIFF', IMMUTABLE_ROWS=true, TTL=%s, COMPRESSION='SNAPPY'"
+
+ val CREATE_TREND_ANOMALY_METRICS_TABLE_SQL: String = "CREATE TABLE IF NOT EXISTS %s (" +
+ "METRIC_UUID BINARY(20) NOT NULL, " +
+ "ANOMALY_PERIOD_START UNSIGNED_LONG NOT NULL, " +
+ "ANOMALY_PERIOD_END UNSIGNED_LONG NOT NULL, " +
+ "TEST_PERIOD_START UNSIGNED_LONG NOT NULL, " +
+ "TEST_PERIOD_END UNSIGNED_LONG NOT NULL, " +
+ "METHOD_NAME VARCHAR, " +
+ "ANOMALY_SCORE DOUBLE, " +
+ "MODEL_SNAPSHOT VARCHAR, " +
+ "DETECTION_TIME UNSIGNED_LONG " +
+ "CONSTRAINT pk PRIMARY KEY (METRIC_UUID, METHOD_NAME, ANOMALY_PERIOD_START, ANOMALY_PERIOD_END, TEST_PERIOD_START, TEST_PERIOD_END)) " +
+ "DATA_BLOCK_ENCODING='FAST_DIFF' IMMUTABLE_ROWS=true, TTL=%s, COMPRESSION='SNAPPY'"
+
+ val CREATE_MODEL_SNAPSHOT_TABLE: String = "CREATE TABLE IF NOT EXISTS %s (" +
+ "METRIC_UUID BINARY(20), " +
+ "METHOD_NAME VARCHAR, " +
+ "METHOD_TYPE VARCHAR, " +
+ "PARAMETERS VARCHAR " +
+ "SNAPSHOT_TIME UNSIGNED LONG NOT NULL "
+ "CONSTRAINT pk PRIMARY KEY (METRIC_UUID, METHOD_NAME)) " +
+ "DATA_BLOCK_ENCODING='FAST_DIFF', IMMUTABLE_ROWS=true, COMPRESSION='SNAPPY'"
+
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+ /* UPSERT statement constants */
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+ val UPSERT_METHOD_PARAMETERS_SQL: String = "UPSERT INTO %s (METHOD_NAME, METHOD_TYPE, PARAMETERS) VALUES (?,?,?)"
+
+ val UPSERT_PIT_ANOMALY_METRICS_SQL: String = "UPSERT INTO %s (METRIC_UUID, ANOMALY_TIMESTAMP, METRIC_VALUE, METHOD_NAME, " +
+ "SEASONAL_INFO, ANOMALY_SCORE, MODEL_SNAPSHOT, DETECTION_TIME) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+
+ val UPSERT_TREND_ANOMALY_METRICS_SQL: String = "UPSERT INTO %s (METRIC_UUID, ANOMALY_PERIOD_START, ANOMALY_PERIOD_END, " +
+ "TEST_PERIOD_START, TEST_PERIOD_END, METHOD_NAME, ANOMALY_SCORE, MODEL_SNAPSHOT, DETECTION_TIME) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+
+ val UPSERT_MODEL_SNAPSHOT_SQL: String = "UPSERT INTO %s (METRIC_UUID, METHOD_NAME, METHOD_TYPE, PARAMETERS) VALUES (?, ?, ?, ?)"
+
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+ /* GET statement constants */
+ //////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
+
+ val GET_METHOD_PAREMETERS_SQL: String = "SELECT METHOD_NAME, METHOD_TYPE, PARAMETERS FROM %s WHERE METHOD_NAME = %s"
+
+ val GET_PIT_ANOMALY_METRIC_SQL: String = "SELECT METRIC_UUID, ANOMALY_TIMESTAMP, METRIC_VALUE, METHOD_NAME, SEASONAL_INFO, " +
+ "ANOMALY_SCORE, MODEL_SNAPSHOT, DETECTION_TIME FROM %s WHERE METRIC_METRIC_UUID = ? AND ANOMALY_TIMESTAMP > ? AND ANOMALY_TIMESTAMP <= ? " +
+ "ORDER BY ANOMALY_SCORE DESC"
+
+ val GET_TREND_ANOMALY_METRIC_SQL: String = "SELECT METRIC_METRIC_UUID, ANOMALY_PERIOD_START, ANOMALY_PERIOD_END, TEST_PERIOD_START, " +
+ "ANOMALY_PERIOD_START, METHOD_NAME, ANOMALY_SCORE, MODEL_SNAPSHOT, DETECTION_TIME FROM %s WHERE METHOD = ? AND ANOMALY_PERIOD_END > ? " +
+ "AND TEST_END_TIME <= ? ORDER BY ANOMALY_SCORE DESC"
+
+ val GET_MODEL_SNAPSHOT_SQL: String = "SELECT METRIC_UUID, METHOD_NAME, METHOD_TYPE, PARAMETERS FROM %s WHERE UUID = %s AND METHOD_NAME = %s"
+
+}
diff --git a/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/db/PhoenixAnomalyStoreAccessor.scala b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/db/PhoenixAnomalyStoreAccessor.scala
new file mode 100644
index 0000000..6f33e56
--- /dev/null
+++ b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/db/PhoenixAnomalyStoreAccessor.scala
@@ -0,0 +1,67 @@
+/*
+ * 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.ambari.metrics.adservice.db
+
+import java.sql.{Connection, SQLException}
+
+import org.apache.ambari.metrics.adservice.common.{ADServiceConfiguration, PhoenixQueryConstants}
+import org.apache.hadoop.hbase.util.RetryCounterFactory
+import org.apache.hadoop.metrics2.sink.timeline.query.{DefaultPhoenixDataSource, PhoenixConnectionProvider}
+import java.util.concurrent.TimeUnit.SECONDS
+
+object PhoenixAnomalyStoreAccessor {
+
+ private var datasource: PhoenixConnectionProvider = _
+
+ def initAnomalyMetricSchema(): Unit = {
+
+ val datasource: PhoenixConnectionProvider = new DefaultPhoenixDataSource(ADServiceConfiguration.getHBaseConf)
+ val retryCounterFactory = new RetryCounterFactory(10, SECONDS.toMillis(3).toInt)
+
+ val ttl = ADServiceConfiguration.getAnomalyDataTtl
+ try {
+ var conn = datasource.getConnectionRetryingOnException(retryCounterFactory)
+ var stmt = conn.createStatement
+
+ val methodParametersSql = String.format(PhoenixQueryConstants.CREATE_METHOD_PARAMETERS_TABLE,
+ PhoenixQueryConstants.METHOD_PARAMETERS_TABLE_NAME)
+ stmt.executeUpdate(methodParametersSql)
+
+ val pointInTimeAnomalySql = String.format(PhoenixQueryConstants.CREATE_PIT_ANOMALY_METRICS_TABLE_SQL,
+ PhoenixQueryConstants.PIT_ANOMALY_METRICS_TABLE_NAME,
+ ttl.asInstanceOf[Object])
+ stmt.executeUpdate(pointInTimeAnomalySql)
+
+ val trendAnomalySql = String.format(PhoenixQueryConstants.CREATE_TREND_ANOMALY_METRICS_TABLE_SQL,
+ PhoenixQueryConstants.TREND_ANOMALY_METRICS_TABLE_NAME,
+ ttl.asInstanceOf[Object])
+ stmt.executeUpdate(trendAnomalySql)
+
+ val snapshotSql = String.format(PhoenixQueryConstants.CREATE_MODEL_SNAPSHOT_TABLE,
+ PhoenixQueryConstants.MODEL_SNAPSHOT)
+ stmt.executeUpdate(snapshotSql)
+
+ conn.commit()
+ } catch {
+ case e: SQLException => throw e
+ }
+ }
+
+ @throws[SQLException]
+ def getConnection: Connection = datasource.getConnection
+}
diff --git a/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/spark/prototype/SparkPhoenixReader.scala b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/spark/prototype/SparkPhoenixReader.scala
index 6e1ae07..ac00764 100644
--- a/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/spark/prototype/SparkPhoenixReader.scala
+++ b/ambari-metrics-anomaly-detection-service/src/main/scala/org/apache/ambari/metrics/adservice/spark/prototype/SparkPhoenixReader.scala
@@ -26,52 +26,52 @@
def main(args: Array[String]) {
- if (args.length < 6) {
- System.err.println("Usage: SparkPhoenixReader <metric_name> <appId> <hostname> <weight> <timessdev> <phoenixConnectionString> <model_dir>")
- System.exit(1)
- }
-
- var metricName = args(0)
- var appId = args(1)
- var hostname = args(2)
- var weight = args(3).toDouble
- var timessdev = args(4).toInt
- var phoenixConnectionString = args(5) //avijayan-ams-3.openstacklocal:61181:/ams-hbase-unsecure
- var modelDir = args(6)
-
- val conf = new SparkConf()
- conf.set("spark.app.name", "AMSAnomalyModelBuilder")
- //conf.set("spark.master", "spark://avijayan-ams-2.openstacklocal:7077")
-
- var sc = new SparkContext(conf)
- val sqlContext = new SQLContext(sc)
-
- val currentTime = System.currentTimeMillis()
- val oneDayBack = currentTime - 24*60*60*1000
-
- val df = sqlContext.load("org.apache.phoenix.spark", Map("table" -> "METRIC_RECORD", "zkUrl" -> phoenixConnectionString))
- df.registerTempTable("METRIC_RECORD")
- val result = sqlContext.sql("SELECT METRIC_NAME, HOSTNAME, APP_ID, SERVER_TIME, METRIC_SUM, METRIC_COUNT FROM METRIC_RECORD " +
- "WHERE METRIC_NAME = '" + metricName + "' AND HOSTNAME = '" + hostname + "' AND APP_ID = '" + appId + "' AND SERVER_TIME < " + currentTime + " AND SERVER_TIME > " + oneDayBack)
-
- var metricValues = new java.util.TreeMap[java.lang.Long, java.lang.Double]
- result.collect().foreach(
- t => metricValues.put(t.getLong(3), t.getDouble(4) / t.getInt(5))
- )
-
- //val seriesName = result.head().getString(0)
- //val hostname = result.head().getString(1)
- //val appId = result.head().getString(2)
-
- val timelineMetric = new TimelineMetric()
- timelineMetric.setMetricName(metricName)
- timelineMetric.setAppId(appId)
- timelineMetric.setHostName(hostname)
- timelineMetric.setMetricValues(metricValues)
-
- var emaModel = new EmaTechnique(weight, timessdev)
- emaModel.test(timelineMetric)
- emaModel.save(sc, modelDir)
+// if (args.length < 6) {
+// System.err.println("Usage: SparkPhoenixReader <metric_name> <appId> <hostname> <weight> <timessdev> <phoenixConnectionString> <model_dir>")
+// System.exit(1)
+// }
+//
+// var metricName = args(0)
+// var appId = args(1)
+// var hostname = args(2)
+// var weight = args(3).toDouble
+// var timessdev = args(4).toInt
+// var phoenixConnectionString = args(5) //avijayan-ams-3.openstacklocal:61181:/ams-hbase-unsecure
+// var modelDir = args(6)
+//
+// val conf = new SparkConf()
+// conf.set("spark.app.name", "AMSAnomalyModelBuilder")
+// //conf.set("spark.master", "spark://avijayan-ams-2.openstacklocal:7077")
+//
+// var sc = new SparkContext(conf)
+// val sqlContext = new SQLContext(sc)
+//
+// val currentTime = System.currentTimeMillis()
+// val oneDayBack = currentTime - 24*60*60*1000
+//
+// val df = sqlContext.load("org.apache.phoenix.spark", Map("table" -> "METRIC_RECORD", "zkUrl" -> phoenixConnectionString))
+// df.registerTempTable("METRIC_RECORD")
+// val result = sqlContext.sql("SELECT METRIC_NAME, HOSTNAME, APP_ID, SERVER_TIME, METRIC_SUM, METRIC_COUNT FROM METRIC_RECORD " +
+// "WHERE METRIC_NAME = '" + metricName + "' AND HOSTNAME = '" + hostname + "' AND APP_ID = '" + appId + "' AND SERVER_TIME < " + currentTime + " AND SERVER_TIME > " + oneDayBack)
+//
+// var metricValues = new java.util.TreeMap[java.lang.Long, java.lang.Double]
+// result.collect().foreach(
+// t => metricValues.put(t.getLong(3), t.getDouble(4) / t.getInt(5))
+// )
+//
+// //val seriesName = result.head().getString(0)
+// //val hostname = result.head().getString(1)
+// //val appId = result.head().getString(2)
+//
+// val timelineMetric = new TimelineMetric()
+// timelineMetric.setMetricName(metricName)
+// timelineMetric.setAppId(appId)
+// timelineMetric.setHostName(hostname)
+// timelineMetric.setMetricValues(metricValues)
+//
+// var emaModel = new EmaTechnique(weight, timessdev)
+// emaModel.test(timelineMetric)
+// emaModel.save(sc, modelDir)
}
diff --git a/ambari-metrics-anomaly-detection-service/src/test/scala/org/apache/ambari/metrics/adservice/common/ADManagerConfigurationTest.scala b/ambari-metrics-anomaly-detection-service/src/test/scala/org/apache/ambari/metrics/adservice/common/ADManagerConfigurationTest.scala
new file mode 100644
index 0000000..535dc9e
--- /dev/null
+++ b/ambari-metrics-anomaly-detection-service/src/test/scala/org/apache/ambari/metrics/adservice/common/ADManagerConfigurationTest.scala
@@ -0,0 +1,23 @@
+package org.apache.ambari.metrics.adservice.common
+
+import org.scalatest.FlatSpec
+
+import scala.collection.mutable
+
+class ADServiceConfigurationTest extends FlatSpec {
+
+ "A Stack" should "pop values in last-in-first-out order" in {
+ val stack = new mutable.Stack[Int]
+ stack.push(1)
+ stack.push(2)
+ assert(stack.pop() === 2)
+ assert(stack.pop() === 1)
+ }
+
+ it should "throw NoSuchElementException if an empty stack is popped" in {
+ val emptyStack = new mutable.Stack[String]
+ assertThrows[NoSuchElementException] {
+ emptyStack.pop()
+ }
+ }
+}
diff --git a/ambari-metrics-anomaly-detection-service/src/test/scala/org/apache/ambari/metrics/adservice/db/PhoenixAnomalyStoreAccessorTest.scala b/ambari-metrics-anomaly-detection-service/src/test/scala/org/apache/ambari/metrics/adservice/db/PhoenixAnomalyStoreAccessorTest.scala
new file mode 100644
index 0000000..142e98a
--- /dev/null
+++ b/ambari-metrics-anomaly-detection-service/src/test/scala/org/apache/ambari/metrics/adservice/db/PhoenixAnomalyStoreAccessorTest.scala
@@ -0,0 +1,26 @@
+/*
+ * 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.ambari.metrics.adservice.db
+
+import org.scalatest.FunSuite
+
+class PhoenixAnomalyStoreAccessorTest extends FunSuite {
+
+ test("testInitAnomalyMetricSchema") {
+ }
+
+}
diff --git a/ambari-metrics-common/pom.xml b/ambari-metrics-common/pom.xml
index 4f08820..5477270 100644
--- a/ambari-metrics-common/pom.xml
+++ b/ambari-metrics-common/pom.xml
@@ -26,6 +26,13 @@
<modelVersion>4.0.0</modelVersion>
<artifactId>ambari-metrics-common</artifactId>
<name>Ambari Metrics Common</name>
+
+ <properties>
+ <hadoop.version>2.7.3.2.6.0.3-8</hadoop.version>
+ <hbase.version>1.1.2.2.6.0.3-8</hbase.version>
+ <phoenix.version>4.7.0.2.6.0.3-8</phoenix.version>
+ </properties>
+
<build>
<plugins>
<plugin>
@@ -126,6 +133,45 @@
<dependencies>
<dependency>
+ <groupId>org.apache.phoenix</groupId>
+ <artifactId>phoenix-core</artifactId>
+ <version>${phoenix.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-annotations</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <version>${hadoop.version}</version>
+ <scope>provided</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-el</groupId>
+ <artifactId>commons-el</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-runtime</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>tomcat</groupId>
+ <artifactId>jasper-compiler</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.mortbay.jetty</groupId>
+ <artifactId>jsp-2.1-jetty</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
<groupId>net.sf.ehcache</groupId>
<artifactId>ehcache</artifactId>
<version>2.10.0</version>
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/ConnectionProvider.java b/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/ConnectionProvider.java
similarity index 79%
rename from ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/ConnectionProvider.java
rename to ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/ConnectionProvider.java
index 24239a0..72e5fb5 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/ConnectionProvider.java
+++ b/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/ConnectionProvider.java
@@ -15,9 +15,11 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query;
+package org.apache.hadoop.metrics2.sink.timeline.query;
+import org.apache.hadoop.hbase.util.RetryCounterFactory;
+
import java.sql.Connection;
import java.sql.SQLException;
@@ -26,4 +28,5 @@
*/
public interface ConnectionProvider {
public Connection getConnection() throws SQLException;
+ public Connection getConnectionRetryingOnException(RetryCounterFactory retryCounterFactory) throws SQLException, InterruptedException;
}
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/DefaultPhoenixDataSource.java b/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/DefaultPhoenixDataSource.java
similarity index 81%
rename from ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/DefaultPhoenixDataSource.java
rename to ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/DefaultPhoenixDataSource.java
index c5761f7..a28a433 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/DefaultPhoenixDataSource.java
+++ b/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/DefaultPhoenixDataSource.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query;
+package org.apache.hadoop.metrics2.sink.timeline.query;
import org.apache.commons.logging.Log;
@@ -23,6 +23,8 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HBaseAdmin;
+import org.apache.hadoop.hbase.util.RetryCounter;
+import org.apache.hadoop.hbase.util.RetryCounterFactory;
import java.io.IOException;
import java.sql.Connection;
@@ -87,4 +89,20 @@
}
}
+ public Connection getConnectionRetryingOnException(RetryCounterFactory retryCounterFactory)
+ 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();
+ }
+ }
}
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixConnectionProvider.java b/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/PhoenixConnectionProvider.java
similarity index 92%
rename from ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixConnectionProvider.java
rename to ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/PhoenixConnectionProvider.java
index cacbcfb..194c769 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/query/PhoenixConnectionProvider.java
+++ b/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/query/PhoenixConnectionProvider.java
@@ -1,4 +1,4 @@
-package org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query;
+package org.apache.hadoop.metrics2.sink.timeline.query;
import org.apache.hadoop.hbase.client.HBaseAdmin;
diff --git a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java b/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java
index f470c58..f8d31f7 100644
--- a/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java
+++ b/ambari-metrics-timelineservice/src/main/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessor.java
@@ -140,8 +140,8 @@
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.discovery.TimelineMetricMetadataKey;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.discovery.TimelineMetricMetadataManager;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.Condition;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.DefaultPhoenixDataSource;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixConnectionProvider;
+import org.apache.hadoop.metrics2.sink.timeline.query.DefaultPhoenixDataSource;
+import org.apache.hadoop.metrics2.sink.timeline.query.PhoenixConnectionProvider;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.SplitByMetricNamesCondition;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.sink.ExternalMetricsSink;
@@ -458,23 +458,6 @@
return mapper.readValue(json, metricValuesTypeRef);
}
- 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
@@ -507,7 +490,7 @@
try {
LOG.info("Initializing metrics schema...");
- conn = getConnectionRetryingOnException();
+ conn = dataSource.getConnectionRetryingOnException(retryCounterFactory);
stmt = conn.createStatement();
// Metadata
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryServer.java b/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryServer.java
index 03205e7..7b70a80 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryServer.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/TestApplicationHistoryServer.java
@@ -29,7 +29,7 @@
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.PhoenixHBaseAccessor;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.TimelineMetricConfiguration;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.availability.MetricCollectorHAController;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.DefaultPhoenixDataSource;
+import org.apache.hadoop.metrics2.sink.timeline.query.DefaultPhoenixDataSource;
import org.apache.zookeeper.ClientCnxn;
import org.easymock.EasyMock;
import org.junit.After;
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/AbstractMiniHBaseClusterTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/AbstractMiniHBaseClusterTest.java
index 3a42db9..40691d6 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/AbstractMiniHBaseClusterTest.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/AbstractMiniHBaseClusterTest.java
@@ -22,13 +22,9 @@
import static org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL.UPSERT_METRICS_SQL;
import static org.apache.phoenix.util.TestUtil.TEST_PROPERTIES;
import static org.assertj.core.api.Assertions.assertThat;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.replay;
import static org.powermock.api.easymock.PowerMock.mockStatic;
-import static org.powermock.api.easymock.PowerMock.replayAll;
import java.io.IOException;
-import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
@@ -48,7 +44,7 @@
import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.aggregators.AggregatorUtils;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixConnectionProvider;
+import org.apache.hadoop.metrics2.sink.timeline.query.PhoenixConnectionProvider;
import org.apache.hadoop.yarn.util.timeline.TimelineUtils;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
diff --git a/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessorTest.java b/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessorTest.java
index 7be3c0d..97d2512 100644
--- a/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessorTest.java
+++ b/ambari-metrics-timelineservice/src/test/java/org/apache/hadoop/yarn/server/applicationhistoryservice/metrics/timeline/PhoenixHBaseAccessorTest.java
@@ -32,19 +32,17 @@
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.discovery.TimelineMetricMetadataManager;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.Condition;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.DefaultCondition;
-import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixConnectionProvider;
+import org.apache.hadoop.metrics2.sink.timeline.query.PhoenixConnectionProvider;
import org.apache.hadoop.yarn.server.applicationhistoryservice.metrics.timeline.query.PhoenixTransactSQL;
import org.apache.phoenix.exception.PhoenixIOException;
import org.easymock.EasyMock;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.powermock.api.easymock.PowerMock;
import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner;
import java.io.IOException;
-import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
