From 5da5639f4833d939f7ccf4a8b98cf10fed0653d5 Mon Sep 17 00:00:00 2001 From: "Samuele E. Locatelli (W11-AI)" Date: Wed, 8 Apr 2026 09:47:46 +0200 Subject: [PATCH] Update metodo di scrittura sul DB dei dati statistiche --- MP.IOC/Services/MetricsDbFlushService.cs | 655 +++++++++++++++-------- 1 file changed, 428 insertions(+), 227 deletions(-) diff --git a/MP.IOC/Services/MetricsDbFlushService.cs b/MP.IOC/Services/MetricsDbFlushService.cs index c6ef4be5..68e01431 100644 --- a/MP.IOC/Services/MetricsDbFlushService.cs +++ b/MP.IOC/Services/MetricsDbFlushService.cs @@ -57,29 +57,10 @@ namespace MP.IOC.Services Log.Info("Starting historical metrics recovery from Redis..."); var baseKey = _config.GetValue("ServerConf:RedisBaseKey") ?? "MP_IOC"; - var lastHour = DateTime.Now.AddHours(-1); var yesterday = DateTime.Now.AddDays(-1).Date; - var methods = await GetAllMethodsAsync(); - - foreach (var method in methods) - { - var hoursIndexKey = $"{baseKey}:stats:hours:{method}"; - var daysIndexKey = $"{baseKey}:stats:days:{method}"; - - if (!_db.KeyExists(hoursIndexKey) && !_db.KeyExists(daysIndexKey)) continue; - - await RecoverHistoricalHourlyAsync(method, hoursIndexKey, lastHour); - - if (deleteConfirmed) - { - await RecoverHistoricalDailyAsync(method, daysIndexKey, yesterday); - } - else - { - Log.Info("[SKIP] Historical daily data for method {method} would be recovered but delete is disabled", method); - } - } + await RecoverHistoricalDailyAsync(baseKey, yesterday, deleteConfirmed); + await RecoverHistoricalHourlyAsync(baseKey, yesterday, deleteConfirmed); Log.Info("Historical metrics recovery completed"); } @@ -89,255 +70,475 @@ namespace MP.IOC.Services } } - private async Task RecoverHistoricalHourlyAsync(string method, string hoursIndexKey, DateTime cutoff) + private async Task RecoverHistoricalHourlyAsync(string baseKey, DateTime cutoff, bool deleteConfirmed) { try { - var allKeys = await _db.SortedSetRangeByRankAsync(hoursIndexKey, 0, -1); - var recordsToInsert = new List(); + var allHoursIndices = await _db.KeyExistsAsync($"{baseKey}:stats:hours:*"); - foreach (var key in allKeys) + if (!allHoursIndices) { - if (!key.HasValue) continue; - - var parts = key.ToString().Split(':'); - if (parts.Length >= 4 && DateTime.TryParseExact(parts[3], "yyyyMMddHH", - CultureInfo.InvariantCulture, DateTimeStyles.None, out var hourFromKey)) - { - if (hourFromKey <= cutoff) - { - var score = await _db.SortedSetScoreAsync(hoursIndexKey, key); - if (score.HasValue) - { - recordsToInsert.Add(new StatsAggregatedModel - { - Hour = hourFromKey.Date.AddHours(hourFromKey.Hour), - Destination = parts.Length > 4 ? parts[4] : "unknown", - RequestCount = (long)score.Value, - AvgDuration = 0, - MinDuration = 0, - MaxDuration = 0, - NoReply = 0 - }); - } - } - } - } - - if (recordsToInsert.Count > 0) - { - await using var scope = _scopeFactory.CreateAsyncScope(); - var _aggrService = scope.ServiceProvider.GetRequiredService(); - await _aggrService.UpsertManyAsync(recordsToInsert, false); - Log.Info($"[HISTORICAL] Recovered {recordsToInsert.Count} hourly records for method {method}"); - - var sampleCount = Math.Min(5, recordsToInsert.Count); - for (int i = 0; i < sampleCount; i++) - { - var r = recordsToInsert[i]; - Log.Info($"[HISTORICAL SAMPLE] Hour={r.Hour:yyyy-MM-dd HH:mm}, Destination={r.Destination}, Count={r.RequestCount}"); - } - } - } - catch (Exception ex) - { - Log.Error(ex, "Error recovering historical hourly data for method {method}", method); - } - } - - private async Task RecoverHistoricalDailyAsync(string method, string daysIndexKey, DateTime cutoff) - { - try - { - var allKeys = await _db.SortedSetRangeByRankAsync(daysIndexKey, 0, -1); - var recordsToInsert = new List(); - - foreach (var key in allKeys) - { - if (!key.HasValue) continue; - - var parts = key.ToString().Split(':'); - if (parts.Length >= 4 && DateTime.TryParseExact(parts[3], "yyyyMMdd", - CultureInfo.InvariantCulture, DateTimeStyles.None, out var dayFromKey)) - { - if (dayFromKey.Date <= cutoff.Date) - { - var score = await _db.SortedSetScoreAsync(daysIndexKey, key); - if (score.HasValue) - { - recordsToInsert.Add(new StatsDetailModel - { - Destination = parts.Length > 4 ? parts[4] : "unknown", - Type = method, - Hour = dayFromKey.Date, - RequestCount = (long)score.Value, - AvgDuration = 0, - MinDuration = 0, - MaxDuration = 0, - NoReply = 0 - }); - } - } - } - } - - if (recordsToInsert.Count > 0) - { - await using var scope = _scopeFactory.CreateAsyncScope(); - var _detailService = scope.ServiceProvider.GetRequiredService(); - await _detailService.UpsertManyAsync(recordsToInsert, false); - Log.Info("[HISTORICAL] Recovered {count} daily records for method {method}", recordsToInsert.Count, method); - - var sampleCount = Math.Min(5, recordsToInsert.Count); - for (int i = 0; i < sampleCount; i++) - { - var r = recordsToInsert[i]; - Log.Info("[HISTORICAL SAMPLE] Env={env}, Type={type}, Date={date}, Count={count}", - r.Destination, r.Type, r.Hour.ToString("yyyy-MM-dd"), r.RequestCount); - } - } - } - catch (Exception ex) - { - Log.Error(ex, "Error recovering historical daily data for method {method}", method); - } - } - - public async Task FlushMetricsAsync() - { - var utcNow = DateTime.Now; - var hourStart = new DateTime(utcNow.Year, utcNow.Month, utcNow.Day, utcNow.Hour, 0, 0); - var dayStart = new DateTime(utcNow.Year, utcNow.Month, utcNow.Day, 0, 0, 0); - - await FlushCurrentDataAsync(utcNow, hourStart, dayStart); - await CleanExpiredRedisKeysAsync(hourStart, dayStart); - } - - private async Task FlushCurrentDataAsync(DateTime utcNow, DateTime hourStart, DateTime dayStart) - { - try - { - var snapshot = _stats.Snapshot(); - - if (snapshot.Count == 0) - { - Log.Debug("No metrics in snapshot to flush"); + Log.Debug("No hourly indices found in Redis"); return; } - var aggrRecords = new List(); - var detailRecords = new List(); + var cursor = 0L; + var pattern = $"{baseKey}:stats:hours:*"; + var allHourIndices = new List(); - foreach (var kv in snapshot) + do { - string dest = "IO"; - string method = "NA"; - // verifico se ho chiave completo (dest+method) o parziale - string rawKey = kv.Key; - if (rawKey.Contains("|")) + var result = await _db.KeyScanAsync(pattern, cursor); + allHourIndices.AddRange(result.Keys.Where(k => k != null)); + cursor = result.Cursor; + } while (cursor > 0 && allHourIndices.Count < 10000); + + var detailRecordsToInsert = new List(); + + foreach (var hoursIndexKey in allHourIndices) + { + var keyStr = hoursIndexKey.ToString(); + var parts = keyStr.Split(':'); + + if (parts.Length < 6) continue; + + var dest = parts[4]; + var method = parts[5]; + var dayHourPart = string.Join(":", parts.Skip(6).TakeWhile(p => DateTime.TryParseExact(p, "yyyyMMddHH", CultureInfo.InvariantCulture, DateTimeStyles.None, out _))); + + if (!DateTime.TryParseExact(dayHourPart, "yyyyMMddHH", CultureInfo.InvariantCulture, DateTimeStyles.None, out var hourFromKey)) + continue; + + if (hourFromKey > cutoff) continue; + + var allBuckets = await _db.SortedSetRangeByRankAsync(hoursIndexKey, 0, -1); + + foreach (var bucketKey in allBuckets) { - var splitVal = rawKey.Split("|"); - dest = splitVal[0]; - method = splitVal[1]; - } - else - { - method = rawKey; - } -#if false - var method = kv.Key; -#endif - var stat = kv.Value; - var count = Interlocked.Read(ref stat.Count); - var totalMs = stat.TotalDuration.TotalMilliseconds; - var maxMs = stat.MaxDuration.TotalMilliseconds; - var minMs = (stat.MinDuration == TimeSpan.MaxValue) ? 0 : stat.MinDuration.TotalMilliseconds; + if (!bucketKey.HasValue) continue; - if (count == 0) continue; + var hashData = await _db.HashGetAllAsync(bucketKey); - var avgDuration = totalMs / count; + if (hashData.Length == 0) continue; - var finalMaxMs = (maxMs == 0) ? avgDuration : maxMs; - var finalMinMs = (minMs == 0) ? avgDuration : minMs; + var countVal = hashData.FirstOrDefault(f => f.Name.ToString() == "count"); + var totalMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "totalMs"); + var maxMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "maxMs"); + var minMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "minMs"); - var aggrRecord = new StatsAggregatedModel - { - Hour = hourStart, - Destination = dest, - RequestCount = count, - AvgDuration = avgDuration, - MaxDuration = finalMaxMs, - MinDuration = finalMinMs, - NoReply = 0 - }; + if (countVal.Value.IsNullOrEmpty || totalMsVal.Value.IsNullOrEmpty) continue; - aggrRecords.Add(aggrRecord); + var count = long.TryParse(countVal.Value.ToString(), out var count) ? count : 0; + var totalMs = double.TryParse(totalMsVal.Value.ToString(), out var totalMs) ? totalMs : 0; + var maxMs = double.TryParse(maxMsVal.Value.ToString(), out var maxMs) ? maxMs : 0; + var minMs = double.TryParse(minMsVal.Value.ToString(), out var minMs) ? minMs : 0; -#if false - foreach (var destStat in stat.Destinations) - { - var detailCount = Interlocked.Read(ref destStat.Value.Count); - var detailTotalMs = destStat.Value.TotalDuration.TotalMilliseconds; - var detailMaxMs = destStat.Value.MaxDuration.TotalMilliseconds; - var detailMinMs = (destStat.Value.MinDuration == TimeSpan.MaxValue) ? 0 : destStat.Value.MinDuration.TotalMilliseconds; + if (count == 0) continue; - if (detailCount == 0) continue; + var avgDuration = totalMs / count; - var finalDetailMaxMs = (detailMaxMs == 0) ? (detailTotalMs / detailCount) : detailMaxMs; - var finalDetailMinMs = (detailMinMs == 0) ? (detailTotalMs / detailCount) : detailMinMs; - - var detailRecord = new StatsDetailModel + detailRecordsToInsert.Add(new StatsDetailModel { - Destination = destStat.Key, + Destination = dest, Type = method, - Hour = hourStart, - RequestCount = detailCount, - AvgDuration = detailTotalMs / detailCount, - MaxDuration = finalDetailMaxMs, - MinDuration = finalDetailMinMs, + Hour = hourFromKey.Date.AddHours(hourFromKey.Hour), + RequestCount = count, + AvgDuration = avgDuration, + MinDuration = minMs, + MaxDuration = maxMs, NoReply = 0 - }; - - detailRecords.Add(detailRecord); + }); } -#endif } - await using var scope = _scopeFactory.CreateAsyncScope(); - if (aggrRecords.Count > 0) - { - var _aggrService = scope.ServiceProvider.GetRequiredService(); - await _aggrService.UpsertManyAsync(aggrRecords, false); - Log.Info("Flushed {count} current aggregated stats records", aggrRecords.Count); - var sampleCount = Math.Min(10, aggrRecords.Count); + if (detailRecordsToInsert.Count > 0) + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var detailService = scope.ServiceProvider.GetRequiredService(); + await detailService.UpsertManyAsync(detailRecordsToInsert, false); + + Log.Info("[HISTORICAL] Recovered {count} hourly detail records from Redis", detailRecordsToInsert.Count); + + var sampleCount = Math.Min(5, detailRecordsToInsert.Count); for (int i = 0; i < sampleCount; i++) { - var r = aggrRecords[i]; - Log.Info("[SAMPLE] INSERT Aggregated: Hour={hour}, Destination={Destination}, Count={count}, Avg={avg:F2}ms, Min={min:F2}ms, Max={max:F2}ms", - r.Hour.ToString("yyyy-MM-dd HH:mm"), r.Destination, r.RequestCount, r.AvgDuration, r.MinDuration, r.MaxDuration); + var r = detailRecordsToInsert[i]; + Log.Info("[HISTORICAL SAMPLE] Env={env}, Type={type}, Hour={hour}, Count={count}, Avg={avg:F2}ms", + r.Destination, r.Type, r.Hour.ToString("yyyy-MM-dd HH:mm"), r.RequestCount, r.AvgDuration); + } + + if (deleteConfirmed) + { + await DeleteExpiredHourlyKeysAsync(allHourIndices); + } + } + } + catch (Exception ex) + { + Log.Error(ex, "Error recovering historical hourly data from Redis"); + } + } + + private async Task RecoverHistoricalDailyAsync(string baseKey, DateTime cutoff, bool deleteConfirmed) + { + try + { + var cursor = 0L; + var pattern = $"{baseKey}:stats:days:*"; + var allDayIndices = new List(); + + do + { + var result = await _db.KeyScanAsync(pattern, cursor); + allDayIndices.AddRange(result.Keys.Where(k => k != null)); + cursor = result.Cursor; + } while (cursor > 0 && allDayIndices.Count < 10000); + + var aggrRecordsToInsert = new List(); + + foreach (var daysIndexKey in allDayIndices) + { + var keyStr = daysIndexKey.ToString(); + var parts = keyStr.Split(':'); + + if (parts.Length < 5) continue; + + var dest = parts[4]; + + var dayPart = string.Join(":", parts.Skip(5).TakeWhile(p => DateTime.TryParseExact(p, "yyyyMMdd", CultureInfo.InvariantCulture, DateTimeStyles.None, out _))); + + if (!DateTime.TryParseExact(dayPart, "yyyyMMdd", CultureInfo.InvariantCulture, DateTimeStyles.None, out var dayFromKey)) + continue; + + if (dayFromKey.Date > cutoff) continue; + + var allBuckets = await _db.SortedSetRangeByRankAsync(daysIndexKey, 0, -1); + + foreach (var bucketKey in allBuckets) + { + if (!bucketKey.HasValue) continue; + + var hashData = await _db.HashGetAllAsync(bucketKey); + + if (hashData.Length == 0) continue; + + var countVal = hashData.FirstOrDefault(f => f.Name.ToString() == "count"); + var totalMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "totalMs"); + var maxMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "maxMs"); + var minMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "minMs"); + + if (countVal.Value.IsNullOrEmpty || totalMsVal.Value.IsNullOrEmpty) continue; + + var count = long.TryParse(countVal.Value.ToString(), out var countValParsed) ? countValParsed : 0; + var totalMs = double.TryParse(totalMsVal.Value.ToString(), out var totalMsValParsed) ? totalMsValParsed : 0; + var maxMs = double.TryParse(maxMsVal.Value.ToString(), out var maxMsValParsed) ? maxMsValParsed : 0; + var minMs = double.TryParse(minMsVal.Value.ToString(), out var minMsValParsed) ? minMsValParsed : 0; + + if (count == 0) continue; + + var avgDuration = totalMs / count; + + aggrRecordsToInsert.Add(new StatsAggregatedModel + { + Destination = dest, + Hour = dayFromKey.Date, + RequestCount = count, + AvgDuration = avgDuration, + MinDuration = minMs, + MaxDuration = maxMs, + NoReply = 0 + }); + } + } + + if (aggrRecordsToInsert.Count > 0) + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var aggrService = scope.ServiceProvider.GetRequiredService(); + await aggrService.UpsertManyAsync(aggrRecordsToInsert, false); + + Log.Info("[HISTORICAL] Recovered {count} daily aggregated records from Redis", aggrRecordsToInsert.Count); + + var sampleCount = Math.Min(5, aggrRecordsToInsert.Count); + for (int i = 0; i < sampleCount; i++) + { + var r = aggrRecordsToInsert[i]; + Log.Info("[HISTORICAL SAMPLE] Env={env}, Date={date}, Count={count}, Avg={avg:F2}ms", + r.Destination, r.Hour.ToString("yyyy-MM-dd"), r.RequestCount, r.AvgDuration); + } + + if (deleteConfirmed) + { + await DeleteExpiredDailyKeysAsync(allDayIndices); + } + } + } + catch (Exception ex) + { + Log.Error(ex, "Error recovering historical daily data from Redis"); + } + } + + private async Task DeleteExpiredHourlyKeysAsync(List allHourIndices) + { + try + { + var keysToDelete = new List(); + + foreach (var hoursIndexKey in allHourIndices) + { + var allBuckets = await _db.SortedSetRangeByRankAsync(hoursIndexKey, 0, -1); + keysToDelete.AddRange(allBuckets.Select(k => k.ToString()).Where(s => !string.IsNullOrEmpty(s))); + } + + if (keysToDelete.Count > 0) + { + var uniqueKeys = keysToDelete.Distinct().ToList(); + var deletedCount = await DeleteKeysAsync(uniqueKeys); + Log.Info("[DELETE] Removed {count} expired hourly bucket keys from Redis", deletedCount); + } + } + catch (Exception ex) + { + Log.Error(ex, "Error deleting expired hourly keys from Redis"); + } + } + + private async Task DeleteExpiredDailyKeysAsync(List allDayIndices) + { + try + { + var keysToDelete = new List(); + + foreach (var daysIndexKey in allDayIndices) + { + var allBuckets = await _db.SortedSetRangeByRankAsync(daysIndexKey, 0, -1); + keysToDelete.AddRange(allBuckets.Select(k => k.ToString()).Where(s => !string.IsNullOrEmpty(s))); + } + + if (keysToDelete.Count > 0) + { + var uniqueKeys = keysToDelete.Distinct().ToList(); + var deletedCount = await DeleteKeysAsync(uniqueKeys); + Log.Info("[DELETE] Removed {count} expired daily bucket keys from Redis", deletedCount); + } + } + catch (Exception ex) + { + Log.Error(ex, "Error deleting expired daily keys from Redis"); + } + } + + public async Task FlushLiveMetricsAsync() + { + try + { + var utcNow = DateTime.Now; + var hourStart = new DateTime(utcNow.Year, utcNow.Month, utcNow.Day, utcNow.Hour, 0, 0); + + await FlushCurrentHourlyDataAsync(utcNow, hourStart); + } + catch (Exception ex) + { + Log.Error(ex, "Error flushing live metrics from Redis"); + } + } + + private async Task FlushCurrentHourlyDataAsync(DateTime utcNow, DateTime hourStart) + { + try + { + var baseKey = _config.GetValue("ServerConf:RedisBaseKey") ?? "MP_IOC"; + + var cursor = 0L; + var pattern = $"{baseKey}:stats:hours:*:{hourStart.ToString("yyyyMMddHH")}"; + var matchingKeys = new List(); + + do + { + var result = await _db.KeyScanAsync(pattern, cursor); + matchingKeys.AddRange(result.Keys.Where(k => k != null)); + cursor = result.Cursor; + } while (cursor > 0 && matchingKeys.Count < 10000); + + if (matchingKeys.Count == 0) + { + Log.Debug("No current hourly buckets found in Redis for flush"); + return; + } + + var detailRecords = new List(); + + foreach (var bucketKey in matchingKeys) + { + var keyStr = bucketKey.ToString(); + var parts = keyStr.Split(':'); + + if (parts.Length < 7) continue; + + var dest = parts[4]; + var method = parts[5]; + + try + { + var hashData = await _db.HashGetAllAsync(bucketKey); + + if (hashData.Length == 0) continue; + + var countVal = hashData.FirstOrDefault(f => f.Name.ToString() == "count"); + var totalMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "totalMs"); + var maxMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "maxMs"); + var minMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "minMs"); + + if (countVal.Value.IsNullOrEmpty || totalMsVal.Value.IsNullOrEmpty) continue; + + var count = long.TryParse(countVal.Value.ToString(), out var countParsed) ? countParsed : 0; + var totalMs = double.TryParse(totalMsVal.Value.ToString(), out var totalMsParsed) ? totalMsParsed : 0; + var maxMs = double.TryParse(maxMsVal.Value.ToString(), out var maxMsParsed) ? maxMsParsed : 0; + var minMs = double.TryParse(minMsVal.Value.ToString(), out var minMsParsed) ? minMsParsed : 0; + + if (count == 0) continue; + + var avgDuration = totalMs / count; + + detailRecords.Add(new StatsDetailModel + { + Destination = dest, + Type = method, + Hour = hourStart, + RequestCount = count, + AvgDuration = avgDuration, + MinDuration = minMs, + MaxDuration = maxMs, + NoReply = 0 + }); + } + catch (Exception ex) + { + Log.Error(ex, "Error processing hourly bucket key {key}", keyStr); } } if (detailRecords.Count > 0) { - var _detailService = scope.ServiceProvider.GetRequiredService(); - await _detailService.UpsertManyAsync(detailRecords, false); - Log.Info("Flushed {count} current detail stats records", detailRecords.Count); + await using var scope = _scopeFactory.CreateAsyncScope(); + var detailService = scope.ServiceProvider.GetRequiredService(); + await detailService.UpsertManyAsync(detailRecords, false); + + Log.Info("[LIVE] Flushed {count} current hourly detail records from Redis", detailRecords.Count); var sampleCount = Math.Min(10, detailRecords.Count); for (int i = 0; i < sampleCount; i++) { var r = detailRecords[i]; - Log.Info("[SAMPLE] INSERT Detail: Env={env}, Type={type}, Hour={hour}, Count={count}, Avg={avg:F2}ms, Min={min:F2}ms, Max={max:F2}ms", + Log.Info("[LIVE SAMPLE] Env={env}, Type={type}, Hour={hour}, Count={count}, Avg={avg:F2}ms, Min={min:F2}ms, Max={max:F2}ms", r.Destination, r.Type, r.Hour.ToString("yyyy-MM-dd HH:mm"), r.RequestCount, r.AvgDuration, r.MinDuration, r.MaxDuration); } - } - _stats.Clear(); + await DeleteCurrentHourlyBucketKeysAsync(matchingKeys); + } } catch (Exception ex) { - Log.Error(ex, "Error processing current metrics for database flush"); + Log.Error(ex, "Error flushing current hourly data from Redis"); + } + } + + private async Task FlushCurrentDailyDataAsync(DateTime utcNow, DateTime dayStart) + { + try + { + var baseKey = _config.GetValue("ServerConf:RedisBaseKey") ?? "MP_IOC"; + + var cursor = 0L; + var pattern = $"{baseKey}:stats:days:*:{dayStart.ToString("yyyyMMdd")}"; + var matchingKeys = new List(); + + do + { + var result = await _db.KeyScanAsync(pattern, cursor); + matchingKeys.AddRange(result.Keys.Where(k => k != null)); + cursor = result.Cursor; + } while (cursor > 0 && matchingKeys.Count < 10000); + + if (matchingKeys.Count == 0) + { + Log.Debug("No current daily buckets found in Redis for flush"); + return; + } + + var aggrRecords = new List(); + + foreach (var bucketKey in matchingKeys) + { + var keyStr = bucketKey.ToString(); + var parts = keyStr.Split(':'); + + if (parts.Length < 6) continue; + + var dest = parts[4]; + + try + { + var hashData = await _db.HashGetAllAsync(bucketKey); + + if (hashData.Length == 0) continue; + + var countVal = hashData.FirstOrDefault(f => f.Name.ToString() == "count"); + var totalMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "totalMs"); + var maxMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "maxMs"); + var minMsVal = hashData.FirstOrDefault(f => f.Name.ToString() == "minMs"); + + if (countVal.Value.IsNullOrEmpty || totalMsVal.Value.IsNullOrEmpty) continue; + + var count = long.TryParse(countVal.Value.ToString(), out var countParsed) ? countParsed : 0; + var totalMs = double.TryParse(totalMsVal.Value.ToString(), out var totalMsParsed) ? totalMsParsed : 0; + var maxMs = double.TryParse(maxMsVal.Value.ToString(), out var maxMsParsed) ? maxMsParsed : 0; + var minMs = double.TryParse(minMsVal.Value.ToString(), out var minMsParsed) ? minMsParsed : 0; + + if (count == 0) continue; + + var avgDuration = totalMs / count; + + aggrRecords.Add(new StatsAggregatedModel + { + Destination = dest, + Hour = dayStart.Date, + RequestCount = count, + AvgDuration = avgDuration, + MinDuration = minMs, + MaxDuration = maxMs, + NoReply = 0 + }); + } + catch (Exception ex) + { + Log.Error(ex, "Error processing daily bucket key {key}", keyStr); + } + } + + if (aggrRecords.Count > 0) + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var aggrService = scope.ServiceProvider.GetRequiredService(); + await aggrService.UpsertManyAsync(aggrRecords, false); + + Log.Info("[LIVE] Flushed {count} current daily aggregated records from Redis", aggrRecords.Count); + + var sampleCount = Math.Min(10, aggrRecords.Count); + for (int i = 0; i < sampleCount; i++) + { + var r = aggrRecords[i]; + Log.Info("[LIVE SAMPLE] Env={env}, Date={date}, Count={count}, Avg={avg:F2}ms", + r.Destination, r.Hour.ToString("yyyy-MM-dd"), r.RequestCount, r.AvgDuration); + } + + await DeleteCurrentDailyBucketKeysAsync(matchingKeys); + } + } + catch (Exception ex) + { + Log.Error(ex, "Error flushing current daily data from Redis"); } }