/*
 * 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.server.controller.metrics;

import static org.easymock.EasyMock.anyObject;
import static org.easymock.EasyMock.createNiceMock;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.internal.PropertyInfo;
import org.apache.ambari.server.controller.internal.ResourceImpl;
import org.apache.ambari.server.controller.jmx.JMXHostProvider;
import org.apache.ambari.server.controller.jmx.JMXPropertyProvider;
import org.apache.ambari.server.controller.jmx.TestStreamProvider;
import org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService;
import org.apache.ambari.server.controller.spi.Request;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.SystemException;
import org.apache.ambari.server.controller.spi.TemporalInfo;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.security.TestAuthenticationFactory;
import org.apache.ambari.server.security.authorization.AuthorizationException;
import org.apache.ambari.server.security.authorization.AuthorizationHelperInitializer;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.services.MetricsRetrievalService;
import org.apache.ambari.server.utils.SynchronousThreadPoolExecutor;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.security.core.context.SecurityContextHolder;

import com.google.inject.Guice;
import com.google.inject.Injector;

/**
 * JMX property provider tests.
 */
public class JMXPropertyProviderTest {
  protected static final String CLUSTER_NAME_PROPERTY_ID = PropertyHelper.getPropertyId("HostRoles", "cluster_name");
  protected static final String HOST_COMPONENT_HOST_NAME_PROPERTY_ID = PropertyHelper.getPropertyId("HostRoles", "host_name");
  protected static final String HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID = PropertyHelper.getPropertyId("HostRoles", "component_name");
  protected static final String HOST_COMPONENT_STATE_PROPERTY_ID = PropertyHelper.getPropertyId("HostRoles", "state");

  public static final int NUMBER_OF_RESOURCES = 400;
  private static final int METRICS_SERVICE_TIMEOUT = 10;

  public static final Map<String, Map<String, PropertyInfo>> jmxPropertyIds = PropertyHelper.getJMXPropertyIds(Resource.Type.HostComponent);
  public static final Map<String, Map<String, PropertyInfo>> jmxPropertyIdsWithHAState;

  static {
    jmxPropertyIdsWithHAState = new HashMap<>(jmxPropertyIds);
    jmxPropertyIdsWithHAState.get("NAMENODE").put("metrics/dfs/FSNamesystem/HAState", new PropertyInfo("Hadoop:service=NameNode,name=FSNamesystem.tag#HAState", false, true));
  }

  private static MetricPropertyProviderFactory metricPropertyProviderFactory;
  private static MetricsRetrievalService metricsRetrievalService;

  @BeforeClass
  public static void setupClass() throws TimeoutException {
    Injector injector = Guice.createInjector(new InMemoryDefaultTestModule());

    // disable request TTL for these tests
    Configuration configuration = injector.getInstance(Configuration.class);
    configuration.setProperty(Configuration.METRIC_RETRIEVAL_SERVICE_REQUEST_TTL_ENABLED.getKey(),
        "false");

    JMXPropertyProvider.init(configuration);

    metricPropertyProviderFactory = injector.getInstance(MetricPropertyProviderFactory.class);

    metricsRetrievalService = injector.getInstance(
        MetricsRetrievalService.class);

    metricsRetrievalService.startAsync();
    metricsRetrievalService.awaitRunning(METRICS_SERVICE_TIMEOUT, TimeUnit.SECONDS);
    metricsRetrievalService.setThreadPoolExecutor(new SynchronousThreadPoolExecutor());
  }

  @AfterClass
  public static void stopService() throws TimeoutException {
    if (metricsRetrievalService != null && metricsRetrievalService.isRunning()) {
      metricsRetrievalService.stopAsync();
      metricsRetrievalService.awaitTerminated(METRICS_SERVICE_TIMEOUT, TimeUnit.SECONDS);
    }
  }

  @Before
  public void setUpCommonMocks() throws AmbariException, NoSuchFieldException, IllegalAccessException {
    AmbariManagementController amc = createNiceMock(AmbariManagementController.class);
    Field field = AmbariServer.class.getDeclaredField("clusterController");
    field.setAccessible(true);
    field.set(null, amc);
    Clusters clusters = createNiceMock(Clusters.class);
    Cluster cluster = createNiceMock(Cluster.class);
    expect(amc.getClusters()).andReturn(clusters).anyTimes();
    expect(clusters.getCluster(CLUSTER_NAME_PROPERTY_ID)).andReturn(cluster).anyTimes();
    expect(cluster.getResourceId()).andReturn(2L).anyTimes();

    try {
      expect(clusters.getCluster(anyObject(String.class))).andReturn(cluster).anyTimes();
    } catch (AmbariException e) {
      e.printStackTrace();
    }

    replay(amc, clusters, cluster);
    AuthorizationHelperInitializer.viewInstanceDAOReturningNull();
  }

  @After
  public void clearAuthentication() {
    SecurityContextHolder.getContext().setAuthentication(null);
  }

  @Test
  public void testJMXPropertyProviderAsClusterAdministrator() throws Exception {
    //Setup user with Role 'ClusterAdministrator'.
    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createClusterAdministrator("ClusterAdmin", 2L));
    testPopulateResources();
    testPopulateResources_singleProperty();
    testPopulateResources_category();
    testPopulateResourcesWithUnknownPort();
    testPopulateResourcesUnhealthyResource();
    testPopulateResourcesMany();
    testPopulateResourcesTimeout();
    testPopulateResources_HAState_request();
  }

  @Test
  public void testJMXPropertyProviderAsAdministrator() throws Exception {
    //Setup user with Role 'Administrator'
    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createAdministrator("Admin"));
    testPopulateResources();
    testPopulateResources_singleProperty();
    testPopulateResources_category();
    testPopulateResourcesWithUnknownPort();
    testPopulateResourcesUnhealthyResource();
    testPopulateResourcesMany();
    testPopulateResourcesTimeout();
    testPopulateResources_HAState_request();
  }

  @Test
  public void testJMXPropertyProviderAsServiceAdministrator() throws Exception {
    //Setup user with 'ServiceAdministrator'
    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createServiceAdministrator("ServiceAdmin", 2L));
    testPopulateResources();
    testPopulateResources_singleProperty();
    testPopulateResources_category();
    testPopulateResourcesWithUnknownPort();
    testPopulateResourcesUnhealthyResource();
    testPopulateResourcesMany();
    testPopulateResourcesTimeout();
    testPopulateResources_HAState_request();
  }

  @Test(expected = AuthorizationException.class)
  public void testJMXPropertyProviderAsViewUser() throws Exception {
    // Setup user with 'ViewUser'
    // ViewUser doesn't have the 'CLUSTER_VIEW_METRICS', 'HOST_VIEW_METRICS' and 'SERVICE_VIEW_METRICS', thus
    // can't retrieve the Metrics.
    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createViewUser("ViewUser", 2L));
    testPopulateResources();
    testPopulateResources_singleProperty();
    testPopulateResources_category();
    testPopulateResourcesWithUnknownPort();
    testPopulateResourcesUnhealthyResource();
    testPopulateResourcesMany();
    testPopulateResourcesTimeout();
    testPopulateResources_HAState_request();
  }

  public void testPopulateResources() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(false);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();
    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      jmxPropertyIdsWithHAState,
      streamProvider,
      hostProvider,
      metricsHostProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"));

    // set the provider timeout to 5000 millis
    propertyProvider.setPopulateTimeout(5000);

    // namenode
    Resource resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-0e-34-e1.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "NAMENODE");
    resource.setProperty(HOST_COMPONENT_STATE_PROPERTY_ID, "STARTED");

    // request with an empty set should get all supported properties
    Request request = PropertyHelper.getReadRequest(Collections.emptySet());
    Assert.assertEquals(1, propertyProvider.populateResources(Collections.singleton(resource), request, null).size());

    List<String> expectedSpecs = new ArrayList<>();
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx"));
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx?get=Hadoop:service=NameNode,name=FSNamesystem::tag.HAState"));
    Assert.assertEquals(expectedSpecs, streamProvider.getSpecs());

    // see test/resources/hdfs_namenode_jmx.json for values
    Assert.assertEquals(13670605, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/rpc", "ReceivedBytes")));
    Assert.assertEquals(28, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/dfs/namenode", "CreateFileOps")));
    Assert.assertEquals(1006632960, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryMax")));
    Assert.assertEquals(473433016, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryUsed")));
    Assert.assertEquals(136314880, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryMax")));
    Assert.assertEquals(23634400, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryUsed")));
    Assert.assertEquals(887717691390L, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/dfs/FSNamesystem", "CapacityTotal")));
    Assert.assertEquals(184320, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/dfs/FSNamesystem", "CapacityUsed")));
    Assert.assertEquals(842207944704L, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/dfs/FSNamesystem", "CapacityRemaining")));
    Assert.assertEquals(45509562366L, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/dfs/FSNamesystem", "CapacityNonDFSUsed")));

    // datanode
    resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-14-ee-b3.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "DATANODE");

    // request with an empty set should get all supported properties
    request = PropertyHelper.getReadRequest(Collections.emptySet());

    propertyProvider.populateResources(Collections.singleton(resource), request, null);

    Assert.assertEquals(propertyProvider.getSpec("http", "domu-12-31-39-14-ee-b3.compute-1.internal", "50075", "/jmx"), streamProvider.getLastSpec());

    // see test/resources/hdfs_datanode_jmx.json for values
    Assert.assertEquals(856, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/rpc", "ReceivedBytes")));
    Assert.assertEquals(954466304, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryMax")));
    Assert.assertEquals(9772616, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryUsed")));
    Assert.assertEquals(136314880, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryMax")));
    Assert.assertEquals(21933376, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryUsed")));

    // hbase master
    resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-14-ee-b3.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "HBASE_MASTER");
    resource.setProperty(HOST_COMPONENT_STATE_PROPERTY_ID, "STARTED");

    // only ask for specific properties
    Set<String> properties = new HashSet<>();
    properties.add(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryMax"));
    properties.add(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryUsed"));
    properties.add(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryMax"));
    properties.add(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryUsed"));
    properties.add(PropertyHelper.getPropertyId("metrics/load", "AverageLoad"));
    request = PropertyHelper.getReadRequest(properties);

    propertyProvider.populateResources(Collections.singleton(resource), request, null);

    Assert.assertEquals(propertyProvider.getSpec("http", "domu-12-31-39-14-ee-b3.compute-1.internal", "60010", "/jmx"), streamProvider.getLastSpec());

    Assert.assertEquals(9, PropertyHelper.getProperties(resource).size());
    Assert.assertEquals(1069416448, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryMax")));
    Assert.assertEquals(4806976, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryUsed")));
    Assert.assertEquals(136314880, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryMax")));
    Assert.assertEquals(28971240, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryUsed")));
    Assert.assertEquals(3.0, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/load", "AverageLoad")));

    Assert.assertNull(resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "gcCount")));
  }

  public void testPopulateResources_singleProperty() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(false);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();

    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      PropertyHelper.getJMXPropertyIds(Resource.Type.HostComponent),
      streamProvider,
      hostProvider,
      metricsHostProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"));

    // namenode
    Resource resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-0e-34-e1.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "NAMENODE");
    resource.setProperty(HOST_COMPONENT_STATE_PROPERTY_ID, "STARTED");

    // only ask for one property
    Map<String, TemporalInfo> temporalInfoMap = new HashMap<>();
    Request request = PropertyHelper.getReadRequest(Collections.singleton("metrics/rpc/ReceivedBytes"), temporalInfoMap);

    Assert.assertEquals(1, propertyProvider.populateResources(Collections.singleton(resource), request, null).size());

    Assert.assertEquals(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx"), streamProvider.getLastSpec());

    // see test/resources/hdfs_namenode_jmx.json for values
    Assert.assertEquals(13670605, resource.getPropertyValue("metrics/rpc/ReceivedBytes"));
    Assert.assertNull(resource.getPropertyValue("metrics/dfs/namenode/CreateFileOps"));
  }

  public void testPopulateResources_category() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(false);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();

    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      PropertyHelper.getJMXPropertyIds(Resource.Type.HostComponent),
      streamProvider,
      hostProvider,
      metricsHostProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"));

    // namenode
    Resource resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-0e-34-e1.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "NAMENODE");
    resource.setProperty(HOST_COMPONENT_STATE_PROPERTY_ID, "STARTED");

    // request with an empty set should get all supported properties
    // only ask for one property
    Map<String, TemporalInfo> temporalInfoMap = new HashMap<>();
    Request request = PropertyHelper.getReadRequest(Collections.singleton("metrics/dfs"), temporalInfoMap);

    Assert.assertEquals(1, propertyProvider.populateResources(Collections.singleton(resource), request, null).size());

    List<String> expectedSpecs = new ArrayList<>();
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx"));
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx?get=Hadoop:service=NameNode,name=FSNamesystem::tag.HAState"));
    Assert.assertEquals(expectedSpecs, streamProvider.getSpecs());

    // see test/resources/hdfs_namenode_jmx.json for values
    Assert.assertEquals(184320, resource.getPropertyValue("metrics/dfs/FSNamesystem/CapacityUsed"));
    Assert.assertEquals(21, resource.getPropertyValue("metrics/dfs/FSNamesystem/UnderReplicatedBlocks"));
    Assert.assertNull(resource.getPropertyValue("metrics/rpc/ReceivedBytes"));
  }

  public void testPopulateResources_HAState_request() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(false);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();

    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      jmxPropertyIdsWithHAState,
      streamProvider,
      hostProvider,
      metricsHostProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"));

    // namenode
    Resource resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-0e-34-e1.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "NAMENODE");
    resource.setProperty(HOST_COMPONENT_STATE_PROPERTY_ID, "STARTED");

    // request with an empty set should get all supported properties
    // only ask for one property
    Map<String, TemporalInfo> temporalInfoMap = new HashMap<>();
    Request request = PropertyHelper.getReadRequest(Collections.singleton("metrics/dfs/FSNamesystem"), temporalInfoMap);

    Assert.assertEquals(1, propertyProvider.populateResources(Collections.singleton(resource), request, null).size());

    List<String> expectedSpecs = new ArrayList<>();
    expectedSpecs.add(propertyProvider.getSpec("http","domu-12-31-39-0e-34-e1.compute-1.internal","50070","/jmx"));
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx?get=Hadoop:service=NameNode,name=FSNamesystem::tag.HAState"));

    Assert.assertEquals(expectedSpecs, streamProvider.getSpecs());

    // see test/resources/hdfs_namenode_jmx.json for values
    Assert.assertEquals(184320, resource.getPropertyValue("metrics/dfs/FSNamesystem/CapacityUsed"));
    Assert.assertEquals(21, resource.getPropertyValue("metrics/dfs/FSNamesystem/UnderReplicatedBlocks"));
    Assert.assertEquals("customState", resource.getPropertyValue("metrics/dfs/FSNamesystem/HAState"));
    Assert.assertNull(resource.getPropertyValue("metrics/rpc/ReceivedBytes"));

    // namenode
    resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-0e-34-e1.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "NAMENODE");
    resource.setProperty(HOST_COMPONENT_STATE_PROPERTY_ID, "STARTED");

    streamProvider.getSpecs().clear();

    // request with an empty set should get all supported properties
    // only ask for one property
    temporalInfoMap = new HashMap<>();
    request = PropertyHelper.getReadRequest(Collections.singleton("metrics/dfs/FSNamesystem/CapacityUsed"), temporalInfoMap);

    Assert.assertEquals(1, propertyProvider.populateResources(Collections.singleton(resource), request, null).size());

    // HAState isn't requested. It shouldn't be retrieved.
    expectedSpecs.clear();
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx"));
    Assert.assertEquals(expectedSpecs, streamProvider.getSpecs());

    // see test/resources/hdfs_namenode_jmx.json for values
    Assert.assertEquals(184320, resource.getPropertyValue("metrics/dfs/FSNamesystem/CapacityUsed"));
    Assert.assertNull(resource.getPropertyValue("metrics/dfs/FSNamesystem/HAState"));
    Assert.assertNull(resource.getPropertyValue("metrics/rpc/ReceivedBytes"));

  }

  public void testPopulateResourcesWithUnknownPort() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(true);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();

    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      PropertyHelper.getJMXPropertyIds(Resource.Type.HostComponent),
      streamProvider,
      hostProvider,
      metricsHostProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"));

    // namenode
    Resource resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-0e-34-e1.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "NAMENODE");

    // request with an empty set should get all supported properties
    Request request = PropertyHelper.getReadRequest(Collections.emptySet());

    Assert.assertEquals(1, propertyProvider.populateResources(Collections.singleton(resource), request, null).size());

    List<String> expectedSpecs = new ArrayList<>();
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx"));
    expectedSpecs.add(propertyProvider.getSpec("http", "domu-12-31-39-0e-34-e1.compute-1.internal", "50070", "/jmx?get=Hadoop:service=NameNode,name=FSNamesystem::tag.HAState"));
    Assert.assertEquals(expectedSpecs, streamProvider.getSpecs());

    // see test/resources/hdfs_namenode_jmx.json for values
    Assert.assertEquals(13670605, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/rpc", "ReceivedBytes")));
    Assert.assertEquals(28, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/dfs/namenode", "CreateFileOps")));
    Assert.assertEquals(1006632960, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryMax")));
    Assert.assertEquals(473433016, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryUsed")));
    Assert.assertEquals(136314880, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryMax")));
    Assert.assertEquals(23634400, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryUsed")));
  }

  public void testPopulateResourcesUnhealthyResource() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(true);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();

    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      PropertyHelper.getJMXPropertyIds(Resource.Type.HostComponent),
      streamProvider,
      hostProvider,
      metricsHostProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"));

    // namenode
    Resource resource = new ResourceImpl(Resource.Type.HostComponent);

    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-0e-34-e1.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "NAMENODE");
    resource.setProperty(HOST_COMPONENT_STATE_PROPERTY_ID, "INSTALLED");

    // request with an empty set should get all supported properties
    Request request = PropertyHelper.getReadRequest(Collections.emptySet());

    Assert.assertEquals(1, propertyProvider.populateResources(Collections.singleton(resource), request, null).size());

    // Assert that the stream provider was never called.
    Assert.assertNull(streamProvider.getLastSpec());
  }

  public void testPopulateResourcesMany() throws Exception {
    // Set the provider to take 50 millis to return the JMX values
    TestStreamProvider streamProvider = new TestStreamProvider(50L);
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(true);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();
    Set<Resource> resources = new HashSet<>();

    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      PropertyHelper.getJMXPropertyIds(Resource.Type.HostComponent),
      streamProvider,
      hostProvider,
      metricsHostProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"));

    // set the provider timeout to 5000 millis
    propertyProvider.setPopulateTimeout(5000);

    for (int i = 0; i < NUMBER_OF_RESOURCES; ++i) {
      // datanode
      Resource resource = new ResourceImpl(Resource.Type.HostComponent);
      resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
      resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-14-ee-b3.compute-1.internal");
      resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "DATANODE");
      resource.setProperty("unique_id", i);

      resources.add(resource);
    }
    // request with an empty set should get all supported properties
    Request request = PropertyHelper.getReadRequest(Collections.emptySet());

    Set<Resource> resourceSet = propertyProvider.populateResources(resources, request, null);

    Assert.assertEquals(NUMBER_OF_RESOURCES, resourceSet.size());
    for (Resource resource : resourceSet) {
      // see test/resources/hdfs_datanode_jmx.json for values
      Assert.assertEquals(856, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/rpc", "ReceivedBytes")));
      Assert.assertEquals(954466304, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryMax")));
      Assert.assertEquals(9772616, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryUsed")));
      Assert.assertEquals(136314880, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryMax")));
      Assert.assertEquals(21933376, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "NonHeapMemoryUsed")));
    }
  }

  public void testPopulateResourcesTimeout() throws Exception {
    // Set the provider to take 100 millis to return the JMX values
    TestStreamProvider streamProvider = new TestStreamProvider(100L);
    TestJMXHostProvider hostProvider = new TestJMXHostProvider(true);
    TestMetricHostProvider metricsHostProvider = new TestMetricHostProvider();
    Set<Resource> resources = new HashSet<>();

    JMXPropertyProvider propertyProvider = metricPropertyProviderFactory.createJMXPropertyProvider(
      PropertyHelper.getJMXPropertyIds(Resource.Type.HostComponent),
      streamProvider,
      hostProvider,
      metricsHostProvider,
      "HostRoles/cluster_name",
      "HostRoles/host_name",
      "HostRoles/component_name",
      "HostRoles/state");

    // set the provider timeout to 50 millis
    propertyProvider.setPopulateTimeout(50L);

    // datanode
    Resource resource = new ResourceImpl(Resource.Type.HostComponent);
    resource.setProperty(CLUSTER_NAME_PROPERTY_ID, "c1");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "domu-12-31-39-14-ee-b3.compute-1.internal");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "DATANODE");

    resources.add(resource);

    // request with an empty set should get all supported properties
    Request request = PropertyHelper.getReadRequest(Collections.emptySet());

    Set<Resource> resourceSet = propertyProvider.populateResources(resources, request, null);

    // make sure that the thread running the stream provider has completed
    Thread.sleep(150L);

    Assert.assertEquals(0, resourceSet.size());

    // assert that properties never get set on the resource
    Assert.assertNull(resource.getPropertyValue("metrics/rpc/ReceivedBytes"));
    Assert.assertNull(resource.getPropertyValue("metrics/jvm/HeapMemoryMax"));
    Assert.assertNull(resource.getPropertyValue("metrics/jvm/HeapMemoryUsed"));
    Assert.assertNull(resource.getPropertyValue("metrics/jvm/NonHeapMemoryMax"));
    Assert.assertNull(resource.getPropertyValue("metrics/jvm/NonHeapMemoryUsed"));
  }

