# Copyright 2015-2019 Prometheus Python Client Developers # Copyright 2019 Matrix.org Foundation C.I.C. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """ This code is based off `prometheus_client/exposition.py` from version 0.7.1. Due to the renaming of metrics in prometheus_client 0.4.0, this customised vendoring of the code will emit both the old versions that Synapse dashboards expect, and the newer "best practice" version of the up-to-date official client. """ import logging import math import threading from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn from typing import Any, Dict, List, Type, Union from urllib.parse import parse_qs, urlparse from prometheus_client import REGISTRY, CollectorRegistry from prometheus_client.core import Sample from twisted.web.resource import Resource from twisted.web.server import Request logger = logging.getLogger(__name__) CONTENT_TYPE_LATEST = "text/plain; version=0.0.4; charset=utf-8" def floatToGoString(d: Union[int, float]) -> str: d = float(d) if d == math.inf: return "+Inf" elif d == -math.inf: return "-Inf" elif math.isnan(d): return "NaN" else: s = repr(d) dot = s.find(".") # Go switches to exponents sooner than Python. # We only need to care about positive values for le/quantile. if d > 0 and dot > 6: mantissa = f"{s[0]}.{s[1:dot]}{s[dot + 1 :]}".rstrip("0.") return f"{mantissa}e+0{dot - 1}" return s def sample_line(line: Sample, name: str) -> str: if line.labels: labelstr = "{{{0}}}".format( ",".join( [ '{}="{}"'.format( k, v.replace("\\", r"\\").replace("\n", r"\n").replace('"', r"\""), ) for k, v in sorted(line.labels.items()) ] ) ) else: labelstr = "" timestamp = "" if line.timestamp is not None: # Convert to milliseconds. timestamp = f" {int(float(line.timestamp) * 1000):d}" return "{}{} {}{}\n".format(name, labelstr, floatToGoString(line.value), timestamp) # Mapping from new metric names to legacy metric names. # We translate these back to their old names when exposing them through our # legacy vendored exporter. # Only this legacy exposition module applies these name changes. LEGACY_METRIC_NAMES = { "synapse_util_caches_cache_hits": "synapse_util_caches_cache:hits", "synapse_util_caches_cache_size": "synapse_util_caches_cache:size", "synapse_util_caches_cache_evicted_size": "synapse_util_caches_cache:evicted_size", "synapse_util_caches_cache": "synapse_util_caches_cache:total", "synapse_util_caches_response_cache_size": "synapse_util_caches_response_cache:size", "synapse_util_caches_response_cache_hits": "synapse_util_caches_response_cache:hits", "synapse_util_caches_response_cache_evicted_size": "synapse_util_caches_response_cache:evicted_size", "synapse_util_caches_response_cache": "synapse_util_caches_response_cache:total", "synapse_federation_client_sent_pdu_destinations": "synapse_federation_client_sent_pdu_destinations:total", "synapse_federation_client_sent_pdu_destinations_count": "synapse_federation_client_sent_pdu_destinations:count", "synapse_admin_mau_current": "synapse_admin_mau:current", "synapse_admin_mau_max": "synapse_admin_mau:max", "synapse_admin_mau_registered_reserved_users": "synapse_admin_mau:registered_reserved_users", } def generate_latest(registry: CollectorRegistry, emit_help: bool = False) -> bytes: """ Generate metrics in legacy format. Modern metrics are generated directly by prometheus-client. """ output = [] for metric in registry.collect(): if not metric.samples: # No samples, don't bother. continue # Translate to legacy metric name if it has one. mname = LEGACY_METRIC_NAMES.get(metric.name, metric.name) mnewname = metric.name mtype = metric.type # OpenMetrics -> Prometheus if mtype == "counter": mnewname = mnewname + "_total" elif mtype == "info": mtype = "gauge" mnewname = mnewname + "_info" elif mtype == "stateset": mtype = "gauge" elif mtype == "gaugehistogram": mtype = "histogram" elif mtype == "unknown": mtype = "untyped" # Output in the old format for compatibility. if emit_help: output.append( "# HELP {} {}\n".format( mname, metric.documentation.replace("\\", r"\\").replace("\n", r"\n"), ) ) output.append(f"# TYPE {mname} {mtype}\n") om_samples: Dict[str, List[str]] = {} for s in metric.samples: for suffix in ["_created", "_gsum", "_gcount"]: if s.name == mname + suffix: # OpenMetrics specific sample, put in a gauge at the end. # (these come from gaugehistograms which don't get renamed, # so no need to faff with mnewname) om_samples.setdefault(suffix, []).append(sample_line(s, s.name)) break else: newname = s.name.replace(mnewname, mname) if ":" in newname and newname.endswith("_total"): newname = newname[: -len("_total")] output.append(sample_line(s, newname)) for suffix, lines in sorted(om_samples.items()): if emit_help: output.append( "# HELP {}{} {}\n".format( mname, suffix, metric.documentation.replace("\\", r"\\").replace("\n", r"\n"), ) ) output.append(f"# TYPE {mname}{suffix} gauge\n") output.extend(lines) # Get rid of the weird colon things while we're at it if mtype == "counter": mnewname = mnewname.replace(":total", "") mnewname = mnewname.replace(":", "_") if mname == mnewname: continue # Also output in the new format, if it's different. if emit_help: output.append( "# HELP {} {}\n".format( mnewname, metric.documentation.replace("\\", r"\\").replace("\n", r"\n"), ) ) output.append(f"# TYPE {mnewname} {mtype}\n") for s in metric.samples: # Get rid of the OpenMetrics specific samples (we should already have # dealt with them above anyway.) for suffix in ["_created", "_gsum", "_gcount"]: if s.name == mname + suffix: break else: sample_name = LEGACY_METRIC_NAMES.get(s.name, s.name) output.append( sample_line(s, sample_name.replace(":total", "").replace(":", "_")) ) return "".join(output).encode("utf-8") class MetricsHandler(BaseHTTPRequestHandler): """HTTP handler that gives metrics from ``REGISTRY``.""" registry = REGISTRY def do_GET(self) -> None: registry = self.registry params = parse_qs(urlparse(self.path).query) if "help" in params: emit_help = True else: emit_help = False try: output = generate_latest(registry, emit_help=emit_help) except Exception: self.send_error(500, "error generating metric output") raise try: self.send_response(200) self.send_header("Content-Type", CONTENT_TYPE_LATEST) self.send_header("Content-Length", str(len(output))) self.end_headers() self.wfile.write(output) except BrokenPipeError as e: logger.warning( "BrokenPipeError when serving metrics (%s). Did Prometheus restart?", e ) def log_message(self, format: str, *args: Any) -> None: """Log nothing.""" @classmethod def factory(cls, registry: CollectorRegistry) -> Type: """Returns a dynamic MetricsHandler class tied to the passed registry. """ # This implementation relies on MetricsHandler.registry # (defined above and defaulted to REGISTRY). # As we have unicode_literals, we need to create a str() # object for type(). cls_name = str(cls.__name__) MyMetricsHandler = type(cls_name, (cls, object), {"registry": registry}) return MyMetricsHandler class _ThreadingSimpleServer(ThreadingMixIn, HTTPServer): """Thread per request HTTP server.""" # Make worker threads "fire and forget". Beginning with Python 3.7 this # prevents a memory leak because ``ThreadingMixIn`` starts to gather all # non-daemon threads in a list in order to join on them at server close. # Enabling daemon threads virtually makes ``_ThreadingSimpleServer`` the # same as Python 3.7's ``ThreadingHTTPServer``. daemon_threads = True def start_http_server( port: int, addr: str = "", registry: CollectorRegistry = REGISTRY ) -> None: """Starts an HTTP server for prometheus metrics as a daemon thread""" CustomMetricsHandler = MetricsHandler.factory(registry) httpd = _ThreadingSimpleServer((addr, port), CustomMetricsHandler) t = threading.Thread(target=httpd.serve_forever) t.daemon = True t.start() class MetricsResource(Resource): """ Twisted ``Resource`` that serves prometheus metrics. """ isLeaf = True def __init__(self, registry: CollectorRegistry = REGISTRY): self.registry = registry def render_GET(self, request: Request) -> bytes: request.setHeader(b"Content-Type", CONTENT_TYPE_LATEST.encode("ascii")) response = generate_latest(self.registry) request.setHeader(b"Content-Length", str(len(response))) return response