AMBARI-26316: Add metrics for Agent Stomp and Api Stomp performance

ambari-funtest/src/test/resources/metrics.properties

@@ -21,3 +21,4 @@
 # Source interval determines how often the metric is sent to sink. Its unit is in seconds
 metric.sources=jvm
 source.jvm.class=org.apache.ambari.server.metrics.system.impl.JvmMetricsSource
+source.stats.class=org.apache.ambari.server.metrics.system.impl.StompStatsMetricsSource

ambari-server/conf/unix/metrics.properties

@@ -19,12 +19,13 @@
 
 #################### Metrics Source Configs #####################
 
-#Metric sources : jvm,database
+#Metric sources : jvm,database, stats
 metric.sources=jvm,event
 
 #### JVM Source Configs ###
 source.jvm.class=org.apache.ambari.server.metrics.system.impl.JvmMetricsSource
 source.event.class=org.apache.ambari.server.metrics.system.impl.StompEventsMetricsSource
+source.stats.class=org.apache.ambari.server.metrics.system.impl.StompStatsMetricsSource
 source.jvm.interval=10
 
 #### Database Source Configs ###

ambari-server/src/main/java/org/apache/ambari/server/controller/AmbariServer.java

@@ -86,6 +86,8 @@
 import org.apache.ambari.server.ldap.LdapModule;
 import org.apache.ambari.server.listeners.WebSocketInitializerListener;
 import org.apache.ambari.server.metrics.system.MetricsService;
+import org.apache.ambari.server.metrics.system.impl.MetricsConfiguration;
+import org.apache.ambari.server.metrics.system.impl.MetricsServiceImpl;
 import org.apache.ambari.server.orm.GuiceJpaInitializer;
 import org.apache.ambari.server.orm.PersistenceType;
 import org.apache.ambari.server.orm.dao.BlueprintDAO;
@@ -166,6 +168,7 @@
 import org.springframework.web.context.support.AnnotationConfigWebApplicationContext;
 import org.springframework.web.filter.DelegatingFilterProxy;
 import org.springframework.web.servlet.DispatcherServlet;
+import org.springframework.web.socket.config.WebSocketMessageBrokerStats;
 
 import com.google.common.base.Joiner;
 import com.google.common.util.concurrent.ServiceManager;
@@ -598,6 +601,12 @@ public void run() throws Exception {
       LOG.info("********* Started Services **********");
 
       if (!configs.isMetricsServiceDisabled()) {
+        if (MetricsConfiguration.isStompStatMetricsConfigured() && metricsService instanceof MetricsServiceImpl) {
+          WebSocketMessageBrokerStats apiStompStats = apiDispatcherContext.getBean(WebSocketMessageBrokerStats.class);
+          ((MetricsServiceImpl) metricsService).setApiStompStats(apiStompStats);
+          WebSocketMessageBrokerStats agentStompStats = agentDispatcherContext.getBean(WebSocketMessageBrokerStats.class);
+          ((MetricsServiceImpl) metricsService).setAgentStompStats(agentStompStats);
+        }
         metricsService.start();
       } else {
         LOG.info("AmbariServer Metrics disabled.");

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

@@ -124,4 +124,30 @@ public static MetricsConfiguration getSubsetConfiguration(MetricsConfiguration m
 
     return new MetricsConfiguration(subsetProperties);
   }
+
+  /**
+   * For enabling stompStats Metrics, append 'stats' in metric.sources.
+   * @return true if StompStatsMetricsSource is configured, false otherwise.
+   */
+  public static boolean isStompStatMetricsConfigured() {
+    MetricsConfiguration configuration = getMetricsConfiguration();
+    String commaSeparatedSources = configuration.getProperty("metric.sources");
+    if (StringUtils.isEmpty(commaSeparatedSources)) {
+      return false;
+    }
+
+    String[] sourceNames = commaSeparatedSources.split(",");
+    for (String sourceName : sourceNames) {
+      if (StringUtils.isEmpty(sourceName)) {
+        continue;
+      }
+      sourceName = sourceName.trim();
+
+      String className = configuration.getProperty("source." + sourceName + ".class");
+      if (className.equals(StompStatsMetricsSource.class.getName())) {
+        return true;
+      }
+    }
+    return false;
+  }
 }

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

@@ -30,6 +30,7 @@
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.web.socket.config.WebSocketMessageBrokerStats;
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
@@ -46,6 +47,9 @@ public class MetricsServiceImpl implements MetricsService {
   @Inject
   STOMPUpdatePublisher STOMPUpdatePublisher;
 
+  private WebSocketMessageBrokerStats apiStompStats;
+  private WebSocketMessageBrokerStats agentStompStats;
+
   @Override
   public void start() {
     LOG.info("********* Initializing AmbariServer Metrics Service **********");
@@ -119,6 +123,11 @@ private void initializeMetricSources() {
           STOMPUpdatePublisher.registerAPI(src);
           STOMPUpdatePublisher.registerAgent(src);
         }
+
+        if (src instanceof StompStatsMetricsSource && apiStompStats != null && agentStompStats != null) {
+          ((StompStatsMetricsSource) src).setApiStompStats(apiStompStats);
+          ((StompStatsMetricsSource) src).setAgentStompStats(agentStompStats);
+        }
         src.start();
       }
 
@@ -134,4 +143,12 @@ public static MetricsSource getSource(String type) {
   public static MetricsSink getSink() {
     return sink;
   }
+
+  public void setApiStompStats(WebSocketMessageBrokerStats apiStompStats) {
+    this.apiStompStats = apiStompStats;
+  }
+
+  public void setAgentStompStats(WebSocketMessageBrokerStats agentStompStats) {
+    this.agentStompStats = agentStompStats;
+  }
 }

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

@@ -0,0 +1,128 @@
+/*
+ * 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.metrics.system.impl;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import org.apache.ambari.server.metrics.system.SingleMetric;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.socket.config.WebSocketMessageBrokerStats;
+
+/**
+ * Gets the metrics about stomp stats connections and publishes to configured
+ * Metric Sink.
+ */
+public class StompStatsMetricsSource extends AbstractMetricsSource {
+    public static final String[] metricsTypes = { "stomp.api", "stomp.agent" };
+    public static final String[] poolMetrics = { "pool_size", "active_threads", "queued_tasks", "completed_tasks" };
+    public static final String[] webSocketMetrics = { "current_client_session", "current_web_socket_session",
+            "current_http_stream", "current_http_polling",
+            "total_sessions_established", "abnormally_closed_session", "connect_failure_session",
+            "send_time_limit_exceeded", "transport_errors_sessions" };
+    public static final String[] stompSubProtocolMetrics = { "connect", "connected", "disconnect" };
+    private WebSocketMessageBrokerStats apiStompStats;
+    private WebSocketMessageBrokerStats agentStompStats;
+    private static final Logger LOG = LoggerFactory.getLogger(StompEventsMetricsSource.class);
+    private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
+
+    @Override
+    public void start() {
+        int interval = 60;
+        LOG.info("Starting stomp stat source.");
+        try {
+            executor.scheduleWithFixedDelay(new Runnable() {
+                @Override
+                public void run() {
+                    List<SingleMetric> events = getStompStatMetrics();
+                    if (!events.isEmpty()) {
+                        sink.publish(events);
+                        LOG.debug("********* Published stomp stat metrics to sink **********");
+                    }
+                }
+            }, interval, interval, TimeUnit.SECONDS);
+        } catch (Exception e) {
+            LOG.info("Failed to start stomp stat source", e);
+        }
+    }
+
+    protected List<SingleMetric> getStompStatMetrics() {
+        List<SingleMetric> metrics = new ArrayList<>();
+        populateStompStatMetrics(metrics, metricsTypes[0], apiStompStats);
+        populateStompStatMetrics(metrics, metricsTypes[1], agentStompStats);
+        return metrics;
+    }
+
+    private void populateStompStatMetrics(List<SingleMetric> metricsList, String metricsPrefix,
+            WebSocketMessageBrokerStats stompStats) {
+        parseStats(metricsList, metricsPrefix + ".websocket.", webSocketMetrics,
+                stompStats.getWebSocketSessionStatsInfo());
+        parseStats(metricsList, metricsPrefix + ".stomp_sub_protocol.", stompSubProtocolMetrics,
+                stompStats.getStompSubProtocolStatsInfo());
+        parseStats(metricsList, metricsPrefix + ".inbound_channel.", poolMetrics,
+                stompStats.getClientInboundExecutorStatsInfo());
+        parseStats(metricsList, metricsPrefix + ".outbound_channel.", poolMetrics,
+                stompStats.getClientOutboundExecutorStatsInfo());
+        parseStats(metricsList, metricsPrefix + ".sockJsScheduler.", poolMetrics,
+                stompStats.getSockJsTaskSchedulerStatsInfo());
+    }
+
+    private void parseStats(List<SingleMetric> metricsList, String metricsPrefix, String[] metricsNamesList,
+            String stats) {
+        if (stats.equals("null")) {
+            LOG.warn("stats for " + metricsPrefix + " is null");
+            return;
+        }
+        List<Long> statsArray = getValuesFromString(stats);
+
+        // sanity check to make sure we have the same number of metrics and values.
+        if (statsArray.size() != metricsNamesList.length) {
+            LOG.error("Number of metrics and stats do not match for " + metricsPrefix);
+            return;
+        }
+        long currentTimeMillis = System.currentTimeMillis();
+        for (int i = 0; i < metricsNamesList.length; i++) {
+            metricsList
+                    .add(new SingleMetric(metricsPrefix + metricsNamesList[i], statsArray.get(i), currentTimeMillis));
+        }
+    }
+
+    private List<Long> getValuesFromString(String stats) {
+        // '\\D+' matches one or more non-digit characters. After that it split the
+        // string and filter out empty strings. Then we convert the string to long.
+        return Arrays.stream(stats.split("\\D+"))
+                .filter(s -> !s.isEmpty())
+                .map(Long::parseLong)
+                .collect(Collectors.toList());
+    }
+
+    public void setApiStompStats(WebSocketMessageBrokerStats apiStompStats) {
+        this.apiStompStats = apiStompStats;
+    }
+
+    public void setAgentStompStats(WebSocketMessageBrokerStats agentStompStats) {
+        this.agentStompStats = agentStompStats;
+    }
+}