Delete job artifact on JobArtifactDeletedEvent

db/migrate/20180103234731_create_geo_job_artifact_deleted_event.rb

+# See http://doc.gitlab.com/ce/development/migration_style_guide.html
+# for more information on how to write migrations for GitLab.
+
+class CreateGeoJobArtifactDeletedEvent < ActiveRecord::Migration
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  def change
+    create_table :geo_job_artifact_deleted_events, id: :bigserial do |t|
+      t.references :job_artifact, references: :ci_job_artifacts, index: true, foreign_key: false, null: false
+      t.string :file_path, null: false
+    end
+
+    add_column :geo_event_log, :job_artifact_deleted_event_id, :integer, limit: 8
+  end
+end

db/migrate/20180104001824_add_geo_job_artifact_deleted_events_foreign_key.rb

+# See http://doc.gitlab.com/ce/development/migration_style_guide.html
+# for more information on how to write migrations for GitLab.
+
+class AddGeoJobArtifactDeletedEventsForeignKey < ActiveRecord::Migration
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_foreign_key :geo_event_log, :geo_job_artifact_deleted_events,
+                               column: :job_artifact_deleted_event_id, on_delete: :cascade
+  end
+
+  def down
+    remove_foreign_key :geo_event_log, column: :job_artifact_deleted_event_id
+  end
+end

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 20171230123729) do
+ActiveRecord::Schema.define(version: 20180104001824) do
 
   # These are extensions that must be enabled in order to support this database
   enable_extension "plpgsql"
@@ -912,6 +912,7 @@ ActiveRecord::Schema.define(version: 20171230123729) do
     t.integer "hashed_storage_migrated_event_id", limit: 8
     t.integer "lfs_object_deleted_event_id", limit: 8
     t.integer "hashed_storage_attachments_event_id", limit: 8
+    t.integer "job_artifact_deleted_event_id", limit: 8
   end
 
   add_index "geo_event_log", ["repositories_changed_event_id"], name: "index_geo_event_log_on_repositories_changed_event_id", using: :btree
@@ -942,6 +943,13 @@ ActiveRecord::Schema.define(version: 20171230123729) do
 
   add_index "geo_hashed_storage_migrated_events", ["project_id"], name: "index_geo_hashed_storage_migrated_events_on_project_id", using: :btree
 
+  create_table "geo_job_artifact_deleted_events", id: :bigserial, force: :cascade do |t|
+    t.integer "job_artifact_id", null: false
+    t.string "file_path", null: false
+  end
+
+  add_index "geo_job_artifact_deleted_events", ["job_artifact_id"], name: "index_geo_job_artifact_deleted_events_on_job_artifact_id", using: :btree
+
   create_table "geo_lfs_object_deleted_events", id: :bigserial, force: :cascade do |t|
     t.integer "lfs_object_id", null: false
     t.string "oid", null: false
@@ -2489,6 +2497,7 @@ ActiveRecord::Schema.define(version: 20171230123729) do
   add_foreign_key "gcp_clusters", "services", on_delete: :nullify
   add_foreign_key "gcp_clusters", "users", on_delete: :nullify
   add_foreign_key "geo_event_log", "geo_hashed_storage_migrated_events", column: "hashed_storage_migrated_event_id", name: "fk_27548c6db3", on_delete: :cascade
+  add_foreign_key "geo_event_log", "geo_job_artifact_deleted_events", column: "job_artifact_deleted_event_id", name: "fk_176d3fbb5d", on_delete: :cascade
   add_foreign_key "geo_event_log", "geo_lfs_object_deleted_events", column: "lfs_object_deleted_event_id", name: "fk_d5af95fcd9", on_delete: :cascade
   add_foreign_key "geo_event_log", "geo_repositories_changed_events", column: "repositories_changed_event_id", name: "fk_4a99ebfd60", on_delete: :cascade
   add_foreign_key "geo_event_log", "geo_repository_created_events", column: "repository_created_event_id", name: "fk_9b9afb1916", on_delete: :cascade

ee/app/models/geo/event_log.rb

@@ -30,6 +30,10 @@ module Geo
       class_name: 'Geo::LfsObjectDeletedEvent',
       foreign_key: :lfs_object_deleted_event_id
 
+    belongs_to :job_artifact_deleted_event,
+      class_name: 'Geo::JobArtifactDeletedEvent',
+      foreign_key: :job_artifact_deleted_event_id
+
     belongs_to :hashed_storage_attachments_event,
       class_name: 'Geo::HashedStorageAttachmentsEvent',
       foreign_key: :hashed_storage_attachments_event_id
@@ -46,6 +50,7 @@ module Geo
         repositories_changed_event ||
         hashed_storage_migrated_event ||
         lfs_object_deleted_event ||
+        job_artifact_deleted_event ||
         hashed_storage_attachments_event
     end
 

ee/app/models/geo/job_artifact_deleted_event.rb

+module Geo
+  class JobArtifactDeletedEvent < ActiveRecord::Base
+    include Geo::Model
+
+    belongs_to :job_artifact, class_name: 'Ci::JobArtifact'
+
+    validates :job_artifact, presence: true
+  end
+end

ee/lib/gitlab/geo/log_cursor/daemon.rb

@@ -216,12 +216,39 @@ module Gitlab
           ::Geo::FileRegistry.lfs_objects.where(file_id: event.lfs_object_id).delete_all
         end
 
+        def handle_job_artifact_deleted_event(event, created_at)
+          file_registry_job_artifacts = ::Geo::FileRegistry.job_artifacts.where(file_id: event.job_artifact_id)
+          return unless file_registry_job_artifacts.any? # avoid race condition
+
+          file_path = File.join(JobArtifactUploader.local_store_path, event.file_path)
+
+          if File.file?(file_path)
+            deleted = delete_file(file_path) # delete synchronously to ensure consistency
+            return unless deleted # do not delete file from registry if deletion failed
+          end
+
+          logger.event_info(
+            created_at,
+            message: 'Deleted job artifact',
+            file_id: event.job_artifact_id,
+            file_path: file_path)
+
+          file_registry_job_artifacts.delete_all
+        end
+
         def find_or_initialize_registry(project_id, attrs)
           registry = ::Geo::ProjectRegistry.find_or_initialize_by(project_id: project_id)
           registry.assign_attributes(attrs)
           registry
         end
 
+        def delete_file(path)
+          File.delete(path)
+        rescue => ex
+          logger.error("Failed to remove file", exception: ex.class.name, details: ex.message, filename: path)
+          false
+        end
+
         # Sleeps for the expired TTL that remains on the lease plus some random seconds.
         #
         # This allows multiple GeoLogCursors to randomly process a batch of events,

spec/ee/spec/lib/gitlab/geo/log_cursor/daemon_spec.rb

@@ -300,5 +300,110 @@ describe Gitlab::Geo::LogCursor::Daemon, :postgresql, :clean_gitlab_redis_shared
         expect { daemon.run_once! }.to change(Geo::FileRegistry.lfs_objects, :count).by(-1)
       end
     end
+
+    context 'when replaying a job artifact event' do
+      let(:event_log) { create(:geo_event_log, :job_artifact_deleted_event) }
+      let!(:event_log_state) { create(:geo_event_log_state, event_id: event_log.id - 1) }
+      let(:job_artifact_deleted_event) { event_log.job_artifact_deleted_event }
+      let(:job_artifact) { job_artifact_deleted_event.job_artifact }
+
+      context 'with a tracking database entry' do
+        before do
+          create(:geo_file_registry, :job_artifact, file_id: job_artifact.id)
+        end
+
+        context 'with a file' do
+          context 'when the delete succeeds' do
+            it 'removes the tracking database entry' do
+              expect { daemon.run_once! }.to change(Geo::FileRegistry.job_artifacts, :count).by(-1)
+            end
+
+            it 'deletes the file' do
+              expect { daemon.run_once! }.to change { File.exist?(job_artifact.file.path) }.from(true).to(false)
+            end
+          end
+
+          context 'when the delete fails' do
+            before do
+              expect(daemon).to receive(:delete_file).and_return(false)
+            end
+
+            it 'does not remove the tracking database entry' do
+              expect { daemon.run_once! }.not_to change(Geo::FileRegistry.job_artifacts, :count)
+            end
+          end
+        end
+
+        context 'without a file' do
+          before do
+            FileUtils.rm(job_artifact.file.path)
+          end
+
+          it 'removes the tracking database entry' do
+            expect { daemon.run_once! }.to change(Geo::FileRegistry.job_artifacts, :count).by(-1)
+          end
+        end
+      end
+
+      context 'without a tracking database entry' do
+        it 'does not create a tracking database entry' do
+          expect { daemon.run_once! }.not_to change(Geo::FileRegistry, :count)
+        end
+
+        it 'does not delete the file (yet, due to possible race condition)' do
+          expect { daemon.run_once! }.not_to change { File.exist?(job_artifact.file.path) }.from(true)
+        end
+      end
+    end
+  end
+
+  describe '#delete_file' do
+    context 'when the file exists' do
+      let!(:file) { fixture_file_upload(Rails.root + "spec/fixtures/dk.png", "`/png") }
+
+      context 'when the delete does not raise an exception' do
+        it 'returns true' do
+          expect(daemon.send(:delete_file, file.path)).to be_truthy
+        end
+
+        it 'does not log an error' do
+          expect(daemon).not_to receive(:logger)
+
+          daemon.send(:delete_file, file.path)
+        end
+      end
+
+      context 'when the delete raises an exception' do
+        before do
+          expect(File).to receive(:delete).and_raise('something went wrong')
+        end
+
+        it 'returns false' do
+          expect(daemon.send(:delete_file, file.path)).to be_falsey
+        end
+
+        it 'logs an error' do
+          logger = double(logger)
+          expect(daemon).to receive(:logger).and_return(logger)
+          expect(logger).to receive(:error).with('Failed to remove file', exception: 'RuntimeError', details: 'something went wrong', filename: file.path)
+
+          daemon.send(:delete_file, file.path)
+        end
+      end
+    end
+
+    context 'when the file does not exist' do
+      it 'returns false' do
+        expect(daemon.send(:delete_file, '/does/not/exist')).to be_falsey
+      end
+
+      it 'logs an error' do
+        logger = double(logger)
+        expect(daemon).to receive(:logger).and_return(logger)
+        expect(logger).to receive(:error).with('Failed to remove file', exception: 'Errno::ENOENT', details: 'No such file or directory @ unlink_internal - /does/not/exist', filename: '/does/not/exist')
+
+        daemon.send(:delete_file, '/does/not/exist')
+      end
+    end
   end
 end

spec/ee/spec/models/geo/event_log_spec.rb

@@ -8,6 +8,9 @@ RSpec.describe Geo::EventLog, type: :model do
     it { is_expected.to belong_to(:repository_renamed_event).class_name('Geo::RepositoryRenamedEvent').with_foreign_key('repository_renamed_event_id') }
     it { is_expected.to belong_to(:repository_updated_event).class_name('Geo::RepositoryUpdatedEvent').with_foreign_key('repository_updated_event_id') }
     it { is_expected.to belong_to(:hashed_storage_migrated_event).class_name('Geo::HashedStorageMigratedEvent').with_foreign_key('hashed_storage_migrated_event_id') }
+    it { is_expected.to belong_to(:hashed_storage_attachments_event).class_name('Geo::HashedStorageAttachmentsEvent').with_foreign_key('hashed_storage_attachments_event_id') }
+    it { is_expected.to belong_to(:lfs_object_deleted_event).class_name('Geo::LfsObjectDeletedEvent').with_foreign_key('lfs_object_deleted_event_id') }
+    it { is_expected.to belong_to(:job_artifact_deleted_event).class_name('Geo::JobArtifactDeletedEvent').with_foreign_key('job_artifact_deleted_event_id') }
   end
 
   describe '#event' do
@@ -57,12 +60,26 @@ RSpec.describe Geo::EventLog, type: :model do
       expect(subject.event).to eq hashed_storage_migrated_event
     end
 
+    it 'returns hashed_storage_attachments_event when set' do
+      hashed_storage_attachments_event = build(:geo_hashed_storage_attachments_event)
+      subject.hashed_storage_attachments_event = hashed_storage_attachments_event
+
+      expect(subject.event).to eq hashed_storage_attachments_event
+    end
+
     it 'returns lfs_object_deleted_event when set' do
       lfs_object_deleted_event = build(:geo_lfs_object_deleted_event)
       subject.lfs_object_deleted_event = lfs_object_deleted_event
 
       expect(subject.event).to eq lfs_object_deleted_event
     end
+
+    it 'returns job_artifact_deleted_event when set' do
+      job_artifact_deleted_event = build(:geo_job_artifact_deleted_event)
+      subject.job_artifact_deleted_event = job_artifact_deleted_event
+
+      expect(subject.event).to eq job_artifact_deleted_event
+    end
   end
 
   describe '#project_id' do

spec/ee/spec/models/geo/job_artifact_deleted_event_spec.rb

+require 'spec_helper'
+
+RSpec.describe Geo::JobArtifactDeletedEvent, type: :model do
+  describe 'relationships' do
+    it { is_expected.to belong_to(:job_artifact) }
+  end
+
+  describe 'validations' do
+    it { is_expected.to validate_presence_of(:job_artifact) }
+  end
+end

spec/factories/geo/event_log.rb

@@ -27,6 +27,10 @@ FactoryBot.define do
     trait :lfs_object_deleted_event do
       lfs_object_deleted_event factory: :geo_lfs_object_deleted_event
     end
+
+    trait :job_artifact_deleted_event do
+      job_artifact_deleted_event factory: :geo_job_artifact_deleted_event
+    end
   end
 
   factory :geo_repository_created_event, class: Geo::RepositoryCreatedEvent do
@@ -96,8 +100,22 @@ FactoryBot.define do
     lfs_object { create(:lfs_object, :with_file) }
 
     after(:build, :stub) do |event, _|
+      local_store_path = Pathname.new(LfsObjectUploader.local_store_path)
+      relative_path = Pathname.new(event.lfs_object.file.path).relative_path_from(local_store_path)
+
       event.oid = event.lfs_object.oid
-      event.file_path = event.lfs_object.file.path
+      event.file_path = relative_path
+    end
+  end
+
+  factory :geo_job_artifact_deleted_event, class: Geo::JobArtifactDeletedEvent do
+    job_artifact { create(:ci_job_artifact, :archive) }
+
+    after(:build, :stub) do |event, _|
+      local_store_path = Pathname.new(JobArtifactUploader.local_store_path)
+      relative_path = Pathname.new(event.job_artifact.file.path).relative_path_from(local_store_path)
+
+      event.file_path = relative_path
     end
   end
 end