Add support for WithStreamingInputSpans (#20)

Falco20019 · web-flow · commit 298c697f1bb0 · 2020-10-30T12:13:33.000+01:00
* Add support for WithStreamingInputSpans
* Fix deprecated fields
* Fix typo
diff --git a/Directory.Build.props b/Directory.Build.props
@@ -2,9 +2,9 @@
   <Import Project="version.props" />
   <PropertyGroup>
     <Authors>Benjamin Krämer</Authors>
-    <PackageIconUrl>https://avatars0.githubusercontent.com/u/15482765</PackageIconUrl>
+    <PackageIcon>opentracing-icon.png</PackageIcon>
     <PackageProjectUrl>https://github.com/opentracing-contrib/csharp-grpc</PackageProjectUrl>
-    <PackageLicenseUrl>https://raw.githubusercontent.com/opentracing-contrib/csharp-grpc/master/LICENSE</PackageLicenseUrl>
+    <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
     <PackageReleaseNotes Condition="'$(Version)' != ''">https://github.com/opentracing-contrib/csharp-grpc/releases/tag/v$(Version)</PackageReleaseNotes>
     <RepositoryType>git</RepositoryType>
     <RepositoryUrl>git://github.com/opentracing-contrib/csharp-grpc</RepositoryUrl>
@@ -16,4 +16,8 @@
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
 
+  <ItemGroup>
+    <None Include="$(SolutionDir)/images/opentracing-icon.png" Pack="true" PackagePath="" />
+  </ItemGroup>
+
 </Project>
diff --git a/README.md b/README.md
@@ -76,7 +76,8 @@ using OpenTracing.Contrib.Grpc;
 A `ServerTracingInterceptor` uses default settings, which you can override by creating it using a `ServerTracingInterceptor.Builder`.
 
 - `WithOperationName(IOperationNameConstructor operationName)`: Define how the operation name is constructed for all spans created for this intercepted server. Default is the name of the RPC method. More details in the `Operation Name` section.
-- `WithStreaming()`: Logs to the server span whenever a message is received. *Note:* This package supports streaming but has not been rigorously tested. If you come across any issues, please let us know.
+- `WithStreaming()`: Logs to the server span whenever a message is is received or a response sent. *Note:* This package supports streaming but has not been rigorously tested. If you come across any issues, please let us know.
+- `WithStreamingInputSpans()`: Creates a child span for each incoming message. This is adviced when using long-running streams as the calls' span is only finished when the connection is closed.
 - `WithVerbosity()`: Logs to the server span additional events, such as message received, headers received and call complete. Default only logs if a call is cancelled.
 - `WithTracedAttributes(params ServerRequestAttribute[] attrs)`: Sets tags on the server span in case you want to track information about the RPC call.
 
@@ -86,6 +87,7 @@ A `ServerTracingInterceptor` uses default settings, which you can override by cr
 ServerTracingInterceptor tracingInterceptor = new ServerTracingInterceptor
     .Builder(tracer)
     .WithStreaming()
+    .WithStreamingInputSpans()
     .WithVerbosity()
     .WithOperationName(new PrefixOperationNameConstructor("Server"))
     .WithTracedAttributes(ServerTracingConfiguration.RequestAttribute.Headers,
@@ -99,6 +101,7 @@ A `ClientTracingInterceptor` also has default settings, which you can override b
 
 - `WithOperationName(IOperationNameConstructor operationName)`: Define how the operation name is constructed for all spans created for this intercepted client. Default is the name of the RPC method. More details in the `Operation Name` section.
 - `WithStreaming()`: Logs to the client span whenever a message is sent or a response is received. *Note:* This package supports streaming but has not been rigorously tested. If you come across any issues, please let us know.
+- `WithStreamingInputSpans()`: Creates a child span for each incoming message. This is adviced when using long-running streams as the calls' span is only finished when the connection is closed.
 - `WithVerbosity()`: Logs to the client span additional events, such as call started, message sent, headers received, response received, and call complete. Default only logs if a call is cancelled.
 - `WithTracedAttributes(params ClientRequestAttribute[] attrs)`: Sets tags on the client span in case you want to track information about the RPC call.
 - `WithWaitForReady()`: Enables WaitForReady on all RPC calls.
@@ -117,6 +120,7 @@ public class CustomOperationNameConstructor : IOperationNameConstructor
 ClientTracingInterceptor tracingInterceptor = new ClientTracingInterceptor
     .Builder(tracer)
     .WithStreaming()
+    .WithStreamingInputSpans()
     .WithVerbosity()
     .WithOperationName(new CustomOperationNameConstructor())
     .WithTracingAttributes(ClientTracingConfiguration.RequestAttribute.AllCallOptions,
diff --git a/images/opentracing-icon.png b/images/opentracing-icon.png
diff --git a/src/OpenTracing.Contrib.Grpc/Configuration/ClientTracingConfiguration.cs b/src/OpenTracing.Contrib.Grpc/Configuration/ClientTracingConfiguration.cs
@@ -26,8 +26,8 @@ internal ClientTracingConfiguration(ITracer tracer) : base(tracer)
             TracedAttributes = new HashSet<RequestAttribute>();
         }
 
-        internal ClientTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool verbose, ISet<RequestAttribute> tracedAttributes, bool waitForReady, CancellationToken fallbackCancellationToken)
-            : base(tracer, operationNameConstructor, streaming, verbose)
+        internal ClientTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool streamingInputSpans, bool verbose, ISet<RequestAttribute> tracedAttributes, bool waitForReady, CancellationToken fallbackCancellationToken)
+            : base(tracer, operationNameConstructor, streaming, streamingInputSpans, verbose)
         {
             TracedAttributes = tracedAttributes ?? new HashSet<RequestAttribute>();
             WaitForReady = waitForReady;
diff --git a/src/OpenTracing.Contrib.Grpc/Configuration/ServerTracingConfiguration.cs b/src/OpenTracing.Contrib.Grpc/Configuration/ServerTracingConfiguration.cs
@@ -20,8 +20,8 @@ internal ServerTracingConfiguration(ITracer tracer) : base(tracer)
             TracedAttributes = new HashSet<RequestAttribute>();
         }
 
-        internal ServerTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool verbose, ISet<RequestAttribute> tracedAttributes)
-            : base(tracer, operationNameConstructor, streaming, verbose)
+        internal ServerTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool streamingInputSpans, bool verbose, ISet<RequestAttribute> tracedAttributes)
+            : base(tracer, operationNameConstructor, streaming, streamingInputSpans, verbose)
         {
             TracedAttributes = tracedAttributes ?? new HashSet<RequestAttribute>();
         }
diff --git a/src/OpenTracing.Contrib.Grpc/Configuration/TracingConfiguration.cs b/src/OpenTracing.Contrib.Grpc/Configuration/TracingConfiguration.cs
@@ -7,13 +7,15 @@ public abstract class TracingConfiguration
         public ITracer Tracer { get; }
         public IOperationNameConstructor OperationNameConstructor { get; }
         public bool Streaming { get; }
+        public bool StreamingInputSpans { get; }
         public bool Verbose { get; }
 
-        protected TracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor = null, bool streaming = false, bool verbose = false)
+        protected TracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor = null, bool streaming = false, bool streamingInputSpans = false, bool verbose = false)
         {
             Tracer = tracer;
             OperationNameConstructor = operationNameConstructor ?? new DefaultOperationNameConstructor();
             Streaming = streaming;
+            StreamingInputSpans = streamingInputSpans;
             Verbose = verbose;
         }
     }
diff --git a/src/OpenTracing.Contrib.Grpc/GrpcTraceLogger.cs b/src/OpenTracing.Contrib.Grpc/GrpcTraceLogger.cs
@@ -12,6 +12,11 @@ internal class GrpcTraceLogger<TRequest, TResponse>
         private readonly ISpan _span;
         private readonly TracingConfiguration _configuration;
 
+        private IScope _scope;
+        private bool _isFinished;
+
+        private ISpan ScopeSpan => _scope?.Span ?? _span;
+
         public GrpcTraceLogger(ISpan span, TracingConfiguration configuration)
         {
             _span = span;
@@ -34,11 +39,27 @@ public void ResponseHeader(Metadata metadata)
             });
         }
 
+        public void BeginScope(string operationName)
+        {
+            if (!(_configuration.StreamingInputSpans || _configuration.Verbose)) return;
+
+            _scope = _configuration.Tracer.BuildSpan(operationName).StartActive(false);
+        }
+
+        public void EndScope()
+        {
+            if (_scope == null || !(_configuration.StreamingInputSpans || _configuration.Verbose)) return;
+
+            _scope.Span.Finish();
+            _scope.Dispose();
+            _scope = null;
+        }
+
         public void Request(TRequest req)
         {
             if (!(_configuration.Streaming || _configuration.Verbose)) return;
 
-            _span.Log(new Dictionary<string, object>
+            ScopeSpan.Log(new Dictionary<string, object>
             {
                 { LogFields.Event, "gRPC request" },
                 { "data", req }
@@ -49,7 +70,7 @@ public void Response(TResponse rsp)
         {
             if (!(_configuration.Streaming || _configuration.Verbose)) return;
 
-            _span.Log(new Dictionary<string, object>
+            ScopeSpan.Log(new Dictionary<string, object>
             {
                 { LogFields.Event, "gRPC response" },
                 { "data", rsp }
@@ -62,7 +83,7 @@ public void FinishSuccess()
             {
                 _span.Log("Call completed");
             }
-            _span.Finish();
+            Finish();
         }
 
         public void FinishException(Exception ex)
@@ -71,8 +92,17 @@ public void FinishException(Exception ex)
             {
                 _span.Log("Call failed");
             }
-            _span.SetException(ex)
-                .Finish();
+            _span.SetException(ex);
+            Finish();
+        }
+
+        private void Finish()
+        {
+            if (_isFinished) return;
+
+            EndScope();
+            _span.Finish();
+            _isFinished = true;
         }
     }
 }
diff --git a/src/OpenTracing.Contrib.Grpc/Handler/InterceptedClientHandler.cs b/src/OpenTracing.Contrib.Grpc/Handler/InterceptedClientHandler.cs
@@ -18,6 +18,7 @@ internal class InterceptedClientHandler<TRequest, TResponse>
         private readonly ClientTracingConfiguration _configuration;
         private readonly ClientInterceptorContext<TRequest, TResponse> _context;
         private readonly GrpcTraceLogger<TRequest, TResponse> _logger;
+        private readonly TracingAsyncStreamReader<TResponse>.StreamActions _streamActions;
 
         public InterceptedClientHandler(ClientTracingConfiguration configuration, ClientInterceptorContext<TRequest, TResponse> context)
         {
@@ -33,6 +34,9 @@ public InterceptedClientHandler(ClientTracingConfiguration configuration, Client
             var span = InitializeSpanWithHeaders();
             _logger = new GrpcTraceLogger<TRequest, TResponse>(span, configuration);
             _configuration.Tracer.Inject(span.Context, BuiltinFormats.HttpHeaders, new MetadataCarrier(_context.Options.Headers));
+
+            var scopeActions = new ScopeActions("new_response", _logger.BeginScope, _logger.EndScope);
+            _streamActions = new TracingAsyncStreamReader<TResponse>.StreamActions(scopeActions, _logger.Response, _logger.FinishSuccess, _logger.FinishException);
         }
 
         private CallOptions ApplyConfigToCallOptions(CallOptions callOptions)
@@ -143,7 +147,7 @@ public AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall(TRequest req
         {
             _logger.Request(request);
             var rspCnt = continuation(request, _context);
-            var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _logger.Response, _logger.FinishSuccess, _logger.FinishException);
+            var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _streamActions);
             var rspHeaderAsync = rspCnt.ResponseHeadersAsync.ContinueWith(LogResponseHeader);
             return new AsyncServerStreamingCall<TResponse>(tracingResponseStream, rspHeaderAsync, rspCnt.GetStatus, rspCnt.GetTrailers, rspCnt.Dispose);
         }
@@ -175,7 +179,7 @@ public AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall(In
         {
             var rspCnt = continuation(_context);
             var tracingRequestStream = new TracingClientStreamWriter<TRequest>(rspCnt.RequestStream, _logger.Request);
-            var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _logger.Response, _logger.FinishSuccess, _logger.FinishException);
+            var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _streamActions);
             var rspHeaderAsync = rspCnt.ResponseHeadersAsync.ContinueWith(LogResponseHeader);
             return new AsyncDuplexStreamingCall<TRequest, TResponse>(tracingRequestStream, tracingResponseStream, rspHeaderAsync, rspCnt.GetStatus, rspCnt.GetTrailers, rspCnt.Dispose);
         }
diff --git a/src/OpenTracing.Contrib.Grpc/Handler/InterceptedServerHandler.cs b/src/OpenTracing.Contrib.Grpc/Handler/InterceptedServerHandler.cs
@@ -16,6 +16,7 @@ internal class InterceptedServerHandler<TRequest, TResponse>
         private readonly ServerTracingConfiguration _configuration;
         private readonly ServerCallContext _context;
         private readonly GrpcTraceLogger<TRequest, TResponse> _logger;
+        private readonly TracingAsyncStreamReader<TRequest>.StreamActions _streamActions;
 
         public InterceptedServerHandler(ServerTracingConfiguration configuration, ServerCallContext context)
         {
@@ -24,6 +25,9 @@ public InterceptedServerHandler(ServerTracingConfiguration configuration, Server
 
             var span = GetSpanFromContext();
             _logger = new GrpcTraceLogger<TRequest, TResponse>(span, configuration);
+
+            var scopeActions = new ScopeActions("new_request", _logger.BeginScope, _logger.EndScope);
+            _streamActions = new TracingAsyncStreamReader<TRequest>.StreamActions(scopeActions, _logger.Request);
         }
 
         private ISpan GetSpanFromContext()
@@ -83,7 +87,7 @@ public async Task<TResponse> ClientStreamingServerHandler(IAsyncStreamReader<TRe
         {
             try
             {
-                var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _logger.Request);
+                var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _streamActions);
                 var response = await continuation(tracingRequestStream, _context).ConfigureAwait(false);
                 _logger.Response(response);
                 _logger.FinishSuccess();
@@ -116,7 +120,7 @@ public async Task DuplexStreamingServerHandler(IAsyncStreamReader<TRequest> requ
         {
             try
             {
-                var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _logger.Request);
+                var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _streamActions);
                 var tracingResponseStream = new TracingServerStreamWriter<TResponse>(responseStream, _logger.Response);
                 await continuation(tracingRequestStream, tracingResponseStream, _context).ConfigureAwait(false);
                 _logger.FinishSuccess();
diff --git a/src/OpenTracing.Contrib.Grpc/Interceptors/ClientTracingInterceptor.cs b/src/OpenTracing.Contrib.Grpc/Interceptors/ClientTracingInterceptor.cs
@@ -60,6 +60,7 @@ public class Builder
             private readonly ITracer _tracer;
             private IOperationNameConstructor _operationNameConstructor;
             private bool _streaming;
+            private bool _streamingInputSpans;
             private bool _verbose;
             private ISet<ClientTracingConfiguration.RequestAttribute> _tracedAttributes;
             private bool _waitForReady;
@@ -88,6 +89,16 @@ public Builder WithStreaming()
                 return this;
             }
 
+            /// <summary>
+            /// Creates a child span for each input message received.
+            /// </summary>
+            /// <returns>this Builder configured to create child spans</returns>
+            public Builder WithStreamingInputSpans()
+            {
+                _streamingInputSpans = true;
+                return this;
+            }
+
             /// <summary>
             /// Logs all request life-cycle events to client spans.
             /// </summary>
@@ -126,7 +137,7 @@ public Builder WithFallbackCancellationToken(CancellationToken cancellationToken
 
             public ClientTracingInterceptor Build()
             {
-                var configuration = new ClientTracingConfiguration(_tracer, _operationNameConstructor, _streaming, _verbose, _tracedAttributes, _waitForReady, _cancellationToken);
+                var configuration = new ClientTracingConfiguration(_tracer, _operationNameConstructor, _streaming, _streamingInputSpans, _verbose, _tracedAttributes, _waitForReady, _cancellationToken);
                 return new ClientTracingInterceptor(configuration);
             }
         }
diff --git a/src/OpenTracing.Contrib.Grpc/Interceptors/ServerTracingInterceptor.cs b/src/OpenTracing.Contrib.Grpc/Interceptors/ServerTracingInterceptor.cs
@@ -53,6 +53,7 @@ public class Builder
             private readonly ITracer _tracer;
             private IOperationNameConstructor _operationNameConstructor;
             private bool _streaming;
+            private bool _streamingInputSpans;
             private bool _verbose;
             private ISet<ServerTracingConfiguration.RequestAttribute> _tracedAttributes;
 
@@ -70,7 +71,7 @@ public Builder WithOperationName(IOperationNameConstructor operationNameConstruc
             }
 
             /// <summary>
-            /// Logs streaming events to client spans.
+            /// Logs streaming events to server spans.
             /// </summary>
             /// <returns>this Builder configured to log streaming events</returns>
             public Builder WithStreaming()
@@ -80,7 +81,17 @@ public Builder WithStreaming()
             }
 
             /// <summary>
-            /// Logs all request life-cycle events to client spans.
+            /// Creates a child span for each input message received.
+            /// </summary>
+            /// <returns>this Builder configured to create child spans</returns>
+            public Builder WithStreamingInputSpans()
+            {
+                _streamingInputSpans = true;
+                return this;
+            }
+
+            /// <summary>
+            /// Logs all request life-cycle events to server spans.
             /// </summary>
             /// <returns>this Builder configured to be verbose</returns>
             public Builder WithVerbosity()
@@ -89,7 +100,7 @@ public Builder WithVerbosity()
                 return this;
             }
 
-            /// <param name="tracedAttributes">to set as tags on client spans created by this intercepter</param>
+            /// <param name="tracedAttributes">to set as tags on server spans created by this intercepter</param>
             /// <returns>this Builder configured to trace attributes</returns>
             public Builder WithTracedAttributes(params ServerTracingConfiguration.RequestAttribute[] tracedAttributes)
             {
@@ -99,7 +110,7 @@ public Builder WithTracedAttributes(params ServerTracingConfiguration.RequestAtt
 
             public ServerTracingInterceptor Build()
             {
-                var configuration = new ServerTracingConfiguration(_tracer, _operationNameConstructor, _streaming, _verbose, _tracedAttributes);
+                var configuration = new ServerTracingConfiguration(_tracer, _operationNameConstructor, _streaming, _streamingInputSpans, _verbose, _tracedAttributes);
                 return new ServerTracingInterceptor(configuration);
             }
         }
diff --git a/src/OpenTracing.Contrib.Grpc/Streaming/ScopeActions.cs b/src/OpenTracing.Contrib.Grpc/Streaming/ScopeActions.cs
@@ -0,0 +1,28 @@
+﻿using System;
+
+namespace OpenTracing.Contrib.Grpc.Streaming
+{
+    internal readonly struct ScopeActions
+    {
+        public string ScopeOperationName { get; }
+        public Action<string> OnBeginScope { get; }
+        public Action OnEndScope { get; }
+
+        public ScopeActions(string scopeOperationName, Action<string> onBeginScope, Action onEndScope)
+        {
+            ScopeOperationName = scopeOperationName;
+            OnBeginScope = onBeginScope;
+            OnEndScope = onEndScope;
+        }
+
+        public void BeginScope()
+        {
+            OnBeginScope?.Invoke(ScopeOperationName);
+        }
+
+        public void EndScope()
+        {
+            OnEndScope?.Invoke();
+        }
+    }
+}
diff --git a/src/OpenTracing.Contrib.Grpc/Streaming/TracingAsyncStreamReader.StreamActions.cs b/src/OpenTracing.Contrib.Grpc/Streaming/TracingAsyncStreamReader.StreamActions.cs
diff --git a/src/OpenTracing.Contrib.Grpc/Streaming/TracingAsyncStreamReader.cs b/src/OpenTracing.Contrib.Grpc/Streaming/TracingAsyncStreamReader.cs

Original file line number	Diff line number	Diff line change
`@@ -26,8 +26,8 @@ internal ClientTracingConfiguration(ITracer tracer) : base(tracer)`
`26`	`26`	`TracedAttributes = new HashSet<RequestAttribute>();`
`27`	`27`	`}`
`28`	`28`
`29`		`- internal ClientTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool verbose, ISet<RequestAttribute> tracedAttributes, bool waitForReady, CancellationToken fallbackCancellationToken)`
`30`		`- : base(tracer, operationNameConstructor, streaming, verbose)`
	`29`	`+ internal ClientTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool streamingInputSpans, bool verbose, ISet<RequestAttribute> tracedAttributes, bool waitForReady, CancellationToken fallbackCancellationToken)`
	`30`	`+ : base(tracer, operationNameConstructor, streaming, streamingInputSpans, verbose)`
`31`	`31`	`{`
`32`	`32`	`TracedAttributes = tracedAttributes ?? new HashSet<RequestAttribute>();`
`33`	`33`	`WaitForReady = waitForReady;`
Original file line number	Diff line number	Diff line change
`@@ -20,8 +20,8 @@ internal ServerTracingConfiguration(ITracer tracer) : base(tracer)`
`20`	`20`	`TracedAttributes = new HashSet<RequestAttribute>();`
`21`	`21`	`}`
`22`	`22`
`23`		`- internal ServerTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool verbose, ISet<RequestAttribute> tracedAttributes)`
`24`		`- : base(tracer, operationNameConstructor, streaming, verbose)`
	`23`	`+ internal ServerTracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor, bool streaming, bool streamingInputSpans, bool verbose, ISet<RequestAttribute> tracedAttributes)`
	`24`	`+ : base(tracer, operationNameConstructor, streaming, streamingInputSpans, verbose)`
`25`	`25`	`{`
`26`	`26`	`TracedAttributes = tracedAttributes ?? new HashSet<RequestAttribute>();`
`27`	`27`	`}`
Original file line number	Diff line number	Diff line change
`@@ -7,13 +7,15 @@ public abstract class TracingConfiguration`
`7`	`7`	`public ITracer Tracer { get; }`
`8`	`8`	`public IOperationNameConstructor OperationNameConstructor { get; }`
`9`	`9`	`public bool Streaming { get; }`
	`10`	`+ public bool StreamingInputSpans { get; }`
`10`	`11`	`public bool Verbose { get; }`
`11`	`12`
`12`		`- protected TracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor = null, bool streaming = false, bool verbose = false)`
	`13`	`+ protected TracingConfiguration(ITracer tracer, IOperationNameConstructor operationNameConstructor = null, bool streaming = false, bool streamingInputSpans = false, bool verbose = false)`
`13`	`14`	`{`
`14`	`15`	`Tracer = tracer;`
`15`	`16`	`OperationNameConstructor = operationNameConstructor ?? new DefaultOperationNameConstructor();`
`16`	`17`	`Streaming = streaming;`
	`18`	`+ StreamingInputSpans = streamingInputSpans;`
`17`	`19`	`Verbose = verbose;`
`18`	`20`	`}`
`19`	`21`	`}`
Original file line number	Diff line number	Diff line change
`@@ -12,6 +12,11 @@ internal class GrpcTraceLogger<TRequest, TResponse>`
`12`	`12`	`private readonly ISpan _span;`
`13`	`13`	`private readonly TracingConfiguration _configuration;`
`14`	`14`
	`15`	`+ private IScope _scope;`
	`16`	`+ private bool _isFinished;`
	`17`	`+`
	`18`	`+ private ISpan ScopeSpan => _scope?.Span ?? _span;`
	`19`	`+`
`15`	`20`	`public GrpcTraceLogger(ISpan span, TracingConfiguration configuration)`
`16`	`21`	`{`
`17`	`22`	`_span = span;`
`@@ -34,11 +39,27 @@ public void ResponseHeader(Metadata metadata)`
`34`	`39`	`});`
`35`	`40`	`}`
`36`	`41`
	`42`	`+ public void BeginScope(string operationName)`
	`43`	`+ {`
	`44`	`+ if (!(_configuration.StreamingInputSpans \|\| _configuration.Verbose)) return;`
	`45`	`+`
	`46`	`+ _scope = _configuration.Tracer.BuildSpan(operationName).StartActive(false);`
	`47`	`+ }`
	`48`	`+`
	`49`	`+ public void EndScope()`
	`50`	`+ {`
	`51`	`+ if (_scope == null \|\| !(_configuration.StreamingInputSpans \|\| _configuration.Verbose)) return;`
	`52`	`+`
	`53`	`+ _scope.Span.Finish();`
	`54`	`+ _scope.Dispose();`
	`55`	`+ _scope = null;`
	`56`	`+ }`
	`57`	`+`
`37`	`58`	`public void Request(TRequest req)`
`38`	`59`	`{`
`39`	`60`	`if (!(_configuration.Streaming \|\| _configuration.Verbose)) return;`
`40`	`61`
`41`		`- _span.Log(new Dictionary<string, object>`
	`62`	`+ ScopeSpan.Log(new Dictionary<string, object>`
`42`	`63`	`{`
`43`	`64`	`{ LogFields.Event, "gRPC request" },`
`44`	`65`	`{ "data", req }`
`@@ -49,7 +70,7 @@ public void Response(TResponse rsp)`
`49`	`70`	`{`
`50`	`71`	`if (!(_configuration.Streaming \|\| _configuration.Verbose)) return;`
`51`	`72`
`52`		`- _span.Log(new Dictionary<string, object>`
	`73`	`+ ScopeSpan.Log(new Dictionary<string, object>`
`53`	`74`	`{`
`54`	`75`	`{ LogFields.Event, "gRPC response" },`
`55`	`76`	`{ "data", rsp }`
`@@ -62,7 +83,7 @@ public void FinishSuccess()`
`62`	`83`	`{`
`63`	`84`	`_span.Log("Call completed");`
`64`	`85`	`}`
`65`		`- _span.Finish();`
	`86`	`+ Finish();`
`66`	`87`	`}`
`67`	`88`
`68`	`89`	`public void FinishException(Exception ex)`
`@@ -71,8 +92,17 @@ public void FinishException(Exception ex)`
`71`	`92`	`{`
`72`	`93`	`_span.Log("Call failed");`
`73`	`94`	`}`
`74`		`- _span.SetException(ex)`
`75`		`- .Finish();`
	`95`	`+ _span.SetException(ex);`
	`96`	`+ Finish();`
	`97`	`+ }`
	`98`	`+`
	`99`	`+ private void Finish()`
	`100`	`+ {`
	`101`	`+ if (_isFinished) return;`
	`102`	`+`
	`103`	`+ EndScope();`
	`104`	`+ _span.Finish();`
	`105`	`+ _isFinished = true;`
`76`	`106`	`}`
`77`	`107`	`}`
`78`	`108`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@ internal class InterceptedClientHandler<TRequest, TResponse>`
`18`	`18`	`private readonly ClientTracingConfiguration _configuration;`
`19`	`19`	`private readonly ClientInterceptorContext<TRequest, TResponse> _context;`
`20`	`20`	`private readonly GrpcTraceLogger<TRequest, TResponse> _logger;`
	`21`	`+ private readonly TracingAsyncStreamReader<TResponse>.StreamActions _streamActions;`
`21`	`22`
`22`	`23`	`public InterceptedClientHandler(ClientTracingConfiguration configuration, ClientInterceptorContext<TRequest, TResponse> context)`
`23`	`24`	`{`
`@@ -33,6 +34,9 @@ public InterceptedClientHandler(ClientTracingConfiguration configuration, Client`
`33`	`34`	`var span = InitializeSpanWithHeaders();`
`34`	`35`	`_logger = new GrpcTraceLogger<TRequest, TResponse>(span, configuration);`
`35`	`36`	`_configuration.Tracer.Inject(span.Context, BuiltinFormats.HttpHeaders, new MetadataCarrier(_context.Options.Headers));`
	`37`	`+`
	`38`	`+ var scopeActions = new ScopeActions("new_response", _logger.BeginScope, _logger.EndScope);`
	`39`	`+ _streamActions = new TracingAsyncStreamReader<TResponse>.StreamActions(scopeActions, _logger.Response, _logger.FinishSuccess, _logger.FinishException);`
`36`	`40`	`}`
`37`	`41`
`38`	`42`	`private CallOptions ApplyConfigToCallOptions(CallOptions callOptions)`
`@@ -143,7 +147,7 @@ public AsyncServerStreamingCall<TResponse> AsyncServerStreamingCall(TRequest req`
`143`	`147`	`{`
`144`	`148`	`_logger.Request(request);`
`145`	`149`	`var rspCnt = continuation(request, _context);`
`146`		`- var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _logger.Response, _logger.FinishSuccess, _logger.FinishException);`
	`150`	`+ var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _streamActions);`
`147`	`151`	`var rspHeaderAsync = rspCnt.ResponseHeadersAsync.ContinueWith(LogResponseHeader);`
`148`	`152`	`return new AsyncServerStreamingCall<TResponse>(tracingResponseStream, rspHeaderAsync, rspCnt.GetStatus, rspCnt.GetTrailers, rspCnt.Dispose);`
`149`	`153`	`}`
`@@ -175,7 +179,7 @@ public AsyncDuplexStreamingCall<TRequest, TResponse> AsyncDuplexStreamingCall(In`
`175`	`179`	`{`
`176`	`180`	`var rspCnt = continuation(_context);`
`177`	`181`	`var tracingRequestStream = new TracingClientStreamWriter<TRequest>(rspCnt.RequestStream, _logger.Request);`
`178`		`- var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _logger.Response, _logger.FinishSuccess, _logger.FinishException);`
	`182`	`+ var tracingResponseStream = new TracingAsyncStreamReader<TResponse>(rspCnt.ResponseStream, _streamActions);`
`179`	`183`	`var rspHeaderAsync = rspCnt.ResponseHeadersAsync.ContinueWith(LogResponseHeader);`
`180`	`184`	`return new AsyncDuplexStreamingCall<TRequest, TResponse>(tracingRequestStream, tracingResponseStream, rspHeaderAsync, rspCnt.GetStatus, rspCnt.GetTrailers, rspCnt.Dispose);`
`181`	`185`	`}`
Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,7 @@ internal class InterceptedServerHandler<TRequest, TResponse>`
`16`	`16`	`private readonly ServerTracingConfiguration _configuration;`
`17`	`17`	`private readonly ServerCallContext _context;`
`18`	`18`	`private readonly GrpcTraceLogger<TRequest, TResponse> _logger;`
	`19`	`+ private readonly TracingAsyncStreamReader<TRequest>.StreamActions _streamActions;`
`19`	`20`
`20`	`21`	`public InterceptedServerHandler(ServerTracingConfiguration configuration, ServerCallContext context)`
`21`	`22`	`{`
`@@ -24,6 +25,9 @@ public InterceptedServerHandler(ServerTracingConfiguration configuration, Server`
`24`	`25`
`25`	`26`	`var span = GetSpanFromContext();`
`26`	`27`	`_logger = new GrpcTraceLogger<TRequest, TResponse>(span, configuration);`
	`28`	`+`
	`29`	`+ var scopeActions = new ScopeActions("new_request", _logger.BeginScope, _logger.EndScope);`
	`30`	`+ _streamActions = new TracingAsyncStreamReader<TRequest>.StreamActions(scopeActions, _logger.Request);`
`27`	`31`	`}`
`28`	`32`
`29`	`33`	`private ISpan GetSpanFromContext()`
`@@ -83,7 +87,7 @@ public async Task<TResponse> ClientStreamingServerHandler(IAsyncStreamReader<TRe`
`83`	`87`	`{`
`84`	`88`	`try`
`85`	`89`	`{`
`86`		`- var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _logger.Request);`
	`90`	`+ var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _streamActions);`
`87`	`91`	`var response = await continuation(tracingRequestStream, _context).ConfigureAwait(false);`
`88`	`92`	`_logger.Response(response);`
`89`	`93`	`_logger.FinishSuccess();`
`@@ -116,7 +120,7 @@ public async Task DuplexStreamingServerHandler(IAsyncStreamReader<TRequest> requ`
`116`	`120`	`{`
`117`	`121`	`try`
`118`	`122`	`{`
`119`		`- var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _logger.Request);`
	`123`	`+ var tracingRequestStream = new TracingAsyncStreamReader<TRequest>(requestStream, _streamActions);`
`120`	`124`	`var tracingResponseStream = new TracingServerStreamWriter<TResponse>(responseStream, _logger.Response);`
`121`	`125`	`await continuation(tracingRequestStream, tracingResponseStream, _context).ConfigureAwait(false);`
`122`	`126`	`_logger.FinishSuccess();`