-
Notifications
You must be signed in to change notification settings - Fork 294
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add poison queues support for RabbitMQ (#648)
## Motivation and Context (Why the change? What's the scenario?) Avoid infinite retries when using RabbitMQ, and move poison messages after N retries. See #408. --------- Co-authored-by: Devis Lucato <[email protected]>
- Loading branch information
1 parent
fb55c22
commit 759cc43
Showing
14 changed files
with
553 additions
and
220 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,90 @@ | ||
| // Copyright (c) Microsoft. All rights reserved. | ||
|
|
||
| using System.Text; | ||
| using Microsoft.KernelMemory; | ||
| using Microsoft.KernelMemory.Diagnostics; | ||
| using Microsoft.KernelMemory.Orchestration.RabbitMQ; | ||
| using Microsoft.KernelMemory.Pipeline.Queue; | ||
| using RabbitMQ.Client; | ||
| using RabbitMQ.Client.Events; | ||
|
|
||
| namespace Microsoft.RabbitMQ.TestApplication; | ||
|
|
||
| internal static class Program | ||
| { | ||
| private const string QueueName = "test queue"; | ||
|
|
||
| public static async Task Main() | ||
| { | ||
| var cfg = new ConfigurationBuilder() | ||
| .AddJsonFile("appsettings.json") | ||
| .AddJsonFile("appsettings.development.json", optional: true) | ||
| .AddJsonFile("appsettings.Development.json", optional: true) | ||
| .Build(); | ||
|
|
||
| var rabbitMQConfig = cfg.GetSection("KernelMemory:Services:RabbitMQ").Get<RabbitMQConfig>(); | ||
| ArgumentNullExceptionEx.ThrowIfNull(rabbitMQConfig, nameof(rabbitMQConfig), "RabbitMQ config not found"); | ||
|
|
||
| DefaultLogger.Factory = LoggerFactory.Create(builder => | ||
| { | ||
| builder.AddConsole(); | ||
| builder.SetMinimumLevel(LogLevel.Warning); | ||
| }); | ||
|
|
||
| var pipeline = new RabbitMQPipeline(rabbitMQConfig, DefaultLogger.Factory); | ||
|
|
||
| var counter = 0; | ||
| pipeline.OnDequeue(async msg => | ||
| { | ||
| Console.WriteLine($"{++counter} Received message: {msg}"); | ||
| await Task.Delay(0); | ||
| return false; | ||
| }); | ||
|
|
||
| await pipeline.ConnectToQueueAsync(QueueName, QueueOptions.PubSub); | ||
|
|
||
| ListenToDeadLetterQueue(rabbitMQConfig); | ||
|
|
||
| await pipeline.EnqueueAsync($"test {DateTimeOffset.Now:T}"); | ||
|
|
||
| while (true) | ||
| { | ||
| await Task.Delay(TimeSpan.FromSeconds(2)); | ||
| } | ||
| } | ||
|
|
||
| private static void ListenToDeadLetterQueue(RabbitMQConfig config) | ||
| { | ||
| var factory = new ConnectionFactory | ||
| { | ||
| HostName = config.Host, | ||
| Port = config.Port, | ||
| UserName = config.Username, | ||
| Password = config.Password, | ||
| VirtualHost = !string.IsNullOrWhiteSpace(config.VirtualHost) ? config.VirtualHost : "/", | ||
| DispatchConsumersAsync = true, | ||
| Ssl = new SslOption | ||
| { | ||
| Enabled = config.SslEnabled, | ||
| ServerName = config.Host, | ||
| } | ||
| }; | ||
|
|
||
| var connection = factory.CreateConnection(); | ||
| var channel = connection.CreateModel(); | ||
| var consumer = new AsyncEventingBasicConsumer(channel); | ||
|
|
||
| consumer.Received += async (object sender, BasicDeliverEventArgs args) => | ||
| { | ||
| byte[] body = args.Body.ToArray(); | ||
| string message = Encoding.UTF8.GetString(body); | ||
| Console.WriteLine($"Poison message received: {message}"); | ||
| await Task.Delay(0); | ||
| }; | ||
|
|
||
| channel.BasicConsume(queue: $"{QueueName}{config.PoisonQueueSuffix}", | ||
| autoAck: true, | ||
| consumer: consumer); | ||
| } | ||
| } |
19 changes: 19 additions & 0 deletions
19
extensions/RabbitMQ/RabbitMQ.TestApplication/RabbitMQ.TestApplication.csproj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk.Web"> | ||
|
|
||
| <PropertyGroup> | ||
| <AssemblyName>Microsoft.RabbitMQ.TestApplication</AssemblyName> | ||
| <RootNamespace>Microsoft.RabbitMQ.TestApplication</RootNamespace> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFramework>net8.0</TargetFramework> | ||
| <RollForward>LatestMajor</RollForward> | ||
| <ImplicitUsings>enable</ImplicitUsings> | ||
| <Nullable>enable</Nullable> | ||
| <IsPackable>false</IsPackable> | ||
| <NoWarn>$(NoWarn);KMEXP00;KMEXP01;KMEXP02;KMEXP03;KMEXP04;</NoWarn> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\..\..\service\Core\Core.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
24 changes: 24 additions & 0 deletions
24
extensions/RabbitMQ/RabbitMQ.TestApplication/appsettings.json
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| { | ||
| "Logging": { | ||
| "LogLevel": { | ||
| "Default": "Error" | ||
| } | ||
| }, | ||
| "KernelMemory": { | ||
| "Services": { | ||
| "RabbitMQ": { | ||
| "Host": "127.0.0.1", | ||
| "Port": "5672", | ||
| "Username": "user", | ||
| "Password": "password", | ||
| "VirtualHost": "/", | ||
| "MessageTTLSecs": 3600, | ||
| "SslEnabled": false, | ||
| // How many times to dequeue a messages and process before moving it to a poison queue | ||
| "MaxRetriesBeforePoisonQueue": 5, | ||
| // Suffix used for the poison queues. | ||
| "PoisonQueueSuffix": "-poison" | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| // Copyright (c) Microsoft. All rights reserved. | ||
|
|
||
| using System.Text; | ||
|
|
||
| #pragma warning disable IDE0130 // reduce number of "using" statements | ||
| // ReSharper disable once CheckNamespace - reduce number of "using" statements | ||
| namespace Microsoft.KernelMemory; | ||
|
|
||
| public class RabbitMQConfig | ||
| { | ||
| /// <summary> | ||
| /// RabbitMQ hostname, e.g. "127.0.0.1" | ||
| /// </summary> | ||
| public string Host { get; set; } = ""; | ||
|
|
||
| /// <summary> | ||
| /// TCP port for the connection, e.g. 5672 | ||
| /// </summary> | ||
| public int Port { get; set; } = 0; | ||
|
|
||
| /// <summary> | ||
| /// Authentication username | ||
| /// </summary> | ||
| public string Username { get; set; } = ""; | ||
|
|
||
| /// <summary> | ||
| /// Authentication password | ||
| /// </summary> | ||
| public string Password { get; set; } = ""; | ||
|
|
||
| /// <summary> | ||
| /// RabbitMQ virtual host name, e.g. "/" | ||
| /// See https://www.rabbitmq.com/docs/vhosts | ||
| /// </summary> | ||
| public string VirtualHost { get; set; } = "/"; | ||
|
|
||
| /// <summary> | ||
| /// How long to retry messages delivery, ie how long to retry, in seconds. | ||
| /// Default: 3600 second, 1 hour. | ||
| /// </summary> | ||
| public int MessageTTLSecs { get; set; } = 3600; | ||
|
|
||
| /// <summary> | ||
| /// Set to true if your RabbitMQ supports SSL. | ||
| /// Default: false | ||
| /// </summary> | ||
| public bool SslEnabled { get; set; } = false; | ||
|
|
||
| /// <summary> | ||
| /// How many times to retry processing a message before moving it to a poison queue. | ||
| /// Example: a value of 20 means that a message will be processed up to 21 times. | ||
| /// Note: this value cannot be changed after queues have been created. In such case | ||
| /// you might need to drain all queues, delete them, and restart the ingestion service(s). | ||
| /// </summary> | ||
| public int MaxRetriesBeforePoisonQueue { get; set; } = 20; | ||
|
|
||
| /// <summary> | ||
| /// Suffix used for the poison queues. | ||
| /// </summary> | ||
| public string PoisonQueueSuffix { get; set; } = "-poison"; | ||
|
|
||
| /// <summary> | ||
| /// Verify that the current state is valid. | ||
| /// </summary> | ||
| public void Validate() | ||
| { | ||
| const int MinTTLSecs = 5; | ||
|
|
||
| if (string.IsNullOrWhiteSpace(this.Host) || this.Host != $"{this.Host}".Trim()) | ||
| { | ||
| throw new ConfigurationException($"RabbitMQ: {nameof(this.Host)} cannot be empty or have leading or trailing spaces"); | ||
| } | ||
|
|
||
| if (this.Port < 1) | ||
| { | ||
| throw new ConfigurationException($"RabbitMQ: {nameof(this.Port)} value {this.Port} is not valid"); | ||
| } | ||
|
|
||
| if (this.MessageTTLSecs < MinTTLSecs) | ||
| { | ||
| throw new ConfigurationException($"RabbitMQ: {nameof(this.MessageTTLSecs)} value {this.MessageTTLSecs} is too low, cannot be less than {MinTTLSecs}"); | ||
| } | ||
|
|
||
| if (string.IsNullOrWhiteSpace(this.PoisonQueueSuffix) || this.PoisonQueueSuffix != $"{this.PoisonQueueSuffix}".Trim()) | ||
| { | ||
| throw new ConfigurationException($"RabbitMQ: {nameof(this.PoisonQueueSuffix)} cannot be empty or have leading or trailing spaces"); | ||
| } | ||
|
|
||
| if (this.MaxRetriesBeforePoisonQueue < 0) | ||
| { | ||
| throw new ConfigurationException($"RabbitMQ: {nameof(this.MaxRetriesBeforePoisonQueue)} cannot be a negative number"); | ||
| } | ||
|
|
||
| if (string.IsNullOrWhiteSpace(this.PoisonQueueSuffix)) | ||
| { | ||
| throw new ConfigurationException($"RabbitMQ: {nameof(this.PoisonQueueSuffix)} is empty"); | ||
| } | ||
|
|
||
| // Queue names can be up to 255 bytes of UTF-8 characters. | ||
| // Allow a max of 60 bytes for the suffix, so there is room for the queue name. | ||
| if (Encoding.UTF8.GetByteCount(this.PoisonQueueSuffix) > 60) | ||
| { | ||
| throw new ConfigurationException($"RabbitMQ: {nameof(this.PoisonQueueSuffix)} can be up to 60 characters length"); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.