AMBARI-6106. Customize the Hadoop metrics sink to write to MySQL stor…

…e. (swagle)

contrib/ambari-scom/metrics-sink/db/Hadoop-Metrics-SQLServer-CREATE.ddl

@@ -550,10 +550,10 @@ BEGIN
 			SELECT @recordIDCutoff = MIN(RecordID) FROM MetricRecord WHERE RecordID IN (SELECT TOP 60000 RecordID FROM MetricRecord WHERE NodeID = @nodeID ORDER BY RecordDate DESC);
 			IF @recordIDCutoff IS NOT NULL
 			BEGIN
-				DELETE FROM MetricPair 
-				FROM MetricPair as mp
+				DELETE FROM MetricPair WHERE RecordID IN (
+				SELECT RecordID FROM MetricPair as mp
 				JOIN MetricRecord as mr ON mp.RecordID = mr.RecordID
-				WHERE mr.RecordID < @recordIDCutoff AND mr.NodeID = @nodeID;
+				WHERE mr.RecordID < @recordIDCutoff AND mr.NodeID = @nodeID);
 
 				DELETE FROM MetricRecord
 				WHERE RecordID < @recordIDCutoff AND NodeID = @nodeID;

contrib/ambari-scom/metrics-sink/pom.xml

@@ -70,6 +70,7 @@
                         <configuration>
                             <excludes>
                                 <exclude>**/SqlServerSinkHadoop2.java</exclude>
+                                <exclude>**/MySqlSinkHadoop2.java</exclude>
                             </excludes>
                             <testExcludes>
                                 <exclude>**/SqlServerSinkHadoop2Test.java</exclude>
@@ -99,6 +100,7 @@
                         <configuration>
                             <excludes>
                                 <exclude>**/SqlServerSinkHadoop1.java</exclude>
+                                <exclude>**/MySqlSinkHadoop1.java</exclude>
                             </excludes>
                             <testExcludes>
                                 <exclude>**/SqlServerSinkHadoop1Test.java</exclude>

contrib/ambari-scom/metrics-sink/src/main/java/org/apache/hadoop/metrics2/sink/MySqlSink.java

@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.metrics2.sink;
+
+import java.lang.Override;
+
+/**
+ * This class stores published metrics to the MySql database.
+ */
+public abstract class MySqlSink extends SqlSink {
+  public MySqlSink(String NAMENODE_URL_KEY, String DFS_BLOCK_SIZE_KEY) {
+    super(NAMENODE_URL_KEY, DFS_BLOCK_SIZE_KEY);
+  }
+
+  @Override
+  protected String getInsertMetricsProcedureName() {
+    return SqlSink.insertMetricProc;
+  }
+
+  @Override
+  protected String getGetMetricsProcedureName() {
+    return SqlSink.getMetricRecordProc;
+  }
+
+  @Override
+  protected String getDatabaseDriverClassName() {
+    return "com.mysql.jdbc.Driver";
+  }
+}

contrib/ambari-scom/metrics-sink/src/main/java/org/apache/hadoop/metrics2/sink/MySqlSinkHadoop1.java

@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.metrics2.sink;
+
+import org.apache.hadoop.metrics2.Metric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+
+public class MySqlSinkHadoop1 extends MySqlSink {
+  public MySqlSinkHadoop1() {
+    super(SqlSink.HADOOP1_NAMENODE_URL_KEY, SqlSink.HADOOP1_DFS_BLOCK_SIZE_KEY);
+  }
+
+  @Override
+  public void putMetrics(MetricsRecord record) {
+    long metricRecordID = getMetricRecordID(record.context(), record.name(),
+      getLocalNodeName(), getLocalNodeIPAddress(), getClusterNodeName(), getCurrentServiceName(),
+      getTagString(record.tags()), record.timestamp());
+    if (metricRecordID < 0)
+      return;
+
+    for (Metric metric : record.metrics()) {
+      insertMetricValue(metricRecordID, metric.name(), String.valueOf(metric
+        .value()));
+      if (metric.name().equals("BlockCapacity")) {
+        insertMetricValue(metricRecordID, "BlockSize", Integer
+          .toString(getBlockSize()));
+      }
+    }
+  }
+}

contrib/ambari-scom/metrics-sink/src/main/java/org/apache/hadoop/metrics2/sink/MySqlSinkHadoop2.java

@@ -0,0 +1,44 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.metrics2.sink;
+
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+
+public class MySqlSinkHadoop2 extends MySqlSink {
+  public MySqlSinkHadoop2() {
+    super(SqlSink.HADOOP2_NAMENODE_URL_KEY, SqlSink.HADOOP2_DFS_BLOCK_SIZE_KEY);
+  }
+
+  @Override
+  public void putMetrics(MetricsRecord record) {
+    long metricRecordID = getMetricRecordID(record.context(), record.name(),
+      getLocalNodeName(), getLocalNodeIPAddress(), getClusterNodeName(), getCurrentServiceName(),
+      getTagString(record.tags()), record.timestamp());
+    if (metricRecordID < 0)
+      return;
+
+    for (AbstractMetric metric : record.metrics()) {
+      insertMetricValue(metricRecordID, metric.name(), String.valueOf(metric.value()));
+      if (metric.name().equals("BlockCapacity")) {
+        insertMetricValue(metricRecordID, "BlockSize", Integer
+          .toString(getBlockSize()));
+      }
+    }
+  }
+}