RavenDB-21956 Add azure queue storage etl

src/Raven.Client/Documents/Operations/ETL/Queue/AzureQueueStorageConnectionSettings.cs

@@ -0,0 +1,86 @@
+using System;
+using System.Linq;
+using Sparrow.Json.Parsing;
+
+namespace Raven.Client.Documents.Operations.ETL.Queue;
+
+public sealed class AzureQueueStorageConnectionSettings
+{
+    public Authentication Authentication;
+
+    public string GetStorageUrl()
+    {
+        string storageAccountName = GetStorageAccountName();
+        return $"https://{storageAccountName}.queue.core.windows.net/";
+    }
+
+    public string GetStorageAccountName()
+    {
+        string storageAccountName = "";
+
+        if (Authentication.ConnectionString != null)
+        {
+            var accountNamePart = Authentication.ConnectionString.Split(';')
+                .FirstOrDefault(part => part.StartsWith("AccountName=", StringComparison.OrdinalIgnoreCase));
+
+            if (accountNamePart == null)
+            {
+                throw new ArgumentException("Storage account name not found in the connection string.",
+                    nameof(Authentication.ConnectionString));
+            }
+
+            storageAccountName = accountNamePart.Substring("AccountName=".Length);
+        }
+        else if (Authentication.EntraId != null)
+        {
+            storageAccountName = Authentication.EntraId.StorageAccountName;
+        }
+
+        return storageAccountName;
+    }
+
+    public DynamicJsonValue ToJson()
+    {
+        var json = new DynamicJsonValue
+        {
+            [nameof(Authentication)] = Authentication == null
+                ? null
+                : new DynamicJsonValue
+                {
+                    [nameof(Authentication.ConnectionString)] = Authentication.ConnectionString,
+                    [nameof(Authentication.Passwordless)] = Authentication.Passwordless,
+                    [nameof(Authentication.EntraId)] =
+                        Authentication.EntraId == null
+                            ? null
+                            : new DynamicJsonValue
+                            {
+                                [nameof(Authentication.EntraId.StorageAccountName)] =
+                                    Authentication?.EntraId?.StorageAccountName,
+                                [nameof(Authentication.EntraId.TenantId)] =
+                                    Authentication?.EntraId?.TenantId,
+                                [nameof(Authentication.EntraId.ClientId)] =
+                                    Authentication?.EntraId?.ClientId,
+                                [nameof(Authentication.EntraId.ClientSecret)] =
+                                    Authentication?.EntraId?.ClientSecret
+                            }
+                }
+        };
+
+        return json;
+    }
+}
+
+public sealed class Authentication
+{
+    public EntraId EntraId { get; set; }
+    public string ConnectionString { get; set; }
+    public bool Passwordless { get; set; }
+}
+
+public sealed class EntraId
+{
+    public string StorageAccountName { get; set; }
+    public string TenantId { get; set; }
+    public string ClientId { get; set; }
+    public string ClientSecret { get; set; }
+}

src/Raven.Client/Documents/Operations/ETL/Queue/QueueBrokerType.cs

@@ -4,5 +4,6 @@ public enum QueueBrokerType
 {
     None,
     Kafka,
-    RabbitMq
+    RabbitMq,
+    AzureQueueStorage
 }

src/Raven.Client/Documents/Operations/ETL/Queue/QueueConnectionString.cs

@@ -12,6 +12,8 @@ public sealed class QueueConnectionString : ConnectionString
     public KafkaConnectionSettings KafkaConnectionSettings { get; set; }
 
     public RabbitMqConnectionSettings RabbitMqConnectionSettings { get; set; }
+
+    public AzureQueueStorageConnectionSettings AzureQueueStorageConnectionSettings { get; set; }
 
     public override ConnectionStringType Type => ConnectionStringType.Queue;
 
@@ -31,6 +33,12 @@ protected override void ValidateImpl(ref List<string> errors)
                     errors.Add($"{nameof(RabbitMqConnectionSettings)} has no valid setting.");
                 }
                 break;
+            case QueueBrokerType.AzureQueueStorage:
+                if (AzureQueueStorageConnectionSettings?.Authentication == null)
+                {
+                    errors.Add($"{nameof(AzureQueueStorageConnectionSettings)} has no valid setting.");
+                }
+                break;
             default:
                 throw new NotSupportedException($"'{BrokerType}' broker is not supported");
         }
@@ -52,6 +60,9 @@ internal string GetUrl()
 
                 url = indexOfStartServerUri != -1 ? connectionString.Substring(indexOfStartServerUri + 1) : null;
                 break;
+            case QueueBrokerType.AzureQueueStorage:
+                url = AzureQueueStorageConnectionSettings.GetStorageUrl();
+                break;
             default:
                 throw new NotSupportedException($"'{BrokerType}' broker is not supported");
         }
@@ -66,6 +77,7 @@ public override DynamicJsonValue ToJson()
         json[nameof(BrokerType)] = BrokerType;
         json[nameof(KafkaConnectionSettings)] = KafkaConnectionSettings?.ToJson();
         json[nameof(RabbitMqConnectionSettings)] = RabbitMqConnectionSettings?.ToJson();
+        json[nameof(AzureQueueStorageConnectionSettings)] = AzureQueueStorageConnectionSettings?.ToJson();
 
         return json;
     }

src/Raven.Client/Documents/Operations/ETL/Queue/QueueEtlConfiguration.cs

@@ -47,7 +47,11 @@ public override bool UsingEncryptedCommunicationChannel()
                     }
                     break;
                 case QueueBrokerType.RabbitMq:
-                    return Connection.RabbitMqConnectionSettings.ConnectionString.StartsWith("amqps", StringComparison.OrdinalIgnoreCase);
+                    return Connection.RabbitMqConnectionSettings.ConnectionString.StartsWith("amqps",
+                        StringComparison.OrdinalIgnoreCase);
+                case QueueBrokerType.AzureQueueStorage:
+                    return Connection.AzureQueueStorageConnectionSettings.GetStorageUrl()
+                        .StartsWith("https", StringComparison.OrdinalIgnoreCase);
                 default:
                     throw new NotSupportedException($"Unknown broker type: {BrokerType}");
             }

src/Raven.Server/Commercial/LicenseStatus.cs

@@ -216,7 +216,9 @@ public bool CanAutoRenewLetsEncryptCertificate
 
         public bool HasElasticSearchEtl => Enabled(LicenseAttribute.ElasticSearchEtl);
 
-        public bool HasQueueEtl => Enabled(LicenseAttribute.QueueEtl);
+        //public bool HasQueueEtl => Enabled(LicenseAttribute.QueueEtl);
+        public bool HasQueueEtl => true;
+
 
         public bool HasPowerBI => Enabled(LicenseAttribute.PowerBI);
 

src/Raven.Server/Config/Categories/EtlConfiguration.cs

@@ -52,5 +52,17 @@ public sealed class EtlConfiguration : ConfigurationCategory
         [TimeUnit(TimeUnit.Seconds)]
         [ConfigurationEntry("ETL.Queue.Kafka.InitTransactionsTimeoutInSec", ConfigurationEntryScope.ServerWideOrPerDatabase)]
         public TimeSetting KafkaInitTransactionsTimeout { get; set; }
+
+        [Description("Lifespan of a message in the queue")]
+        [DefaultValue(604800)]
+        [TimeUnit(TimeUnit.Seconds)]
+        [ConfigurationEntry("ETL.Queue.AzureQueueStorage.TimeToLiveInSec", ConfigurationEntryScope.ServerWideOrPerDatabase)]
+        public TimeSetting AzureQueueStorageTimeToLive{ get; set; }
+
+        [Description("How long a message is hidden after being retrieved but not deleted")]
+        [DefaultValue(0)]
+        [TimeUnit(TimeUnit.Seconds)]
+        [ConfigurationEntry("ETL.Queue.AzureQueueStorage.VisibilityTimeoutInSec", ConfigurationEntryScope.ServerWideOrPerDatabase)]
+        public TimeSetting AzureQueueStorageVisibilityTimeout{ get; set; }
     }
 }

src/Raven.Server/Dashboard/DatabasesInfoRetriever.cs

@@ -349,6 +349,10 @@ private static DatabaseOngoingTasksInfoItem GetOngoingTasksInfoItem(DocumentData
             long rabbitMqEtlCountOnNode = GetTaskCountOnNode<QueueEtlConfiguration>(database, dbRecord, serverStore, database.EtlLoader.QueueDestinations,
                 task => EtlLoader.GetProcessState(task.Transforms, database, task.Name), task => task.BrokerType == QueueBrokerType.RabbitMq);
 
+            var azureQueueStorageEtlCount = database.EtlLoader.GetQueueDestinationCountByBroker(QueueBrokerType.AzureQueueStorage);
+            long azureQueueStorageEtlCountOnNode = GetTaskCountOnNode<QueueEtlConfiguration>(database, dbRecord, serverStore, database.EtlLoader.QueueDestinations,
+                task => EtlLoader.GetProcessState(task.Transforms, database, task.Name), task => task.BrokerType == QueueBrokerType.AzureQueueStorage);
+
             var periodicBackupCount = database.PeriodicBackupRunner.PeriodicBackups.Count;
             long periodicBackupCountOnNode = BackupUtils.GetTasksCountOnNode(serverStore, database.Name, context);
 
@@ -364,8 +368,8 @@ private static DatabaseOngoingTasksInfoItem GetOngoingTasksInfoItem(DocumentData
                 task => QueueSinkLoader.GetProcessState(task.Scripts, database, task.Name), task => task.BrokerType == QueueBrokerType.RabbitMq);
 
             ongoingTasksCount = extRepCount + replicationHubCount + replicationSinkCount +
-                                ravenEtlCount + sqlEtlCount + elasticSearchEtlCount + olapEtlCount + kafkaEtlCount + rabbitMqEtlCount +
-                                periodicBackupCount + subscriptionCount +
+                                ravenEtlCount + sqlEtlCount + elasticSearchEtlCount + olapEtlCount + kafkaEtlCount +
+                                rabbitMqEtlCount + azureQueueStorageEtlCount + periodicBackupCount + subscriptionCount +
                                 kafkaSinkCount + rabbitMqSinkCount;
 
             return new DatabaseOngoingTasksInfoItem
@@ -380,6 +384,7 @@ private static DatabaseOngoingTasksInfoItem GetOngoingTasksInfoItem(DocumentData
                 OlapEtlCount = olapEtlCountOnNode,
                 KafkaEtlCount = kafkaEtlCountOnNode,
                 RabbitMqEtlCount = rabbitMqEtlCountOnNode,
+                AzureQueueStorageEtlCount = azureQueueStorageEtlCountOnNode,
                 PeriodicBackupCount = periodicBackupCountOnNode,
                 SubscriptionCount = subscriptionCountOnNode,
                 KafkaSinkCount = kafkaSinkCountOnNode,

src/Raven.Server/Dashboard/DatabasesOngoingTasksInfo.cs

@@ -34,6 +34,8 @@ public sealed class DatabaseOngoingTasksInfoItem : IDynamicJson
         public long KafkaEtlCount { get; set; }
 
         public long RabbitMqEtlCount { get; set; }
+
+        public long AzureQueueStorageEtlCount { get; set; }
 
         public long PeriodicBackupCount { get; set; }
 
@@ -57,6 +59,7 @@ public DynamicJsonValue ToJson()
                 [nameof(OlapEtlCount)] = OlapEtlCount,
                 [nameof(KafkaEtlCount)] = KafkaEtlCount,
                 [nameof(RabbitMqEtlCount)] = RabbitMqEtlCount,
+                [nameof(AzureQueueStorageEtlCount)] = AzureQueueStorageEtlCount,
                 [nameof(PeriodicBackupCount)] = PeriodicBackupCount,
                 [nameof(SubscriptionCount)] = SubscriptionCount,
                 [nameof(KafkaSinkCount)] = KafkaSinkCount,

src/Raven.Server/Documents/ETL/EtlLoader.cs

@@ -18,6 +18,7 @@
 using Raven.Server.Documents.ETL.Providers.ElasticSearch;
 using Raven.Server.Documents.ETL.Providers.OLAP;
 using Raven.Server.Documents.ETL.Providers.Queue;
+using Raven.Server.Documents.ETL.Providers.Queue.AzureQueueStorage;
 using Raven.Server.Documents.ETL.Providers.Queue.Kafka;
 using Raven.Server.Documents.ETL.Providers.Queue.RabbitMq;
 using Raven.Server.Documents.ETL.Providers.Raven;
@@ -646,6 +647,29 @@ public void HandleDatabaseRecordChange(DatabaseRecord record)
 
                             break;
                         }
+                    case AzureQueueStorageEtl azureQueueStorageEtl:
+                    {
+                        QueueEtlConfiguration existing = null;
+
+                        foreach (var config in myQueueEtl)
+                        {
+                            var diff = azureQueueStorageEtl.Configuration.Compare(config);
+
+                            if (diff == EtlConfigurationCompareDifferences.None)
+                            {
+                                existing = config;
+                                break;
+                            }
+                        }
+
+                        if (existing != null)
+                        {
+                            toRemove.Remove(processesPerConfig.Key);
+                            myQueueEtl.Remove(existing);
+                        }
+
+                        break;
+                    }
                     case ElasticSearchEtl elasticSearchEtl:
                         {
                             ElasticSearchEtlConfiguration existing = null;
@@ -794,6 +818,13 @@ private static string GetStopReason(
                 if (existing != null)
                     differences = rabbitMqEtl.Configuration.Compare(existing, transformationDiffs);
             }
+            else if (process is AzureQueueStorageEtl azureQueueStorageEtl)
+            {
+                var existing = myQueueEtl.FirstOrDefault(x => x.Name.Equals(azureQueueStorageEtl.ConfigurationName, StringComparison.OrdinalIgnoreCase));
+
+                if (existing != null)
+                    differences = azureQueueStorageEtl.Configuration.Compare(existing, transformationDiffs);
+            }
             else
             {
                 throw new InvalidOperationException($"Unknown ETL process type: " + process.GetType().FullName);

src/Raven.Server/Documents/ETL/EtlProcess.cs

@@ -23,6 +23,7 @@
 using Raven.Server.Documents.ETL.Providers.OLAP;
 using Raven.Server.Documents.ETL.Providers.OLAP.Test;
 using Raven.Server.Documents.ETL.Providers.Queue;
+using Raven.Server.Documents.ETL.Providers.Queue.AzureQueueStorage;
 using Raven.Server.Documents.ETL.Providers.Queue.Kafka;
 using Raven.Server.Documents.ETL.Providers.Queue.RabbitMq;
 using Raven.Server.Documents.ETL.Providers.Raven;
@@ -1328,6 +1329,21 @@ public static IDisposable TestScript<TC, TCS>(
                                         result = rabbitMqEtl.RunTest(results, context);
                                         result.DebugOutput = debugOutput;
 
+                                        return tx;
+                                    }
+                                case AzureQueueStorageEtl azureQueueStorageEtl:
+                                    using (azureQueueStorageEtl.EnterTestMode(out debugOutput))
+                                    {
+                                        azureQueueStorageEtl.EnsureThreadAllocationStats();
+
+                                        var queueItem = new QueueItem(document, docCollection);
+
+                                        var results = azureQueueStorageEtl.Transform(new[] { queueItem }, context, new EtlStatsScope(new EtlRunStats()),
+                                            new EtlProcessState());
+
+                                        result = azureQueueStorageEtl.RunTest(results, context);
+                                        result.DebugOutput = debugOutput;
+
                                         return tx;
                                     }
                                 default:

src/Raven.Server/Documents/ETL/Handlers/Processors/EtlHandlerProcessorForProgress.cs

@@ -2,6 +2,7 @@
 using System.Threading.Tasks;
 using JetBrains.Annotations;
 using Raven.Client.Documents.Operations.ETL.Queue;
+using Raven.Server.Documents.ETL.Providers.Queue.AzureQueueStorage;
 using Raven.Server.Documents.ETL.Providers.Queue.Kafka;
 using Raven.Server.Documents.ETL.Providers.Queue.RabbitMq;
 using Raven.Server.Documents.ETL.Stats;
@@ -37,6 +38,7 @@ protected override async ValueTask HandleCurrentNodeAsync()
                 {
                     RabbitMqEtl => QueueBrokerType.RabbitMq,
                     KafkaEtl => QueueBrokerType.Kafka,
+                    AzureQueueStorageEtl => QueueBrokerType.AzureQueueStorage,
                     _ => null
                 }
             }).ToArray();