Monitor by tag_columns (defaults with name)

README.md

@@ -391,14 +391,16 @@ QUEUE=tracking rake delayed:monitor
 QUEUES=mailers,tasks rake delayed:monitor
 ```
 
-The following events will be emitted, grouped by priority name (e.g. "interactive") and queue name,
-and the metric's "`:value`" will be available in the event's payload.  **This means that there will
-be one value _per_ unique combination of queue & priority**, and totals must be computed via
-downstream aggregation (e.g. as a StatsD "gauge" metric).
+The following events will be emitted, grouped by priority name (e.g. "interactive"), queue name,
+and the values of any configured `tag_columns`. By default, job `name` is included. The
+metric's "`:value`" will be available in the event's payload.  **This means that there will be one
+value _per_ unique combination of queue, priority, and tag column values**, and totals must be
+computed via downstream aggregation (e.g. as a StatsD "gauge" metric, summed or maxed by tag).
 
 - **delayed.job.count** - the total number of jobs
 - **delayed.job.future_count** - jobs where run_at is in the future
 - **delayed.job.working_count** - jobs that are currently being worked off (excludes failed jobs)
+- **delayed.job.locked_count** - jobs that are currently locked by a worker (equivalent to working_count)
 - **delayed.job.workable_count** - jobs that are waiting to be worked off
 - **delayed.job.erroring_count** - jobs where attempts > 0
 - **delayed.job.failed_count** - jobs where failed_at is not nil
@@ -409,8 +411,66 @@ An additional _experimental_ metric is available, intended for use with applicat
 
 - **delayed.job.alert_age_percent** - the _percent_ to which the oldest job has reached the "age alert" threshold. (See the [Alerting Threshholds](#priority-based-alerting-threshholds) section above.)
 
-All of these events may be subscribed to via a single regular expression (again, in your application
-config or in an initializer):
+By default, these events are also tagged with the job's `name` (when the jobs table has a `name`
+column — see [Database Setup](#database-setup)) so that downstream aggregation can answer
+"_which_ job is stuck?" when `delayed.job.max_age` alerts (e.g. `max by {queue, priority, name}`
+in Datadog).
+
+The set of tagged columns is driven by `Delayed::Monitor.tag_columns`, which defaults to
+`%i(name)` when the jobs table has a `name` column (and to `[]` otherwise). You can include
+columns your application adds to the jobs table (populated at enqueue time). For example, if your
+jobs table has an `owner` column you wish to also monitor:
+
+```ruby
+Delayed::Monitor.tag_columns = %i(name owner)
+```
+
+A few behavioral notes:
+
+- Rows whose value was never populated for a tagged column are reported under the value `'unset'`
+  (e.g. jobs enqueued before the `name` column existed, mid-upgrade).
+- Configured columns must exist on the jobs table: assigning a missing column to `tag_columns`
+  raises an `ArgumentError` immediately, rather than the column being silently skipped. Because
+  the assignment validates against the schema, setting `tag_columns` in an initializer requires a
+  database connection at boot, in every process that loads it. See the rollout steps below.
+- Tag values cannot be enumerated in advance, so a tagged series is only emitted while matching
+  jobs are present. Separately, an untagged zero value is always emitted for every
+  (priority, queue) combination, so that each metric maintains a baseline series even when no
+  matching jobs are enqueued. For example, `delayed.job.count` with a single enqueued job would
+  emit the following series:
+
+  ```ruby
+  { priority: 'interactive', queue: 'default', name: 'SimpleJob', value: 1 }
+  { priority: 'interactive', queue: 'default', value: 0 }
+  { priority: 'user_visible', queue: 'default', value: 0 }
+  { priority: 'eventual', queue: 'default', value: 0 }
+  { priority: 'reporting', queue: 'default', value: 0 }
+  ```
+- Each column multiplies a metric's series cardinality by its number of distinct values (though in
+  practice a job's `name` tends to determine its `priority` and any ownership tags, making the
+  number of distinct job names the effective upper bound). If cardinality is a concern for your
+  metrics provider, tagging can be disabled entirely with `Delayed::Monitor.tag_columns = []`.
+
+#### Rolling out a new tag column
+
+Because assignment fails loudly on a missing column, a new tag column should be rolled out in
+three separate deploys, each fully released before the next begins:
+
+1. Migrate the column onto the jobs table (nullable — no backfill required).
+2. Deploy the code that populates the column at enqueue time.
+3. Add the column to `Delayed::Monitor.tag_columns` in an initializer, and deploy.
+
+Adding the column to `tag_columns` before the migration has run everywhere would raise at boot in
+every process that loads the initializer. Jobs enqueued before step 2 will report under the
+`'unset'` tag value until they are worked off (or backfilled).
+
+The default `name` tag needs no such rollout: it applies only when the jobs table already has a
+`name` column, so a monitor running against an older schema simply emits untagged metrics until
+the generated migrations (see [Database Setup](#database-setup)) have run and the monitor process
+has restarted (the default is resolved once per process).
+
+All of these events may be subscribed to via a single regular expression (again, in your
+application config or in an initializer):
 
 ```ruby
 ActiveSupport::Notifications.subscribe(/delayed\.job\..*_(count|age|percent)/) do |*args|
@@ -423,7 +483,7 @@ ActiveSupport::Notifications.subscribe(/delayed\.job\..*_(count|age|percent)/) d
 end
 ```
 
-Additionally, the monitor process with emit a **delayed.monitor.run** event with a duration
+Additionally, the monitor process will emit a **delayed.monitor.run** event with a duration
 attached, so that you can monitor the time it takes to emit these aggregate metrics.
 
 ```ruby

lib/delayed/monitor.rb

@@ -17,6 +17,19 @@ class Monitor
 
     cattr_accessor :sleep_delay, instance_writer: false, default: 60
 
+    def self.tag_columns
+      @tag_columns ||= (Job.column_names.include?('name') ? %i(name) : []).freeze
+    end
+
+    def self.tag_columns=(columns)
+      if columns.any? { |column| Job.column_names.exclude?(column.to_s) }
+        raise ArgumentError, "Delayed::Monitor.tag_columns includes columns missing from #{Job.table_name}. " \
+                             "Available columns: #{Job.column_names.join(', ')}"
+      end
+
+      @tag_columns = columns.map(&:to_sym).freeze
+    end
+
     def initialize
       @jobs = Job.group(:priority, :queue)
       @jobs = @jobs.where(queue: Worker.queues) if Worker.queues.any?
@@ -67,10 +80,13 @@ def self.parse_utc_time(string)
     attr_reader :jobs
 
     def emit_metric!(metric)
-      query_for(metric).reverse_merge(default_results).each do |(priority, queue), value|
+      query_for(metric)
+        .merge!(default_results) { |_key, existing, _default| existing }
+        .each do |(priority, queue, *column_values), value|
+        tags = column_values.zip(self.class.tag_columns).to_h { |val, column| [column, val.nil? ? 'unset' : val] }
         ActiveSupport::Notifications.instrument(
           "delayed.job.#{metric}",
-          default_tags.merge(priority: Priority.new(priority).to_s, queue: queue, value: value),
+          default_tags.merge(priority: Priority.new(priority).to_s, queue: queue, **tags, value: value),
         )
       end
     end
@@ -96,24 +112,30 @@ def default_tags
     end
 
     # This method generates a query that scans the specified scope, groups by
-    # priority and queue, and calculates the specified aggregates. An outer
-    # query is executed for priority bucketing and appending db_now_utc (to
-    # avoid running these computations for each tuple in the inner query).
-    def grouped_query(scope, include_db_time: false, **kwargs)
+    # priority and queue (plus any extra_group_columns), and calculates the
+    # specified aggregates. An outer query is executed for priority bucketing
+    # and appending db_now_utc (to avoid running these computations for each
+    # tuple in the inner query).
+    def grouped_query(scope, include_db_time: false, extra_group_columns: [], **kwargs)
       inner_selects = kwargs.map { |key, (agg, expr)| as_expression(agg, expr, key) }
       outer_selects = kwargs.map { |key, (agg, _)| as_expression(agg == :count ? :sum : agg, key, key) }
       outer_selects << "#{self.class.sql_now_in_utc} AS db_now_utc" if include_db_time
 
       Delayed::Job
-        .from(scope.select(:priority, :queue, *inner_selects).group(:priority, :queue))
-        .group(priority_case_statement, :queue).select(
+        .from(scope.select(:priority, :queue, *extra_group_columns, *inner_selects).group(:priority, :queue, *extra_group_columns))
+        .group(priority_case_statement, :queue, *extra_group_columns).select(
           *outer_selects,
           "#{priority_case_statement} AS priority",
           'queue AS queue',
-        ).group_by { |j| [j.priority.to_i, j.queue] }
+          *extra_group_columns.map { |column| "#{column} AS #{column}" },
+        ).group_by { |j| result_key(j, extra_group_columns) }
         .transform_values(&:first)
     end
 
+    def result_key(record, extra_group_columns)
+      [record.priority.to_i, record.queue, *extra_group_columns.map { |column| record[column] }]
+    end
+
     def as_expression(aggregate_function, aggregate_expression, column_name)
       "#{aggregate_function.to_s.upcase}(#{aggregate_expression}) AS #{column_name}"
     end
@@ -151,10 +173,10 @@ def max_age_grouped
     end
 
     def alert_age_percent_grouped
-      pending_counts.each_with_object({}) do |((priority, queue), j), metrics|
+      pending_counts.each_with_object({}) do |(key, j), metrics|
         max_age = time_ago(db_now(j), j.run_at)
-        alert_age = Priority.new(priority).alert_age
-        metrics[[priority, queue]] = [max_age / alert_age * 100, 100].min if alert_age
+        alert_age = Priority.new(key.first).alert_age
+        metrics[key] = [max_age / alert_age * 100, 100].min if alert_age
       end
     end
 
@@ -175,6 +197,7 @@ def oldest_workable_job_grouped
     def live_counts
       @memo[:live_counts] ||= grouped_query(
         jobs.live,
+        extra_group_columns: self.class.tag_columns,
         count: [:count, '*'],
         future_count: [:sum, case_when(Job.future_clause.to_sql)],
         erroring_count: [:sum, case_when(Job.erroring_clause.to_sql)],
@@ -185,6 +208,7 @@ def pending_counts
       @memo[:pending_counts] ||= grouped_query(
         jobs.pending,
         include_db_time: true,
+        extra_group_columns: self.class.tag_columns,
         claimed_count: [:sum, case_when(Job.claimed_clause.to_sql)],
         claimable_count: [:sum, case_when(Job.claimable_clause.to_sql)],
         locked_at: [:min, case_when(Job.claimed_clause.to_sql, 'locked_at')],
@@ -193,7 +217,8 @@ def pending_counts
     end
 
     def failed_counts
-      @memo[:failed_counts] ||= grouped_query(jobs.failed, count: [:count, '*'])
+      @memo[:failed_counts] ||=
+        grouped_query(jobs.failed, extra_group_columns: self.class.tag_columns, count: [:count, '*'])
     end
 
     def db_now(record)