main - Replace log panel with 5-min health stats and anomaly alerts

Raw log tailing on the monitoring page duplicated HF's own log viewer and
wrote every log line to disk per worker for no real benefit. Removed that
entirely (_FileLogHandler, _trim_log_files, get_log_lines, /logs) and
replaced it with per-worker in-process instrumentation: a rolling 5-minute
window of per-image latency/success and cache evictions, aggregated into
throughput/error-rate/latency health stats and four anomaly alerts (worker
not reporting, RSS spike past the cache warm-up grace period, cache
thrashing, error-rate spike). All aggregation stays local per worker and
piggybacks on the existing 5s stats-reporter tick, so it adds negligible
overhead. Also dropped the dead `Model` import and unused `os` import in
main.py.

main.py

@@ -1,4 +1,3 @@
-import os
 from contextlib import asynccontextmanager
 from typing import List
 
@@ -7,7 +6,7 @@ from fastapi.responses import HTMLResponse
 
 from models import RequestModel, ResponseModel
 from similarity import (
-    check_similarity, get_all_memory_stats, get_machine_stats, get_log_lines,
+    check_similarity, get_dashboard_data, get_machine_stats,
     start_memory_stats_reporter, start_memory_trimmer,
 )
 
@@ -54,19 +53,19 @@ h1{font-family:'Barlow Condensed',sans-serif;font-weight:800;font-style:italic;
 .req{font-size:.75rem;color:#94a3b8;font-family:'SF Mono','Fira Code',monospace}
 .track{height:5px;background:#1e2942;border-radius:99px;margin-top:.45rem;overflow:hidden}
 .fill{height:100%;border-radius:99px;opacity:.7}
-.log-section{margin-top:.875rem;border-top:1px solid #1a1a2e;padding-top:.75rem;
+.alerts-section{margin-top:.875rem;border-top:1px solid #1a1a2e;padding-top:.75rem;
   display:flex;flex-direction:column;flex:1;min-height:0}
-.log-label{font-size:.65rem;color:#475569;font-family:'SF Mono','Fira Code',monospace;
+.alerts-label{font-size:.65rem;color:#475569;font-family:'SF Mono','Fira Code',monospace;
   text-transform:uppercase;letter-spacing:.08em;margin-bottom:.4rem;
   display:flex;align-items:center;gap:.5rem;flex-shrink:0}
-.log-dot{width:6px;height:6px;border-radius:50%;background:#4ade80;
-  box-shadow:0 0 6px #4ade80;animation:pulse 2s infinite}
+.alerts-dot{width:6px;height:6px;border-radius:50%;background:#4ade80;box-shadow:0 0 6px #4ade80}
+.alerts-dot.warn{background:#fab219;box-shadow:0 0 6px #fab219;animation:pulse 2s infinite}
 @keyframes pulse{0%,100%{opacity:1}50%{opacity:.4}}
-.log-box{background:#060b14;border:1px solid #1e2942;border-radius:6px;
-  flex:1;min-height:420px;overflow-y:auto;padding:.65rem .875rem;
-  font-family:ui-monospace,'SF Mono',Menlo,Consolas,'Fira Code',monospace;
-  font-size:.78rem;line-height:1.85;-webkit-font-smoothing:antialiased}
-.log-line{white-space:pre-wrap;word-break:break-all;color:#f4f4f2}
+.alerts-box{background:#111827;border:1px solid #1e2942;border-radius:6px;
+  flex:1;min-height:180px;overflow-y:auto;padding:.75rem .875rem;
+  font-family:'Inter',system-ui,sans-serif;font-size:.82rem;line-height:2}
+.alert-line{display:flex;align-items:baseline;gap:.55rem;color:#f4f4f2}
+.alert-empty{color:#475569;font-family:'SF Mono','Fira Code',monospace;font-size:.75rem}
 </style>
 </head><body>
 <header>
@@ -74,23 +73,41 @@ h1{font-family:'Barlow Condensed',sans-serif;font-weight:800;font-style:italic;
   <span id="meta" class="meta"></span>
 </header>
 <div id="grid" class="grid"></div>
-<div class="log-section">
-  <div class="log-label"><span class="log-dot"></span>Application logs</div>
-  <div id="logs" class="log-box"></div>
+<div class="alerts-section">
+  <div class="alerts-label"><span id="alerts-dot" class="alerts-dot"></span>Alertas</div>
+  <div id="alerts" class="alerts-box"></div>
 </div>
 <script>
 var MAX=__CACHE_MAX__;
 function col(pct){return pct<50?'#4ade80':pct<80?'#fbbf24':'#f87171'}
 function workerCol(mb){return mb<800?'#4ade80':mb<1000?'#fbbf24':'#f87171'}
+function esc(s){return s.replace(/&/g,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;')}
+function alertColor(s){return /sem reportar|taxa de erro/i.test(s)?'#d03b3b':'#fab219'}
+function renderAlerts(alerts){
+  var box=document.getElementById('alerts'),dot=document.getElementById('alerts-dot');
+  if(!alerts||!alerts.length){
+    box.innerHTML='<div class="alert-empty">sem alertas</div>';
+    dot.className='alerts-dot';
+    return;
+  }
+  dot.className='alerts-dot warn';
+  box.innerHTML=alerts.map(function(a){
+    return '<div class="alert-line"><span style="color:'+alertColor(a)+'">●</span><span>'+esc(a)+'</span></div>';
+  }).join('');
+}
 function render(d){
-  var w=d.workers,m=d.machine;
+  var w=d.workers,m=d.machine,h=d.health;
   var gb=(w.reduce(function(s,x){return s+x.rss_mb},0)/1024).toFixed(1);
   var rc=col(m.ram_percent),cc=col(m.cpu_percent);
+  var ec=h.error_rate_pct>5?'#d03b3b':'#94a3b8';
   document.getElementById('meta').innerHTML=
     '<b>'+w.length+'</b> workers &nbsp;&middot;&nbsp; '+
     '<b>'+gb+' GB</b> RSS &nbsp;&middot;&nbsp; '+
     'RAM <b style="color:'+rc+'">'+m.ram_used_gb+'/'+m.ram_total_gb+' GB</b> &nbsp;&middot;&nbsp; '+
     'CPU <b style="color:'+cc+'">'+m.cpu_percent+'%</b> &nbsp;&middot;&nbsp; '+
+    '<b>'+h.throughput_per_min+'</b> req/min &nbsp;&middot;&nbsp; '+
+    '<b style="color:'+ec+'">'+h.error_rate_pct+'%</b> erros &nbsp;&middot;&nbsp; '+
+    '<b>'+h.avg_latency_ms+'</b> ms &nbsp;&middot;&nbsp; '+
     new Date().toTimeString().slice(0,8);
   var maxRss=Math.max.apply(null,w.map(function(x){return x.rss_mb}).concat([1]));
   document.getElementById('grid').innerHTML=w.map(function(x){
@@ -102,59 +119,28 @@ function render(d){
       '<div class="track"><div class="fill" style="width:'+p+'%;background:'+c+'"></div></div>'+
       '</div>';
   }).join('');
+  renderAlerts(d.alerts);
 }
 function refresh(){
   var xhr=new XMLHttpRequest();
   xhr.open('GET','/stats');
   xhr.onload=function(){
     if(xhr.status===200){
-      try{var d=JSON.parse(xhr.responseText);if(d&&d.workers&&d.machine)render(d);}catch(e){}
+      try{var d=JSON.parse(xhr.responseText);if(d&&d.workers&&d.machine&&d.health)render(d);}catch(e){}
     }
   };
   xhr.send();
 }
 refresh();setInterval(refresh,5000);
-
-function esc(s){return s.replace(/&/g,'&amp;').replace(/</g,'&lt;').replace(/>/g,'&gt;')}
-function lineCol(s){
-  if(/error|exception|traceback|critical/i.test(s)) return '#d03b3b';
-  if(/warn/i.test(s)) return '#fab219';
-  return '#f4f4f2';
-}
-function refreshLogs(){
-  var xhr=new XMLHttpRequest();
-  xhr.open('GET','/logs?n=400');
-  xhr.onload=function(){
-    if(xhr.status===200){
-      try{
-        var d=JSON.parse(xhr.responseText);
-        if(d&&d.lines&&d.lines.length){
-          var el=document.getElementById('logs');
-          var atBottom=el.scrollHeight-el.scrollTop-el.clientHeight<60;
-          var lines=d.lines.slice().reverse();
-          el.innerHTML=lines.map(function(l){
-            return '<div class="log-line" style="color:'+lineCol(l)+'">'+esc(l)+'</div>';
-          }).join('');
-          if(atBottom) el.scrollTop=el.scrollHeight;
-        }
-      }catch(e){}
-    }
-  };
-  xhr.send();
-}
-refreshLogs();setInterval(refreshLogs,2000);
 </script>
 </body></html>""").replace("__CACHE_MAX__", str(_CACHE_MAX))
 
 
 @app.get("/stats", include_in_schema=False)
 async def stats_endpoint():
-    return {"workers": get_all_memory_stats(), "machine": get_machine_stats()}
-
-
-@app.get("/logs", include_in_schema=False)
-async def logs_endpoint(n: int = 100):
-    return {"lines": get_log_lines(n=min(n, 1000))}
+    data = get_dashboard_data()
+    data["machine"] = get_machine_stats()
+    return data
 
 
 @app.get("/", include_in_schema=False)

similarity.py

@@ -11,12 +11,11 @@ import psutil
 import json
 import time
 import ctypes
-from collections import OrderedDict
+from collections import OrderedDict, deque
 from concurrent.futures import ThreadPoolExecutor
 from typing import List
 from tensorflow.keras.applications import MobileNetV2
 from tensorflow.keras.applications.mobilenet_v2 import preprocess_input
-from tensorflow.keras.models import Model
 from tensorflow.keras.preprocessing.image import img_to_array
 from sklearn.metrics.pairwise import cosine_similarity
 from skimage.metrics import structural_similarity as ssim
@@ -30,58 +29,6 @@ register_heif_opener()
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 logging.getLogger("urllib3.connection").setLevel(logging.ERROR)
 
-_LOG_DIR = "/tmp/similarity_logs"
-os.makedirs(_LOG_DIR, exist_ok=True)
-
-
-class _FileLogHandler(logging.Handler):
-    def __init__(self):
-        super().__init__(logging.INFO)
-        self.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s', datefmt='%H:%M:%S'))
-        self._file = os.path.join(_LOG_DIR, f"{os.getpid()}.log")
-
-    def emit(self, record):
-        try:
-            with open(self._file, 'a') as f:
-                f.write(self.format(record) + "\n")
-        except Exception:
-            pass
-
-
-logging.getLogger().addHandler(_FileLogHandler())
-
-
-def _trim_log_files(max_lines=1000):
-    try:
-        for fname in os.listdir(_LOG_DIR):
-            fpath = os.path.join(_LOG_DIR, fname)
-            try:
-                with open(fpath, 'r') as f:
-                    lines = f.readlines()
-                if len(lines) > max_lines:
-                    with open(fpath, 'w') as f:
-                        f.writelines(lines[-max_lines:])
-            except (OSError, IOError):
-                pass
-    except OSError:
-        pass
-
-
-def get_log_lines(n=100):
-    all_lines = []
-    try:
-        for fname in os.listdir(_LOG_DIR):
-            fpath = os.path.join(_LOG_DIR, fname)
-            try:
-                with open(fpath, 'r') as f:
-                    all_lines.extend(f.readlines())
-            except (OSError, IOError):
-                continue
-    except OSError:
-        return []
-    all_lines.sort()
-    return [l.rstrip() for l in all_lines[-n:]][::-1]
-
 _mobilenet = None
 _mobilenet_lock = threading.Lock()
 
@@ -93,6 +40,69 @@ _request_count = 0
 
 _process = psutil.Process()
 
+# Rolling 5-minute window of per-image-comparison events, cache evictions, and
+# RSS samples, used to compute health stats and anomaly alerts. Bounded by both
+# maxlen (hard cap against a traffic spike) and periodic time-based pruning.
+_events_lock = threading.Lock()
+_request_events = deque(maxlen=5000)   # (timestamp, duration_ms, success)
+_eviction_events = deque(maxlen=2000)  # timestamp
+_rss_history = deque(maxlen=90)        # (timestamp, rss_mb), one per 5s reporter tick
+
+_WINDOW_SECONDS = 300
+_RSS_SPIKE_WINDOW_SECONDS = 120
+_RSS_SPIKE_MB = 150
+_RSS_SPIKE_MIN_UPTIME_SECONDS = 600
+_CACHE_THRASH_EVICTIONS_PER_MIN = 10
+_ERROR_RATE_ALERT_PCT = 5
+_STALE_AFTER_SECONDS = 15
+_STALE_FILE_DELETE_AFTER_SECONDS = 300
+
+
+def _record_eviction():
+    with _events_lock:
+        _eviction_events.append(time.time())
+
+
+def _record_request_event(duration_ms, success):
+    with _events_lock:
+        _request_events.append((time.time(), duration_ms, success))
+
+
+def _compute_window_metrics():
+    cutoff = time.time() - _WINDOW_SECONDS
+    with _events_lock:
+        while _request_events and _request_events[0][0] < cutoff:
+            _request_events.popleft()
+        while _eviction_events and _eviction_events[0] < cutoff:
+            _eviction_events.popleft()
+        total = len(_request_events)
+        failed = sum(1 for _, _, ok in _request_events if not ok)
+        latency_sum = sum(d for _, d, _ in _request_events)
+        evictions = len(_eviction_events)
+
+    minutes = _WINDOW_SECONDS / 60
+    return {
+        "throughput_per_min": round(total / minutes, 1),
+        "error_rate_pct": round((failed / total * 100) if total else 0.0, 1),
+        "avg_latency_ms": round((latency_sum / total) if total else 0.0, 1),
+        "evictions_per_min": round(evictions / minutes, 1),
+        "request_count_window": total,
+    }
+
+
+def _record_rss_sample(rss_mb):
+    now = time.time()
+    target_ts = now - _RSS_SPIKE_WINDOW_SECONDS
+    with _events_lock:
+        _rss_history.append((now, rss_mb))
+        rss_2min_ago = None
+        for ts, value in _rss_history:
+            if ts <= target_ts:
+                rss_2min_ago = value
+            else:
+                break
+    return round(rss_mb - rss_2min_ago, 1) if rss_2min_ago is not None else 0.0
+
 _cpu_cache = {"pct": 0.0, "usage": None, "ts": None}
 
 
@@ -200,15 +210,22 @@ def get_machine_stats():
 
 
 def get_memory_stats():
-    rss_mb = _process.memory_info().rss / (1024 * 1024)
+    rss_mb = round(_process.memory_info().rss / (1024 * 1024), 1)
     with _image_cache_lock:
         cache_size = len(_image_cache)
+
+    rss_delta_2min_mb = _record_rss_sample(rss_mb)
+    window = _compute_window_metrics()
+
     stats = {
         "pid": os.getpid(),
-        "rss_mb": round(rss_mb, 1),
+        "rss_mb": rss_mb,
         "cache_size": cache_size,
         "requests": _request_count,
+        "uptime_seconds": round(time.time() - _process.create_time(), 1),
+        "rss_delta_2min_mb": rss_delta_2min_mb,
         "updated_at": time.time(),
+        **window,
     }
     with open(os.path.join(_MEM_STATS_DIR, f"{stats['pid']}.json"), "w") as f:
         json.dump(stats, f)
@@ -236,21 +253,74 @@ def start_memory_trimmer(interval_seconds=300):
         while True:
             time.sleep(interval_seconds)
             _trim_memory()
-            _trim_log_files()
     threading.Thread(target=_trim_loop, daemon=True).start()
 
 
-def get_all_memory_stats():
-    get_memory_stats()
-    all_stats = []
+def _read_all_worker_files():
+    entries = []
     for filename in os.listdir(_MEM_STATS_DIR):
+        fpath = os.path.join(_MEM_STATS_DIR, filename)
         try:
-            with open(os.path.join(_MEM_STATS_DIR, filename)) as f:
-                all_stats.append(json.load(f))
+            with open(fpath) as f:
+                entries.append((fpath, json.load(f)))
         except (OSError, json.JSONDecodeError):
             continue
-    all_stats.sort(key=lambda s: s["pid"])
-    return all_stats
+    return entries
+
+
+def get_dashboard_data():
+    get_memory_stats()  # refresh this worker's own file before merging
+
+    now = time.time()
+    alive, alerts = [], []
+
+    for fpath, stats in _read_all_worker_files():
+        age = now - stats.get("updated_at", 0)
+
+        if age > _STALE_FILE_DELETE_AFTER_SECONDS:
+            try:
+                os.remove(fpath)
+            except OSError:
+                pass
+            continue
+
+        if age > _STALE_AFTER_SECONDS:
+            alerts.append(f"worker {stats['pid']} sem reportar há {int(age)}s")
+            continue
+
+        alive.append(stats)
+
+        if (stats.get("uptime_seconds", 0) > _RSS_SPIKE_MIN_UPTIME_SECONDS
+                and stats.get("rss_delta_2min_mb", 0) > _RSS_SPIKE_MB):
+            alerts.append(f"worker {stats['pid']} RAM subiu {stats['rss_delta_2min_mb']:.0f} MB em 2 min")
+
+        if stats.get("evictions_per_min", 0) > _CACHE_THRASH_EVICTIONS_PER_MIN:
+            alerts.append(f"worker {stats['pid']} cache a saturar (~{stats['evictions_per_min']:.0f} evicções/min)")
+
+    alive.sort(key=lambda s: s["pid"])
+
+    total_requests = sum(s.get("request_count_window", 0) for s in alive)
+    throughput_per_min = round(sum(s.get("throughput_per_min", 0) for s in alive), 1)
+    evictions_per_min = round(sum(s.get("evictions_per_min", 0) for s in alive), 1)
+    if total_requests:
+        error_rate_pct = round(
+            sum(s.get("error_rate_pct", 0) * s.get("request_count_window", 0) for s in alive) / total_requests, 1)
+        avg_latency_ms = round(
+            sum(s.get("avg_latency_ms", 0) * s.get("request_count_window", 0) for s in alive) / total_requests, 1)
+    else:
+        error_rate_pct = 0.0
+        avg_latency_ms = 0.0
+
+    if error_rate_pct > _ERROR_RATE_ALERT_PCT:
+        alerts.append(f"taxa de erro em {error_rate_pct:.0f}% nos últimos 5 min")
+
+    health = {
+        "throughput_per_min": throughput_per_min,
+        "error_rate_pct": error_rate_pct,
+        "avg_latency_ms": avg_latency_ms,
+        "evictions_per_min": evictions_per_min,
+    }
+    return {"workers": alive, "health": health, "alerts": alerts}
 
 def get_mobilenet():
     global _mobilenet
@@ -369,6 +439,7 @@ def _cache_put(asset_code, image):
     with _image_cache_lock:
         if len(_image_cache) >= _IMAGE_CACHE_MAX:
             _image_cache.popitem(last=False)
+            _record_eviction()
         _image_cache[asset_code] = image
 
 
@@ -512,6 +583,8 @@ def change_similarity_from_quadrants(main_image, main_source, source_image , sou
 
 def _process_image(args):
     i, original_image, original_image_shape, images, quadrant = args
+    started_at = time.time()
+    success = True
     try:
         image = load_image_url(images[i].source, images[i].assetCode, images[i].contentType)
         image = cv2.resize(image, original_image_shape[::-1])
@@ -532,6 +605,9 @@ def _process_image(args):
         similarity_score = 0
         similarity_orb_score = 0
         similarity_mobilenet_score = 0
+        success = False
+
+    _record_request_event((time.time() - started_at) * 1000, success)
 
     return ResponseModel(originId=images[i].originId, sequence=images[i].sequence,
                          assetCode=images[i].assetCode, similarity=similarity_score, similarityOrb=similarity_orb_score, similarityMobileNet=similarity_mobilenet_score)