AMBARI-4840 - Need 2 versions of metrics sink

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

@@ -44,13 +44,72 @@
             <version>3.1</version>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.apache.hadoop</groupId>
-            <artifactId>hadoop-core</artifactId>
-            <version>[1.0,2.0)</version>
-        </dependency>
     </dependencies>
 
+    <profiles>
+        <profile>
+            <id>hadoop1</id>
+            <activation>
+                <activeByDefault>false</activeByDefault>
+                <property>
+                    <name>hadoop1</name>
+                </property>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-core</artifactId>
+                    <version>[1.0,2.0)</version>
+                </dependency>
+            </dependencies>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-compiler-plugin</artifactId>
+                        <configuration>
+                            <excludes>
+                                <exclude>**/SqlServerSinkHadoop2.java</exclude>
+                            </excludes>
+                            <testExcludes>
+                                <exclude>**/SqlServerSinkHadoop2Test.java</exclude>
+                            </testExcludes>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+        <profile>
+            <id>hadoop2</id>
+            <activation>
+                <activeByDefault>true</activeByDefault>
+            </activation>
+            <dependencies>
+                <dependency>
+                    <groupId>org.apache.hadoop</groupId>
+                    <artifactId>hadoop-common</artifactId>
+                    <version>2.2.0</version>
+                </dependency>
+            </dependencies>
+            <build>
+                <plugins>
+                    <plugin>
+                        <groupId>org.apache.maven.plugins</groupId>
+                        <artifactId>maven-compiler-plugin</artifactId>
+                        <configuration>
+                            <excludes>
+                                <exclude>**/SqlServerSinkHadoop1.java</exclude>
+                            </excludes>
+                            <testExcludes>
+                                <exclude>**/SqlServerSinkHadoop1Test.java</exclude>
+                            </testExcludes>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
+
     <build>
         <pluginManagement>
             <plugins>

contrib/ambari-scom/metrics-sink/src/Test/java/org/apache/hadoop/metrics2/sink/SqlServerSinkHadoop1Test.java

@@ -0,0 +1,97 @@
+/**
+ * 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 com.microsoft.sqlserver.jdbc.SQLServerDriver;
+import org.apache.commons.configuration.SubsetConfiguration;
+import org.apache.hadoop.metrics2.Metric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.junit.Test;
+
+import java.sql.CallableStatement;
+import java.sql.Connection;
+import java.util.Collections;
+import java.util.HashSet;
+
+import static org.easymock.EasyMock.*;
+
+public class SqlServerSinkHadoop1Test extends SqlServerSinkTest<SqlServerSinkHadoop1> {
+    @Override
+    public SqlServerSinkHadoop1 createInstance() throws InstantiationException, IllegalAccessException {
+        return new SqlServerSinkHadoop1();
+    }
+
+    @Override
+    @Test
+    public void testPutMetrics() throws Exception {
+        SubsetConfiguration configuration = createNiceMock(SubsetConfiguration.class);
+        Connection connection = createNiceMock(Connection.class);
+        CallableStatement cstmt = createNiceMock(CallableStatement.class);
+        MetricsRecord record = createNiceMock(MetricsRecord.class);
+        Metric metric = createNiceMock(Metric.class);
+
+        // set expectations
+        expect(configuration.getParent()).andReturn(null);
+        expect(configuration.getPrefix()).andReturn("prefix");
+        expect(configuration.getString("databaseUrl")).andReturn("url");
+
+        expect(record.context()).andReturn("context");
+        expect(record.name()).andReturn("typeName");
+        expect(record.tags()).andReturn(new HashSet<MetricsTag>());
+        expect(record.timestamp()).andReturn(9999L);
+
+        expect(record.metrics()).andReturn(Collections.singleton(metric));
+
+        expect(metric.name()).andReturn("name").anyTimes();
+        expect(metric.value()).andReturn(1234);
+
+        expect(connection.prepareCall("{call dbo.uspGetMetricRecord(?, ?, ?, ?, ?, ?, ?, ?, ?)}")).andReturn(cstmt);
+        cstmt.setNString(1, "context");
+        cstmt.setNString(2, "typeName");
+        cstmt.setNString(eq(3), (String) anyObject());
+        cstmt.setNString(eq(4), (String) anyObject());
+        cstmt.setNString(eq(5), (String) anyObject());
+        cstmt.setNString(6, "prefix");
+        cstmt.setNString(7, "sourceName:prefix");
+        cstmt.setLong(8, 9999L);
+        cstmt.registerOutParameter(9, java.sql.Types.BIGINT);
+        expect(cstmt.execute()).andReturn(true);
+        expect(cstmt.getLong(9)).andReturn(99L);
+        expect(cstmt.wasNull()).andReturn(false);
+
+        expect(connection.prepareCall("{call dbo.uspInsertMetricValue(?, ?, ?)}")).andReturn(cstmt);
+        cstmt.setLong(1, 99L);
+        cstmt.setNString(2, "name");
+        cstmt.setNString(3, "1234");
+        expect(cstmt.execute()).andReturn(true);
+
+        // replay
+        replay(configuration, connection, cstmt, record, metric);
+
+        SqlServerSink sink = createInstance();
+
+        sink.init(configuration);
+
+        SQLServerDriver.setConnection(connection);
+
+        sink.putMetrics(record);
+
+        verify(configuration, connection, cstmt, record, metric);
+    }
+}

contrib/ambari-scom/metrics-sink/src/Test/java/org/apache/hadoop/metrics2/sink/SqlServerSinkHadoop2Test.java

@@ -0,0 +1,97 @@
+/**
+ * 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 com.microsoft.sqlserver.jdbc.SQLServerDriver;
+import org.apache.commons.configuration.SubsetConfiguration;
+import org.apache.hadoop.metrics2.AbstractMetric;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.junit.Test;
+
+import java.sql.CallableStatement;
+import java.sql.Connection;
+import java.util.Collections;
+import java.util.HashSet;
+
+import static org.easymock.EasyMock.*;
+
+public class SqlServerSinkHadoop2Test extends SqlServerSinkTest<SqlServerSinkHadoop2> {
+    @Override
+    public SqlServerSinkHadoop2 createInstance() throws InstantiationException, IllegalAccessException {
+        return new SqlServerSinkHadoop2();
+    }
+
+    @Override
+    @Test
+    public void testPutMetrics() throws Exception {
+        SubsetConfiguration configuration = createNiceMock(SubsetConfiguration.class);
+        Connection connection = createNiceMock(Connection.class);
+        CallableStatement cstmt = createNiceMock(CallableStatement.class);
+        MetricsRecord record = createNiceMock(MetricsRecord.class);
+        AbstractMetric metric = createNiceMock(AbstractMetric.class);
+
+        // set expectations
+        expect(configuration.getParent()).andReturn(null);
+        expect(configuration.getPrefix()).andReturn("prefix");
+        expect(configuration.getString("databaseUrl")).andReturn("url");
+
+        expect(record.context()).andReturn("context");
+        expect(record.name()).andReturn("typeName");
+        expect(record.tags()).andReturn(new HashSet<MetricsTag>());
+        expect(record.timestamp()).andReturn(9999L);
+
+        expect(record.metrics()).andReturn(Collections.singleton(metric));
+
+        expect(metric.name()).andReturn("name").anyTimes();
+        expect(metric.value()).andReturn(1234);
+
+        expect(connection.prepareCall("{call dbo.uspGetMetricRecord(?, ?, ?, ?, ?, ?, ?, ?, ?)}")).andReturn(cstmt);
+        cstmt.setNString(1, "context");
+        cstmt.setNString(2, "typeName");
+        cstmt.setNString(eq(3), (String) anyObject());
+        cstmt.setNString(eq(4), (String) anyObject());
+        cstmt.setNString(eq(5), (String) anyObject());
+        cstmt.setNString(6, "prefix");
+        cstmt.setNString(7, "sourceName:prefix");
+        cstmt.setLong(8, 9999L);
+        cstmt.registerOutParameter(9, java.sql.Types.BIGINT);
+        expect(cstmt.execute()).andReturn(true);
+        expect(cstmt.getLong(9)).andReturn(99L);
+        expect(cstmt.wasNull()).andReturn(false);
+
+        expect(connection.prepareCall("{call dbo.uspInsertMetricValue(?, ?, ?)}")).andReturn(cstmt);
+        cstmt.setLong(1, 99L);
+        cstmt.setNString(2, "name");
+        cstmt.setNString(3, "1234");
+        expect(cstmt.execute()).andReturn(true);
+
+        // replay
+        replay(configuration, connection, cstmt, record, metric);
+
+        SqlServerSink sink = createInstance();
+
+        sink.init(configuration);
+
+        SQLServerDriver.setConnection(connection);
+
+        sink.putMetrics(record);
+
+        verify(configuration, connection, cstmt, record, metric);
+    }
+}

contrib/ambari-scom/metrics-sink/src/Test/java/org/apache/hadoop/metrics2/sink/SqlServerSinkTest.java

@@ -20,20 +20,13 @@ package org.apache.hadoop.metrics2.sink;
 
 import com.microsoft.sqlserver.jdbc.SQLServerDriver;
 import org.apache.commons.configuration.SubsetConfiguration;
-import org.apache.hadoop.metrics2.Metric;
-import org.apache.hadoop.metrics2.MetricsRecord;
-import org.apache.hadoop.metrics2.MetricsTag;
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.sql.CallableStatement;
 import java.sql.Connection;
-import java.util.Collections;
-import java.util.HashSet;
 
-import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.createNiceMock;
-import static org.easymock.EasyMock.eq;
 import static org.easymock.EasyMock.expect;
 import static org.easymock.EasyMock.replay;
 import static org.easymock.EasyMock.verify;
@@ -41,7 +34,10 @@ import static org.easymock.EasyMock.verify;
 /**
  * SqlServerSink Tests.
  */
-public class SqlServerSinkTest {
+public abstract class SqlServerSinkTest<T extends SqlServerSink> {
+
+  public abstract T createInstance() throws InstantiationException, IllegalAccessException;
+
   @Test
   public void testInit() throws Exception {
 
@@ -55,7 +51,7 @@ public class SqlServerSinkTest {
     // replay
     replay(configuration);
 
-    SqlServerSink sink = new SqlServerSink();
+    SqlServerSink sink = createInstance();
 
     sink.init(configuration);
 
@@ -76,11 +72,11 @@ public class SqlServerSinkTest {
     // replay
     replay(configuration, connection);
 
-    SqlServerSink sink = new SqlServerSink();
+    SqlServerSink sink = createInstance();
 
     sink.init(configuration);
 
-    Assert.assertFalse(sink.ensureConnection());
+    Assert.assertTrue(sink.ensureConnection());
 
     SQLServerDriver.setConnection(connection);
     Assert.assertTrue(sink.ensureConnection());
@@ -103,7 +99,7 @@ public class SqlServerSinkTest {
     // replay
     replay(configuration, connection);
 
-    SqlServerSink sink = new SqlServerSink();
+    SqlServerSink sink = createInstance();
 
     sink.init(configuration);
 
@@ -142,7 +138,7 @@ public class SqlServerSinkTest {
     // replay
     replay(configuration, connection, cstmt);
 
-    SqlServerSink sink = new SqlServerSink();
+    SqlServerSink sink = createInstance();
 
     sink.init(configuration);
 
@@ -182,7 +178,7 @@ public class SqlServerSinkTest {
     // replay
     replay(configuration, connection, cstmt);
 
-    SqlServerSink sink = new SqlServerSink();
+    SqlServerSink sink = createInstance();
 
     sink.init(configuration);
 
@@ -214,7 +210,7 @@ public class SqlServerSinkTest {
     // replay
     replay(configuration, connection, cstmt);
 
-    SqlServerSink sink = new SqlServerSink();
+    SqlServerSink sink = createInstance();
 
     sink.init(configuration);
 
@@ -225,60 +221,5 @@ public class SqlServerSinkTest {
     verify(configuration, connection, cstmt);
   }
 
-  @Test
-  public void testPutMetrics() throws Exception {
-    SubsetConfiguration configuration = createNiceMock(SubsetConfiguration.class);
-    Connection connection = createNiceMock(Connection.class);
-    CallableStatement cstmt = createNiceMock(CallableStatement.class);
-    MetricsRecord record = createNiceMock(MetricsRecord.class);
-    Metric metric = createNiceMock(Metric.class);
-
-    // set expectations
-    expect(configuration.getParent()).andReturn(null);
-    expect(configuration.getPrefix()).andReturn("prefix");
-    expect(configuration.getString("databaseUrl")).andReturn("url");
-
-    expect(record.context()).andReturn("context");
-    expect(record.name()).andReturn("typeName");
-    expect(record.tags()).andReturn(new HashSet<MetricsTag>());
-    expect(record.timestamp()).andReturn(9999L);
-
-    expect(record.metrics()).andReturn(Collections.singleton(metric));
-
-    expect(metric.name()).andReturn("name").anyTimes();
-    expect(metric.value()).andReturn(1234);
-
-    expect(connection.prepareCall("{call dbo.uspGetMetricRecord(?, ?, ?, ?, ?, ?, ?, ?, ?)}")).andReturn(cstmt);
-    cstmt.setNString(1, "context");
-    cstmt.setNString(2, "typeName");
-    cstmt.setNString(eq(3), (String) anyObject());
-    cstmt.setNString(eq(4), (String) anyObject());
-    cstmt.setNString(eq(5), (String) anyObject());
-    cstmt.setNString(6, "prefix");
-    cstmt.setNString(7, "sourceName:prefix");
-    cstmt.setLong(8, 9999L);
-    cstmt.registerOutParameter(9, java.sql.Types.BIGINT);
-    expect(cstmt.execute()).andReturn(true);
-    expect(cstmt.getLong(9)).andReturn(99L);
-    expect(cstmt.wasNull()).andReturn(false);
-
-    expect(connection.prepareCall("{call dbo.uspInsertMetricValue(?, ?, ?)}")).andReturn(cstmt);
-    cstmt.setLong(1, 99L);
-    cstmt.setNString(2, "name");
-    cstmt.setNString(3, "1234");
-    expect(cstmt.execute()).andReturn(true);
-
-    // replay
-    replay(configuration, connection, cstmt, record, metric);
-
-    SqlServerSink sink = new SqlServerSink();
-
-    sink.init(configuration);
-
-    SQLServerDriver.setConnection(connection);
-
-    sink.putMetrics(record);
-
-    verify(configuration, connection, cstmt, record, metric);
-  }
+  public abstract void testPutMetrics() throws Exception;
 }

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

@@ -18,7 +18,13 @@
 
 package org.apache.hadoop.metrics2.sink;
 
-import java.lang.Exception;
+import org.apache.commons.configuration.SubsetConfiguration;
+import org.apache.hadoop.metrics2.MetricsException;
+import org.apache.hadoop.metrics2.MetricsRecord;
+import org.apache.hadoop.metrics2.MetricsSink;
+import org.apache.hadoop.metrics2.MetricsTag;
+import org.apache.log4j.Logger;
+
 import java.net.InetAddress;
 import java.sql.CallableStatement;
 import java.sql.Connection;
@@ -27,18 +33,10 @@ import java.sql.SQLException;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import org.apache.commons.configuration.SubsetConfiguration;
-import org.apache.hadoop.metrics2.Metric;
-import org.apache.hadoop.metrics2.MetricsException;
-import org.apache.hadoop.metrics2.MetricsRecord;
-import org.apache.hadoop.metrics2.MetricsSink;
-import org.apache.hadoop.metrics2.MetricsTag;
-import org.apache.log4j.Logger;
-
 /**
  * This class stores published metrics to the SQL Server database.
  */
-public class SqlServerSink implements MetricsSink {
+public abstract class SqlServerSink implements MetricsSink {
   private static final String DATABASE_URL_KEY = "databaseUrl";
   private static final boolean DEBUG = true;
   private static final String NAME_URL_KEY = "fs.default.name";
@@ -109,22 +107,7 @@ public class SqlServerSink implements MetricsSink {
   }
 
   @Override
-  public void putMetrics(MetricsRecord record) {
-    long metricRecordID = getMetricRecordID(record.context(), record.name(),
-        getLocalNodeName(), getLocalNodeIPAddress(), getClusterNodeName(), currentServiceName,
-        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(blockSize));
-      }
-    }
-  }
+  public abstract void putMetrics(MetricsRecord record);
 
   @Override
   public void flush() {
@@ -311,4 +294,12 @@ public class SqlServerSink implements MetricsSink {
       }
     }
   }
+
+  public String getCurrentServiceName() {
+    return currentServiceName;
+  }
+
+  public int getBlockSize() {
+    return blockSize;
+  }
 }

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

@@ -0,0 +1,41 @@
+/**
+ * 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 SqlServerSinkHadoop1 extends SqlServerSink {
+  @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/SqlServerSinkHadoop2.java

@@ -0,0 +1,40 @@
+/**
+ * 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 SqlServerSinkHadoop2 extends SqlServerSink {
+  @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()));
+      }
+    }
+  }
+}