92 lines
3.2 KiB
C#
92 lines
3.2 KiB
C#
using Confluent.Kafka;
|
|
using Confluent.Kafka.Admin;
|
|
using Microsoft.Extensions.Options;
|
|
|
|
public sealed class KafkaConsumerLagCollector : BackgroundService
|
|
{
|
|
private static readonly string[] Groups =
|
|
{
|
|
"es-indexer",
|
|
"sepsis-engine",
|
|
"notification-publisher",
|
|
"data-lake-writer",
|
|
};
|
|
|
|
private readonly KafkaOptions _kafkaOptions;
|
|
private readonly ClinicalMetrics _metrics;
|
|
private readonly ILogger<KafkaConsumerLagCollector> _logger;
|
|
|
|
public KafkaConsumerLagCollector(
|
|
IOptions<KafkaOptions> kafkaOptions,
|
|
ClinicalMetrics metrics,
|
|
ILogger<KafkaConsumerLagCollector> logger)
|
|
{
|
|
_kafkaOptions = kafkaOptions.Value;
|
|
_metrics = metrics;
|
|
_logger = logger;
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken ct)
|
|
{
|
|
// Initial delay so Kafka is reachable before the first collection.
|
|
await Task.Delay(TimeSpan.FromSeconds(20), ct);
|
|
|
|
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(30));
|
|
while (await timer.WaitForNextTickAsync(ct))
|
|
{
|
|
foreach (var group in Groups)
|
|
{
|
|
try { await CollectGroupLagAsync(group, ct); }
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogDebug(ex, "KafkaConsumerLagCollector: failed for group {Group}", group);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task CollectGroupLagAsync(string groupId, CancellationToken ct)
|
|
{
|
|
var adminConfig = new AdminClientConfig
|
|
{ BootstrapServers = _kafkaOptions.BootstrapServers }
|
|
.ApplySecurity(_kafkaOptions);
|
|
|
|
using var admin = new AdminClientBuilder(adminConfig).Build();
|
|
|
|
// List the committed offsets for this consumer group across all its partitions.
|
|
var result = await admin.ListConsumerGroupOffsetsAsync(
|
|
new[] { new ConsumerGroupTopicPartitions(groupId, null) },
|
|
new ListConsumerGroupOffsetsOptions { RequireStableOffsets = false });
|
|
|
|
var partitions = result.FirstOrDefault()?.Partitions ?? [];
|
|
if (!partitions.Any())
|
|
{
|
|
_metrics.KafkaConsumerLag.WithLabels(groupId).Set(0);
|
|
return;
|
|
}
|
|
|
|
// Query high watermarks using a temporary consumer (does not join the group).
|
|
using var tempConsumer = new ConsumerBuilder<Ignore, Ignore>(new ConsumerConfig
|
|
{
|
|
BootstrapServers = _kafkaOptions.BootstrapServers,
|
|
GroupId = $"__lag-probe",
|
|
}.ApplySecurity(_kafkaOptions)).Build();
|
|
|
|
long totalLag = 0;
|
|
foreach (var tpo in partitions)
|
|
{
|
|
if (tpo.Error.IsError || tpo.Offset == Offset.Unset) continue;
|
|
|
|
var watermarks = tempConsumer.QueryWatermarkOffsets(
|
|
tpo.TopicPartition, TimeSpan.FromSeconds(5));
|
|
|
|
// Committed offset is the next offset the consumer will read.
|
|
// High watermark is the latest available offset + 1.
|
|
// Lag = messages the consumer has not yet read.
|
|
var lag = watermarks.High.Value - tpo.Offset.Value;
|
|
totalLag += Math.Max(0L, lag);
|
|
}
|
|
|
|
_metrics.KafkaConsumerLag.WithLabels(groupId).Set(totalLag);
|
|
}
|
|
} |