Remove expired artifacts periodically

Rename

Introduce Destroy expired job artifacts service

Revert a bit

Add changelog

Use expired

Improve

Fix spec

Fix spec

Use bang for destroy

Introduce iteration limit

Update comment

Simplify more

Refacor

Remove unnecessary thing

Fix comments

Fix coding offence

Make loop helper exception free

app/models/ci/job_artifact.rb

@@ -73,6 +73,8 @@ module Ci
       where(file_type: types)
     end
 
+    scope :expired, -> (limit) { where('expire_at < ?', Time.now).limit(limit) }
+
     delegate :filename, :exists?, :open, to: :file
 
     enum file_type: {

app/services/ci/destroy_expired_job_artifacts_service.rb

+# frozen_string_literal: true
+
+module Ci
+  class DestroyExpiredJobArtifactsService
+    include ::Gitlab::ExclusiveLeaseHelpers
+    include ::Gitlab::LoopHelpers
+
+    BATCH_SIZE = 100
+    LOOP_TIMEOUT = 45.minutes
+    LOOP_LIMIT = 1000
+    EXCLUSIVE_LOCK_KEY = 'expired_job_artifacts:destroy:lock'
+    LOCK_TIMEOUT = 50.minutes
+
+    ##
+    # Destroy expired job artifacts on GitLab instance
+    #
+    # This destroy process cannot run for more than 45 minutes. This is for
+    # preventing multiple `ExpireBuildArtifactsWorker` CRON jobs run concurrently,
+    # which is scheduled at every hour.
+    def execute
+      in_lock(EXCLUSIVE_LOCK_KEY, ttl: LOCK_TIMEOUT, retries: 1) do
+        loop_until(timeout: LOOP_TIMEOUT, limit: LOOP_LIMIT) do
+          destroy_batch
+        end
+      end
+    end
+
+    private
+
+    def destroy_batch
+      artifacts = Ci::JobArtifact.expired(BATCH_SIZE).to_a
+
+      return false if artifacts.empty?
+
+      artifacts.each(&:destroy!)
+    end
+  end
+end

app/workers/expire_build_artifacts_worker.rb

@@ -4,8 +4,20 @@ class ExpireBuildArtifactsWorker
   include ApplicationWorker
   include CronjobQueue
 
-  # rubocop: disable CodeReuse/ActiveRecord
   def perform
+    if Feature.enabled?(:ci_new_expire_job_artifacts_service, default_enabled: true)
+      perform_efficient_artifacts_removal
+    else
+      perform_legacy_artifacts_removal
+    end
+  end
+
+  def perform_efficient_artifacts_removal
+    Ci::DestroyExpiredJobArtifactsService.new.execute
+  end
+
+  # rubocop: disable CodeReuse/ActiveRecord
+  def perform_legacy_artifacts_removal
     Rails.logger.info 'Scheduling removal of build artifacts'
 
     build_ids = Ci::Build.with_expired_artifacts.pluck(:id)

changelogs/unreleased/expire-job-artifacts-worker.yml

+---
+title: Efficiently remove expired artifacts in `ExpireBuildArtifactsWorker`
+merge_request: 24450
+author:
+type: performance

lib/gitlab/loop_helpers.rb

+# frozen_string_literal: true
+
+module Gitlab
+  module LoopHelpers
+    ##
+    # This helper method repeats the same task until it's expired.
+    #
+    # Note: ExpiredLoopError does not happen until the given block finished.
+    #       Please do not use this method for heavy or asynchronous operations.
+    def loop_until(timeout: nil, limit: 1_000_000)
+      raise ArgumentError unless limit
+
+      start = Time.now
+
+      limit.times do
+        return true unless yield
+
+        return false if timeout && (Time.now - start) > timeout
+      end
+
+      false
+    end
+  end
+end

spec/lib/gitlab/loop_helpers_spec.rb

+require 'spec_helper'
+
+describe Gitlab::LoopHelpers do
+  let(:class_instance) { (Class.new { include ::Gitlab::LoopHelpers }).new }
+
+  describe '#loop_until' do
+    subject do
+      class_instance.loop_until(**params) { true }
+    end
+
+    context 'when limit is not given' do
+      let(:params) { { limit: nil } }
+
+      it 'raises an error' do
+        expect { subject }.to raise_error(ArgumentError)
+      end
+    end
+
+    context 'when timeout is specified' do
+      let(:params) { { timeout: 1.second } }
+
+      it "returns false after it's expired" do
+        is_expected.to be_falsy
+      end
+
+      it 'executes the block at least once' do
+        expect { |b| class_instance.loop_until(**params, &b) }
+          .to yield_control.at_least(1)
+      end
+    end
+
+    context 'when iteration limit is specified' do
+      let(:params) { { limit: 1 } }
+
+      it "returns false after it's expired" do
+        is_expected.to be_falsy
+      end
+
+      it 'executes the block once' do
+        expect { |b| class_instance.loop_until(**params, &b) }
+          .to yield_control.once
+      end
+    end
+  end
+end

spec/services/ci/destroy_expired_job_artifacts_service_spec.rb

+require 'spec_helper'
+
+describe Ci::DestroyExpiredJobArtifactsService, :clean_gitlab_redis_shared_state do
+  include ExclusiveLeaseHelpers
+
+  describe '.execute' do
+    subject { service.execute }
+
+    let(:service) { described_class.new }
+    let!(:artifact) { create(:ci_job_artifact, expire_at: 1.day.ago) }
+
+    it 'destroys expired job artifacts' do
+      expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
+    end
+
+    context 'when artifact is not expired' do
+      let!(:artifact) { create(:ci_job_artifact, expire_at: 1.day.since) }
+
+      it 'does not destroy expired job artifacts' do
+        expect { subject }.not_to change { Ci::JobArtifact.count }
+      end
+    end
+
+    context 'when artifact is permanent' do
+      let!(:artifact) { create(:ci_job_artifact, expire_at: nil) }
+
+      it 'does not destroy expired job artifacts' do
+        expect { subject }.not_to change { Ci::JobArtifact.count }
+      end
+    end
+
+    context 'when failed to destroy artifact' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::LOOP_LIMIT', 10)
+
+        allow_any_instance_of(Ci::JobArtifact)
+          .to receive(:destroy!)
+          .and_raise(ActiveRecord::RecordNotDestroyed)
+      end
+
+      it 'raises an exception and stop destroying' do
+        expect { subject }.to raise_error(ActiveRecord::RecordNotDestroyed)
+      end
+    end
+
+    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 destroying' do
+        expect { subject }.to raise_error(Gitlab::ExclusiveLeaseHelpers::FailedToObtainLockError)
+      end
+    end
+
+    context 'when timeout happens' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::LOOP_TIMEOUT', 1.second)
+        allow_any_instance_of(described_class).to receive(:destroy_batch) { true }
+      end
+
+      it 'returns false and does not continue destroying' do
+        is_expected.to be_falsy
+      end
+    end
+
+    context 'when loop reached loop limit' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::LOOP_LIMIT', 1)
+        stub_const('Ci::DestroyExpiredJobArtifactsService::BATCH_SIZE', 1)
+      end
+
+      let!(:artifact) { create_list(:ci_job_artifact, 2, expire_at: 1.day.ago) }
+
+      it 'raises an error and does not continue destroying' do
+        is_expected.to be_falsy
+      end
+
+      it 'destroys one artifact' do
+        expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
+      end
+    end
+
+    context 'when there are no artifacts' do
+      let!(:artifact) { }
+
+      it 'does not raise error' do
+        expect { subject }.not_to raise_error
+      end
+    end
+
+    context 'when there are artifacts more than batch sizes' do
+      before do
+        stub_const('Ci::DestroyExpiredJobArtifactsService::BATCH_SIZE', 1)
+      end
+
+      let!(:artifact) { create_list(:ci_job_artifact, 2, expire_at: 1.day.ago) }
+
+      it 'destroys all expired artifacts' do
+        expect { subject }.to change { Ci::JobArtifact.count }.by(-2)
+      end
+    end
+  end
+end

spec/workers/expire_build_artifacts_worker_spec.rb

@@ -11,6 +11,7 @@ describe ExpireBuildArtifactsWorker do
 
   describe '#perform' do
     before do
+      stub_feature_flags(ci_new_expire_job_artifacts_service: false)
       build
     end
 
@@ -47,4 +48,17 @@ describe ExpireBuildArtifactsWorker do
       Sidekiq::Queues.jobs_by_worker['ExpireBuildInstanceArtifactsWorker']
     end
   end
+
+  describe '#perform with ci_new_expire_job_artifacts_service feature flag' do
+    before do
+      stub_feature_flags(ci_new_expire_job_artifacts_service: true)
+    end
+
+    it 'executes a service' do
+      expect_any_instance_of(Ci::DestroyExpiredJobArtifactsService).to receive(:execute)
+      expect(ExpireBuildInstanceArtifactsWorker).not_to receive(:bulk_perform_async)
+
+      worker.perform
+    end
+  end
 end