SQLPropertyProvider extends AbstractPropertyProvider { @Override public Set<Resource> populateResources(Set<Resource> resources, Request request, Predicate predicate) throws SystemException { Set<Resource> keepers = new HashSet<Resource>(); try { Connection connection = connectionFactory.getConnection(); try { Statement statement = connection.createStatement(); try { for (Resource resource : resources) { if (populateResource(resource, request, predicate, statement)) { keepers.add(resource); } } } finally { statement.close(); } } finally { connection.close(); } } catch (SQLException e) { if (LOG.isErrorEnabled()) { LOG.error("Error during populateResources call."); LOG.debug("Error during populateResources call : caught exception", e); } } return keepers; } SQLPropertyProvider(
    Map<String, Map<String, PropertyInfo>> componentPropertyInfoMap,
    HostInfoProvider hostProvider,
    String clusterNamePropertyId,
    String hostNamePropertyId,
    String componentNamePropertyId,
    String serviceNamePropertyId,
    ConnectionFactory connectionFactory); @Override Set<Resource> populateResources(Set<Resource> resources, Request request, Predicate predicate);  }
@Test public void testPopulateResources() throws Exception { ConnectionFactory connectionFactory = createNiceMock(ConnectionFactory.class); Connection connection = createNiceMock(Connection.class); Statement statement = createNiceMock(Statement.class); ResultSet resultSet = createNiceMock(ResultSet.class); expect(connectionFactory.getConnection()).andReturn(connection).once(); expect(connection.createStatement()).andReturn(statement).once(); expect(statement.executeQuery(anyObject(String.class))).andReturn(resultSet).once(); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999990L); expect(resultSet.getNString("MetricValue")).andReturn("0"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999991L); expect(resultSet.getNString("MetricValue")).andReturn("1"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999992L); expect(resultSet.getNString("MetricValue")).andReturn("2"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999993L); expect(resultSet.getNString("MetricValue")).andReturn("3"); expect(resultSet.next()).andReturn(false); replay(connectionFactory, connection, statement, resultSet); SQLPropertyProvider provider = new SQLPropertyProvider( PropertyHelper.getGangliaPropertyIds(Resource.Type.HostComponent), new TestHostInfoProvider(), CLUSTER_NAME_PROPERTY_ID, HOST_NAME_PROPERTY_ID, COMPONENT_NAME_PROPERTY_ID, SERVICE_NAME_PROPERTY_ID, connectionFactory); Resource resource = new ResourceImpl(Resource.Type.HostComponent); resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1"); resource.setProperty(HOST_NAME_PROPERTY_ID, "domU-12-31-39-0E-34-E1.compute-1.internal"); resource.setProperty(COMPONENT_NAME_PROPERTY_ID, "DATANODE"); resource.setProperty(SERVICE_NAME_PROPERTY_ID, "HDFS"); Map<String, TemporalInfo> temporalInfoMap = new HashMap<String, TemporalInfo>(); temporalInfoMap.put(PROPERTY_ID_1, new TemporalInfoImpl(10L, 20L, 1L)); Request request = PropertyHelper.getReadRequest(Collections.singleton(PROPERTY_ID_1), temporalInfoMap); Assert.assertEquals(1, provider.populateResources(Collections.singleton(resource), request, null).size()); Assert.assertTrue(resource.getPropertyValue(PROPERTY_ID_1) instanceof Number[][]); Number[][] datapoints = (Number[][]) resource.getPropertyValue(PROPERTY_ID_1); for (int i = 0; i < datapoints.length; ++i) { Assert.assertEquals((long) i, datapoints[i][0]); Assert.assertEquals((999990L + i)/1000, datapoints[i][1]); } verify(connectionFactory, connection, statement, resultSet); }
@Test public void testPopulateResources_temporalStartTimeOnly() throws Exception { ConnectionFactory connectionFactory = createNiceMock(ConnectionFactory.class); Connection connection = createNiceMock(Connection.class); Statement statement = createNiceMock(Statement.class); ResultSet resultSet = createNiceMock(ResultSet.class); expect(connectionFactory.getConnection()).andReturn(connection).once(); expect(connection.createStatement()).andReturn(statement).once(); expect(statement.executeQuery(anyObject(String.class))).andReturn(resultSet).once(); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999990L); expect(resultSet.getNString("MetricValue")).andReturn("0"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999991L); expect(resultSet.getNString("MetricValue")).andReturn("1"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999992L); expect(resultSet.getNString("MetricValue")).andReturn("2"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999993L); expect(resultSet.getNString("MetricValue")).andReturn("3"); expect(resultSet.next()).andReturn(false); replay(connectionFactory, connection, statement, resultSet); TestHostInfoProvider hostProvider = new TestHostInfoProvider(); SQLPropertyProvider provider = new SQLPropertyProvider( PropertyHelper.getGangliaPropertyIds(Resource.Type.HostComponent), hostProvider, CLUSTER_NAME_PROPERTY_ID, HOST_NAME_PROPERTY_ID, COMPONENT_NAME_PROPERTY_ID, SERVICE_NAME_PROPERTY_ID, connectionFactory); Resource resource = new ResourceImpl(Resource.Type.HostComponent); resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1"); resource.setProperty(HOST_NAME_PROPERTY_ID, "domU-12-31-39-0E-34-E1.compute-1.internal"); resource.setProperty(COMPONENT_NAME_PROPERTY_ID, "DATANODE"); resource.setProperty(SERVICE_NAME_PROPERTY_ID, "HDFS"); Map<String, TemporalInfo> temporalInfoMap = new HashMap<String, TemporalInfo>(); temporalInfoMap.put(PROPERTY_ID_1, new TemporalInfoImpl(10L, -1L, -1L)); Request request = PropertyHelper.getReadRequest(Collections.singleton(PROPERTY_ID_1), temporalInfoMap); Assert.assertEquals(1, provider.populateResources(Collections.singleton(resource), request, null).size()); Assert.assertTrue(resource.getPropertyValue(PROPERTY_ID_1) instanceof Number[][]); Number[][] datapoints = (Number[][]) resource.getPropertyValue(PROPERTY_ID_1); for (int i = 0; i < datapoints.length; ++i) { Assert.assertEquals((long) i, datapoints[i][0]); Assert.assertEquals((999990L + i) / 1000, datapoints[i][1]); } verify(connectionFactory, connection, statement, resultSet); }
@Test public void testPopulateResources_hostNameProperty() throws Exception { ConnectionFactory connectionFactory = createNiceMock(ConnectionFactory.class); Connection connection = createNiceMock(Connection.class); Statement statement = createNiceMock(Statement.class); ResultSet resultSet = createNiceMock(ResultSet.class); expect(connectionFactory.getConnection()).andReturn(connection).once(); expect(connection.createStatement()).andReturn(statement).once(); expect(statement.executeQuery(anyObject(String.class))).andReturn(resultSet).once(); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999990L); expect(resultSet.getNString("MetricValue")).andReturn("0"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999991L); expect(resultSet.getNString("MetricValue")).andReturn("1"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999992L); expect(resultSet.getNString("MetricValue")).andReturn("2"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999993L); expect(resultSet.getNString("MetricValue")).andReturn("3"); expect(resultSet.next()).andReturn(false); replay(connectionFactory, connection, statement, resultSet); TestHostInfoProvider hostProvider = new TestHostInfoProvider(); SQLPropertyProvider provider = new SQLPropertyProvider( PropertyHelper.getGangliaPropertyIds(Resource.Type.HostComponent), hostProvider, CLUSTER_NAME_PROPERTY_ID, HOST_NAME_PROPERTY_ID, COMPONENT_NAME_PROPERTY_ID, SERVICE_NAME_PROPERTY_ID, connectionFactory); Resource resource = new ResourceImpl(Resource.Type.HostComponent); resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1"); resource.setProperty(HOST_NAME_PROPERTY_ID, "domU-12-31-39-0E-34-E1.compute-1.internal"); resource.setProperty(COMPONENT_NAME_PROPERTY_ID, "DATANODE"); resource.setProperty(SERVICE_NAME_PROPERTY_ID, "HDFS"); Map<String, TemporalInfo> temporalInfoMap = new HashMap<String, TemporalInfo>(); temporalInfoMap.put(PROPERTY_ID_1, new TemporalInfoImpl(10L, -1L, -1L)); Request request = PropertyHelper.getReadRequest(Collections.singleton(PROPERTY_ID_1), temporalInfoMap); provider.populateResources(Collections.singleton(resource), request, null); Assert.assertEquals("domU-12-31-39-0E-34-E1.compute-1.internal", hostProvider.getHostId()); Assert.assertNull(hostProvider.getClusterName()); Assert.assertNull(hostProvider.getComponentName()); verify(connectionFactory, connection, statement, resultSet); }
@Test public void testPopulateResources_noHostNameProperty() throws Exception { ConnectionFactory connectionFactory = createNiceMock(ConnectionFactory.class); Connection connection = createNiceMock(Connection.class); Statement statement = createNiceMock(Statement.class); ResultSet resultSet = createNiceMock(ResultSet.class); expect(connectionFactory.getConnection()).andReturn(connection).once(); expect(connection.createStatement()).andReturn(statement).once(); expect(statement.executeQuery(anyObject(String.class))).andReturn(resultSet).once(); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999990L); expect(resultSet.getNString("MetricValue")).andReturn("0"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999991L); expect(resultSet.getNString("MetricValue")).andReturn("1"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999992L); expect(resultSet.getNString("MetricValue")).andReturn("2"); expect(resultSet.next()).andReturn(true); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999993L); expect(resultSet.getNString("MetricValue")).andReturn("3"); expect(resultSet.next()).andReturn(false); replay(connectionFactory, connection, statement, resultSet); TestHostInfoProvider hostProvider = new TestHostInfoProvider(); SQLPropertyProvider provider = new SQLPropertyProvider( PropertyHelper.getGangliaPropertyIds(Resource.Type.HostComponent), hostProvider, CLUSTER_NAME_PROPERTY_ID, null, COMPONENT_NAME_PROPERTY_ID, SERVICE_NAME_PROPERTY_ID, connectionFactory); Resource resource = new ResourceImpl(Resource.Type.HostComponent); resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1"); resource.setProperty(COMPONENT_NAME_PROPERTY_ID, "DATANODE"); resource.setProperty(SERVICE_NAME_PROPERTY_ID, "HDFS"); Map<String, TemporalInfo> temporalInfoMap = new HashMap<String, TemporalInfo>(); temporalInfoMap.put(PROPERTY_ID_1, new TemporalInfoImpl(10L, -1L, -1L)); Request request = PropertyHelper.getReadRequest(Collections.singleton(PROPERTY_ID_1), temporalInfoMap); provider.populateResources(Collections.singleton(resource), request, null); Assert.assertNull(hostProvider.getHostId()); Assert.assertEquals("c1", hostProvider.getClusterName()); Assert.assertEquals("DATANODE", hostProvider.getComponentName()); verify(connectionFactory, connection, statement, resultSet); }
@Test public void testPopulateResources_pointInTime() throws Exception { ConnectionFactory connectionFactory = createNiceMock(ConnectionFactory.class); Connection connection = createNiceMock(Connection.class); Statement statement = createNiceMock(Statement.class); ResultSet resultSet = createNiceMock(ResultSet.class); expect(connectionFactory.getConnection()).andReturn(connection).once(); expect(connection.createStatement()).andReturn(statement).once(); expect(statement.executeQuery(anyObject(String.class))).andReturn(resultSet).once(); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcSlowResponse_num_ops"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getNString("MetricValue")).andReturn("0"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999990L); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcSlowResponse_num_ops"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getNString("MetricValue")).andReturn("1"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999991L); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcSlowResponse_num_ops"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getNString("MetricValue")).andReturn("2"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999992L); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcSlowResponse_num_ops"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getNString("MetricValue")).andReturn("3"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999993L); expect(resultSet.next()).andReturn(false); replay(connectionFactory, connection, statement, resultSet); SQLPropertyProvider provider = new SQLPropertyProvider( PropertyHelper.getGangliaPropertyIds(Resource.Type.HostComponent), new TestHostInfoProvider(), CLUSTER_NAME_PROPERTY_ID, HOST_NAME_PROPERTY_ID, COMPONENT_NAME_PROPERTY_ID, SERVICE_NAME_PROPERTY_ID, connectionFactory); Resource resource = new ResourceImpl(Resource.Type.HostComponent); resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1"); resource.setProperty(HOST_NAME_PROPERTY_ID, "domU-12-31-39-0E-34-E1.compute-1.internal"); resource.setProperty(COMPONENT_NAME_PROPERTY_ID, "DATANODE"); resource.setProperty(SERVICE_NAME_PROPERTY_ID, "HDFS"); Map<String, TemporalInfo> temporalInfoMap = new HashMap<String, TemporalInfo>(); Request request = PropertyHelper.getReadRequest(Collections.singleton(PROPERTY_ID_2), temporalInfoMap); Assert.assertEquals(1, provider.populateResources(Collections.singleton(resource), request, null).size()); Assert.assertEquals( 3L, resource.getPropertyValue(PROPERTY_ID_2)); verify(connectionFactory, connection, statement, resultSet); }
@Test public void testPopulateResources_multi() throws Exception { ConnectionFactory connectionFactory = createNiceMock(ConnectionFactory.class); Connection connection = createNiceMock(Connection.class); Statement statement = createNiceMock(Statement.class); ResultSet resultSet = createNiceMock(ResultSet.class); expect(connectionFactory.getConnection()).andReturn(connection).once(); expect(connection.createStatement()).andReturn(statement).once(); expect(statement.executeQuery(anyObject(String.class))).andReturn(resultSet).once(); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcQueueTime_avg_time"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999990L); expect(resultSet.getNString("MetricValue")).andReturn("0"); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcQueueTime_avg_time"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999991L); expect(resultSet.getNString("MetricValue")).andReturn("1"); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcSlowResponse_num_ops"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999992L); expect(resultSet.getNString("MetricValue")).andReturn("2"); expect(resultSet.next()).andReturn(true); expect(resultSet.getString("RecordTypeContext")).andReturn("rpc"); expect(resultSet.getString("RecordTypeName")).andReturn("rpc"); expect(resultSet.getString("TagPairs")).andReturn(""); expect(resultSet.getString("MetricName")).andReturn("RpcSlowResponse_num_ops"); expect(resultSet.getString("ServiceName")).andReturn("datanode"); expect(resultSet.getString("NodeName")).andReturn("host1"); expect(resultSet.getLong("RecordTimeStamp")).andReturn(999993L); expect(resultSet.getNString("MetricValue")).andReturn("3"); expect(resultSet.next()).andReturn(false); replay(connectionFactory, connection, statement, resultSet); SQLPropertyProvider provider = new SQLPropertyProvider( PropertyHelper.getGangliaPropertyIds(Resource.Type.HostComponent), new TestHostInfoProvider(), CLUSTER_NAME_PROPERTY_ID, HOST_NAME_PROPERTY_ID, COMPONENT_NAME_PROPERTY_ID, SERVICE_NAME_PROPERTY_ID, connectionFactory); Resource resource = new ResourceImpl(Resource.Type.HostComponent); resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1"); resource.setProperty(HOST_NAME_PROPERTY_ID, "domU-12-31-39-0E-34-E1.compute-1.internal"); resource.setProperty(COMPONENT_NAME_PROPERTY_ID, "DATANODE"); resource.setProperty(SERVICE_NAME_PROPERTY_ID, "HDFS"); Map<String, TemporalInfo> temporalInfoMap = new HashMap<String, TemporalInfo>(); temporalInfoMap.put(PROPERTY_ID_1, new TemporalInfoImpl(10L, 20L, 1L)); Set<String> propertyIds = new LinkedHashSet<String>(); propertyIds.add(PROPERTY_ID_1); propertyIds.add(PROPERTY_ID_2); Request request = PropertyHelper.getReadRequest(propertyIds, temporalInfoMap); Assert.assertEquals(1, provider.populateResources(Collections.singleton(resource), request, null).size()); Assert.assertTrue(resource.getPropertyValue(PROPERTY_ID_1) instanceof Number[][]); Number[][] datapoints = (Number[][]) resource.getPropertyValue(PROPERTY_ID_1); for (int i = 0; i < datapoints.length; ++i) { Assert.assertEquals((long) i, datapoints[i][0]); Assert.assertEquals((999990L + i) / 1000, datapoints[i][1]); } Assert.assertEquals( 3L, resource.getPropertyValue(PROPERTY_ID_2)); verify(connectionFactory, connection, statement, resultSet); }