diff --git a/MP.IOC/Services/MetricsCalcService.cs b/MP.IOC/Services/MetricsCalcService.cs index 80700269..c12f78bc 100644 --- a/MP.IOC/Services/MetricsCalcService.cs +++ b/MP.IOC/Services/MetricsCalcService.cs @@ -26,6 +26,173 @@ namespace MP.IOC.Services #region Protected Methods + /// + /// Valore sentinella x valore minimo in reids/Lua + /// Usiamo un valore molto grande (es. 10 milioni di secondi = ~115 giorni). + /// È impossibile che una singola richiesta HTTP duri così tanto, quindi usarlo come "sentinella" è sicuro. + /// + private const string SentinelValue = "999999999"; + + /// + /// Script update Redis in Lua, definito come costante per efficienza. + /// Lo script gestisce l'incremento e la logica condizionale per Min/Max in un'unica operazione atomica. + /// + private const string RedisUpdateScript = @" + local key = KEYS[1] + local countInc = tonumber(ARGV[1]) + local totalMsInc = tonumber(ARGV[2]) + local newMax = tonumber(ARGV[3]) + local newMin = tonumber(ARGV[4]) + local sentinel = tonumber(ARGV[5]) + + -- 1. Incremento Count e TotalDuration + redis.call('HINCRBY', key, 'count', countInc) + redis.call('HINCRBYFLOAT', key, 'totalMs', totalMsInc) + + -- 2. Aggiornamento Max (solo se il nuovo valore è maggiore del precedente E non è la sentinella) + local currentMax = redis.call('HGET', key, 'maxMs') + if newMax < sentinel and (not currentMax or newMax > tonumber(currentMax)) then + redis.call('HSET', key, 'maxMs', tostring(newMax)) + end + + -- 3. Aggiornamento Min (solo se il nuovo valore è minore del precedente E non è la sentinella) + local currentMin = redis.call('HGET', key, 'minMs') + if newMin < sentinel and (not currentMin or newMin < tonumber(currentMin)) then + redis.call('HSET', key, 'minMs', tostring(newMin)) + end + return 1 + "; + + // Classe di supporto per l'aggregazione locale dei valori Daily + private class AggregatedStats + { + public long Count; + public double TotalMs; + public double MaxMs; + public double MinMs = double.MaxValue; + } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var interval = _config.GetValue("RouteMan:MetricCalcIntervalSeconds", 20); + + while (!stoppingToken.IsCancellationRequested) + { + try + { + await Task.Delay(TimeSpan.FromSeconds(interval), stoppingToken); + var snapshot = _stats.Snapshot(); + if (snapshot.Count == 0) continue; + + var adesso = DateTime.Now; + var hourStart = new DateTime(adesso.Year, adesso.Month, adesso.Day, adesso.Hour, 0, 0); + var dayStart = new DateTime(adesso.Year, adesso.Month, adesso.Day, 0, 0, 0); + + if (_db == null) continue; + + var batch = _db.CreateBatch(); + var tasks = new List(); + + // Dizionario per aggregare i valori Daily prima di inviarli a Redis + var dailyAggregates = new Dictionary(); + + foreach (var kv in snapshot) + { + string dest = "IO"; + string method = "NA"; + string rawKey = kv.Key; + if (rawKey.Contains("|")) + { + var splitVal = rawKey.Split("|"); + dest = splitVal[0]; + method = splitVal[1]; + } + else { method = rawKey; } + + var stat = kv.Value; + var count = Interlocked.Read(ref stat.Count); + var totalMs = stat.TotalDuration.TotalMilliseconds; + var maxMs = stat.MaxDuration.TotalMilliseconds; + // Se la durata è ancora MaxValue, usiamo la sentinella per dire a Redis "non aggiornare" + var minMs = (stat.MinDuration == TimeSpan.MaxValue) + ? double.Parse(SentinelValue) + : stat.MinDuration.TotalMilliseconds; + + // --- LOGICA HOURLY (Per ogni route/method è una chiave distinta) --- + var hourKey = HourBucketKey(dest, method, hourStart); + var hoursIndex = HoursIndexKey(dest, method); + var hourScore = ToEpochSeconds(hourStart); + + // Usiamo lo script Lua per l'aggiornamento atomico dell'ora + tasks.Add(batch.ScriptEvaluateAsync(RedisUpdateScript, + new RedisKey[] { hourKey }, + new RedisValue[] { + count.ToString(), + totalMs.ToString(), + maxMs.ToString(), + minMs.ToString(), + SentinelValue + })); + + tasks.Add(batch.SortedSetAddAsync(hoursIndex, hourKey, hourScore)); + + // --- LOGICA DAILY (Aggregazione locale per evitare sovrascritture nel loop) --- + var dayKey = DayBucketKey(dest, dayStart); + var daysIndex = DaysIndexKey(dest); + var dayScore = ToEpochSeconds(dayStart); + + if (!dailyAggregates.TryGetValue(dayKey, out var agg)) + { + agg = new AggregatedStats(); + dailyAggregates[dayKey] = agg; + } + + agg.Count += count; + agg.TotalMs += totalMs; + agg.MaxMs = Math.Max(agg.MaxMs, maxMs); + if (minMs != double.MaxValue) + agg.MinMs = Math.Min(agg.MinMs, minMs); + + // Aggiungiamo l'indice al batch + tasks.Add(batch.SortedSetAddAsync(daysIndex, dayKey, dayScore)); + } + + // --- INVIO AGGREGATI DAILY A REDIS --- + foreach (var dayKV in dailyAggregates) + { + var key = dayKV.Key; + var agg = dayKV.Value; + // Se agg.MinMs è inizializzato a double.MaxValue, lo portiamo alla sentinella + double finalMin = (agg.MinMs == double.MaxValue || agg.MinMs >= double.Parse(SentinelValue)) + ? double.Parse(SentinelValue) + : agg.MinMs; + + tasks.Add(batch.ScriptEvaluateAsync(RedisUpdateScript, + new RedisKey[] { key }, + new RedisValue[] { + agg.Count.ToString(), + agg.TotalMs.ToString(), + agg.MaxMs.ToString(), + finalMin.ToString(), + SentinelValue + })); + } + + // Esegui tutto il batch in un unico round-trip + batch.Execute(); + await Task.WhenAll(tasks); + + _stats.Clear(); + } + catch (TaskCanceledException) { } + catch (Exception ex) + { + Log.Error(ex, "Error flushing metrics"); + } + } + } + +#if false protected override async Task ExecuteAsync(CancellationToken stoppingToken) { var interval = _config.GetValue("RouteMan:MetricCalcIntervalSeconds", 20); @@ -116,6 +283,7 @@ namespace MP.IOC.Services } } } +#endif #endregion Protected Methods