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