Egalware/mapo-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update metodo di scrittura sul DB dei dati statistiche

Browse Source
This commit is contained in:
Samuele E. Locatelli (W11-AI)
2026-04-08 09:47:46 +02:00
parent f8d5d1820c
commit 5da5639f48
1 changed files with 428 additions and 227 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MP.IOC/Services/MetricsDbFlushService.cs
+428 -227
View File
@@ -57,29 +57,10 @@ namespace MP.IOC.Services
Log.Info("Starting historical metrics recovery from Redis...");
var baseKey = _config.GetValue<string>("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<StatsAggregatedModel>();
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<IStatsAggrService>();
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<StatsDetailModel>();
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<IStatsDetailService>();
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<StatsAggregatedModel>();
var detailRecords = new List<StatsDetailModel>();
var cursor = 0L;
var pattern = $"{baseKey}:stats:hours:*";
var allHourIndices = new List<RedisKey>();
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<StatsDetailModel>();
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<IStatsAggrService>();
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<IStatsDetailService>();
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<RedisKey>();
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<StatsAggregatedModel>();
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<IStatsAggrService>();
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<RedisKey> allHourIndices)
{
try
{
var keysToDelete = new List<string>();
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<RedisKey> allDayIndices)
{
try
{
var keysToDelete = new List<string>();
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<string>("ServerConf:RedisBaseKey") ?? "MP_IOC";
var cursor = 0L;
var pattern = $"{baseKey}:stats:hours:*:{hourStart.ToString("yyyyMMddHH")}";
var matchingKeys = new List<RedisKey>();
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<StatsDetailModel>();
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<IStatsDetailService>();
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<IStatsDetailService>();
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<string>("ServerConf:RedisBaseKey") ?? "MP_IOC";
var cursor = 0L;
var pattern = $"{baseKey}:stats:days:*:{dayStart.ToString("yyyyMMdd")}";
var matchingKeys = new List<RedisKey>();
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<StatsAggregatedModel>();
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<IStatsAggrService>();
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");
}
}