Updates stats API response to expose also 1m, 5m and 15m average batch metrics

Author: andsel

logstash-core/lib/logstash/api/commands/stats.rb

@@ -176,25 +176,52 @@ def refine_batch_metrics(stats)
             # current is a tuple of [event_count, byte_size] store the reference locally to avoid repeatedly
             # reading and retrieve unrelated values
             current_data_point = stats[:batch][:current]
-            {
+            # average return a FlowMetric which and we need to invoke getValue to obtain the map with metric details.
+            event_count_average_flow_metric = stats[:batch][:event_count][:average].value
+            event_count_average_lifetime = event_count_average_flow_metric["lifetime"] ? event_count_average_flow_metric["lifetime"].round : 0
+            byte_size_average_flow_metric = stats[:batch][:byte_size][:average].value
+            byte_size_average_lifetime = byte_size_average_flow_metric["lifetime"] ? byte_size_average_flow_metric["lifetime"].round : 0
+            result = {
               :event_count => {
                 # current_data_point is an instance of org.logstash.instrument.metrics.gauge.LazyDelegatingGauge so need to invoke getValue() to obtain the actual value
                 :current => current_data_point.value[0],
                 :average => {
-                  # average return a FlowMetric which and we need to invoke getValue to obtain the map with metric details.
-                  :lifetime => stats[:batch][:event_count][:average].value["lifetime"] ? stats[:batch][:event_count][:average].value["lifetime"].round : 0
+                  :lifetime => event_count_average_lifetime
                 }
               },
               :byte_size => {
                 :current => current_data_point.value[1],
                 :average => {
-                  :lifetime => stats[:batch][:byte_size][:average].value["lifetime"] ? stats[:batch][:byte_size][:average].value["lifetime"].round : 0
+                  :lifetime => byte_size_average_lifetime
                 }
               }
             }
+            # Enrich byte_size and event_count averages with the last 1, 5, 15 minutes averages if available
+            publish_average_count_flow_metric(event_count_average_flow_metric, result, :last_1_minute)
+            publish_average_count_flow_metric(event_count_average_flow_metric, result, :last_5_minutes)
+            publish_average_count_flow_metric(event_count_average_flow_metric, result, :last_15_minutes)
+            publish_average_size_flow_metric(byte_size_average_flow_metric, result, :last_1_minute)
+            publish_average_size_flow_metric(byte_size_average_flow_metric, result, :last_5_minutes)
+            publish_average_size_flow_metric(byte_size_average_flow_metric, result, :last_15_minutes)
+            result
           end
           private :refine_batch_metrics
 
+
+          def publish_average_count_flow_metric(average_flow_metric, result, time_window)
+            if average_flow_metric[time_window.to_s]
+              result[:event_count][:average][time_window] = average_flow_metric[time_window.to_s].round
+            end
+          end
+          private :publish_average_count_flow_metric
+
+          def publish_average_size_flow_metric(average_flow_metric, result, time_window)
+            if average_flow_metric[time_window.to_s]
+              result[:byte_size][:average][time_window] = average_flow_metric[time_window.to_s].round
+            end
+          end
+          private :publish_average_size_flow_metric
+
           def report(stats, extended_stats = nil, opts = {})
             ret = {
               :events => stats[:events],