Geo - Ensure that upload deletions are communicated to the secondary

app/models/upload.rb

 class Upload < ActiveRecord::Base
+  prepend EE::Upload
+
   # Upper limit for foreground checksum processing
   CHECKSUM_THRESHOLD = 100.megabytes
 

db/migrate/20171130151759_create_geo_upload_deleted_events.rb

+class CreateGeoUploadDeletedEvents < ActiveRecord::Migration
+  DOWNTIME = false
+
+  def change
+    create_table :geo_upload_deleted_events, id: :bigserial do |t|
+      # If an upload is deleted, we need to retain this entry
+      t.references :upload, index: true, foreign_key: false, null: false
+      t.string :file_path, null: false
+      t.integer :model_id, null: false
+      t.string :model_type, null: false
+      t.string :uploader, null: false
+    end
+
+    add_column :geo_event_log, :upload_deleted_event_id, :integer, limit: 8
+  end
+end

db/migrate/20171130152602_add_geo_upload_deleted_events_foreign_key.rb

+class AddGeoUploadDeletedEventsForeignKey < ActiveRecord::Migration
+  include Gitlab::Database::MigrationHelpers
+
+  DOWNTIME = false
+
+  disable_ddl_transaction!
+
+  def up
+    add_concurrent_foreign_key :geo_event_log, :geo_upload_deleted_events,
+                               column: :upload_deleted_event_id, on_delete: :cascade
+  end
+
+  def down
+    remove_foreign_key :geo_event_log, column: :upload_deleted_event_id
+  end
+end

db/schema.rb

@@ -916,6 +916,7 @@ ActiveRecord::Schema.define(version: 20180204200836) do
     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
+    t.integer "upload_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
@@ -1080,6 +1081,16 @@ ActiveRecord::Schema.define(version: 20180204200836) do
   add_index "geo_repository_updated_events", ["project_id"], name: "index_geo_repository_updated_events_on_project_id", using: :btree
   add_index "geo_repository_updated_events", ["source"], name: "index_geo_repository_updated_events_on_source", using: :btree
 
+  create_table "geo_upload_deleted_events", id: :bigserial, force: :cascade do |t|
+    t.integer "upload_id", null: false
+    t.string "file_path", null: false
+    t.integer "model_id", null: false
+    t.string "model_type", null: false
+    t.string "uploader", null: false
+  end
+
+  add_index "geo_upload_deleted_events", ["upload_id"], name: "index_geo_upload_deleted_events_on_upload_id", using: :btree
+
   create_table "gpg_key_subkeys", force: :cascade do |t|
     t.integer "gpg_key_id", null: false
     t.binary "keyid"
@@ -2532,6 +2543,7 @@ ActiveRecord::Schema.define(version: 20180204200836) do
   add_foreign_key "geo_event_log", "geo_repository_deleted_events", column: "repository_deleted_event_id", name: "fk_c4b1c1f66e", on_delete: :cascade
   add_foreign_key "geo_event_log", "geo_repository_renamed_events", column: "repository_renamed_event_id", name: "fk_86c84214ec", on_delete: :cascade
   add_foreign_key "geo_event_log", "geo_repository_updated_events", column: "repository_updated_event_id", on_delete: :cascade
+  add_foreign_key "geo_event_log", "geo_upload_deleted_events", column: "upload_deleted_event_id", name: "fk_c1f241c70d", on_delete: :cascade
   add_foreign_key "geo_hashed_storage_attachments_events", "projects", on_delete: :cascade
   add_foreign_key "geo_hashed_storage_migrated_events", "projects", on_delete: :cascade
   add_foreign_key "geo_node_namespace_links", "geo_nodes", on_delete: :cascade

ee/app/models/ee/upload.rb

+module EE
+  # Upload EE mixin
+  #
+  # This module is intended to encapsulate EE-specific model logic
+  # and be prepended in the `Upload` model
+  module Upload
+    extend ActiveSupport::Concern
+
+    prepended do
+      after_destroy :log_geo_event
+    end
+
+    private
+
+    def log_geo_event
+      ::Geo::UploadDeletedEventStore.new(self).create
+    end
+  end
+end

ee/app/models/geo/event_log.rb

@@ -26,6 +26,10 @@ module Geo
       class_name: 'Geo::HashedStorageMigratedEvent',
       foreign_key: :hashed_storage_migrated_event_id
 
+    belongs_to :hashed_storage_attachments_event,
+      class_name: 'Geo::HashedStorageAttachmentsEvent',
+      foreign_key: :hashed_storage_attachments_event_id
+
     belongs_to :lfs_object_deleted_event,
       class_name: 'Geo::LfsObjectDeletedEvent',
       foreign_key: :lfs_object_deleted_event_id
@@ -34,9 +38,9 @@ module Geo
       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
+    belongs_to :upload_deleted_event,
+      class_name: 'Geo::UploadDeletedEvent',
+      foreign_key: :upload_deleted_event_id
 
     def self.latest_event
       order(id: :desc).first
@@ -49,9 +53,10 @@ module Geo
         repository_renamed_event ||
         repositories_changed_event ||
         hashed_storage_migrated_event ||
+        hashed_storage_attachments_event ||
         lfs_object_deleted_event ||
         job_artifact_deleted_event ||
-        hashed_storage_attachments_event
+        upload_deleted_event
     end
 
     def project_id

ee/app/models/geo/upload_deleted_event.rb

+module Geo
+  class UploadDeletedEvent < ActiveRecord::Base
+    include Geo::Model
+
+    belongs_to :upload
+
+    validates :upload, :file_path, :model_id, :model_type, :uploader, presence: true
+
+    def upload_type
+      uploader&.sub(/Uploader\z/, '')&.underscore
+    end
+  end
+end

ee/app/services/geo/upload_deleted_event_store.rb

+module Geo
+  class UploadDeletedEventStore < EventStore
+    self.event_type = :upload_deleted_event
+
+    attr_reader :upload
+
+    def initialize(upload)
+      @upload = upload
+    end
+
+    def create
+      return unless upload.local?
+
+      super
+    end
+
+    private
+
+    def build_event
+      Geo::UploadDeletedEvent.new(
+        upload: upload,
+        file_path: upload.path,
+        model_id: upload.model_id,
+        model_type: upload.model_type,
+        uploader: upload.uploader
+      )
+    end
+
+    # This is called by ProjectLogHelpers to build json log with context info
+    #
+    # @see ::Gitlab::Geo::ProjectLogHelpers
+    def base_log_data(message)
+      {
+        class: self.class.name,
+        upload_id: upload.id,
+        file_path: upload.path,
+        model_id: upload.model_id,
+        model_type: upload.model_type,
+        uploader: upload.uploader,
+        message: message
+      }
+    end
+  end
+end

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

@@ -236,6 +236,19 @@ module Gitlab
           file_registry_job_artifacts.delete_all
         end
 
+        def handle_upload_deleted_event(event, created_at)
+          logger.event_info(
+            created_at,
+            message: 'Deleted upload file',
+            upload_id: event.upload_id,
+            upload_type: event.upload_type,
+            file_path: event.file_path,
+            model_id: event.model_id,
+            model_type: event.model_type)
+
+          ::Geo::FileRegistry.where(file_id: event.upload_id, file_type: event.upload_type).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)

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

@@ -306,6 +306,25 @@ describe Gitlab::Geo::LogCursor::Daemon, :postgresql, :clean_gitlab_redis_shared
       end
     end
 
+    context 'when replaying a upload deleted event' do
+      context 'with default handling' do
+        let(:event_log) { create(:geo_event_log, :upload_deleted_event) }
+        let!(:event_log_state) { create(:geo_event_log_state, event_id: event_log.id - 1) }
+        let(:upload_deleted_event) { event_log.upload_deleted_event }
+        let(:upload) { upload_deleted_event.upload }
+
+        it 'does not create a tracking database entry' do
+          expect { daemon.run_once! }.not_to change(Geo::FileRegistry, :count)
+        end
+
+        it 'removes the tracking database entry if exist' do
+          create(:geo_file_registry, :avatar, file_id: upload.id)
+
+          expect { daemon.run_once! }.to change(Geo::FileRegistry.attachments, :count).by(-1)
+        end
+      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) }

spec/ee/spec/models/ee/upload_spec.rb

+require 'spec_helper'
+
+describe Upload do
+  describe '#destroy' do
+    subject { create(:upload, checksum: '8710d2c16809c79fee211a9693b64038a8aae99561bc86ce98a9b46b45677fe4') }
+
+    context 'when running in a Geo primary node' do
+      set(:primary) { create(:geo_node, :primary) }
+      set(:secondary) { create(:geo_node) }
+
+      it 'logs an event to the Geo event log' do
+        expect { subject.destroy }.to change(Geo::UploadDeletedEvent, :count).by(1)
+      end
+    end
+  end
+end

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

@@ -80,6 +80,13 @@ RSpec.describe Geo::EventLog, type: :model do
 
       expect(subject.event).to eq job_artifact_deleted_event
     end
+
+    it 'returns upload_deleted_event when set' do
+      upload_deleted_event = build(:geo_upload_deleted_event)
+      subject.upload_deleted_event = upload_deleted_event
+
+      expect(subject.event).to eq upload_deleted_event
+    end
   end
 
   describe '#project_id' do

spec/ee/spec/models/geo/upload_deleted_event_spec.rb

+require 'spec_helper'
+
+RSpec.describe Geo::UploadDeletedEvent, type: :model do
+  describe 'relationships' do
+    it { is_expected.to belong_to(:upload) }
+  end
+
+  describe 'validations' do
+    it { is_expected.to validate_presence_of(:upload) }
+    it { is_expected.to validate_presence_of(:file_path) }
+    it { is_expected.to validate_presence_of(:model_id) }
+    it { is_expected.to validate_presence_of(:model_type) }
+    it { is_expected.to validate_presence_of(:uploader) }
+  end
+
+  describe '#upload_type' do
+    it 'returns nil when uploader is not set' do
+      subject.uploader = nil
+
+      expect(subject.upload_type).to be_nil
+    end
+
+    it 'returns uploader type when uploader is set' do
+      subject.uploader = 'PersonalFileUploader'
+
+      expect(subject.upload_type).to eq 'personal_file'
+    end
+  end
+end

spec/ee/spec/services/geo/upload_deleted_event_store_spec.rb

+require 'spec_helper'
+
+describe Geo::UploadDeletedEventStore do
+  set(:secondary_node) { create(:geo_node) }
+  let(:upload) { create(:upload) }
+
+  subject(:event_store) { described_class.new(upload) }
+
+  describe '#create' do
+    it 'does not create an event when not running on a primary node' do
+      allow(Gitlab::Geo).to receive(:primary?) { false }
+
+      expect { event_store.create }.not_to change(Geo::UploadDeletedEvent, :count)
+    end
+
+    context 'when running on a primary node' do
+      before do
+        allow(Gitlab::Geo).to receive(:primary?) { true }
+      end
+
+      it 'does not create an event when there are no secondary nodes' do
+        allow(Gitlab::Geo).to receive(:secondary_nodes) { [] }
+
+        expect { event_store.create }.not_to change(Geo::UploadDeletedEvent, :count)
+      end
+
+      it 'creates an upload deleted event' do
+        expect { event_store.create }.to change(Geo::UploadDeletedEvent, :count).by(1)
+      end
+
+      it 'does not create an event when the upload does not use local storage' do
+        allow(upload).to receive(:local?).and_return(false)
+
+        expect { event_store.create }.not_to change(Geo::UploadDeletedEvent, :count)
+      end
+
+      it 'tracks upload attributes' do
+        event_store.create
+        event = Geo::UploadDeletedEvent.last
+
+        expect(event).to have_attributes(
+          upload_id: upload.id,
+          file_path: upload.path,
+          model_id: upload.model_id,
+          model_type: upload.model_type,
+          uploader: upload.uploader
+        )
+      end
+    end
+  end
+end

spec/factories/geo/event_log.rb

@@ -31,6 +31,10 @@ FactoryBot.define do
     trait :job_artifact_deleted_event do
       job_artifact_deleted_event factory: :geo_job_artifact_deleted_event
     end
+
+    trait :upload_deleted_event do
+      upload_deleted_event factory: :geo_upload_deleted_event
+    end
   end
 
   factory :geo_repository_created_event, class: Geo::RepositoryCreatedEvent do
@@ -118,4 +122,24 @@ FactoryBot.define do
       event.file_path = relative_path
     end
   end
+
+  factory :geo_upload_deleted_event, class: Geo::UploadDeletedEvent do
+    upload { create(:upload) }
+    file_path { upload.path }
+    model_id { upload.model_id }
+    model_type { upload.model_type }
+    uploader { upload.uploader }
+
+    trait :issuable_upload do
+      upload { create(:upload, :issuable_upload) }
+    end
+
+    trait :personal_snippet do
+      upload { create(:upload, :personal_snippet) }
+    end
+
+    trait :namespace_upload do
+      upload { create(:upload, :namespace_upload) }
+    end
+  end
 end