port the object storage to CE

9 files changed, 351 insertions, 19 deletions

diff --git a/app/models/appearance.rb b/app/models/appearance.rb
index dcd14c08f3c..2a6406d63c7 100644
--- a/app/models/appearance.rb
+++ b/app/models/appearance.rb
@@ -1,5 +1,7 @@
 class Appearance < ActiveRecord::Base
   include CacheMarkdownField
+  include AfterCommitQueue
+  include ObjectStorage::BackgroundMove
 
   cache_markdown_field :description
   cache_markdown_field :new_project_guidelines
diff --git a/app/models/ci/job_artifact.rb b/app/models/ci/job_artifact.rb
index 2dfd8d4ef58..df57b4f65e3 100644
--- a/app/models/ci/job_artifact.rb
+++ b/app/models/ci/job_artifact.rb
@@ -1,6 +1,7 @@
 module Ci
   class JobArtifact < ActiveRecord::Base
     include AfterCommitQueue
+    include ObjectStorage::BackgroundMove
     extend Gitlab::Ci::Model
 
     belongs_to :project
@@ -8,15 +9,11 @@ module Ci
 
     before_save :set_size, if: :file_changed?
 
-    mount_uploader :file, JobArtifactUploader
+    scope :with_files_stored_locally, -> { where(file_store: [nil, ::JobArtifactUploader::Store::LOCAL]) }
 
-    after_save if: :file_changed?, on: [:create, :update] do
-      run_after_commit do
-        file.schedule_migration_to_object_storage
-      end
-    end
+    mount_uploader :file, JobArtifactUploader
 
-    delegate :open, :exists?, to: :file
+    delegate :exists?, :open, to: :file
 
     enum file_type: {
       archive: 1,
@@ -28,6 +25,10 @@ module Ci
       self.where(project: project).sum(:size)
     end
 
+    def local_store?
+      [nil, ::JobArtifactUploader::Store::LOCAL].include?(self.file_store)
+    end
+
     def set_size
       self.size = file.size
     end
diff --git a/app/models/concerns/avatarable.rb b/app/models/concerns/avatarable.rb
index d35e37935fb..4d40a2c483e 100644
--- a/app/models/concerns/avatarable.rb
+++ b/app/models/concerns/avatarable.rb
@@ -3,6 +3,7 @@ module Avatarable
 
   included do
     prepend ShadowMethods
+    include ObjectStorage::BackgroundMove
 
     validate :avatar_type, if: ->(user) { user.avatar.present? && user.avatar_changed? }
     validates :avatar, file_size: { maximum: 200.kilobytes.to_i }
diff --git a/app/models/lfs_object.rb b/app/models/lfs_object.rb
index 65c157d61ca..04c75d827e0 100644
--- a/app/models/lfs_object.rb
+++ b/app/models/lfs_object.rb
@@ -1,10 +1,12 @@
 class LfsObject < ActiveRecord::Base
-  prepend EE::LfsObject
   include AfterCommitQueue
+  include ObjectStorage::BackgroundMove
 
   has_many :lfs_objects_projects, dependent: :destroy # rubocop:disable Cop/ActiveRecordDependent
   has_many :projects, through: :lfs_objects_projects
 
+  scope :with_files_stored_locally, -> { where(file_store: [nil, LfsObjectUploader::Store::LOCAL]) }
+
   validates :oid, presence: true, uniqueness: true
 
   scope :with_files_stored_locally, -> { where(file_store: [nil, LfsObjectUploader::Store::LOCAL]) }
@@ -21,6 +23,10 @@ class LfsObject < ActiveRecord::Base
     projects.exists?(project.lfs_storage_project.id)
   end
 
+  def local_store?
+    [nil, LfsObjectUploader::Store::LOCAL].include?(self.file_store)
+  end
+
   def self.destroy_unreferenced
     joins("LEFT JOIN lfs_objects_projects ON lfs_objects_projects.lfs_object_id = #{table_name}.id")
         .where(lfs_objects_projects: { id: nil })
diff --git a/app/models/upload.rb b/app/models/upload.rb
index 3aca452616c..cf71a7b76fc 100644
--- a/app/models/upload.rb
+++ b/app/models/upload.rb
@@ -9,6 +9,8 @@ class Upload < ActiveRecord::Base
   validates :model, presence: true
   validates :uploader, presence: true
 
+  scope :with_files_stored_locally, -> { where(store: [nil, ObjectStorage::Store::LOCAL]) }
+
   before_save  :calculate_checksum!, if: :foreground_checksummable?
   after_commit :schedule_checksum,   if: :checksummable?
 
@@ -34,8 +36,8 @@ class Upload < ActiveRecord::Base
     self.checksum = Digest::SHA256.file(absolute_path).hexdigest
   end
 
-  def build_uploader
-    uploader_class.new(model, mount_point, **uploader_context).tap do |uploader|
+  def build_uploader(mounted_as = nil)
+    uploader_class.new(model, mounted_as || mount_point).tap do |uploader|
       uploader.upload = self
       uploader.retrieve_from_store!(identifier)
     end
@@ -52,6 +54,12 @@ class Upload < ActiveRecord::Base
     }.compact
   end
 
+  def local?
+    return true if store.nil?
+
+    store == ObjectStorage::Store::LOCAL
+  end
+
   private
 
   def delete_file!
@@ -62,12 +70,6 @@ class Upload < ActiveRecord::Base
     checksum.nil? && local? && exist?
   end
 
-  def local?
-    return true if store.nil?
-
-    store == ObjectStorage::Store::LOCAL
-  end
-
   def foreground_checksummable?
     checksummable? && size <= CHECKSUM_THRESHOLD
   end
diff --git a/app/uploaders/attachment_uploader.rb b/app/uploaders/attachment_uploader.rb
index cd819dc9bff..11e038f9327 100644
--- a/app/uploaders/attachment_uploader.rb
+++ b/app/uploaders/attachment_uploader.rb
@@ -2,7 +2,6 @@ class AttachmentUploader < GitlabUploader
   include RecordsUploads::Concern
   include ObjectStorage::Concern
   prepend ObjectStorage::Extension::RecordsUploads
-  include UploaderHelper
 
   private
 
diff --git a/app/uploaders/file_uploader.rb b/app/uploaders/file_uploader.rb
index 051f1b19938..0e2da64de6a 100644
--- a/app/uploaders/file_uploader.rb
+++ b/app/uploaders/file_uploader.rb
@@ -15,10 +15,12 @@ class FileUploader < GitlabUploader
   MARKDOWN_PATTERN = %r{\!?\[.*?\]\(/uploads/(?<secret>[0-9a-f]{32})/(?<file>.*?)\)}
   DYNAMIC_PATH_PATTERN = %r{(?<secret>\h{32})/(?<identifier>.*)}
 
-  attr_accessor :model
-
   after :remove, :prune_store_dir
 
+  # FileUploader do not run in a model transaction, so we can simply
+  # enqueue a job after the :store hook.
+  after :store, :schedule_background_upload
+
   def self.root
     File.join(options.storage_path, 'uploads')
   end
diff --git a/app/uploaders/object_storage.rb b/app/uploaders/object_storage.rb
new file mode 100644
index 00000000000..55f07967dfc
--- /dev/null
+++ b/app/uploaders/object_storage.rb
@@ -0,0 +1,314 @@
+require 'fog/aws'
+require 'carrierwave/storage/fog'
+
+#
+# This concern should add object storage support
+# to the GitlabUploader class
+#
+module ObjectStorage
+  RemoteStoreError = Class.new(StandardError)
+  UnknownStoreError = Class.new(StandardError)
+  ObjectStorageUnavailable = Class.new(StandardError)
+
+  module Store
+    LOCAL = 1
+    REMOTE = 2
+  end
+
+  module Extension
+    # this extension is the glue between the ObjectStorage::Concern and RecordsUploads::Concern
+    module RecordsUploads
+      extend ActiveSupport::Concern
+
+      prepended do |base|
+        raise "#{base} must include ObjectStorage::Concern to use extensions." unless base < Concern
+
+        base.include(::RecordsUploads::Concern)
+      end
+
+      def retrieve_from_store!(identifier)
+        paths = store_dirs.map { |store, path| File.join(path, identifier) }
+
+        unless current_upload_satisfies?(paths, model)
+          # the upload we already have isn't right, find the correct one
+          self.upload = uploads.find_by(model: model, path: paths)
+        end
+
+        super
+      end
+
+      def build_upload
+        super.tap do |upload|
+          upload.store = object_store
+        end
+      end
+
+      def upload=(upload)
+        return unless upload
+
+        self.object_store = upload.store
+        super
+      end
+
+      def schedule_background_upload(*args)
+        return unless schedule_background_upload?
+
+        ObjectStorage::BackgroundMoveWorker.perform_async(self.class.name,
+                                                upload.class.to_s,
+                                                mounted_as,
+                                                upload.id)
+      end
+
+      private
+
+      def current_upload_satisfies?(paths, model)
+        return false unless upload
+        return false unless model
+
+        paths.include?(upload.path) &&
+          upload.model_id == model.id &&
+          upload.model_type == model.class.base_class.sti_name
+      end
+    end
+  end
+
+  # Add support for automatic background uploading after the file is stored.
+  #
+  module BackgroundMove
+    extend ActiveSupport::Concern
+
+    def background_upload(mount_points = [])
+      return unless mount_points.any?
+
+      run_after_commit do
+        mount_points.each { |mount| send(mount).schedule_background_upload } # rubocop:disable GitlabSecurity/PublicSend
+      end
+    end
+
+    def changed_mounts
+      self.class.uploaders.select do |mount, uploader_class|
+        mounted_as = uploader_class.serialization_column(self.class, mount)
+        mount if send(:"#{mounted_as}_changed?") # rubocop:disable GitlabSecurity/PublicSend
+      end.keys
+    end
+
+    included do
+      after_save on: [:create, :update] do
+        background_upload(changed_mounts)
+      end
+    end
+  end
+
+  module Concern
+    extend ActiveSupport::Concern
+
+    included do |base|
+      base.include(ObjectStorage)
+
+      before :store, :verify_license!
+      after :migrate, :delete_migrated_file
+    end
+
+    class_methods do
+      def object_store_options
+        options.object_store
+      end
+
+      def object_store_enabled?
+        object_store_options.enabled
+      end
+
+      def background_upload_enabled?
+        object_store_options.background_upload
+      end
+
+      def object_store_credentials
+        object_store_options.connection.to_hash.deep_symbolize_keys
+      end
+
+      def remote_store_path
+        object_store_options.remote_directory
+      end
+
+      def licensed?
+        License.feature_available?(:object_storage)
+      end
+
+      def serialization_column(model_class, mount_point)
+        model_class.uploader_options.dig(mount_point, :mount_on) || mount_point
+      end
+    end
+
+    def file_storage?
+      storage.is_a?(CarrierWave::Storage::File)
+    end
+
+    def file_cache_storage?
+      cache_storage.is_a?(CarrierWave::Storage::File)
+    end
+
+    def object_store
+      @object_store ||= model.try(store_serialization_column) || Store::LOCAL
+    end
+
+    # rubocop:disable Gitlab/ModuleWithInstanceVariables
+    def object_store=(value)
+      @object_store = value || Store::LOCAL
+      @storage = storage_for(object_store)
+    end
+    # rubocop:enable Gitlab/ModuleWithInstanceVariables
+
+    # Return true if the current file is part or the model (i.e. is mounted in the model)
+    #
+    def persist_object_store?
+      model.respond_to?(:"#{store_serialization_column}=")
+    end
+
+    # Save the current @object_store to the model <mounted_as>_store column
+    def persist_object_store!
+      return unless persist_object_store?
+
+      updated = model.update_column(store_serialization_column, object_store)
+      raise ActiveRecordError unless updated
+    end
+
+    def use_file
+      if file_storage?
+        return yield path
+      end
+
+      begin
+        cache_stored_file!
+        yield cache_path
+      ensure
+        cache_storage.delete_dir!(cache_path(nil))
+      end
+    end
+
+    def filename
+      super || file&.filename
+    end
+
+    #
+    # Move the file to another store
+    #
+    #   new_store: Enum (Store::LOCAL, Store::REMOTE)
+    #
+    def migrate!(new_store)
+      return unless object_store != new_store
+      return unless file
+
+      new_file = nil
+      file_to_delete = file
+      from_object_store = object_store
+      self.object_store = new_store # changes the storage and file
+
+      cache_stored_file! if file_storage?
+
+      with_callbacks(:migrate, file_to_delete) do
+        with_callbacks(:store, file_to_delete) do # for #store_versions!
+          new_file = storage.store!(file)
+          persist_object_store!
+          self.file = new_file
+        end
+      end
+
+      file
+    rescue => e
+      # in case of failure delete new file
+      new_file.delete unless new_file.nil?
+      # revert back to the old file
+      self.object_store = from_object_store
+      self.file = file_to_delete
+      raise e
+    end
+
+    def schedule_background_upload(*args)
+      return unless schedule_background_upload?
+
+      ObjectStorage::BackgroundMoveWorker.perform_async(self.class.name,
+                                                          model.class.name,
+                                                          mounted_as,
+                                                          model.id)
+    end
+
+    def fog_directory
+      self.class.remote_store_path
+    end
+
+    def fog_credentials
+      self.class.object_store_credentials
+    end
+
+    def fog_public
+      false
+    end
+
+    def delete_migrated_file(migrated_file)
+      migrated_file.delete if exists?
+    end
+
+    def verify_license!(_file)
+      return if file_storage?
+
+      raise(ObjectStorageUnavailable, 'Object Storage feature is missing') unless self.class.licensed?
+    end
+
+    def exists?
+      file.present?
+    end
+
+    def store_dir(store = nil)
+      store_dirs[store || object_store]
+    end
+
+    def store_dirs
+      {
+        Store::LOCAL => File.join(base_dir, dynamic_segment),
+        Store::REMOTE => File.join(dynamic_segment)
+      }
+    end
+
+    private
+
+    def schedule_background_upload?
+      self.class.object_store_enabled? &&
+        self.class.background_upload_enabled? &&
+        self.class.licensed? &&
+        self.file_storage?
+    end
+
+    # this is a hack around CarrierWave. The #migrate method needs to be
+    # able to force the current file to the migrated file upon success.
+    def file=(file)
+      @file = file # rubocop:disable Gitlab/ModuleWithInstanceVariables
+    end
+
+    def serialization_column
+      self.class.serialization_column(model.class, mounted_as)
+    end
+
+    # Returns the column where the 'store' is saved
+    #   defaults to 'store'
+    def store_serialization_column
+      [serialization_column, 'store'].compact.join('_').to_sym
+    end
+
+    def storage
+      @storage ||= storage_for(object_store)
+    end
+
+    def storage_for(store)
+      case store
+      when Store::REMOTE
+        raise 'Object Storage is not enabled' unless self.class.object_store_enabled?
+
+        CarrierWave::Storage::Fog.new(self)
+      when Store::LOCAL
+        CarrierWave::Storage::File.new(self)
+      else
+        raise UnknownStoreError
+      end
+    end
+  end
+end
diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml
index 28a5e5da037..0a7656f69f0 100644
--- a/app/workers/all_queues.yml
+++ b/app/workers/all_queues.yml
@@ -38,6 +38,9 @@
 - github_importer:github_import_stage_import_pull_requests
 - github_importer:github_import_stage_import_repository
 
+- object_storage:object_storage_background_move
+- object_storage:object_storage_migrate_uploads
+
 - pipeline_cache:expire_job_cache
 - pipeline_cache:expire_pipeline_cache
 - pipeline_creation:create_pipeline
@@ -102,3 +105,5 @@
 - update_user_activity
 - upload_checksum
 - web_hook
+
+