CloudWatch Embedded Metrics

Author: metaskills

lib/lambdakiq.rb

@@ -32,6 +32,10 @@ def client
     @client ||= Client.new
   end
 
+  def config
+    Lambdakiq::Railtie.config.lambdakiq
+  end
+
   extend self
 
 end

lib/lambdakiq/job.rb

@@ -22,7 +22,10 @@ def initialize(record)
     end
 
     def job_data
-      @job_data ||= JSON.parse(record.body)
+      @job_data ||= JSON.parse(record.body).tap do |data|
+        data['provider_job_id'] = record.message_id
+        data['executions'] = record.receive_count - 1
+      end
     end
 
     def active_job
@@ -34,16 +37,16 @@ def queue
     end
 
     def perform
-      if queue.fifo? && record.fifo_delay_seconds?
-        delay_fifo_message_visibility
-      else
-        ActiveJob::Base.execute(job_data)
-      end
+      fifo_delay? ? fifo_delay : execute
       delete_message
     rescue Exception => e
       perform_error(e)
     end
 
+    def execute
+      ActiveJob::Base.execute(job_data)
+    end
+
     private
 
     def client_params
@@ -79,7 +82,11 @@ def max_receive_count?
       record.max_receive_count? || record.receive_count >= queue.max_receive_count
     end
 
-    def delay_fifo_message_visibility
+    def fifo_delay?
+      queue.fifo? && record.fifo_delay_seconds?
+    end
+
+    def fifo_delay
       params = client_params.merge visibility_timeout: record.fifo_delay_visibility_timeout
       client.change_message_visibility(params)
     end

lib/lambdakiq/metrics.rb

@@ -1,38 +1,72 @@
 module Lambdakiq
   class Metrics
+    attr_reader :event
 
-    def initialize
-      @logger = ActiveJob::Base.logger
-      @namespace = Rails.application.class.name.split('::').first
-      @dimensions = Concurrent::Array.new
-      @metrics = Concurrent::Array.new
-      @properties = Concurrent::Hash.new
+    class << self
+      def log(event)
+        new(event).log
+      end
     end
 
-    def metrics
-      yield(self)
-    ensure
-      flush
+    def initialize(event)
+      @event = event
+      @metrics = []
+      @properties = {}
+      instrument!
     end
 
-    def flush
-      @logger.info(message) unless empty?
+    def log
+      logger.info JSON.dump(message)
     end
 
-    def benchmark
-      value = nil
-      seconds = Benchmark.realtime { value = yield }
-      milliseconds = (seconds * 1000).to_i
-      [value, milliseconds]
+    private
+
+    def job
+      event.payload[:job]
     end
 
-    def put_dimension(name, value)
-      @dimensions << { name => value }
-      self
+    def job_name
+      job.class.name
+    end
+
+    def logger
+      Lambdakiq.config.metrics_logger
+    end
+
+    def namespace
+      Lambdakiq.config.metrics_namespace
+    end
+
+    def exception
+      event.payload[:exception].try(:first)
+    end
+
+    def dimensions
+      [
+        { AppName: rails_app_name },
+        { JobEvent: event.name },
+        { JobName: job_name }
+      ]
+    end
+
+    def instrument!
+      put_metric 'Duration', event.duration.to_i, 'Milliseconds'
+      put_metric job_name, 1, 'Count'
+      put_metric 'Exceptions', 1, 'Count' if exception
+      set_property 'JobId', job.job_id
+      set_property 'JobName', job_name
+      set_property 'QueueName', job.queue_name
+      set_property 'MessageId', job.provider_job_id if job.provider_job_id
+      set_property 'Exception', exception if exception
+      set_property 'EnqueuedAt', job.enqueued_at if job.enqueued_at
+      set_property 'Executions', job.executions if job.executions
+      job.arguments.each_with_index do |argument, index|
+        set_property "JobArg#{index+1}", argument
+      end
     end
 
     def put_metric(name, value, unit = nil)
-      @metrics << { 'Name' => name }.tap do |m|
+      @metrics << { 'Name': name }.tap do |m|
         m['Unit'] = unit if unit
       end
       set_property name, value
@@ -43,24 +77,20 @@ def set_property(name, value)
       self
     end
 
-    def empty?
-      [@dimensions, @metrics, @properties].all?(&:empty?)
-    end
-
     def message
       {
-        '_aws' => {
-          'Timestamp' => timestamp,
-          'CloudWatchMetrics' => [
+        '_aws': {
+          'Timestamp': timestamp,
+          'CloudWatchMetrics': [
             {
-              'Namespace' => @namespace,
-              'Dimensions' => [@dimensions.map(&:keys).flatten],
-              'Metrics' => @metrics
+              'Namespace': namespace,
+              'Dimensions': [dimensions.map(&:keys).flatten],
+              'Metrics': @metrics
             }
           ]
         }
       }.tap do |m|
-        @dimensions.each { |dim| m.merge!(dim) }
+        dimensions.each { |d| m.merge!(d) }
         m.merge!(@properties)
       end
     end
@@ -69,5 +99,10 @@ def timestamp
       Time.current.strftime('%s%3N').to_i
     end
 
+    def rails_app_name
+      Rails.application.class.name.split('::').first
+    end
+
   end
 end
+

lib/lambdakiq/railtie.rb

@@ -2,5 +2,18 @@ module Lambdakiq
   class Railtie < ::Rails::Railtie
     config.lambdakiq = ActiveSupport::OrderedOptions.new
     config.lambdakiq.max_retries = 12
+    config.lambdakiq.metrics_namespace = 'Lambdakiq'
+
+    config.after_initialize do
+      config.active_job.logger = Rails.logger
+      config.lambdakiq.metrics_logger = Rails.logger
+    end
+
+    initializer "lambdakiq.metrics" do |app|
+      ActiveSupport::Notifications.subscribe(/active_job/) do |*args|
+        event = ActiveSupport::Notifications::Event.new *args
+        Lambdakiq::Metrics.log(event)
+      end
+    end
   end
 end

lib/lambdakiq/record.rb

@@ -11,6 +11,10 @@ def body
       data['body']
     end
 
+    def message_id
+      data['messageId']
+    end
+
     def receipt_handle
       data['receiptHandle']
     end
@@ -47,7 +51,7 @@ def receive_count
     end
 
     def max_receive_count?
-      receive_count >= 12
+      receive_count >= Lambdakiq.config.max_retries
     end
 
     def next_visibility_timeout

test/cases/job_test.rb

@@ -6,8 +6,21 @@ class JobTest < LambdakiqSpec
     expect(delete_message).must_be :present?
     expect(change_message_visibility).must_be_nil
     expect(perform_buffer_last_value).must_equal 'BasicJob with: "test"'
-    expect(active_job_log).must_include 'Performing TestHelper::Jobs::BasicJob'
-    expect(active_job_log).must_include 'Performed TestHelper::Jobs::BasicJob'
+    expect(logger).must_include 'Performing TestHelper::Jobs::BasicJob'
+    expect(logger).must_include 'Performed TestHelper::Jobs::BasicJob'
+  end
+
+  it 'logs cloudwatch embedded metrics' do
+    Lambdakiq::Job.handler(event_basic)
+    metric = logged_metric('perform.active_job')
+    expect(metric).must_be :present?
+    expect(metric['AppName']).must_equal 'Dummy'
+    expect(metric['JobName']).must_equal 'TestHelper::Jobs::BasicJob'
+    expect(metric['Duration']).must_equal 0
+    expect(metric['JobId']).must_equal '527cd37e-08f4-4aa8-9834-a46220cdc5a3'
+    expect(metric['QueueName']).must_equal 'lambdakiq-JobsQueue-TESTING123.fifo'
+    expect(metric['MessageId']).must_equal '9081fe74-bc79-451f-a03a-2fe5c6e2f807'
+    expect(metric['JobArg1']).must_equal 'test'
   end
 
   it 'must change message visibility to next value for failed jobs' do
@@ -16,8 +29,8 @@ class JobTest < LambdakiqSpec
     expect(change_message_visibility).must_be :present?
     expect(change_message_visibility_params[:visibility_timeout]).must_equal 1416
     expect(perform_buffer_last_value).must_equal 'ErrorJob with: "test"'
-    expect(active_job_log).must_include 'Performing TestHelper::Jobs::ErrorJob'
-    expect(active_job_log).must_include 'Error performing TestHelper::Jobs::ErrorJob'
+    expect(logger).must_include 'Performing TestHelper::Jobs::ErrorJob'
+    expect(logger).must_include 'Error performing TestHelper::Jobs::ErrorJob'
   end
 
   it 'wraps returned errors with no backtrace which avoids excessive/duplicate cloudwatch logging' do
@@ -26,17 +39,17 @@ class JobTest < LambdakiqSpec
     expect(error.class.name).must_equal 'Lambdakiq::JobError'
     expect(error.backtrace).must_equal []
     expect(perform_buffer_last_value).must_equal 'ErrorJob with: "test"'
-    expect(active_job_log).must_include 'Performing TestHelper::Jobs::ErrorJob'
-    expect(active_job_log).must_include 'Error performing TestHelper::Jobs::ErrorJob'
+    expect(logger).must_include 'Performing TestHelper::Jobs::ErrorJob'
+    expect(logger).must_include 'Error performing TestHelper::Jobs::ErrorJob'
   end
 
   it 'must delete message for failed jobs at the end of the queue/message max receive count' do
     event = event_basic attributes: { ApproximateReceiveCount: '8' }, job_class: 'TestHelper::Jobs::ErrorJob'
     Lambdakiq::Job.handler(event)
     expect(delete_message).must_be :present?
     expect(perform_buffer_last_value).must_equal 'ErrorJob with: "test"'
-    expect(active_job_log).must_include 'Performing TestHelper::Jobs::ErrorJob'
-    expect(active_job_log).must_include 'Error performing TestHelper::Jobs::ErrorJob'
+    expect(logger).must_include 'Performing TestHelper::Jobs::ErrorJob'
+    expect(logger).must_include 'Error performing TestHelper::Jobs::ErrorJob'
   end
 
   it 'must not perform and allow fifo queue to use message visibility as delay' do
@@ -45,7 +58,7 @@ class JobTest < LambdakiqSpec
     expect(change_message_visibility).must_be :present?
     expect(change_message_visibility_params[:visibility_timeout]).must_equal 6.minutes
     expect(perform_buffer_last_value).must_be_nil
-    expect(active_job_log).must_be :blank?
+    expect(logger).must_be :blank?
   end
 
   it 'must not perform and allow fifo queue to use message visibility as delay (using SentTimestamp)' do
@@ -54,7 +67,7 @@ class JobTest < LambdakiqSpec
     expect(change_message_visibility).must_be :present?
     expect(change_message_visibility_params[:visibility_timeout]).must_equal 8.minutes
     expect(perform_buffer_last_value).must_be_nil
-    expect(active_job_log).must_be :blank?
+    expect(logger).must_be :blank?
   end
 
   it 'must perform and allow fifo queue to use message visibility as delay but not when SentTimestamp is too far in the past' do
@@ -63,7 +76,7 @@ class JobTest < LambdakiqSpec
     expect(delete_message).must_be :present?
     expect(change_message_visibility).must_be_nil
     expect(perform_buffer_last_value).must_equal 'BasicJob with: "test"'
-    expect(active_job_log).must_include 'Performing TestHelper::Jobs::BasicJob'
-    expect(active_job_log).must_include 'Performed TestHelper::Jobs::BasicJob'
+    expect(logger).must_include 'Performing TestHelper::Jobs::BasicJob'
+    expect(logger).must_include 'Performed TestHelper::Jobs::BasicJob'
   end
 end

test/cases/record_test.rb

@@ -25,7 +25,7 @@ class RecordTest < LambdakiqSpec
 
   it '#sent_at' do
     sent_at = record.sent_at
-    expect(sent_at).must_be_instance_of Time
+    expect(sent_at).must_be_instance_of ActiveSupport::TimeWithZone
     expect(sent_at.year).must_equal 2020
     expect(sent_at.month).must_equal 11
     expect(sent_at.day).must_equal 30

test/dummy_app/config/.keep

@@ -0,0 +1,14 @@
+require 'rails/all'
+
+module Dummy
+  class Application < ::Rails::Application
+    config.root = File.join __FILE__, '..'
+    config.eager_load = true
+    logger = ActiveSupport::Logger.new(StringIO.new)
+    logger.formatter = ActiveSupport::Logger::SimpleFormatter.new
+    config.logger = logger
+    config.active_job.queue_adapter = :lambdakiq
+  end
+end
+
+Dummy::Application.initialize!

test/dummy_app/init.rb

@@ -10,9 +10,8 @@
 require 'minitest/focus'
 require 'mocha/minitest'
 Dir['test/test_helper/*.{rb}'].each { |f| require_relative "../#{f}" }
-
-ActiveJob::Base.queue_adapter = :lambdakiq
 Lambdakiq::Client.default_options.merge! stub_responses: true
+require_relative './dummy_app/init'
 
 class LambdakiqSpec < Minitest::Spec
 
@@ -26,7 +25,7 @@ class LambdakiqSpec < Minitest::Spec
   before do
     client_reset!
     client_stub_responses
-    reset_active_job_logger!
+    logger_reset!
     perform_buffer_clear!
   end
 

test/test_helper.rb

@@ -3,13 +3,23 @@ module LogHelpers
     extend ActiveSupport::Concern
 
     included do
-      let(:active_job_log)  { ActiveJob::Base.logger.instance_variable_get(:@logdev).instance_variable_get(:@dev).string }
+      let(:logger)  { Rails.logger.instance_variable_get(:@logdev).instance_variable_get(:@dev).string }
     end
 
     private
 
-    def reset_active_job_logger!
-      ActiveJob::Base.logger = Logger.new(StringIO.new)
+    def logged_metric(event)
+      metric = logged_metrics.reverse.detect { |l| l.include?(event) }
+      JSON.parse(metric) if metric
     end
+
+    def logged_metrics
+      logger.each_line.select { |l| l.include? 'CloudWatchMetrics' }
+    end
+
+    def logger_reset!
+      Rails.logger.instance_variable_get(:@logdev).instance_variable_get(:@dev).truncate 0
+    end
+
   end
 end