Prevent concurrent execution of PipelineScheduleWorker

Currently, PipelineScheduleWorker is fired in a short period on
our production server. We can stop this behavior by locking the
execution thread with in_lock method.

app/workers/pipeline_schedule_worker.rb

@@ -3,20 +3,26 @@
 class PipelineScheduleWorker
   include ApplicationWorker
   include CronjobQueue
+  include ::Gitlab::ExclusiveLeaseHelpers
+
+  EXCLUSIVE_LOCK_KEY = 'pipeline_schedules:run:lock'
+  LOCK_TIMEOUT = 50.minutes
 
   # rubocop: disable CodeReuse/ActiveRecord
   def perform
-    Ci::PipelineSchedule.active.where("next_run_at < ?", Time.now)
-      .preload(:owner, :project).find_each do |schedule|
-
-      Ci::CreatePipelineService.new(schedule.project,
-                                    schedule.owner,
-                                    ref: schedule.ref)
-        .execute!(:schedule, ignore_skip_ci: true, save_on_errors: true, schedule: schedule)
-    rescue => e
-      error(schedule, e)
-    ensure
-      schedule.schedule_next_run!
+    in_lock(EXCLUSIVE_LOCK_KEY, ttl: LOCK_TIMEOUT, retries: 1) do
+      Ci::PipelineSchedule.active.where("next_run_at < ?", Time.now)
+        .preload(:owner, :project).find_each do |schedule|
+
+        schedule.schedule_next_run!
+
+        Ci::CreatePipelineService.new(schedule.project,
+                                      schedule.owner,
+                                      ref: schedule.ref)
+          .execute!(:schedule, ignore_skip_ci: true, save_on_errors: true, schedule: schedule)
+      rescue => e
+        error(schedule, e)
+      end
     end
   end
   # rubocop: enable CodeReuse/ActiveRecord

changelogs/unreleased/lock-pipeline-schedule-worker.yml

+---
+title: Prevent concurrent execution of PipelineScheduleWorker
+merge_request: 27781
+author:
+type: performance

spec/workers/pipeline_schedule_worker_spec.rb

@@ -3,6 +3,8 @@
 require 'spec_helper'
 
 describe PipelineScheduleWorker do
+  include ExclusiveLeaseHelpers
+
   subject { described_class.new.perform }
 
   set(:project) { create(:project, :repository) }
@@ -39,6 +41,16 @@ describe PipelineScheduleWorker do
 
       it_behaves_like 'successful scheduling'
 
+      context 'when exclusive lease has already been taken by the other instance' do
+        before do
+          stub_exclusive_lease_taken(described_class::EXCLUSIVE_LOCK_KEY, timeout: described_class::LOCK_TIMEOUT)
+        end
+
+        it 'raises an error and does not start creating pipelines' do
+          expect { subject }.to raise_error(Gitlab::ExclusiveLeaseHelpers::FailedToObtainLockError)
+        end
+      end
+
       context 'when the latest commit contains [ci skip]' do
         before do
           allow_any_instance_of(Ci::Pipeline)