Add structured logs and Prometheus metrics for blobstore and packer operations

lib/cloud_controller/blobstore/base_client.rb

@@ -5,15 +5,39 @@ module CloudController
   module Blobstore
     class BaseClient
       def cp_r_to_blobstore(source_dir)
-        Find.find(source_dir).each do |path|
-          next unless File.file?(path)
-          next unless within_limits?(File.size(path))
-          next unless File.stat(path).mode.to_s(8)[3..5].to_i(8) >= 0o600
+        counters = new_cp_r_counters
+        traversal_completed = false
 
-          sha1 = Digester.new.digest_path(path)
-          next if exists?(sha1)
+        begin
+          unless File.directory?(source_dir)
+            logger.error('blobstore.cp_r.source_dir_missing', source_dir: source_dir)
+            raise Errno::ENOENT.new(source_dir)
+          end
 
-          cp_to_blobstore(path, sha1)
+          Find.find(source_dir).each do |path|
+            begin
+              stat = File.stat(path)
+            rescue SystemCallError
+              next
+            end
+            next unless stat.file?
+            next unless within_limits?(stat.size)
+            next unless stat.mode.to_s(8)[3..5].to_i(8) >= 0o600
+
+            cp_r_digest_file(path, counters)
+            sha1 = counters[:last_sha1]
+
+            next unless sha1
+
+            exists = cp_r_check_exists(sha1, counters)
+            next if exists.nil? # exists? raised, already logged
+            next if exists
+
+            cp_r_upload_file(path, sha1, counters)
+          end
+          traversal_completed = true
+        ensure
+          log_cp_r_phases(counters, traversal_completed)
         end
       end
 
@@ -23,6 +47,67 @@ def cp_to_blobstore(_, _)
 
       private
 
+      def new_cp_r_counters
+        { digest_duration: 0, exists_duration: 0, upload_duration: 0,
+          digest_count: 0, exists_count: 0, upload_count: 0,
+          upload_attempts: 0, exists_errors: 0, last_sha1: nil }
+      end
+
+      def cp_r_digest_file(path, counters)
+        t = Time.now.utc
+        counters[:digest_count] += 1
+        counters[:last_sha1] = Digester.new.digest_path(path)
+      ensure
+        counters[:digest_duration] += Time.now.utc - t
+      end
+
+      def cp_r_check_exists(sha1, counters)
+        t = Time.now.utc
+        counters[:exists_count] += 1
+        exists?(sha1)
+      rescue StandardError => e
+        counters[:exists_errors] += 1
+        logger.error('blobstore.cp_r.exists.error', sha1: sha1, error: e.message)
+        nil
+      ensure
+        counters[:exists_duration] += Time.now.utc - t
+      end
+
+      def cp_r_upload_file(path, sha1, counters)
+        t = Time.now.utc
+        counters[:upload_attempts] += 1
+        cp_to_blobstore(path, sha1)
+        counters[:upload_count] += 1
+      ensure
+        counters[:upload_duration] += Time.now.utc - t
+      end
+
+      def log_cp_r_phases(ctrs, completed)
+        logger.info('blobstore.cp_r.digest',
+                    files_scanned: ctrs[:digest_count],
+                    duration_seconds: ctrs[:digest_duration],
+                    completed: completed)
+        ec = ctrs[:exists_count]
+        ea = ctrs[:upload_attempts]
+        ee = ctrs[:exists_errors]
+        logger.info('blobstore.cp_r.exists',
+                    files_checked: ec,
+                    duration_seconds: ctrs[:exists_duration],
+                    avg_duration_seconds: ec > 0 ? ctrs[:exists_duration] / ec : 0,
+                    cache_hits: ec - ea - ee,
+                    cache_misses: ea,
+                    cache_hit_rate: (ec - ee) > 0 ? (ec - ea - ee).to_f / (ec - ee) : 0,
+                    process_spawns: ec,
+                    error_count: ee,
+                    completed: completed)
+        logger.info('blobstore.cp_r.upload',
+                    files_to_upload: ea,
+                    files_uploaded: ctrs[:upload_count],
+                    files_failed: ea - ctrs[:upload_count],
+                    duration_seconds: ctrs[:upload_duration],
+                    completed: completed)
+      end
+
       def partitioned_key(key)
         key             = key.to_s.downcase
         partitioned_key = BlobKeyGenerator.full_path_from_key(key)
@@ -33,6 +118,10 @@ def partitioned_key(key)
       def within_limits?(size)
         size >= @min_size && (@max_size.nil? || size <= @max_size)
       end
+
+      def logger
+        @logger ||= Steno.logger('cc.blobstore')
+      end
     end
   end
 end

lib/cloud_controller/blobstore/client_provider.rb

@@ -4,6 +4,8 @@
 require 'cloud_controller/blobstore/local/local_client'
 require 'cloud_controller/blobstore/safe_delete_client'
 require 'cloud_controller/blobstore/storage_cli/storage_cli_client'
+require 'cloud_controller/execution_context'
+require 'cloud_controller/metrics/prometheus_updater'
 
 module CloudController
   module Blobstore
@@ -49,7 +51,8 @@ def provide_storage_cli(options, directory_key, root_dir, resource_type)
             resource_type: resource_type,
             root_dir: root_dir,
             min_size: options[:minimum_size],
-            max_size: options[:maximum_size]
+            max_size: options[:maximum_size],
+            metrics_updater: metrics_updater_if_registered
           )
 
           logger = Steno.logger('cc.blobstore.storage_cli_client')
@@ -58,6 +61,13 @@ def provide_storage_cli(options, directory_key, root_dir, resource_type)
 
           Client.new(SafeDeleteClient.new(retryable_client, root_dir))
         end
+
+        def metrics_updater_if_registered
+          ctx = VCAP::CloudController::ExecutionContext.from_process_type_env
+          return nil unless VCAP::CloudController::Metrics::PrometheusUpdater::STORAGE_CLI_METRICS_EXECUTION_CONTEXTS.include?(ctx)
+
+          CloudController::DependencyLocator.instance.prometheus_updater
+        end
       end
     end
   end

lib/cloud_controller/blobstore/storage_cli/storage_cli_client.rb

@@ -20,7 +20,7 @@ class StorageCliClient < BaseClient
       # Native storage-cli type names supported by CC
       STORAGE_CLI_TYPES = %w[azurebs alioss s3 gcs dav].freeze
 
-      def initialize(directory_key:, resource_type:, root_dir:, min_size: nil, max_size: nil)
+      def initialize(directory_key:, resource_type:, root_dir:, min_size: nil, max_size: nil, metrics_updater: nil)
         raise 'Missing resource_type' if resource_type.nil?
 
         config_file_path = config_path_for(resource_type)
@@ -40,6 +40,7 @@ def initialize(directory_key:, resource_type:, root_dir:, min_size: nil, max_siz
         @root_dir = root_dir
         @min_size = min_size || 0
         @max_size = max_size
+        @metrics_updater = metrics_updater
       end
 
       def fetch_config(resource_type)
@@ -102,23 +103,55 @@ def download_from_blobstore(source_key, destination_path, mode: nil)
       def cp_to_blobstore(source_path, destination_key)
         start     = Time.now.utc
         log_entry = 'cp-skip'
-        size      = -1
+        size      = nil
+        metric_emitted = false
 
         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'
+        begin
+          File.open(source_path) do |file|
+            size = file.size
+            if within_limits?(size)
+              dest_key = partitioned_key(destination_key)
+              begin
+                run_cli('put', source_path, dest_key, blob_size: size)
+              ensure
+                metric_emitted = true
+              end
+              log_entry = 'cp-finish'
+            else
+              safe_emit_metrics('put', -4, Time.now.utc - start, 'cp_to_blobstore skip')
+              metric_emitted = true
+              log_entry = 'cp-skip'
+            end
+          end
+        rescue Interrupt, SignalException
+          log_entry = 'cp-error'
+          safe_emit_metrics('put', -5, Time.now.utc - start, 'signal before CLI') unless metric_emitted
+          raise
+        rescue StandardError
+          log_entry = 'cp-error'
+          safe_emit_metrics('put', -3, Time.now.utc - start, 'cp_to_blobstore') unless metric_emitted
+          raise
+        rescue Exception # rubocop:disable Lint/RescueException
+          log_entry = 'cp-error'
+          raise
+        ensure
+          begin
+            logger.info(log_entry,
+                        destination_key: destination_key,
+                        duration_seconds: Time.now.utc - start,
+                        size: size)
+          rescue StandardError => e
+            logger.warn("cp_to_blobstore ensure log failed: #{e.inspect}")
+          end
         end
+      end
 
-        duration = Time.now.utc - start
-        logger.info(log_entry,
-                    destination_key: destination_key,
-                    duration_seconds: duration,
-                    size: size)
+      def safe_emit_metrics(operation, exit_code, duration, context, blob_size: nil)
+        emit_storage_cli_metrics(operation, exit_code, duration, blob_size: blob_size)
+      rescue StandardError => e
+        logger.warn("emit_storage_cli_metrics failed in #{context}: #{e.inspect}")
       end
 
       def cp_file_between_keys(source_key, destination_key)
@@ -170,25 +203,117 @@ def additional_flags
         flags_string.split
       end
 
-      def run_cli(command, *args, allow_exit_code_three: false)
-        logger.debug("running storage-cli: #{@cli_path} -s #{@storage_type} -c #{@config_file} #{additional_flags.join(' ')} #{command} #{args.join(' ')}")
-
+      def run_cli(command, *args, allow_exit_code_three: false, blob_size: nil)
+        t = Time.now.utc
+        emitted = false
+        capture3_called = false
         begin
-          stdout, stderr, status = Open3.capture3(@cli_path, '-s', @storage_type, '-c', @config_file, *additional_flags, command, *args)
-          stderr.split("\n").each do |line|
-            logger.info("[INFO] storage-cli: #{line}")
-          end
+          flags = additional_flags
+          logger.debug("running storage-cli: #{@cli_path} -s #{@storage_type} -c #{@config_file} #{flags.join(' ')} #{command} #{args.join(' ')}")
+          capture3_called = true
+          stdout, stderr, status = Open3.capture3(@cli_path, '-s', @storage_type, '-c', @config_file, *flags, command, *args)
         rescue StandardError => e
+          duration = Time.now.utc - t
+          # -1: spawn failure (binary missing/permission); -3: setup error before capture3 (e.g. config load)
+          pre_spawn_code = capture3_called ? -1 : -3
+          safe_emit_metrics(command, pre_spawn_code, duration, 'spawn error')
+          emitted = true # set before logger.error so a logging failure can't trigger a second metric
+          logger.error("storage_cli.call.error #{command}#{log_message_arg_suffix(command, args)}",
+                       log_fields(command, args, blob_size: blob_size).merge('duration_seconds' => duration, 'error' => e.inspect))
           raise BlobstoreError.new(e.inspect)
+        ensure
+          unless emitted
+            duration = Time.now.utc - t
+            if status
+              exit_code = resolve_exit_code(status)
+              success = status.success? || (allow_exit_code_three && status.exitstatus == 3)
+              metric_exit_code = allow_exit_code_three && exit_code == 3 ? 0 : exit_code
+              safe_emit_metrics(command, metric_exit_code, duration, 'run_cli', blob_size: blob_size)
+              logger.info("storage_cli.call.finish #{command}#{log_message_arg_suffix(command, args)}",
+                          log_fields(command, args, blob_size: blob_size).merge('duration_seconds' => duration,
+                                                                                'exit_code' => exit_code,
+                                                                                'success' => success))
+            else
+              mid_exec_code = mid_execution_exit_code(capture3_called)
+              safe_emit_metrics(command, mid_exec_code, duration, 'run_cli no status')
+              logger.error("storage_cli.call.error #{command}#{log_message_arg_suffix(command, args)}",
+                           log_fields(command, args, blob_size: blob_size).merge('duration_seconds' => duration,
+                                                                                 'exit_code' => mid_exec_code,
+                                                                                 'error' => $!.inspect)) # rubocop:disable Style/SpecialGlobalVars
+            end
+            true
+          end
         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
+        stderr.split("\n").each { |line| logger.info("[INFO] storage-cli: #{line}") }
+
+        raise "storage-cli #{command} failed with exit code #{exit_code.inspect}, output: '#{stdout}', error: '#{stderr}'" unless success
 
         [stdout, status]
       end
 
+      def log_message_arg_suffix(command, args)
+        case command
+        when 'put', 'get', 'copy'
+          " src=#{File.basename(args[0].to_s)} dest=#{File.basename(args[1].to_s)}"
+        when 'exists', 'delete', 'properties'
+          " key=#{File.basename(args[0].to_s)}"
+        when 'sign', 'sign-internal', 'sign-public'
+          " key=#{File.basename(args[0].to_s)} verb=#{args[1]} expires_in=#{args[2]}"
+        when 'list'
+          " prefix=#{File.basename(args[0].to_s)}"
+        else
+          args.empty? ? '' : " #{args.map { |a| File.basename(a.to_s) }.join(' ')}"
+        end
+      end
+
+      def log_fields(command, args, blob_size: nil)
+        fields = { 'operation' => command, 'bucket' => @directory_key, 'resource_type' => @resource_type }
+        case command
+        when 'put'
+          fields['source_path'] = args[0]
+          fields['destination_key'] = args[1]
+          fields['blob_size'] = blob_size unless blob_size.nil?
+        when 'get'
+          fields['source_key'] = args[0]
+          fields['destination_path'] = args[1]
+        when 'exists', 'delete', 'properties'
+          fields['key'] = args[0]
+        when 'sign', 'sign-internal', 'sign-public'
+          fields['key'] = args[0]
+          fields['verb'] = args[1]
+          fields['expires_in'] = args[2]
+        when 'copy'
+          fields['source_key'] = args[0]
+          fields['destination_key'] = args[1]
+        when 'list'
+          fields['prefix'] = args[0]
+        when 'delete-recursive'
+          fields['path'] = args[0]
+        end
+        fields
+      end
+
+      def resolve_exit_code(status)
+        if status.exitstatus
+          status.exitstatus
+        elsif status.termsig
+          128 + status.termsig
+        else
+          # exit_code=-2: process exited with neither exitstatus nor termsig (platform anomaly).
+          -2
+        end
+      end
+
+      def mid_execution_exit_code(capture3_called)
+        signal = $!.is_a?(Interrupt) || $!.is_a?(SignalException) # rubocop:disable Style/SpecialGlobalVars
+        if capture3_called
+          signal ? -6 : -7
+        else
+          signal ? -5 : -7
+        end
+      end
+
       def sign_url(key, verb:, expires_in_seconds:, use_internal_url: false)
         command =
           if @storage_type == 'dav'
@@ -228,6 +353,12 @@ def tmpdir
       def logger
         @logger ||= Steno.logger('cc.blobstore.storage_cli_client')
       end
+
+      def emit_storage_cli_metrics(operation, exit_code, duration_seconds, blob_size: nil)
+        return if @metrics_updater.nil?
+
+        @metrics_updater.report_storage_cli_metrics(operation:, exit_code:, duration_seconds:, blob_size:)
+      end
     end
   end
 end