Add Redis member store for ACA provider and manage stale members

src/Proto.Cluster.AzureContainerApps/Actors/AzureContainerAppsClusterMonitor.cs

@@ -0,0 +1,295 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Azure.ResourceManager.AppContainers;
+using JetBrains.Annotations;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Proto.Cluster.AzureContainerApps.Contracts;
+using Proto.Cluster.AzureContainerApps.Messages;
+using Proto.Cluster.AzureContainerApps.Models;
+using Proto.Cluster.AzureContainerApps.Options;
+using Proto.Cluster.AzureContainerApps.Stores.ResourceTags;
+using Proto.Timers;
+using Proto.Utils;
+
+namespace Proto.Cluster.AzureContainerApps.Actors;
+
+/// <summary>
+/// An actor that periodically updates the member list. 
+/// </summary>
+public class AzureContainerAppsClusterMonitor : IActor
+{
+    private readonly Cluster _cluster;
+    private readonly IArmClientProvider _armClientProvider;
+    private readonly IClusterMemberStore _clusterMemberStore;
+    private readonly IContainerAppMetadataAccessor _containerAppMetadataAccessor;
+    private readonly ISystemClock _systemClock;
+    private readonly IOptions<AzureContainerAppsProviderOptions> _options;
+    private readonly ILogger _logger;
+
+    private string _memberId = default!;
+    private string _host = default!;
+    private int _port;
+    private string _address = default!;
+    private ICollection<string> _kinds = default!;
+    private ContainerAppMetadata _containerAppMetadata = default!;
+    [CanBeNull] private ContainerAppResource _containerApp;
+    [CanBeNull] private CancellationTokenSource _scheduledTask;
+    private bool _stopping;
+
+    /// <summary>
+    /// Initializes a new instance of <see cref="AzureContainerAppsClusterMonitor"/>.
+    /// </summary>
+    public AzureContainerAppsClusterMonitor(
+        Cluster cluster,
+        IArmClientProvider armClientProvider,
+        IClusterMemberStore clusterMemberStore,
+        IContainerAppMetadataAccessor containerAppMetadataAccessor,
+        ISystemClock systemClock,
+        IOptions<AzureContainerAppsProviderOptions> options,
+        ILogger<AzureContainerAppsClusterMonitor> logger
+    )
+    {
+        _cluster = cluster;
+        _armClientProvider = armClientProvider;
+        _clusterMemberStore = clusterMemberStore;
+        _containerAppMetadataAccessor = containerAppMetadataAccessor;
+        _systemClock = systemClock;
+        _options = options;
+        _logger = logger;
+    }
+
+    private string ClusterName => _cluster.Config.ClusterName;
+
+    /// <inheritdoc />
+    public Task ReceiveAsync(IContext context)
+    {
+        var cancellationToken = context.CancellationToken;
+
+        var task = context.Message switch
+        {
+            Started => OnStarted(cancellationToken),
+            RegisterMember command => OnRegisterMember(context, command),
+            UpdateMembers => OnUpdateMembers(context),
+            UnregisterMember => OnUnregisterMember(),
+            _ => Task.CompletedTask
+        };
+
+        return task;
+    }
+
+    private async Task OnStarted(CancellationToken cancellationToken)
+    {
+        _containerAppMetadata = await _containerAppMetadataAccessor.GetMetadataAsync(cancellationToken).ConfigureAwait(false);
+    }
+
+    private Task OnRegisterMember(IContext context, RegisterMember command)
+    {
+        // Store the member ID.
+        _memberId = command.MemberId;
+        _host = command.Host;
+        _port = command.Port;
+        _kinds = command.Kinds;
+        _address = $"{_host}:{_port}";
+
+        var registerMemberTask = RegisterMemberInternal();
+
+        // Reenter after the member has been registered.
+        context.ReenterAfter(registerMemberTask, _ =>
+        {
+            // Schedule the first update.
+            ScheduleUpdate(context);
+        });
+
+        return Task.CompletedTask;
+    }
+
+    private async Task OnUnregisterMember()
+    {
+        await Retry.Try(UnregisterMemberInternal, onError: OnError, onFailed: OnFailed).ConfigureAwait(false);
+        void OnError(int attempt, Exception exception) => _logger.LogWarning(exception, "Failed to unregister member");
+        void OnFailed(Exception exception) => _logger.LogError(exception, "Failed to unregister member");
+    }
+
+    private Task OnUpdateMembers(IContext context)
+    {
+        if (_stopping)
+            return Task.CompletedTask;
+
+        var updateMembersTask = UpdateMembersAsync();
+        context.ReenterAfter(updateMembersTask, _ =>
+        {
+            // Schedule the next update.
+            ScheduleUpdate(context);
+        });
+
+        return Task.CompletedTask;
+    }
+
+    private async Task RegisterMemberInternal()
+    {
+        // Register the member in the store.
+        await AddMember().ConfigureAwait(false);
+    }
+
+    private async Task UpdateMembersAsync()
+    {
+        var storeName = _clusterMemberStore.GetType().Name;
+        _logger.LogInformation("Looking for members in {Store}", storeName);
+
+        var now = _systemClock.UtcNow;
+
+        try
+        {
+            var ttl = _options.Value.PollInterval + _options.Value.MemberTimeToLive;
+            var storedMembers = await GetMembersFromStore();
+            var activeMembers = storedMembers.Where(m => m.UpdatedAt + ttl >= now).ToList();
+            var expiredMembers = storedMembers.Where(m => m.UpdatedAt + ttl < now).ToList();
+
+            LogStoredMembers(storedMembers);
+            await UpdateCurrentMember(activeMembers, now);
+            await RemoveExpiredMembers(expiredMembers);
+
+            UpdateClusterTopology(activeMembers);
+        }
+        catch (Exception x)
+        {
+            _logger.LogError(x, "Failed to get members from {Store}", storeName);
+        }
+    }
+
+    private async Task<IList<StoredMember>> GetMembersFromStore()
+    {
+        return (await _clusterMemberStore.ListAsync().ConfigureAwait(false)).ToList();
+    }
+
+    private void LogStoredMembers(ICollection<StoredMember> storedMembers)
+    {
+        if (storedMembers.Any())
+            _logger.LogInformation("Got members {Members}", storedMembers.Count);
+        else
+            _logger.LogWarning("Did not get any members from {Store}", _clusterMemberStore.GetType().Name);
+    }
+
+    private async Task UpdateCurrentMember(ICollection<StoredMember> activeMembers, DateTimeOffset now)
+    {
+        var currentMember = activeMembers.FirstOrDefault(m => m.Id == _memberId);
+
+        if (currentMember is not null)
+        {
+            var canReceiveTraffic = await CanReceiveTrafficAsync().ConfigureAwait(false);
+            var revisionName = _containerAppMetadata.RevisionName;
+
+            if (!canReceiveTraffic)
+            {
+                _logger.LogInformation("Revision {RevisionName} is not active", revisionName);
+                activeMembers.Remove(currentMember);
+            }
+            else
+            {
+                _logger.LogInformation("Updating current member {MemberId} on {MemberAddress}", currentMember.Id, currentMember.Address);
+                currentMember = currentMember with { UpdatedAt = now };
+                await _clusterMemberStore.UpdateAsync(currentMember).ConfigureAwait(false);
+            }
+        }
+    }
+
+    private async Task RemoveExpiredMembers(IList<StoredMember> expiredMembers)
+    {
+        if (expiredMembers.Any())
+        {
+            _logger.LogInformation("Removing {Members} expired members", expiredMembers.Count);
+
+            foreach (var expiredMember in expiredMembers)
+            {
+                _logger.LogInformation("Expired member {MemberId} on {MemberAddress}", expiredMember.Id, expiredMember.Address);
+                await _clusterMemberStore.UnregisterAsync(expiredMember.Id).ConfigureAwait(false);
+            }
+        }
+    }
+
+    private void UpdateClusterTopology(IEnumerable<StoredMember> activeMembers)
+    {
+        var members = activeMembers.Select(x => x.ToMember()).ToList();
+        _cluster.MemberList.UpdateClusterTopology(members);
+    }
+
+    private async Task AddMember()
+    {
+        await Retry.Try(AddMemberInternal, retryCount: Retry.Forever, onError: OnError, onFailed: OnFailed).ConfigureAwait(false);
+
+        void OnError(int attempt, Exception exception) => _logger.LogWarning(exception, "Failed to register member");
+        void OnFailed(Exception exception) => _logger.LogError(exception, "Failed to register member");
+    }
+
+    private async Task AddMemberInternal()
+    {
+        var canReceiveTraffic = await CanReceiveTrafficAsync().ConfigureAwait(false);
+
+        if (!canReceiveTraffic)
+        {
+            _logger.LogInformation("Revision {RevisionName} is not active", _containerAppMetadata.RevisionName);
+            return;
+        }
+
+        var member = new Member
+        {
+            Id = _memberId,
+            Host = _host,
+            Port = _port,
+        };
+
+        _logger.LogInformation(
+            "[Cluster][AzureContainerAppsProvider] Registering service {ReplicaName} on {IpAddress}",
+            _containerAppMetadata.ReplicaName,
+            _address);
+
+        member.Kinds.AddRange(_kinds);
+
+        await _clusterMemberStore.RegisterAsync(ClusterName, member).ConfigureAwait(false);
+    }
+
+    private void ScheduleUpdate(IContext context)
+    {
+        if (_stopping)
+            return;
+
+        var pollInterval = _options.Value.PollInterval;
+        _scheduledTask = context.Scheduler().SendOnce(pollInterval, context.Self, new UpdateMembers());
+    }
+
+    private async Task<bool> CanReceiveTrafficAsync()
+    {
+        var containerApp = await GetContainerAppAsync();
+        var revisionName = _containerAppMetadata.RevisionName;
+        var revision = await containerApp.GetContainerAppRevisionAsync(revisionName).ConfigureAwait(false);
+        return revision.Value.Data.TrafficWeight.GetValueOrDefault() > 0;
+    }
+
+    private async Task<ContainerAppResource> GetContainerAppAsync()
+    {
+        if (_containerApp is not null)
+            return _containerApp;
+
+        var subscriptionId = _options.Value.SubscriptionId;
+        var resourceGroupName = _options.Value.ResourceGroupName;
+        var client = await _armClientProvider.CreateClientAsync().ConfigureAwait(false);
+        var resourceGroup = await client.GetResourceGroupByNameAsync(resourceGroupName, subscriptionId).ConfigureAwait(false);
+        var containerAppName = _containerAppMetadata.ContainerAppName;
+        var response = await resourceGroup.GetContainerAppAsync(containerAppName).ConfigureAwait(false);
+
+        _containerApp = response.Value;
+        return _containerApp;
+    }
+
+    private async Task UnregisterMemberInternal()
+    {
+        _stopping = true;
+        _scheduledTask?.Cancel();
+        _logger.LogInformation("[Cluster][AzureContainerAppsProvider] Unregistering member {ReplicaName} on {IpAddress}", _containerAppMetadata.RevisionName, _address);
+        await _clusterMemberStore.UnregisterAsync(_memberId).ConfigureAwait(false);
+    }
+}

src/Proto.Cluster.AzureContainerApps/DefaultAzureCredentialArmClientProvider.cs → src/Proto.Cluster.AzureContainerApps/ArmClientProviders/DefaultAzureCredentialArmClientProvider.cs

@@ -3,8 +3,9 @@
 using Azure.Identity;
 using Azure.ResourceManager;
 using JetBrains.Annotations;
+using Proto.Cluster.AzureContainerApps.Contracts;
 
-namespace Proto.Cluster.AzureContainerApps;
+namespace Proto.Cluster.AzureContainerApps.ArmClientProviders;
 
 /// <summary>
 /// Provides an <see cref="ArmClient"/> instance using <see cref="Azure.Identity.DefaultAzureCredential"/>