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

Update metodi calcolo e flush statistiche

Browse Source
This commit is contained in:
Samuele E. Locatelli (W11-AI)
2026-04-07 11:56:06 +02:00
parent f7e0dc87fd
commit bf57560b30
2 changed files with 200 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MP.IOC/Services/MetricsDbFlushService.cs
+169 -182
View File
@@ -59,160 +59,8 @@ namespace MP.IOC.Services
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 ProcessExpiredDataAsync(utcNow, hourStart, dayStart);
await FlushCurrentDataAsync(utcNow, hourStart, dayStart);
}
private async Task ProcessExpiredDataAsync(DateTime utcNow, DateTime hourStart, DateTime dayStart)
{
try
{
var thirtyDaysAgo = utcNow.AddDays(-30);
var lastHour = hourStart.AddHours(-1);
var yesterday = dayStart.AddDays(-1);
var aggRange = new DtUtils.Periodo(DtUtils.PeriodSet.Today)
{
Inizio = thirtyDaysAgo,
Fine = utcNow
};
var aggrRecords = await _aggrService.GetFiltAsync(thirtyDaysAgo, utcNow);
var detailRecords = await _detailService.GetFiltAsync(thirtyDaysAgo, utcNow);
if (aggrRecords.Count == 0 && detailRecords.Count == 0)
{
Log.Debug("No expired data found to process");
return;
}
var keysToDelete = new List<string>();
var logMessages = new List<string>();
foreach (var method in aggrRecords.Select(r => r.Type).Distinct())
{
var hoursIndexKey = GetHoursIndexKey(method);
var daysIndexKey = GetDaysIndexKey(method);
if (_db.KeyExists(hoursIndexKey))
{
var lastHourKeys = await GetExpiredHourKeysAsync(method, lastHour);
if (lastHourKeys.Count > 0)
{
keysToDelete.AddRange(lastHourKeys);
logMessages.Add($"[PREVIEW] Would delete {lastHourKeys.Count} expired hourly buckets for method {method}");
}
}
if (_db.KeyExists(daysIndexKey))
{
var yesterdayKeys = await GetExpiredDayKeysAsync(method, yesterday);
if (yesterdayKeys.Count > 0)
{
keysToDelete.AddRange(yesterdayKeys);
logMessages.Add($"[PREVIEW] Would delete {yesterdayKeys.Count} expired daily buckets for method {method}");
}
}
}
Log.Info("Expired data preview: {count} records found", aggrRecords.Count + detailRecords.Count);
foreach (var msg in logMessages)
{
Log.Info(msg);
}
var deleteConfirmed = _config.GetValue<bool>("RouteMan:DeleteExpiredMetrics", false);
if (deleteConfirmed && keysToDelete.Count > 0)
{
await DeleteKeysAsync(keysToDelete.Distinct().ToList());
Log.Info("Deleted {count} expired Redis keys", keysToDelete.Count);
}
ProcessAggregatedRecords(aggrRecords, utcNow, hourStart);
ProcessDetailRecords(detailRecords, utcNow, hourStart);
}
catch (Exception ex)
{
Log.Error(ex, "Error processing expired data");
}
}
private async Task<List<string>> GetExpiredHourKeysAsync(string method, DateTime cutoff)
{
var keys = new List<string>();
var hoursIndexKey = GetHoursIndexKey(method);
if (!_db.KeyExists(hoursIndexKey)) return keys;
var allKeys = await _db.SortedSetRangeByRankAsync(hoursIndexKey, 0, -1);
foreach (var key in allKeys)
{
if (!key.HasValue) continue;
var hourFromKey = ExtractHourFromKey(key);
if (hourFromKey.HasValue && hourFromKey.Value < cutoff)
{
keys.Add(key.ToString());
}
}
return keys;
}
private async Task<List<string>> GetExpiredDayKeysAsync(string method, DateTime cutoff)
{
var keys = new List<string>();
var daysIndexKey = GetDaysIndexKey(method);
if (!_db.KeyExists(daysIndexKey)) return keys;
var allKeys = await _db.SortedSetRangeByRankAsync(daysIndexKey, 0, -1);
foreach (var key in allKeys)
{
if (!key.HasValue) continue;
var dayFromKey = ExtractDayFromKey(key);
if (dayFromKey.HasValue && dayFromKey.Value.Date < cutoff.Date)
{
keys.Add(key.ToString());
}
}
return keys;
}
private async Task DeleteKeysAsync(List<string> keys)
{
var batch = _db.CreateBatch();
var tasks = keys.Select(k => batch.KeyDeleteAsync(k)).ToList();
batch.Execute();
await Task.WhenAll(tasks.ToArray());
}
private static DateTime? ExtractHourFromKey(string key)
{
try
{
var parts = key.Split(':');
if (parts.Length >= 4 && DateTime.TryParseExact(parts[3], "yyyyMMddHH", null, DateTimeStyles.None, out var dt))
{
return dt;
}
}
catch { }
return null;
}
private static DateTime? ExtractDayFromKey(string key)
{
try
{
var parts = key.Split(':');
if (parts.Length >= 4 && DateTime.TryParseExact(parts[3], "yyyyMMdd", null, DateTimeStyles.None, out var dt))
{
return dt;
}
}
catch { }
return null;
await CleanExpiredRedisKeysAsync(hourStart, dayStart);
}
private async Task FlushCurrentDataAsync(DateTime utcNow, DateTime hourStart, DateTime dayStart)
@@ -236,6 +84,8 @@ namespace MP.IOC.Services
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 (count == 0) continue;
@@ -247,24 +97,31 @@ namespace MP.IOC.Services
Type = method,
RequestCount = count,
AvgDuration = avgDuration,
MaxDuration = totalMs,
MinDuration = 0,
MaxDuration = maxMs,
MinDuration = minMs,
NoReply = 0
};
aggrRecords.Add(aggrRecord);
foreach (var dest in stat.Destinations)
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 (detailCount == 0) continue;
var detailRecord = new StatsDetailModel
{
Environment = dest.Key,
Environment = destStat.Key,
Type = method,
Hour = hourStart,
RequestCount = dest.Value,
AvgDuration = totalMs / dest.Value,
MaxDuration = totalMs,
MinDuration = 0,
RequestCount = detailCount,
AvgDuration = detailTotalMs / detailCount,
MaxDuration = detailMaxMs,
MinDuration = detailMinMs,
NoReply = 0
};
@@ -292,46 +149,176 @@ namespace MP.IOC.Services
}
}
private void ProcessAggregatedRecords(List<StatsAggregatedModel> records, DateTime utcNow, DateTime hourStart)
private async Task CleanExpiredRedisKeysAsync(DateTime hourStart, DateTime dayStart)
{
var expiredToday = records.Where(r => r.Hour < hourStart && r.Hour.Date == utcNow.Date).ToList();
if (expiredToday.Count > 0)
try
{
Log.Info("Processing {count} hourly records from previous hour", expiredToday.Count);
}
var lastHour = hourStart.AddHours(-1);
var yesterday = dayStart.AddDays(-1);
var deleteConfirmed = _config.GetValue<bool>("RouteMan:DeleteExpiredMetrics", false);
var toUpsert = expiredToday.Concat(records.Where(r => r.Hour < hourStart)).ToList();
if (toUpsert.Count > 0)
if (!deleteConfirmed)
{
Log.Debug("Auto-delete of expired Redis keys disabled, skipping cleanup");
return;
}
var previewedKeys = new List<string>();
var deletedCount = 0;
var baseKey = _config.GetValue<string>("ServerConf:RedisBaseKey") ?? "MP_IOC";
// Get all unique methods from current snapshot and Redis indices
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;
// Preview expired hourly keys
if (_db.KeyExists(hoursIndexKey))
{
var expiredHourKeys = await GetExpiredHourKeysAsync(method, lastHour);
previewedKeys.AddRange(expiredHourKeys);
Log.Info("[PREVIEW] Would delete {count} expired hourly buckets for method {method}",
expiredHourKeys.Count, method);
}
// Preview expired daily keys
if (_db.KeyExists(daysIndexKey))
{
var expiredDayKeys = await GetExpiredDayKeysAsync(method, yesterday);
previewedKeys.AddRange(expiredDayKeys);
Log.Info("[PREVIEW] Would delete {count} expired daily buckets for method {method}",
expiredDayKeys.Count, method);
}
}
// Delete keys if any found
if (previewedKeys.Count > 0)
{
var uniqueKeys = previewedKeys.Distinct().ToList();
deletedCount = await DeleteKeysAsync(uniqueKeys);
Log.Info("Deleted {count} expired Redis keys", deletedCount);
}
else
{
Log.Debug("No expired Redis keys to delete");
}
}
catch (Exception ex)
{
_aggrService.UpsertManyAsync(toUpsert, false).Wait();
Log.Error(ex, "Error cleaning expired Redis keys");
}
}
private void ProcessDetailRecords(List<StatsDetailModel> records, DateTime utcNow, DateTime hourStart)
private async Task<List<string>> GetAllMethodsAsync()
{
var expiredToday = records.Where(r => r.Hour < hourStart && r.Hour.Date == utcNow.Date).ToList();
if (expiredToday.Count > 0)
var methods = new HashSet<string>();
var snapshot = _stats.Snapshot();
foreach (var method in snapshot.Keys)
{
Log.Info("Processing {count} detail records from previous hour", expiredToday.Count);
methods.Add(method);
}
var toUpsert = expiredToday.Concat(records.Where(r => r.Hour < hourStart)).ToList();
if (toUpsert.Count > 0)
{
_detailService.UpsertManyAsync(toUpsert, false).Wait();
}
// Try to get more methods from Redis indices if needed
return methods.ToList();
}
private string GetHoursIndexKey(string method)
private async Task<List<string>> GetExpiredHourKeysAsync(string method, DateTime cutoff)
{
var keys = new List<string>();
var baseKey = _config.GetValue<string>("ServerConf:RedisBaseKey") ?? "MP_IOC";
return $"{baseKey}:stats:hours:{method}";
var hoursIndexKey = $"{baseKey}:stats:hours:{method}";
if (!_db.KeyExists(hoursIndexKey)) return keys;
try
{
var allKeys = await _db.SortedSetRangeByRankAsync(hoursIndexKey, 0, -1);
foreach (var key in allKeys)
{
if (key == null) 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)
{
keys.Add(key.ToString());
}
}
}
}
catch (Exception ex)
{
Log.Error(ex, "Error reading expired hourly keys for method {method}", method);
}
return keys;
}
private string GetDaysIndexKey(string method)
private async Task<List<string>> GetExpiredDayKeysAsync(string method, DateTime cutoff)
{
var keys = new List<string>();
var baseKey = _config.GetValue<string>("ServerConf:RedisBaseKey") ?? "MP_IOC";
return $"{baseKey}:stats:days:{method}";
var daysIndexKey = $"{baseKey}:stats:days:{method}";
if (!_db.KeyExists(daysIndexKey)) return keys;
try
{
var allKeys = await _db.SortedSetRangeByRankAsync(daysIndexKey, 0, -1);
foreach (var key in allKeys)
{
if (key == null) 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)
{
keys.Add(key.ToString());
}
}
}
}
catch (Exception ex)
{
Log.Error(ex, "Error reading expired daily keys for method {method}", method);
}
return keys;
}
private async Task<int> DeleteKeysAsync(List<string> keys)
{
if (keys.Count == 0) return 0;
var successCount = 0;
var batch = _db.CreateBatch();
var tasks = new List<Task<long>>();
foreach (var key in keys)
{
tasks.Add(batch.KeyDeleteAsync(key));
}
batch.Execute();
var results = await Task.WhenAll(tasks);
successCount = results.Sum(r => r > 0 ? 1 : 0);
return successCount;
}
}
}
MP.IOC/Services/RouteStatsManager.cs
+31 -7
View File
@@ -1,19 +1,25 @@
using System.Collections.Concurrent;
using System.Collections.Concurrent;
namespace MP.IOC.Services
{
public class RouteStats
{
public long Count;
public TimeSpan TotalDuration = TimeSpan.Zero;
public TimeSpan MaxDuration = TimeSpan.Zero;
public TimeSpan MinDuration = TimeSpan.MaxValue;
public ConcurrentDictionary<string, long> Destinations = new();
public ConcurrentDictionary<string, DestinationStats> Destinations = new();
public ConcurrentDictionary<int, long> StatusCodes = new();
}
public class DestinationStats
{
public long Count;
public TimeSpan TotalDuration = TimeSpan.Zero;
public TimeSpan MaxDuration = TimeSpan.Zero;
public TimeSpan MinDuration = TimeSpan.MaxValue;
}
public class RouteStatsManager
{
private readonly ConcurrentDictionary<string, RouteStats> _map = new();
@@ -22,7 +28,9 @@ namespace MP.IOC.Services
{
var stat = _map.GetOrAdd(method, _ => new RouteStats());
Interlocked.Increment(ref stat.Count);
stat.Destinations.AddOrUpdate(destination, 1, (_, v) => v + 1);
var destStat = stat.Destinations.GetOrAdd(destination, _ => new DestinationStats());
Interlocked.Increment(ref destStat.Count);
}
public void RecordDuration(string method, TimeSpan duration)
@@ -32,7 +40,7 @@ namespace MP.IOC.Services
lock (stat)
{
stat.TotalDuration += duration;
// aggiorno valori min/max se necessario
if (stat.MaxDuration < duration)
{
stat.MaxDuration = duration;
@@ -41,6 +49,23 @@ namespace MP.IOC.Services
{
stat.MinDuration = duration;
}
foreach (var destStat in stat.Destinations.Values)
{
lock (destStat)
{
destStat.TotalDuration += duration;
if (destStat.MaxDuration < duration)
{
destStat.MaxDuration = duration;
}
if (destStat.MinDuration > duration)
{
destStat.MinDuration = duration;
}
}
}
}
}
}
@@ -63,5 +88,4 @@ namespace MP.IOC.Services
_map.Clear();
}
}
}