Support for tagged measurements (#54)

CHANGELOG.md

@@ -1,3 +1,6 @@
+### master
+* Add support for tagged measurements (#54)
+
 ### Version 1.1.0
 * Fix deprecation warnings in ruby 2.4 (#57, Ben Radler)
 

README.md

@@ -49,7 +49,7 @@ If you don't have a Metrics account already, [sign up](https://metrics.librato.c
 
 By default you can use `LIBRATO_USER` and `LIBRATO_TOKEN` to pass your account data to the middleware. While these are the only required variables, there are a few more optional environment variables you may find useful.
 
-* `LIBRATO_SOURCE` - the default source to use for submitted metrics. If this is not set, hostname of the executing machine will be the default source
+* `LIBRATO_TAGS` - the default tags to use for submitted metrics. Format is comma-separated key=value pairs, e.g. `region=us-east,az=b`. If not set, `host` of the executing machine is detected and set as default tag
 * `LIBRATO_SUITES` - manage which metrics librato-rack will report. See more in [metrics suites](#metric-suites).
 * `LIBRATO_PREFIX` - a prefix which will be prepended to all metric names
 * `LIBRATO_LOG_LEVEL` - see logging section for more
@@ -74,19 +74,36 @@ See the [configuration class](https://github.com/librato/librato-rack/blob/maste
 
 If you are using the [Librato Metrics Heroku addon](https://addons.heroku.com/librato), your `LIBRATO_USER` and `LIBRATO_TOKEN` environment variables will already be set in your Heroku environment. If you are running without the addon you will need to provide them yourself.
 
-You must also specify a custom source for your app to track properly. If an explicit source is not set, `librato-rack` will not start. You can set the source in your environment:
+NOTE: if Heroku idles your application no measurements will be sent until it receives another request and is restarted. If you see intermittent gaps in your measurements during periods of low traffic this is the most likely cause.
 
-    heroku config:add LIBRATO_SOURCE=myappname
+## Default Tags
 
-NOTE: if Heroku idles your application no measurements will be sent until it receives another request and is restarted. If you see intermittent gaps in your measurements during periods of low traffic this is the most likely cause.
+Librato Metrics supports tagged measurements that are associated with a metric, one or more tag pairs, and a point in time. For more information on tagged measurements, visit our [API documentation](https://www.librato.com/docs/api/#measurements).
+
+##### Detected Tags
+
+By default, `host` is detected and applied as a default tag for submitted measurements. Optionally, you can override the detected values, e.g. `LIBRATO_TAGS=host=myapp-prod-1`
+
+##### Custom Tags
+
+In addition to the default tags, you can also provide custom tags:
+
+```ruby
+config = Librato::Rack::Configuration.new
+config.user = '[email protected]'
+config.token = 'mytoken'
+config.tags = { service: 'myapp', environment: 'production', host: 'myapp-prod-1' }
+
+use Librato::Rack, :config => config
+```
 
 ##### Metric Suites
 
 The metrics recorded by `librato-rack` are organized into named metric suites that can be selectively enabled/disabled:
 
 * `rack`: The `rack.request.total`, `rack.request.time`, `rack.request.slow`, and `rack.request.queue.time` metrics
-* `rack_status`: All of the `rack.request.status` metrics
-* `rack_method`: All of the `rack.request.method` metrics
+* `rack_status`: `rack.request.status` metric with `status` tag name and HTTP status code tag value, e.g. `status=200`
+* `rack_method`: `rack.request.method` metric with `method` tag name and HTTP method tag value, e.g. `method=POST`
 
 All three of the metric suites listed above are enabled by default.
 
@@ -116,34 +133,43 @@ Tracking anything that interests you is easy with Metrics. There are four primar
 
 Use for tracking a running total of something _across_ requests, examples:
 
-    # increment the 'sales_completed' metric by one
-    Librato.increment 'sales.completed'
+```ruby
+# increment the 'sales_completed' metric by one
+Librato.increment 'sales.completed'
+# => {:host=>"myapp-prod-1"}
 
-    # increment by five
-    Librato.increment 'items.purchased', :by => 5
+# increment by five
+Librato.increment 'items.purchased', by: 5
+# => {:host=>"myapp-prod-1"}
 
-    # increment with a custom source
-    Librato.increment 'user.purchases', :source => user.id
+# increment with custom per-measurement tags
+Librato.increment 'user.purchases', tags: { user_id: user.id, currency: 'USD' }
+# => {:user_id=>43, :currency=>"USD"}
+
+# increment with custom per-measurement tags and inherited default tags
+Librato.increment 'user.purchases', tags: { user_id: user.id, currency: 'USD' }, inherit_tags: true
+# => {:host=>"myapp-prod-1", :user_id=>43, :currency=>"USD"}
+```
 
 Other things you might track this way: user signups, requests of a certain type or to a certain route, total jobs queued or processed, emails sent or received
 
 ###### Sporadic Increment Reporting
 
 Note that `increment` is primarily used for tracking the rate of occurrence of some event. Given this increment metrics are _continuous by default_: after being called on a metric once they will report on every interval, reporting zeros for any interval when increment was not called on the metric.
 
-Especially with custom sources you may want the opposite behavior - reporting a measurement only during intervals where `increment` was called on the metric:
+Especially with custom per-measurement tags you may want the opposite behavior - reporting a measurement only during intervals where `increment` was called on the metric:
 
     # report a value for 'user.uploaded_file' only during non-zero intervals
-    Librato.increment 'user.uploaded_file', :source => user.id, :sporadic => true
+    Librato.increment 'user.uploaded_file', tags: { user: user.id, bucket: bucket.name }, sporadic: true
 
 #### measure
 
 Use when you want to track an average value _per_-request. Examples:
 
     Librato.measure 'user.social_graph.nodes', 212
 
-    # report from a custom source
-    Librato.measure 'jobs.queued', 3, :source => 'worker.12'
+    # report from custom per-measurement tags
+    Librato.measure 'jobs.queued', 3, tags: { priority: 'high', worker: 'worker.12' }
 
 #### timing
 
@@ -205,14 +231,11 @@ If you are using `librato-rack` with sidekiq, [see these notes about setup](http
 
 ## Cross-Process Aggregation
 
-`librato-rack` submits measurements back to the Librato platform on a _per-process_ basis. By default these measurements are then combined into a single measurement per source (default is your hostname) before persisting the data.
+`librato-rack` submits measurements back to the Librato platform on a _per-process_ basis. By default these measurements are then combined into a single measurement per default tags (detects `host`) before persisting the data.
 
 For example if you have 4 hosts with 8 unicorn instances each (i.e. 32 processes total), on the Metrics site you'll find 4 data streams (1 per host) instead of 32.
 Current pricing applies after aggregation, so in this case you will be charged for 4 streams instead of 32.
 
-If you want to report per-process instead, you can set `source_pids` to `true` in
-your config, which will append the process id to the source name used by each thread.
-
 ## Troubleshooting
 
 Note that it may take 2-3 minutes for the first results to show up in your Metrics account after you have started your servers with `librato-rack` enabled and the first request has been received.

lib/librato/collector.rb

@@ -10,14 +10,20 @@ class Collector
     def_delegators :counters, :increment
     def_delegators :aggregate, :measure, :timing
 
+    attr_reader :tags
+
+    def initialize(options={})
+      @tags = options[:tags]
+    end
+
     # access to internal aggregator object
     def aggregate
-      @aggregator_cache ||= Aggregator.new(prefix: @prefix)
+      @aggregator_cache ||= Aggregator.new(prefix: @prefix, default_tags: @tags)
     end
 
     # access to internal counters object
     def counters
-      @counter_cache ||= CounterCache.new
+      @counter_cache ||= CounterCache.new(default_tags: @tags)
     end
 
     # remove any accumulated but unsent metrics
@@ -48,4 +54,4 @@ def prefix
 require_relative 'collector/aggregator'
 require_relative 'collector/counter_cache'
 require_relative 'collector/exceptions'
-require_relative 'collector/group'
+require_relative 'collector/group'

lib/librato/collector/aggregator.rb

@@ -5,16 +5,19 @@ class Collector
     # maintains storage of timing and measurement type measurements
     #
     class Aggregator
-      SOURCE_SEPARATOR = '$$'
+      SEPARATOR = "$$"
 
       extend Forwardable
 
       def_delegators :@cache, :empty?, :prefix, :prefix=
 
+      attr_reader :default_tags
+
       def initialize(options={})
         @cache = Librato::Metrics::Aggregator.new(prefix: options[:prefix])
         @percentiles = {}
         @lock = Mutex.new
+        @default_tags = options.fetch(:default_tags, {})
       end
 
       def [](key)
@@ -26,13 +29,13 @@ def [](key)
       def fetch(key, options={})
         return nil if @cache.empty?
         return fetch_percentile(key, options) if options[:percentile]
-        gauges = nil
-        source = options[:source]
-        @lock.synchronize { gauges = @cache.queued[:gauges] }
-        gauges.each do |metric|
+        measurements = nil
+        tags = options[:tags] || @default_tags
+        @lock.synchronize { measurements = @cache.queued[:measurements] }
+        measurements.each do |metric|
           if metric[:name] == key.to_s
-            return metric if !source && !metric[:source]
-            return metric if source.to_s == metric[:source]
+            return metric if !tags && !metric[:tags]
+            return metric if tags == metric[:tags]
           end
         end
         nil
@@ -89,18 +92,27 @@ def measure(*args, &block)
         if args.length > 1 and args[-1].respond_to?(:each)
           options = args[-1]
         end
-        source = options[:source]
-        percentiles = Array(options[:percentile])
 
-        @lock.synchronize do
-          if source
-            @cache.add event => {source: source, value: value}
+        percentiles = Array(options[:percentile])
+        source = options[:source]
+        tags_option = options[:tags]
+        tags_option = { source: source } if source && !tags_option
+        tags =
+          if tags_option && options[:inherit_tags]
+            @default_tags.merge(tags_option)
+          elsif tags_option
+            tags_option
           else
-            @cache.add event => value
+            @default_tags
           end
 
+        @lock.synchronize do
+          payload = { value: value }
+          payload.merge!({ tags: tags }) if tags
+          @cache.add event => payload
+
           percentiles.each do |perc|
-            store = fetch_percentile_store(event, source)
+            store = fetch_percentile_store(event, payload)
             store[:reservoir] << value
             track_percentile(store, perc)
           end
@@ -120,30 +132,38 @@ def clear_storage
       end
 
       def fetch_percentile(key, options)
-        store = fetch_percentile_store(key, options[:source])
+        store = fetch_percentile_store(key, options)
         return nil unless store
         store[:reservoir].percentile(options[:percentile])
       end
 
-      def fetch_percentile_store(event, source)
-        keyname = source ? "#{event}#{SOURCE_SEPARATOR}#{source}" : event
+      def fetch_percentile_store(event, options)
+        keyname = event
+
+        if options[:tags]
+          keyname = Librato::Metrics::Util.build_key_for(keyname, options[:tags])
+        end
+
         @percentiles[keyname] ||= {
+          name: event,
           reservoir: Hetchy::Reservoir.new(size: 1000),
           percs: Set.new
         }
+        @percentiles[keyname].merge!({ tags: options[:tags] }) if options && options[:tags]
+        @percentiles[keyname]
       end
 
       def flush_percentiles(queue, opts)
         @percentiles.each do |key, val|
-          metric, source = key.split(SOURCE_SEPARATOR)
           val[:percs].each do |perc|
             perc_name = perc.to_s[0,5].gsub('.','')
-            payload = if source
-              { value: val[:reservoir].percentile(perc), source: source }
-            else
-              val[:reservoir].percentile(perc)
-            end
-            queue.add "#{metric}.p#{perc_name}" => payload
+            payload =
+              if val[:tags]
+                { value: val[:reservoir].percentile(perc), tags: val[:tags] }
+              else
+                val[:reservoir].percentile(perc)
+              end
+            queue.add "#{val[:name]}.p#{perc_name}" => payload
           end
         end
       end
@@ -158,4 +178,4 @@ def track_percentile(store, perc)
     end
 
   end
-end
+end