AMBARI-21244 Add https support to local metrics aggregator application (dsen)

ambari-logsearch/ambari-logsearch-logfeeder/src/main/java/org/apache/ambari/logfeeder/metrics/LogFeederAMSClient.java

@@ -99,6 +99,11 @@ public class LogFeederAMSClient extends AbstractTimelineMetricsSink {
     return 0;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return "http";
+  }
+
   @Override
   protected boolean emitMetrics(TimelineMetrics metrics) {
     return super.emitMetrics(metrics);

ambari-metrics/ambari-metrics-common/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractTimelineMetricsSink.java

@@ -81,6 +81,7 @@ public abstract class AbstractTimelineMetricsSink {
   public static final String SSL_KEYSTORE_PASSWORD_PROPERTY = "truststore.password";
   public static final String HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY = "host_in_memory_aggregation";
   public static final String HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY = "host_in_memory_aggregation_port";
+  public static final String HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY = "host_in_memory_aggregation_protocol";
   public static final String COLLECTOR_LIVE_NODES_PATH = "/ws/v1/timeline/metrics/livenodes";
   public static final String INSTANCE_ID_PROPERTY = "instanceId";
   public static final String SET_INSTANCE_ID_PROPERTY = "set.instanceId";
@@ -293,7 +294,11 @@ public abstract class AbstractTimelineMetricsSink {
     boolean validCollectorHost = true;
 
     if (isHostInMemoryAggregationEnabled()) {
-      connectUrl = constructTimelineMetricUri("http", "localhost", String.valueOf(getHostInMemoryAggregationPort()));
+      String hostname = "localhost";
+      if (getHostInMemoryAggregationProtocol().equalsIgnoreCase("https")) {
+        hostname = getHostname();
+      }
+      connectUrl = constructTimelineMetricUri(getHostInMemoryAggregationProtocol(), hostname, String.valueOf(getHostInMemoryAggregationPort()));
     } else {
       String collectorHost  = getCurrentCollectorHost();
       if (collectorHost == null) {
@@ -647,4 +652,10 @@ public abstract class AbstractTimelineMetricsSink {
    * @return
    */
   abstract protected int getHostInMemoryAggregationPort();
+
+  /**
+   * In memory aggregation protocol
+   * @return
+   */
+  abstract protected String getHostInMemoryAggregationProtocol();
 }

ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/availability/AbstractTimelineMetricSinkTest.java

@@ -99,6 +99,11 @@ public class AbstractTimelineMetricSinkTest {
       return 61888;
     }
 
+    @Override
+    protected String getHostInMemoryAggregationProtocol() {
+      return "http";
+    }
+
     @Override
     public boolean emitMetrics(TimelineMetrics metrics) {
       super.init();

ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/availability/MetricCollectorHATest.java

@@ -202,5 +202,10 @@ public class MetricCollectorHATest {
     protected int getHostInMemoryAggregationPort() {
       return 61888;
     }
+
+    @Override
+    protected String getHostInMemoryAggregationProtocol() {
+      return "http";
+    }
   }
 }

ambari-metrics/ambari-metrics-common/src/test/java/org/apache/hadoop/metrics2/sink/timeline/cache/HandleConnectExceptionTest.java

@@ -145,6 +145,11 @@ public class HandleConnectExceptionTest {
       return 61888;
     }
 
+    @Override
+    protected String getHostInMemoryAggregationProtocol() {
+      return "http";
+    }
+
     @Override
     public boolean emitMetrics(TimelineMetrics metrics) {
       super.init();

ambari-metrics/ambari-metrics-flume-sink/src/main/conf/flume-metrics2.properties.j2

@@ -19,7 +19,13 @@
 collector=http://localhost:6188
 collectionFrequency=60000
 maxRowCacheSize=10000
-sendInterval=59000
+sendInterval={{metrics_report_interval}}000
+clusterReporterAppId=nimbus
+host_in_memory_aggregation = {{host_in_memory_aggregation}}
+host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # Metric names having type COUNTER
 counters=EventTakeSuccessCount,EventPutSuccessCount,EventTakeAttemptCount,EventPutAttemptCount

ambari-metrics/ambari-metrics-flume-sink/src/main/java/org/apache/hadoop/metrics2/sink/flume/FlumeTimelineMetricsSink.java

@@ -65,6 +65,7 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
   private String instanceId;
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
+  private String hostInMemoryAggregationProtocol;
 
 
   @Override
@@ -114,12 +115,13 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
     setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY, "false"));
     instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY, "");
 
-    hostInMemoryAggregationEnabled = Boolean.getBoolean(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY));
-    hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY));
+    hostInMemoryAggregationEnabled = Boolean.getBoolean(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, "false"));
+    hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, "61888"));
+    hostInMemoryAggregationProtocol = configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
     // Initialize the collector write strategy
     super.init();
 
-    if (protocol.contains("https")) {
+    if (protocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
       String trustStorePath = configuration.getProperty(SSL_KEYSTORE_PATH_PROPERTY).trim();
       String trustStoreType = configuration.getProperty(SSL_KEYSTORE_TYPE_PROPERTY).trim();
       String trustStorePwd = configuration.getProperty(SSL_KEYSTORE_PASSWORD_PROPERTY).trim();
@@ -178,6 +180,11 @@ public class FlumeTimelineMetricsSink extends AbstractTimelineMetricsSink implem
     return hostInMemoryAggregationPort;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return hostInMemoryAggregationProtocol;
+  }
+
   public void setPollFrequency(long pollFrequency) {
     this.pollFrequency = pollFrequency;
   }

ambari-metrics/ambari-metrics-hadoop-sink/src/main/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSink.java

@@ -77,6 +77,7 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
   });
   private int hostInMemoryAggregationPort;
   private boolean hostInMemoryAggregationEnabled;
+  private String hostInMemoryAggregationProtocol;
 
   @Override
   public void init(SubsetConfiguration conf) {
@@ -109,12 +110,13 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
     protocol = conf.getString(COLLECTOR_PROTOCOL, "http");
     collectorHosts = parseHostsStringArrayIntoCollection(conf.getStringArray(COLLECTOR_HOSTS_PROPERTY));
     port = conf.getString(COLLECTOR_PORT, "6188");
-    hostInMemoryAggregationEnabled = conf.getBoolean(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY);
-    hostInMemoryAggregationPort = conf.getInt(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY);
+    hostInMemoryAggregationEnabled = conf.getBoolean(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, false);
+    hostInMemoryAggregationPort = conf.getInt(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, 61888);
+    hostInMemoryAggregationProtocol = conf.getString(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
     if (collectorHosts.isEmpty()) {
       LOG.error("No Metric collector configured.");
     } else {
-      if (protocol.contains("https")) {
+      if (protocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
         String trustStorePath = conf.getString(SSL_KEYSTORE_PATH_PROPERTY).trim();
         String trustStoreType = conf.getString(SSL_KEYSTORE_TYPE_PROPERTY).trim();
         String trustStorePwd = conf.getString(SSL_KEYSTORE_PASSWORD_PROPERTY).trim();
@@ -261,6 +263,11 @@ public class HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink imple
     return hostInMemoryAggregationPort;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return hostInMemoryAggregationProtocol;
+  }
+
   @Override
   public void putMetrics(MetricsRecord record) {
     try {

ambari-metrics/ambari-metrics-hadoop-sink/src/test/java/org/apache/hadoop/metrics2/sink/timeline/HadoopTimelineMetricsSinkTest.java

@@ -60,6 +60,7 @@ import static org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSi
 import static org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink.MAX_METRIC_ROW_CACHE_SIZE;
 import static org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink.METRICS_SEND_INTERVAL;
 import static org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink.SET_INSTANCE_ID_PROPERTY;
+import static org.apache.hadoop.metrics2.sink.timeline.AbstractTimelineMetricsSink.HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY;
 import static org.easymock.EasyMock.anyInt;
 import static org.easymock.EasyMock.anyObject;
 import static org.easymock.EasyMock.anyString;
@@ -116,6 +117,7 @@ public class HadoopTimelineMetricsSinkTest {
     expect(conf.getInt(eq(METRICS_SEND_INTERVAL), anyInt())).andReturn(1000).anyTimes();
     expect(conf.getBoolean(eq(SET_INSTANCE_ID_PROPERTY), eq(false))).andReturn(true).anyTimes();
     expect(conf.getString(eq(INSTANCE_ID_PROPERTY), anyString())).andReturn("instanceId").anyTimes();
+    expect(conf.getString(eq(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY), anyString())).andReturn("http").anyTimes();
 
     conf.setListDelimiterHandler(new DefaultListDelimiterHandler(eq(',')));
     expectLastCall().anyTimes();
@@ -188,6 +190,7 @@ public class HadoopTimelineMetricsSinkTest {
     expect(conf.getString(eq("serviceName-prefix"), eq(""))).andReturn("").anyTimes();
     expect(conf.getString(eq(COLLECTOR_PROTOCOL), eq("http"))).andReturn("http").anyTimes();
     expect(conf.getString(eq(COLLECTOR_PORT), eq("6188"))).andReturn("6188").anyTimes();
+    expect(conf.getString(eq(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY), anyString())).andReturn("http").anyTimes();
 
     expect(conf.getInt(eq(MAX_METRIC_ROW_CACHE_SIZE), anyInt())).andReturn(10).anyTimes();
     // Return eviction time smaller than time diff for first 3 entries
@@ -326,6 +329,7 @@ public class HadoopTimelineMetricsSinkTest {
 
     expect(conf.getInt(eq(MAX_METRIC_ROW_CACHE_SIZE), anyInt())).andReturn(10).anyTimes();
     expect(conf.getInt(eq(METRICS_SEND_INTERVAL), anyInt())).andReturn(10).anyTimes();
+    expect(conf.getString(eq(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY), anyString())).andReturn("http").anyTimes();
 
     conf.setListDelimiterHandler(new DefaultListDelimiterHandler(eq(',')));
     expectLastCall().anyTimes();

ambari-metrics/ambari-metrics-host-aggregator/pom.xml

@@ -101,6 +101,16 @@
             <version>4.2</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-server</artifactId>
+            <version>9.2.11.v20150529</version>
+        </dependency>
+        <dependency>
+            <groupId>org.eclipse.jetty</groupId>
+            <artifactId>jetty-webapp</artifactId>
+            <version>9.2.11.v20150529</version>
+        </dependency>
     </dependencies>
 
     <build>

ambari-metrics/ambari-metrics-host-aggregator/src/main/java/org/apache/hadoop/metrics2/host/aggregator/AggregatorApplication.java

@@ -22,20 +22,23 @@ import com.sun.jersey.api.core.PackagesResourceConfig;
 import com.sun.jersey.api.core.ResourceConfig;
 import com.sun.net.httpserver.HttpServer;
 
+import javax.net.ssl.SSLContext;
 import javax.ws.rs.core.UriBuilder;
-import java.io.IOException;
 import java.net.InetAddress;
 import java.net.URI;
 import java.net.URL;
 import java.net.UnknownHostException;
 import java.util.HashMap;
 
+import com.sun.net.httpserver.HttpsConfigurator;
+import com.sun.net.httpserver.HttpsServer;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.metrics2.sink.timeline.AbstractMetricPublisher;
 import org.apache.hadoop.metrics2.sink.timeline.AggregatedMetricsPublisher;
 import org.apache.hadoop.metrics2.sink.timeline.RawMetricsPublisher;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
 
 /**
  * WEB application with 2 publisher threads that processes received metrics and submits results to the collector
@@ -45,10 +48,12 @@ public class AggregatorApplication
     private static final int STOP_SECONDS_DELAY = 0;
     private static final int JOIN_SECONDS_TIMEOUT = 5;
     private static final String METRICS_SITE_CONFIGURATION_FILE = "ams-site.xml";
+    private static final String METRICS_SSL_SERVER_CONFIGURATION_FILE = "ssl-server.xml";
     private Log LOG;
     private final int webApplicationPort;
     private final int rawPublishingInterval;
     private final int aggregationInterval;
+    private final String webServerProtocol;
     private Configuration configuration;
     private Thread aggregatePublisherThread;
     private Thread rawPublisherThread;
@@ -65,10 +70,11 @@ public class AggregatorApplication
         this.aggregationInterval = configuration.getInt("timeline.metrics.host.aggregator.minute.interval", 300);
         this.rawPublishingInterval = configuration.getInt("timeline.metrics.sink.report.interval", 60);
         this.webApplicationPort = configuration.getInt("timeline.metrics.host.inmemory.aggregation.port", 61888);
+        this.webServerProtocol = configuration.get("timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY").equalsIgnoreCase("HTTP_ONLY") ? "http" : "https";
         this.timelineMetricsHolder = TimelineMetricsHolder.getInstance(rawPublishingInterval, aggregationInterval);
         try {
             this.httpServer = createHttpServer();
-        } catch (IOException e) {
+        } catch (Exception e) {
             LOG.error("Exception while starting HTTP server. Exiting", e);
             System.exit(1);
         }
@@ -88,13 +94,20 @@ public class AggregatorApplication
 
         URL amsResUrl = classLoader.getResource(METRICS_SITE_CONFIGURATION_FILE);
         LOG.info("Found metric service configuration: " + amsResUrl);
+        URL sslConfUrl = classLoader.getResource(METRICS_SSL_SERVER_CONFIGURATION_FILE);
+        LOG.info("Found metric service configuration: " + sslConfUrl);
         if (amsResUrl == null) {
-            throw new IllegalStateException("Unable to initialize the metrics " +
-                    "subsystem. No ams-site present in the classpath.");
+            throw new IllegalStateException(String.format("Unable to initialize the metrics " +
+                    "subsystem. No %s present in the classpath.", METRICS_SITE_CONFIGURATION_FILE));
+        }
+        if (sslConfUrl == null) {
+            throw new IllegalStateException(String.format("Unable to initialize the metrics " +
+                    "subsystem. No %s present in the classpath.", METRICS_SSL_SERVER_CONFIGURATION_FILE));
         }
 
         try {
             configuration.addResource(amsResUrl.toURI().toURL());
+            configuration.addResource(sslConfUrl.toURI().toURL());
         } catch (Exception e) {
             LOG.error("Couldn't init configuration. ", e);
             System.exit(1);
@@ -112,17 +125,41 @@ public class AggregatorApplication
     }
 
     protected URI getURI() {
-        URI uri = UriBuilder.fromUri("http://" + getHostName() + "/").port(this.webApplicationPort).build();
+        URI uri = UriBuilder.fromUri("/").scheme(this.webServerProtocol).host(getHostName()).port(this.webApplicationPort).build();
         LOG.info(String.format("Web server at %s", uri));
         return uri;
     }
 
-    protected HttpServer createHttpServer() throws IOException {
+    protected HttpServer createHttpServer() throws Exception {
         ResourceConfig resourceConfig = new PackagesResourceConfig("org.apache.hadoop.metrics2.host.aggregator");
         HashMap<String, Object> params = new HashMap();
         params.put("com.sun.jersey.api.json.POJOMappingFeature", "true");
         resourceConfig.setPropertiesAndFeatures(params);
-        return HttpServerFactory.create(getURI(), resourceConfig);
+        HttpServer server = HttpServerFactory.create(getURI(), resourceConfig);
+
+        if (webServerProtocol.equalsIgnoreCase("https")) {
+            HttpsServer httpsServer = (HttpsServer) server;
+            SslContextFactory sslContextFactory = new SslContextFactory();
+            String keyStorePath = configuration.get("ssl.server.keystore.location");
+            String keyStorePassword = configuration.get("ssl.server.keystore.password");
+            String keyManagerPassword = configuration.get("ssl.server.keystore.keypassword");
+            String trustStorePath = configuration.get("ssl.server.truststore.location");
+            String trustStorePassword = configuration.get("ssl.server.truststore.password");
+
+            sslContextFactory.setKeyStorePath(keyStorePath);
+            sslContextFactory.setKeyStorePassword(keyStorePassword);
+            sslContextFactory.setKeyManagerPassword(keyManagerPassword);
+            sslContextFactory.setTrustStorePath(trustStorePath);
+            sslContextFactory.setTrustStorePassword(trustStorePassword);
+
+            sslContextFactory.start();
+            SSLContext sslContext = sslContextFactory.getSslContext();
+            sslContextFactory.stop();
+            HttpsConfigurator httpsConfigurator = new HttpsConfigurator(sslContext);
+            httpsServer.setHttpsConfigurator(httpsConfigurator);
+            server = httpsServer;
+        }
+        return server;
     }
 
     private void startWebServer() {

ambari-metrics/ambari-metrics-host-aggregator/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AbstractMetricPublisher.java

@@ -30,9 +30,9 @@ import java.util.Map;
  */
 public abstract class AbstractMetricPublisher extends AbstractTimelineMetricsSink implements Runnable {
 
-    private static final String AMS_SITE_SSL_KEYSTORE_PATH_PROPERTY = "ssl.server.truststore.location";
-    private static final String AMS_SITE_SSL_KEYSTORE_TYPE_PROPERTY = "ssl.server.truststore.password";
-    private static final String AMS_SITE_SSL_KEYSTORE_PASSWORD_PROPERTY = "ssl.server.truststore.type";
+    private static final String AMS_SITE_SSL_TRUSTSTORE_PATH_PROPERTY = "ssl.server.truststore.location";
+    private static final String AMS_SITE_SSL_TRUSTSTORE_TYPE_PROPERTY = "ssl.server.truststore.type";
+    private static final String AMS_SITE_SSL_TRUSTSTORE_PASSWORD_PROPERTY = "ssl.server.truststore.password";
     private static final String AMS_SITE_HTTP_POLICY_PROPERTY = "timeline.metrics.service.http.policy";
     private static final String AMS_SITE_COLLECTOR_WEBAPP_ADDRESS_PROPERTY = "timeline.metrics.service.webapp.address";
     private static final String PUBLISHER_COLLECTOR_HOSTS_PROPERTY = "timeline.metrics.collector.hosts";
@@ -68,9 +68,9 @@ public abstract class AbstractMetricPublisher extends AbstractTimelineMetricsSin
             LOG.error("No Metric collector configured.");
         } else {
             if (collectorProtocol.contains("https")) {
-                String trustStorePath = configuration.get(AMS_SITE_SSL_KEYSTORE_PATH_PROPERTY).trim();
-                String trustStoreType = configuration.get(AMS_SITE_SSL_KEYSTORE_TYPE_PROPERTY).trim();
-                String trustStorePwd = configuration.get(AMS_SITE_SSL_KEYSTORE_PASSWORD_PROPERTY).trim();
+                String trustStorePath = configuration.get(AMS_SITE_SSL_TRUSTSTORE_PATH_PROPERTY).trim();
+                String trustStoreType = configuration.get(AMS_SITE_SSL_TRUSTSTORE_TYPE_PROPERTY).trim();
+                String trustStorePwd = configuration.get(AMS_SITE_SSL_TRUSTSTORE_PASSWORD_PROPERTY).trim();
                 loadTruststore(trustStorePath, trustStoreType, trustStorePwd);
             }
         }

ambari-metrics/ambari-metrics-host-aggregator/src/main/java/org/apache/hadoop/metrics2/sink/timeline/AggregatedMetricsPublisher.java

@@ -100,4 +100,9 @@ public class AggregatedMetricsPublisher extends AbstractMetricPublisher {
     protected String getPostUrl() {
         return BASE_POST_URL + AGGREGATED_POST_PREFIX;
     }
+
+    @Override
+    protected String getHostInMemoryAggregationProtocol() {
+        return "http";
+    }
 }

ambari-metrics/ambari-metrics-host-aggregator/src/main/java/org/apache/hadoop/metrics2/sink/timeline/RawMetricsPublisher.java

@@ -62,4 +62,9 @@ public class RawMetricsPublisher extends AbstractMetricPublisher {
     protected String getPostUrl() {
         return BASE_POST_URL;
     }
+
+    @Override
+    protected String getHostInMemoryAggregationProtocol() {
+        return "http";
+    }
 }

ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/aggregator.py

@@ -59,7 +59,7 @@ class Aggregator(threading.Thread):
   def stop(self):
     self.stopped = True
     if self._aggregator_process :
-      logger.info('Stopping Aggregator thread.')
+      logger.info('Shutting down Aggregator thread.')
       self._aggregator_process.terminate()
       self._aggregator_process = None
 
@@ -71,7 +71,7 @@ class AggregatorWatchdog(threading.Thread):
     threading.Thread.__init__(self)
     self._config = config
     self._stop_handler = stop_handler
-    self.URL = 'http://localhost:' + self._config.get_inmemory_aggregation_port() + self.AMS_AGGREGATOR_METRICS_CHECK_URL
+    self.URL = self._config.get_inmemory_aggregation_protocol() + '://localhost:' + self._config.get_inmemory_aggregation_port() + self.AMS_AGGREGATOR_METRICS_CHECK_URL
     self._is_ok = threading.Event()
     self.set_is_ok(True)
     self.stopped = False
@@ -106,7 +106,7 @@ class AggregatorWatchdog(threading.Thread):
 
 
   def stop(self):
-    logger.info('Stopping watcher thread.')
+    logger.info('Shutting down watcher thread.')
     self.stopped = True
 
 

ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/config_reader.py

@@ -258,17 +258,20 @@ class Configuration:
   def get_max_queue_size(self):
     return int(self.get("collector", "max_queue_size", 5000))
 
-  def is_server_https_enabled(self):
+  def is_collector_https_enabled(self):
     return "true" == str(self.get("collector", "https_enabled")).lower()
 
   def get_java_home(self):
     return self.get("aggregation", "java_home")
 
   def is_inmemory_aggregation_enabled(self):
-    return "true" == str(self.get("aggregation", "host_in_memory_aggregation")).lower()
+    return "true" == str(self.get("aggregation", "host_in_memory_aggregation", "false")).lower()
 
   def get_inmemory_aggregation_port(self):
-    return self.get("aggregation", "host_in_memory_aggregation_port")
+    return self.get("aggregation", "host_in_memory_aggregation_port", "61888")
+
+  def get_inmemory_aggregation_protocol(self):
+    return self.get("aggregation", "host_in_memory_aggregation_protocol", "http")
 
   def get_aggregator_jvm_agrs(self):
     hosts = self.get("aggregation", "jvm_arguments", "-Xmx256m -Xms128m -XX:PermSize=68m")

ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/emitter.py

@@ -54,17 +54,19 @@ class Emitter(threading.Thread):
     self.application_metric_map = application_metric_map
     self.collector_port = config.get_server_port()
     self.all_metrics_collector_hosts = config.get_metrics_collector_hosts_as_list()
-    self.is_server_https_enabled = config.is_server_https_enabled()
+    self.is_collector_https_enabled = config.is_collector_https_enabled()
+    self.collector_protocol = "https" if self.is_collector_https_enabled else "http"
     self.set_instanceid = config.is_set_instanceid()
     self.instanceid = config.get_instanceid()
     self.is_inmemory_aggregation_enabled = config.is_inmemory_aggregation_enabled()
 
     if self.is_inmemory_aggregation_enabled:
-      self.collector_port = config.get_inmemory_aggregation_port()
-      self.all_metrics_collector_hosts = ['localhost']
-      self.is_server_https_enabled = False
+      self.inmemory_aggregation_port = config.get_inmemory_aggregation_port()
+      self.inmemory_aggregation_protocol = config.get_inmemory_aggregation_protocol()
+      if self.inmemory_aggregation_protocol == "https":
+        self.ca_certs = config.get_ca_certs()
 
-    if self.is_server_https_enabled:
+    if self.is_collector_https_enabled:
       self.ca_certs = config.get_ca_certs()
 
     # TimedRoundRobinSet
@@ -101,22 +103,26 @@ class Emitter(threading.Thread):
 
   def push_metrics(self, data):
     success = False
-    while self.active_collector_hosts.get_actual_size() > 0:
+    if self.is_inmemory_aggregation_enabled:
+      success = self.try_with_collector(self.inmemory_aggregation_protocol, "localhost", self.inmemory_aggregation_port, data)
+      if not success:
+        logger.warning("Failed to submit metrics to local aggregator. Trying to post them to collector...")
+    while not success and self.active_collector_hosts.get_actual_size() > 0:
       collector_host = self.get_collector_host_shard()
-      success = self.try_with_collector_host(collector_host, data)
-      if success:
-        break
+      success = self.try_with_collector(self.collector_protocol, collector_host, self.collector_port, data)
     pass
 
     if not success:
       logger.info('No valid collectors found...')
       for collector_host in self.active_collector_hosts:
-        success = self.try_with_collector_host(collector_host, data)
+        success = self.try_with_collector(self.collector_protocol, collector_host, self.ollector_port, data)
+        if success:
+          break
       pass
 
-  def try_with_collector_host(self, collector_host, data):
+  def try_with_collector(self, collector_protocol, collector_host, collector_port, data):
     headers = {"Content-Type" : "application/json", "Accept" : "*/*"}
-    connection = self.get_connection(collector_host)
+    connection = self.get_connection(collector_protocol, collector_host, collector_port)
     logger.debug("message to send: %s" % data)
 
     try:
@@ -169,16 +175,16 @@ class Emitter(threading.Thread):
       logger.warn("Metric collector host {0} was blacklisted.".format(collector_host))
       return False
 
-  def get_connection(self, collector_host):
+  def get_connection(self, protocol, host, port):
     timeout = int(self.send_interval - 10)
-    if self.is_server_https_enabled:
-      connection = CachedHTTPSConnection(collector_host,
-                                         self.collector_port,
+    if protocol == "https":
+      connection = CachedHTTPSConnection(host,
+                                         port,
                                          timeout=timeout,
                                          ca_certs=self.ca_certs)
     else:
-      connection = CachedHTTPConnection(collector_host,
-                                        self.collector_port,
+      connection = CachedHTTPConnection(host,
+                                        port,
                                         timeout=timeout)
     return connection
 

ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/host_info.py

@@ -265,7 +265,6 @@ class HostInfo():
 
     skip_disk_patterns = self.__config.get_disk_metrics_skip_pattern()
     logger.debug('skip_disk_patterns: %s' % skip_disk_patterns)
-    print skip_disk_patterns
     if not skip_disk_patterns or skip_disk_patterns == 'None':
       io_counters = psutil.disk_io_counters()
       print io_counters

ambari-metrics/ambari-metrics-host-monitoring/src/main/python/core/stop_handler.py

@@ -78,7 +78,7 @@ class StopHandlerWindows(StopHandler):
       raise FatalException(-1, "Error waiting for stop event: " + str(result))
     if (win32event.WAIT_TIMEOUT == result):
       return -1
-      logger.info("Stop event received")
+      logger.debug("Stop event received")
     return result # 0 -> stop
 
 
@@ -119,7 +119,7 @@ class StopHandlerLinux(StopHandler):
     # Stop process when stop event received
     self.stop_event.wait(timeout)
     if self.stop_event.isSet():
-      logger.info("Stop event received")
+      logger.debug("Stop event received")
       return 0
     # Timeout
     return -1

ambari-metrics/ambari-metrics-kafka-sink/src/main/java/org/apache/hadoop/metrics2/sink/kafka/KafkaTimelineMetricsReporter.java

@@ -74,6 +74,7 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
   private static final String TIMELINE_METRICS_KAFKA_SET_INSTANCE_ID_PROPERTY = TIMELINE_METRICS_KAFKA_PREFIX + SET_INSTANCE_ID_PROPERTY;
   private static final String TIMELINE_METRICS_KAFKA_HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY = TIMELINE_METRICS_KAFKA_PREFIX + HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY;
   private static final String TIMELINE_METRICS_KAFKA_HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY = TIMELINE_METRICS_KAFKA_PREFIX + HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY;
+  private static final String TIMELINE_METRICS_KAFKA_HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY = TIMELINE_METRICS_KAFKA_PREFIX + HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY;
   private static final String TIMELINE_DEFAULT_HOST = "localhost";
   private static final String TIMELINE_DEFAULT_PORT = "6188";
   private static final String TIMELINE_DEFAULT_PROTOCOL = "http";
@@ -100,6 +101,7 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
   private Set<String> excludedMetrics = new HashSet<>();
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
+  private String hostInMemoryAggregationProtocol;
 
   @Override
   protected String getCollectorUri(String host) {
@@ -147,6 +149,11 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
     return hostInMemoryAggregationPort;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return hostInMemoryAggregationProtocol;
+  }
+
   public void setMetricsCache(TimelineMetricsCache metricsCache) {
     this.metricsCache = metricsCache;
   }
@@ -186,9 +193,10 @@ public class KafkaTimelineMetricsReporter extends AbstractTimelineMetricsSink
 
         hostInMemoryAggregationEnabled = props.getBoolean(TIMELINE_METRICS_KAFKA_HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, false);
         hostInMemoryAggregationPort = props.getInt(TIMELINE_METRICS_KAFKA_HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, 61888);
+        hostInMemoryAggregationProtocol = props.getString(TIMELINE_METRICS_KAFKA_HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
         setMetricsCache(new TimelineMetricsCache(maxRowCacheSize, metricsSendInterval));
 
-        if (metricCollectorProtocol.contains("https")) {
+        if (metricCollectorProtocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
           String trustStorePath = props.getString(TIMELINE_METRICS_SSL_KEYSTORE_PATH_PROPERTY).trim();
           String trustStoreType = props.getString(TIMELINE_METRICS_SSL_KEYSTORE_TYPE_PROPERTY).trim();
           String trustStorePwd = props.getString(TIMELINE_METRICS_SSL_KEYSTORE_PASSWORD_PROPERTY).trim();

ambari-metrics/ambari-metrics-storm-sink-legacy/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java

@@ -57,6 +57,7 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
   private int timeoutSeconds;
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
+  private String hostInMemoryAggregationProtocol;
 
   public StormTimelineMetricsReporter() {
 
@@ -107,6 +108,11 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
     return hostInMemoryAggregationPort;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return hostInMemoryAggregationProtocol;
+  }
+
   @Override
   public void prepare(Map conf) {
     LOG.info("Preparing Storm Metrics Reporter");
@@ -144,11 +150,15 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
         setInstanceId = Boolean.getBoolean(cf.get(SET_INSTANCE_ID_PROPERTY).toString());
         instanceId = cf.get(INSTANCE_ID_PROPERTY).toString();
       }
-      hostInMemoryAggregationEnabled = Boolean.valueOf(cf.get(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY).toString());
-      hostInMemoryAggregationPort = Integer.valueOf(cf.get(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY).toString());
+      hostInMemoryAggregationEnabled = Boolean.valueOf(cf.get(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY) != null ?
+        cf.get(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY).toString() : "false");
+      hostInMemoryAggregationPort = Integer.valueOf(cf.get(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY) != null ?
+        cf.get(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY).toString() : "61888");
+      hostInMemoryAggregationProtocol = cf.get(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY) != null ?
+        cf.get(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY).toString() : "http";
 
       collectorUri = constructTimelineMetricUri(protocol, findPreferredCollectHost(), port);
-      if (protocol.contains("https")) {
+      if (protocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
         String trustStorePath = cf.get(SSL_KEYSTORE_PATH_PROPERTY).toString().trim();
         String trustStoreType = cf.get(SSL_KEYSTORE_TYPE_PROPERTY).toString().trim();
         String trustStorePwd = cf.get(SSL_KEYSTORE_PASSWORD_PROPERTY).toString().trim();

ambari-metrics/ambari-metrics-storm-sink-legacy/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsSink.java

@@ -63,6 +63,7 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
   private String instanceId;
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
+  private String hostInMemoryAggregationProtocol;
 
   @Override
   protected String getCollectorUri(String host) {
@@ -109,6 +110,11 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
     return hostInMemoryAggregationPort;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return hostInMemoryAggregationProtocol;
+  }
+
   @Override
   public void prepare(Map map, Object o, TopologyContext topologyContext, IErrorReporter iErrorReporter) {
     LOG.info("Preparing Storm Metrics Sink");
@@ -138,12 +144,13 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
 
     instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY, null);
     setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY, "false"));
-    hostInMemoryAggregationEnabled = Boolean.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY));
-    hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY));
+    hostInMemoryAggregationEnabled = Boolean.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, "false"));
+    hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, "61888"));
+    hostInMemoryAggregationProtocol = configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
     // Initialize the collector write strategy
     super.init();
 
-    if (protocol.contains("https")) {
+    if (protocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
       String trustStorePath = configuration.getProperty(SSL_KEYSTORE_PATH_PROPERTY).trim();
       String trustStoreType = configuration.getProperty(SSL_KEYSTORE_TYPE_PROPERTY).trim();
       String trustStorePwd = configuration.getProperty(SSL_KEYSTORE_PASSWORD_PROPERTY).trim();

ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsReporter.java

@@ -52,6 +52,7 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
   private int timeoutSeconds;
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
+  private String hostInMemoryAggregationProtocol;
 
   public StormTimelineMetricsReporter() {
 
@@ -102,6 +103,11 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
     return hostInMemoryAggregationPort;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return hostInMemoryAggregationProtocol;
+  }
+
   @Override
   public void prepare(Object registrationArgument) {
     LOG.info("Preparing Storm Metrics Reporter");
@@ -132,10 +138,11 @@ public class StormTimelineMetricsReporter extends AbstractTimelineMetricsSink
       setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY));
       instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY);
 
-      hostInMemoryAggregationEnabled = Boolean.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY));
-      hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY));
+      hostInMemoryAggregationEnabled = Boolean.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, "false"));
+      hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, "61888"));
+      hostInMemoryAggregationProtocol = configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
 
-      if (protocol.contains("https")) {
+      if (protocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
         String trustStorePath = configuration.getProperty(SSL_KEYSTORE_PATH_PROPERTY).trim();
         String trustStoreType = configuration.getProperty(SSL_KEYSTORE_TYPE_PROPERTY).trim();
         String trustStorePwd = configuration.getProperty(SSL_KEYSTORE_PASSWORD_PROPERTY).trim();

ambari-metrics/ambari-metrics-storm-sink/src/main/java/org/apache/hadoop/metrics2/sink/storm/StormTimelineMetricsSink.java

@@ -72,6 +72,7 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
   private boolean setInstanceId;
   private boolean hostInMemoryAggregationEnabled;
   private int hostInMemoryAggregationPort;
+  private String hostInMemoryAggregationProtocol;
 
   @Override
   protected String getCollectorUri(String host) {
@@ -118,6 +119,11 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
     return hostInMemoryAggregationPort;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return hostInMemoryAggregationProtocol;
+  }
+
   @Override
   public void prepare(Map map, Object o, TopologyContext topologyContext, IErrorReporter iErrorReporter) {
     LOG.info("Preparing Storm Metrics Sink");
@@ -150,13 +156,14 @@ public class StormTimelineMetricsSink extends AbstractTimelineMetricsSink implem
     instanceId = configuration.getProperty(INSTANCE_ID_PROPERTY, null);
     setInstanceId = Boolean.valueOf(configuration.getProperty(SET_INSTANCE_ID_PROPERTY, "false"));
 
-    hostInMemoryAggregationEnabled = Boolean.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY));
-    hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY));
+    hostInMemoryAggregationEnabled = Boolean.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_ENABLED_PROPERTY, "false"));
+    hostInMemoryAggregationPort = Integer.valueOf(configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PORT_PROPERTY, "61888"));
+    hostInMemoryAggregationProtocol = configuration.getProperty(HOST_IN_MEMORY_AGGREGATION_PROTOCOL_PROPERTY, "http");
 
     // Initialize the collector write strategy
     super.init();
 
-    if (protocol.contains("https")) {
+    if (protocol.contains("https") || hostInMemoryAggregationProtocol.contains("https")) {
       String trustStorePath = configuration.getProperty(SSL_KEYSTORE_PATH_PROPERTY).trim();
       String trustStoreType = configuration.getProperty(SSL_KEYSTORE_TYPE_PROPERTY).trim();
       String trustStorePwd = configuration.getProperty(SSL_KEYSTORE_PASSWORD_PROPERTY).trim();

ambari-server/src/main/java/org/apache/ambari/server/metrics/system/impl/AmbariMetricSinkImpl.java

@@ -310,6 +310,11 @@ public class AmbariMetricSinkImpl extends AbstractTimelineMetricsSink implements
     return 0;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return "http";
+  }
+
   private List<TimelineMetric> getFilteredMetricList(List<SingleMetric> metrics) {
     final List<TimelineMetric> metricList = new ArrayList<>();
     for (SingleMetric metric : metrics) {

ambari-server/src/main/resources/common-services/ACCUMULO/1.6.1.2.2.0/package/scripts/params.py

@@ -155,6 +155,12 @@ metrics_report_interval = default("/configurations/ams-site/timeline.metrics.sin
 metrics_collection_period = default("/configurations/ams-site/timeline.metrics.sink.collection.period", 10)
 host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 
 # if accumulo is selected accumulo_tserver_hosts should not be empty, but still default just in case
 if 'slave_hosts' in config['clusterHostInfo']:

ambari-server/src/main/resources/common-services/ACCUMULO/1.6.1.2.2.0/package/templates/hadoop-metrics2-accumulo.properties.j2

@@ -18,6 +18,9 @@
 
 *.host_in_memory_aggregation = {{host_in_memory_aggregation}}
 *.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+*.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 {% if has_metric_collector %}
 

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/configuration/ams-site.xml

@@ -849,6 +849,10 @@
       <selection-cardinality>1</selection-cardinality>
     </value-attributes>
   </property>
+  <property>
+    <name>timeline.metrics.host.inmemory.aggregation.http.policy</name>
+    <value>HTTP_ONLY</value>
+  </property>
   <property>
     <name>timeline.metrics.downsampler.event.metric.patterns</name>
     <value></value>

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/scripts/ams.py

@@ -163,19 +163,37 @@ def ams(name=None):
               create_parents = True
     )
 
-    if params.host_in_memory_aggregation and params.log4j_props is not None:
-      File(os.path.join(params.ams_monitor_conf_dir, "log4j.properties"),
-           owner=params.ams_user,
-           content=params.log4j_props
-           )
+    if params.host_in_memory_aggregation:
+      if params.log4j_props is not None:
+        File(os.path.join(params.ams_monitor_conf_dir, "log4j.properties"),
+             owner=params.ams_user,
+             content=params.log4j_props
+             )
+        pass
+
+      XmlConfig("ams-site.xml",
+                conf_dir=params.ams_monitor_conf_dir,
+                configurations=params.config['configurations']['ams-site'],
+                configuration_attributes=params.config['configurationAttributes']['ams-site'],
+                owner=params.ams_user,
+                group=params.user_group
+                )
 
-    XmlConfig("ams-site.xml",
+      XmlConfig("ams-site.xml",
               conf_dir=params.ams_monitor_conf_dir,
               configurations=params.config['configurations']['ams-site'],
               configuration_attributes=params.config['configurationAttributes']['ams-site'],
               owner=params.ams_user,
               group=params.user_group
               )
+      XmlConfig("ssl-server.xml",
+              conf_dir=params.ams_monitor_conf_dir,
+              configurations=params.config['configurations']['ams-ssl-server'],
+              configuration_attributes=params.config['configurationAttributes']['ams-ssl-server'],
+              owner=params.ams_user,
+              group=params.user_group
+              )
+      pass
 
     TemplateConfig(
       os.path.join(params.ams_monitor_conf_dir, "metric_monitor.ini"),
@@ -393,13 +411,21 @@ def ams(name=None, action=None):
            content=InlineTemplate(params.log4j_props)
            )
 
-    XmlConfig("ams-site.xml",
+      XmlConfig("ams-site.xml",
               conf_dir=params.ams_monitor_conf_dir,
               configurations=params.config['configurations']['ams-site'],
               configuration_attributes=params.config['configurationAttributes']['ams-site'],
               owner=params.ams_user,
               group=params.user_group
               )
+      XmlConfig("ssl-server.xml",
+              conf_dir=params.ams_monitor_conf_dir,
+              configurations=params.config['configurations']['ams-ssl-server'],
+              configuration_attributes=params.config['configuration_attributes']['ams-ssl-server'],
+              owner=params.ams_user,
+              group=params.user_group
+              )
+      pass
 
     Execute(format("{sudo} chown -R {ams_user}:{user_group} {ams_monitor_log_dir}")
             )
@@ -440,7 +466,7 @@ def ams(name=None, action=None):
          content=InlineTemplate(params.ams_env_sh_template)
     )
 
-    if params.metric_collector_https_enabled:
+    if params.metric_collector_https_enabled or params.is_aggregation_https_enabled:
       export_ca_certs(params.ams_monitor_conf_dir)
 
     pass

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/scripts/params.py

@@ -234,8 +234,14 @@ metrics_collector_heapsize = check_append_heap_property(str(metrics_collector_he
 master_heapsize = check_append_heap_property(str(master_heapsize), "m")
 regionserver_heapsize = check_append_heap_property(str(regionserver_heapsize), "m")
 
-host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
+host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", False)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 host_in_memory_aggregation_jvm_arguments = default("/configurations/ams-env/timeline.metrics.host.inmemory.aggregation.jvm.arguments",
                                                    "-Xmx256m -Xms128m -XX:PermSize=68m")
 

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/templates/hadoop-metrics2-hbase.properties.j2

@@ -16,67 +16,47 @@
 # limitations under the License.
 #}
 
-# 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.
+hbase.extendedperiod = 3600
 
-# syntax: [prefix].[source|sink|jmx].[instance].[options]
-# See package.html for org.apache.hadoop.metrics2 for details
+hbase.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
+hbase.period=30
+hbase.collector.hosts={{ams_collector_hosts}}
+hbase.port={{metric_collector_port}}
+hbase.protocol={{metric_collector_protocol}}
 
-# HBase-specific configuration to reset long-running stats (e.g. compactions)
-# If this variable is left out, then the default is no expiration.
+jvm.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
+jvm.period=30
+jvm.collector.hosts={{ams_collector_hosts}}
+jvm.port={{metric_collector_port}}
+jvm.protocol={{metric_collector_protocol}}
 
-# Disable metrics since AMS 2.X doesn't work with Hadoop 3.x sink
+rpc.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
+rpc.period=30
+rpc.collector.hosts={{ams_collector_hosts}}
+rpc.port={{metric_collector_port}}
+rpc.protocol={{metric_collector_protocol}}
 
-#hbase.extendedperiod = 3600
-#
-#hbase.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
-#hbase.period=30
-#hbase.collector.hosts={{ams_collector_hosts}}
-#hbase.port={{metric_collector_port}}
-#hbase.protocol={{metric_collector_protocol}}
-#
-#jvm.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
-#jvm.period=30
-#jvm.collector.hosts={{ams_collector_hosts}}
-#jvm.port={{metric_collector_port}}
-#jvm.protocol={{metric_collector_protocol}}
-#
-#rpc.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
-#rpc.period=30
-#rpc.collector.hosts={{ams_collector_hosts}}
-#rpc.port={{metric_collector_port}}
-#rpc.protocol={{metric_collector_protocol}}
-#
-#*.timeline.plugin.urls=file:///usr/lib/ambari-metrics-hadoop-sink/ambari-metrics-hadoop-sink.jar
-#*.sink.timeline.slave.host.name={{hostname}}
-#*.host_in_memory_aggregation = {{host_in_memory_aggregation}}
-#*.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
-#
-#hbase.sink.timeline.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
-#hbase.sink.timeline.period={{metrics_collection_period}}
-#hbase.sink.timeline.sendInterval={{metrics_report_interval}}000
-#hbase.sink.timeline.collector.hosts={{ams_collector_hosts}}
-#hbase.sink.timeline.port={{metric_collector_port}}
-#hbase.sink.timeline.protocol={{metric_collector_protocol}}
-#hbase.sink.timeline.serviceName-prefix=ams
-#
-## HTTPS properties
-#hbase.sink.timeline.truststore.path = {{metric_truststore_path}}
-#hbase.sink.timeline.truststore.type = {{metric_truststore_type}}
-#hbase.sink.timeline.truststore.password = {{metric_truststore_password}}
-#
-## Switch off metrics generation on a per region basis
-#*.source.filter.class=org.apache.hadoop.metrics2.filter.RegexFilter
-#hbase.*.source.filter.exclude=.*(Regions|Users|Tables).*
+*.timeline.plugin.urls=file:///usr/lib/ambari-metrics-hadoop-sink/ambari-metrics-hadoop-sink.jar
+*.sink.timeline.slave.host.name={{hostname}}
+*.host_in_memory_aggregation = {{host_in_memory_aggregation}}
+*.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+*.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
+
+hbase.sink.timeline.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
+hbase.sink.timeline.period={{metrics_collection_period}}
+hbase.sink.timeline.sendInterval={{metrics_report_interval}}000
+hbase.sink.timeline.collector.hosts={{ams_collector_hosts}}
+hbase.sink.timeline.port={{metric_collector_port}}
+hbase.sink.timeline.protocol={{metric_collector_protocol}}
+hbase.sink.timeline.serviceName-prefix=ams
+
+# HTTPS properties
+hbase.sink.timeline.truststore.path = {{metric_truststore_path}}
+hbase.sink.timeline.truststore.type = {{metric_truststore_type}}
+hbase.sink.timeline.truststore.password = {{metric_truststore_password}}
+
+# Switch off metrics generation on a per region basis
+*.source.filter.class=org.apache.hadoop.metrics2.filter.RegexFilter
+hbase.*.source.filter.exclude=.*(Regions|Users|Tables).*

ambari-server/src/main/resources/common-services/AMBARI_METRICS/0.1.0/package/templates/metric_monitor.ini.j2

@@ -44,6 +44,9 @@ https_enabled = {{metric_collector_https_enabled}}
 [aggregation]
 host_in_memory_aggregation = {{host_in_memory_aggregation}}
 host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 java_home = {{java64_home}}
 jvm_arguments = {{host_in_memory_aggregation_jvm_arguments}}
 ams_monitor_log_dir = {{ams_monitor_log_dir}}

ambari-server/src/main/resources/common-services/FLUME/1.4.0.2.0/package/scripts/params.py

@@ -126,6 +126,12 @@ metrics_collection_period = default("/configurations/ams-site/timeline.metrics.s
 
 host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 
 # Cluster Zookeeper quorum
 zookeeper_quorum = None

ambari-server/src/main/resources/common-services/FLUME/1.4.0.2.0/package/templates/flume-metrics2.properties.j2

@@ -25,6 +25,9 @@ maxRowCacheSize=10000
 sendInterval={{metrics_report_interval}}000
 host_in_memory_aggregation = {{host_in_memory_aggregation}}
 host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 truststore.path = {{metric_truststore_path}}

ambari-server/src/main/resources/common-services/HBASE/0.96.0.2.0/package/scripts/params_linux.py

@@ -193,6 +193,12 @@ metrics_collection_period = default("/configurations/ams-site/timeline.metrics.s
 
 host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 
 # if hbase is selected the hbase_rs_hosts, should not be empty, but still default just in case
 if 'slave_hosts' in config['clusterHostInfo']:

ambari-server/src/main/resources/common-services/HBASE/0.96.0.2.0/package/templates/hadoop-metrics2-hbase.properties-GANGLIA-MASTER.j2

@@ -78,6 +78,9 @@ hbase.sink.timeline.protocol={{metric_collector_protocol}}
 hbase.sink.timeline.port={{metric_collector_port}}
 hbase.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
 hbase.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+hbase.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 hbase.sink.timeline.truststore.path = {{metric_truststore_path}}

ambari-server/src/main/resources/common-services/HBASE/0.96.0.2.0/package/templates/hadoop-metrics2-hbase.properties-GANGLIA-RS.j2

@@ -76,6 +76,9 @@ hbase.sink.timeline.protocol={{metric_collector_protocol}}
 hbase.sink.timeline.port={{metric_collector_port}}
 hbase.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
 hbase.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+hbase.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 hbase.sink.timeline.truststore.path = {{metric_truststore_path}}

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/scripts/params_linux.py

@@ -580,6 +580,12 @@ metrics_collection_period = default("/configurations/ams-site/timeline.metrics.s
 
 host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 ########################################################
 ############# Atlas related params #####################
 ########################################################

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/templates/hadoop-metrics2-hivemetastore.properties.j2

@@ -53,6 +53,8 @@
   hivemetastore.sink.timeline.protocol={{metric_collector_protocol}}
   hivemetastore.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
   hivemetastore.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
-
+  {% if is_aggregation_https_enabled %}
+    hivemetastore.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
 {% endif %}

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/templates/hadoop-metrics2-hiveserver2.properties.j2

@@ -53,5 +53,8 @@
   hiveserver2.sink.timeline.protocol={{metric_collector_protocol}}
   hiveserver2.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
   hiveserver2.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    hiveserver2.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
 {% endif %}

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/templates/hadoop-metrics2-llapdaemon.j2

@@ -52,5 +52,8 @@
   llapdaemon.sink.timeline.protocol={{metric_collector_protocol}}
   llapdaemon.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
   llapdaemon.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    llapdaemon.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
 {% endif %}

ambari-server/src/main/resources/common-services/HIVE/0.12.0.2.0/package/templates/hadoop-metrics2-llaptaskscheduler.j2

@@ -52,5 +52,8 @@
   llaptaskscheduler.sink.timeline.protocol={{metric_collector_protocol}}
   llaptaskscheduler.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
   llaptaskscheduler.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    llaptaskscheduler.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
 {% endif %}

ambari-server/src/main/resources/common-services/KAFKA/0.8.1/configuration/kafka-broker.xml

@@ -424,4 +424,9 @@
     <value>{{host_in_memory_aggregation_port}}</value>
     <on-ambari-upgrade add="true"/>
   </property>
+  <property>
+    <name>kafka.timeline.metrics.host_in_memory_aggregation_protocol</name>
+    <value>{{host_in_memory_aggregation_protocol}}</value>
+    <on-ambari-upgrade add="true"/>
+  </property>
 </configuration>

ambari-server/src/main/resources/common-services/KAFKA/0.8.1/package/scripts/params.py

@@ -154,6 +154,12 @@ if has_metric_collector:
 
   host_in_memory_aggregation = str(default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)).lower()
   host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+  is_aggregation_https_enabled = False
+  if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+    host_in_memory_aggregation_protocol = 'https'
+    is_aggregation_https_enabled = True
+  else:
+    host_in_memory_aggregation_protocol = 'http'
   pass
 
 # Security-related params

ambari-server/src/main/resources/common-services/STORM/0.9.1/package/scripts/params_linux.py

@@ -210,6 +210,12 @@ metric_collector_sink_jar = "/usr/lib/storm/lib/ambari-metrics-storm-sink-with-c
 metric_collector_legacy_sink_jar = "/usr/lib/storm/lib/ambari-metrics-storm-sink-legacy-with-common-*.jar"
 host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 
 
 # Cluster Zookeeper quorum

ambari-server/src/main/resources/common-services/STORM/0.9.1/package/templates/config.yaml.j2

@@ -61,8 +61,11 @@ metrics_collector:
   protocol: "{{metric_collector_protocol}}"
   port: "{{metric_collector_port}}"
   appId: "{{metric_collector_app_id}}"
-  host_in_memory_aggregation = {{host_in_memory_aggregation}}
-  host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  host_in_memory_aggregation: {{host_in_memory_aggregation}}
+  host_in_memory_aggregation_port: {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    host_in_memory_aggregation_protocol: {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
   # HTTPS settings
   truststore.path : "{{metric_truststore_path}}"

ambari-server/src/main/resources/common-services/STORM/0.9.1/package/templates/storm-metrics2.properties.j2

@@ -25,6 +25,9 @@ sendInterval={{metrics_report_interval}}000
 clusterReporterAppId=nimbus
 host_in_memory_aggregation = {{host_in_memory_aggregation}}
 host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 truststore.path = {{metric_truststore_path}}

ambari-server/src/main/resources/stack-hooks/before-START/scripts/params.py

@@ -169,6 +169,12 @@ metrics_collection_period = default("/configurations/ams-site/timeline.metrics.s
 
 host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 
 # Cluster Zookeeper quorum
 zookeeper_quorum = None

ambari-server/src/main/resources/stack-hooks/before-START/templates/hadoop-metrics2.properties.j2

@@ -77,6 +77,9 @@ resourcemanager.sink.ganglia.tagsForPrefix.yarn=Queue
 *.sink.timeline.port={{metric_collector_port}}
 *.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
 *.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+*.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 *.sink.timeline.truststore.path = {{metric_truststore_path}}

ambari-server/src/main/resources/stacks/HDP/2.1.GlusterFS/services/STORM/package/templates/config.yaml.j2

@@ -1,21 +1,3 @@
-{#
-# 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.
-#}
-
 # 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.
@@ -62,4 +44,32 @@ ganglia:
   # an <IP>:<HOSTNAME> pair to spoof
   # this allows us to simulate storm cluster metrics coming from a specific host
   #spoof: "192.168.1.1:storm"
-{% endif %}
+{% endif %}
+
+{% if has_metric_collector and stack_supports_storm_ams %}
+enableGanglia: False
+
+ganglia:
+  reportInterval: {{metric_collector_report_interval}}
+
+enableMetricsSink: True
+
+metrics_collector:
+
+  reportInterval: {{metric_collector_report_interval}}
+  collector.hosts: "{{ams_collector_hosts}}"
+  protocol: "{{metric_collector_protocol}}"
+  port: "{{metric_collector_port}}"
+  appId: "{{metric_collector_app_id}}"
+  host_in_memory_aggregation: {{host_in_memory_aggregation}}
+  host_in_memory_aggregation_port: {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    host_in_memory_aggregation_protocol: {{host_in_memory_aggregation_protocol}}
+  {% endif %}
+
+  # HTTPS settings
+  truststore.path : "{{metric_truststore_path}}"
+  truststore.type : "{{metric_truststore_type}}"
+  truststore.password : "{{metric_truststore_password}}"
+
+{% endif %}

ambari-server/src/main/resources/stacks/HDP/2.6/services/HDFS/configuration/hadoop-metrics2.properties.xml

@@ -88,6 +88,9 @@ resourcemanager.sink.ganglia.tagsForPrefix.yarn=Queue
 *.sink.timeline.port={{metric_collector_port}}
 *.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
 *.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+*.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 *.sink.timeline.truststore.path = {{metric_truststore_path}}

ambari-server/src/main/resources/stacks/PERF/1.0/services/FAKEHDFS/configuration/hadoop-metrics2.properties.xml

@@ -86,6 +86,11 @@ resourcemanager.sink.ganglia.tagsForPrefix.yarn=Queue
 *.sink.timeline.zookeeper.quorum={{zookeeper_quorum}}
 *.sink.timeline.protocol={{metric_collector_protocol}}
 *.sink.timeline.port={{metric_collector_port}}
+*.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
+*.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+*.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 *.sink.timeline.truststore.path = {{metric_truststore_path}}

ambari-server/src/test/java/org/apache/ambari/server/metric/system/impl/TestAmbariMetricsSinkImpl.java

@@ -83,6 +83,11 @@ public class TestAmbariMetricsSinkImpl extends AbstractTimelineMetricsSink imple
     return 61888;
   }
 
+  @Override
+  protected String getHostInMemoryAggregationProtocol() {
+    return "http";
+  }
+
   @Override
   public void init(MetricsConfiguration configuration) {
 

ambari-server/src/test/python/stacks/2.0.6/AMBARI_METRICS/test_metrics_monitor.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python
+
+'''
+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.
+'''
+
+from mock.mock import MagicMock, patch
+from stacks.utils.RMFTestCase import *
+
+@patch("tempfile.mkdtemp", new = MagicMock(return_value='/some_tmp_dir'))
+@patch("os.path.exists", new = MagicMock(return_value=True))
+@patch("platform.linux_distribution", new = MagicMock(return_value="Linux"))
+class TestMetricsMonitor(RMFTestCase):
+  COMMON_SERVICES_PACKAGE_DIR = "AMBARI_METRICS/0.1.0/package"
+  STACK_VERSION = "2.0.6"
+  DEFAULT_IMMUTABLE_PATHS = ['/apps/hive/warehouse', '/apps/falcon', '/mr-history/done', '/app-logs', '/tmp']
+
+  def test_start_default_with_aggregator_https(self):
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/metrics_monitor.py",
+                       classname = "AmsMonitor",
+                       command = "start",
+                       config_file="default.json",
+                       stack_version = self.STACK_VERSION,
+                       target = RMFTestCase.TARGET_COMMON_SERVICES
+                       )
+    self.maxDiff=None
+    self.assert_ams(inmemory_aggregation = False)
+    self.assertResourceCalled('Execute', 'ambari-sudo.sh /usr/jdk64/jdk1.7.0_45/bin/keytool -importkeystore -srckeystore /etc/security/clientKeys/all.jks -destkeystore /some_tmp_dir/truststore.p12 -srcalias c6402.ambari.apache.org -deststoretype PKCS12 -srcstorepass bigdata -deststorepass bigdata',
+                              )
+    self.assertResourceCalled('Execute', 'ambari-sudo.sh openssl pkcs12 -in /some_tmp_dir/truststore.p12 -out /etc/ambari-metrics-monitor/conf/ca.pem -cacerts -nokeys -passin pass:bigdata',
+                              )
+    self.assertResourceCalled('Execute', ('chown', u'ams:hadoop', '/etc/ambari-metrics-monitor/conf/ca.pem'),
+                              sudo=True
+                              )
+    self.assertResourceCalled('Execute', ('chmod', '644', '/etc/ambari-metrics-monitor/conf/ca.pem'),
+                              sudo=True)
+    self.assertResourceCalled('Execute', 'ambari-sudo.sh rm -rf /some_tmp_dir',
+                              )
+    self.assertResourceCalled('Execute', '/usr/sbin/ambari-metrics-monitor --config /etc/ambari-metrics-monitor/conf start',
+                              user = 'ams'
+                              )
+    self.assertNoMoreResources()
+
+  def test_start_inmemory_aggregator(self):
+    self.executeScript(self.COMMON_SERVICES_PACKAGE_DIR + "/scripts/metrics_monitor.py",
+                       classname = "AmsMonitor",
+                       command = "start",
+                       config_file="default_ams_embedded.json",
+                       stack_version = self.STACK_VERSION,
+                       target = RMFTestCase.TARGET_COMMON_SERVICES
+                       )
+    self.maxDiff=None
+    self.assert_ams(inmemory_aggregation = True)
+
+    self.assertResourceCalled('Execute', '/usr/sbin/ambari-metrics-monitor --config /etc/ambari-metrics-monitor/conf start',
+                              user = 'ams'
+                              )
+    self.assertNoMoreResources()
+
+  def assert_ams(self, inmemory_aggregation=False):
+    self.assertResourceCalled('Directory', '/etc/ambari-metrics-monitor/conf',
+                              owner = 'ams',
+                              group = 'hadoop',
+                              create_parents = True
+                              )
+
+    self.assertResourceCalled('Directory', '/var/log/ambari-metrics-monitor',
+                              owner = 'ams',
+                              group = 'hadoop',
+                              mode = 0755,
+                              create_parents = True
+                              )
+
+    if inmemory_aggregation:
+      self.assertResourceCalled('File', '/etc/ambari-metrics-monitor/conf/log4j.properties',
+                                owner = 'ams',
+                                group = 'hadoop',
+                                content = InlineTemplate(self.getConfig()['configurations']['ams-log4j']['content']),
+                                mode=0644,
+                                )
+      self.assertResourceCalled('XmlConfig', 'ams-site.xml',
+                                owner = 'ams',
+                                group = 'hadoop',
+                                conf_dir = '/etc/ambari-metrics-monitor/conf',
+                                configurations = self.getConfig()['configurations']['ams-site'],
+                                configuration_attributes = self.getConfig()['configuration_attributes']['ams-hbase-site']
+                                )
+
+      self.assertResourceCalled('XmlConfig', 'ssl-server.xml',
+                              owner = 'ams',
+                              group = 'hadoop',
+                              conf_dir = '/etc/ambari-metrics-monitor/conf',
+                              configurations = self.getConfig()['configurations']['ams-ssl-server'],
+                              configuration_attributes = self.getConfig()['configuration_attributes']['ams-ssl-server']
+                              )
+      pass
+
+    self.assertResourceCalled('Execute', 'ambari-sudo.sh chown -R ams:hadoop /var/log/ambari-metrics-monitor')
+    self.assertResourceCalled('Directory', '/var/run/ambari-metrics-monitor',
+                              owner = 'ams',
+                              group = 'hadoop',
+                              mode = 0755,
+                              cd_access = 'a',
+                              create_parents = True
+                              )
+    self.assertResourceCalled('Directory', '/usr/lib/python2.6/site-packages/resource_monitoring/psutil/build',
+                              owner = 'ams',
+                              group = 'hadoop',
+                              cd_access = 'a',
+                              create_parents = True
+                              )
+
+    self.assertResourceCalled('Execute', 'ambari-sudo.sh chown -R ams:hadoop /usr/lib/python2.6/site-packages/resource_monitoring')
+    self.assertResourceCalled('TemplateConfig', '/etc/ambari-metrics-monitor/conf/metric_monitor.ini',
+                              owner = 'ams',
+                              group = 'hadoop',
+                              template_tag = None,
+                              )
+    self.assertResourceCalled('TemplateConfig', '/etc/ambari-metrics-monitor/conf/metric_groups.conf',
+                              owner = 'ams',
+                              group = 'hadoop',
+                              template_tag = None,
+                              )
+    self.assertResourceCalled('File', '/etc/ambari-metrics-monitor/conf/ams-env.sh',
+                              owner = 'ams',
+                              content = InlineTemplate(self.getConfig()['configurations']['ams-env']['content'])
+                              )
+

ambari-server/src/test/python/stacks/2.0.6/configs/default_ams_embedded.json

@@ -956,6 +956,7 @@
             "content": "\n"
         },
         "ams-site": {
+            "timeline.metrics.host.inmemory.aggregation": "true",
             "timeline.metrics.host.aggregator.minute.ttl": "604800",
             "timeline.metrics.cluster.aggregator.daily.checkpointCutOffMultiplier": "1",
             "timeline.metrics.cluster.aggregator.daily.ttl": "63072000",

contrib/management-packs/hdf-ambari-mpack/src/main/resources/stacks/HDF/2.0/hooks/before-START/scripts/params.py

@@ -137,6 +137,12 @@ metrics_collection_period = default("/configurations/ams-site/timeline.metrics.s
 
 host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
 host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
 #hadoop params
 
 if has_namenode or dfs_type == 'HCFS':

contrib/management-packs/hdf-ambari-mpack/src/main/resources/stacks/HDF/2.0/hooks/before-START/templates/hadoop-metrics2.properties.j2

@@ -71,7 +71,15 @@ resourcemanager.sink.ganglia.tagsForPrefix.yarn=Queue
 *.sink.timeline.class=org.apache.hadoop.metrics2.sink.timeline.HadoopTimelineMetricsSink
 *.sink.timeline.period={{metrics_collection_period}}
 *.sink.timeline.sendInterval={{metrics_report_interval}}000
-*.sink.timeline.slave.host.name = {{hostname}}
+*.sink.timeline.slave.host.name={{hostname}}
+*.sink.timeline.zookeeper.quorum={{zookeeper_quorum}}
+*.sink.timeline.protocol={{metric_collector_protocol}}
+*.sink.timeline.port={{metric_collector_port}}
+*.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
+*.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+{% if is_aggregation_https_enabled %}
+*.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+{% endif %}
 
 # HTTPS properties
 *.sink.timeline.truststore.path = {{metric_truststore_path}}

contrib/management-packs/odpi-ambari-mpack/src/main/resources/stacks/ODPi/2.0/hooks/before-START/scripts/params.py

@@ -134,6 +134,25 @@ if has_metric_collector:
   pass
 metrics_report_interval = default("/configurations/ams-site/timeline.metrics.sink.report.interval", 60)
 metrics_collection_period = default("/configurations/ams-site/timeline.metrics.sink.collection.period", 10)
+host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
+host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
+
+# Cluster Zookeeper quorum
+zookeeper_quorum = None
+if has_zk_host:
+  if 'zoo.cfg' in config['configurations'] and 'clientPort' in config['configurations']['zoo.cfg']:
+    zookeeper_clientPort = config['configurations']['zoo.cfg']['clientPort']
+  else:
+    zookeeper_clientPort = '2181'
+  zookeeper_quorum = (':' + zookeeper_clientPort + ',').join(config['clusterHostInfo']['zookeeper_hosts'])
+  # last port config
+  zookeeper_quorum += ':' + zookeeper_clientPort
 
 #hadoop params
 

contrib/management-packs/odpi-ambari-mpack/src/main/resources/stacks/ODPi/2.0/services/HIVE/package/scripts/params_linux.py

@@ -509,6 +509,15 @@ if has_metric_collector:
 metrics_report_interval = default("/configurations/ams-site/timeline.metrics.sink.report.interval", 60)
 metrics_collection_period = default("/configurations/ams-site/timeline.metrics.sink.collection.period", 10)
 
+host_in_memory_aggregation = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation", True)
+host_in_memory_aggregation_port = default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.port", 61888)
+is_aggregation_https_enabled = False
+if default("/configurations/ams-site/timeline.metrics.host.inmemory.aggregation.http.policy", "HTTP_ONLY") == "HTTPS_ONLY":
+  host_in_memory_aggregation_protocol = 'https'
+  is_aggregation_https_enabled = True
+else:
+  host_in_memory_aggregation_protocol = 'http'
+
 ########################################################
 ############# Atlas related params #####################
 ########################################################

contrib/management-packs/odpi-ambari-mpack/src/main/resources/stacks/ODPi/2.0/services/HIVE/package/templates/hadoop-metrics2-hivemetastore.properties.j2

@@ -48,7 +48,13 @@
   *.sink.timeline.truststore.type = {{metric_truststore_type}}
   *.sink.timeline.truststore.password = {{metric_truststore_password}}
 
-  hivemetastore.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
-
+  hivemetastore.sink.timeline.collector.hosts={{ams_collector_hosts}}
+  hivemetastore.sink.timeline.port={{metric_collector_port}}
+  hivemetastore.sink.timeline.protocol={{metric_collector_protocol}}
+  hivemetastore.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
+  hivemetastore.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    hivemetastore.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
 {% endif %}

contrib/management-packs/odpi-ambari-mpack/src/main/resources/stacks/ODPi/2.0/services/HIVE/package/templates/hadoop-metrics2-hiveserver2.properties.j2

@@ -48,7 +48,13 @@
   *.sink.timeline.truststore.type = {{metric_truststore_type}}
   *.sink.timeline.truststore.password = {{metric_truststore_password}}
 
-  hiveserver2.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
-
+  hiveserver2.sink.timeline.collector.hosts={{ams_collector_hosts}}
+  hiveserver2.sink.timeline.port={{metric_collector_port}}
+  hiveserver2.sink.timeline.protocol={{metric_collector_protocol}}
+  hiveserver2.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
+  hiveserver2.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    hiveserver2.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
 {% endif %}

contrib/management-packs/odpi-ambari-mpack/src/main/resources/stacks/ODPi/2.0/services/HIVE/package/templates/hadoop-metrics2-llapdaemon.j2

@@ -47,6 +47,13 @@
   *.sink.timeline.truststore.type = {{metric_truststore_type}}
   *.sink.timeline.truststore.password = {{metric_truststore_password}}
 
-  llapdaemon.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+  llapdaemon.sink.timeline.collector.hosts={{ams_collector_hosts}}
+  llapdaemon.sink.timeline.port={{metric_collector_port}}
+  llapdaemon.sink.timeline.protocol={{metric_collector_protocol}}
+  llapdaemon.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
+  llapdaemon.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    llapdaemon.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
-{% endif %}
+{% endif %}

contrib/management-packs/odpi-ambari-mpack/src/main/resources/stacks/ODPi/2.0/services/HIVE/package/templates/hadoop-metrics2-llaptaskscheduler.j2

@@ -47,6 +47,13 @@
   *.sink.timeline.truststore.type = {{metric_truststore_type}}
   *.sink.timeline.truststore.password = {{metric_truststore_password}}
 
-  llaptaskscheduler.sink.timeline.collector={{metric_collector_protocol}}://{{metric_collector_host}}:{{metric_collector_port}}
+  llaptaskscheduler.sink.timeline.collector.hosts={{ams_collector_hosts}}
+  llaptaskscheduler.sink.timeline.port={{metric_collector_port}}
+  llaptaskscheduler.sink.timeline.protocol={{metric_collector_protocol}}
+  llaptaskscheduler.sink.timeline.host_in_memory_aggregation = {{host_in_memory_aggregation}}
+  llaptaskscheduler.sink.timeline.host_in_memory_aggregation_port = {{host_in_memory_aggregation_port}}
+  {% if is_aggregation_https_enabled %}
+    llaptaskscheduler.sink.timeline.host_in_memory_aggregation_protocol = {{host_in_memory_aggregation_protocol}}
+  {% endif %}
 
 {% endif %}