mirror of
https://github.com/bitwarden/server
synced 2025-12-21 18:53:41 +00:00
[PM-17562] Add support for retries on event integrations (#5795)
* [PM-17562] Add support for retires on event integrations * Add additional test coverage * Fixed missing await call * Remove debug organization id * Respond to PR feedback * Change NotBeforeUtc to DelayUntilDate. Adjust comments. * Respond to PR feedback
This commit is contained in:
Brant DeBow
GitHub
@@ -0,0 +1,83 @@
|
||||
using System.Text.Json;
|
||||
using Bit.Core.AdminConsole.Models.Data.Integrations;
|
||||
using Bit.Core.AdminConsole.Utilities;
|
||||
using Bit.Core.Enums;
|
||||
using Bit.Core.Models.Data;
|
||||
using Bit.Core.Repositories;
|
||||
|
||||
namespace Bit.Core.Services;
|
||||
|
||||
#nullable enable
|
||||
|
||||
public class EventIntegrationHandler<T>(
|
||||
IntegrationType integrationType,
|
||||
IIntegrationPublisher integrationPublisher,
|
||||
IOrganizationIntegrationConfigurationRepository configurationRepository,
|
||||
IUserRepository userRepository,
|
||||
IOrganizationRepository organizationRepository)
|
||||
: IEventMessageHandler
|
||||
{
|
||||
public async Task HandleEventAsync(EventMessage eventMessage)
|
||||
{
|
||||
if (eventMessage.OrganizationId is not Guid organizationId)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var configurations = await configurationRepository.GetConfigurationDetailsAsync(
|
||||
organizationId,
|
||||
integrationType,
|
||||
eventMessage.Type);
|
||||
|
||||
foreach (var configuration in configurations)
|
||||
{
|
||||
var template = configuration.Template ?? string.Empty;
|
||||
var context = await BuildContextAsync(eventMessage, template);
|
||||
var renderedTemplate = IntegrationTemplateProcessor.ReplaceTokens(template, context);
|
||||
|
||||
var config = configuration.MergedConfiguration.Deserialize<T>()
|
||||
?? throw new InvalidOperationException($"Failed to deserialize to {typeof(T).Name}");
|
||||
|
||||
var message = new IntegrationMessage<T>
|
||||
{
|
||||
IntegrationType = integrationType,
|
||||
Configuration = config,
|
||||
RenderedTemplate = renderedTemplate,
|
||||
RetryCount = 0,
|
||||
DelayUntilDate = null
|
||||
};
|
||||
|
||||
await integrationPublisher.PublishAsync(message);
|
||||
}
|
||||
}
|
||||
|
||||
public async Task HandleManyEventsAsync(IEnumerable<EventMessage> eventMessages)
|
||||
{
|
||||
foreach (var eventMessage in eventMessages)
|
||||
{
|
||||
await HandleEventAsync(eventMessage);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<IntegrationTemplateContext> BuildContextAsync(EventMessage eventMessage, string template)
|
||||
{
|
||||
var context = new IntegrationTemplateContext(eventMessage);
|
||||
|
||||
if (IntegrationTemplateProcessor.TemplateRequiresUser(template) && eventMessage.UserId.HasValue)
|
||||
{
|
||||
context.User = await userRepository.GetByIdAsync(eventMessage.UserId.Value);
|
||||
}
|
||||
|
||||
if (IntegrationTemplateProcessor.TemplateRequiresActingUser(template) && eventMessage.ActingUserId.HasValue)
|
||||
{
|
||||
context.ActingUser = await userRepository.GetByIdAsync(eventMessage.ActingUserId.Value);
|
||||
}
|
||||
|
||||
if (IntegrationTemplateProcessor.TemplateRequiresOrganization(template) && eventMessage.OrganizationId.HasValue)
|
||||
{
|
||||
context.Organization = await organizationRepository.GetByIdAsync(eventMessage.OrganizationId.Value);
|
||||
}
|
||||
|
||||
return context;
|
||||
}
|
||||
}
|
||||
@@ -1,8 +1,8 @@
|
||||
using System.Text.Json.Nodes;
|
||||
using Bit.Core.AdminConsole.Models.Data.Integrations;
|
||||
using Bit.Core.AdminConsole.Utilities;
|
||||
using Bit.Core.Enums;
|
||||
using Bit.Core.Models.Data;
|
||||
using Bit.Core.Models.Data.Integrations;
|
||||
using Bit.Core.Repositories;
|
||||
|
||||
namespace Bit.Core.Services;
|
||||
|
||||
@@ -29,7 +29,7 @@ public class RabbitMqEventListenerService : EventLoggingListenerService
|
||||
UserName = globalSettings.EventLogging.RabbitMq.Username,
|
||||
Password = globalSettings.EventLogging.RabbitMq.Password
|
||||
};
|
||||
_exchangeName = globalSettings.EventLogging.RabbitMq.ExchangeName;
|
||||
_exchangeName = globalSettings.EventLogging.RabbitMq.EventExchangeName;
|
||||
_logger = logger;
|
||||
_queueName = queueName;
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@ public class RabbitMqEventWriteService : IEventWriteService, IAsyncDisposable
|
||||
UserName = globalSettings.EventLogging.RabbitMq.Username,
|
||||
Password = globalSettings.EventLogging.RabbitMq.Password
|
||||
};
|
||||
_exchangeName = globalSettings.EventLogging.RabbitMq.ExchangeName;
|
||||
_exchangeName = globalSettings.EventLogging.RabbitMq.EventExchangeName;
|
||||
|
||||
_lazyConnection = new Lazy<Task<IConnection>>(CreateConnectionAsync);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
using System.Text;
|
||||
using Bit.Core.Settings;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
|
||||
namespace Bit.Core.Services;
|
||||
|
||||
public class RabbitMqIntegrationListenerService : BackgroundService
|
||||
{
|
||||
private const string _deadLetterRoutingKey = "dead-letter";
|
||||
private IChannel _channel;
|
||||
private IConnection _connection;
|
||||
private readonly string _exchangeName;
|
||||
private readonly string _queueName;
|
||||
private readonly string _retryQueueName;
|
||||
private readonly string _deadLetterQueueName;
|
||||
private readonly string _routingKey;
|
||||
private readonly string _retryRoutingKey;
|
||||
private readonly int _maxRetries;
|
||||
private readonly IIntegrationHandler _handler;
|
||||
private readonly ConnectionFactory _factory;
|
||||
private readonly ILogger<RabbitMqIntegrationListenerService> _logger;
|
||||
private readonly int _retryTiming;
|
||||
|
||||
public RabbitMqIntegrationListenerService(IIntegrationHandler handler,
|
||||
string routingKey,
|
||||
string queueName,
|
||||
string retryQueueName,
|
||||
string deadLetterQueueName,
|
||||
GlobalSettings globalSettings,
|
||||
ILogger<RabbitMqIntegrationListenerService> logger)
|
||||
{
|
||||
_handler = handler;
|
||||
_routingKey = routingKey;
|
||||
_retryRoutingKey = $"{_routingKey}-retry";
|
||||
_queueName = queueName;
|
||||
_retryQueueName = retryQueueName;
|
||||
_deadLetterQueueName = deadLetterQueueName;
|
||||
_logger = logger;
|
||||
_exchangeName = globalSettings.EventLogging.RabbitMq.IntegrationExchangeName;
|
||||
_maxRetries = globalSettings.EventLogging.RabbitMq.MaxRetries;
|
||||
_retryTiming = globalSettings.EventLogging.RabbitMq.RetryTiming;
|
||||
|
||||
_factory = new ConnectionFactory
|
||||
{
|
||||
HostName = globalSettings.EventLogging.RabbitMq.HostName,
|
||||
UserName = globalSettings.EventLogging.RabbitMq.Username,
|
||||
Password = globalSettings.EventLogging.RabbitMq.Password
|
||||
};
|
||||
}
|
||||
|
||||
public override async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_connection = await _factory.CreateConnectionAsync(cancellationToken);
|
||||
_channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
|
||||
|
||||
await _channel.ExchangeDeclareAsync(exchange: _exchangeName,
|
||||
type: ExchangeType.Direct,
|
||||
durable: true,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Declare main queue
|
||||
await _channel.QueueDeclareAsync(queue: _queueName,
|
||||
durable: true,
|
||||
exclusive: false,
|
||||
autoDelete: false,
|
||||
arguments: null,
|
||||
cancellationToken: cancellationToken);
|
||||
await _channel.QueueBindAsync(queue: _queueName,
|
||||
exchange: _exchangeName,
|
||||
routingKey: _routingKey,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Declare retry queue (Configurable TTL, dead-letters back to main queue)
|
||||
await _channel.QueueDeclareAsync(queue: _retryQueueName,
|
||||
durable: true,
|
||||
exclusive: false,
|
||||
autoDelete: false,
|
||||
arguments: new Dictionary<string, object>
|
||||
{
|
||||
{ "x-dead-letter-exchange", _exchangeName },
|
||||
{ "x-dead-letter-routing-key", _routingKey },
|
||||
{ "x-message-ttl", _retryTiming }
|
||||
},
|
||||
cancellationToken: cancellationToken);
|
||||
await _channel.QueueBindAsync(queue: _retryQueueName,
|
||||
exchange: _exchangeName,
|
||||
routingKey: _retryRoutingKey,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Declare dead letter queue
|
||||
await _channel.QueueDeclareAsync(queue: _deadLetterQueueName,
|
||||
durable: true,
|
||||
exclusive: false,
|
||||
autoDelete: false,
|
||||
arguments: null,
|
||||
cancellationToken: cancellationToken);
|
||||
await _channel.QueueBindAsync(queue: _deadLetterQueueName,
|
||||
exchange: _exchangeName,
|
||||
routingKey: _deadLetterRoutingKey,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
await base.StartAsync(cancellationToken);
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var consumer = new AsyncEventingBasicConsumer(_channel);
|
||||
consumer.ReceivedAsync += async (_, ea) =>
|
||||
{
|
||||
var json = Encoding.UTF8.GetString(ea.Body.Span);
|
||||
|
||||
try
|
||||
{
|
||||
var result = await _handler.HandleAsync(json);
|
||||
var message = result.Message;
|
||||
|
||||
if (result.Success)
|
||||
{
|
||||
// Successful integration send. Acknowledge message delivery and return
|
||||
await _channel.BasicAckAsync(ea.DeliveryTag, false, cancellationToken);
|
||||
return;
|
||||
}
|
||||
|
||||
if (result.Retryable)
|
||||
{
|
||||
// Integration failed, but is retryable - apply delay and check max retries
|
||||
message.ApplyRetry(result.DelayUntilDate);
|
||||
|
||||
if (message.RetryCount < _maxRetries)
|
||||
{
|
||||
// Publish message to the retry queue. It will be re-published for retry after a delay
|
||||
await _channel.BasicPublishAsync(
|
||||
exchange: _exchangeName,
|
||||
routingKey: _retryRoutingKey,
|
||||
body: Encoding.UTF8.GetBytes(message.ToJson()),
|
||||
cancellationToken: cancellationToken);
|
||||
}
|
||||
else
|
||||
{
|
||||
// Exceeded the max number of retries; fail and send to dead letter queue
|
||||
await PublishToDeadLetterAsync(message.ToJson());
|
||||
_logger.LogWarning("Max retry attempts reached. Sent to DLQ.");
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
// Fatal error (i.e. not retryable) occurred. Send message to dead letter queue without any retries
|
||||
await PublishToDeadLetterAsync(message.ToJson());
|
||||
_logger.LogWarning("Non-retryable failure. Sent to DLQ.");
|
||||
}
|
||||
|
||||
// Message has been sent to retry or dead letter queues.
|
||||
// Acknowledge receipt so Rabbit knows it's been processed
|
||||
await _channel.BasicAckAsync(ea.DeliveryTag, false, cancellationToken);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// Unknown error occurred. Acknowledge so Rabbit doesn't keep attempting. Log the error
|
||||
_logger.LogError(ex, "Unhandled error processing integration message.");
|
||||
await _channel.BasicAckAsync(ea.DeliveryTag, false, cancellationToken);
|
||||
}
|
||||
};
|
||||
|
||||
await _channel.BasicConsumeAsync(queue: _queueName, autoAck: false, consumer: consumer, cancellationToken: cancellationToken);
|
||||
}
|
||||
|
||||
private async Task PublishToDeadLetterAsync(string json)
|
||||
{
|
||||
await _channel.BasicPublishAsync(
|
||||
exchange: _exchangeName,
|
||||
routingKey: _deadLetterRoutingKey,
|
||||
body: Encoding.UTF8.GetBytes(json));
|
||||
}
|
||||
|
||||
public override async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
await _channel.CloseAsync(cancellationToken);
|
||||
await _connection.CloseAsync(cancellationToken);
|
||||
await base.StopAsync(cancellationToken);
|
||||
}
|
||||
|
||||
public override void Dispose()
|
||||
{
|
||||
_channel.Dispose();
|
||||
_connection.Dispose();
|
||||
base.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
using System.Text;
|
||||
using Bit.Core.AdminConsole.Models.Data.Integrations;
|
||||
using Bit.Core.Enums;
|
||||
using Bit.Core.Settings;
|
||||
using RabbitMQ.Client;
|
||||
|
||||
namespace Bit.Core.Services;
|
||||
|
||||
public class RabbitMqIntegrationPublisher : IIntegrationPublisher, IAsyncDisposable
|
||||
{
|
||||
private readonly ConnectionFactory _factory;
|
||||
private readonly Lazy<Task<IConnection>> _lazyConnection;
|
||||
private readonly string _exchangeName;
|
||||
|
||||
public RabbitMqIntegrationPublisher(GlobalSettings globalSettings)
|
||||
{
|
||||
_factory = new ConnectionFactory
|
||||
{
|
||||
HostName = globalSettings.EventLogging.RabbitMq.HostName,
|
||||
UserName = globalSettings.EventLogging.RabbitMq.Username,
|
||||
Password = globalSettings.EventLogging.RabbitMq.Password
|
||||
};
|
||||
_exchangeName = globalSettings.EventLogging.RabbitMq.IntegrationExchangeName;
|
||||
|
||||
_lazyConnection = new Lazy<Task<IConnection>>(CreateConnectionAsync);
|
||||
}
|
||||
|
||||
public async Task PublishAsync(IIntegrationMessage message)
|
||||
{
|
||||
var routingKey = message.IntegrationType.ToRoutingKey();
|
||||
var connection = await _lazyConnection.Value;
|
||||
await using var channel = await connection.CreateChannelAsync();
|
||||
|
||||
await channel.ExchangeDeclareAsync(exchange: _exchangeName, type: ExchangeType.Direct, durable: true);
|
||||
|
||||
var body = Encoding.UTF8.GetBytes(message.ToJson());
|
||||
|
||||
await channel.BasicPublishAsync(exchange: _exchangeName, routingKey: routingKey, body: body);
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_lazyConnection.IsValueCreated)
|
||||
{
|
||||
var connection = await _lazyConnection.Value;
|
||||
await connection.DisposeAsync();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<IConnection> CreateConnectionAsync()
|
||||
{
|
||||
return await _factory.CreateConnectionAsync();
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Nodes;
|
||||
using Bit.Core.AdminConsole.Models.Data.Integrations;
|
||||
using Bit.Core.Enums;
|
||||
using Bit.Core.Models.Data.Integrations;
|
||||
using Bit.Core.Repositories;
|
||||
|
||||
#nullable enable
|
||||
|
||||
@@ -0,0 +1,19 @@
|
||||
using Bit.Core.AdminConsole.Models.Data.Integrations;
|
||||
|
||||
namespace Bit.Core.Services;
|
||||
|
||||
public class SlackIntegrationHandler(
|
||||
ISlackService slackService)
|
||||
: IntegrationHandlerBase<SlackIntegrationConfigurationDetails>
|
||||
{
|
||||
public override async Task<IntegrationHandlerResult> HandleAsync(IntegrationMessage<SlackIntegrationConfigurationDetails> message)
|
||||
{
|
||||
await slackService.SendSlackMessageByChannelIdAsync(
|
||||
message.Configuration.token,
|
||||
message.RenderedTemplate,
|
||||
message.Configuration.channelId
|
||||
);
|
||||
|
||||
return new IntegrationHandlerResult(success: true, message: message);
|
||||
}
|
||||
}
|
||||
@@ -1,8 +1,8 @@
|
||||
using System.Text;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Nodes;
|
||||
using Bit.Core.AdminConsole.Models.Data.Integrations;
|
||||
using Bit.Core.Enums;
|
||||
using Bit.Core.Models.Data.Integrations;
|
||||
using Bit.Core.Repositories;
|
||||
|
||||
#nullable enable
|
||||
@@ -25,7 +25,7 @@ public class WebhookEventHandler(
|
||||
protected override async Task ProcessEventIntegrationAsync(JsonObject mergedConfiguration,
|
||||
string renderedTemplate)
|
||||
{
|
||||
var config = mergedConfiguration.Deserialize<WebhookIntegrationConfigurationDetils>();
|
||||
var config = mergedConfiguration.Deserialize<WebhookIntegrationConfigurationDetails>();
|
||||
if (config is null || string.IsNullOrEmpty(config.url))
|
||||
{
|
||||
return;
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
using System.Globalization;
|
||||
using System.Net;
|
||||
using System.Text;
|
||||
using Bit.Core.AdminConsole.Models.Data.Integrations;
|
||||
|
||||
#nullable enable
|
||||
|
||||
namespace Bit.Core.Services;
|
||||
|
||||
public class WebhookIntegrationHandler(IHttpClientFactory httpClientFactory)
|
||||
: IntegrationHandlerBase<WebhookIntegrationConfigurationDetails>
|
||||
{
|
||||
private readonly HttpClient _httpClient = httpClientFactory.CreateClient(HttpClientName);
|
||||
|
||||
public const string HttpClientName = "WebhookIntegrationHandlerHttpClient";
|
||||
|
||||
public override async Task<IntegrationHandlerResult> HandleAsync(IntegrationMessage<WebhookIntegrationConfigurationDetails> message)
|
||||
{
|
||||
var content = new StringContent(message.RenderedTemplate, Encoding.UTF8, "application/json");
|
||||
var response = await _httpClient.PostAsync(message.Configuration.url, content);
|
||||
var result = new IntegrationHandlerResult(success: response.IsSuccessStatusCode, message);
|
||||
|
||||
switch (response.StatusCode)
|
||||
{
|
||||
case HttpStatusCode.TooManyRequests:
|
||||
case HttpStatusCode.RequestTimeout:
|
||||
case HttpStatusCode.InternalServerError:
|
||||
case HttpStatusCode.BadGateway:
|
||||
case HttpStatusCode.ServiceUnavailable:
|
||||
case HttpStatusCode.GatewayTimeout:
|
||||
result.Retryable = true;
|
||||
result.FailureReason = response.ReasonPhrase;
|
||||
|
||||
if (response.Headers.TryGetValues("Retry-After", out var values))
|
||||
{
|
||||
var value = values.FirstOrDefault();
|
||||
if (int.TryParse(value, out var seconds))
|
||||
{
|
||||
// Retry-after was specified in seconds. Adjust DelayUntilDate by the requested number of seconds.
|
||||
result.DelayUntilDate = DateTime.UtcNow.AddSeconds(seconds);
|
||||
}
|
||||
else if (DateTimeOffset.TryParseExact(value,
|
||||
"r", // "r" is the round-trip format: RFC1123
|
||||
CultureInfo.InvariantCulture,
|
||||
DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal,
|
||||
out var retryDate))
|
||||
{
|
||||
// Retry-after was specified as a date. Adjust DelayUntilDate to the specified date.
|
||||
result.DelayUntilDate = retryDate.UtcDateTime;
|
||||
}
|
||||
}
|
||||
break;
|
||||
default:
|
||||
result.Retryable = false;
|
||||
result.FailureReason = response.ReasonPhrase;
|
||||
break;
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user