Fix Gitaly duration timings of BlobService RPCs

For many Gitaly RPCs, previously the `gitaly_duration_s` log timings
only accounted for the initial request/response time. We now measure the
total time it takes to consume the streaming response for the following
RPCs:

1. GetBlob
2. GetLFSPointers
3. GetBlobs
4. GetNewLFSPointers

This came up in https://gitlab.com/gitlab-org/gitlab/-/issues/222247.

Part of https://gitlab.com/gitlab-org/gitlab/-/issues/30334

changelogs/unreleased/sh-fix-gitaly-blob-service-durations.yml

+---
+title: Fix Gitaly duration timings of BlobService RPCs
+merge_request: 34906
+author:
+type: other

lib/gitlab/gitaly_client/blob_service.rb

@@ -15,28 +15,9 @@ module Gitlab
           oid: oid,
           limit: limit
         )
-        response = GitalyClient.call(@gitaly_repo.storage_name, :blob_service, :get_blob, request, timeout: GitalyClient.fast_timeout)
-
-        data = []
-        blob = nil
-        response.each do |msg|
-          if blob.nil?
-            blob = msg
-          end
-
-          data << msg.data
+        GitalyClient.streaming_call(@gitaly_repo.storage_name, :blob_service, :get_blob, request, timeout: GitalyClient.fast_timeout) do |response|
+          consume_blob_response(response)
         end
-
-        return if blob.oid.blank?
-
-        data = data.join
-
-        Gitlab::Git::Blob.new(
-          id: blob.oid,
-          size: blob.size,
-          data: data,
-          binary: Gitlab::Git::Blob.binary?(data)
-        )
       end
 
       def batch_lfs_pointers(blob_ids)
@@ -47,10 +28,10 @@ module Gitlab
           blob_ids: blob_ids
         )
 
-        response = GitalyClient.call(@gitaly_repo.storage_name, :blob_service, :get_lfs_pointers, request, timeout: GitalyClient.medium_timeout)
-
+        GitalyClient.streaming_call(@gitaly_repo.storage_name, :blob_service, :get_lfs_pointers, request, timeout: GitalyClient.medium_timeout) do |response|
           map_lfs_pointers(response)
         end
+      end
 
       def get_blobs(revision_paths, limit = -1)
         return [] if revision_paths.empty?
@@ -65,16 +46,16 @@ module Gitlab
           limit: limit
         )
 
-        response = GitalyClient.call(
+        GitalyClient.streaming_call(
           @gitaly_repo.storage_name,
           :blob_service,
           :get_blobs,
           request,
           timeout: GitalyClient.fast_timeout
-        )
-
+        ) do |response|
           GitalyClient::BlobsStitcher.new(response)
         end
+      end
 
       def get_blob_types(revision_paths, limit = -1)
         return {} if revision_paths.empty?
@@ -89,16 +70,16 @@ module Gitlab
           limit: limit
         )
 
-        response = GitalyClient.call(
+        GitalyClient.streaming_call(
           @gitaly_repo.storage_name,
           :blob_service,
           :get_blobs,
           request,
           timeout: GitalyClient.fast_timeout
-        )
-
+        ) do |response|
           map_blob_types(response)
         end
+      end
 
       def get_new_lfs_pointers(revision, limit, not_in, dynamic_timeout = nil)
         request = Gitaly::GetNewLFSPointersRequest.new(
@@ -120,29 +101,52 @@ module Gitlab
             GitalyClient.medium_timeout
           end
 
-        response = GitalyClient.call(
+        GitalyClient.streaming_call(
           @gitaly_repo.storage_name,
           :blob_service,
           :get_new_lfs_pointers,
           request,
           timeout: timeout
-        )
-
+        ) do |response|
           map_lfs_pointers(response)
         end
+      end
 
       def get_all_lfs_pointers
         request = Gitaly::GetAllLFSPointersRequest.new(
           repository: @gitaly_repo
         )
 
-        response = GitalyClient.call(@gitaly_repo.storage_name, :blob_service, :get_all_lfs_pointers, request, timeout: GitalyClient.medium_timeout)
-
+        GitalyClient.streaming_call(@gitaly_repo.storage_name, :blob_service, :get_all_lfs_pointers, request, timeout: GitalyClient.medium_timeout) do |response|
           map_lfs_pointers(response)
         end
+      end
 
       private
 
+      def consume_blob_response(response)
+        data = []
+        blob = nil
+        response.each do |msg|
+          if blob.nil?
+            blob = msg
+          end
+
+          data << msg.data
+        end
+
+        return if blob.oid.blank?
+
+        data = data.join
+
+        Gitlab::Git::Blob.new(
+          id: blob.oid,
+          size: blob.size,
+          data: data,
+          binary: Gitlab::Git::Blob.binary?(data)
+        )
+      end
+
       def map_lfs_pointers(response)
         response.flat_map do |message|
           message.lfs_pointers.map do |lfs_pointer|