/*
 * 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.internal;

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.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.H2DatabaseCleaner;
import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
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.jmx.JMXPropertyProvider;
import org.apache.ambari.server.controller.jmx.TestStreamProvider;
import org.apache.ambari.server.controller.metrics.JMXPropertyProviderTest;
import org.apache.ambari.server.controller.metrics.MetricsServiceProvider;
import org.apache.ambari.server.controller.metrics.ganglia.GangliaPropertyProviderTest.TestGangliaHostProvider;
import org.apache.ambari.server.controller.metrics.ganglia.GangliaPropertyProviderTest.TestGangliaServiceProvider;
import org.apache.ambari.server.controller.metrics.timeline.MetricsRequestHelper;
import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheEntryFactory;
import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.PropertyProvider;
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.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.orm.GuiceJpaInitializer;
import org.apache.ambari.server.orm.InMemoryDefaultTestModule;
import org.apache.ambari.server.orm.OrmTestHelper;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.security.TestAuthenticationFactory;
import org.apache.ambari.server.security.authorization.AuthorizationException;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.services.MetricsRetrievalService;
import org.apache.ambari.server.state.stack.Metric;
import org.apache.ambari.server.utils.SynchronousThreadPoolExecutor;
import org.junit.After;
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.Binder;
import com.google.inject.Guice;
import com.google.inject.Injector;
import com.google.inject.Module;
import com.google.inject.util.Modules;

/**
 * Tests the stack defined property provider.
 */
public class StackDefinedPropertyProviderTest {
  private static final String HOST_COMPONENT_HOST_NAME_PROPERTY_ID = "HostRoles/host_name";
  private static final String HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID = "HostRoles/component_name";
  private static final String HOST_COMPONENT_STATE_PROPERTY_ID = "HostRoles/state";
  private static final String CLUSTER_NAME_PROPERTY_ID = PropertyHelper.getPropertyId("HostRoles", "cluster_name");
  private static final int METRICS_SERVICE_TIMEOUT = 10;

  private Clusters clusters = null;
  private Injector injector = null;
  private OrmTestHelper helper = null;

  private static TimelineMetricCacheEntryFactory cacheEntryFactory;
  private static TimelineMetricCacheProvider cacheProvider;
  private static MetricsRetrievalService metricsRetrievalService;

  @BeforeClass
  public static void setupCache() {
    cacheEntryFactory = new TimelineMetricCacheEntryFactory(new Configuration());
    cacheProvider = new TimelineMetricCacheProvider(new Configuration(), cacheEntryFactory);
    Injector injector = Guice.createInjector(new InMemoryDefaultTestModule());
    JMXPropertyProvider.init(injector.getInstance(Configuration.class));
  }

  public class TestModuleWithCacheProvider implements Module {
    @Override
    public void configure(Binder binder) {
      binder.bind(TimelineMetricCacheProvider.class).toInstance(cacheProvider);
    }
  }

  @Before
  public void setup() throws Exception {
    InMemoryDefaultTestModule module = new InMemoryDefaultTestModule();
    // Use the same cache provider to ensure there is only once instance of
    // Cache available. The @org.ehcache.CacheManager is a singleton and
    // does not allow multiple instance with same cache name to be registered.
    injector = Guice.createInjector(Modules.override(module).with(new TestModuleWithCacheProvider()));
    injector.getInstance(GuiceJpaInitializer.class);
    StackDefinedPropertyProvider.init(injector);

    metricsRetrievalService = injector.getInstance(
        MetricsRetrievalService.class);

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

    helper = injector.getInstance(OrmTestHelper.class);

    clusters = injector.getInstance(Clusters.class);
    StackId stackId = new StackId("HDP-2.0.5");

    clusters.addCluster("c2", stackId);

    Cluster cluster = clusters.getCluster("c2");

    cluster.setDesiredStackVersion(stackId);
    RepositoryVersionEntity repositoryVersion = helper.getOrCreateRepositoryVersion(stackId, stackId.getStackVersion());
    Service service = cluster.addService("HDFS", repositoryVersion);
    service.addServiceComponent("NAMENODE");
    service.addServiceComponent("DATANODE");
    service.addServiceComponent("JOURNALNODE");

    service = cluster.addService("YARN", repositoryVersion);
    service.addServiceComponent("RESOURCEMANAGER");

    service = cluster.addService("HBASE", repositoryVersion);
    service.addServiceComponent("HBASE_MASTER");
    service.addServiceComponent("HBASE_REGIONSERVER");

    stackId = new StackId("HDP-2.1.1");
    repositoryVersion = helper.getOrCreateRepositoryVersion(stackId, stackId.getStackVersion());

    service = cluster.addService("STORM", repositoryVersion);
    service.addServiceComponent("STORM_REST_API");

    clusters.addHost("h1");
    Host host = clusters.getHost("h1");
    Map<String, String> hostAttributes = new HashMap<>();
    hostAttributes.put("os_family", "redhat");
    hostAttributes.put("os_release_version", "6.3");
    host.setHostAttributes(hostAttributes);

    clusters.mapHostToCluster("h1", "c2");

    // Setting up Mocks for Controller, Clusters etc, queried as part of user's Role context
    // while fetching Metrics.
    AmbariManagementController amc = createNiceMock(AmbariManagementController.class);
    Field field = AmbariServer.class.getDeclaredField("clusterController");
    field.setAccessible(true);
    field.set(null, amc);
    Clusters clustersMock = createNiceMock(Clusters.class);
    Cluster clusterMock = createNiceMock(Cluster.class);
    expect(amc.getAmbariEventPublisher()).andReturn(createNiceMock(AmbariEventPublisher.class)).anyTimes();
    expect(amc.getClusters()).andReturn(clustersMock).anyTimes();
    expect(clustersMock.getCluster(CLUSTER_NAME_PROPERTY_ID)).andReturn(clusterMock).anyTimes();
    expect(clusterMock.getResourceId()).andReturn(2L).anyTimes();
    try {
      expect(clustersMock.getCluster(anyObject(String.class))).andReturn(clusterMock).anyTimes();
    } catch (AmbariException e) {
      e.printStackTrace();
    }
    replay(amc, clustersMock, clusterMock);
  }

  @After
  public void teardown() throws Exception {
    if (metricsRetrievalService != null && metricsRetrievalService.isRunning()) {
      metricsRetrievalService.stopAsync();
      metricsRetrievalService.awaitTerminated(METRICS_SERVICE_TIMEOUT, TimeUnit.SECONDS);
    }
    H2DatabaseCleaner.clearDatabaseAndStopPersistenceService(injector);
  }

  @Test
  public void testStackDefinedPropertyProviderAsClusterAdministrator() throws Exception {
    //Setup user with Role 'ClusterAdministrator'.
    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createClusterAdministrator("ClusterAdmin", 2L));

    testPopulateHostComponentResources();
    testCustomProviders();
    testPopulateResources_HDP2();
    testPopulateResources_HDP2_params();
    testPopulateResources_HDP2_params_singleProperty();
    testPopulateResources_HDP2_params_category();
    testPopulateResources_HDP2_params_category2();
    testPopulateResources_jmx_JournalNode();
    testPopulateResources_jmx_Storm();
    testPopulateResources_NoRegionServer();
    testPopulateResources_HBaseMaster2();
    testPopulateResources_params_category5();
    testPopulateResources_ganglia_JournalNode();
    testPopulateResources_resourcemanager_clustermetrics();
    testPopulateResourcesWithAggregateFunctionMetrics();
  }

  @Test
  public void testStackDefinedPropertyProviderAsAdministrator() throws Exception {
    //Setup user with Role 'Administrator'
    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createAdministrator("Admin"));

    testPopulateHostComponentResources();
    testCustomProviders();
    testPopulateResources_HDP2();
    testPopulateResources_HDP2_params();
    testPopulateResources_HDP2_params_singleProperty();
    testPopulateResources_HDP2_params_category();
    testPopulateResources_HDP2_params_category2();
    testPopulateResources_jmx_JournalNode();
    testPopulateResources_jmx_Storm();
    testPopulateResources_NoRegionServer();
    testPopulateResources_HBaseMaster2();
    testPopulateResources_params_category5();
    testPopulateResources_ganglia_JournalNode();
    testPopulateResources_resourcemanager_clustermetrics();
    testPopulateResourcesWithAggregateFunctionMetrics();
  }

  @Test
  public void testStackDefinedPropertyProviderAsServiceAdministrator() throws Exception {
    //Setup user with 'ServiceAdministrator'
    SecurityContextHolder.getContext().setAuthentication(TestAuthenticationFactory.createServiceAdministrator("ServiceAdmin", 2L));

    testPopulateHostComponentResources();
    testCustomProviders();
    testPopulateResources_HDP2();
    testPopulateResources_HDP2_params();
    testPopulateResources_HDP2_params_singleProperty();
    testPopulateResources_HDP2_params_category();
    testPopulateResources_HDP2_params_category2();
    testPopulateResources_jmx_JournalNode();
    testPopulateResources_jmx_Storm();
    testPopulateResources_NoRegionServer();
    testPopulateResources_HBaseMaster2();
    testPopulateResources_params_category5();
    testPopulateResources_ganglia_JournalNode();
    testPopulateResources_resourcemanager_clustermetrics();
    testPopulateResourcesWithAggregateFunctionMetrics();
  }

  @Test(expected = AuthorizationException.class)
  public void testStackDefinedPropertyProviderAsViewUser() 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));

    testPopulateHostComponentResources();
    testCustomProviders();
    testPopulateResources_HDP2();
    testPopulateResources_HDP2_params();
    testPopulateResources_HDP2_params_singleProperty();
    testPopulateResources_HDP2_params_category();
    testPopulateResources_HDP2_params_category2();
    testPopulateResources_jmx_JournalNode();
    testPopulateResources_jmx_Storm();
    testPopulateResources_NoRegionServer();
    testPopulateResources_HBaseMaster2();
    testPopulateResources_params_category5();
    testPopulateResources_ganglia_JournalNode();
    testPopulateResources_resourcemanager_clustermetrics();
    testPopulateResourcesWithAggregateFunctionMetrics();
  }

  public void testPopulateHostComponentResources() throws Exception {
    TestJMXProvider tj = new TestJMXProvider(true);
    JMXPropertyProviderTest.TestMetricHostProvider tm = new JMXPropertyProviderTest.TestMetricHostProvider();
    TestGangliaServiceProvider serviceProvider = new TestGangliaServiceProvider();

    StackDefinedPropertyProvider sdpp = new StackDefinedPropertyProvider(
        Resource.Type.HostComponent, tj, tm, serviceProvider,
        new CombinedStreamProvider(), "HostRoles/cluster_name",
        "HostRoles/host_name", "HostRoles/component_name", "HostRoles/state",
        null, null);

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

    resource.setProperty("HostRoles/cluster_name", "c2");
    resource.setProperty("HostRoles/host_name", "h1");
    resource.setProperty("HostRoles/component_name", "NAMENODE");
    resource.setProperty("HostRoles/state", "STARTED");

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

    Set<Resource> set = sdpp.populateResources(Collections.singleton(resource), request, null);
    Assert.assertEquals(1, set.size());

    Resource res = set.iterator().next();

    Map<String, Map<String, Object>> values = res.getPropertiesMap();

    Assert.assertTrue("Expected JMX metric 'metrics/dfs/FSNamesystem'", values.containsKey("metrics/dfs/FSNamesystem"));
    Assert.assertTrue("Expected JMX metric 'metrics/dfs/namenode'", values.containsKey("metrics/dfs/namenode"));
    Assert.assertTrue("Expected Ganglia metric 'metrics/rpcdetailed/client'", values.containsKey("metrics/rpcdetailed/client"));
    Assert.assertTrue("Expected Ganglia metric 'metrics/rpc/client'", values.containsKey("metrics/rpc/client"));
  }


  public void testCustomProviders() throws Exception {

    StackDefinedPropertyProvider sdpp = new StackDefinedPropertyProvider(
      Resource.Type.HostComponent, null, null, null, new CombinedStreamProvider(),
      "HostRoles/cluster_name", "HostRoles/host_name", "HostRoles/component_name", "HostRoles/state",
      new EmptyPropertyProvider(), new EmptyPropertyProvider());

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

    resource.setProperty("HostRoles/cluster_name", "c2");
    resource.setProperty("HostRoles/host_name", "h1");
    resource.setProperty("HostRoles/component_name", "DATANODE");
    resource.setProperty("HostRoles/state", "STARTED");

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

    Set<Resource> set = sdpp.populateResources(Collections.singleton(resource), request, null);
    Assert.assertEquals(1, set.size());

    Resource res = set.iterator().next();

    Map<String, Map<String, Object>> values = res.getPropertiesMap();
    Assert.assertTrue(values.containsKey("foo/type1"));
    Assert.assertTrue(values.containsKey("foo/type2"));
    Assert.assertTrue(values.containsKey("foo/type3"));
    Assert.assertFalse(values.containsKey("foo/type4"));

    Assert.assertTrue(values.get("foo/type1").containsKey("name"));
    Assert.assertTrue(values.get("foo/type2").containsKey("name"));
    Assert.assertTrue(values.get("foo/type3").containsKey("name"));

    Assert.assertEquals("value1", values.get("foo/type1").get("name"));

  }


  private static class CombinedStreamProvider extends URLStreamProvider {

    public CombinedStreamProvider() {
      super(1000, 1000, ComponentSSLConfiguration.instance());
    }

    @Override
    public InputStream readFrom(String spec) throws IOException {
      if (spec.indexOf("jmx") > -1) {
        // jmx
        return ClassLoader.getSystemResourceAsStream("hdfs_namenode_jmx.json");
      } else {
        // ganglia
        return ClassLoader.getSystemResourceAsStream("temporal_ganglia_data.txt");
      }
    }

    @Override
    public InputStream readFrom(String spec, String requestMethod, String params)
      throws IOException {
      return readFrom(spec);
    }
  }

  private static class EmptyPropertyProvider implements PropertyProvider {

    @Override
    public Set<Resource> populateResources(Set<Resource> resources,
                                           Request request, Predicate predicate) throws SystemException {
      // TODO Auto-generated method stub
      return null;
    }

    @Override
    public Set<String> checkPropertyIds(Set<String> propertyIds) {
      // TODO Auto-generated method stub
      return null;
    }

  }

  /**
   * Test for empty constructor.  Public since instantiated via reflection.
   */
  public static class CustomMetricProvider1 implements PropertyProvider {
    @Override
    public Set<Resource> populateResources(Set<Resource> resources,
                                           Request request, Predicate predicate) throws SystemException {

      for (Resource r : resources) {
        r.setProperty("foo/type1/name", "value1");
      }

      return resources;
    }

    @Override
    public Set<String> checkPropertyIds(Set<String> propertyIds) {
      return Collections.emptySet();
    }
  }

  /**
   * Test map constructors.  Public since instantiated via reflection.
   */
  public static class CustomMetricProvider2 implements PropertyProvider {
    private Map<String, String> providerProperties = null;

    public CustomMetricProvider2(Map<String, String> properties, Map<String, Metric> metrics) {
      providerProperties = properties;
    }

    @Override
    public Set<Resource> populateResources(Set<Resource> resources,
                                           Request request, Predicate predicate) throws SystemException {
      for (Resource r : resources) {
        r.setProperty("foo/type2/name", providerProperties.get("Type2.Metric.Name"));
      }
      return resources;
    }

    @Override
    public Set<String> checkPropertyIds(Set<String> propertyIds) {
      return Collections.emptySet();
    }
  }

  /**
   * Test singleton accessor.  Public since instantiated via reflection.
   */
  public static class CustomMetricProvider3 implements PropertyProvider {
    private static CustomMetricProvider3 instance = null;
    private Map<String, String> providerProperties = new HashMap<>();

    public static CustomMetricProvider3 getInstance(Map<String, String> properties, Map<String, Metric> metrics) {
      if (null == instance) {
        instance = new CustomMetricProvider3();
        instance.providerProperties.putAll(properties);
      }
      return instance;
    }

    @Override
    public Set<Resource> populateResources(Set<Resource> resources,
                                           Request request, Predicate predicate) throws SystemException {
      for (Resource r : resources) {
        r.setProperty("foo/type3/name", providerProperties.get("Type3.Metric.Name"));
      }
      return resources;
    }

    @Override
    public Set<String> checkPropertyIds(Set<String> propertyIds) {
      return Collections.emptySet();
    }
  }

  public void testPopulateResources_HDP2() throws Exception {

    URLStreamProvider streamProvider = new TestStreamProvider();
    TestJMXProvider hostProvider = new TestJMXProvider(true);
    JMXPropertyProviderTest.TestMetricHostProvider metricsHostProvider = new JMXPropertyProviderTest.TestMetricHostProvider();
    TestGangliaServiceProvider serviceProvider = new TestGangliaServiceProvider();

    StackDefinedPropertyProvider propertyProvider = new StackDefinedPropertyProvider(
      Resource.Type.HostComponent,
      hostProvider,
      metricsHostProvider,
      serviceProvider,
      streamProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"),
      new EmptyPropertyProvider(),
      new EmptyPropertyProvider());

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

    resource.setProperty("HostRoles/cluster_name", "c2");
    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, "RESOURCEMANAGER");
    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());

    // see test/resources/resourcemanager_jmx.json for values
    Assert.assertEquals(6, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AggregateContainersAllocated")));
    Assert.assertEquals(6, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AggregateContainersReleased")));
    Assert.assertEquals(8192, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableMB")));
    Assert.assertEquals(1, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableVCores")));
    Assert.assertEquals(2, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AppsSubmitted")));

    Assert.assertEquals(1, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/ClusterMetrics", "NumActiveNMs")));
    Assert.assertEquals(0, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/ClusterMetrics", "NumDecommissionedNMs")));
    Assert.assertEquals(0, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/ClusterMetrics", "NumLostNMs")));
    Assert.assertEquals(0, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/ClusterMetrics", "NumUnhealthyNMs")));
    Assert.assertEquals(0, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/ClusterMetrics", "NumRebootedNMs")));

    Assert.assertEquals(932118528, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/jvm", "HeapMemoryMax")));

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

    resource.setProperty("HostRoles/cluster_name", "c2");
    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 = PropertyHelper.getReadRequest(Collections.emptySet());

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

  public void testPopulateResources_HDP2_params() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXProvider hostProvider = new TestJMXProvider(false);
    JMXPropertyProviderTest.TestMetricHostProvider metricsHostProvider = new JMXPropertyProviderTest.TestMetricHostProvider();
    TestGangliaServiceProvider serviceProvider = new TestGangliaServiceProvider();

    StackDefinedPropertyProvider propertyProvider = new StackDefinedPropertyProvider(
      Resource.Type.HostComponent,
      hostProvider,
      metricsHostProvider,
      serviceProvider,
      streamProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"),
      new EmptyPropertyProvider(),
      new EmptyPropertyProvider());

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

    resource.setProperty("HostRoles/cluster_name", "c2");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "h1");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "RESOURCEMANAGER");
    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());

    // see test/resources/resourcemanager_jmx.json for values
    Assert.assertEquals(6, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AggregateContainersAllocated")));
    Assert.assertEquals(6, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AggregateContainersReleased")));
    Assert.assertEquals(8192, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableMB")));
    Assert.assertEquals(1, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableVCores")));
    Assert.assertEquals(2, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AppsSubmitted")));

    Assert.assertEquals(15, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/default", "AggregateContainersAllocated")));
    Assert.assertEquals(12, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/default", "AggregateContainersReleased")));
    Assert.assertEquals(8192, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/default", "AvailableMB")));
    Assert.assertEquals(1, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/default", "AvailableVCores")));
    Assert.assertEquals(47, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/default", "AppsSubmitted")));

    Assert.assertEquals(4, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/second_queue", "AggregateContainersAllocated")));
    Assert.assertEquals(4, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/second_queue", "AggregateContainersReleased")));
    Assert.assertEquals(6048, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/second_queue", "AvailableMB")));
    Assert.assertEquals(1, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/second_queue", "AvailableVCores")));
    Assert.assertEquals(1, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/second_queue", "AppsSubmitted")));
  }


  public void testPopulateResources_HDP2_params_singleProperty() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXProvider hostProvider = new TestJMXProvider(false);
    JMXPropertyProviderTest.TestMetricHostProvider metricsHostProvider = new JMXPropertyProviderTest.TestMetricHostProvider();
    TestGangliaServiceProvider serviceProvider = new TestGangliaServiceProvider();

    StackDefinedPropertyProvider propertyProvider = new StackDefinedPropertyProvider(
      Resource.Type.HostComponent,
      hostProvider,
      metricsHostProvider,
      serviceProvider,
      streamProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"),
      new EmptyPropertyProvider(),
      new EmptyPropertyProvider());

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

    resource.setProperty("HostRoles/cluster_name", "c2");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "h1");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "RESOURCEMANAGER");
    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/yarn/Queue/root/AvailableMB"), temporalInfoMap);

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

    // see test/resources/resourcemanager_jmx.json for values
    Assert.assertEquals(8192, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableMB")));
    Assert.assertNull(resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableVCores")));
  }

  public void testPopulateResources_HDP2_params_category() throws Exception {
    TestStreamProvider streamProvider = new TestStreamProvider();
    TestJMXProvider hostProvider = new TestJMXProvider(false);
    JMXPropertyProviderTest.TestMetricHostProvider metricsHostProvider = new JMXPropertyProviderTest.TestMetricHostProvider();
    TestGangliaServiceProvider serviceProvider = new TestGangliaServiceProvider();

    StackDefinedPropertyProvider propertyProvider = new StackDefinedPropertyProvider(
      Resource.Type.HostComponent,
      hostProvider,
      metricsHostProvider,
      serviceProvider,
      streamProvider,
      PropertyHelper.getPropertyId("HostRoles", "cluster_name"),
      PropertyHelper.getPropertyId("HostRoles", "host_name"),
      PropertyHelper.getPropertyId("HostRoles", "component_name"),
      PropertyHelper.getPropertyId("HostRoles", "state"),
      new EmptyPropertyProvider(),
      new EmptyPropertyProvider());

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

    resource.setProperty("HostRoles/cluster_name", "c2");
    resource.setProperty(HOST_COMPONENT_HOST_NAME_PROPERTY_ID, "h1");
    resource.setProperty(HOST_COMPONENT_COMPONENT_NAME_PROPERTY_ID, "RESOURCEMANAGER");
    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/yarn/Queue"), temporalInfoMap);

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

    // see test/resources/resourcemanager_jmx.json for values
    Assert.assertEquals(6, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AggregateContainersAllocated")));
    Assert.assertEquals(6, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AggregateContainersReleased")));
    Assert.assertEquals(8192, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableMB")));
    Assert.assertEquals(1, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AvailableVCores")));
    Assert.assertEquals(2, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root", "AppsSubmitted")));

    Assert.assertEquals(15, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/default", "AggregateContainersAllocated")));
    Assert.assertEquals(12, resource.getPropertyValue(PropertyHelper.getPropertyId("metrics/yarn/Queue/root/default", "AggregateContainersReleased")));
