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

ancora update metodi DbFlush

Browse Source
This commit is contained in:
Samuele Locatelli
2026-04-08 13:13:40 +02:00
parent d6d7b9ff61
commit 7f825f2a09
9 changed files with 44 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
MP.Data/Repository/Utils/StatsAggrRepository.cs
+1 -36
View File
@@ -81,44 +81,9 @@ namespace MP.Data.Repository.Utils
}
}
// ora preparo inserimento massivo
var existingRecords = await dbCtx
.DbSetStatsAggr
.Where(x => listRecords.Select(r => r.Hour).Contains(x.Hour) &&
listRecords.Select(r => r.Destination).Contains(x.Destination))
.ToListAsync();
if (existingRecords.Count > 0)
{
var existingHours = existingRecords.Select(r => new { r.Hour, r.Destination }).ToHashSet();
var recordsToUpdate = listRecords.Where(r =>
existingHours.Contains(new { Hour = r.Hour, Destination = r.Destination })).ToList();
var recordsToAdd = listRecords.Where(r =>
!existingHours.Contains(new { Hour = r.Hour, Destination = r.Destination })).ToList();
foreach (var updateRec in recordsToUpdate)
{
var existing = existingRecords.First(x => x.Hour == updateRec.Hour && x.Destination == updateRec.Destination);
existing.RequestCount = updateRec.RequestCount;
existing.AvgDuration = updateRec.AvgDuration;
existing.MaxDuration = updateRec.MaxDuration;
existing.MinDuration = updateRec.MinDuration;
existing.NoReply = updateRec.NoReply;
}
if (recordsToAdd.Count > 0)
{
await dbCtx.DbSetStatsAggr.AddRangeAsync(recordsToAdd);
}
}
else
{
await dbCtx
await dbCtx
.DbSetStatsAggr
.AddRangeAsync(listRecords);
}
// salvo!
answ = await dbCtx.SaveChangesAsync();
MP.Data/Repository/Utils/StatsDetailRepository.cs
+1 -35
View File
@@ -102,43 +102,9 @@ namespace MP.Data.Repository.Utils
}
}
// ora preparo inserimento massivo
var existingRecords = await dbCtx
.DbSetStatsDet
.Where(x => listRecords.Select(r => new { r.Hour, r.Destination, r.Type }).Contains(new { Hour = x.Hour, Destination = x.Destination, Type = x.Type }))
.ToListAsync();
if (existingRecords.Count > 0)
{
var existingKeys = existingRecords.Select(r => new { r.Hour, r.Destination, r.Type }).ToHashSet();
var recordsToUpdate = listRecords.Where(r =>
existingKeys.Contains(new { Hour = r.Hour, Destination = r.Destination, Type = r.Type })).ToList();
var recordsToAdd = listRecords.Where(r =>
!existingKeys.Contains(new { Hour = r.Hour, Destination = r.Destination, Type = r.Type })).ToList();
foreach (var updateRec in recordsToUpdate)
{
var existing = Enumerable.First<StatsDetailModel>((IEnumerable<StatsDetailModel>)existingRecords, (Func<StatsDetailModel, bool>)(x => x.Hour == updateRec.Hour && x.Destination == updateRec.Destination && x.Type == updateRec.Type));
existing.RequestCount = updateRec.RequestCount;
existing.AvgDuration = updateRec.AvgDuration;
existing.MaxDuration = updateRec.MaxDuration;
existing.MinDuration = updateRec.MinDuration;
existing.NoReply = updateRec.NoReply;
}
if (recordsToAdd.Count > 0)
{
await dbCtx.DbSetStatsDet.AddRangeAsync(recordsToAdd);
}
}
else
{
await dbCtx
await dbCtx
.DbSetStatsDet
.AddRangeAsync(listRecords);
}
// salvo!
answ = await dbCtx.SaveChangesAsync();
MP.IOC/MP.IOC.csproj
-4
View File
@@ -6,10 +6,6 @@
<ImplicitUsings>enable</ImplicitUsings>
</PropertyGroup>
<ItemGroup>
<Compile Remove="Services\MetricsDbFlushService.cs" />
</ItemGroup>
<ItemGroup>
<_WebToolingArtifacts Remove="Properties\PublishProfiles\IIS01.pubxml" />
<_WebToolingArtifacts Remove="Properties\PublishProfiles\IIS03.pubxml" />
MP.IOC/Program.cs
+1 -1
View File
@@ -62,8 +62,8 @@ logger.Info("YARP reverse proxy configured");
builder.Services.AddSingleton<PreserveBodyTransformer>();
builder.Services.AddSingleton<RouteStatsManager>();
builder.Services.AddHostedService<MetricsCalcService>();
builder.Services.AddHostedService<MetricsDbFlushService>();
#if false
builder.Services.AddHostedService<MetricsDbFlushService>();
#endif
// MP.Data DbContext for Stats repositories
MP.IOC/Resources/ChangeLog.html
+1 -1
View File
@@ -1,6 +1,6 @@
<body>
<i>Modulo MP-IOC </i>
<h4>Versione: 6.16.2604.812</h4>
<h4>Versione: 6.16.2604.813</h4>
<br /> Note di rilascio:
<ul>
<li>
MP.IOC/Resources/VersNum.txt
+1 -1
View File
@@ -1 +1 @@
6.16.2604.812
6.16.2604.813
MP.IOC/Resources/manifest.xml
+1 -1
View File
@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="UTF-8"?>
<item>
<version>6.16.2604.812</version>
<version>6.16.2604.813</version>
<url>https://nexus.steamware.net/repository/SWS/MP-IOC/stable/LAST/MP.IOC.zip</url>
<changelog>https://nexus.steamware.net/repository/SWS/MP-IOC/stable/LAST/ChangeLog.html</changelog>
<mandatory>false</mandatory>
MP.IOC/Services/MetricsDbFlushService.cs
+36 -44
View File
@@ -27,14 +27,15 @@ namespace MP.IOC.Services
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
{
var interval = _intOrDefault("RouteMan:MetricFlushIntervalSeconds", 60);
//var interval = _config.GetValue<int>("RouteMan:MetricFlushIntervalSeconds", 120);
var interval = _config.GetValue<int>("RouteMan:MetricFlushIntervalSeconds", 300);
while (!stoppingToken.IsCancellationRequested)
{
try
{
await Task.Delay(TimeSpan.FromSeconds(interval), stoppingToken);
await FlushLiveMetricsAsync();
await Task.Delay(TimeSpan.FromSeconds(interval), stoppingToken);
}
catch (TaskCanceledException) { break; }
catch (Exception ex)
@@ -48,15 +49,12 @@ namespace MP.IOC.Services
#region Private Fields
// Helper per il parsing delle date
private static readonly CultureInfo DateTimeIntl = CultureInfo.InvariantCulture;
private const double SentinelValue = 999999999;
private static readonly Logger Log = LogManager.GetCurrentClassLogger();
private readonly IConfiguration _config;
private readonly IDatabase _db;
private readonly IConnectionMultiplexer _mux;
private readonly IServiceScopeFactory _scopeFactory;
private double SentinelValue = 999999999;
#endregion Private Fields
@@ -68,18 +66,15 @@ namespace MP.IOC.Services
#region Private Methods
private int _intOrDefault(string key, int defaultValue)
=> _config.GetValue<int>(key, defaultValue);
private async Task FlushLiveMetricsAsync()
{
var detailRecordsToInsert = new List<StatsDetailModel>();
var keysToDelete = new List<RedisKey>();
// 1. Configurazione
bool deleteConfirmed = _config.GetValue<bool>("RouteMan:DeleteExpiredMetrics", false);
DateTime now = DateTime.Now;
// Definiamo i confini dell' "attuale" (ciò che non deve essere toccato)
// Confini temporali per proteggere i dati in corso
DateTime currentHourStart = new DateTime(now.Year, now.Month, now.Day, now.Hour, 0, 0);
DateTime currentDayStart = new DateTime(now.Year, now.Month, now.Day, 0, 0, 0);
@@ -88,6 +83,12 @@ namespace MP.IOC.Services
foreach (var endpoint in endpoints)
{
var server = _mux.GetServer(endpoint);
if (server.IsReplica)
{
continue;
}
string[] patternsToScan = {
$"{_redisBaseKey}:stats:hours:*",
$"{_redisBaseKey}:stats:days:*"
@@ -95,36 +96,33 @@ namespace MP.IOC.Services
foreach (var pattern in patternsToScan)
{
// Nota: KeyScanAsync/KeysAsync e' disponibile su IServer
await foreach (var indexKey in server.KeysAsync(pattern: pattern))
{
if (string.IsNullOrEmpty($"{indexKey}")) continue;
// Leggiamo i membri dell'indice (le chiavi delle singole ore/giorni)
var memberKeys = await _db.SortedSetRangeByRankAsync(indexKey);
// CORREZIONE: Utilizzo di SortedSetRangeByRankAsync con range 0 a -1 per prendere tutto il set
var memberKeys = await _db.SortedSetRangeByRankAsync(indexKey, 0, -1);
foreach (var statKey in memberKeys)
{
if (!TryParseKeyMetadata(statKey, out string dest, out string method, out DateTime timestamp, out bool isHourType))
var sKey = (RedisKey)$"{statKey}";
if (!TryParseKeyMetadata(sKey, out string dest, out string method, out DateTime timestamp, out bool isHourType))
continue;
// --- LOGICA DI CONTROLLO SCADENZA ---
// Una chiave è "scaduta" se il suo timestamp è precedente all'inizio dell'intervallo corrente
// Verifica se la chiave � scaduta rispetto all'orario corrente
bool isExpired = isHourType
? timestamp < currentHourStart
: timestamp < currentDayStart;
// Se la chiave è scaduta e l'utente ha abilitato la pulizia, segniamola per la cancellazione
// Se � scaduta e abbiamo il permesso, segnamola per la cancellazione
if (isExpired && deleteConfirmed)
{
keysToDelete.Add(statKey);
keysToDelete.Add(sKey);
}
// --- LOGICA DI RECUPERO DATI PER IL DB ---
// Non processiamo per il DB i dati che stiamo per cancellare (per evitare duplicati o errori di race condition)
// In realtà, processiamo tutto ciò che è nell'indice, ma la logica di "sicurezza"
// è che se è scaduto lo carichiamo al DB e poi lo eliminiamo.
var hashData = await _db.HashGetAllAsync(statKey);
// Recupero dati dalla Hash
var hashData = await _db.HashGetAllAsync(sKey);
if (hashData.Length == 0) continue;
var dict = hashData.ToDictionary(x => x.Name.ToString(), x => x.Value.ToString());
@@ -133,13 +131,17 @@ namespace MP.IOC.Services
dict.TryGetValue("totalMs", out var totalMsStr))
{
long count = long.Parse(countStr);
if (count <= 0) continue;
count = long.Parse(countStr);
double totalMs = double.Parse(totalMsStr, CultureInfo.InvariantCulture);
double maxMs = dict.ContainsKey("maxMs") ? double.Parse(dict["maxMs"], CultureInfo.InvariantCulture) : 0;
double minMs = dict.ContainsKey("minMs") ? double.Parse(dict["minMs"], CultureInfo.InvariantCulture) : 0;
if (minMs >= SentinelValue) minMs = SentinelValue; // Reset SentinelValue
// TUA CORREZIONE: Reset sentinella per evitare valori fuori scala nel DB
if (minMs >= SentinelValue) minMs = SentinelValue;
if (maxMs >= SentinelValue) maxMs = SentinelValue;
if (count <= 0) continue;
detailRecordsToInsert.Add(new StatsDetailModel
{
@@ -158,21 +160,20 @@ namespace MP.IOC.Services
}
}
// --- FASE 2: UPSERT DB ---
// --- FASE UPSERT DB ---
if (detailRecordsToInsert.Count > 0)
{
await using var scope = _scopeFactory.CreateAsyncScope();
var detailService = scope.ServiceProvider.GetRequiredService<IStatsDetailService>();
await detailService.UpsertManyAsync(detailRecordsToInsert, false);
await detailService.UpsertManyAsync(detailRecordsToInsert, true);
Log.Info("[HISTORICAL] Upserted {count} records to DB", detailRecordsToInsert.Count);
}
// --- FASE 3: PULIZIA REDIS (Se confermata e se ci sono chiavi scadute) ---
// --- FASE PULIZIA REDIS ---
if (deleteConfirmed && keysToDelete.Count > 0)
{
int deletedCount = 0;
// Usiamo un batch per cancellare in modo efficiente
var batch = _db.CreateBatch();
int deletedCount = 0;
foreach (var key in keysToDelete)
{
_ = batch.KeyDeleteAsync(key);
@@ -183,23 +184,15 @@ namespace MP.IOC.Services
}
}
/// <summary>
/// Analizza la chiave per estrarre metadata e determinare se è un'ora o un giorno.
/// </summary>
private bool TryParseKeyMetadata(RedisKey key, out string dest, out string method, out DateTime timestamp, out bool isHourType)
{
dest = "NA"; method = "NA"; timestamp = DateTime.MinValue; isHourType = true;
try
{
string k = key.ToString();
// Rimuoviamo il prefisso base per analizzare la struttura: stats:{type}:{dest}...
string relativeKey = k.Replace($"{_redisBaseKey}:", "");
var parts = relativeKey.Split(':');
// Formato Hour: stats : hour : dest : method : yyyyMMddHH (parts count 5)
// Formato Day: stats : day : dest : yyyyMMdd (parts count 4)
if (parts.Length < 4) return false;
string type = parts[1]; // "hour" o "day"
@@ -209,7 +202,7 @@ namespace MP.IOC.Services
{
isHourType = true;
method = parts[3];
if (parts.Length >= 5 && DateTime.TryParseExact(parts[4], "yyyyMMddHH", DateTimeIntl, DateTimeStyles.None, out var dt))
if (parts.Length >= 5 && DateTime.TryParseExact(parts[4], "yyyyMMddHH", CultureInfo.InvariantCulture, DateTimeStyles.None, out var dt))
{
timestamp = dt;
return true;
@@ -219,15 +212,14 @@ namespace MP.IOC.Services
{
isHourType = false;
method = "DAILY";
if (DateTime.TryParseExact(parts[3], "yyyyMMdd", DateTimeIntl, DateTimeStyles.None, out var dt))
if (DateTime.TryParseExact(parts[3], "yyyyMMdd", CultureInfo.InvariantCulture, DateTimeStyles.None, out var dt))
{
timestamp = dt;
return true;
}
}
}
catch { /* Log error if needed */ }
catch { }
return false;
}
MP.IOC/appsettings.json
+2 -2
View File
@@ -27,7 +27,7 @@
"archiveEvery": "Day",
"archiveFileName": "${basedir}/logs/old/${shortdate}_{#}.log",
"archiveNumbering": "DateAndSequence",
"archiveAboveSize": "1024000",
"archiveAboveSize": "10240000",
"archiveDateFormat": "HH",
"maxArchiveFiles": "60",
"maxArchiveDays": "30"
@@ -71,7 +71,7 @@
},
"RouteMan": {
"MetricCalcIntervalSeconds": 10,
"MetricFlushIntervalSeconds": 60,
"MetricFlushIntervalSeconds": 600,
"DefaultWeightOld": 100,
"DefaultWeightNew": 0,
"DeleteExpiredMetrics": false