Add :most_recent aggregation to DirectFileStore

This reports the value that was set by a process most recently.

The way this works is by tagging each value in the files with the
timestamp of when they were set.

For all existing aggregations, we ignore that timestamp and do
what we've been doing so far.
For `:most_recent`, we take the "maximum" entry according to
its timestamp (i.e. the latest) and then return its value

Signed-off-by: Stefan Sundin <stefan@stefansundin.com>
Signed-off-by: Daniel Magliola <danielmagliola@gocardless.com>

Author: stefansundin

README.md

@@ -313,9 +313,9 @@ When instantiating metrics, there is an optional `store_settings` attribute. Thi
 to set up store-specific settings for each metric. For most stores, this is not used, but
 for multi-process stores, this is used to specify how to aggregate the values of each
 metric across multiple processes. For the most part, this is used for Gauges, to specify
-whether you want to report the `SUM`, `MAX` or `MIN` value observed across all processes.
-For almost all other cases, you'd leave the default (`SUM`). More on this on the 
-*Aggregation* section below.
+whether you want to report the `SUM`, `MAX`, `MIN`, or `MOST_RECENT` value observed across
+all processes. For almost all other cases, you'd leave the default (`SUM`). More on this
+on the *Aggregation* section below.
 
 Custom stores may also accept extra parameters besides `:aggregation`. See the
 documentation of each store for more details.
@@ -348,8 +348,8 @@ use case, you may need to control how this works. When using this store,
 each Metric allows you to specify an `:aggregation` setting, defining how
 to aggregate the multiple possible values we can get for each labelset. By default,
 Counters, Histograms and Summaries are `SUM`med, and Gauges report all their values (one
-for each process), tagged with a `pid` label. You can also select `SUM`, `MAX` or `MIN`
-for your gauges, depending on your use case.
+for each process), tagged with a `pid` label. You can also select `SUM`, `MAX`, `MIN`, or
+`MOST_RECENT` for your gauges, depending on your use case.
 
 **Memory Usage**: When scraped by Prometheus, this store will read all these files, get all
 the values and aggregate them. We have notice this can have a noticeable effect on memory

lib/prometheus/client/data_stores/direct_file_store.rb

@@ -29,7 +29,7 @@ module DataStores
 
       class DirectFileStore
         class InvalidStoreSettingsError < StandardError; end
-        AGGREGATION_MODES = [MAX = :max, MIN = :min, SUM = :sum, ALL = :all]
+        AGGREGATION_MODES = [MAX = :max, MIN = :min, SUM = :sum, ALL = :all, MOST_RECENT = :most_recent]
         DEFAULT_METRIC_SETTINGS = { aggregation: SUM }
         DEFAULT_GAUGE_SETTINGS = { aggregation: ALL }
 
@@ -121,15 +121,15 @@ def all_values
             stores_for_metric.each do |file_path|
               begin
                 store = FileMappedDict.new(file_path, true)
-                store.all_values.each do |(labelset_qs, v)|
+                store.all_values.each do |(labelset_qs, v, ts)|
                   # Labels come as a query string, and CGI::parse returns arrays for each key
                   # "foo=bar&x=y" => { "foo" => ["bar"], "x" => ["y"] }
                   # Turn the keys back into symbols, and remove the arrays
                   label_set = CGI::parse(labelset_qs).map do |k, vs|
                     [k.to_sym, vs.first]
                   end.to_h
 
-                  stores_data[label_set] << v
+                  stores_data[label_set] << [v, ts]
                 end
               ensure
                 store.close if store
@@ -181,30 +181,41 @@ def process_id
           end
 
           def aggregate_values(values)
-            if @values_aggregation_mode == SUM
-              values.inject { |sum, element| sum + element }
-            elsif @values_aggregation_mode == MAX
-              values.max
-            elsif @values_aggregation_mode == MIN
-              values.min
-            elsif @values_aggregation_mode == ALL
-              values.first
+            # Each entry in the `values` array is a tuple of `value` and `timestamp`,
+            # so for all aggregations except `MOST_RECENT`, we need to only take the
+            # first value in each entry and ignore the second.
+            if @values_aggregation_mode == MOST_RECENT
+              latest_tuple = values.max { |a,b| a[1] <=> b[1] }
+              latest_tuple.first # return the value without the timestamp
             else
-              raise InvalidStoreSettingsError,
-                    "Invalid Aggregation Mode: #{ @values_aggregation_mode }"
+              values = values.map(&:first) # Discard timestamps
+
+              if @values_aggregation_mode == SUM
+                values.inject { |sum, element| sum + element }
+              elsif @values_aggregation_mode == MAX
+                values.max
+              elsif @values_aggregation_mode == MIN
+                values.min
+              elsif @values_aggregation_mode == ALL
+                values.first
+              else
+                raise InvalidStoreSettingsError,
+                      "Invalid Aggregation Mode: #{ @values_aggregation_mode }"
+              end
             end
           end
         end
 
         private_constant :MetricStore
 
-        # A dict of doubles, backed by an file we access directly a a byte array.
+        # A dict of doubles, backed by an file we access directly as a byte array.
         #
         # The file starts with a 4 byte int, indicating how much of it is used.
         # Then 4 bytes of padding.
         # There's then a number of entries, consisting of a 4 byte int which is the
         # size of the next field, a utf-8 encoded string key, padding to an 8 byte
-        # alignment, and then a 8 byte float which is the value.
+        # alignment, and then a 8 byte float which is the value, and then a 8 byte
+        # float which is the unix timestamp when the value was set.
         class FileMappedDict
           INITIAL_FILE_SIZE = 1024*1024
 
@@ -236,7 +247,8 @@ def all_values
               @positions.map do |key, pos|
                 @f.seek(pos)
                 value = @f.read(8).unpack('d')[0]
-                [key, value]
+                timestamp = @f.read(8).unpack('d')[0]
+                [key, value, timestamp]
               end
             end
           end
@@ -258,7 +270,7 @@ def write_value(key, value)
 
             pos = @positions[key]
             @f.seek(pos)
-            @f.write([value].pack('d'))
+            @f.write([value, Time.now.to_f].pack('dd'))
             @f.flush
           end
 
@@ -299,7 +311,7 @@ def resize_file(new_capacity)
           def init_value(key)
             # Pad to be 8-byte aligned.
             padded = key + (' ' * (8 - (key.length + 4) % 8))
-            value = [padded.length, padded, 0.0].pack("lA#{padded.length}d")
+            value = [padded.length, padded, 0.0, 0.0].pack("lA#{padded.length}dd")
             while @used + value.length > @capacity
               @capacity *= 2
               resize_file(@capacity)
@@ -310,7 +322,7 @@ def init_value(key)
             @f.seek(0)
             @f.write([@used].pack('l'))
             @f.flush
-            @positions[key] = @used - 8
+            @positions[key] = @used - 16
           end
 
           # Read position of all keys. No locking is performed.
@@ -320,7 +332,7 @@ def populate_positions
               padded_len = @f.read(4).unpack('l')[0]
               key = @f.read(padded_len).unpack("A#{padded_len}")[0].strip
               @positions[key] = @f.pos
-              @f.seek(8, :CUR)
+              @f.seek(16, :CUR)
             end
           end
         end

spec/prometheus/client/data_stores/direct_file_store_spec.rb

@@ -267,6 +267,41 @@
     end
   end
 
+  context "with a metric that takes MOST_RECENT instead of SUM" do
+    it "reports the most recently written value from different processes" do
+      metric_store1 = subject.for_metric(
+        :metric_name,
+        metric_type: :gauge,
+        metric_settings: { aggregation: :most_recent }
+      )
+      metric_store2 = subject.for_metric(
+        :metric_name,
+        metric_type: :gauge,
+        metric_settings: { aggregation: :most_recent }
+      )
+
+      allow(Process).to receive(:pid).and_return(12345)
+      metric_store1.set(labels: { foo: "bar" }, val: 1)
+
+      allow(Process).to receive(:pid).and_return(23456)
+      metric_store2.set(labels: { foo: "bar" }, val: 3) # Supercedes 'bar' in PID 12345
+      metric_store2.set(labels: { foo: "baz" }, val: 2)
+      metric_store2.set(labels: { foo: "zzz" }, val: 1)
+
+      allow(Process).to receive(:pid).and_return(12345)
+      metric_store1.set(labels: { foo: "baz" }, val: 4) # Supercedes 'baz' in PID 23456
+
+      expect(metric_store1.all_values).to eq(
+        { foo: "bar" } => 3.0,
+        { foo: "baz" } => 4.0,
+        { foo: "zzz" } => 1.0,
+      )
+
+      # Both processes should return the same value
+      expect(metric_store1.all_values).to eq(metric_store2.all_values)
+    end
+  end
+
   it "resizes the File if metrics get too big" do
      truncate_calls_count = 0
      allow_any_instance_of(Prometheus::Client::DataStores::DirectFileStore::FileMappedDict).