RavenDB-21956 Add azure queue storage etl

RavenDB-21956 wip RavenDB-21956 RavenDB-21956 wip RavenDB-21956 tests wip RavenDB-21956 wip RavenDB-21956 AzureStorageQueue -> AzureQueueStorage RavenDB-21956 Add passwordless option
djordjedjukic · Mar 26, 2024 · e034cc8 · e034cc8
1 parent 2ed871d
commit e034cc8
Show file tree

Hide file tree

Showing 15 changed files with 724 additions and 84 deletions.
diff --git a/src/Raven.Client/Documents/Operations/ETL/Queue/AzureQueueStorageConnectionSettings.cs b/src/Raven.Client/Documents/Operations/ETL/Queue/AzureQueueStorageConnectionSettings.cs
@@ -1,26 +1,54 @@
+using System;
+using System.Linq;
 using Sparrow.Json.Parsing;
 
 namespace Raven.Client.Documents.Operations.ETL.Queue;
 
 public sealed class AzureQueueStorageConnectionSettings
 {
-    // TODO djordje: what should be input, minutes?
-    public int TimeToLive { get; set; }
-
-    public int VisibilityTimeout { get; set; }
-
     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(TimeToLive)] = TimeToLive,
-            [nameof(VisibilityTimeout)] = VisibilityTimeout,
             [nameof(Authentication)] = Authentication == null
                 ? null
                 : new DynamicJsonValue
                 {
+                    [nameof(Authentication.ConnectionString)] = Authentication.ConnectionString,
+                    [nameof(Authentication.Passwordless)] = Authentication.Passwordless,
                     [nameof(Authentication.EntraId)] =
                         Authentication.EntraId == null
                             ? null
@@ -34,13 +62,7 @@ public DynamicJsonValue ToJson()
                                     Authentication?.EntraId?.ClientId,
                                 [nameof(Authentication.EntraId.ClientSecret)] =
                                     Authentication?.EntraId?.ClientSecret
-                            },
-                    [nameof(Authentication.ConnectionString)] = Authentication.ConnectionString == null
-                        ? null
-                        : new DynamicJsonValue
-                        {
-                            [nameof(Authentication.ConnectionString)] = Authentication?.ConnectionString
-                        }
+                            }
                 }
         };
 
@@ -52,6 +74,7 @@ public sealed class Authentication
 {
     public EntraId EntraId { get; set; }
     public string ConnectionString { get; set; }
+    public bool Passwordless { get; set; }
 }
 
 public sealed class EntraId

diff --git a/src/Raven.Client/Documents/Operations/ETL/Queue/QueueConnectionString.cs b/src/Raven.Client/Documents/Operations/ETL/Queue/QueueConnectionString.cs
@@ -46,7 +46,6 @@ protected override void ValidateImpl(ref List<string> errors)
 
     internal string GetUrl()
     {
-        // TODO djordje: what to do with this, no connection string defined in this moment?
         string url;
 
         switch (BrokerType)
@@ -62,7 +61,7 @@ internal string GetUrl()
                 url = indexOfStartServerUri != -1 ? connectionString.Substring(indexOfStartServerUri + 1) : null;
                 break;
             case QueueBrokerType.AzureQueueStorage:
-                url = "azure-queue-storage";
+                url = AzureQueueStorageConnectionSettings.GetStorageUrl();
                 break;
             default:
                 throw new NotSupportedException($"'{BrokerType}' broker is not supported");

diff --git a/src/Raven.Client/Documents/Operations/ETL/Queue/QueueEtlConfiguration.cs b/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}");
             }

diff --git a/src/Raven.Server/Config/Categories/EtlConfiguration.cs b/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; }
     }
 }
diff --git a/src/Raven.Server/Dashboard/DatabasesInfoRetriever.cs b/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,

diff --git a/src/Raven.Server/Dashboard/DatabasesOngoingTasksInfo.cs b/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,

diff --git a/src/Raven.Server/Documents/ETL/EtlProcess.cs b/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:

diff --git a/src/Raven.Server/Documents/ETL/Handlers/Processors/EtlHandlerProcessorForProgress.cs b/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();