diff --git a/app/jobs/stats_collector_job.rb b/app/jobs/stats_collector_job.rb
index 1d18258f268..b75f68c7993 100644
--- a/app/jobs/stats_collector_job.rb
+++ b/app/jobs/stats_collector_job.rb
@@ -47,7 +47,8 @@ def perform
   def run_collectors(stats_collectors)
     stats_collectors.each do |collector_name, collector|
       start_time = Time.zone.now
-      collector.new.collect_stats&.each { |metric_name, value| emit(metric_name, value) }
+
+      collector.new.collect_stats&.each { |obj| emit_or_fail(obj) }
     rescue StandardError => error
       log_error(collector_name, error)
     ensure
@@ -55,13 +56,38 @@ def run_collectors(stats_collectors)
     end
   end
 
-  def emit(name, value, attrs: {})
+  def emit_or_fail(hash)
+    # when hash is { metric: metric_name", value: 123, "some_tag": "type1" }
+    return emit_tagged_hash(hash) if tagged_hash?(hash)
+
+    # when hash is { "metric_name" => 123 }
+    return emit_untagged_hash(hash) if hash.size == 1
+
+    fail "Unexpect metric object: #{hash}"
+  end
+
+  # :reek:FeatureEnvy
+  def tagged_hash?(hash)
+    hash[:metric] && hash[:value]
+  end
+
+  # :reek:FeatureEnvy
+  def emit_tagged_hash(hash)
+    emit(hash[:metric], hash[:value], tags: hash.except(:metric, :value))
+  end
+
+  # :reek:FeatureEnvy
+  def emit_untagged_hash(hash)
+    emit(hash.first[0], hash.first[1])
+  end
+
+  def emit(name, value, tags: {})
     DataDogService.emit_gauge(
       metric_group: METRIC_GROUP_NAME,
       metric_name: name,
       metric_value: value,
       app_name: APP_NAME,
-      attrs: attrs
+      attrs: tags
     )
   end
 
diff --git a/app/services/collectors/daily_counts_stats_collector.rb b/app/services/collectors/daily_counts_stats_collector.rb
index 1dcdd966c63..b87aeb92cf6 100644
--- a/app/services/collectors/daily_counts_stats_collector.rb
+++ b/app/services/collectors/daily_counts_stats_collector.rb
@@ -3,102 +3,83 @@
 # Collect daily stats for basic objects. Each query should complete within 10 seconds.
 # This collector is used by StatsCollectorJob.
 class Collectors::DailyCountsStatsCollector
+  include Collectors::StatsCollector
+
   TOTALS_METRIC_NAME_PREFIX = "daily_counts.totals"
   INCREMENTS_METRIC_NAME_PREFIX = "daily_counts.increments"
 
   def collect_stats
-    {}.tap do |stats|
-      stats.merge! flatten_stats total_counts_hash(TOTALS_METRIC_NAME_PREFIX)
-      stats.merge! flatten_stats daily_counts(INCREMENTS_METRIC_NAME_PREFIX)
+    [].tap do |stats|
+      stats.concat flatten_stats(TOTALS_METRIC_NAME_PREFIX, total_counts_hash)
+      stats.concat flatten_stats(INCREMENTS_METRIC_NAME_PREFIX, daily_counts)
     end
   end
 
   private
 
-  def flatten_stats(stats_hash)
-    {}.tap do |stats|
-      stats_hash.each do |key_namespace, counts_hash|
-        stats.merge! prepend_namespace_to_keys(key_namespace, counts_hash)
-      end
-    end
-  end
-
-  def prepend_namespace_to_keys(namespace, hash)
-    hash.transform_keys { |key| "#{namespace}.#{key_to_name(key)}" }
-  end
-
-  def key_to_name(key)
-    return "nil" unless key
-
-    return key.map(&:parameterize).map(&:underscore).join(".") if key.instance_of?(Array)
-
-    return key.parameterize.underscore if key.instance_of?(String)
-
-    key.to_s.parameterize.underscore
-  end
-
-  def total_counts_hash(prefix)
+  def total_counts_hash
     {
-      "#{prefix}.claimant.decision_review" => Claimant.group(:decision_review_type).count,
+      "claimant.decision_review" => Claimant.group(:decision_review_type).count,
       # => {"HigherLevelReview"=>75769, "SupplementalClaim"=>336809, "Appeal"=>48793}
-      "#{prefix}.claimant.payee_code" => Claimant.group(:payee_code).count,
+      "claimant.payee_code" => Claimant.group(:payee_code).count,
       # => {nil=>430595, "13"=>45, "17"=>1, "11"=>669, "00"=>14258, "60"=>75, "29"=>1, "10"=>15599, "12"=>89, ...
 
-      "#{prefix}.certification.office" => Certification.group(:certifying_office).count,
+      "certification.office" => Certification.group(:certifying_office).count,
       # => {nil=>39579, "Nashville, TN"=>2657, "Little Rock, AR"=>1796, "Indianapolis, IN"=>2607, ...
-      "#{prefix}.certification.rep" => Certification.group(:vacols_representative_type).count,
+      "certification.rep" => Certification.group(:vacols_representative_type).count,
       # => {nil=>55738, "Service Organization"=>77601, "Agent"=>3246, "None"=>8588, "Attorney"=>36092, ...
 
-      "#{prefix}.hearing" => Hearing.group(:disposition).count,
+      "hearing.disposition" => Hearing.group(:disposition).count,
       # => {nil=>432, "postponed"=>577, "no_show"=>125, "held"=>1253, "cancelled"=>219}
 
-      "#{prefix}.hearing_virtual" => VirtualHearing.group(:hearing_type).count,
+      "hearing_virtual" => VirtualHearing.group(:hearing_type).count,
       # => {"LegacyHearing"=>9}
 
-      "#{prefix}.distribution" => Distribution.all.group(:status).count,
+      "distribution" => Distribution.all.group(:status).count,
       # => {"completed"=>5090, "error"=>1431}
 
-      "#{prefix}.case_review" => {
+      "case_review" => {
         "judge" => JudgeCaseReview.count,
         "attorney" => AttorneyCaseReview.count
       },
 
-      "#{prefix}.decision_doc" => DecisionDocument.group(:appeal_type).count,
+      "decision_doc" => DecisionDocument.group(:appeal_type).count,
       # => {"Appeal"=>8047, "LegacyAppeal"=>63192}
 
-      "#{prefix}.claim_establishment" => ClaimEstablishment.group(:decision_type).count,
+      "claim_establishment" => ClaimEstablishment.group(:decision_type).count,
       # => {nil=>40, "full_grant"=>46292, "partial_grant"=>29677, "remand"=>111624}
 
-      "#{prefix}.req_issues" => RequestIssue.group(:decision_review_type).count,
+      "req_issue" => RequestIssue.group(:decision_review_type).count,
       # => {nil=>78, "HigherLevelReview"=>172753, "Appeal"=>125898, "SupplementalClaim"=>717169}
-      "#{prefix}.decision_issues" => DecisionIssue.group(:benefit_type, :disposition).count,
+      "decision_issue" => DecisionIssue.group(:benefit_type, :disposition).count,
       # => {["fiduciary", "remanded"]=>2, ["pension", "dismissed_matter_of_law"]=>31, ["vha", "Granted"]=>70, ...
 
-      "#{prefix}.dispatch" => Dispatch::Task.group(:type, :aasm_state).count
+      "dispatch" => Dispatch::Task.group(:type, :aasm_state).count
       # => {["EstablishClaim", "started"]=>5, ["EstablishClaim", "unprepared"]=>64, ...
     }
   end
 
-  def daily_counts(prefix)
+  def daily_counts
     {
-      "#{prefix}.appeal" => query_yesterdays(Appeal).group(:docket_type).count,
-      # => {"evidence_submission"=>52, "hearing"=>269, "direct_review"=>36}
-
-      prefix => {
+      "counts" => {
+        "appeal" => query_yesterdays(Appeal).count,
         "legacy_appeal" => query_yesterdays(LegacyAppeal).count,
         "appeal_series" => query_yesterdays(AppealSeries).count,
         "hearing" => query_yesterdays(Hearing).count,
         "legacy_hearing" => query_yesterdays(LegacyHearing).count
       },
 
-      "#{prefix}.appeal.status" => count_groups(query_yesterdays(Appeal)) { |record| record.status.status },
+      "appeal" => query_yesterdays(Appeal).group(:docket_type).count,
+      # => {"evidence_submission"=>52, "hearing"=>269, "direct_review"=>36}
+
+      "appeal.status" => count_groups(query_yesterdays(Appeal)) { |record| record.status.status },
       # => {:not_distributed=>349, :distributed_to_judge=>4, :cancelled=>2, :assigned_to_attorney=>2}
 
-      "#{prefix}.appeal_series.num_of_appeals" =>
+      "appeal_series.num_of_appeals" =>
           count_groups(query_yesterdays(AppealSeries)) { |record| record.appeals.count },
       # => {1=>555, 2=>56, 4=>12, 5=>3, 3=>20, 6=>1, 7=>2}
 
-      "#{prefix}.distribution" => query_yesterdays(Distribution).group(:status).count
+      "distribution" => query_yesterdays(Distribution).group(:status).count
       # => {"completed"=>13, "error"=>3}
     }
   end
diff --git a/app/services/collectors/request_issues_stats_collector.rb b/app/services/collectors/request_issues_stats_collector.rb
index f6d83aede98..f20f44f9e7c 100644
--- a/app/services/collectors/request_issues_stats_collector.rb
+++ b/app/services/collectors/request_issues_stats_collector.rb
@@ -5,40 +5,71 @@
 # The stats are typically used for monthly reporting.
 # This collector is used by StatsCollectorJob.
 class Collectors::RequestIssuesStatsCollector
-  METRIC_NAME_PREFIX = "request_issues.unidentified_with_contention"
+  include Collectors::StatsCollector
+
+  METRIC_NAME_PREFIX = "req_issues.w_contentions"
 
   # :reek:FeatureEnvy
   def collect_stats
-    start_of_month = Time.zone.now.prev_month.beginning_of_month
-    req_issues = unidentified_request_issues_with_contention(start_of_month, start_of_month.next_month)
+    [].tap do |stats|
+      group_count_hash = {
+        "status" => request_issues_with_contention.group(:closed_status).count,
+        "benefit" => request_issues_with_contention.group(:benefit_type).count,
+        "decis_review" => request_issues_with_contention.group(:decision_review_type).count,
 
-    {}.tap do |stats|
-      stats[METRIC_NAME_PREFIX] = req_issues.count
+        "report" => monthly_report_hash
+      }
+      stats.concat flatten_stats(METRIC_NAME_PREFIX, group_count_hash)
+    end
+  end
 
-      req_issues.group(:closed_status).count.each do |status, count|
-        stats["#{METRIC_NAME_PREFIX}.closed_status.#{status || 'nil'}"] = count
-      end
+  private
 
-      req_issues.group(:benefit_type).count.each do |ben_type, count|
-        stats["#{METRIC_NAME_PREFIX}.benefit.#{ben_type}"] = count
-      end
+  def monthly_report_hash
+    {
+      "hlr_established" => HigherLevelReview
+        .where("establishment_processed_at >= ?", start_date)
+        .where("establishment_processed_at < ?", end_date).count,
+      "sc_established" => SupplementalClaim
+        .where("establishment_processed_at >= ?", start_date)
+        .where("establishment_processed_at < ?", end_date).count,
 
-      dr_counts_by_type = req_issues.group(:decision_review_type).count
-      dr_counts_by_type.each do |dr_type, count|
-        stats["#{METRIC_NAME_PREFIX}.decision_review.#{dr_type}"] = count
-      end
+      "030_end_products_established" => endproduct_establishment.where(source_type: "HigherLevelReview").count,
+      "040_end_products_established" => endproduct_establishment.where(source_type: "SupplementalClaim").count,
+
+      "created" => request_issues_with_contention.count,
+      "edited" => edited_request_issues_with_contention.count,
+      "unidentified_created" => request_issues_with_contention.where(is_unidentified: true).count,
 
       # Could use `req_issues.group(:veteran_participant_id).count.count` but there's a count discrepancy
-      stats["#{METRIC_NAME_PREFIX}.vet_count"] = req_issues.map(&:decision_review).map(&:veteran_file_number).uniq.count
-    end
+      "vet_count" => request_issues_with_contention.map(&:decision_review).map(&:veteran_file_number).uniq.count
+    }
   end
 
-  private
+  def start_date
+    @start_date ||= Time.zone.now.prev_month.beginning_of_month
+  end
+
+  def end_date
+    @end_date ||= start_date.next_month
+  end
 
-  def unidentified_request_issues_with_contention(start_date, end_date)
-    RequestIssue.where.not(contention_reference_id: nil)
-      .where(is_unidentified: true)
+  def endproduct_establishment
+    EndProductEstablishment.where.not(reference_id: nil)
+      .where("committed_at >= ?", start_date)
+      .where("committed_at < ?", end_date)
+  end
+
+  def request_issues_with_contention
+    @request_issues_with_contention ||= RequestIssue.where.not(contention_reference_id: nil)
       .where("created_at >= ?", start_date)
       .where("created_at < ?", end_date)
   end
+
+  def edited_request_issues_with_contention
+    @edited_request_issues_with_contention ||= RequestIssue.where.not(contention_reference_id: nil)
+      .where.not(contention_updated_at: nil)
+      .where("contention_updated_at >= ?", start_date)
+      .where("contention_updated_at < ?", end_date)
+  end
 end
diff --git a/app/services/collectors/stats_collector.rb b/app/services/collectors/stats_collector.rb
new file mode 100644
index 00000000000..3400eae37de
--- /dev/null
+++ b/app/services/collectors/stats_collector.rb
@@ -0,0 +1,62 @@
+# frozen_string_literal: true
+
+module Collectors::StatsCollector
+  # Given metric_name_prefix and a hash like { "hearing.disposition" => Hearing.group(:disposition).count },
+  # the result will be an array of metrics like:
+  # [ { :metric => "#{metric_name_prefix}.hearing.disposition", :value => 300, "disposition" => "held" },
+  #   { :metric => "#{metric_name_prefix}.hearing.disposition", :value => 100, "disposition" => "postponed" }
+  # ]
+  def flatten_stats(metric_name_prefix, stats_hash)
+    [].tap do |stats|
+      stats_hash.each do |metric_name, counts_hash|
+        unless valid_metric_name?(metric_name)
+          fail "Invalid metric name #{metric_name}; "\
+            "see https://docs.datadoghq.com/developers/metrics/#naming-custom-metrics"
+        end
+
+        stats.concat add_tags_to_group_counts(metric_name_prefix, metric_name, counts_hash)
+      end
+    end
+  end
+
+  protected
+
+  def add_tags_to_group_counts(prefix, metric_name, group_counts)
+    tag_key = to_valid_tag_key(metric_name.split(".").last)
+
+    group_counts.map do |key, count|
+      { :metric => "#{prefix}.#{metric_name}", :value => count, tag_key => to_valid_tag(group_key_to_name(key)) }
+    end
+  end
+
+  # See valid tag name rules at https://docs.datadoghq.com/tagging/#defining-tags
+  def to_valid_tag(name)
+    name.gsub(/[^a-zA-Z_\-\:\.\d\/]/, "__")
+  end
+
+  def to_valid_tag_key(name)
+    return "#{name}_" if %w[host device source service].include?(name)
+
+    to_valid_tag(name)
+  end
+
+  def valid_metric_name?(metric_name)
+    # Actual limit is 200 but since the actual metric name in DataDog has
+    # "dsva_appeals.stats_collector_job." prepended, let's just stick with a 150 character limit.
+    return false if metric_name.length > 150
+
+    return true if metric_name =~ /\A[a-zA-Z][a-zA-Z\d_\.]*\Z/
+
+    false
+  end
+
+  def group_key_to_name(key)
+    return "nil" unless key
+
+    return key.map(&:underscore).map(&:parameterize).join(".") if key.instance_of?(Array)
+
+    return key.underscore.parameterize if key.instance_of?(String)
+
+    key.to_s.underscore.parameterize
+  end
+end
diff --git a/spec/jobs/stats_collector_job_spec.rb b/spec/jobs/stats_collector_job_spec.rb
index 077c5d2bfe6..9085ecc9dd5 100644
--- a/spec/jobs/stats_collector_job_spec.rb
+++ b/spec/jobs/stats_collector_job_spec.rb
@@ -18,11 +18,11 @@
 
   class ExampleDailyCollector
     def collect_stats
-      {
-        "ex_metric_daily" => 9,
-        "ex_metric_daily.stats1" => 2,
-        "ex_metric_daily.statsA.metric2" => 4
-      }
+      [
+        { "ex_metric_daily" => 9 },
+        { "ex_metric_daily.stats1" => 2 },
+        { "ex_metric_daily.statsA.metric2" => 4 }
+      ]
     end
   end
 
@@ -88,7 +88,7 @@ def collect_stats
       let(:emitted_gauges) { [] }
       let(:collector_gauges) do
         emitted_gauges.select { |gauge| gauge[:metric_group] == StatsCollectorJob.name.underscore }
-          .map { |gauge| [gauge[:metric_name], gauge] }.to_h
+          .group_by { |gauge| gauge[:metric_name] }
       end
       let(:runtime_gauges) do
         emitted_gauges.select { |gauge| gauge[:metric_name] == "runtime" }
@@ -104,9 +104,9 @@ def collect_stats
           expect(runtime_gauges["stats_collector_job"][:metric_value]).not_to be_nil
           expect(runtime_gauges["stats_collector_job.test_daily_collector"][:metric_value]).not_to be_nil
 
-          expect(collector_gauges["ex_metric_daily"][:metric_value]).to eq(9)
-          expect(collector_gauges["ex_metric_daily.stats1"][:metric_value]).to eq(2)
-          expect(collector_gauges["ex_metric_daily.statsA.metric2"][:metric_value]).to eq(4)
+          expect(collector_gauges["ex_metric_daily"].first[:metric_value]).to eq(9)
+          expect(collector_gauges["ex_metric_daily.stats1"].first[:metric_value]).to eq(2)
+          expect(collector_gauges["ex_metric_daily.statsA.metric2"].first[:metric_value]).to eq(4)
         end
       end
 
@@ -143,9 +143,41 @@ def collect_stats
           expect(runtime_gauges["stats_collector_job"][:metric_value]).not_to be_nil
           expect(runtime_gauges["stats_collector_job.test_daily_collector"][:metric_value]).not_to be_nil
 
-          expect(collector_gauges["ex_metric_daily"][:metric_value]).to eq(9)
-          expect(collector_gauges["ex_metric_daily.stats1"][:metric_value]).to eq(2)
-          expect(collector_gauges["ex_metric_daily.statsA.metric2"][:metric_value]).to eq(4)
+          expect(collector_gauges["ex_metric_daily"].first[:metric_value]).to eq(9)
+          expect(collector_gauges["ex_metric_daily.stats1"].first[:metric_value]).to eq(2)
+          expect(collector_gauges["ex_metric_daily.statsA.metric2"].first[:metric_value]).to eq(4)
+        end
+      end
+
+      context "with tagging collector" do
+        class ExampleTaggingCollector
+          def collect_stats
+            [
+              { "ex_tagged_metric" => 9 },
+              { metric: "ex_tagged_metric.status", value: 3, "status": "open" },
+              { metric: "ex_tagged_metric.status", value: 5, "status": "closed" }
+            ]
+          end
+        end
+        let(:daily_collectors) { { "tagging_collector" => ExampleTaggingCollector }.freeze }
+
+        let(:status_gauge) do
+          collector_gauges["ex_tagged_metric.status"].group_by { |gauge| gauge[:attrs][:status] }
+        end
+
+        it "records stats with tags when provided in collector results" do
+          allow(DataDogService).to receive(:emit_gauge) { |args| emitted_gauges.push(args) }
+
+          slack_msg = []
+          allow_any_instance_of(SlackService).to receive(:send_notification) { |_, first_arg| slack_msg << first_arg }
+
+          described_class.perform_now
+
+          expect(slack_msg).not_to include(/Fatal error/)
+
+          expect(collector_gauges["ex_tagged_metric"].first[:metric_value]).to eq(9)
+          expect(status_gauge["open"].first[:metric_value]).to eq(3)
+          expect(status_gauge["closed"].first[:metric_value]).to eq(5)
         end
       end
     end
diff --git a/spec/services/collectors/daily_counts_stats_collector_spec.rb b/spec/services/collectors/daily_counts_stats_collector_spec.rb
index 38d1dc2dbc1..64a4366458c 100644
--- a/spec/services/collectors/daily_counts_stats_collector_spec.rb
+++ b/spec/services/collectors/daily_counts_stats_collector_spec.rb
@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require "stats_collector_job"
+
 describe Collectors::DailyCountsStatsCollector do
   def random_time_of_day
     rand(0..23).hours + rand(0..59).minutes + rand(0..59).seconds
@@ -27,20 +29,63 @@ def random_time_of_day
                contention_reference_id: i,
                decision_review: create(:higher_level_review))
       end
+
+      create(:establish_claim, aasm_state: :unassigned, created_at: yesterday + random_time_of_day)
     end
 
     it "returns daily stats" do
-      expect(subject.collect_stats).to include(
-        "daily_counts.totals.claimant.decision_review.appeal" => 5,
-        "daily_counts.totals.claimant.payee_code.00" => 5,
-        "daily_counts.totals.hearing.nil" => 3,
-        "daily_counts.totals.case_review.judge" => 1,
-        "daily_counts.totals.case_review.attorney" => 2,
-        "daily_counts.totals.req_issues.higherlevelreview" => 2,
-        "daily_counts.increments.appeal.evidence_submission" => 2,
-        "daily_counts.increments.hearing" => 3,
-        "daily_counts.increments.appeal.status.unknown" => 2
+      expect(subject.collect_stats).to match_array(
+        [
+          { :metric => "daily_counts.totals.claimant.decision_review", :value => 5, "decision_review" => "appeal" },
+          { :metric => "daily_counts.totals.claimant.payee_code", :value => 5, "payee_code" => "00" },
+          { :metric => "daily_counts.totals.hearing.disposition", :value => 3, "disposition" => "nil" },
+          { :metric => "daily_counts.totals.case_review", :value => 1, "case_review" => "judge" },
+          { :metric => "daily_counts.totals.case_review", :value => 2, "case_review" => "attorney" },
+          { :metric => "daily_counts.totals.req_issue", :value => 2, "req_issue" => "higher_level_review" },
+          { :metric => "daily_counts.totals.claim_establishment", :value => 1, "claim_establishment" => "nil" },
+          { :metric => "daily_counts.totals.dispatch", :value => 1, "dispatch" => "establish_claim.unassigned" },
+          { :metric => "daily_counts.increments.counts", :value => 2, "counts" => "appeal" },
+          { :metric => "daily_counts.increments.counts", :value => 0, "counts" => "legacy_appeal" },
+          { :metric => "daily_counts.increments.counts", :value => 0, "counts" => "appeal_series" },
+          { :metric => "daily_counts.increments.counts", :value => 3, "counts" => "hearing" },
+          { :metric => "daily_counts.increments.counts", :value => 0, "counts" => "legacy_hearing" },
+          { :metric => "daily_counts.increments.appeal", :value => 2, "appeal" => "evidence_submission" },
+          { :metric => "daily_counts.increments.appeal.status", :value => 2, "status" => "unknown" }
+        ]
       )
     end
+
+    context "when run in the StatsCollectorJob" do
+      before do
+        stub_const("StatsCollectorJob::DAILY_COLLECTORS",  { "my_collector" => described_class }.freeze)
+        stub_const("StatsCollectorJob::WEEKLY_COLLECTORS", {})
+        stub_const("StatsCollectorJob::MONTHLY_COLLECTORS", {})
+      end
+
+      let(:emitted_gauges) { [] }
+      let(:collector_gauges) do
+        emitted_gauges.select { |gauge| gauge[:metric_group] == StatsCollectorJob.name.underscore }
+          .group_by { |gauge| gauge[:metric_name] }
+      end
+
+      let(:case_review_gauge) do
+        collector_gauges["daily_counts.totals.case_review"].group_by { |gauge| gauge[:attrs]["case_review"] }
+      end
+
+      it "records stats with tags" do
+        allow(DataDogService).to receive(:emit_gauge) { |args| emitted_gauges.push(args) }
+
+        slack_msg = []
+        allow_any_instance_of(SlackService).to receive(:send_notification) { |_, first_arg| slack_msg << first_arg }
+
+        StatsCollectorJob.new.perform_now
+
+        expect(slack_msg).not_to include(/Fatal error/)
+
+        expect(collector_gauges["daily_counts.increments.appeal"].first[:metric_value]).to eq(2)
+        expect(case_review_gauge["judge"].first[:metric_value]).to eq(1)
+        expect(case_review_gauge["attorney"].first[:metric_value]).to eq(2)
+      end
+    end
   end
 end
diff --git a/spec/services/collectors/request_issues_stats_collector_spec.rb b/spec/services/collectors/request_issues_stats_collector_spec.rb
index 5af6cfb5623..2f5fae0e566 100644
--- a/spec/services/collectors/request_issues_stats_collector_spec.rb
+++ b/spec/services/collectors/request_issues_stats_collector_spec.rb
@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require "stats_collector_job"
+
 describe Collectors::RequestIssuesStatsCollector do
   context "when unidentified RequestIssues with contentions exist" do
     before do
@@ -32,17 +34,57 @@
     it "records stats on unidentified request issues that have contentions" do
       stats = subject.collect_stats
 
-      metric_name_prefix = described_class::METRIC_NAME_PREFIX
-      expect(stats[metric_name_prefix]).to eq(9)
-      expect(stats["#{metric_name_prefix}.vet_count"]).to be_between(1, 9)
+      expected_stats = [
+        { :metric => "req_issues.w_contentions.status", :value => 2, "status" => "removed" },
+        { :metric => "req_issues.w_contentions.status", :value => 7, "status" => "nil" },
+        { :metric => "req_issues.w_contentions.benefit", :value => 4, "benefit" => "compensation" },
+        { :metric => "req_issues.w_contentions.benefit", :value => 5, "benefit" => "pension" },
+        { :metric => "req_issues.w_contentions.decis_review", :value => 3, "decis_review" => "higher_level_review" },
+        { :metric => "req_issues.w_contentions.decis_review", :value => 6, "decis_review" => "supplemental_claim" },
+        { :metric => "req_issues.w_contentions.report", :value => 0, "report" => "hlr_established" },
+        { :metric => "req_issues.w_contentions.report", :value => 0, "report" => "sc_established" },
+        { :metric => "req_issues.w_contentions.report", :value => 0, "report" => "030_end_products_established" },
+        { :metric => "req_issues.w_contentions.report", :value => 0, "report" => "040_end_products_established" },
+        { :metric => "req_issues.w_contentions.report", :value => 9, "report" => "created" },
+        { :metric => "req_issues.w_contentions.report", :value => 0, "report" => "edited" },
+        { :metric => "req_issues.w_contentions.report", :value => 9, "report" => "unidentified_created" }
+      ]
+      expect(stats).to include(*expected_stats)
+
+      vet_count_stat = stats.detect { |stat| stat["report"] == "vet_count" }
+      expect(vet_count_stat[:value]).to be_between(1, 9)
+    end
+
+    context "when run in the StatsCollectorJob" do
+      before do
+        stub_const("StatsCollectorJob::DAILY_COLLECTORS",  { "my_collector" => described_class }.freeze)
+        stub_const("StatsCollectorJob::WEEKLY_COLLECTORS", {})
+        stub_const("StatsCollectorJob::MONTHLY_COLLECTORS", {})
+      end
+
+      let(:emitted_gauges) { [] }
+      let(:collector_gauges) do
+        emitted_gauges.select { |gauge| gauge[:metric_group] == StatsCollectorJob.name.underscore }
+          .group_by { |gauge| gauge[:metric_name] }
+      end
+
+      let(:issue_status_gauge) do
+        collector_gauges["req_issues.w_contentions.status"].group_by { |gauge| gauge[:attrs]["status"] }
+      end
 
-      expect(stats["#{metric_name_prefix}.closed_status.removed"]).to eq(2)
+      it "records stats with tags" do
+        allow(DataDogService).to receive(:emit_gauge) { |args| emitted_gauges.push(args) }
 
-      expect(stats["#{metric_name_prefix}.benefit.pension"]).to eq(5)
-      expect(stats["#{metric_name_prefix}.benefit.compensation"]).to eq(4)
+        slack_msg = []
+        allow_any_instance_of(SlackService).to receive(:send_notification) { |_, first_arg| slack_msg << first_arg }
 
-      expect(stats["#{metric_name_prefix}.decision_review.HigherLevelReview"]).to eq(3)
-      expect(stats["#{metric_name_prefix}.decision_review.SupplementalClaim"]).to eq(6)
+        StatsCollectorJob.new.perform_now
+
+        expect(slack_msg).not_to include(/Fatal error/)
+
+        expect(issue_status_gauge["removed"].first[:metric_value]).to eq(2)
+        expect(issue_status_gauge["nil"].first[:metric_value]).to eq(7)
+      end
     end
   end
 end