Use new locked column for Ci::JobArtifacts::DestroyAllExpiredService

Fixes https://gitlab.com/gitlab-org/gitlab/-/issues/327281

app/services/ci/job_artifacts/destroy_all_expired_service.rb

@@ -24,7 +24,11 @@ module Ci
       # which is scheduled every 7 minutes.
       def execute
         in_lock(EXCLUSIVE_LOCK_KEY, ttl: LOCK_TIMEOUT, retries: 1) do
-          destroy_job_artifacts_with_slow_iteration(Time.current)
+          if ::Feature.enabled?(:ci_destroy_unlocked_job_artifacts)
+            destroy_unlocked_job_artifacts(Time.current)
+          else
+            destroy_job_artifacts_with_slow_iteration(Time.current)
+          end
         end
 
         @removed_artifacts_count
@@ -32,13 +36,21 @@ module Ci
 
       private
 
+      def destroy_unlocked_job_artifacts(start_at)
+        loop_until(timeout: LOOP_TIMEOUT, limit: LOOP_LIMIT) do
+          artifacts = Ci::JobArtifact.expired_before(start_at).artifact_unlocked.limit(BATCH_SIZE)
+          service_response = destroy_batch(artifacts)
+          @removed_artifacts_count += service_response[:destroyed_artifacts_count]
+        end
+      end
+
       def destroy_job_artifacts_with_slow_iteration(start_at)
         Ci::JobArtifact.expired_before(start_at).each_batch(of: BATCH_SIZE, column: :expire_at, order: :desc) do |relation, index|
           # For performance reasons, join with ci_pipelines after the batch is queried.
           # See: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/47496
           artifacts = relation.unlocked
 
-          service_response = destroy_batch_async(artifacts)
+          service_response = destroy_batch(artifacts)
           @removed_artifacts_count += service_response[:destroyed_artifacts_count]
 
           break if loop_timeout?(start_at)
@@ -46,7 +58,7 @@ module Ci
         end
       end
 
-      def destroy_batch_async(artifacts)
+      def destroy_batch(artifacts)
         Ci::JobArtifacts::DestroyBatchService.new(artifacts).execute
       end
 

app/services/ci/job_artifacts/destroy_batch_service.rb

@@ -34,7 +34,7 @@ module Ci
 
         # This is executed outside of the transaction because it depends on Redis
         update_project_statistics! if update_stats
-        increment_monitoring_statistics(artifacts_count)
+        increment_monitoring_statistics(artifacts_count, artifacts_bytes)
 
         success(destroyed_artifacts_count: artifacts_count,
           statistics_updates: affected_project_statistics)
@@ -63,8 +63,9 @@ module Ci
         end
       end
 
-      def increment_monitoring_statistics(size)
-        metrics.increment_destroyed_artifacts(size)
+      def increment_monitoring_statistics(size, bytes)
+        metrics.increment_destroyed_artifacts_count(size)
+        metrics.increment_destroyed_artifacts_bytes(bytes)
       end
 
       def metrics
@@ -76,6 +77,12 @@ module Ci
           @job_artifacts.count
         end
       end
+
+      def artifacts_bytes
+        strong_memoize(:artifacts_bytes) do
+          @job_artifacts.sum { |artifact| artifact.try(:size) || 0 }
+        end
+      end
     end
   end
 end

config/feature_flags/development/ci_destroy_unlocked_job_artifacts.yml

+---
+name: ci_destroy_unlocked_job_artifacts
+introduced_by_url: https://gitlab.com/gitlab-org/gitlab/-/merge_requests/72406
+rollout_issue_url: https://gitlab.com/gitlab-org/gitlab/-/issues/338165
+milestone: '14.5'
+type: development
+group: group::testing
+default_enabled: false

ee/spec/services/ee/ci/job_artifacts/destroy_all_expired_service_spec.rb

@@ -10,53 +10,48 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
 
     let(:service) { described_class.new }
 
-    let_it_be(:artifact) { create(:ci_job_artifact, expire_at: 1.day.ago) }
-    let_it_be(:security_scan) { create(:security_scan, build: artifact.job) }
+    let_it_be(:locked_pipeline) { create(:ci_pipeline, :artifacts_locked) }
+    let_it_be(:pipeline) { create(:ci_pipeline, :unlocked) }
+    let_it_be(:locked_job) { create(:ci_build, :success, pipeline: locked_pipeline) }
+    let_it_be(:job) { create(:ci_build, :success, pipeline: pipeline) }
+    let_it_be(:security_scan) { create(:security_scan, build: job) }
     let_it_be(:security_finding) { create(:security_finding, scan: security_scan) }
 
-    before(:all) do
-      artifact.job.pipeline.unlocked!
-    end
-
     context 'when artifact is expired' do
       context 'when artifact is not locked' do
+        let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
+
         it 'destroys job artifact and the security finding' do
           expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
-                            .and change { Security::Finding.count }.from(1).to(0)
+                            .and change { Security::Finding.count }.by(-1)
         end
       end
 
       context 'when artifact is locked' do
-        before do
-          artifact.job.pipeline.reload.artifacts_locked!
-        end
+        let!(:artifact) { create(:ci_job_artifact, :expired, job: locked_job, locked: locked_job.pipeline.locked) }
 
         it 'does not destroy job artifact' do
           expect { subject }.to not_change { Ci::JobArtifact.count }
-                            .and not_change { Security::Finding.count }.from(1)
+                            .and not_change { Security::Finding.count }
         end
       end
     end
 
     context 'when artifact is not expired' do
-      before do
-        artifact.update_column(:expire_at, 1.day.since)
-      end
+      let!(:artifact) { create(:ci_job_artifact, job: job, locked: job.pipeline.locked) }
 
       it 'does not destroy expired job artifacts' do
         expect { subject }.to not_change { Ci::JobArtifact.count }
-                          .and not_change { Security::Finding.count }.from(1)
+                          .and not_change { Security::Finding.count }
       end
     end
 
     context 'when artifact is permanent' do
-      before do
-        artifact.update_column(:expire_at, nil)
-      end
+      let!(:artifact) { create(:ci_job_artifact, expire_at: nil, job: job, locked: job.pipeline.locked) }
 
       it 'does not destroy expired job artifacts' do
         expect { subject }.to not_change { Ci::JobArtifact.count }
-                          .and not_change { Security::Finding.count }.from(1)
+                          .and not_change { Security::Finding.count }
       end
     end
 
@@ -69,41 +64,38 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
           .and_raise(ActiveRecord::RecordNotDestroyed)
       end
 
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
+
       it 'raises an exception and stop destroying' do
         expect { subject }.to raise_error(ActiveRecord::RecordNotDestroyed)
-                          .and not_change { Security::Finding.count }.from(1)
+                          .and not_change { Security::Finding.count }
       end
     end
 
     context 'when there are artifacts more than batch sizes' do
       before do
         stub_const('Ci::JobArtifacts::DestroyAllExpiredService::BATCH_SIZE', 1)
-
-        second_artifact.job.pipeline.unlocked!
       end
 
-      let!(:second_artifact) { create(:ci_job_artifact, expire_at: 1.day.ago) }
-      let!(:second_security_scan) { create(:security_scan, build: second_artifact.job) }
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
+      let!(:second_job) { create(:ci_build, :success, pipeline: pipeline) }
+      let!(:second_artifact) { create(:ci_job_artifact, :expired, job: second_job, locked: second_job.pipeline.locked) }
+      let!(:second_security_scan) { create(:security_scan, build: second_job) }
       let!(:second_security_finding) { create(:security_finding, scan: second_security_scan) }
 
       it 'destroys all expired artifacts' do
         expect { subject }.to change { Ci::JobArtifact.count }.by(-2)
-                          .and change { Security::Finding.count }.from(2).to(0)
+                          .and change { Security::Finding.count }.by(-2)
       end
     end
 
     context 'when some artifacts are locked' do
-      before do
-        pipeline = create(:ci_pipeline, locked: :artifacts_locked)
-        job = create(:ci_build, pipeline: pipeline)
-        create(:ci_job_artifact, expire_at: 1.day.ago, job: job)
-        security_scan = create(:security_scan, build: job)
-        create(:security_finding, scan: security_scan)
-      end
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
+      let!(:second_artifact) { create(:ci_job_artifact, :expired, job: locked_job, locked: locked_job.pipeline.locked) }
 
       it 'destroys only unlocked artifacts' do
         expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
-                          .and change { Security::Finding.count }.from(2).to(1)
+                          .and change { Security::Finding.count }.by(-1)
       end
     end
   end

lib/gitlab/ci/artifacts/metrics.rb

@@ -6,10 +6,14 @@ module Gitlab
       class Metrics
         include Gitlab::Utils::StrongMemoize
 
-        def increment_destroyed_artifacts(size)
+        def increment_destroyed_artifacts_count(size)
           destroyed_artifacts_counter.increment({}, size.to_i)
         end
 
+        def increment_destroyed_artifacts_bytes(bytes)
+          destroyed_artifacts_bytes_counter.increment({}, bytes)
+        end
+
         private
 
         def destroyed_artifacts_counter
@@ -20,6 +24,15 @@ module Gitlab
             ::Gitlab::Metrics.counter(name, comment)
           end
         end
+
+        def destroyed_artifacts_bytes_counter
+          strong_memoize(:destroyed_artifacts_bytes_counter) do
+            name = :destroyed_job_artifacts_bytes_total
+            comment = 'Counter of bytes of destroyed expired job artifacts'
+
+            ::Gitlab::Metrics.counter(name, comment)
+          end
+        end
       end
     end
   end

spec/lib/gitlab/ci/artifacts/metrics_spec.rb

@@ -10,9 +10,9 @@ RSpec.describe Gitlab::Ci::Artifacts::Metrics, :prometheus do
       let(:counter) { metrics.send(:destroyed_artifacts_counter) }
 
       it 'increments a single counter' do
-        subject.increment_destroyed_artifacts(10)
-        subject.increment_destroyed_artifacts(20)
-        subject.increment_destroyed_artifacts(30)
+        subject.increment_destroyed_artifacts_count(10)
+        subject.increment_destroyed_artifacts_count(20)
+        subject.increment_destroyed_artifacts_count(30)
 
         expect(counter.get).to eq 60
         expect(counter.values.count).to eq 1

spec/services/ci/job_artifacts/destroy_all_expired_service_spec.rb

@@ -16,26 +16,43 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
     let_it_be(:job) { create(:ci_build, :success, pipeline: pipeline) }
 
     context 'when artifact is expired' do
-      let!(:artifact) { create(:ci_job_artifact, :expired, job: job) }
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
 
       context 'with preloaded relationships' do
         before do
           stub_const("#{described_class}::LOOP_LIMIT", 1)
         end
 
-        it 'performs the smallest number of queries for job_artifacts' do
-          log = ActiveRecord::QueryRecorder.new { subject }
+        context 'with ci_destroy_unlocked_job_artifacts feature flag disabled' do
+          before do
+            stub_feature_flags(ci_destroy_unlocked_job_artifacts: false)
+          end
+
+          it 'performs the smallest number of queries for job_artifacts' do
+            log = ActiveRecord::QueryRecorder.new { subject }
+
+            # SELECT expired ci_job_artifacts - 3 queries from each_batch
+            # PRELOAD projects, routes, project_statistics
+            # BEGIN
+            # INSERT into ci_deleted_objects
+            # DELETE loaded ci_job_artifacts
+            # DELETE security_findings  -- for EE
+            # COMMIT
+            # SELECT next expired ci_job_artifacts
+
+            expect(log.count).to be_within(1).of(10)
+          end
+        end
 
-          # SELECT expired ci_job_artifacts - 3 queries from each_batch
-          # PRELOAD projects, routes, project_statistics
-          # BEGIN
-          # INSERT into ci_deleted_objects
-          # DELETE loaded ci_job_artifacts
-          # DELETE security_findings  -- for EE
-          # COMMIT
-          # SELECT next expired ci_job_artifacts
+        context 'with ci_destroy_unlocked_job_artifacts feature flag enabled' do
+          before do
+            stub_feature_flags(ci_destroy_unlocked_job_artifacts: true)
+          end
 
-          expect(log.count).to be_within(1).of(10)
+          it 'performs the smallest number of queries for job_artifacts' do
+            log = ActiveRecord::QueryRecorder.new { subject }
+            expect(log.count).to be_within(1).of(8)
+          end
         end
       end
 
@@ -53,7 +70,7 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
         end
 
         context 'when the artifact has a file attached to it' do
-          let!(:artifact) { create(:ci_job_artifact, :expired, :zip, job: job) }
+          let!(:artifact) { create(:ci_job_artifact, :expired, :zip, job: job, locked: job.pipeline.locked) }
 
           it 'creates a deleted object' do
             expect { subject }.to change { Ci::DeletedObject.count }.by(1)
@@ -74,7 +91,7 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
       end
 
       context 'when artifact is locked' do
-        let!(:artifact) { create(:ci_job_artifact, :expired, job: locked_job) }
+        let!(:artifact) { create(:ci_job_artifact, :expired, job: locked_job, locked: locked_job.pipeline.locked) }
 
         it 'does not destroy job artifact' do
           expect { subject }.not_to change { Ci::JobArtifact.count }
@@ -83,7 +100,7 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
     end
 
     context 'when artifact is not expired' do
-      let!(:artifact) { create(:ci_job_artifact, job: job) }
+      let!(:artifact) { create(:ci_job_artifact, job: job, locked: job.pipeline.locked) }
 
       it 'does not destroy expired job artifacts' do
         expect { subject }.not_to change { Ci::JobArtifact.count }
@@ -91,7 +108,7 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
     end
 
     context 'when artifact is permanent' do
-      let!(:artifact) { create(:ci_job_artifact, expire_at: nil, job: job) }
+      let!(:artifact) { create(:ci_job_artifact, expire_at: nil, job: job, locked: job.pipeline.locked) }
 
       it 'does not destroy expired job artifacts' do
         expect { subject }.not_to change { Ci::JobArtifact.count }
@@ -99,7 +116,7 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
     end
 
     context 'when failed to destroy artifact' do
-      let!(:artifact) { create(:ci_job_artifact, :expired, job: job) }
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
 
       before do
         stub_const("#{described_class}::LOOP_LIMIT", 10)
@@ -135,7 +152,7 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
     end
 
     context 'when exclusive lease has already been taken by the other instance' do
-      let!(:artifact) { create(:ci_job_artifact, :expired, job: job) }
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
 
       before do
         stub_exclusive_lease_taken(described_class::EXCLUSIVE_LOCK_KEY, timeout: described_class::LOCK_TIMEOUT)
@@ -149,8 +166,8 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
 
     context 'with a second artifact and batch size of 1' do
       let(:second_job) { create(:ci_build, :success, pipeline: pipeline) }
-      let!(:second_artifact) { create(:ci_job_artifact, :archive, expire_at: 1.day.ago, job: second_job) }
-      let!(:artifact) { create(:ci_job_artifact, :expired, job: job) }
+      let!(:second_artifact) { create(:ci_job_artifact, :archive, expire_at: 1.day.ago, job: second_job, locked: job.pipeline.locked) }
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
 
       before do
         stub_const("#{described_class}::BATCH_SIZE", 1)
@@ -206,8 +223,8 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
     end
 
     context 'when some artifacts are locked' do
-      let!(:artifact) { create(:ci_job_artifact, :expired, job: job) }
-      let!(:locked_artifact) { create(:ci_job_artifact, :expired, job: locked_job) }
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: job, locked: job.pipeline.locked) }
+      let!(:locked_artifact) { create(:ci_job_artifact, :expired, job: locked_job, locked: locked_job.pipeline.locked) }
 
       it 'destroys only unlocked artifacts' do
         expect { subject }.to change { Ci::JobArtifact.count }.by(-1)
@@ -216,7 +233,7 @@ RSpec.describe Ci::JobArtifacts::DestroyAllExpiredService, :clean_gitlab_redis_s
     end
 
     context 'when all artifacts are locked' do
-      let!(:artifact) { create(:ci_job_artifact, :expired, job: locked_job) }
+      let!(:artifact) { create(:ci_job_artifact, :expired, job: locked_job, locked: locked_job.pipeline.locked) }
 
       it 'destroys no artifacts' do
         expect { subject }.to change { Ci::JobArtifact.count }.by(0)

spec/services/ci/job_artifacts/destroy_batch_service_spec.rb

@@ -29,7 +29,8 @@ RSpec.describe Ci::JobArtifacts::DestroyBatchService do
 
       it 'reports metrics for destroyed artifacts' do
         expect_next_instance_of(Gitlab::Ci::Artifacts::Metrics) do |metrics|
-          expect(metrics).to receive(:increment_destroyed_artifacts).with(1).and_call_original
+          expect(metrics).to receive(:increment_destroyed_artifacts_count).with(1).and_call_original
+          expect(metrics).to receive(:increment_destroyed_artifacts_bytes).with(107464).and_call_original
         end
 
         execute