[Part3.4][Live-scale] Implement add endpoint. (#841)

* implement add endpoint. * minor fix * remove Wait() * minor fix * move some default timespan to contants * simplify serverids and resolve comments * fix merge error. * rename and minor improve
Azure · Mar 17, 2020 · 1b1901c · 1b1901c
1 parent 53aedfb
commit 1b1901c
Show file tree

Hide file tree

Showing 18 changed files with 558 additions and 302 deletions.
diff --git a/src/Microsoft.Azure.SignalR.AspNet/ServerConnections/ServiceConnectionManager.cs b/src/Microsoft.Azure.SignalR.AspNet/ServerConnections/ServiceConnectionManager.cs
@@ -127,7 +127,7 @@ public Task StopGetServersPing()
             return Task.WhenAll(GetConnections().Select(s => s.StopGetServersPing()));
         }
 
-        public HashSet<string> GlobalServerIds => throw new NotSupportedException();
+        public string ServersTag => throw new NotSupportedException();
 
         public bool HasClients => throw new NotSupportedException();
 

diff --git a/src/Microsoft.Azure.SignalR.AspNet/ServiceOptions.cs b/src/Microsoft.Azure.SignalR.AspNet/ServiceOptions.cs
@@ -42,7 +42,7 @@ public class ServiceOptions : IServiceEndpointOptions
         /// Gets or sets the lifetime of auto-generated access token, which will be used to authenticate with Azure SignalR Service.
         /// Default value is one hour.
         /// </summary>
-        public TimeSpan AccessTokenLifetime { get; set; } = Constants.DefaultAccessTokenLifetime;
+        public TimeSpan AccessTokenLifetime { get; set; } = Constants.Periods.DefaultAccessTokenLifetime;
 
         /// <summary>
         /// Gets or sets the access token generate algorithm, supports <see cref="SecurityAlgorithms.HmacSha256"/> or <see cref="SecurityAlgorithms.HmacSha512"/>

diff --git a/src/Microsoft.Azure.SignalR.Common/Constants.cs b/src/Microsoft.Azure.SignalR.Common/Constants.cs
@@ -21,18 +21,23 @@ public static class Keys
             public static readonly string ConnectionStringSecondaryKeyPrefix = $"{ConnectionStringSecondaryKey}:";
         }
 
-        public const int DefaultShutdownTimeoutInSeconds = 30;
-        public const int DefaultScaleTimeoutInSeconds = 300;
-
         public const string AsrsMigrateFrom = "Asrs-Migrate-From";
         public const string AsrsMigrateTo = "Asrs-Migrate-To";
 
         public const string AsrsUserAgent = "Asrs-User-Agent";
         public const string AsrsInstanceId = "Asrs-Instance-Id";
 
 
-        // Default access token lifetime
-        public static readonly TimeSpan DefaultAccessTokenLifetime = TimeSpan.FromHours(1);
+        public static class Periods
+        {
+            public static readonly TimeSpan DefaultAccessTokenLifetime = TimeSpan.FromHours(1);
+            public static readonly TimeSpan DefaultScaleTimeout = TimeSpan.FromMinutes(5);
+            public static readonly TimeSpan DefaultShutdownTimeout = TimeSpan.FromSeconds(30);
+            public static readonly TimeSpan RemoveFromServiceTimeout = TimeSpan.FromSeconds(5);
+
+            public static readonly TimeSpan DefaultStatusPingInterval = TimeSpan.FromSeconds(10);
+            public static readonly TimeSpan DefaultServersPingInterval = TimeSpan.FromSeconds(5);
+        }
 
         public static class ClaimType
         {

diff --git a/src/Microsoft.Azure.SignalR.Common/Endpoints/ScaleOperation.cs b/src/Microsoft.Azure.SignalR.Common/Endpoints/ScaleOperation.cs
@@ -6,7 +6,6 @@ namespace Microsoft.Azure.SignalR
     internal enum ScaleOperation
     {
         Add,
-        Remove,
-        Rename
+        Remove
     }
 }
diff --git a/src/Microsoft.Azure.SignalR.Common/Endpoints/ServiceEndpointManagerBase.cs b/src/Microsoft.Azure.SignalR.Common/Endpoints/ServiceEndpointManagerBase.cs
@@ -144,9 +144,9 @@ private async Task AddServiceEndpointsAsync(IEnumerable<ServiceEndpoint> endpoin
                 {
                     var hubEndpoints = CreateHubServiceEndpoints(endpoints, true);
 
-                    await Task.WhenAll(hubEndpoints.Select(e => AddHubServiceEndpointAsync(e, cancellationToken)));
+                    await Task.WhenAll(hubEndpoints.SelectMany(h => h.Value.Select(e => AddHubServiceEndpointAsync(e, cancellationToken))));
 
-                    // TODO: update local store for negotiation
+                    UpdateNegotiationEndpointsStore(hubEndpoints, ScaleOperation.Add);
                 }
                 catch (Exception ex)
                 {
@@ -155,6 +155,29 @@ private async Task AddServiceEndpointsAsync(IEnumerable<ServiceEndpoint> endpoin
             }
         }
 
+        private void UpdateNegotiationEndpointsStore(Dictionary<string, List<HubServiceEndpoint>> endpoints, ScaleOperation scaleOperation)
+        {
+            foreach (var hubEndpoint in _endpointsPerHub)
+            {
+                var updatedEndpoints = endpoints[hubEndpoint.Key];
+                if (updatedEndpoints.Count == 0)
+                {
+                    return;
+                }
+                var oldEndpoints = hubEndpoint.Value;
+                var newEndpoints = oldEndpoints.ToList();
+                switch (scaleOperation)
+                {
+                    case ScaleOperation.Add:
+                        newEndpoints.AddRange(updatedEndpoints);
+                        break;
+                    default:
+                        break;
+                }
+                _endpointsPerHub.TryUpdate(hubEndpoint.Key, newEndpoints, oldEndpoints);
+            }
+        }
+
         private async Task RemoveServiceEndpointsAsync(IEnumerable<ServiceEndpoint> endpoints, CancellationToken cancellationToken)
         {
             if (endpoints.Count() > 0)
@@ -165,7 +188,7 @@ private async Task RemoveServiceEndpointsAsync(IEnumerable<ServiceEndpoint> endp
 
                     // TODO: update local store for negotiation
 
-                    await Task.WhenAll(hubEndpoints.Select(e => RemoveHubServiceEndpointAsync(e, cancellationToken)));
+                    await Task.WhenAll(hubEndpoints.SelectMany(h => h.Value.Select(e => RemoveHubServiceEndpointAsync(e, cancellationToken))));
                 }
                 catch (Exception ex)
                 {
@@ -186,15 +209,15 @@ private IReadOnlyList<HubServiceEndpoint> CreateHubServiceEndpoints(string hub,
             return endpoints.Select(e => CreateHubServiceEndpoint(hub, e, needScaleTcs)).ToList();
         }
 
-        private IReadOnlyList<HubServiceEndpoint> CreateHubServiceEndpoints(IEnumerable<ServiceEndpoint> endpoints, bool needScaleTcs)
+        private Dictionary<string, List<HubServiceEndpoint>> CreateHubServiceEndpoints(IEnumerable<ServiceEndpoint> endpoints, bool needScaleTcs)
         {
             var hubEndpoints = new List<HubServiceEndpoint>();
             var hubs = _endpointsPerHub.Keys;
             foreach (var hub in hubs)
             {
                 hubEndpoints.AddRange(CreateHubServiceEndpoints(hub, endpoints, needScaleTcs));
             }
-            return hubEndpoints;
+            return hubEndpoints.GroupBy(k => k.Hub).ToDictionary(k => k.Key, v => v.ToList());
         }
 
         private async Task AddHubServiceEndpointAsync(HubServiceEndpoint endpoint, CancellationToken cancellationToken)

diff --git a/src/Microsoft.Azure.SignalR.Common/Interfaces/IServiceConnectionContainer.cs b/src/Microsoft.Azure.SignalR.Common/Interfaces/IServiceConnectionContainer.cs
@@ -28,7 +28,7 @@ internal interface IServiceConnectionContainer
 
         Task ConnectionInitializedTask { get; }
 
-        HashSet<string> GlobalServerIds { get; }
+        string ServersTag { get; }
 
         bool HasClients { get; }
     }
-Original file line number
+Diff line change
@@ Expand Up / @@ -6,7 +6,6 @@ namespace Microsoft.Azure.SignalR @@
         internal enum ScaleOperation
         {
             Add,
-            Remove,
-            Rename
+            Remove
         }
     }