Merge pull request #77 from zendesk/support_for_an_events_callback

Support for an events callback

README.md

@@ -48,6 +48,19 @@ semantics instead. In this mode, the model will be updated in the cache as well
 Kasket.setup(write_through: true)
 ```
 
+#### Events Callback
+
+You can configure a callable object to listen to events. This can be useful to emit metrics and observe Kasket's behaviour.
+
+```ruby
+Kasket.setup(events_callback: -> (event, ar_klass) do
+  MyMetrics.increase_some_counter("kasket.#{event}", tags: ["table:#{ar_klass.table_name}"])
+end)
+```
+
+The following events are emitted:
+* `"cache_hit"`, when Kasket has found some record's data in the cache, which can be returned.
+
 ## Configuring caching of your models
 
 You can configure Kasket for any ActiveRecord model, and subclasses will automatically inherit the caching

gemfiles/common.rb

@@ -6,7 +6,7 @@
 gem "byebug",      "~> 11.1"
 gem "pry-byebug",  "= 3.9.0"
 gem "bump",        "~> 0.10"
-gem "minitest",    "~> 5.1"
+gem "minitest",    "~> 5.16.2" # temporary evil until https://github.com/zendesk/kasket/pull/82
 gem "minitest-rg", "~> 5.2"
 gem "mocha",       "~> 1.13"
 gem "rake",        "~> 13"

lib/kasket.rb

@@ -14,21 +14,35 @@ module Kasket
   autoload :Visitor,                'kasket/visitor'
   autoload :SelectManagerMixin,     'kasket/select_manager_mixin'
   autoload :RelationMixin,          'kasket/relation_mixin'
+  autoload :Events,                 'kasket/events'
 
   CONFIGURATION = { # rubocop:disable Style/MutableConstant
     max_collection_size: 100,
     write_through: false,
-    default_expires_in: nil
+    default_expires_in: nil,
+    events_callback: nil,
   }
 
   module_function
 
+  # Configure Kasket.
+  #
+  # @param [Hash] options the configuration options for Kasket.
+  # @option options [Integer] :max_collection_size max size limit for a cacheable
+  #   collection of records.
+  # @option options [Boolean] :write_through
+  # @option options [Integer, nil] :default_expires_in the cache TTL.
+  # @option options [#call] :events_callback a callable object used to instrument
+  #   Kasket operations. It is invoked with two arguments: the name of the event,
+  #   as a String, and the Klass of the ActiveRecord model the event is about.
+  #
   def setup(options = {})
     return if ActiveRecord::Base.respond_to?(:has_kasket)
 
     CONFIGURATION[:max_collection_size] = options[:max_collection_size] if options[:max_collection_size]
     CONFIGURATION[:write_through]       = options[:write_through]       if options[:write_through]
     CONFIGURATION[:default_expires_in]  = options[:default_expires_in]  if options[:default_expires_in]
+    CONFIGURATION[:events_callback]     = options[:events_callback]     if options[:events_callback]
 
     ActiveRecord::Base.extend(Kasket::ConfigurationMixin)
     ActiveRecord::Relation.include(Kasket::RelationMixin)

lib/kasket/events.rb

@@ -0,0 +1,34 @@
+# frozen_string_literal: true
+
+module Kasket
+  # Interface to the internal instrumentation event.
+  module Events
+    class << self
+      # Invokes the configured events callback, if provided.
+      #
+      # The callback behaves like a listener, and receives the same arguments
+      # that are passed to this `report` method.
+      #
+      # @param [String] event the type of event being instrumented.
+      # @param [class] ar_klass the ActiveRecord::Base subclass that the event
+      #   refers to.
+      #
+      # @return [nil]
+      #
+      def report(event, ar_klass)
+        return unless fn
+
+        fn.call(event, ar_klass)
+        nil
+      end
+
+      private
+
+      def fn
+        return @fn if defined?(@fn)
+
+        @fn = Kasket::CONFIGURATION[:events_callback]
+      end
+    end
+  end
+end

lib/kasket/read_mixin.rb

@@ -36,8 +36,15 @@ def find_by_sql_with_kasket(sql, binds = [], *restargs, **kwargs, &blk)
             result_set = if value.is_a?(TrueClass)
               find_by_sql_without_kasket(sql, binds, *restargs, **kwargs, &blk)
             elsif value.is_a?(Array)
+              # The data from the Kasket cache is a list of keys to other Kasket entries.
+              # This usually happens when we're trying to load a collection association,
+              # e.g. a list of comments using their post_id in the query.
+              # Do not report a cache hit yet, and defer it until we've verified that at
+              # least one of the retrieved keys is actually in the cache.
               filter_pending_records(find_by_sql_with_kasket_on_id_array(value))
             else
+              # Direct cache hit for the key.
+              Events.report("cache_hit", self)
               filter_pending_records(Array.wrap(value).collect { |record| instantiate(record.dup, &blk) })
             end
 
@@ -60,6 +67,9 @@ def find_by_sql_with_kasket_on_id_array(keys, &blk)
       key_attributes_map = Kasket.cache.read_multi(*keys)
 
       found_keys, missing_keys = keys.partition {|k| key_attributes_map[k] }
+      # Only report a cache hit if at least some keys were found in the cache.
+      Events.report("cache_hit", self) if found_keys.any?
+
       found_keys.each {|k| key_attributes_map[k] = instantiate(key_attributes_map[k].dup, &blk) }
       key_attributes_map.merge!(missing_records_from_db(missing_keys))
 

lib/kasket/version.rb

@@ -1,6 +1,6 @@
 # frozen_string_literal: true
 module Kasket
-  VERSION = '4.13.0'
+  VERSION = '4.14.0'
   class Version
     MAJOR = Kasket::VERSION.split('.')[0]
     MINOR = Kasket::VERSION.split('.')[1]

test/events_test.rb

@@ -0,0 +1,52 @@
+# frozen_string_literal: true
+
+require_relative "helper"
+
+describe Kasket::Events do
+  before do
+    @previous = Kasket::CONFIGURATION[:events_callback]
+    Kasket::Events.remove_instance_variable(:@fn) if Kasket::Events.instance_variable_defined?(:@fn)
+  end
+
+  after { Kasket::CONFIGURATION[:events_callback] = @previous }
+
+  describe ".report" do
+    describe "when there is no stats callback configured" do
+      before do
+        Kasket::CONFIGURATION[:events_callback] = nil
+      end
+
+      it "does nothing and returns safely" do
+        assert_nil Kasket::Events.report("something", Author)
+      end
+    end
+
+    describe "when a stats callback is configured" do
+      before do
+        @event = nil
+        @ar_klass = nil
+
+        callback = proc do |event, ar_klass|
+          @event = event
+          @ar_klass = ar_klass
+        end
+
+        Kasket::CONFIGURATION[:events_callback] = callback
+      end
+
+      it "returns safely" do
+        assert_nil Kasket::Events.report("something", Author)
+      end
+
+      it "invokes the callback" do
+        assert_nil @event
+        assert_nil @ar_klass
+
+        Kasket::Events.report("something", Author)
+
+        assert_equal "something", @event
+        assert_equal Author, @ar_klass
+      end
+    end
+  end
+end

test/read_mixin_test.rb

@@ -233,6 +233,60 @@
         assert_not_equal @record, same_record
       end
     end
+
+    describe "emitting stats" do
+      before do
+        @previous = Kasket::CONFIGURATION[:events_callback]
+        Kasket::Events.remove_instance_variable(:@fn) if Kasket::Events.instance_variable_defined?(:@fn)
+
+        @emitted_events = []
+        callback = proc do |event, ar_klass|
+          @emitted_events << [event, ar_klass]
+        end
+
+        Kasket::CONFIGURATION[:events_callback] = callback
+      end
+
+      after { Kasket::CONFIGURATION[:events_callback] = @previous }
+
+      describe "event: cache_hit" do
+        it "is emitted when retrieving one record from the cache" do
+          Kasket.cache.write("#{Post.kasket_key_prefix}id=1", @post_database_result)
+
+          assert_empty @emitted_events
+          Post.find_by_sql('SELECT * FROM `posts` WHERE (id = 1)')
+          assert_equal 1, @emitted_events.length
+          assert_equal ["cache_hit", Post], @emitted_events[0]
+        end
+
+        it "is emitted when retrieving multiple records from the cache" do
+          # The Comment-by-post-id key points to the two Comment-by-id keys.
+          # Each Comment-by-id key contains its own record data.
+          Kasket.cache.write("#{Comment.kasket_key_prefix}post_id=1", [
+            "#{Comment.kasket_key_prefix}id=1",
+            "#{Comment.kasket_key_prefix}id=2"
+          ])
+          Kasket.cache.write("#{Comment.kasket_key_prefix}id=1", @comment_database_result[0])
+          Kasket.cache.write("#{Comment.kasket_key_prefix}id=2", @comment_database_result[1])
+
+          assert_empty @emitted_events
+          Comment.find_by_sql('SELECT * FROM `comments` WHERE (post_id = 1)')
+          assert_equal 1, @emitted_events.length
+          assert_equal ["cache_hit", Comment], @emitted_events[0]
+        end
+
+        it "is NOT emitted when nothing is in the cache the cache" do
+          assert_nil Kasket.cache.read("#{Post.kasket_key_prefix}id=1")
+
+          assert_empty @emitted_events
+          Post.find_by_sql('SELECT * FROM `posts` WHERE (id = 1)')
+          # Other events might have been emitted.
+          @emitted_events.each do |emitted_event|
+            refute_equal ["cache_hit", Post], emitted_event
+          end
+        end
+      end
+    end
   end
 
   it "support serialized attributes" do