Add 'shard' label to 'job_queue_duration_seconds' metric

app/services/ci/register_job_service.rb

@@ -8,6 +8,8 @@ module Ci
 
     JOB_QUEUE_DURATION_SECONDS_BUCKETS = [1, 3, 10, 30, 60, 300, 900, 1800, 3600].freeze
     JOBS_RUNNING_FOR_PROJECT_MAX_BUCKET = 5.freeze
+    METRICS_SHARD_TAG_PREFIX = 'metrics_shard::'.freeze
+    DEFAULT_METRICS_SHARD = 'default'.freeze
 
     Result = Struct.new(:build, :valid?)
 
@@ -193,7 +195,13 @@ module Ci
 
     def register_success(job)
       labels = { shared_runner: runner.instance_type?,
-                 jobs_running_for_project: jobs_running_for_project(job) }
+                 jobs_running_for_project: jobs_running_for_project(job),
+                 shard: DEFAULT_METRICS_SHARD }
+
+      if runner.instance_type?
+        shard = runner.tag_list.sort.find { |name| name.starts_with?(METRICS_SHARD_TAG_PREFIX) }
+        labels[:shard] = shard.gsub(METRICS_SHARD_TAG_PREFIX, '') if shard
+      end
 
       job_queue_duration_seconds.observe(labels, Time.now - job.queued_at) unless job.queued_at.nil?
       attempt_counter.increment

spec/services/ci/register_job_service_spec.rb

@@ -612,7 +612,8 @@ module Ci
           allow(attempt_counter).to receive(:increment)
           expect(job_queue_duration_seconds).to receive(:observe)
             .with({ shared_runner: expected_shared_runner,
-                    jobs_running_for_project: expected_jobs_running_for_project_first_job }, 1800)
+                    jobs_running_for_project: expected_jobs_running_for_project_first_job,
+                    shard: expected_shard }, 1800)
 
           execute(runner)
         end
@@ -625,7 +626,8 @@ module Ci
             allow(attempt_counter).to receive(:increment)
             expect(job_queue_duration_seconds).to receive(:observe)
               .with({ shared_runner: expected_shared_runner,
-                      jobs_running_for_project: expected_jobs_running_for_project_third_job }, 1800)
+                      jobs_running_for_project: expected_jobs_running_for_project_third_job,
+                      shard: expected_shard }, 1800)
 
             execute(runner)
           end
@@ -638,13 +640,28 @@ module Ci
       end
 
       context 'when shared runner is used' do
-        let(:runner) { shared_runner }
+        let(:runner) { create(:ci_runner, :instance, tag_list: %w(tag1 tag2)) }
         let(:expected_shared_runner) { true }
+        let(:expected_shard) { Ci::RegisterJobService::DEFAULT_METRICS_SHARD }
         let(:expected_jobs_running_for_project_first_job) { 0 }
         let(:expected_jobs_running_for_project_third_job) { 2 }
 
         it_behaves_like 'metrics collector'
 
+        context 'when metrics_shard tag is defined' do
+          let(:runner) { create(:ci_runner, :instance, tag_list: %w(tag1 metrics_shard::shard_tag tag2)) }
+          let(:expected_shard) { 'shard_tag' }
+
+          it_behaves_like 'metrics collector'
+        end
+
+        context 'when multiple metrics_shard tag is defined' do
+          let(:runner) { create(:ci_runner, :instance, tag_list: %w(tag1 metrics_shard::shard_tag metrics_shard::shard_tag_2 tag2)) }
+          let(:expected_shard) { 'shard_tag' }
+
+          it_behaves_like 'metrics collector'
+        end
+
         context 'when pending job with queued_at=nil is used' do
           before do
             pending_job.update(queued_at: nil)
@@ -662,8 +679,9 @@ module Ci
       end
 
       context 'when specific runner is used' do
-        let(:runner) { specific_runner }
+        let(:runner) { create(:ci_runner, :project, projects: [project], tag_list: %w(tag1 metrics_shard::shard_tag tag2)) }
         let(:expected_shared_runner) { false }
+        let(:expected_shard) { Ci::RegisterJobService::DEFAULT_METRICS_SHARD }
         let(:expected_jobs_running_for_project_first_job) { '+Inf' }
         let(:expected_jobs_running_for_project_third_job) { '+Inf' }