Merge pull request #4106 from tbetbetbe/grpc_ruby_fix_flaky_ruby_interop_test

Grpc ruby fix flaky ruby interop test

src/ruby/lib/grpc/generic/rpc_server.rb

@@ -418,11 +418,11 @@ module GRPC
           an_rpc = @server.request_call(@cq, loop_tag, INFINITE_FUTURE)
           break if (!an_rpc.nil?) && an_rpc.call.nil?
 
-          c = new_active_server_call(an_rpc)
-          unless c.nil?
-            mth = an_rpc.method.to_sym
-            @pool.schedule(c) do |call|
-              rpc_descs[mth].run_server_method(call, rpc_handlers[mth])
+          active_call = new_active_server_call(an_rpc)
+          unless active_call.nil?
+            @pool.schedule(active_call) do |ac|
+              c, mth = ac
+              rpc_descs[mth].run_server_method(c, rpc_handlers[mth])
             end
           end
         rescue Core::CallError, RuntimeError => e
@@ -442,6 +442,7 @@ module GRPC
       # allow the metadata to be accessed from the call
       handle_call_tag = Object.new
       an_rpc.call.metadata = an_rpc.metadata  # attaches md to call for handlers
+      GRPC.logger.debug("call md is #{an_rpc.metadata}")
       connect_md = nil
       unless @connect_md_proc.nil?
         connect_md = @connect_md_proc.call(an_rpc.method, an_rpc.metadata)
@@ -454,9 +455,11 @@ module GRPC
       # Create the ActiveCall
       GRPC.logger.info("deadline is #{an_rpc.deadline}; (now=#{Time.now})")
       rpc_desc = rpc_descs[an_rpc.method.to_sym]
-      ActiveCall.new(an_rpc.call, @cq,
-                     rpc_desc.marshal_proc, rpc_desc.unmarshal_proc(:input),
-                     an_rpc.deadline)
+      c = ActiveCall.new(an_rpc.call, @cq,
+                         rpc_desc.marshal_proc, rpc_desc.unmarshal_proc(:input),
+                         an_rpc.deadline)
+      mth = an_rpc.method.to_sym
+      [c, mth]
     end
 
     protected

src/ruby/pb/test/client.rb

@@ -46,6 +46,7 @@ $LOAD_PATH.unshift(pb_dir) unless $LOAD_PATH.include?(pb_dir)
 $LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
 
 require 'optparse'
+require 'logger'
 
 require 'grpc'
 require 'googleauth'
@@ -59,6 +60,22 @@ require 'signet/ssl_config'
 
 AUTH_ENV = Google::Auth::CredentialsLoader::ENV_VAR
 
+# RubyLogger defines a logger for gRPC based on the standard ruby logger.
+module RubyLogger
+  def logger
+    LOGGER
+  end
+
+  LOGGER = Logger.new(STDOUT)
+  LOGGER.level = Logger::INFO
+end
+
+# GRPC is the general RPC module
+module GRPC
+  # Inject the noop #logger if no module-level logger method has been injected.
+  extend RubyLogger
+end
+
 # AssertionError is use to indicate interop test failures.
 class AssertionError < RuntimeError; end
 

src/ruby/pb/test/server.rb

@@ -45,6 +45,7 @@ $LOAD_PATH.unshift(pb_dir) unless $LOAD_PATH.include?(pb_dir)
 $LOAD_PATH.unshift(this_dir) unless $LOAD_PATH.include?(this_dir)
 
 require 'forwardable'
+require 'logger'
 require 'optparse'
 
 require 'grpc'
@@ -53,6 +54,60 @@ require 'test/proto/empty'
 require 'test/proto/messages'
 require 'test/proto/test_services'
 
+# DebugIsTruncated extends the default Logger to truncate debug messages
+class DebugIsTruncated < Logger
+  def debug(s)
+    super(truncate(s, 1024))
+  end
+
+  # Truncates a given +text+ after a given <tt>length</tt> if +text+ is longer than <tt>length</tt>:
+  #
+  #   'Once upon a time in a world far far away'.truncate(27)
+  #   # => "Once upon a time in a wo..."
+  #
+  # Pass a string or regexp <tt>:separator</tt> to truncate +text+ at a natural break:
+  #
+  #   'Once upon a time in a world far far away'.truncate(27, separator: ' ')
+  #   # => "Once upon a time in a..."
+  #
+  #   'Once upon a time in a world far far away'.truncate(27, separator: /\s/)
+  #   # => "Once upon a time in a..."
+  #
+  # The last characters will be replaced with the <tt>:omission</tt> string (defaults to "...")
+  # for a total length not exceeding <tt>length</tt>:
+  #
+  #   'And they found that many people were sleeping better.'.truncate(25, omission: '... (continued)')
+  #   # => "And they f... (continued)"
+  def truncate(s, truncate_at, options = {})
+    return s unless s.length > truncate_at
+    omission = options[:omission] || '...'
+    with_extra_room = truncate_at - omission.length
+    stop = \
+      if options[:separator]
+        rindex(options[:separator], with_extra_room) || with_extra_room
+      else
+        with_extra_room
+      end
+    "#{s[0, stop]}#{omission}"
+  end
+end
+
+# RubyLogger defines a logger for gRPC based on the standard ruby logger.
+module RubyLogger
+  def logger
+    LOGGER
+  end
+
+  LOGGER = DebugIsTruncated.new(STDOUT)
+  LOGGER.level = Logger::WARN
+end
+
+# GRPC is the general RPC module
+module GRPC
+  # Inject the noop #logger if no module-level logger method has been injected.
+  extend RubyLogger
+end
+
 # loads the certificates by the test server.
 def load_test_certs
   this_dir = File.expand_path(File.dirname(__FILE__))
@@ -113,7 +168,7 @@ class TestTarget < Grpc::Testing::TestService::Service
 
   def streaming_input_call(call)
     sizes = call.each_remote_read.map { |x| x.payload.body.length }
-    sum = sizes.inject { |s, x| s + x }
+    sum = sizes.inject(0) { |s, x| s + x }
     StreamingInputCallResponse.new(aggregated_payload_size: sum)
   end