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 · e7b6167 · e7b6167
1 parent 0593ffb
commit e7b6167
Show file tree

Hide file tree

Showing 15 changed files with 677 additions and 64 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();

diff --git a/src/Raven.Server/Documents/ETL/Providers/Queue/AzureQueueStorage/AzureQueueStorageEtl.cs b/src/Raven.Server/Documents/ETL/Providers/Queue/AzureQueueStorage/AzureQueueStorageEtl.cs
@@ -8,6 +8,7 @@
 using Raven.Client.Documents.Operations.ETL;
 using Raven.Client.Documents.Operations.ETL.Queue;
 using Raven.Server.Documents.ETL.Stats;
+using Raven.Server.Exceptions.ETL.QueueEtl;
 using Raven.Server.ServerWide;
 using Raven.Server.ServerWide.Context;
 using Sparrow.Json;
@@ -41,18 +42,17 @@ protected override int PublishMessages(List<QueueWithItems<AzureQueueStorageItem
             return 0;
         }
 
-        idsToDelete = new List<string>();
-
+        idsToDelete = [];
         int count = 0;
 
         foreach (QueueWithItems<AzureQueueStorageItem> queue in itemsPerQueue)
         {
             string queueName = queue.Name.ToLower();
-            
+
             if (_queueClients.TryGetValue(queueName, out QueueClient queueClient) == false)
             {
                 queueClient =
-                    QueueBrokerConnectionHelper.CreateAzureStorageQueueClient(
+                    QueueBrokerConnectionHelper.CreateAzureQueueStorageClient(
                         Configuration.Connection.AzureQueueStorageConnectionSettings, queueName);
                 _queueClients.Add(queueName, queueClient);
             }
@@ -62,27 +62,43 @@ protected override int PublishMessages(List<QueueWithItems<AzureQueueStorageItem
 
             foreach (AzureQueueStorageItem queueItem in queue.Items)
             {
-                var cloudEvent = CreateCloudEvent(queueItem);
-                byte[] cloudEventBytes = JsonSerializer.SerializeToUtf8Bytes(cloudEvent, new JsonSerializerOptions { Converters = { CloudEventConverter.Instance } });
-                string base64CloudEvent = Convert.ToBase64String(cloudEventBytes);
-
-                TimeSpan? timeToLive = Configuration.Connection.AzureQueueStorageConnectionSettings.TimeToLive != 0
-                    ? TimeSpan.FromDays(Configuration.Connection.AzureQueueStorageConnectionSettings.TimeToLive)
-                    : null;
-                TimeSpan? visibilityTimeout =
-                    Configuration.Connection.AzureQueueStorageConnectionSettings.VisibilityTimeout != 0
-                        ? TimeSpan.FromDays(Configuration.Connection.AzureQueueStorageConnectionSettings
-                            .VisibilityTimeout)
-                        : null;
-                var result = queueClient.SendMessage(base64CloudEvent, visibilityTimeout, timeToLive);
-
-                // TODO djordje: handle errors
+                string base64CloudEvent = CreateBase64CloudEvent(queueItem);
+
+                TimeSpan? timeToLive = Database.Configuration.Etl.AzureQueueStorageTimeToLive.AsTimeSpan;
+                TimeSpan? visibilityTimeout = Database.Configuration.Etl.AzureQueueStorageVisibilityTimeout.AsTimeSpan;
+
+                try
+                {
+                    queueClient.SendMessage(base64CloudEvent, visibilityTimeout, timeToLive);
+                    count++;
+
+                    if (queue.DeleteProcessedDocuments)
+                        idsToDelete.Add(queueItem.DocumentId);
+                }
+                catch (Azure.RequestFailedException ex)
+                {
+                    throw new QueueLoadException(
+                        $"Failed to deliver message, Azure error code: '{ex.ErrorCode}', error reason: '{ex.Message}' for document with id: '{queueItem.DocumentId}'",
+                        ex);
+                }
+                catch (Exception ex)
+                {
+                    throw new QueueLoadException($"Failed to deliver message, error reason: '{ex.Message}'", ex);
+                }
             }
         }
 
         return count;
     }
 
+    private string CreateBase64CloudEvent(AzureQueueStorageItem queueItem)
+    {
+        var cloudEvent = CreateCloudEvent(queueItem);
+        var options = new JsonSerializerOptions { Converters = { CloudEventConverter.Instance } };
+        byte[] cloudEventBytes = JsonSerializer.SerializeToUtf8Bytes(cloudEvent, options);
+        return Convert.ToBase64String(cloudEventBytes);
+    }
+
     protected override void OnProcessStopped()
     {
         _queueClients.Clear();
@@ -91,8 +107,16 @@ protected override void OnProcessStopped()
     private void CreateQueueIfNotExists(QueueClient queueClient)
     {
         if (_alreadyCreatedQueues.Contains(queueClient.Name)) return;
-        queueClient.CreateIfNotExists();
-        _alreadyCreatedQueues.Add(queueClient.Name);
+        try
+        {
+            queueClient.CreateIfNotExists();
+            _alreadyCreatedQueues.Add(queueClient.Name);
+        }
+        catch (Azure.RequestFailedException ex)
+        {
+            throw new QueueLoadException(
+                $"Failed to deliver message, Azure error code: '{ex.ErrorCode}', error reason: '{ex.Message}'", ex);
+        }
     }
 
     private sealed class CloudEventConverter : JsonConverter<CloudEvent>

diff --git a/...s/Queue/Handlers/Processors/QueueEtlHandlerProcessorForTestAzureQueueStorageConnection.cs b/...s/Queue/Handlers/Processors/QueueEtlHandlerProcessorForTestAzureQueueStorageConnection.cs
@@ -0,0 +1,64 @@
+using System;
+using System.IO;
+using System.Threading.Tasks;
+using Azure.Storage.Queues;
+using JetBrains.Annotations;
+using Newtonsoft.Json;
+using Raven.Client.Documents.Operations.ETL.Queue;
+using Raven.Server.Documents.Handlers.Processors;
+using Raven.Server.Web.System;
+using Sparrow.Json;
+using Sparrow.Json.Parsing;
+
+namespace Raven.Server.Documents.ETL.Providers.Queue.Handlers.Processors;
+
+internal sealed class
+    QueueEtlHandlerProcessorForTestAzureQueueStorageConnection<TRequestHandler, TOperationContext> :
+    AbstractDatabaseHandlerProcessor<TRequestHandler, TOperationContext>
+    where TOperationContext : JsonOperationContext
+    where TRequestHandler : AbstractDatabaseRequestHandler<TOperationContext>
+{
+    public QueueEtlHandlerProcessorForTestAzureQueueStorageConnection([NotNull] TRequestHandler requestHandler) :
+        base(requestHandler)
+    {
+    }
+
+    public override async ValueTask ExecuteAsync()
+    {
+        try
+        {
+            string authenticationJson = await new StreamReader(HttpContext.Request.Body).ReadToEndAsync();
+            Authentication authentication = JsonConvert.DeserializeObject<Authentication>(authenticationJson);
+
+            var connectionSettings = new AzureQueueStorageConnectionSettings() { Authentication = authentication };
+
+            QueueServiceClient client =
+                QueueBrokerConnectionHelper.CreateAzureQueueStorageServiceClient(connectionSettings);
+
+            await client.GetStatisticsAsync();
+
+            DynamicJsonValue result = new() { [nameof(NodeConnectionTestResult.Success)] = true, };
+            using (ServerStore.ContextPool.AllocateOperationContext(out JsonOperationContext context))
+            await using (AsyncBlittableJsonTextWriter writer = new(context, RequestHandler.ResponseBodyStream()))
+            {
+                context.Write(writer, result);
+            }
+        }
+        catch (Exception ex)
+        {
+            using (ServerStore.ContextPool.AllocateOperationContext(out JsonOperationContext context))
+            {
+                await using (var writer =
+                             new AsyncBlittableJsonTextWriter(context, RequestHandler.ResponseBodyStream()))
+                {
+                    context.Write(writer,
+                        new DynamicJsonValue
+                        {
+                            [nameof(NodeConnectionTestResult.Success)] = false,
+                            [nameof(NodeConnectionTestResult.Error)] = ex.ToString()
+                        });
+                }
+            }
+        }
+    }
+}