feat(pubsub): Support adhoc debug logging (#32404)

Co-authored-by: Charlotte Yun <yejiyun@google.com>
Co-authored-by: Yoshi Automation Bot <yoshi-automation@google.com>

Author: 3 people

google-cloud-pubsub/README.md

@@ -92,6 +92,43 @@ module GRPC
 end
 ```
 
+### Enabling library level logging
+
+This library includes an opt-in logging mechanism that provides detailed information about high-level operations. These logs are useful for troubleshooting and monitoring the client's behavior. When enabled, logs are tagged with subtags to indicate the operation type.
+
+The following subtags are used:
+
+*   `callback-delivery`: Logs when a message is delivered to the user-provided callback.
+*   `callback-exceptions`: Logs any exceptions raised from the user callback.
+*   `ack-nack`: Logs when a message is acknowledged (`ack`) or negatively acknowledged (`nack`).
+*   `ack-batch`: Logs the reason and size of acknowledgement batches sent to the server.
+*   `publish-batch`: Logs the reason and size of message batches sent to the server for publishing.
+*   `expiry`: Logs when a message's lease expires and it is dropped from client-side lease management.
+*   `subscriber-streams`: Logs key events in the subscriber's streaming connection, such as opening, closing, and errors.
+*   `subscriber-flow-control`: Logs when the subscriber's client-side flow control is paused or resumed.
+
+**WARNING:** These logs may contain message data in plaintext, which could include sensitive information. Ensure you are practicing good data hygiene with your application logs. It is recommended to enable this logging only for debugging purposes and not permanently in production.
+
+To enable these debug logs, you must provide a logger with the `progname` set to `"pubsub"`.
+
+```ruby
+require "google/cloud/pubsub"
+require "logger"
+
+# Create a logger and set the progname to "pubsub" to enable library-level logging
+logger = Logger.new($stdout)
+logger.progname = "pubsub"
+
+# Configure the logger globally
+Google::Cloud.configure.pubsub.logger = logger
+
+# Or provide it directly to the client
+pubsub = Google::Cloud::PubSub.new logger: logger
+```
+
+If the logger's `progname` is not set to `"pubsub"`, these debug logs will be suppressed, even if the logger is provided.
+
+
 ## Supported Ruby Versions
 
 This library is supported on Ruby 3.1+.

google-cloud-pubsub/lib/google-cloud-pubsub.rb

@@ -23,6 +23,7 @@
 require "google/cloud" unless defined? Google::Cloud.new
 require "google/cloud/config"
 require "googleauth"
+require "logger"
 
 module Google
   module Cloud
@@ -142,6 +143,8 @@ def self.pubsub project_id = nil,
     "https://www.googleapis.com/auth/pubsub"
   ]
 
+  default_logger = Logger.new $stdout
+
   config.add_field! :project_id, default_project, match: String, allow_nil: true
   config.add_alias! :project, :project_id
   config.add_field! :credentials, default_creds, match: [String, Hash, Google::Auth::Credentials], allow_nil: true
@@ -153,4 +156,5 @@ def self.pubsub project_id = nil,
   config.add_field! :on_error, nil, match: Proc
   config.add_field! :endpoint, nil, match: String
   config.add_field! :universe_domain, nil, match: String
+  config.add_field! :logger, default_logger, match: Logger, allow_nil: true
 end

google-cloud-pubsub/lib/google/cloud/pubsub.rb

@@ -14,6 +14,7 @@
 
 
 require "google-cloud-pubsub"
+require "google/cloud/pubsub/internal_logger"
 require "google/cloud/pubsub/project"
 require "google/cloud/config"
 require "google/cloud/env"
@@ -70,11 +71,16 @@ module PubSub
       #
       #   * `https://www.googleapis.com/auth/pubsub`
       # @param [Numeric] timeout Default timeout to use in requests. Optional.
+      # @param [String] universe_domain A custom universe domain. Optional.
       # @param [String] endpoint Override of the endpoint host name. Optional.
       #   If the param is nil, uses the default endpoint.
       # @param [String] emulator_host Pub/Sub emulator host. Optional.
       #   If the param is nil, uses the value of the `emulator_host` config.
-      # @param universe_domain [String] A custom universe domain. Optional.
+      # @param [Logger] logger Optional Logger instance for emitting
+      #   library-level debug logs. If not provided, it will default to
+      #   configure.logger, which defaults to Logger.new STDOUT if not set. To
+      #   enable logging, set environment variable GOOGLE_SDK_RUBY_LOGGING_GEMS
+      #   to "all" or a comma separated list of gem names, including "pubsub".
       #
       # @return [Google::Cloud::PubSub::Project]
       #
@@ -92,13 +98,15 @@ def self.new project_id: nil,
                    timeout: nil,
                    universe_domain: nil,
                    endpoint: nil,
-                   emulator_host: nil
+                   emulator_host: nil,
+                   logger: nil
         project_id ||= default_project_id
         scope ||= configure.scope
         timeout ||= configure.timeout
         endpoint ||= configure.endpoint
         universe_domain ||= configure.universe_domain
         emulator_host ||= configure.emulator_host
+        logger ||= configure.logger
 
         if emulator_host
           credentials = :this_channel_is_insecure
@@ -114,10 +122,12 @@ def self.new project_id: nil,
         project_id = project_id.to_s # Always cast to a string
         raise ArgumentError, "project_id is missing" if project_id.empty?
 
+        logger = Google::Cloud::PubSub::InternalLogger.new logger
         service = PubSub::Service.new project_id, credentials,
                                       host: endpoint,
                                       timeout: timeout,
-                                      universe_domain: universe_domain
+                                      universe_domain: universe_domain,
+                                      logger: logger
         PubSub::Project.new service
       end
 

google-cloud-pubsub/lib/google/cloud/pubsub/async_publisher.rb

@@ -157,7 +157,7 @@ def publish data = nil, attributes = nil, ordering_key: nil, **extra_attrs, &cal
             end
             batch_action = batch.add msg, callback
             if batch_action == :full
-              publish_batches!
+              publish_batches! reason: "batch full"
             elsif @published_at.nil?
               # Set initial time to now to start the background counter
               @published_at = Time.now
@@ -180,7 +180,7 @@ def stop
             break if @stopped
 
             @stopped = true
-            publish_batches! stop: true
+            publish_batches! stop: true, reason: "shutdown"
             @cond.signal
             @publish_thread_pool.shutdown
           end
@@ -234,7 +234,7 @@ def stop! timeout = nil
         # @return [AsyncPublisher] returns self so calls can be chained.
         def flush
           synchronize do
-            publish_batches!
+            publish_batches! reason: "manual flush"
             @cond.signal
           end
 
@@ -313,7 +313,7 @@ def run_background
               time_since_first_publish = Time.now - @published_at
               if time_since_first_publish > @interval
                 # interval met, flush the batches...
-                publish_batches!
+                publish_batches! reason: "interval timeout"
                 @cond.wait
               else
                 # still waiting for the interval to publish the batch...
@@ -347,28 +347,28 @@ def stop_publish ordering_key, err
           end
         end
 
-        def publish_batches! stop: nil
+        def publish_batches! stop: nil, reason: "unknown"
           @batches.reject! { |_ordering_key, batch| batch.empty? }
           @batches.each_value do |batch|
             ready = batch.publish! stop: stop
-            publish_batch_async @topic_name, batch if ready
+            publish_batch_async @topic_name, batch, reason: reason if ready
           end
           # Set published_at to nil to wait indefinitely
           @published_at = nil
         end
 
-        def publish_batch_async topic_name, batch
+        def publish_batch_async topic_name, batch, reason: "unknown"
           # TODO: raise unless @publish_thread_pool.running?
           return unless @publish_thread_pool.running?
 
           Concurrent::Promises.future_on(
-            @publish_thread_pool, topic_name, batch
-          ) { |t, b| publish_batch_sync t, b }
+            @publish_thread_pool, topic_name, batch, reason
+          ) { |t, b, r| publish_batch_sync t, b, reason: r }
         end
 
         # rubocop:disable Metrics/AbcSize
 
-        def publish_batch_sync topic_name, batch
+        def publish_batch_sync topic_name, batch, reason: "unknown"
           # The only batch methods that are safe to call from the loop are
           # rebalance! and reset! because they are the only methods that are
           # synchronized.
@@ -379,6 +379,7 @@ def publish_batch_sync topic_name, batch
               grpc = @service.publish topic_name,
                                       items.map(&:msg),
                                       compress: compress && batch.total_message_bytes >= compression_bytes_threshold
+              service.logger.log_batch "publish-batch", reason, "publish", items.count, items.sum(&:bytesize)
               items.zip Array(grpc.message_ids) do |item, id|
                 @flow_controller.release item.bytesize
                 next unless item.callback

google-cloud-pubsub/lib/google/cloud/pubsub/batch_publisher.rb

@@ -35,6 +35,7 @@ module PubSub
       #   end
       #
       class BatchPublisher
+
         ##
         # @private The messages to publish
         attr_reader :messages
@@ -117,10 +118,11 @@ def to_gcloud_messages message_ids
 
         ##
         # @private  Call the publish API with arrays of data and attrs.
-        def publish_batch_messages topic_name, service
+        def publish_batch_messages topic_name, service, reason: "unknown"
           grpc = service.publish topic_name,
                                  messages,
                                  compress: compress && total_message_bytes >= compression_bytes_threshold
+          service.logger.log_batch "publish-batch", reason, "publish", messages.count, @total_message_bytes
           to_gcloud_messages Array(grpc.message_ids)
         end
       end

google-cloud-pubsub/lib/google/cloud/pubsub/internal_logger.rb

@@ -0,0 +1,76 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+require "logger"
+
+require "google/cloud/config"
+
+module Google
+  module Cloud
+    module PubSub
+      ##
+      # @private
+      class InternalLogger
+        LOG_NAME = "pubsub".freeze
+        VALID_LOG_LEVELS = [:debug, :info, :warn, :error, :fatal].freeze
+        private_constant :VALID_LOG_LEVELS, :LOG_NAME
+
+        ##
+        # @private
+        # rubocop:disable Naming/BlockForwarding
+        def log level, subtag, &message_block
+          return unless VALID_LOG_LEVELS.include?(level) && block_given?
+          # Only log if the logger is explicitly tagged for 'pubsub'.
+          return unless @logger && @logger.progname == LOG_NAME
+
+          @logger.public_send(level, "#{LOG_NAME}:#{subtag}", &message_block)
+        end
+        # rubocop:enable Naming/BlockForwarding
+
+        ##
+        # @private
+        def log_batch logger_name, reason, type, num_messages, total_bytes
+          log :info, logger_name do
+            "#{reason} triggered #{type} batch of #{num_messages} messages, a total of #{total_bytes} bytes"
+          end
+        end
+
+        ##
+        # @private
+        def log_ack_nack ack_ids, type
+          ack_ids.each do |ack_id|
+            log :info, "ack-nack" do
+              "message (ackID #{ack_id}) #{type}"
+            end
+          end
+        end
+
+        ##
+        # @private
+        def log_expiry expired
+          expired.each do |ack_id, item|
+            log :info, "expiry" do
+              "message (ID #{item.message_id}, ackID #{ack_id}) has been dropped from leasing due to a timeout"
+            end
+          end
+        end
+
+        private
+
+        def initialize logger
+          @logger = logger || Logger.new(nil)
+        end
+      end
+    end
+  end
+end

google-cloud-pubsub/lib/google/cloud/pubsub/message_listener.rb

@@ -14,6 +14,7 @@
 
 
 require "google/cloud/pubsub/service"
+require "google/cloud/pubsub/subscriber"
 require "google/cloud/pubsub/message_listener/stream"
 require "google/cloud/pubsub/message_listener/timed_unary_buffer"
 require "monitor"

google-cloud-pubsub/lib/google/cloud/pubsub/message_listener/inventory.rb

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 require "monitor"
 
 module Google
@@ -22,9 +21,9 @@ class MessageListener
         ##
         # @private
         class Inventory
-          InventoryItem = Struct.new :bytesize, :pulled_at do
+          InventoryItem = Struct.new :message_id, :bytesize, :pulled_at do
             def self.from rec_msg
-              new rec_msg.to_proto.bytesize, Time.now
+              new rec_msg.message.message_id, rec_msg.to_proto.bytesize, Time.now
             end
           end
 
@@ -70,18 +69,24 @@ def add *rec_msgs
           def remove *ack_ids
             ack_ids.flatten!
             ack_ids.compact!
-            return if ack_ids.empty?
+            return {} if ack_ids.empty?
 
+            removed_items = {}
             synchronize do
-              @inventory.delete_if { |ack_id, _| ack_ids.include? ack_id }
+              removed, keep = @inventory.partition { |ack_id, _| ack_ids.include? ack_id }
+              @inventory = keep.to_h
+              removed_items = removed.to_h
               @wait_cond.broadcast
             end
+            removed_items
           end
 
           def remove_expired!
             synchronize do
               extension_time = Time.new - extension
-              @inventory.delete_if { |_ack_id, item| item.pulled_at < extension_time }
+              expired, keep = @inventory.partition { |_ack_id, item| item.pulled_at < extension_time }
+              @inventory = keep.to_h
+              stream.subscriber.service.logger.log_expiry expired
               @wait_cond.broadcast
             end
           end