Add more load balancing logging (#1791)

JamesNK · web-flow · commit a3717a358779 · 2022-07-30T16:28:07.000+08:00
diff --git a/src/Grpc.Net.Client/Balancer/Subchannel.cs b/src/Grpc.Net.Client/Balancer/Subchannel.cs
@@ -19,6 +19,7 @@
 #if SUPPORT_LOAD_BALANCING
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Net;
 using System.Threading;
@@ -56,12 +57,13 @@ public sealed class Subchannel : IDisposable
         /// </summary>
         internal ConnectivityState State => _state;
 
-        private readonly ConnectionManager _manager;
+        internal readonly ConnectionManager _manager;
         private readonly ILogger _logger;
 
         private ConnectContext? _connectContext;
         private ConnectivityState _state;
         private TaskCompletionSource<object?>? _delayInterruptTcs;
+        private int _currentRegistrationId;
 
         /// <summary>
         /// Gets the current connected address.
@@ -101,25 +103,40 @@ public IDisposable OnStateChanged(Action<SubchannelState> callback)
             return registration;
         }
 
+        private string GetNextRegistrationId()
+        {
+            var registrationId = Interlocked.Increment(ref _currentRegistrationId);
+            return Id + "-" + registrationId;
+        }
+
         private sealed class StateChangedRegistration : IDisposable
         {
             private readonly Subchannel _subchannel;
             private readonly Action<SubchannelState> _callback;
 
+            public string RegistrationId { get; }
+
             public StateChangedRegistration(Subchannel subchannel, Action<SubchannelState> callback)
             {
                 _subchannel = subchannel;
                 _callback = callback;
+                RegistrationId = subchannel.GetNextRegistrationId();
+
+                SubchannelLog.StateChangedRegistrationCreated(_subchannel._logger, _subchannel.Id, RegistrationId);
             }
 
             public void Invoke(SubchannelState state)
             {
+                SubchannelLog.ExecutingStateChangedRegistration(_subchannel._logger, _subchannel.Id, RegistrationId);
                 _callback(state);
             }
 
             public void Dispose()
             {
-                _subchannel._stateChangedRegistrations.Remove(this);
+                if (_subchannel._stateChangedRegistrations.Remove(this))
+                {
+                    SubchannelLog.StateChangedRegistrationRemoved(_subchannel._logger, _subchannel.Id, RegistrationId);
+                }
             }
         }
 
@@ -348,6 +365,10 @@ internal void RaiseStateChanged(ConnectivityState state, Status status)
                     registration.Invoke(subchannelState);
                 }
             }
+            else
+            {
+                SubchannelLog.NoStateChangedRegistrations(_logger, Id);
+            }
         }
 
         /// <inheritdocs />
@@ -379,6 +400,11 @@ public IReadOnlyList<BalancerAddress> GetAddresses()
         public void Dispose()
         {
             UpdateConnectivityState(ConnectivityState.Shutdown, "Subchannel disposed.");
+
+            foreach (var registration in _stateChangedRegistrations)
+            {
+                SubchannelLog.StateChangedRegistrationRemoved(_logger, Id, registration.RegistrationId);
+            }
             _stateChangedRegistrations.Clear();
 
             CancelInProgressConnect();
@@ -421,6 +447,21 @@ internal static class SubchannelLog
         private static readonly Action<ILogger, int, ConnectivityState, string, Exception?> _subchannelStateChanged =
             LoggerMessage.Define<int, ConnectivityState, string>(LogLevel.Debug, new EventId(11, "SubchannelStateChanged"), "Subchannel id '{SubchannelId}' state changed to {State}. Detail: '{Detail}'.");
 
+        private static readonly Action<ILogger, int, string, Exception?> _stateChangedRegistrationCreated =
+            LoggerMessage.Define<int, string>(LogLevel.Trace, new EventId(12, "StateChangedRegistrationCreated"), "Subchannel id '{SubchannelId}' state changed registration '{RegistrationId}' created.");
+
+        private static readonly Action<ILogger, int, string, Exception?> _stateChangedRegistrationRemoved =
+            LoggerMessage.Define<int, string>(LogLevel.Trace, new EventId(13, "StateChangedRegistrationRemoved"), "Subchannel id '{SubchannelId}' state changed registration '{RegistrationId}' removed.");
+
+        private static readonly Action<ILogger, int, string, Exception?> _executingStateChangedRegistration =
+            LoggerMessage.Define<int, string>(LogLevel.Trace, new EventId(14, "ExecutingStateChangedRegistration"), "Subchannel id '{SubchannelId}' executing state changed registration '{RegistrationId}'.");
+
+        private static readonly Action<ILogger, int, Exception?> _noStateChangedRegistrations =
+            LoggerMessage.Define<int>(LogLevel.Trace, new EventId(15, "NoStateChangedRegistrations"), "Subchannel id '{SubchannelId}' has no state changed registrations.");
+
+        private static readonly Action<ILogger, int, BalancerAddress, Exception?> _subchannelPreserved =
+            LoggerMessage.Define<int, BalancerAddress>(LogLevel.Trace, new EventId(16, "SubchannelPreserved"), "Subchannel id '{SubchannelId}' matches address '{Address}' and is preserved.");
+
         public static void SubchannelCreated(ILogger logger, int subchannelId, IReadOnlyList<BalancerAddress> addresses)
         {
             if (logger.IsEnabled(LogLevel.Debug))
@@ -479,6 +520,31 @@ public static void SubchannelStateChanged(ILogger logger, int subchannelId, Conn
         {
             _subchannelStateChanged(logger, subchannelId, state, status.Detail, status.DebugException);
         }
+
+        public static void ExecutingStateChangedRegistration(ILogger logger, int subchannelId, string registrationId)
+        {
+            _executingStateChangedRegistration(logger, subchannelId, registrationId, null);
+        }
+
+        public static void NoStateChangedRegistrations(ILogger logger, int subchannelId)
+        {
+            _noStateChangedRegistrations(logger, subchannelId, null);
+        }
+
+        public static void StateChangedRegistrationCreated(ILogger logger, int subchannelId, string registrationId)
+        {
+            _stateChangedRegistrationCreated(logger, subchannelId, registrationId, null);
+        }
+
+        public static void StateChangedRegistrationRemoved(ILogger logger, int subchannelId, string registrationId)
+        {
+            _stateChangedRegistrationRemoved(logger, subchannelId, registrationId, null);
+        }
+
+        public static void SubchannelPreserved(ILogger logger, int subchannelId, BalancerAddress address)
+        {
+            _subchannelPreserved(logger, subchannelId, address, null);
+        }
     }
 }
 #endif
diff --git a/src/Grpc.Net.Client/Balancer/SubchannelsLoadBalancer.cs b/src/Grpc.Net.Client/Balancer/SubchannelsLoadBalancer.cs
@@ -19,6 +19,7 @@
 #if SUPPORT_LOAD_BALANCING
 using System;
 using System.Collections.Generic;
+using System.Globalization;
 using System.Linq;
 using System.Net;
 using Grpc.Core;
@@ -133,15 +134,17 @@ public override void UpdateChannelState(ChannelState state)
                 // Check existing subchannels for a match.
                 var i = FindSubchannelByAddress(currentSubchannels, address);
 
-                AddressSubchannel newOrCurrentSubConnection;
+                AddressSubchannel newOrCurrentSubchannel;
                 if (i != null)
                 {
                     // There is a match so take current subchannel.
-                    newOrCurrentSubConnection = currentSubchannels[i.GetValueOrDefault()];
+                    newOrCurrentSubchannel = currentSubchannels[i.GetValueOrDefault()];
 
                     // Remove from current collection because any subchannels
                     // remaining in this collection at the end will be disposed.
                     currentSubchannels.RemoveAt(i.GetValueOrDefault());
+
+                    SubchannelLog.SubchannelPreserved(_logger, newOrCurrentSubchannel.Subchannel.Id, address);
                 }
                 else
                 {
@@ -150,10 +153,10 @@ public override void UpdateChannelState(ChannelState state)
                     c.OnStateChanged(s => UpdateSubchannelState(c, s));
 
                     newSubchannels.Add(c);
-                    newOrCurrentSubConnection = new AddressSubchannel(c, address);
+                    newOrCurrentSubchannel = new AddressSubchannel(c, address);
                 }
 
-                allUpdatedSubchannels.Add(newOrCurrentSubConnection);
+                allUpdatedSubchannels.Add(newOrCurrentSubchannel);
             }
 
             // Any sub-connections still in this collection are no longer returned by the resolver.
@@ -227,6 +230,7 @@ private void UpdateBalancingState(Status status)
             }
             else
             {
+                SubchannelsLoadBalancerLog.CreatingReadyPicker(_logger, readySubchannels);
                 UpdateChannelState(ConnectivityState.Ready, CreatePicker(readySubchannels));
             }
         }
@@ -298,7 +302,7 @@ protected override void Dispose(bool disposing)
         /// <returns>A subchannel picker.</returns>
         protected abstract SubchannelPicker CreatePicker(IReadOnlyList<Subchannel> readySubchannels);
 
-        private class AddressSubchannel
+        private sealed class AddressSubchannel
         {
             private ConnectivityState _lastKnownState;
 
@@ -340,6 +344,9 @@ internal static class SubchannelsLoadBalancerLog
         private static readonly Action<ILogger, int, ConnectivityState, Exception?> _requestingConnectionForSubchannel =
             LoggerMessage.Define<int, ConnectivityState>(LogLevel.Trace, new EventId(5, "RequestingConnectionForSubchannel"), "Requesting connection for subchannel id '{SubchannelId}' because it is in state {State}.");
 
+        private static readonly Action<ILogger, int, string, Exception?> _creatingReadyPicker =
+            LoggerMessage.Define<int, string>(LogLevel.Trace, new EventId(6, "CreatingReadyPicker"), "Creating ready picker with {SubchannelCount} subchannels: {Subchannels}");
+
         public static void ProcessingSubchannelStateChanged(ILogger logger, int subchannelId, ConnectivityState state, Status status)
         {
             _processingSubchannelStateChanged(logger, subchannelId, state, status.Detail, status.DebugException);
@@ -364,6 +371,14 @@ public static void RequestingConnectionForSubchannel(ILogger logger, int subchan
         {
             _requestingConnectionForSubchannel(logger, subchannelId, state, null);
         }
+
+        public static void CreatingReadyPicker(ILogger logger, List<Subchannel> readySubchannels)
+        {
+            if (logger.IsEnabled(LogLevel.Trace))
+            {
+                _creatingReadyPicker(logger, readySubchannels.Count, string.Join(", ", readySubchannels.Select(s => $"id '{s.Id}' ({string.Join(",", s.GetAddresses())})")), null);
+            }
+        }
     }
 }
 #endif
diff --git a/test/FunctionalTests/Balancer/BalancerHelpers.cs b/test/FunctionalTests/Balancer/BalancerHelpers.cs
@@ -259,7 +259,7 @@ public TestSubchannelTransportFactory(TimeSpan socketPingInterval, TimeSpan? con
             public ISubchannelTransport Create(Subchannel subchannel)
             {
 #if NET5_0_OR_GREATER
-                return new SocketConnectivitySubchannelTransport(subchannel, _socketPingInterval, _connectTimeout, NullLoggerFactory.Instance);
+                return new SocketConnectivitySubchannelTransport(subchannel, _socketPingInterval, _connectTimeout, subchannel._manager.LoggerFactory);
 #else
                 return new PassiveSubchannelTransport(subchannel);
 #endif

Original file line number	Diff line number	Diff line change
`@@ -19,6 +19,7 @@`
`19`	`19`	`#if SUPPORT_LOAD_BALANCING`
`20`	`20`	`using System;`
`21`	`21`	`using System.Collections.Generic;`
	`22`	`+using System.Globalization;`
`22`	`23`	`using System.Linq;`
`23`	`24`	`using System.Net;`
`24`	`25`	`using Grpc.Core;`
`@@ -133,15 +134,17 @@ public override void UpdateChannelState(ChannelState state)`
`133`	`134`	`// Check existing subchannels for a match.`
`134`	`135`	`var i = FindSubchannelByAddress(currentSubchannels, address);`
`135`	`136`
`136`		`- AddressSubchannel newOrCurrentSubConnection;`
	`137`	`+ AddressSubchannel newOrCurrentSubchannel;`
`137`	`138`	`if (i != null)`
`138`	`139`	`{`
`139`	`140`	`// There is a match so take current subchannel.`
`140`		`- newOrCurrentSubConnection = currentSubchannels[i.GetValueOrDefault()];`
	`141`	`+ newOrCurrentSubchannel = currentSubchannels[i.GetValueOrDefault()];`
`141`	`142`
`142`	`143`	`// Remove from current collection because any subchannels`
`143`	`144`	`// remaining in this collection at the end will be disposed.`
`144`	`145`	`currentSubchannels.RemoveAt(i.GetValueOrDefault());`
	`146`	`+`
	`147`	`+ SubchannelLog.SubchannelPreserved(_logger, newOrCurrentSubchannel.Subchannel.Id, address);`
`145`	`148`	`}`
`146`	`149`	`else`
`147`	`150`	`{`
`@@ -150,10 +153,10 @@ public override void UpdateChannelState(ChannelState state)`
`150`	`153`	`c.OnStateChanged(s => UpdateSubchannelState(c, s));`
`151`	`154`
`152`	`155`	`newSubchannels.Add(c);`
`153`		`- newOrCurrentSubConnection = new AddressSubchannel(c, address);`
	`156`	`+ newOrCurrentSubchannel = new AddressSubchannel(c, address);`
`154`	`157`	`}`
`155`	`158`
`156`		`- allUpdatedSubchannels.Add(newOrCurrentSubConnection);`
	`159`	`+ allUpdatedSubchannels.Add(newOrCurrentSubchannel);`
`157`	`160`	`}`
`158`	`161`
`159`	`162`	`// Any sub-connections still in this collection are no longer returned by the resolver.`
`@@ -227,6 +230,7 @@ private void UpdateBalancingState(Status status)`
`227`	`230`	`}`
`228`	`231`	`else`
`229`	`232`	`{`
	`233`	`+ SubchannelsLoadBalancerLog.CreatingReadyPicker(_logger, readySubchannels);`
`230`	`234`	`UpdateChannelState(ConnectivityState.Ready, CreatePicker(readySubchannels));`
`231`	`235`	`}`
`232`	`236`	`}`
`@@ -298,7 +302,7 @@ protected override void Dispose(bool disposing)`
`298`	`302`	`/// <returns>A subchannel picker.</returns>`
`299`	`303`	`protected abstract SubchannelPicker CreatePicker(IReadOnlyList<Subchannel> readySubchannels);`
`300`	`304`
`301`		`- private class AddressSubchannel`
	`305`	`+ private sealed class AddressSubchannel`
`302`	`306`	`{`
`303`	`307`	`private ConnectivityState _lastKnownState;`
`304`	`308`
`@@ -340,6 +344,9 @@ internal static class SubchannelsLoadBalancerLog`
`340`	`344`	`private static readonly Action<ILogger, int, ConnectivityState, Exception?> _requestingConnectionForSubchannel =`
`341`	`345`	`LoggerMessage.Define<int, ConnectivityState>(LogLevel.Trace, new EventId(5, "RequestingConnectionForSubchannel"), "Requesting connection for subchannel id '{SubchannelId}' because it is in state {State}.");`
`342`	`346`
	`347`	`+ private static readonly Action<ILogger, int, string, Exception?> _creatingReadyPicker =`
	`348`	`+ LoggerMessage.Define<int, string>(LogLevel.Trace, new EventId(6, "CreatingReadyPicker"), "Creating ready picker with {SubchannelCount} subchannels: {Subchannels}");`
	`349`	`+`
`343`	`350`	`public static void ProcessingSubchannelStateChanged(ILogger logger, int subchannelId, ConnectivityState state, Status status)`
`344`	`351`	`{`
`345`	`352`	`_processingSubchannelStateChanged(logger, subchannelId, state, status.Detail, status.DebugException);`
`@@ -364,6 +371,14 @@ public static void RequestingConnectionForSubchannel(ILogger logger, int subchan`
`364`	`371`	`{`
`365`	`372`	`_requestingConnectionForSubchannel(logger, subchannelId, state, null);`
`366`	`373`	`}`
	`374`	`+`
	`375`	`+ public static void CreatingReadyPicker(ILogger logger, List<Subchannel> readySubchannels)`
	`376`	`+ {`
	`377`	`+ if (logger.IsEnabled(LogLevel.Trace))`
	`378`	`+ {`
	`379`	`+ _creatingReadyPicker(logger, readySubchannels.Count, string.Join(", ", readySubchannels.Select(s => $"id '{s.Id}' ({string.Join(",", s.GetAddresses())})")), null);`
	`380`	`+ }`
	`381`	`+ }`
`367`	`382`	`}`
`368`	`383`	`}`
`369`	`384`	`#endif`
Original file line number	Diff line number	Diff line change
`@@ -259,7 +259,7 @@ public TestSubchannelTransportFactory(TimeSpan socketPingInterval, TimeSpan? con`
`259`	`259`	`public ISubchannelTransport Create(Subchannel subchannel)`
`260`	`260`	`{`
`261`	`261`	`#if NET5_0_OR_GREATER`
`262`		`- return new SocketConnectivitySubchannelTransport(subchannel, _socketPingInterval, _connectTimeout, NullLoggerFactory.Instance);`
	`262`	`+ return new SocketConnectivitySubchannelTransport(subchannel, _socketPingInterval, _connectTimeout, subchannel._manager.LoggerFactory);`
`263`	`263`	`#else`
`264`	`264`	`return new PassiveSubchannelTransport(subchannel);`
`265`	`265`	`#endif`