#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Prometheus 风格的指标导出器（双端口）。
- 端口 8000: 应用指标 (app_registry)
- 端口 8001: 系统指标 (sys_registry)
"""
import time
import random
import threading
import logging
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(message)s',
)
L = logging.getLogger(__name__)
class Metric:
"""指标基类。"""
def __init__(self, name, help_text, mtype, labelnames=None):
self.name = name
self.help = help_text
self.type = mtype
self.labelnames = labelnames or []
self._lock = threading.Lock()
def _fmt(self, labels):
if not labels:
return ''
parts = [f'{k}="{str(labels.get(k, ""))}"' for k in self.labelnames]
return '{' + ','.join(parts) + '}'
def hdr(self):
return [
f'# HELP {self.name} {self.help}',
f'# TYPE {self.name} {self.type}',
]
class Counter(Metric):
"""单调递增计数器。"""
def __init__(self, name, help_text, labelnames=None):
super().__init__(name, help_text, 'counter', labelnames)
self._values = {}
def _key(self, labels):
return tuple(labels.get(k, '') for k in self.labelnames)
def inc(self, amount=1, **labels):
with self._lock:
k = self._key(labels)
self._values[k] = self._values.get(k, 0) + amount
def expose(self):
lines = self.hdr()
with self._lock:
if not self._values and not self.labelnames:
lines.append(f'{self.name} 0')
for k, v in self._values.items():
lbl = dict(zip(self.labelnames, k))
lines.append(f'{self.name}{self._fmt(lbl)} {v}')
return lines
class Gauge(Metric):
"""可增可减的瞬时值。"""
def __init__(self, name, help_text, labelnames=None):
super().__init__(name, help_text, 'gauge', labelnames)
self._values = {}
def _key(self, labels):
return tuple(labels.get(k, '') for k in self.labelnames)
def set(self, value, **labels):
with self._lock:
self._values[self._key(labels)] = value
def expose(self):
lines = self.hdr()
with self._lock:
if not self._values and not self.labelnames:
lines.append(f'{self.name} 0')
for k, v in self._values.items():
lbl = dict(zip(self.labelnames, k))
lines.append(f'{self.name}{self._fmt(lbl)} {v}')
return lines
class Registry:
"""指标注册表，负责收集并输出所有指标的文本格式。"""
def __init__(self):
self._metrics = []
self._lock = threading.Lock()
def register(self, metric):
with self._lock:
self._metrics.append(metric)
return metric
def expose(self):
lines = []
with self._lock:
for m in self._metrics:
lines.extend(m.expose())
return '\n'.join(lines) + '\n'
# ---------- 应用指标 ----------
app_registry = Registry()
http_requests = app_registry.register(
Counter('myapp_http_requests_total', 'Total requests', ['method', 'status'])
)
active_conns = app_registry.register(
Gauge('myapp_active_connections', 'Active connections')
)
queue = app_registry.register(
Gauge('myapp_queue_size', 'Queue size')
)
# ---------- 系统指标 ----------
sys_registry = Registry()
cpu = sys_registry.register(
Gauge('system_cpu_usage_percent', 'CPU usage', ['core'])
)
mem = sys_registry.register(
Gauge('system_memory_usage_bytes', 'Memory usage', ['type'])
)
uptime_metric = sys_registry.register(
Gauge('system_uptime_seconds', 'Uptime')
)
class MetricsHandler(BaseHTTPRequestHandler):
registry = None
def do_GET(self):
path = urlparse(self.path).path
if path == '/metrics':
body = self.registry.expose().encode('utf-8')
self.send_response(200)
self.send_header('Content-Type', 'text/plain')
self.send_header('Content-Length', str(len(body)))
self.end_headers()
self.wfile.write(body)
else:
self.send_response(404)
self.end_headers()
def log_message(self, fmt, *args):
# 关闭默认访问日志
pass
def make_handler(registry):
"""为指定的 registry 生成一个 handler 类。"""
class _Handler(MetricsHandler):
pass
_Handler.registry = registry
return _Handler
def update_app():
"""模拟更新应用指标。"""
while True:
try:
http_requests.inc(
method=random.choice(['GET', 'POST']),
status=random.choice(['200', '500']),
)
active_conns.set(random.randint(10, 200))
queue.set(random.randint(0, 50))
time.sleep(1)
except Exception as e:
L.error(f'app:{e}')
time.sleep(1)
def update_sys():
"""模拟更新系统指标。"""
start = time.time()
while True:
try:
for i in range(4):
cpu.set(random.uniform(0, 100), core=f'cpu{i}')
total = 16 * 1024 ** 3
mem.set(total, type='total')
mem.set(total * random.uniform(0.3, 0.8), type='used')
uptime_metric.set(time.time() - start)
time.sleep(2)
except Exception as e:
L.error(f'sys:{e}')
time.sleep(1)
def start_server(port, registry):
"""启动一个 HTTP 服务，暴露指定 registry 的 /metrics。"""
httpd = ThreadingHTTPServer(('0.0.0.0', port), make_handler(registry))
threading.Thread(target=httpd.serve_forever, daemon=True).start()
L.info(f'Server started: port {port}')
def main():
L.info('Starting dual-port exporter')
start_server(8000, app_registry)
start_server(8001, sys_registry)
threading.Thread(target=update_app, daemon=True).start()
threading.Thread(target=update_sys, daemon=True).start()
while True:
time.sleep(1)
if __name__ == '__main__':
main()
