consultoria-as
a386ca5b4a
- F0 blindaje: Redis/Celery, pg_hba sin trust LAN, secretos a env 600, backups, webhook auth, jobs muertos - F1 caja: bugs POS, reembolsos al corte, apartados, CFDI, topes de descuento en cotizacion/apartado - F2/F7 IA de catalogo: autocategorizacion, dedup de SKUs con embeddings, OCR de facturas, descripciones, claves SAT, limpieza de fitment - F3 IA en el flujo: vision Yonke, preguntas MeLi, venta cruzada, anomalias, sinonimos, busqueda semantica (rerank/embeddings) - F4/F8 escala y cartera: /metrics+Prometheus+Grafana, pool, score de credito, churn, metricas multi-tenant - Nexus Yonke (M1-M6): modelo de datos 1-de-1, alta por donador, vitrina publica, MeLi usado, consignacion, garantia por dias, reportes, etiquetas - Marketplace B2B entre yonkes (catalogo + pedidos cross-tenant) - Seguridad: auth en /api/admin/* (35 endpoints), alertas por tenant - Migraciones v4.27/v4.28; timers expire-demos y rollup-metrics
256 lines
10 KiB
Python
256 lines
10 KiB
Python
"""Expositor Prometheus mínimo, sin dependencias externas.
|
|
|
|
El Python de producción no trae ``prometheus_client`` (y no se instala), así que
|
|
el formato de texto de Prometheus 0.0.4 se arma a mano.
|
|
|
|
El registro vive en memoria del proceso y es thread-safe (``threading.Lock``);
|
|
con gunicorn multiproceso cada worker lleva sus propios contadores y Prometheus
|
|
scrapea al que responda. Para tasas y latencias eso es suficiente; si algún día
|
|
se necesita un contador global exacto hay que habilitar el modo multiproceso con
|
|
un directorio compartido.
|
|
|
|
Uso::
|
|
|
|
from services.prom_metrics import metrics
|
|
|
|
metrics.configure('pos')
|
|
metrics.request_started() # before_request
|
|
metrics.request_finished(method, path, status, duration) # after_request
|
|
metrics.record_exception() # teardown_request (exc)
|
|
metrics.request_aborted() # teardown si no hubo after_request
|
|
body = metrics.render() # GET /metrics
|
|
"""
|
|
|
|
import os
|
|
import threading
|
|
import time
|
|
|
|
CONTENT_TYPE = 'text/plain; version=0.0.4; charset=utf-8'
|
|
|
|
DEFAULT_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0)
|
|
|
|
# Tope de series distintas por métrica; lo que se salga de aquí se agrupa en
|
|
# path="/otras" para que una ruta inesperada no infle la memoria.
|
|
MAX_SERIES = 400
|
|
|
|
# La consulta a pg_stat_activity es best-effort y se cachea unos segundos.
|
|
DB_GAUGE_TTL = 5.0
|
|
|
|
|
|
def escape_label(value):
|
|
return str(value).replace('\\', '\\\\').replace('"', '\\"').replace('\n', '\\n')
|
|
|
|
|
|
def format_labels(pairs):
|
|
if not pairs:
|
|
return ''
|
|
body = ','.join(f'{k}="{escape_label(v)}"' for k, v in pairs)
|
|
return '{' + body + '}'
|
|
|
|
|
|
def normalize_path(rule, path):
|
|
"""Colapsa una petición a un valor de label acotado.
|
|
|
|
Si Flask resolvió una regla (``request.url_rule.rule``) se usa tal cual, que
|
|
ya viene en forma plantilla (``/pos/api/inventory/items/<int:item_id>``).
|
|
Para 404 (sin regla) sólo se conserva el primer segmento.
|
|
"""
|
|
if rule:
|
|
return rule
|
|
segments = [s for s in (path or '/').split('/') if s]
|
|
if not segments:
|
|
return '/'
|
|
if len(segments) == 1:
|
|
return '/' + segments[0]
|
|
return '/' + segments[0] + '/...'
|
|
|
|
|
|
def _master_dsn():
|
|
dsn = os.environ.get('MASTER_DB_URL') or os.environ.get('DATABASE_URL')
|
|
if dsn:
|
|
return dsn
|
|
try:
|
|
from config import MASTER_DB_URL
|
|
return MASTER_DB_URL
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
class Metrics:
|
|
def __init__(self, app='app', version=None, buckets=DEFAULT_BUCKETS):
|
|
self._lock = threading.Lock()
|
|
self._app = app
|
|
self._version = version or os.environ.get('NEXUS_APP_VERSION', 'dev')
|
|
self._buckets = tuple(sorted(buckets))
|
|
self._start = time.time()
|
|
self._requests = {} # (method, path, status_class) -> count
|
|
self._errors = {} # (method, path, status_class) -> count (4xx/5xx)
|
|
self._durations = {} # (method, path) -> [sum, count, [bucket counts]]
|
|
self._exceptions = 0
|
|
self._in_flight = 0
|
|
self._db_cache = (0.0, None)
|
|
|
|
# ─── Instrumentación ────────────────────────────────────────────────
|
|
def configure(self, app=None, version=None):
|
|
with self._lock:
|
|
if app:
|
|
self._app = app
|
|
if version:
|
|
self._version = version
|
|
|
|
def request_started(self):
|
|
with self._lock:
|
|
self._in_flight += 1
|
|
|
|
def request_finished(self, method, path, status, duration):
|
|
method = method or 'GET'
|
|
path = path or '/'
|
|
try:
|
|
status_class = f'{int(status) // 100}xx'
|
|
except (TypeError, ValueError):
|
|
status_class = 'other'
|
|
duration = max(0.0, float(duration or 0.0))
|
|
|
|
with self._lock:
|
|
self._in_flight = max(0, self._in_flight - 1)
|
|
|
|
key = (method, path, status_class)
|
|
if key not in self._requests and len(self._requests) >= MAX_SERIES:
|
|
key = (method, '/otras', status_class)
|
|
self._requests[key] = self._requests.get(key, 0) + 1
|
|
|
|
if status_class in ('4xx', '5xx'):
|
|
ekey = (method, path, status_class)
|
|
if ekey not in self._errors and len(self._errors) >= MAX_SERIES:
|
|
ekey = (method, '/otras', status_class)
|
|
self._errors[ekey] = self._errors.get(ekey, 0) + 1
|
|
|
|
dkey = (method, path)
|
|
if dkey not in self._durations and len(self._durations) >= MAX_SERIES:
|
|
dkey = (method, '/otras')
|
|
entry = self._durations.get(dkey)
|
|
if entry is None:
|
|
entry = [0.0, 0, [0] * len(self._buckets)]
|
|
self._durations[dkey] = entry
|
|
entry[0] += duration
|
|
entry[1] += 1
|
|
for i, upper in enumerate(self._buckets):
|
|
if duration <= upper:
|
|
entry[2][i] += 1
|
|
|
|
def request_aborted(self):
|
|
"""Petición que no pasó por after_request (excepción que no se convirtió en respuesta)."""
|
|
with self._lock:
|
|
self._in_flight = max(0, self._in_flight - 1)
|
|
|
|
def record_exception(self):
|
|
with self._lock:
|
|
self._exceptions += 1
|
|
|
|
# ─── Gauge de conexiones (consulta en el momento del scrape) ────────
|
|
def _db_connections(self):
|
|
now = time.time()
|
|
cached_at, cached_value = self._db_cache
|
|
if now - cached_at < DB_GAUGE_TTL:
|
|
return cached_value
|
|
|
|
value = None
|
|
dsn = _master_dsn()
|
|
if dsn:
|
|
try:
|
|
import psycopg2
|
|
conn = psycopg2.connect(dsn, connect_timeout=2)
|
|
try:
|
|
conn.autocommit = True
|
|
cur = conn.cursor()
|
|
cur.execute('SET statement_timeout = 2000')
|
|
cur.execute('SELECT count(*) FROM pg_stat_activity')
|
|
row = cur.fetchone()
|
|
cur.close()
|
|
if row:
|
|
value = int(row[0])
|
|
finally:
|
|
conn.close()
|
|
except Exception:
|
|
value = None
|
|
|
|
with self._lock:
|
|
self._db_cache = (now, value)
|
|
return value
|
|
|
|
# ─── Render ─────────────────────────────────────────────────────────
|
|
def render(self):
|
|
with self._lock:
|
|
app = self._app
|
|
version = self._version
|
|
start = self._start
|
|
requests = dict(self._requests)
|
|
errors = dict(self._errors)
|
|
durations = {k: (v[0], v[1], list(v[2])) for k, v in self._durations.items()}
|
|
exceptions = self._exceptions
|
|
in_flight = self._in_flight
|
|
|
|
db_connections = self._db_connections()
|
|
now = time.time()
|
|
out = []
|
|
|
|
def head(name, kind, help_text):
|
|
out.append(f'# HELP {name} {help_text}')
|
|
out.append(f'# TYPE {name} {kind}')
|
|
|
|
head('nexus_app_info', 'gauge', 'Application build information.')
|
|
out.append(f'nexus_app_info{format_labels([("app", app), ("version", version)])} 1')
|
|
|
|
head('nexus_uptime_seconds', 'gauge', 'Seconds since the process started.')
|
|
out.append(f'nexus_uptime_seconds{format_labels([("app", app)])} {now - start:.3f}')
|
|
|
|
head('nexus_process_start_time_seconds', 'gauge', 'Unix time the process started.')
|
|
out.append(f'nexus_process_start_time_seconds{format_labels([("app", app)])} {start:.3f}')
|
|
|
|
head('nexus_http_requests_in_flight', 'gauge', 'Requests currently being processed.')
|
|
out.append(f'nexus_http_requests_in_flight{format_labels([("app", app)])} {in_flight}')
|
|
|
|
head('nexus_http_requests_total', 'counter',
|
|
'HTTP requests by method, normalized path and status class.')
|
|
for method, path, status_class in sorted(requests):
|
|
labels = format_labels([('app', app), ('method', method),
|
|
('path', path), ('status', status_class)])
|
|
out.append(f'nexus_http_requests_total{labels} {requests[(method, path, status_class)]}')
|
|
|
|
head('nexus_http_errors_total', 'counter',
|
|
'HTTP responses with a 4xx or 5xx status class.')
|
|
for method, path, status_class in sorted(errors):
|
|
labels = format_labels([('app', app), ('method', method),
|
|
('path', path), ('status', status_class)])
|
|
out.append(f'nexus_http_errors_total{labels} {errors[(method, path, status_class)]}')
|
|
|
|
if durations:
|
|
head('nexus_http_request_duration_seconds', 'histogram',
|
|
'HTTP request latency by method and normalized path.')
|
|
for method, path in sorted(durations):
|
|
total, count, buckets = durations[(method, path)]
|
|
base = [('app', app), ('method', method), ('path', path)]
|
|
for i, upper in enumerate(self._buckets):
|
|
labels = format_labels(base + [('le', str(upper))])
|
|
out.append(f'nexus_http_request_duration_seconds_bucket{labels} {buckets[i]}')
|
|
inf = format_labels(base + [('le', '+Inf')])
|
|
out.append(f'nexus_http_request_duration_seconds_bucket{inf} {count}')
|
|
labels = format_labels(base)
|
|
out.append(f'nexus_http_request_duration_seconds_sum{labels} {total:.6f}')
|
|
out.append(f'nexus_http_request_duration_seconds_count{labels} {count}')
|
|
|
|
head('nexus_app_exceptions_total', 'counter',
|
|
'Unhandled exceptions raised while serving a request.')
|
|
out.append(f'nexus_app_exceptions_total{format_labels([("app", app)])} {exceptions}')
|
|
|
|
if db_connections is not None:
|
|
head('nexus_db_connections', 'gauge',
|
|
'Open connections on the master database (pg_stat_activity).')
|
|
labels = format_labels([('app', app), ('database', 'master')])
|
|
out.append(f'nexus_db_connections{labels} {db_connections}')
|
|
|
|
return '\n'.join(out) + '\n'
|
|
|
|
|
|
metrics = Metrics()
|