bosh-azure-storage-cli based blobstore client

lib/cloud_controller/blobstore/client_provider.rb

@@ -1,9 +1,10 @@
 require 'cloud_controller/blobstore/client'
 require 'cloud_controller/blobstore/retryable_client'
 require 'cloud_controller/blobstore/fog/fog_client'
-require 'cloud_controller/blobstore/fog/error_handling_client'
+require 'cloud_controller/blobstore/error_handling_client'
 require 'cloud_controller/blobstore/webdav/dav_client'
 require 'cloud_controller/blobstore/safe_delete_client'
+require 'cloud_controller/blobstore/storage_cli/storage_cli_client'
 require 'google/apis/errors'
 
 module CloudController
@@ -12,6 +13,9 @@ class ClientProvider
       def self.provide(options:, directory_key:, root_dir: nil, resource_type: nil)
         if options[:blobstore_type].blank? || (options[:blobstore_type] == 'fog')
           provide_fog(options, directory_key, root_dir)
+        elsif options[:blobstore_type] == 'storage-cli'
+          # storage-cli is an experimental feature and not yet fully implemented. !!! DO NOT USE IN PRODUCTION !!!
+          provide_storage_cli(options, directory_key, root_dir)
         else
           provide_webdav(options, directory_key, root_dir)
         end
@@ -65,6 +69,22 @@ def provide_webdav(options, directory_key, root_dir)
 
           Client.new(SafeDeleteClient.new(retryable_client, root_dir))
         end
+
+        def provide_storage_cli(options, directory_key, root_dir)
+          raise BlobstoreError.new('connection_config for storage-cli is not provided') unless options[:connection_config]
+
+          client = StorageCliClient.build(connection_config: options.fetch(:connection_config),
+                                          directory_key: directory_key,
+                                          root_dir: root_dir,
+                                          min_size: options[:minimum_size],
+                                          max_size: options[:maximum_size])
+
+          logger = Steno.logger('cc.blobstore.storage_cli_client')
+          errors = [StandardError]
+          retryable_client = RetryableClient.new(client:, errors:, logger:)
+
+          Client.new(SafeDeleteClient.new(retryable_client, root_dir))
+        end
       end
     end
   end

lib/cloud_controller/blobstore/storage_cli/azure_storage_cli_client.rb

@@ -0,0 +1,20 @@
+module CloudController
+  module Blobstore
+    class AzureStorageCliClient < StorageCliClient
+      def cli_path
+        ENV['AZURE_STORAGE_CLI_PATH'] || '/var/vcap/packages/azure-storage-cli/bin/azure-storage-cli'
+      end
+
+      def build_config(connection_config)
+        {
+          account_name: connection_config[:azure_storage_account_name],
+          account_key: connection_config[:azure_storage_access_key],
+          container_name: @directory_key,
+          environment: connection_config[:environment]
+        }.compact
+      end
+
+      CloudController::Blobstore::StorageCliClient.register('AzureRM', AzureStorageCliClient)
+    end
+  end
+end

lib/cloud_controller/blobstore/storage_cli/storage_cli_blob.rb

@@ -0,0 +1,43 @@
+module CloudController
+  module Blobstore
+    class StorageCliBlob < Blob
+      attr_reader :key
+
+      def initialize(key, properties: nil, signed_url: nil)
+        @key = key
+        @signed_url = signed_url if signed_url
+        # Set properties to an empty hash if nil to avoid nil errors
+        @properties = properties || {}
+      end
+
+      def internal_download_url
+        signed_url
+      end
+
+      def public_download_url
+        signed_url
+      end
+
+      def attributes(*keys)
+        @attributes ||= {
+          etag: @properties.fetch('etag', nil),
+          last_modified: @properties.fetch('last_modified', nil),
+          content_length: @properties.fetch('content_length', nil),
+          created_at: @properties.fetch('created_at', nil)
+        }
+
+        return @attributes if keys.empty?
+
+        @attributes.select { |key, _| keys.include? key }
+      end
+
+      private
+
+      def signed_url
+        raise BlobstoreError.new('StorageCliBlob not configured with a signed URL') unless @signed_url
+
+        @signed_url
+      end
+    end
+  end
+end

lib/cloud_controller/blobstore/storage_cli/storage_cli_client.rb

@@ -0,0 +1,221 @@
+require 'open3'
+require 'tempfile'
+require 'tmpdir'
+require 'fileutils'
+require 'cloud_controller/blobstore/base_client'
+require 'cloud_controller/blobstore/storage_cli/storage_cli_blob'
+
+module CloudController
+  module Blobstore
+    class StorageCliClient < BaseClient
+      attr_reader :root_dir, :min_size, :max_size
+
+      @registry = {}
+
+      class << self
+        attr_reader :registry
+
+        def register(provider, klass)
+          registry[provider] = klass
+        end
+
+        def build(connection_config:, directory_key:, root_dir:, min_size: nil, max_size: nil)
+          provider = connection_config[:provider]
+          raise 'Missing connection_config[:provider]' if provider.nil?
+
+          impl_class = registry[provider]
+          raise "No storage CLI client registered for provider #{provider}" unless impl_class
+
+          impl_class.new(connection_config:, directory_key:, root_dir:, min_size:, max_size:)
+        end
+      end
+
+      def initialize(connection_config:, directory_key:, root_dir:, min_size: nil, max_size: nil)
+        @cli_path = cli_path
+        @directory_key = directory_key
+        @root_dir = root_dir
+        @min_size = min_size || 0
+        @max_size = max_size
+        config = build_config(connection_config)
+        @config_file = write_config_file(config)
+        @fork = connection_config.fetch(:fork, false)
+      end
+
+      def local?
+        false
+      end
+
+      def exists?(blobstore_key)
+        key = partitioned_key(blobstore_key)
+        _, status = run_cli('exists', key, allow_exit_code_three: true)
+
+        if status.exitstatus == 0
+          return true
+        elsif status.exitstatus == 3
+          return false
+        end
+
+        false
+      end
+
+      def download_from_blobstore(source_key, destination_path, mode: nil)
+        FileUtils.mkdir_p(File.dirname(destination_path))
+        run_cli('get', partitioned_key(source_key), destination_path)
+
+        File.chmod(mode, destination_path) if mode
+      end
+
+      def cp_to_blobstore(source_path, destination_key)
+        start     = Time.now.utc
+        log_entry = 'cp-skip'
+        size      = -1
+
+        logger.info('cp-start', destination_key: destination_key, source_path: source_path, bucket: @directory_key)
+
+        File.open(source_path) do |file|
+          size = file.size
+          next unless within_limits?(size)
+
+          run_cli('put', source_path, partitioned_key(destination_key))
+          log_entry = 'cp-finish'
+        end
+
+        duration = Time.now.utc - start
+        logger.info(log_entry,
+                    destination_key: destination_key,
+                    duration_seconds: duration,
+                    size: size)
+      end
+
+      def cp_file_between_keys(source_key, destination_key)
+        if @fork
+          run_cli('copy', partitioned_key(source_key), partitioned_key(destination_key))
+        else
+          # Azure CLI doesn't support server-side copy yet, so fallback to local copy
+          Tempfile.create('blob-copy') do |tmp|
+            download_from_blobstore(source_key, tmp.path)
+            cp_to_blobstore(tmp.path, destination_key)
+          end
+        end
+      end
+
+      def delete_all(_=nil)
+        # page_size is currently not considered. Azure SDK / API has a limit of 5000
+        pass unless @fork
+
+        # Currently, storage-cli does not support bulk deletion.
+        run_cli('delete-recursive', @root_dir)
+      end
+
+      def delete_all_in_path(path)
+        pass unless @fork
+
+        # Currently, storage-cli does not support bulk deletion.
+        run_cli('delete-recursive', partitioned_key(path))
+      end
+
+      def delete(key)
+        run_cli('delete', partitioned_key(key))
+      end
+
+      def delete_blob(blob)
+        delete(blob.key)
+      end
+
+      def blob(key)
+        if @fork
+          properties = properties(key)
+          return nil if properties.nil? || properties.empty?
+
+          signed_url = sign_url(partitioned_key(key), verb: 'get', expires_in_seconds: 3600)
+          StorageCliBlob.new(key, properties:, signed_url:)
+        elsif exists?(key)
+          # Azure CLI does not support getting blob properties directly, so fallback to local check
+          signed_url = sign_url(partitioned_key(key), verb: 'get', expires_in_seconds: 3600)
+          StorageCliBlob.new(key, signed_url:)
+        end
+      end
+
+      def files_for(prefix, _ignored_directory_prefixes=[])
+        return nil unless @fork
+
+        files, _status = run_cli('list', prefix)
+        files.split("\n").map(&:strip).reject(&:empty?).map { |file| StorageCliBlob.new(file) }
+      end
+
+      def ensure_bucket_exists
+        return unless @fork
+
+        run_cli('ensure-bucket-exists')
+      end
+
+      private
+
+      def run_cli(command, *args, allow_exit_code_three: false)
+        logger.info("[storage_cli_client] Running storage-cli: #{@cli_path} -c #{@config_file} #{command} #{args.join(' ')}")
+
+        begin
+          stdout, stderr, status = Open3.capture3(@cli_path, '-c', @config_file, command, *args)
+        rescue StandardError => e
+          raise BlobstoreError.new(e.inspect)
+        end
+
+        unless status.success? || (allow_exit_code_three && status.exitstatus == 3)
+          raise "storage-cli #{command} failed with exit code #{status.exitstatus}, output: '#{stdout}', error: '#{stderr}'"
+        end
+
+        [stdout, status]
+      end
+
+      def sign_url(key, verb:, expires_in_seconds:)
+        stdout, _status = run_cli('sign', key, verb.to_s.downcase, "#{expires_in_seconds}s")
+        stdout.strip
+      end
+
+      def properties(key)
+        stdout, _status = run_cli('properties', partitioned_key(key))
+        # stdout is expected to be in JSON format - raise an error if it is nil, empty or something unexpected
+        raise BlobstoreError.new("Properties command returned empty output for key: #{key}") if stdout.nil? || stdout.empty?
+
+        begin
+          properties = Oj.load(stdout)
+        rescue StandardError => e
+          raise BlobstoreError.new("Failed to parse json properties for key: #{key}, error: #{e.message}")
+        end
+
+        properties
+      end
+
+      def cli_path
+        raise NotImplementedError
+      end
+
+      def build_config(connection_config)
+        raise NotImplementedError
+      end
+
+      def write_config_file(config)
+        # TODO: Consider to move the config generation into capi-release
+        config_dir = File.join(tmpdir, 'blobstore-configs')
+        FileUtils.mkdir_p(config_dir)
+
+        config_file_path = File.join(config_dir, "#{@directory_key}.json")
+        File.open(config_file_path, 'w', 0o600) do |f|
+          f.write(Oj.dump(config.transform_keys(&:to_s)))
+        end
+        config_file_path
+      end
+
+      def tmpdir
+        VCAP::CloudController::Config.config.get(:directories, :tmpdir)
+      rescue StandardError
+        # Fallback to a temporary directory if the config is not set (e.g. for cc-deployment-updater
+        Dir.mktmpdir('cc_blobstore')
+      end
+
+      def logger
+        @logger ||= Steno.logger('cc.blobstore.storage_cli_client')
+      end
+    end
+  end
+end

lib/cloud_controller/config_schemas/api_schema.rb

@@ -198,13 +198,15 @@ class ApiSchema < VCAP::Config
             minimum_size: Integer,
             resource_directory_key: String,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },
 
           buildpacks: {
             buildpack_directory_key: String,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },
@@ -214,6 +216,7 @@ class ApiSchema < VCAP::Config
             max_valid_packages_stored: Integer,
             app_package_directory_key: String,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },
@@ -222,6 +225,7 @@ class ApiSchema < VCAP::Config
             droplet_directory_key: String,
             max_staged_droplets_stored: Integer,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },

lib/cloud_controller/config_schemas/clock_schema.rb

@@ -109,13 +109,15 @@ class ClockSchema < VCAP::Config
             minimum_size: Integer,
             resource_directory_key: String,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },
 
           buildpacks: {
             buildpack_directory_key: String,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },
@@ -124,13 +126,15 @@ class ClockSchema < VCAP::Config
             max_package_size: Integer,
             app_package_directory_key: String,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },
 
           droplets: {
             droplet_directory_key: String,
             fog_connection: Hash,
+            optional(:connection_config) => Hash,
             fog_aws_storage_options: Hash,
             fog_gcp_storage_options: Hash
           },