Expand usage of MetricsContainer in other collectors (#268)

* Update activerecord collector
* Add extra AR collector test
* Add expiration test for resque collector
* Add process collector test file
* Adds the basic process collection tests (individually for the collector class)
* Fixes hostname collection (hostname was previously ignored, see #272 for details)
* Add unicorn collector test

Author: sosedoff

lib/prometheus_exporter/server/active_record_collector.rb

@@ -2,7 +2,8 @@
 
 module PrometheusExporter::Server
   class ActiveRecordCollector < TypeCollector
-    MAX_ACTIVERECORD_METRIC_AGE = 60
+    MAX_METRIC_AGE = 60
+
     ACTIVE_RECORD_GAUGES = {
       connections: "Total connections in pool",
       busy: "Connections in use in pool",
@@ -13,7 +14,12 @@ class ActiveRecordCollector < TypeCollector
     }
 
     def initialize
-      @active_record_metrics = []
+      @active_record_metrics = MetricsContainer.new(ttl: MAX_METRIC_AGE)
+      @active_record_metrics.filter = -> (new_metric, old_metric) do
+        new_metric["pid"] == old_metric["pid"] &&
+        new_metric["hostname"] == old_metric["hostname"] &&
+        new_metric["metric_labels"]["pool_name"] == old_metric["metric_labels"]["pool_name"]
+      end
     end
 
     def type
@@ -42,16 +48,6 @@ def metrics
     end
 
     def collect(obj)
-      now = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
-
-      obj["created_at"] = now
-
-      @active_record_metrics.delete_if do |current|
-        (obj["pid"] == current["pid"] && obj["hostname"] == current["hostname"] &&
-         obj["metric_labels"]["pool_name"] == current["metric_labels"]["pool_name"]) ||
-          (current["created_at"] + MAX_ACTIVERECORD_METRIC_AGE < now)
-      end
-
       @active_record_metrics << obj
     end
   end

lib/prometheus_exporter/server/process_collector.rb

@@ -3,7 +3,8 @@
 module PrometheusExporter::Server
 
   class ProcessCollector < TypeCollector
-    MAX_PROCESS_METRIC_AGE = 60
+    MAX_METRIC_AGE = 60
+
     PROCESS_GAUGES = {
       heap_free_slots: "Free ruby heap slots.",
       heap_live_slots: "Used ruby heap slots.",
@@ -21,7 +22,10 @@ class ProcessCollector < TypeCollector
     }
 
     def initialize
-      @process_metrics = []
+      @process_metrics = MetricsContainer.new(ttl: MAX_METRIC_AGE)
+      @process_metrics.filter = -> (new_metric, old_metric) do
+        new_metric["pid"] == old_metric["pid"] && new_metric["hostname"] == old_metric["hostname"]
+      end
     end
 
     def type
@@ -34,8 +38,8 @@ def metrics
       metrics = {}
 
       @process_metrics.map do |m|
-        metric_key = m["metric_labels"].merge("pid" => m["pid"])
-        metric_key.merge!(m["custom_labels"] || {})
+        metric_key = (m["metric_labels"] || {}).merge("pid" => m["pid"], "hostname" => m["hostname"])
+        metric_key.merge!(m["custom_labels"]) if m["custom_labels"]
 
         PROCESS_GAUGES.map do |k, help|
           k = k.to_s
@@ -58,15 +62,6 @@ def metrics
     end
 
     def collect(obj)
-      now = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
-
-      obj["created_at"] = now
-
-      @process_metrics.delete_if do |current|
-        (obj["pid"] == current["pid"] && obj["hostname"] == current["hostname"]) ||
-          (current["created_at"] + MAX_PROCESS_METRIC_AGE < now)
-      end
-
       @process_metrics << obj
     end
   end

lib/prometheus_exporter/server/puma_collector.rb

@@ -14,7 +14,7 @@ class PumaCollector < TypeCollector
     }
 
     def initialize
-      @puma_metrics = MetricsContainer.new
+      @puma_metrics = MetricsContainer.new(ttl: MAX_PUMA_METRIC_AGE)
       @puma_metrics.filter = -> (new_metric, old_metric) do
         new_metric["pid"] == old_metric["pid"] && new_metric["hostname"] == old_metric["hostname"]
       end

lib/prometheus_exporter/server/resque_collector.rb

@@ -2,7 +2,7 @@
 
 module PrometheusExporter::Server
   class ResqueCollector < TypeCollector
-    MAX_RESQUE_METRIC_AGE = 30
+    MAX_METRIC_AGE = 30
     RESQUE_GAUGES = {
       processed_jobs: "Total number of processed Resque jobs.",
       failed_jobs: "Total number of failed Resque jobs.",
@@ -13,7 +13,7 @@ class ResqueCollector < TypeCollector
     }
 
     def initialize
-      @resque_metrics = []
+      @resque_metrics = MetricsContainer.new(ttl: MAX_METRIC_AGE)
       @gauges = {}
     end
 
@@ -40,11 +40,7 @@ def metrics
     end
 
     def collect(object)
-      now = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
-
-      object["created_at"] = now
-      resque_metrics.delete_if { |metric| metric["created_at"] + MAX_RESQUE_METRIC_AGE < now }
-      resque_metrics << object
+      @resque_metrics << object
     end
 
     private

lib/prometheus_exporter/server/sidekiq_process_collector.rb

@@ -2,7 +2,7 @@
 
 module PrometheusExporter::Server
   class SidekiqProcessCollector < PrometheusExporter::Server::TypeCollector
-    MAX_SIDEKIQ_METRIC_AGE = 60
+    MAX_METRIC_AGE = 60
 
     SIDEKIQ_PROCESS_GAUGES = {
       'busy' => 'Number of running jobs',
@@ -12,7 +12,7 @@ class SidekiqProcessCollector < PrometheusExporter::Server::TypeCollector
     attr_reader :sidekiq_metrics, :gauges
 
     def initialize
-      @sidekiq_metrics = MetricsContainer.new(ttl: MAX_SIDEKIQ_METRIC_AGE)
+      @sidekiq_metrics = MetricsContainer.new(ttl: MAX_METRIC_AGE)
       @gauges = {}
     end
 

lib/prometheus_exporter/server/sidekiq_queue_collector.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 module PrometheusExporter::Server
   class SidekiqQueueCollector < TypeCollector
-    MAX_SIDEKIQ_METRIC_AGE = 60
+    MAX_METRIC_AGE = 60
 
     SIDEKIQ_QUEUE_GAUGES = {
       'backlog' => 'Size of the sidekiq queue.',
@@ -11,7 +11,7 @@ class SidekiqQueueCollector < TypeCollector
     attr_reader :sidekiq_metrics, :gauges
 
     def initialize
-      @sidekiq_metrics = MetricsContainer.new
+      @sidekiq_metrics = MetricsContainer.new(ttl: MAX_METRIC_AGE)
       @gauges = {}
     end
 

lib/prometheus_exporter/server/sidekiq_stats_collector.rb

@@ -2,7 +2,7 @@
 
 module PrometheusExporter::Server
   class SidekiqStatsCollector < TypeCollector
-    MAX_SIDEKIQ_METRIC_AGE = 60
+    MAX_METRIC_AGE = 60
 
     SIDEKIQ_STATS_GAUGES = {
       'dead_size' => 'Size of dead the queue',
@@ -18,7 +18,7 @@ class SidekiqStatsCollector < TypeCollector
     attr_reader :sidekiq_metrics, :gauges
 
     def initialize
-      @sidekiq_metrics = MetricsContainer.new(ttl: MAX_SIDEKIQ_METRIC_AGE)
+      @sidekiq_metrics = MetricsContainer.new(ttl: MAX_METRIC_AGE)
       @gauges = {}
     end
 

lib/prometheus_exporter/server/unicorn_collector.rb

@@ -2,47 +2,46 @@
 
 # custom type collector for prometheus_exporter for handling the metrics sent from
 # PrometheusExporter::Instrumentation::Unicorn
-class PrometheusExporter::Server::UnicornCollector < PrometheusExporter::Server::TypeCollector
-  MAX_UNICORN_METRIC_AGE = 60
-
-  UNICORN_GAUGES = {
-    workers: 'Number of unicorn workers.',
-    active_workers: 'Number of active unicorn workers',
-    request_backlog: 'Number of requests waiting to be processed by a unicorn worker.'
-  }.freeze
-
-  def initialize
-    @unicorn_metrics = []
-  end
+module PrometheusExporter::Server
+  class UnicornCollector < PrometheusExporter::Server::TypeCollector
+    MAX_METRIC_AGE = 60
+
+    UNICORN_GAUGES = {
+      workers: 'Number of unicorn workers.',
+      active_workers: 'Number of active unicorn workers',
+      request_backlog: 'Number of requests waiting to be processed by a unicorn worker.'
+    }.freeze
+
+    def initialize
+      @unicorn_metrics = MetricsContainer.new(ttl: MAX_METRIC_AGE)
+    end
 
-  def type
-    'unicorn'
-  end
+    def type
+      'unicorn'
+    end
 
-  def metrics
-    return [] if @unicorn_metrics.length.zero?
+    def metrics
+      return [] if @unicorn_metrics.length.zero?
 
-    metrics = {}
+      metrics = {}
 
-    @unicorn_metrics.map do |m|
-      labels = m["custom_labels"] || {}
+      @unicorn_metrics.map do |m|
+        labels = m["custom_labels"] || {}
 
-      UNICORN_GAUGES.map do |k, help|
-        k = k.to_s
-        if (v = m[k])
-          g = metrics[k] ||= PrometheusExporter::Metric::Gauge.new("unicorn_#{k}", help)
-          g.observe(v, labels)
+        UNICORN_GAUGES.map do |k, help|
+          k = k.to_s
+          if (v = m[k])
+            g = metrics[k] ||= PrometheusExporter::Metric::Gauge.new("unicorn_#{k}", help)
+            g.observe(v, labels)
+          end
         end
       end
-    end
 
-    metrics.values
-  end
+      metrics.values
+    end
 
-  def collect(obj)
-    now = ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
-    obj["created_at"] = now
-    @unicorn_metrics.delete_if { |m| m['created_at'] + MAX_UNICORN_METRIC_AGE < now }
-    @unicorn_metrics << obj
+    def collect(obj)
+      @unicorn_metrics << obj
+    end
   end
 end

test/server/active_record_collector_test.rb

@@ -6,6 +6,8 @@
 require 'prometheus_exporter/instrumentation'
 
 class PrometheusActiveRecordCollectorTest < Minitest::Test
+  include CollectorHelper
+
   def collector
     @collector ||= PrometheusExporter::Server::ActiveRecordCollector.new
   end
@@ -27,7 +29,6 @@ def test_collecting_metrics
   end
 
   def test_collecting_metrics_with_custom_labels
-
     collector.collect(
       "type" => "active_record",
       "pid" => "1000",
@@ -49,7 +50,6 @@ def test_collecting_metrics_with_custom_labels
   end
 
   def test_collecting_metrics_with_client_default_labels
-
     collector.collect(
       "type" => "active_record",
       "pid" => "1000",
@@ -108,4 +108,56 @@ def test_collecting_metrics_for_multiple_pools
     assert(metrics.first.metric_text.include?('active_record_connection_pool_connections{pool_name="primary",pid="1000",hostname="localhost"} 50'))
     assert(metrics.first.metric_text.include?('active_record_connection_pool_connections{pool_name="other",pid="1000",hostname="localhost"} 5'))
   end
+
+  def test_metrics_deduplication
+    data = {
+      "pid" => "1000",
+      "hostname" => "localhost",
+      "metric_labels" => { "pool_name" => "primary" },
+      "connections" => 100
+    }
+
+    collector.collect(data)
+    collector.collect(data.merge("connections" => 200))
+    collector.collect(data.merge("pid" => "2000", "connections" => 300))
+    collector.collect(data.merge("pid" => "3000", "connections" => 400))
+    collector.collect(data.merge("hostname" => "localhost2", "pid" => "2000", "connections" => 500))
+
+    metrics = collector.metrics
+    metrics_lines = metrics.map(&:metric_text).join.split("\n")
+
+    assert_equal 1, metrics.size
+    assert_equal [
+      'active_record_connection_pool_connections{pool_name="primary",pid="1000",hostname="localhost"} 200',
+      'active_record_connection_pool_connections{pool_name="primary",pid="2000",hostname="localhost"} 300',
+      'active_record_connection_pool_connections{pool_name="primary",pid="3000",hostname="localhost"} 400',
+      'active_record_connection_pool_connections{pool_name="primary",pid="2000",hostname="localhost2"} 500'
+    ], metrics_lines
+  end
+
+  def test_metrics_expiration
+    data = {
+      "pid" => "1000",
+      "hostname" => "localhost",
+      "connections" => 50,
+      "busy" => 20,
+      "dead" => 10,
+      "idle" => 20,
+      "waiting" => 0,
+      "size" => 120,
+      "metric_labels" => {
+        "pool_name" => "primary"
+      }
+    }
+
+    stub_monotonic_clock(0) do
+      collector.collect(data)
+      collector.collect(data.merge("pid" => "1001", "hostname" => "localhost2"))
+      assert_equal 6, collector.metrics.size
+    end
+
+    stub_monotonic_clock(max_metric_age + 1) do
+      assert_equal 0, collector.metrics.size
+    end
+  end
 end

test/server/collector_test.rb

@@ -727,7 +727,7 @@ def test_it_can_collect_process_metrics
 
     text = collector.prometheus_metrics_text
 
-    v8_str = "v8_heap_count{type=\"web\",pid=\"#{collected[:pid]}\"} #{collected[:v8_heap_count]}"
+    v8_str = "v8_heap_count{type=\"web\",pid=\"#{collected[:pid]}\",hostname=\"#{PrometheusExporter.hostname}\"} #{collected[:v8_heap_count]}"
 
     assert(text.include?(v8_str), "must include v8 metric")
     assert(text.include?("minor_gc_ops_total"), "must include counters")