Permalink
Browse files

Split common functions of Queue and Aggregator into Processor module.

  • Loading branch information...
1 parent c085889 commit c79e1139edb822a95fcd6b8f3787f016551f5136 @nextmat nextmat committed Apr 17, 2012
Showing with 105 additions and 157 deletions.
  1. +0 −1 lib/librato/metrics.rb
  2. +22 −87 lib/librato/metrics/aggregator.rb
  3. +78 −0 lib/librato/metrics/processor.rb
  4. +5 −69 lib/librato/metrics/queue.rb
View
@@ -1,7 +1,6 @@
$:.unshift(File.dirname(__FILE__)) unless
$:.include?(File.dirname(__FILE__)) || $:.include?(File.expand_path(File.dirname(__FILE__)))
-require 'aggregate'
require 'base64'
require 'multi_json'
@@ -1,14 +1,18 @@
+require 'aggregate'
+require 'metrics/processor'
+
module Librato
module Metrics
class Aggregator
+ include Processor
+
attr_reader :source
def initialize(options={})
- # XXX: When isn't this needed?
- @aggregated ||= {}
- @client = options.delete(:client) || Librato::Metrics.client
- @source = options.delete(:source)
+ @aggregated = {}
+ @client = options[:client] || Librato::Metrics.client
+ @source = options[:source]
end
# Add a metric entry to the metric set:
@@ -22,6 +26,20 @@ def add(args)
@aggregated[k] << value
end
end
+
+ # Returns true if aggregate contains no measurements
+ #
+ # @return Boolean
+ def empty?
+ @aggregated.empty?
+ end
+
+ # Remove all queued metrics
+ #
+ def flush
+ @aggregated = {}
+ end
+ alias :clear :flush
def queued
gauges = []
@@ -40,94 +58,11 @@ def queued
end
req = { :gauges => gauges }
-
req[:source] = @source if @source
req
end
- # Remove all queued metrics
- #
- def flush
- @aggregated = {}
- end
- alias :clear :flush
-
- # Returns true if aggregate contains no measurements
- #
- # @return Boolean
- def empty?
- @aggregated.empty?
- end
-
- #### COPIED BELOW
-
- # The current Client instance this queue is using to authenticate
- # and connect to Librato Metrics. This will default to the primary
- # client used by the Librato::Metrics module unless it has been
- # set to something else.
- #
- # @return [Librato::Metrics::Client]
- def client
- @client ||= Librato::Metrics.client
- end
-
- # Persist currently queued metrics
- #
- # @return Boolean
- def submit
- raise NoMetricsQueued if self.empty?
- if persister.persist(self.client, self.queued)
- flush and return true
- end
- false
- end
-
- # Capture execution time for a block and queue
- # it as the value for a metric. Times are recorded
- # in milliseconds.
- #
- # Options are the same as for {#add}.
- #
- # @example Queue API request response time
- # queue.time :api_request_time do
- # # API request..
- # end
- #
- # @example Queue API request response time w/ source
- # queue.time :api_request_time, :source => 'app1' do
- # # API request..
- # end
- #
- # @param [Symbol|String] name Metric name
- # @param [Hash] options Metric options
- def time(name, options={})
- start = Time.now
- yield
- duration = (Time.now - start) * 1000.0 # milliseconds
- metric = {name => options.merge({:value => duration})}
- add metric
- end
-
- alias :benchmark :time
-
- # The object this MetricSet will use to persist
- #
- def persister
- @persister ||= create_persister
- end
-
- private
-
- def create_persister
- type = self.client.persistence.to_s.capitalize
- Librato::Metrics::Persistence.const_get(type).new
- end
-
- def epoch_time
- Time.now.to_i
- end
-
end
end
end
@@ -0,0 +1,78 @@
+module Librato
+ module Metrics
+
+ module Processor
+ MEASUREMENTS_PER_REQUEST = 500
+
+ attr_reader :per_request
+
+ # The current Client instance this queue is using to authenticate
+ # and connect to Librato Metrics. This will default to the primary
+ # client used by the Librato::Metrics module unless it has been
+ # set to something else.
+ #
+ # @return [Librato::Metrics::Client]
+ def client
+ @client ||= Librato::Metrics.client
+ end
+
+ # The object this MetricSet will use to persist
+ #
+ def persister
+ @persister ||= create_persister
+ end
+
+ # Persist currently queued metrics
+ #
+ # @return Boolean
+ def submit
+ raise NoMetricsQueued if self.queued.empty?
+ options = {:per_request => @per_request}
+ if persister.persist(self.client, self.queued, options)
+ flush and return true
+ end
+ false
+ end
+
+ # Capture execution time for a block and queue
+ # it as the value for a metric. Times are recorded
+ # in milliseconds.
+ #
+ # Options are the same as for {#add}.
+ #
+ # @example Queue API request response time
+ # queue.time :api_request_time do
+ # # API request..
+ # end
+ #
+ # @example Queue API request response time w/ source
+ # queue.time :api_request_time, :source => 'app1' do
+ # # API request..
+ # end
+ #
+ # @param [Symbol|String] name Metric name
+ # @param [Hash] options Metric options
+ def time(name, options={})
+ start = Time.now
+ yield
+ duration = (Time.now - start) * 1000.0 # milliseconds
+ metric = {name => options.merge({:value => duration})}
+ add metric
+ end
+ alias :benchmark :time
+
+ private
+
+ def create_persister
+ type = self.client.persistence.to_s.capitalize
+ Librato::Metrics::Persistence.const_get(type).new
+ end
+
+ def epoch_time
+ Time.now.to_i
+ end
+
+ end
+
+ end
+end
@@ -1,16 +1,18 @@
+require 'metrics/processor'
+
+
module Librato
module Metrics
class Queue
- MEASUREMENTS_PER_REQUEST = 500
+ include Processor
- attr_reader :per_request
attr_accessor :skip_measurement_times
def initialize(options={})
@queued = {}
+ @client = options[:client] || Librato::Metrics.client
@per_request = options[:per_request] || MEASUREMENTS_PER_REQUEST
@skip_measurement_times = options[:skip_measurement_times]
- @client = options[:client] || Librato::Metrics.client
end
# Add a metric entry to the metric set:
@@ -37,16 +39,6 @@ def add(args)
queued
end
- # The current Client instance this queue is using to authenticate
- # and connect to Librato Metrics. This will default to the primary
- # client used by the Librato::Metrics module unless it has been
- # set to something else.
- #
- # @return [Librato::Metrics::Client]
- def client
- @client ||= Librato::Metrics.client
- end
-
# Currently queued counters
#
# @return [Array]
@@ -69,12 +61,6 @@ def flush
alias :clear :flush
alias :flush_queued :flush
- # The object this MetricSet will use to persist
- #
- def persister
- @persister ||= create_persister
- end
-
# Currently queued gauges
#
# @return Array
@@ -97,56 +83,6 @@ def size
end
alias :length :size
- # Persist currently queued metrics
- #
- # @return Boolean
- def submit
- raise NoMetricsQueued if self.queued.empty?
- options = {:per_request => @per_request}
- if persister.persist(self.client, self.queued, options)
- flush and return true
- end
- false
- end
-
- # Capture execution time for a block and queue
- # it as the value for a metric. Times are recorded
- # in milliseconds.
- #
- # Options are the same as for {#add}.
- #
- # @example Queue API request response time
- # queue.time :api_request_time do
- # # API request..
- # end
- #
- # @example Queue API request response time w/ source
- # queue.time :api_request_time, :source => 'app1' do
- # # API request..
- # end
- #
- # @param [Symbol|String] name Metric name
- # @param [Hash] options Metric options
- def time(name, options={})
- start = Time.now
- yield
- duration = (Time.now - start) * 1000.0 # milliseconds
- metric = {name => options.merge({:value => duration})}
- add metric
- end
- alias :benchmark :time
-
- private
-
- def create_persister
- type = self.client.persistence.to_s.capitalize
- Librato::Metrics::Persistence.const_get(type).new
- end
-
- def epoch_time
- Time.now.to_i
- end
-
end
end
end

0 comments on commit c79e113

Please sign in to comment.