From 0ff791a0b9b3ed37cb75ad6bcc090fd2739a2ec4 Mon Sep 17 00:00:00 2001 From: raymonddenhaan <155616759+raymonddenhaan@users.noreply.github.com> Date: Thu, 20 Jun 2024 22:48:03 +0200 Subject: [PATCH] Implement automated cleanup for Azure Service Bus subscriptions (#5636) * Implement automated cleanup for Azure Service Bus subscriptions An automated cleanup process has been added for subscriptions that do not have queues connected to Azure Service Bus. This new feature deletes orphaned topics and cleans up other remnants within the namespace. * Removed string interpolation from log message. * Update Program.cs Use Memory transport to ensure docker image for demo purposes functions correctly. * Update AzureServiceBusFeature.cs * Update CleanupSubscriptions.cs * Refactor notifications to commands in MassTransit module Replaced the usage of notifications with commands in the Elsa.MassTransit.AzureServiceBus module. This included changing notification handler to command handler in methods and updating services to use the new command handlers. --------- Co-authored-by: Sipke Schoorstra --- .../Elsa.ServerAndStudio.Web/Program.cs | 6 +- .../Commands/CleanupSubscriptions.cs | 8 +++ .../Features/AzureServiceBusFeature.cs | 33 ++++++++++- .../Handlers/CleanupSubscriptions.cs | 59 +++++++++++++++++++ .../Handlers/RemoveOrphanedSubscriptions.cs | 7 ++- .../CleanSubscriptionsWithoutQueues.cs | 57 ++++++++++++++++++ .../Options/SubscriptionCleanupOptions.cs | 6 ++ 7 files changed, 170 insertions(+), 6 deletions(-) create mode 100644 src/modules/Elsa.MassTransit.AzureServiceBus/Commands/CleanupSubscriptions.cs create mode 100644 src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/CleanupSubscriptions.cs create mode 100644 src/modules/Elsa.MassTransit.AzureServiceBus/HostedServices/CleanSubscriptionsWithoutQueues.cs create mode 100644 src/modules/Elsa.MassTransit.AzureServiceBus/Options/SubscriptionCleanupOptions.cs diff --git a/src/bundles/Elsa.ServerAndStudio.Web/Program.cs b/src/bundles/Elsa.ServerAndStudio.Web/Program.cs index 68d57ce17..a6ca02528 100644 --- a/src/bundles/Elsa.ServerAndStudio.Web/Program.cs +++ b/src/bundles/Elsa.ServerAndStudio.Web/Program.cs @@ -120,7 +120,11 @@ services switch (useMassTransitBroker) { case MassTransitBroker.AzureServiceBus: - massTransit.UseAzureServiceBus(azureServiceBusConnectionString); + massTransit.UseAzureServiceBus(azureServiceBusConnectionString, asb => + { + asb.SubscriptionCleanupOptions = options => options.Interval = TimeSpan.FromMinutes(5); + asb.EnableAutomatedSubscriptionCleanup = true; + }); break; case MassTransitBroker.RabbitMq: massTransit.UseRabbitMq(rabbitMqConnectionString); diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Commands/CleanupSubscriptions.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Commands/CleanupSubscriptions.cs new file mode 100644 index 000000000..398ebcd49 --- /dev/null +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Commands/CleanupSubscriptions.cs @@ -0,0 +1,8 @@ +using Elsa.Mediator.Contracts; + +namespace Elsa.MassTransit.AzureServiceBus.Commands; + +/// +/// Notification to clean up the Azure Service Bus subscriptions. +/// +public record CleanupSubscriptions : ICommand; \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs index 7b2073d47..0fc93e992 100644 --- a/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Features/AzureServiceBusFeature.cs @@ -6,6 +6,7 @@ using Elsa.Features.Services; using Elsa.Hosting.Management.Contracts; using Elsa.Hosting.Management.Features; using Elsa.MassTransit.AzureServiceBus.Handlers; +using Elsa.MassTransit.AzureServiceBus.HostedServices; using Elsa.MassTransit.AzureServiceBus.Models; using Elsa.MassTransit.AzureServiceBus.Options; using Elsa.MassTransit.AzureServiceBus.Services; @@ -34,6 +35,16 @@ public class AzureServiceBusFeature : FeatureBase /// An Azure Service Bus connection string. public string? ConnectionString { get; set; } + /// + /// Delete subscriptions where their connected queues could not be found. + /// Topics without any subscriptions will be deleted as well. + /// + /// + /// - All subscriptions will be cleaned up, including those that were not created by Elsa. + /// - Queues in other namespaces will not be found and the subscription will therefore be removed. + /// + public bool EnableAutomatedSubscriptionCleanup { get; set; } + /// /// A delegate that configures the Azure Service Bus transport options. /// @@ -44,6 +55,11 @@ public class AzureServiceBusFeature : FeatureBase /// public Action AzureServiceBusOptions { get; set; } = _ => { }; + /// + /// A delegate to configure . + /// + public Action SubscriptionCleanupOptions { get; set; } = _ => { }; + /// /// A delegate to create a instance. /// @@ -87,7 +103,9 @@ public class AzureServiceBusFeature : FeatureBase foreach (var consumer in temporaryConsumers) { - var queueName = $"{consumer.Name}-{instanceNameProvider.GetName()}"; + // Start with the instance name since we will be using that to delete queues / subscriptions that are no longer needed. + // This is the only way to guarantee we can match subscriptions to an application instance, since the Azure Service Bus transport for MassTransit trims names that are too large. + var queueName = $"{instanceNameProvider.GetName()}-{consumer.Name}"; configurator.ReceiveEndpoint(queueName, endpointConfigurator => { endpointConfigurator.AutoDeleteOnIdle = options.TemporaryQueueTtl ?? TimeSpan.FromHours(1); @@ -112,9 +130,19 @@ public class AzureServiceBusFeature : FeatureBase public override void Apply() { Services.Configure(AzureServiceBusOptions); + Services.Configure(SubscriptionCleanupOptions); Services.AddSingleton(ServiceBusAdministrationClientFactory); } + /// + public override void ConfigureHostedServices() + { + if (EnableAutomatedSubscriptionCleanup) + { + Module.ConfigureHostedService(); + } + } + private static string GetConnectionString(IServiceProvider serviceProvider) { var options = serviceProvider.GetRequiredService>().Value; @@ -135,5 +163,6 @@ public class AzureServiceBusFeature : FeatureBase Services.AddSingleton(new MessageTopologyProvider(subscriptionTopology)); Services.AddNotificationHandler(); + Services.AddCommandHandler(); } -} \ No newline at end of file +} diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/CleanupSubscriptions.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/CleanupSubscriptions.cs new file mode 100644 index 000000000..144d963fe --- /dev/null +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/CleanupSubscriptions.cs @@ -0,0 +1,59 @@ +using Azure.Messaging.ServiceBus.Administration; +using Elsa.MassTransit.AzureServiceBus.Commands; +using Elsa.Mediator.Contracts; +using Elsa.Mediator.Models; +using JetBrains.Annotations; +using Microsoft.Extensions.Logging; + +namespace Elsa.MassTransit.AzureServiceBus.Handlers; + +/// +/// Cleans up Azure Service Bus resources for which no children can be found. +/// +/// +/// This will clean up topics without subscriptions and subscription of which the queue can not be found. +/// - This handler is unable to differentiate between topologies created by Elsa and others. It will clean up all topology in the namespace. +/// - Queues in other namespaces will not be found by this handler and the subscription will therefore be removed. +/// +[UsedImplicitly] +public class CleanupOrphanedTopology(ServiceBusAdministrationClient client, ILogger logger) : ICommandHandler +{ + /// + public async Task HandleAsync(CleanupSubscriptions notification, CancellationToken cancellationToken) + { + var queues = await client.GetQueuesAsync().ToListAsync(cancellationToken); + var queueNames = queues.Select(q => q.Name).ToList(); + + await foreach (var topic in client.GetTopicsAsync(cancellationToken)) + { + // Only clean up any topics that have the elsa prefix. + if (!topic.Name.StartsWith("elsa")) + continue; + + var subscriptions = await client.GetSubscriptionsAsync(topic.Name, cancellationToken).ToListAsync(cancellationToken); + + // Delete topics which have no active subscriptions. + if (subscriptions.Count == 0) + { + logger.LogWarning("Deleting topic {name}", topic.Name); + await client.DeleteTopicAsync(topic.Name, cancellationToken); + continue; + } + + var subscriptionsToDelete = subscriptions.Where(sub => + { + var queueName = sub.ForwardTo[(sub.ForwardTo.LastIndexOf('/') + 1)..]; + return !queueNames.Contains(queueName); + }).ToList(); + + // Remove subscriptions for which the queue to be forwarded to can not be found in the same namespace. + foreach (SubscriptionProperties? asbSubscription in subscriptionsToDelete) + { + logger.LogWarning("Deleting subscription {subscriptionName} on topic {topicName}", asbSubscription.SubscriptionName, topic.Name); + await client.DeleteSubscriptionAsync(topic.Name, asbSubscription.SubscriptionName, cancellationToken); + } + } + + return Unit.Instance; + } +} diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/RemoveOrphanedSubscriptions.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/RemoveOrphanedSubscriptions.cs index e7276a8d6..a95cb6af9 100644 --- a/src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/RemoveOrphanedSubscriptions.cs +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Handlers/RemoveOrphanedSubscriptions.cs @@ -9,7 +9,8 @@ using Microsoft.Extensions.Logging; namespace Elsa.MassTransit.AzureServiceBus.Handlers; /// -/// A handler for the notification that removes orphaned subscriptions from Azure Service Bus. +/// A handler for the notification that removes subscriptions from Azure Service Bus +/// when there has not been any heartbeat received lately for this instance. /// [UsedImplicitly] public class RemoveOrphanedSubscriptions(MessageTopologyProvider topologyProvider, @@ -28,7 +29,7 @@ public class RemoveOrphanedSubscriptions(MessageTopologyProvider topologyProvide { try { - // Get subscriptions based on topics instead of the topology since when names are longer than 50 characters. + // Get subscriptions based on topics instead of the topology since when names are longer than 50 characters // MassTransit automatically truncates them. await foreach (var asbSubscription in client.GetSubscriptionsAsync(subscription.TopicName, cancellationToken)) { @@ -44,7 +45,7 @@ public class RemoveOrphanedSubscriptions(MessageTopologyProvider topologyProvide } catch (ServiceBusException ex) when(ex.Reason == ServiceBusFailureReason.MessagingEntityNotFound) { - logger.LogWarning(ex, $"Service bus entity {ex.EntityPath} was not found"); + logger.LogWarning(ex, "Service bus entity {entityPath} was not found", ex.EntityPath); } } } diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/HostedServices/CleanSubscriptionsWithoutQueues.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/HostedServices/CleanSubscriptionsWithoutQueues.cs new file mode 100644 index 000000000..b83fff786 --- /dev/null +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/HostedServices/CleanSubscriptionsWithoutQueues.cs @@ -0,0 +1,57 @@ +using Elsa.MassTransit.AzureServiceBus.Commands; +using Elsa.MassTransit.AzureServiceBus.Options; +using Elsa.Mediator.Contracts; +using Medallion.Threading; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Options; + +namespace Elsa.MassTransit.AzureServiceBus.HostedServices; + +/// +/// Represents a hosted service that cleans up subscriptions without queues in Azure Service Bus. +/// +public class CleanSubscriptionsWithoutQueues(IServiceProvider serviceProvider, IOptions subscriptionCleanupOptions) : IHostedService, IDisposable +{ + private Timer? _timer; + + /// + public Task StartAsync(CancellationToken cancellationToken) + { + // Start the service a bit later to make sure it is not running simultaneously with the Heartbeat on startup. + _timer = new Timer(CleanUpSubscriptions, null, TimeSpan.FromMinutes(2), subscriptionCleanupOptions.Value.Interval); + return Task.CompletedTask; + } + + /// + public Task StopAsync(CancellationToken cancellationToken) + { + _timer?.Change(Timeout.Infinite, Timeout.Infinite); + return Task.CompletedTask; + } + + /// + public void Dispose() + { + _timer?.Dispose(); + } + + private void CleanUpSubscriptions(object? state) + { + _ = Task.Run(async () => await CleanUpSubscriptionsAsync()); + } + + private async Task CleanUpSubscriptionsAsync() + { + using var scope = serviceProvider.CreateScope(); + var lockProvider = scope.ServiceProvider.GetRequiredService(); + var commandSender = scope.ServiceProvider.GetRequiredService(); + + const string lockKey = "SubscriptionCleanupService"; + await using var monitorLock = await lockProvider.TryAcquireLockAsync(lockKey, TimeSpan.Zero); + if (monitorLock == null) + return; + + await commandSender.SendAsync(new CleanupSubscriptions()); + } +} \ No newline at end of file diff --git a/src/modules/Elsa.MassTransit.AzureServiceBus/Options/SubscriptionCleanupOptions.cs b/src/modules/Elsa.MassTransit.AzureServiceBus/Options/SubscriptionCleanupOptions.cs new file mode 100644 index 000000000..2d20a6f84 --- /dev/null +++ b/src/modules/Elsa.MassTransit.AzureServiceBus/Options/SubscriptionCleanupOptions.cs @@ -0,0 +1,6 @@ +namespace Elsa.MassTransit.AzureServiceBus.Options; + +public class SubscriptionCleanupOptions +{ + public TimeSpan Interval { get; set; } = TimeSpan.FromDays(7); +} \ No newline at end of file