/*
 * 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 java.util.Collections.emptyMap;
import static org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService.GANGLIA;
import static org.apache.ambari.server.controller.metrics.MetricsServiceProvider.MetricsService.TIMELINE_METRICS;

import java.util.Collections;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.ambari.server.controller.jmx.JMXHostProvider;
import org.apache.ambari.server.controller.metrics.MetricHostProvider;
import org.apache.ambari.server.controller.metrics.MetricPropertyProviderFactory;
import org.apache.ambari.server.controller.metrics.MetricsCollectorHAManager;
import org.apache.ambari.server.controller.metrics.MetricsPropertyProvider;
import org.apache.ambari.server.controller.metrics.MetricsReportPropertyProvider;
import org.apache.ambari.server.controller.metrics.MetricsServiceProvider;
import org.apache.ambari.server.controller.metrics.timeline.cache.TimelineMetricCacheProvider;
import org.apache.ambari.server.controller.spi.NoSuchParentResourceException;
import org.apache.ambari.server.controller.spi.NoSuchResourceException;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.PropertyProvider;
import org.apache.ambari.server.controller.spi.ProviderModule;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.controller.spi.ResourceProvider;
import org.apache.ambari.server.controller.spi.SystemException;
import org.apache.ambari.server.controller.spi.UnsupportedPropertyException;
import org.apache.ambari.server.controller.utilities.PredicateBuilder;
import org.apache.ambari.server.controller.utilities.PropertyHelper;
import org.apache.ambari.server.controller.utilities.StreamProvider;
import org.apache.ambari.server.events.ClusterConfigChangedEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.eventbus.Subscribe;
import com.google.inject.Inject;

/**
 * An abstract provider module implementation.
 */
public abstract class AbstractProviderModule implements ProviderModule,
    ResourceProviderObserver, JMXHostProvider, MetricHostProvider,
    MetricsServiceProvider {

  private static final int PROPERTY_REQUEST_CONNECT_TIMEOUT = 5000;
  private static final int PROPERTY_REQUEST_READ_TIMEOUT    = 10000;

  private static final String GANGLIA_SERVER                            = "GANGLIA_SERVER";
  private static final String METRIC_SERVER                             = "METRICS_COLLECTOR";
  private static final String PROPERTIES_CATEGORY = "properties";
  private static final Map<String, String> serviceConfigVersions = new ConcurrentHashMap<>();
  private static final Map<Service.Type, String> serviceConfigTypes = new EnumMap<>(Service.Type.class);
  private static final Map<Service.Type, Map<String, String[]>> serviceDesiredProperties = new EnumMap<>(Service.Type.class);
  private static final Map<String, Service.Type> componentServiceMap = new HashMap<>();

  private static final Map<String, List<HttpPropertyProvider.HttpPropertyRequest>> HTTP_PROPERTY_REQUESTS = new HashMap<>();

  private static final String PROPERTY_HDFS_HTTP_POLICY_VALUE_HTTPS_ONLY = "HTTPS_ONLY";

  private static final String COLLECTOR_DEFAULT_PORT = "6188";

  private static final Map<String, Map<String, String[]>> jmxDesiredProperties = new HashMap<>();
  private static final Map<String, Map<String, String[]>> jmxDesiredRpcSuffixProperties = new ConcurrentHashMap<>();
  private volatile Map<String, Map<String, Map<String, String>>> jmxDesiredRpcSuffixes = new HashMap<>();
  private volatile Map<String, String> clusterHdfsSiteConfigVersionMap = new HashMap<>();
  private volatile Map<String, String> clusterJmxProtocolMap = new ConcurrentHashMap<>();
  private volatile String clusterMetricServerPort = null;
  private volatile String clusterMetricServerVipPort = null;
  private volatile String clusterMetricserverVipHost = null;

  static {
    serviceConfigTypes.put(Service.Type.HDFS, "hdfs-site");
    serviceConfigTypes.put(Service.Type.HBASE, "hbase-site");
    serviceConfigTypes.put(Service.Type.YARN, "yarn-site");
    serviceConfigTypes.put(Service.Type.MAPREDUCE2, "mapred-site");
    serviceConfigTypes.put(Service.Type.AMBARI_METRICS, "ams-site");

    componentServiceMap.put("NAMENODE", Service.Type.HDFS);
    componentServiceMap.put("DATANODE", Service.Type.HDFS);
    componentServiceMap.put("JOURNALNODE", Service.Type.HDFS);
    componentServiceMap.put("HBASE_MASTER", Service.Type.HBASE);
    componentServiceMap.put("HBASE_REGIONSERVER", Service.Type.HBASE);
    componentServiceMap.put("RESOURCEMANAGER", Service.Type.YARN);
    componentServiceMap.put("NODEMANAGER", Service.Type.YARN);
    componentServiceMap.put("HISTORYSERVER", Service.Type.MAPREDUCE2);

    Map<String, String[]> initPropMap = new HashMap<>();
    initPropMap.put("NAMENODE", new String[]{"dfs.http.address", "dfs.namenode.http-address"});
    initPropMap.put("NAMENODE-HTTPS", new String[]{"dfs.namenode.https-address", "dfs.https.port"});
    initPropMap.put("NAMENODE-HA", new String[]{"dfs.namenode.http-address.%s.%s"});
    initPropMap.put("NAMENODE-HTTPS-HA", new String[]{"dfs.namenode.https-address.%s.%s"});
    initPropMap.put("DATANODE", new String[]{"dfs.datanode.http.address"});
    initPropMap.put("DATANODE-HTTPS", new String[]{"dfs.datanode.https.address"});
    initPropMap.put("JOURNALNODE-HTTPS", new String[]{"dfs.journalnode.https-address"});
    initPropMap.put("JOURNALNODE", new String[]{"dfs.journalnode.http-address"});
    serviceDesiredProperties.put(Service.Type.HDFS, initPropMap);


    initPropMap = new HashMap<>();
    initPropMap.put("HBASE_MASTER", new String[]{"hbase.master.info.port"});
    initPropMap.put("HBASE_REGIONSERVER", new String[]{"hbase.regionserver.info.port"});
    initPropMap.put("HBASE_MASTER-HTTPS", new String[]{"hbase.master.info.port"});
    initPropMap.put("HBASE_REGIONSERVER-HTTPS", new String[]{"hbase.regionserver.info.port"});
    serviceDesiredProperties.put(Service.Type.HBASE, initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("RESOURCEMANAGER", new String[]{"yarn.resourcemanager.webapp.address"});
    initPropMap.put("RESOURCEMANAGER-HTTPS", new String[]{"yarn.resourcemanager.webapp.https.address"});
    initPropMap.put("NODEMANAGER", new String[]{"yarn.nodemanager.webapp.address"});
    initPropMap.put("NODEMANAGER-HTTPS", new String[]{"yarn.nodemanager.webapp.https.address"});
    serviceDesiredProperties.put(Service.Type.YARN, initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("HISTORYSERVER", new String[]{"mapreduce.jobhistory.webapp.address"});
    initPropMap.put("HISTORYSERVER-HTTPS", new String[]{"mapreduce.jobhistory.webapp.https.address"});
    serviceDesiredProperties.put(Service.Type.MAPREDUCE2, initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("NAMENODE", new String[]{"dfs.http.policy"});
    jmxDesiredProperties.put("NAMENODE", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("DATANODE", new String[]{"dfs.http.policy"});
    jmxDesiredProperties.put("DATANODE", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("JOURNALNODE", new String[]{"dfs.http.policy"});
    jmxDesiredProperties.put("JOURNALNODE", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("RESOURCEMANAGER", new String[]{"yarn.http.policy"});
    jmxDesiredProperties.put("RESOURCEMANAGER", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("HBASE_MASTER", new String[]{"hbase.ssl.enabled"});
    jmxDesiredProperties.put("HBASE_MASTER", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("HBASE_REGIONSERVER", new String[]{"hbase.ssl.enabled"});
    jmxDesiredProperties.put("HBASE_REGIONSERVER", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("NODEMANAGER", new String[]{"yarn.http.policy"});
    jmxDesiredProperties.put("NODEMANAGER", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("HISTORYSERVER", new String[]{"mapreduce.jobhistory.http.policy"});
    jmxDesiredProperties.put("HISTORYSERVER", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("client", new String[]{"dfs.namenode.rpc-address"});
    initPropMap.put("datanode", new String[]{"dfs.namenode.servicerpc-address"});
    initPropMap.put("healthcheck", new String[]{"dfs.namenode.lifeline.rpc-address"});
    jmxDesiredRpcSuffixProperties.put("NAMENODE", initPropMap);

    initPropMap = new HashMap<>();
    initPropMap.put("client", new String[]{"dfs.namenode.rpc-address.%s.%s"});
    initPropMap.put("datanode", new String[]{"dfs.namenode.servicerpc-address.%s.%s"});
    initPropMap.put("healthcheck", new String[]{"dfs.namenode.lifeline.rpc-address.%s.%s"});
    jmxDesiredRpcSuffixProperties.put("NAMENODE-HA", initPropMap);

    HTTP_PROPERTY_REQUESTS.put("RESOURCEMANAGER",
        Collections.singletonList(new ResourceManagerHttpPropertyRequest()));

    HTTP_PROPERTY_REQUESTS.put("ATLAS_SERVER",
        Collections.singletonList(new AtlasServerHttpPropertyRequest()));
  }

  /**
   * The map of resource providers.
   */
  private final Map<Resource.Type, ResourceProvider> resourceProviders = new HashMap<>();

  /**
   * The map of lists of property providers.
   */
  private final Map<Resource.Type, List<PropertyProvider>> propertyProviders = new HashMap<>();

  /*
   * TODO: Instantiation for the concrete impl of this class is not done through
   * dependency injector (guice) so none of these field initialization
   * are going to work unless refactoring is complete.
   */
  @Inject
  protected AmbariManagementController managementController;

  @Inject
  TimelineMetricCacheProvider metricCacheProvider;

  @Inject
  MetricsCollectorHAManager metricsCollectorHAManager;

  /**
   * A factory used to retrieve Guice-injected instances of a metric
   * {@link PropertyProvider}.
   */
  @Inject
  private MetricPropertyProviderFactory metricPropertyProviderFactory;

  /**
   * Used to respond to the following events:
   * <ul>
   * <li>{@link ClusterConfigChangedEvent}
   */
  @Inject
  protected AmbariEventPublisher eventPublisher;

  @Inject
  private Clusters clusters;

  /**
   * The map of host components.
   */
  private Map<String, Map<String, String>> clusterHostComponentMap;

  /**
   * The host name of the Ganglia collector.
   */
  private Map<String, String> clusterGangliaCollectorMap;

  /**
   * JMX ports read from the configs
   */
  private final Map<String, ConcurrentMap<String, ConcurrentMap<String, String>>> jmxPortMap = new ConcurrentHashMap<>(1);

  private volatile boolean initialized = false;

  private static final Logger LOG =
      LoggerFactory.getLogger(AbstractProviderModule.class);


  // ----- Constructors ------------------------------------------------------

  /**
   * Create a default provider module.
   */
  public AbstractProviderModule() {
    if (managementController == null) {
      managementController = AmbariServer.getController();
    }

    if (metricCacheProvider == null && managementController != null) {
      metricCacheProvider = managementController.getTimelineMetricCacheProvider();
    }

    if (metricPropertyProviderFactory == null && managementController != null) {
      metricPropertyProviderFactory = managementController.getMetricPropertyProviderFactory();
    }

    if (null == eventPublisher && null != managementController) {
      eventPublisher = managementController.getAmbariEventPublisher();
      eventPublisher.register(this);
    }

    if (null == metricsCollectorHAManager && null != managementController) {
      metricsCollectorHAManager = managementController.getMetricsCollectorHAManager();
    }

    if (null == clusters && null != managementController) {
      clusters = managementController.getClusters();
    }
  }


  // ----- ProviderModule ----------------------------------------------------

  @Override
  public ResourceProvider getResourceProvider(Resource.Type type) {
    if (!resourceProviders.containsKey(type)) {
      registerResourceProvider(type);
    }
    return resourceProviders.get(type);
  }

  @Override
  public List<PropertyProvider>  getPropertyProviders(Resource.Type type) {

    if (!propertyProviders.containsKey(type)) {
      createPropertyProviders(type);
    }
    return propertyProviders.get(type);
  }


  // ----- ResourceProviderObserver ------------------------------------------

  @Override
  public void update(ResourceProviderEvent event) {
    Resource.Type type = event.getResourceType();

    if (type == Resource.Type.Cluster ||
        type == Resource.Type.Host ||
        type == Resource.Type.HostComponent) {
      resetInit();
    }
  }

  // ----- MetricsServiceProvider ---------------------------------------------

  /**
   * Get type of Metrics system installed.
   * @return @MetricsService, null if none found.
   */
  @Override
  public MetricsService getMetricsServiceType() {
    try {
      checkInit();
    } catch (SystemException e) {
      LOG.error("Exception during checkInit.", e);
    }

    if (!metricsCollectorHAManager.isEmpty()) {
      return TIMELINE_METRICS;
    } else if (!clusterGangliaCollectorMap.isEmpty()) {
      return GANGLIA;
    }
    return null;
  }

  // ----- MetricsHostProvider ------------------------------------------------

  @Override
  public String getCollectorHostName(String clusterName, MetricsService service)
    throws SystemException {

    checkInit();
    if (service.equals(GANGLIA)) {
      return clusterGangliaCollectorMap.get(clusterName);
    } else if (service.equals(TIMELINE_METRICS)) {
      return getMetricsCollectorHostName(clusterName);
    }
    return null;
  }


  private void checkAndAddExternalCollectorHosts(String clusterName) throws SystemException {
    try {
      // try to get vip properties from cluster-env
      String configType = "cluster-env";
      String currentConfigVersion = getDesiredConfigVersion(clusterName, configType);
      String oldConfigVersion = serviceConfigVersions.get(configType);
      if (!currentConfigVersion.equals(oldConfigVersion)) {
        serviceConfigVersions.put(configType, currentConfigVersion);
        Map<String, String> configProperties = getDesiredConfigMap
          (clusterName, currentConfigVersion, configType,
            Collections.singletonMap("METRICS_COLLECTOR",
              new String[]{"metrics_collector_external_hosts"}));

        if (!configProperties.isEmpty()) {
          clusterMetricserverVipHost = configProperties.get("METRICS_COLLECTOR");
          if (StringUtils.isNotEmpty(clusterMetricserverVipHost)) {
            for (String collectorHost : StringUtils.split(clusterMetricserverVipHost, ",")) {
              metricsCollectorHAManager.addExternalMetricsCollectorHost(clusterName, collectorHost);
            }
            LOG.info("Setting Metrics Collector External Host : " + clusterMetricserverVipHost);
          }
        }
        // updating the port value, because both vip properties are stored in
        // cluster-env
        configProperties = getDesiredConfigMap
          (clusterName, currentConfigVersion, configType,
            Collections.singletonMap("METRICS_COLLECTOR",
              new String[]{"metrics_collector_external_port"}));

        if (!configProperties.isEmpty()) {
          clusterMetricServerVipPort = configProperties.getOrDefault("METRICS_COLLECTOR", "6188");
        }
      }
    } catch (NoSuchParentResourceException | UnsupportedPropertyException e) {
      LOG.warn("Failed to retrieve collector hostname.", e);
    }
  }

  private String getMetricsCollectorHostName(String clusterName)
    throws SystemException {

    checkAndAddExternalCollectorHosts(clusterName);
    String currentCollectorHost = metricsCollectorHAManager.getCollectorHost(clusterName);
    LOG.debug("Cluster Metrics Vip Host : " + clusterMetricserverVipHost);

    return currentCollectorHost;
  }

  @Override
  public boolean isCollectorHostExternal(String clusterName) {
    return metricsCollectorHAManager.isExternalCollector();
  }

  @Override
  public String getCollectorPort(String clusterName, MetricsService service) throws SystemException {
    checkInit();
    if (service.equals(GANGLIA)) {
      return "80"; // Not called by the provider
    } else if (service.equals(TIMELINE_METRICS)) {
      try {
        if (clusterMetricServerVipPort == null) {
          String configType = serviceConfigTypes.get(Service.Type.AMBARI_METRICS);
          String currentConfigVersion = getDesiredConfigVersion(clusterName, configType);
          String oldConfigVersion = serviceConfigVersions.get(configType);
          if (!currentConfigVersion.equals(oldConfigVersion)) {
            serviceConfigVersions.put(configType, currentConfigVersion);

            Map<String, String> configProperties = getDesiredConfigMap(clusterName,
              currentConfigVersion, configType,
              Collections.singletonMap("METRICS_COLLECTOR",
                new String[]{"timeline.metrics.service.webapp.address"}));

            if (!configProperties.isEmpty()) {
              clusterMetricServerPort = getPortString(configProperties.get("METRICS_COLLECTOR"));
            } else {
              clusterMetricServerPort = COLLECTOR_DEFAULT_PORT;
            }
          }
        }
      } catch (NoSuchParentResourceException | UnsupportedPropertyException e) {
        LOG.warn("Failed to retrieve collector port.", e);
      }
    }
    return clusterMetricServerVipPort != null ? clusterMetricServerVipPort : clusterMetricServerPort;
  }

  @Override
  public boolean isCollectorHostLive(String clusterName, MetricsService service) throws SystemException {
    return metricsCollectorHAManager.isCollectorHostLive(clusterName);
  }

  @Override
  public String getHostName(String clusterName, String componentName) throws SystemException {
    checkInit();
    return clusterHostComponentMap.get(clusterName).get(componentName);
  }

  @Override
  public String getPublicHostName(String clusterName, String hostName) {
    Host host = getHost(clusterName, hostName);
    return host == null? hostName : host.getPublicHostName();
  }

  @Override
  public Set<String> getHostNames(String clusterName, String componentName) {
    Set<String> hosts = null;
    try {
      Cluster cluster = managementController.getClusters().getCluster(clusterName);
      String serviceName = managementController.findServiceName(cluster, componentName);
      hosts = cluster.getService(serviceName).getServiceComponent(componentName).getServiceComponentHosts().keySet();
    } catch (Exception e) {
      LOG.warn("Exception in getting host names for jmx metrics: ", e);
    }
    return hosts;
  }

  private Host getHost(String clusterName, String hostName) {
    Host host = null;
    try {
      Cluster cluster = managementController.getClusters().getCluster(clusterName);
      if(cluster != null) {
        host = cluster.getHost(hostName);
      }
    } catch (Exception e) {
      LOG.warn("Exception in getting host info for jmx metrics: ", e);
    }
    return host;
  }


  @Override
  public boolean isCollectorComponentLive(String clusterName, MetricsService service) throws SystemException {

    final String collectorHostName = getCollectorHostName(clusterName, service);

    if (service.equals(GANGLIA)) {
      return HostStatusHelper.isHostComponentLive(managementController, clusterName, collectorHostName, "GANGLIA",
        Role.GANGLIA_SERVER.name());
    } else if (service.equals(TIMELINE_METRICS)) {
      return metricsCollectorHAManager.isCollectorComponentLive(clusterName);
    }
    return false;
  }

  // ----- JMXHostProvider ---------------------------------------------------

  @Override
  public String getPort(String clusterName, String componentName, String hostName, boolean httpsEnabled) {
    ConcurrentMap<String, ConcurrentMap<String, String>> clusterJmxPorts;
    // Still need double check to ensure single init
    if (!jmxPortMap.containsKey(clusterName)) {
      synchronized (jmxPortMap) {
        if (!jmxPortMap.containsKey(clusterName)) {
          clusterJmxPorts = new ConcurrentHashMap<>();
          jmxPortMap.put(clusterName, clusterJmxPorts);
        }
      }
    }
    clusterJmxPorts = jmxPortMap.get(clusterName);
    Service.Type service = componentServiceMap.get(componentName);

    if (service != null) {
      try {
        String configType = serviceConfigTypes.get(service);
        String currVersion = getDesiredConfigVersion(clusterName, configType);
        String oldVersion = serviceConfigVersions.get(configType);

        // We only update port map when a config version updates,
        // Since concurrent thread access is expected we err on the side of
        // performance with a ConcurrentHashMap and maybe get default/existing
        // ports for a few calls.
        if (!StringUtils.equals(currVersion, oldVersion) ||
            !(clusterJmxPorts.containsKey(hostName) && clusterJmxPorts.get(hostName).containsKey(componentName))) {

          serviceConfigVersions.put(configType, currVersion);

          Map<String, Object> configProperties = getConfigProperties(
              clusterName,
              currVersion,
              serviceConfigTypes.get(service)
          );

          String publicHostName = getPublicHostName(clusterName, hostName);
          Map<String, String[]> componentPortsProperties = new HashMap<>();
          componentPortsProperties.put(
              componentName,
              getPortProperties(service,
                  componentName,
                  hostName,
                  publicHostName,
                  configProperties,
                  httpsEnabled
              )
          );

          Map<String, String> portMap = getDesiredConfigMap(clusterName,
              currVersion, serviceConfigTypes.get(service),
              componentPortsProperties, configProperties);

          for (Entry<String, String> entry : portMap.entrySet()) {
            // portString will be null if the property defined for the component doesn't exist
            // this will trigger using the default port for the component
            String portString = getPortString(entry.getValue());
            if (null != portString) {
              clusterJmxPorts.putIfAbsent(hostName, new ConcurrentHashMap<>());
              clusterJmxPorts.get(hostName).put(entry.getKey(), portString);
            }
          }

          initRpcSuffixes(clusterName, componentName, configType, currVersion, hostName, publicHostName);
        }
      } catch (Exception e) {
        LOG.error("Exception initializing jmx port maps. ", e);
      }
    }

    LOG.debug("jmxPortMap -> {}", jmxPortMap);

    ConcurrentMap<String, String> hostJmxPorts = clusterJmxPorts.get(hostName);
    if (hostJmxPorts == null) {
      LOG.debug("Jmx ports not loaded from properties: clusterName={}, componentName={}, hostName={}, " +
          "clusterJmxPorts={}, jmxPortMap.get(clusterName)={}",
          clusterName, componentName, hostName, clusterJmxPorts, jmxPortMap.get(clusterName));
      //returning null is acceptable in cases when property with port not present
      //or loading from property is not supported for specific component
      return null;
    }
    return hostJmxPorts.get(componentName);
  }

  /**
   * Computes properties that contains proper port for {@code componentName} on {@code hostName}.
   * Must contain custom logic for different configurations(like NAMENODE HA).
   * @param service service type
   * @param componentName component name
   * @param hostName host which contains requested component
   * @param properties properties which used for special cases(like NAMENODE HA)
   * @param httpsEnabled indicates if https enabled for component
   * @return property name that contain port for {@code componentName} on {@code hostName}
   */
  String[] getPortProperties(Service.Type service, String componentName,
    String hostName, String publicHostName, Map<String, Object> properties, boolean httpsEnabled) {
    componentName = httpsEnabled ? componentName + "-HTTPS" : componentName;
    if(componentName.startsWith("NAMENODE") && properties.containsKey("dfs.internal.nameservices")) {
      componentName += "-HA";
      return getNamenodeHaProperty(
        properties, serviceDesiredProperties.get(service).get(componentName), hostName, publicHostName);
    }
    return serviceDesiredProperties.get(service).get(componentName);
  }

  private String[] getNamenodeHaProperty(Map<String, Object> properties, String pattern[],
    String hostName, String publicHostName) {

    // iterate over nameservices and namenodes, to find out namenode http(s) property for concrete host
    for(String nameserviceId : ((String)properties.get("dfs.internal.nameservices")).split(",")) {
      if(properties.containsKey("dfs.ha.namenodes."+nameserviceId)) {
        for (String namenodeId : ((String)properties.get("dfs.ha.namenodes." + nameserviceId)).split(",")) {
          String propertyName = String.format(
            pattern[0],
            nameserviceId,
            namenodeId
          );
          if (properties.containsKey(propertyName)) {
            String propertyValue = (String)properties.get(propertyName);
            String propHostName = propertyValue.split(":")[0];
            if (propHostName.equals(hostName) || propHostName.equals(publicHostName)) {
              return new String[] {propertyName};
            }
          }
        }
      }
    }
    return pattern;
  }

  /**
   * Post process property value. If value has one ore some substrings
   * started with "${" and ended with "}" these substrings will replace
   * with properties from current propertiesMap. It is doing recursively.
   *
   * @param key        - properties name
   * @param value      - properties value
   * @param properties - map with properties
   */
  private String postProcessPropertyValue(String key, String value, Map<String, String> properties, Set<String> prevProps) {
    if (value != null && key != null && value.contains("${")) {
      if (prevProps == null) {
        prevProps = new HashSet<>();
      }
      if (prevProps.contains(key)) {
        return value;
      }
      prevProps.add(key);
      String refValueString = value;
      Map<String, String> refMap = new HashMap<>();
      while (refValueString.contains("${")) {
        int startValueRef = refValueString.indexOf("${") + 2;
        int endValueRef = refValueString.indexOf('}');
        String valueRef = refValueString.substring(startValueRef, endValueRef);
        refValueString = refValueString.substring(endValueRef + 1);
        String trueValue = postProcessPropertyValue(valueRef, properties.get(valueRef), properties, prevProps);
        if (trueValue != null) {
          refMap.put("${" + valueRef + '}', trueValue);
        }
      }
      for (Entry<String, String> entry : refMap.entrySet()) {
        refValueString = entry.getValue();
        value = value.replace(entry.getKey(), refValueString);
      }
      properties.put(key, value);
    }
    return value;
  }

  // ----- utility methods ---------------------------------------------------

  protected abstract ResourceProvider createResourceProvider(Resource.Type type);

  protected void registerResourceProvider(Resource.Type type) {
    ResourceProvider resourceProvider = createResourceProvider(type);

    if (resourceProvider instanceof ObservableResourceProvider) {
      ((ObservableResourceProvider) resourceProvider).addObserver(this);
    }

    putResourceProvider(type, resourceProvider);
  }

  protected void putResourceProvider(Resource.Type type, ResourceProvider resourceProvider) {
    resourceProviders.put(type, resourceProvider);
  }

  protected void putPropertyProviders(Resource.Type type, List<PropertyProvider> providers) {
    propertyProviders.put(type, providers);
  }

  protected void createPropertyProviders(Resource.Type type) {

    List<PropertyProvider> providers = new LinkedList<>();

    ComponentSSLConfiguration configuration = ComponentSSLConfiguration.instance();
    URLStreamProvider streamProvider = new URLStreamProvider(
        PROPERTY_REQUEST_CONNECT_TIMEOUT, PROPERTY_REQUEST_READ_TIMEOUT,
        configuration);

    if (type.isInternalType()) {
      switch (type.getInternalType()) {
        case Cluster:
          providers.add(createMetricsReportPropertyProvider(
            type,
            streamProvider,
            ComponentSSLConfiguration.instance(),
            this,
            this,
            PropertyHelper.getPropertyId("Clusters", "cluster_name")));
          providers.add(new AlertSummaryPropertyProvider(type,
            "Clusters/cluster_name", null));
          break;
        case Service:
          providers.add(new AlertSummaryPropertyProvider(type,
              "ServiceInfo/cluster_name", "ServiceInfo/service_name"));
          break;
        case Host:
          providers.add(createMetricsHostPropertyProvider(
            type,
            streamProvider,
            ComponentSSLConfiguration.instance(),
            this,
            this,
            PropertyHelper.getPropertyId("Hosts", "cluster_name"),
            PropertyHelper.getPropertyId("Hosts", "host_name")
          ));
          providers.add(new AlertSummaryPropertyProvider(type,
              "Hosts/cluster_name", "Hosts/host_name"));
          break;
        case Component: {
          // TODO as we fill out stack metric definitions, these can be phased out
          PropertyProvider jpp = createJMXPropertyProvider(
              type,
              streamProvider,
              this,
              this,
              PropertyHelper.getPropertyId("ServiceComponentInfo", "cluster_name"),
              null,
              PropertyHelper.getPropertyId("ServiceComponentInfo", "component_name"),
              PropertyHelper.getPropertyId("ServiceComponentInfo", "state"));
          PropertyProvider gpp = null;
          gpp = createMetricsComponentPropertyProvider(
              type,
              streamProvider,
              ComponentSSLConfiguration.instance(),
              this,
              this,
              PropertyHelper.getPropertyId("ServiceComponentInfo", "cluster_name"),
              PropertyHelper.getPropertyId("ServiceComponentInfo", "component_name"));
          providers.add(new StackDefinedPropertyProvider(
              type,
              this,
              this,
              this,
              streamProvider,
              PropertyHelper.getPropertyId("ServiceComponentInfo", "cluster_name"),
