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

Test update logica Db scan redis

Browse Source
This commit is contained in:
Samuele E. Locatelli (W11-AI)
2026-04-08 11:12:24 +02:00
parent 421a9d769e
commit 4fbc11438b
1 changed files with 75 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MP.IOC/Services/MetricsDbFlushService.cs
+75 -64
View File
@@ -75,73 +75,84 @@ namespace MP.IOC.Services
try
{
var detailRecordsToInsert = new List<StatsDetailModel>();
var allHourIndexKeys = new List<string>();
var cursor = 0L;
var hourIndexPattern = $"{baseKey}:stats:hours:*";
var allHourIndices = new List<RedisKey>();
do
// Get all hours index keys by pattern matching on existing indices
var dests = await GetAllDestinationsAsync(baseKey);
foreach (var dest in dests)
{
var keysResult = await _db.KeysAsync(hourIndexPattern, cursor);
allHourIndices.AddRange(keysResult.Where(k => k != null));
cursor = keysResult.Cursor;
} while (cursor > 0 && allHourIndices.Count < 10000);
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 dayPart = parts[6];
var hourPart = parts[7];
if (!DateTime.TryParseExact($"{dayPart}{hourPart}", "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 hourIndexKey = $"{baseKey}:stats:hours:{dest}:*";
try
{
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 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;
detailRecordsToInsert.Add(new StatsDetailModel
var allKeys = await _db.KeysAsync(hourIndexKey, 0, -1);
foreach (var key in allKeys)
{
Destination = dest,
Type = method,
Hour = hourFromKey.Date.AddHours(hourFromKey.Hour),
RequestCount = count,
AvgDuration = avgDuration,
MinDuration = minMs,
MaxDuration = maxMs,
NoReply = 0
});
if (!key.IsNullOrEmpty)
{
allHourIndexKeys.Add(key.ToString());
// Read bucket keys from sorted set index
var bucketKeys = await _db.SortedSetRangeByRankAsync(key, 0, -1);
foreach (var bucketKeyRedis in bucketKeys)
{
if (bucketKeyRedis.IsNullOrEmpty) continue;
var bucketStr = bucketKeyRedis.ToString();
var parts = bucketStr.Split(':');
if (parts.Length < 7) continue;
// Parse date from key: {base}:stats:hour:{dest}:{method}:{yyyyMMddHH}
var dayHourPart = string.Join(":", parts.Skip(6));
if (!DateTime.TryParseExact(dayHourPart, "yyyyMMddHH",
CultureInfo.InvariantCulture, DateTimeStyles.None, out var hourFromKey))
continue;
if (hourFromKey > cutoff) continue;
// Read hash data from bucket
var hashData = await _db.HashGetAllAsync(bucketKeyRedis);
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 cp) ? cp : 0;
var totalMs = double.TryParse(totalMsVal.Value.ToString(), out var tp) ? tp : 0;
var maxMs = double.TryParse(maxMsVal.Value.ToString(), out var mp) ? mp : 0;
var minMs = double.TryParse(minMsVal.Value.ToString(), out var mip) ? mip : 0;
if (count == 0) continue;
var avgDuration = totalMs / count;
detailRecordsToInsert.Add(new StatsDetailModel
{
Destination = dest,
Type = parts[5],
Hour = hourFromKey.Date.AddHours(hourFromKey.Hour),
RequestCount = count,
AvgDuration = avgDuration,
MinDuration = minMs,
MaxDuration = maxMs,
NoReply = 0
});
}
}
}
}
catch (Exception ex)
{
Log.Error(ex, "Error reading hourly index for dest {dest}", dest);
}
}
@@ -163,7 +174,7 @@ namespace MP.IOC.Services
if (deleteConfirmed)
{
await DeleteExpiredHourlyKeysAsync(allHourIndices);
await DeleteExpiredHourlyKeysAsync(allHourIndexKeys);
}
}
}