Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-27 02:38
📄 __init__.py #!/usr/bin/python from . import exposition from . import gc_collector from . import metrics from . import metrics_core from . import platform_collector from . import process_collector from . import registry __all__ = ['Counter', 'Gauge', 'Summary', 'Histogram', 'Info', 'Enum'] CollectorRegistry = registry.CollectorRegistry REGISTRY = registry.REGISTRY Metric = metrics_core.Metric Counter = metrics.Counter Gauge = metrics.Gauge Summary = metrics.Summary Histogram = metrics.Histogram Info = metrics.Info Enum = metrics.Enum CONTENT_TYPE_LATEST = exposition.CONTENT_TYPE_LATEST generate_latest = exposition.generate_latest MetricsHandler = exposition.MetricsHandler make_wsgi_app = exposition.make_wsgi_app try: # Python >3.5 only make_asgi_app = exposition.make_asgi_app except: pass start_http_server = exposition.start_http_server start_wsgi_server = exposition.start_wsgi_server write_to_textfile = exposition.write_to_textfile push_to_gateway = exposition.push_to_gateway pushadd_to_gateway = exposition.pushadd_to_gateway delete_from_gateway = exposition.delete_from_gateway instance_ip_grouping_key = exposition.instance_ip_grouping_key ProcessCollector = process_collector.ProcessCollector PROCESS_COLLECTOR = process_collector.PROCESS_COLLECTOR PlatformCollector = platform_collector.PlatformCollector PLATFORM_COLLECTOR = platform_collector.PLATFORM_COLLECTOR GCCollector = gc_collector.GCCollector GC_COLLECTOR = gc_collector.GC_COLLECTOR if __name__ == '__main__': c = Counter('cc', 'A counter') c.inc() g = Gauge('gg', 'A gauge') g.set(17) s = Summary('ss', 'A summary', ['a', 'b']) s.labels('c', 'd').observe(17) h = Histogram('hh', 'A histogram') h.observe(.6) start_http_server(8000) import time while True: time.sleep(1)	1.77 KB	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 asgi.py from urllib.parse import parse_qs from .exposition import _bake_output from .registry import REGISTRY def make_asgi_app(registry=REGISTRY): """Create a ASGI app which serves the metrics from a registry.""" async def prometheus_app(scope, receive, send): assert scope.get("type") == "http" # Prepare parameters params = parse_qs(scope.get('query_string', b'')) accept_header = "Accept: " + ",".join([ value.decode("utf8") for (name, value) in scope.get('headers') if name.decode("utf8") == 'accept' ]) # Bake output status, header, output = _bake_output(registry, accept_header, params) # Return output payload = await receive() if payload.get("type") == "http.request": await send( { "type": "http.response.start", "status": int(status.split(' ')[0]), "headers": [ tuple(x.encode('utf8') for x in header) ] } ) await send({"type": "http.response.body", "body": output}) return prometheus_app	1.16 KB	2026-01-20 12:58
📁 bridge	-	2026-03-21 15:45
📄 context_managers.py from __future__ import unicode_literals from timeit import default_timer from .decorator import decorate class ExceptionCounter(object): def __init__(self, counter, exception): self._counter = counter self._exception = exception def __enter__(self): pass def __exit__(self, typ, value, traceback): if isinstance(value, self._exception): self._counter.inc() def __call__(self, f): def wrapped(func, args, kwargs): with self: return func(args, *kwargs) return decorate(f, wrapped) class InprogressTracker(object): def __init__(self, gauge): self._gauge = gauge def __enter__(self): self._gauge.inc() def __exit__(self, typ, value, traceback): self._gauge.dec() def __call__(self, f): def wrapped(func, args, *kwargs): with self: return func(args, *kwargs) return decorate(f, wrapped) class Timer(object): def __init__(self, callback): self._callback = callback def _new_timer(self): return self.__class__(self._callback) def __enter__(self): self._start = default_timer() def __exit__(self, typ, value, traceback): # Time can go backwards. duration = max(default_timer() - self._start, 0) self._callback(duration) def __call__(self, f): def wrapped(func, args, *kwargs): # Obtaining new instance of timer every time # ensures thread safety and reentrancy. with self._new_timer(): return func(args, **kwargs) return decorate(f, wrapped)	1.65 KB	2026-01-20 12:58
📄 core.py from __future__ import unicode_literals from .metrics import Counter, Enum, Gauge, Histogram, Info, Summary from .metrics_core import ( CounterMetricFamily, GaugeHistogramMetricFamily, GaugeMetricFamily, HistogramMetricFamily, InfoMetricFamily, Metric, StateSetMetricFamily, SummaryMetricFamily, UnknownMetricFamily, UntypedMetricFamily) from .registry import CollectorRegistry, REGISTRY from .samples import Exemplar, Sample, Timestamp __all__ = ( 'CollectorRegistry', 'Counter', 'CounterMetricFamily', 'Enum', 'Exemplar', 'Gauge', 'GaugeHistogramMetricFamily', 'GaugeMetricFamily', 'Histogram', 'HistogramMetricFamily', 'Info', 'InfoMetricFamily', 'Metric', 'REGISTRY', 'Sample', 'StateSetMetricFamily', 'Summary', 'SummaryMetricFamily', 'Timestamp', 'UnknownMetricFamily', 'UntypedMetricFamily', )	895 B	2026-01-20 12:58
📄 decorator.py # ######################### LICENSE ############################ # # Copyright (c) 2005-2016, Michele Simionato # All rights reserved. # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are # met: # Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # Redistributions in bytecode form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in # the documentation and/or other materials provided with the # distribution. # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT # HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS # OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR # TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE # USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH # DAMAGE. """ Decorator module, see http://pypi.python.org/pypi/decorator for the documentation. """ from __future__ import print_function import collections import inspect import itertools import operator import re import sys __version__ = '4.0.10' if sys.version_info >= (3,): from inspect import getfullargspec def get_init(cls): return cls.__init__ else: class getfullargspec(object): "A quick and dirty replacement for getfullargspec for Python 2.X" def __init__(self, f): self.args, self.varargs, self.varkw, self.defaults = \ inspect.getargspec(f) self.kwonlyargs = [] self.kwonlydefaults = None def __iter__(self): yield self.args yield self.varargs yield self.varkw yield self.defaults getargspec = inspect.getargspec def get_init(cls): return cls.__init__.__func__ # getargspec has been deprecated in Python 3.5 ArgSpec = collections.namedtuple( 'ArgSpec', 'args varargs varkw defaults') def getargspec(f): """A replacement for inspect.getargspec""" spec = getfullargspec(f) return ArgSpec(spec.args, spec.varargs, spec.varkw, spec.defaults) DEF = re.compile(r'\sdef\s([_\w][_\w\d])\s\(') # basic functionality class FunctionMaker(object): """ An object with the ability to create functions with a given signature. It has attributes name, doc, module, signature, defaults, dict and methods update and make. """ # Atomic get-and-increment provided by the GIL _compile_count = itertools.count() def __init__(self, func=None, name=None, signature=None, defaults=None, doc=None, module=None, funcdict=None): self.shortsignature = signature if func: # func can be a class or a callable, but not an instance method self.name = func.__name__ if self.name == '<lambda>': # small hack for lambda functions self.name = '_lambda_' self.doc = func.__doc__ self.module = func.__module__ if inspect.isfunction(func): argspec = getfullargspec(func) self.annotations = getattr(func, '__annotations__', {}) for a in ('args', 'varargs', 'varkw', 'defaults', 'kwonlyargs', 'kwonlydefaults'): setattr(self, a, getattr(argspec, a)) for i, arg in enumerate(self.args): setattr(self, 'arg%d' % i, arg) if sys.version_info < (3,): # easy way self.shortsignature = self.signature = ( inspect.formatargspec( formatvalue=lambda val: "", argspec)[1:-1]) else: # Python 3 way allargs = list(self.args) allshortargs = list(self.args) if self.varargs: allargs.append('' + self.varargs) allshortargs.append('' + self.varargs) elif self.kwonlyargs: allargs.append('') # single star syntax for a in self.kwonlyargs: allargs.append('%s=None' % a) allshortargs.append('%s=%s' % (a, a)) if self.varkw: allargs.append('' + self.varkw) allshortargs.append('' + self.varkw) self.signature = ', '.join(allargs) self.shortsignature = ', '.join(allshortargs) self.dict = func.__dict__.copy() # func=None happens when decorating a caller if name: self.name = name if signature is not None: self.signature = signature if defaults: self.defaults = defaults if doc: self.doc = doc if module: self.module = module if funcdict: self.dict = funcdict # check existence required attributes assert hasattr(self, 'name') if not hasattr(self, 'signature'): raise TypeError('You are decorating a non function: %s' % func) def update(self, func, kw): "Update the signature of func with the data in self" func.__name__ = self.name func.__doc__ = getattr(self, 'doc', None) func.__dict__ = getattr(self, 'dict', {}) func.__defaults__ = getattr(self, 'defaults', ()) func.__kwdefaults__ = getattr(self, 'kwonlydefaults', None) func.__annotations__ = getattr(self, 'annotations', None) try: frame = sys._getframe(3) except AttributeError: # for IronPython and similar implementations callermodule = '?' else: callermodule = frame.f_globals.get('__name__', '?') func.__module__ = getattr(self, 'module', callermodule) func.__dict__.update(kw) def make(self, src_templ, evaldict=None, addsource=False, attrs): "Make a new function from a given template and update the signature" src = src_templ % vars(self) # expand name and signature evaldict = evaldict or {} mo = DEF.match(src) if mo is None: raise SyntaxError('not a valid function template\n%s' % src) name = mo.group(1) # extract the function name names = set([name] + [arg.strip(' ') for arg in self.shortsignature.split(',')]) for n in names: if n in ('_func_', '_call_'): raise NameError('%s is overridden in\n%s' % (n, src)) if not src.endswith('\n'): # add a newline for old Pythons src += '\n' # Ensure each generated function has a unique filename for profilers # (such as cProfile) that depend on the tuple of (<filename>, # <definition line>, <function name>) being unique. filename = '<decorator-gen-%d>' % (next(self._compile_count),) try: code = compile(src, filename, 'single') exec(code, evaldict) except: print('Error in generated code:', file=sys.stderr) print(src, file=sys.stderr) raise func = evaldict[name] if addsource: attrs['__source__'] = src self.update(func, attrs) return func @classmethod def create(cls, obj, body, evaldict, defaults=None, doc=None, module=None, addsource=True, attrs): """ Create a function from the strings name, signature and body. evaldict is the evaluation dictionary. If addsource is true an attribute __source__ is added to the result. The attributes attrs are added, if any. """ if isinstance(obj, str): # "name(signature)" name, rest = obj.strip().split('(', 1) signature = rest[:-1] # strip a right parens func = None else: # a function name = None signature = None func = obj self = cls(func, name, signature, defaults, doc, module) ibody = '\n'.join(' ' + line for line in body.splitlines()) return self.make('def %(name)s(%(signature)s):\n' + ibody, evaldict, addsource, attrs) def decorate(func, caller): """ decorate(func, caller) decorates a function using a caller. """ evaldict = dict(_call_=caller, _func_=func) fun = FunctionMaker.create( func, "return _call_(_func_, %(shortsignature)s)", evaldict, __wrapped__=func) if hasattr(func, '__qualname__'): fun.__qualname__ = func.__qualname__ return fun def decorator(caller, _func=None): """decorator(caller) converts a caller function into a decorator""" if _func is not None: # return a decorated function # this is obsolete behavior; you should use decorate instead return decorate(_func, caller) # else return a decorator function if inspect.isclass(caller): name = caller.__name__.lower() doc = 'decorator(%s) converts functions/generators into ' \ 'factories of %s objects' % (caller.__name__, caller.__name__) elif inspect.isfunction(caller): if caller.__name__ == '<lambda>': name = '_lambda_' else: name = caller.__name__ doc = caller.__doc__ else: # assume caller is an object with a __call__ method name = caller.__class__.__name__.lower() doc = caller.__call__.__doc__ evaldict = dict(_call_=caller, _decorate_=decorate) return FunctionMaker.create( '%s(func)' % name, 'return _decorate_(func, _call_)', evaldict, doc=doc, module=caller.__module__, __wrapped__=caller) # ####################### contextmanager ####################### # try: # Python >= 3.2 from contextlib import _GeneratorContextManager except ImportError: # Python >= 2.5 from contextlib import GeneratorContextManager as _GeneratorContextManager class ContextManager(_GeneratorContextManager): def __call__(self, func): """Context manager decorator""" return FunctionMaker.create( func, "with _self_: return _func_(%(shortsignature)s)", dict(_self_=self, _func_=func), __wrapped__=func) init = getfullargspec(_GeneratorContextManager.__init__) n_args = len(init.args) if n_args == 2 and not init.varargs: # (self, genobj) Python 2.7 def __init__(self, g, a, *k): return _GeneratorContextManager.__init__(self, g(a, *k)) ContextManager.__init__ = __init__ elif n_args == 2 and init.varargs: # (self, gen, a, *k) Python 3.4 pass elif n_args == 4: # (self, gen, args, kwds) Python 3.5 def __init__(self, g, a, *k): return _GeneratorContextManager.__init__(self, g, a, k) ContextManager.__init__ = __init__ contextmanager = decorator(ContextManager) # ############################ dispatch_on ############################ # def append(a, vancestors): """ Append ``a`` to the list of the virtual ancestors, unless it is already included. """ add = True for j, va in enumerate(vancestors): if issubclass(va, a): add = False break if issubclass(a, va): vancestors[j] = a add = False if add: vancestors.append(a) # inspired from simplegeneric by P.J. Eby and functools.singledispatch def dispatch_on(dispatch_args): """ Factory of decorators turning a function into a generic function dispatching on the given arguments. """ assert dispatch_args, 'No dispatch args passed' dispatch_str = '(%s,)' % ', '.join(dispatch_args) def check(arguments, wrong=operator.ne, msg=''): """Make sure one passes the expected number of arguments""" if wrong(len(arguments), len(dispatch_args)): raise TypeError('Expected %d arguments, got %d%s' % (len(dispatch_args), len(arguments), msg)) def gen_func_dec(func): """Decorator turning a function into a generic function""" # first check the dispatch arguments argset = set(getfullargspec(func).args) if not set(dispatch_args) <= argset: raise NameError('Unknown dispatch arguments %s' % dispatch_str) typemap = {} def vancestors(types): """ Get a list of sets of virtual ancestors for the given types """ check(types) ras = [[] for _ in range(len(dispatch_args))] for types_ in typemap: for t, type_, ra in zip(types, types_, ras): if issubclass(t, type_) and type_ not in t.__mro__: append(type_, ra) return [set(ra) for ra in ras] def ancestors(types): """ Get a list of virtual MROs, one for each type """ check(types) lists = [] for t, vas in zip(types, vancestors(types)): n_vas = len(vas) if n_vas > 1: raise RuntimeError( 'Ambiguous dispatch for %s: %s' % (t, vas)) elif n_vas == 1: va, = vas mro = type('t', (t, va), {}).__mro__[1:] else: mro = t.__mro__ lists.append(mro[:-1]) # discard t and object return lists def register(types): """ Decorator to register an implementation for the given types """ check(types) def dec(f): check(getfullargspec(f).args, operator.lt, ' in ' + f.__name__) typemap[types] = f return f return dec def dispatch_info(types): """ An utility to introspect the dispatch algorithm """ check(types) lst = [] for anc in itertools.product(ancestors(types)): lst.append(tuple(a.__name__ for a in anc)) return lst def _dispatch(dispatch_args, args, *kw): types = tuple(type(arg) for arg in dispatch_args) try: # fast path f = typemap[types] except KeyError: pass else: return f(args, *kw) combinations = itertools.product(ancestors(types)) next(combinations) # the first one has been already tried for types_ in combinations: f = typemap.get(types_) if f is not None: return f(args, *kw) # else call the default implementation return func(args, **kw) return FunctionMaker.create( func, 'return _f_(%s, %%(shortsignature)s)' % dispatch_str, dict(_f_=_dispatch), register=register, default=func, typemap=typemap, vancestors=vancestors, ancestors=ancestors, dispatch_info=dispatch_info, __wrapped__=func) gen_func_dec.__name__ = 'dispatch_on' + dispatch_str return gen_func_dec	15.43 KB	2026-01-20 12:58
📄 exposition.py from __future__ import unicode_literals import base64 from contextlib import closing import os import socket import sys import threading from wsgiref.simple_server import make_server, WSGIServer, WSGIRequestHandler from .openmetrics import exposition as openmetrics from .registry import REGISTRY from .utils import floatToGoString try: from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer from SocketServer import ThreadingMixIn from urllib2 import build_opener, Request, HTTPHandler from urllib import quote_plus from urlparse import parse_qs, urlparse except ImportError: # Python 3 from http.server import BaseHTTPRequestHandler, HTTPServer from socketserver import ThreadingMixIn from urllib.request import build_opener, Request, HTTPHandler from urllib.parse import quote_plus, parse_qs, urlparse CONTENT_TYPE_LATEST = str('text/plain; version=0.0.4; charset=utf-8') """Content type of the latest text format""" PYTHON26_OR_OLDER = sys.version_info < (2, 7) PYTHON376_OR_NEWER = sys.version_info > (3, 7, 5) def _bake_output(registry, accept_header, params): """Bake output for metrics output.""" encoder, content_type = choose_encoder(accept_header) if 'name[]' in params: registry = registry.restricted_registry(params['name[]']) output = encoder(registry) return str('200 OK'), (str('Content-Type'), content_type), output def make_wsgi_app(registry=REGISTRY): """Create a WSGI app which serves the metrics from a registry.""" def prometheus_app(environ, start_response): # Prepare parameters accept_header = environ.get('HTTP_ACCEPT') params = parse_qs(environ.get('QUERY_STRING', '')) # Bake output status, header, output = _bake_output(registry, accept_header, params) # Return output start_response(status, [header]) return [output] return prometheus_app class _SilentHandler(WSGIRequestHandler): """WSGI handler that does not log requests.""" def log_message(self, format, args): """Log nothing.""" class ThreadingWSGIServer(ThreadingMixIn, WSGIServer): """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. daemon_threads = True def start_wsgi_server(port, addr='', registry=REGISTRY): """Starts a WSGI server for prometheus metrics as a daemon thread.""" app = make_wsgi_app(registry) httpd = make_server(addr, port, app, ThreadingWSGIServer, handler_class=_SilentHandler) t = threading.Thread(target=httpd.serve_forever) t.daemon = True t.start() start_http_server = start_wsgi_server def generate_latest(registry=REGISTRY): """Returns the metrics from the registry in latest text format as a string.""" def sample_line(line): if line.labels: labelstr = '{{{0}}}'.format(','.join( ['{0}="{1}"'.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 = ' {0:d}'.format(int(float(line.timestamp) 1000)) return '{0}{1} {2}{3}\n'.format( line.name, labelstr, floatToGoString(line.value), timestamp) output = [] for metric in registry.collect(): try: mname = metric.name mtype = metric.type # Munging from OpenMetrics into Prometheus format. if mtype == 'counter': mname = mname + '_total' elif mtype == 'info': mname = mname + '_info' mtype = 'gauge' elif mtype == 'stateset': mtype = 'gauge' elif mtype == 'gaugehistogram': # A gauge histogram is really a gauge, # but this captures the structure better. mtype = 'histogram' elif mtype == 'unknown': mtype = 'untyped' output.append('# HELP {0} {1}\n'.format( mname, metric.documentation.replace('\\', r'\\').replace('\n', r'\n'))) output.append('# TYPE {0} {1}\n'.format(mname, mtype)) om_samples = {} for s in metric.samples: for suffix in ['_created', '_gsum', '_gcount']: if s.name == metric.name + suffix: # OpenMetrics specific sample, put in a gauge at the end. om_samples.setdefault(suffix, []).append(sample_line(s)) break else: output.append(sample_line(s)) except Exception as exception: exception.args = (exception.args or ('',)) + (metric,) raise for suffix, lines in sorted(om_samples.items()): output.append('# HELP {0}{1} {2}\n'.format(metric.name, suffix, metric.documentation.replace('\\', r'\\').replace('\n', r'\n'))) output.append('# TYPE {0}{1} gauge\n'.format(metric.name, suffix)) output.extend(lines) return ''.join(output).encode('utf-8') def choose_encoder(accept_header): accept_header = accept_header or '' for accepted in accept_header.split(','): if accepted.split(';')[0].strip() == 'application/openmetrics-text': return (openmetrics.generate_latest, openmetrics.CONTENT_TYPE_LATEST) return generate_latest, CONTENT_TYPE_LATEST class MetricsHandler(BaseHTTPRequestHandler): """HTTP handler that gives metrics from ``REGISTRY``.""" registry = REGISTRY def do_GET(self): # Prepare parameters registry = self.registry accept_header = self.headers.get('Accept') params = parse_qs(urlparse(self.path).query) # Bake output status, header, output = _bake_output(registry, accept_header, params) # Return output self.send_response(int(status.split(' ')[0])) self.send_header(header) self.end_headers() self.wfile.write(output) def log_message(self, format, args): """Log nothing.""" @classmethod def factory(cls, registry): """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 def write_to_textfile(path, registry): """Write metrics to the given path. This is intended for use with the Node exporter textfile collector. The path must end in .prom for the textfile collector to process it.""" tmppath = '%s.%s.%s' % (path, os.getpid(), threading.current_thread().ident) with open(tmppath, 'wb') as f: f.write(generate_latest(registry)) # rename(2) is atomic. os.rename(tmppath, path) def default_handler(url, method, timeout, headers, data): """Default handler that implements HTTP/HTTPS connections. Used by the push_to_gateway functions. Can be re-used by other handlers.""" def handle(): request = Request(url, data=data) request.get_method = lambda: method for k, v in headers: request.add_header(k, v) resp = build_opener(HTTPHandler).open(request, timeout=timeout) if resp.code >= 400: raise IOError("error talking to pushgateway: {0} {1}".format( resp.code, resp.msg)) return handle def basic_auth_handler(url, method, timeout, headers, data, username=None, password=None): """Handler that implements HTTP/HTTPS connections with Basic Auth. Sets auth headers using supplied 'username' and 'password', if set. Used by the push_to_gateway functions. Can be re-used by other handlers.""" def handle(): """Handler that implements HTTP Basic Auth. """ if username is not None and password is not None: auth_value = '{0}:{1}'.format(username, password).encode('utf-8') auth_token = base64.b64encode(auth_value) auth_header = b'Basic ' + auth_token headers.append(['Authorization', auth_header]) default_handler(url, method, timeout, headers, data)() return handle def push_to_gateway( gateway, job, registry, grouping_key=None, timeout=30, handler=default_handler): """Push metrics to the given pushgateway. `gateway` the url for your push gateway. Either of the form 'http://pushgateway.local', or 'pushgateway.local'. Scheme defaults to 'http' if none is provided `job` is the job label to be attached to all pushed metrics `registry` is an instance of CollectorRegistry `grouping_key` please see the pushgateway documentation for details. Defaults to None `timeout` is how long push will attempt to connect before giving up. Defaults to 30s, can be set to None for no timeout. `handler` is an optional function which can be provided to perform requests to the 'gateway'. Defaults to None, in which case an http or https request will be carried out by a default handler. If not None, the argument must be a function which accepts the following arguments: url, method, timeout, headers, and content May be used to implement additional functionality not supported by the built-in default handler (such as SSL client certicates, and HTTP authentication mechanisms). 'url' is the URL for the request, the 'gateway' argument described earlier will form the basis of this URL. 'method' is the HTTP method which should be used when carrying out the request. 'timeout' requests not successfully completed after this many seconds should be aborted. If timeout is None, then the handler should not set a timeout. 'headers' is a list of ("header-name","header-value") tuples which must be passed to the pushgateway in the form of HTTP request headers. The function should raise an exception (e.g. IOError) on failure. 'content' is the data which should be used to form the HTTP Message Body. This overwrites all metrics with the same job and grouping_key. This uses the PUT HTTP method.""" _use_gateway('PUT', gateway, job, registry, grouping_key, timeout, handler) def pushadd_to_gateway( gateway, job, registry, grouping_key=None, timeout=30, handler=default_handler): """PushAdd metrics to the given pushgateway. `gateway` the url for your push gateway. Either of the form 'http://pushgateway.local', or 'pushgateway.local'. Scheme defaults to 'http' if none is provided `job` is the job label to be attached to all pushed metrics `registry` is an instance of CollectorRegistry `grouping_key` please see the pushgateway documentation for details. Defaults to None `timeout` is how long push will attempt to connect before giving up. Defaults to 30s, can be set to None for no timeout. `handler` is an optional function which can be provided to perform requests to the 'gateway'. Defaults to None, in which case an http or https request will be carried out by a default handler. See the 'prometheus_client.push_to_gateway' documentation for implementation requirements. This replaces metrics with the same name, job and grouping_key. This uses the POST HTTP method.""" _use_gateway('POST', gateway, job, registry, grouping_key, timeout, handler) def delete_from_gateway( gateway, job, grouping_key=None, timeout=30, handler=default_handler): """Delete metrics from the given pushgateway. `gateway` the url for your push gateway. Either of the form 'http://pushgateway.local', or 'pushgateway.local'. Scheme defaults to 'http' if none is provided `job` is the job label to be attached to all pushed metrics `grouping_key` please see the pushgateway documentation for details. Defaults to None `timeout` is how long delete will attempt to connect before giving up. Defaults to 30s, can be set to None for no timeout. `handler` is an optional function which can be provided to perform requests to the 'gateway'. Defaults to None, in which case an http or https request will be carried out by a default handler. See the 'prometheus_client.push_to_gateway' documentation for implementation requirements. This deletes metrics with the given job and grouping_key. This uses the DELETE HTTP method.""" _use_gateway('DELETE', gateway, job, None, grouping_key, timeout, handler) def _use_gateway(method, gateway, job, registry, grouping_key, timeout, handler): gateway_url = urlparse(gateway) # See https://bugs.python.org/issue27657 for details on urlparse in py>=3.7.6. if not gateway_url.scheme or ( (PYTHON376_OR_NEWER or PYTHON26_OR_OLDER) and gateway_url.scheme not in ['http', 'https'] ): gateway = 'http://{0}'.format(gateway) url = '{0}/metrics/{1}/{2}'.format(gateway, _escape_grouping_key("job", job)) data = b'' if method != 'DELETE': data = generate_latest(registry) if grouping_key is None: grouping_key = {} url += ''.join( '/{0}/{1}'.format(_escape_grouping_key(str(k), str(v))) for k, v in sorted(grouping_key.items())) handler( url=url, method=method, timeout=timeout, headers=[('Content-Type', CONTENT_TYPE_LATEST)], data=data, )() def _escape_grouping_key(k, v): if v == "" : # Per https://github.com/prometheus/pushgateway/pull/346. return k + "@base64", "=" elif '/' in v: # Added in Pushgateway 0.9.0. return k + "@base64", base64.urlsafe_b64encode(v.encode("utf-8")).decode("utf-8") else: return k, quote_plus(v) def instance_ip_grouping_key(): """Grouping key with instance set to the IP Address of this host.""" with closing(socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) as s: s.connect(('localhost', 0)) return {'instance': s.getsockname()[0]} try: # Python >3.5 only from .asgi import make_asgi_app except: pass	14.92 KB	2026-01-20 12:58
📄 gc_collector.py from __future__ import unicode_literals import gc import platform from .metrics_core import CounterMetricFamily from .registry import REGISTRY class GCCollector(object): """Collector for Garbage collection statistics.""" def __init__(self, registry=REGISTRY): if not hasattr(gc, 'get_stats') or platform.python_implementation() != 'CPython': return registry.register(self) def collect(self): collected = CounterMetricFamily( 'python_gc_objects_collected', 'Objects collected during gc', labels=['generation'], ) uncollectable = CounterMetricFamily( 'python_gc_objects_uncollectable', 'Uncollectable object found during GC', labels=['generation'], ) collections = CounterMetricFamily( 'python_gc_collections', 'Number of times this generation was collected', labels=['generation'], ) for generation, stat in enumerate(gc.get_stats()): generation = str(generation) collected.add_metric([generation], value=stat['collected']) uncollectable.add_metric([generation], value=stat['uncollectable']) collections.add_metric([generation], value=stat['collections']) return [collected, uncollectable, collections] GC_COLLECTOR = GCCollector() """Default GCCollector in default Registry REGISTRY."""	1.42 KB	2026-01-20 12:58
📄 metrics.py import sys from threading import Lock import time import types from . import values # retain this import style for testability from .context_managers import ExceptionCounter, InprogressTracker, Timer from .metrics_core import ( Metric, METRIC_LABEL_NAME_RE, METRIC_NAME_RE, RESERVED_METRIC_LABEL_NAME_RE, ) from .registry import REGISTRY from .utils import floatToGoString, INF if sys.version_info > (3,): unicode = str create_bound_method = types.MethodType else: def create_bound_method(func, obj): return types.MethodType(func, obj, obj.__class__) def _build_full_name(metric_type, name, namespace, subsystem, unit): full_name = '' if namespace: full_name += namespace + '_' if subsystem: full_name += subsystem + '_' full_name += name if metric_type == 'counter' and full_name.endswith('_total'): full_name = full_name[:-6] # Munge to OpenMetrics. if unit and not full_name.endswith("_" + unit): full_name += "_" + unit if unit and metric_type in ('info', 'stateset'): raise ValueError('Metric name is of a type that cannot have a unit: ' + full_name) return full_name def _validate_labelnames(cls, labelnames): labelnames = tuple(labelnames) for l in labelnames: if not METRIC_LABEL_NAME_RE.match(l): raise ValueError('Invalid label metric name: ' + l) if RESERVED_METRIC_LABEL_NAME_RE.match(l): raise ValueError('Reserved label metric name: ' + l) if l in cls._reserved_labelnames: raise ValueError('Reserved label metric name: ' + l) return labelnames class MetricWrapperBase(object): _type = None _reserved_labelnames = () def _is_observable(self): # Whether this metric is observable, i.e. # * a metric without label names and values, or # * the child of a labelled metric. return not self._labelnames or (self._labelnames and self._labelvalues) def _raise_if_not_observable(self): # Functions that mutate the state of the metric, for example incrementing # a counter, will fail if the metric is not observable, because only if a # metric is observable will the value be initialized. if not self._is_observable(): raise ValueError('%s metric is missing label values' % str(self._type)) def _is_parent(self): return self._labelnames and not self._labelvalues def _get_metric(self): return Metric(self._name, self._documentation, self._type, self._unit) def describe(self): return [self._get_metric()] def collect(self): metric = self._get_metric() for suffix, labels, value in self._samples(): metric.add_sample(self._name + suffix, labels, value) return [metric] def __str__(self): return "{0}:{1}".format(self._type, self._name) def __repr__(self): metric_type = type(self) return "{0}.{1}({2})".format(metric_type.__module__, metric_type.__name__, self._name) def __init__(self, name, documentation, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY, labelvalues=None, ): self._name = _build_full_name(self._type, name, namespace, subsystem, unit) self._labelnames = _validate_labelnames(self, labelnames) self._labelvalues = tuple(labelvalues or ()) self._kwargs = {} self._documentation = documentation self._unit = unit if not METRIC_NAME_RE.match(self._name): raise ValueError('Invalid metric name: ' + self._name) if self._is_parent(): # Prepare the fields needed for child metrics. self._lock = Lock() self._metrics = {} if self._is_observable(): self._metric_init() if not self._labelvalues: # Register the multi-wrapper parent metric, or if a label-less metric, the whole shebang. if registry: registry.register(self) def labels(self, labelvalues, labelkwargs): """Return the child for the given labelset. All metrics can have labels, allowing grouping of related time series. Taking a counter as an example: from prometheus_client import Counter c = Counter('my_requests_total', 'HTTP Failures', ['method', 'endpoint']) c.labels('get', '/').inc() c.labels('post', '/submit').inc() Labels can also be provided as keyword arguments: from prometheus_client import Counter c = Counter('my_requests_total', 'HTTP Failures', ['method', 'endpoint']) c.labels(method='get', endpoint='/').inc() c.labels(method='post', endpoint='/submit').inc() See the best practices on [naming](http://prometheus.io/docs/practices/naming/) and [labels](http://prometheus.io/docs/practices/instrumentation/#use-labels). """ if not self._labelnames: raise ValueError('No label names were set when constructing %s' % self) if self._labelvalues: raise ValueError('%s already has labels set (%s); can not chain calls to .labels()' % ( self, dict(zip(self._labelnames, self._labelvalues)) )) if labelvalues and labelkwargs: raise ValueError("Can't pass both args and kwargs") if labelkwargs: if sorted(labelkwargs) != sorted(self._labelnames): raise ValueError('Incorrect label names') labelvalues = tuple(unicode(labelkwargs[l]) for l in self._labelnames) else: if len(labelvalues) != len(self._labelnames): raise ValueError('Incorrect label count') labelvalues = tuple(unicode(l) for l in labelvalues) with self._lock: if labelvalues not in self._metrics: self._metrics[labelvalues] = self.__class__( self._name, documentation=self._documentation, labelnames=self._labelnames, unit=self._unit, labelvalues=labelvalues, self._kwargs ) return self._metrics[labelvalues] def remove(self, *labelvalues): if not self._labelnames: raise ValueError('No label names were set when constructing %s' % self) """Remove the given labelset from the metric.""" if len(labelvalues) != len(self._labelnames): raise ValueError('Incorrect label count (expected %d, got %s)' % (len(self._labelnames), labelvalues)) labelvalues = tuple(unicode(l) for l in labelvalues) with self._lock: del self._metrics[labelvalues] def _samples(self): if self._is_parent(): return self._multi_samples() else: return self._child_samples() def _multi_samples(self): with self._lock: metrics = self._metrics.copy() for labels, metric in metrics.items(): series_labels = list(zip(self._labelnames, labels)) for suffix, sample_labels, value in metric._samples(): yield (suffix, dict(series_labels + list(sample_labels.items())), value) def _child_samples(self): # pragma: no cover raise NotImplementedError('_child_samples() must be implemented by %r' % self) def _metric_init(self): # pragma: no cover """ Initialize the metric object as a child, i.e. when it has labels (if any) set. This is factored as a separate function to allow for deferred initialization. """ raise NotImplementedError('_metric_init() must be implemented by %r' % self) class Counter(MetricWrapperBase): """A Counter tracks counts of events or running totals. Example use cases for Counters: - Number of requests processed - Number of items that were inserted into a queue - Total amount of data that a system has processed Counters can only go up (and be reset when the process restarts). If your use case can go down, you should use a Gauge instead. An example for a Counter: from prometheus_client import Counter c = Counter('my_failures_total', 'Description of counter') c.inc() # Increment by 1 c.inc(1.6) # Increment by given value There are utilities to count exceptions raised: @c.count_exceptions() def f(): pass with c.count_exceptions(): pass # Count only one type of exception with c.count_exceptions(ValueError): pass """ _type = 'counter' def _metric_init(self): self._value = values.ValueClass(self._type, self._name, self._name + '_total', self._labelnames, self._labelvalues) self._created = time.time() def inc(self, amount=1): """Increment counter by the given amount.""" if amount < 0: raise ValueError('Counters can only be incremented by non-negative amounts.') self._value.inc(amount) def count_exceptions(self, exception=Exception): """Count exceptions in a block of code or function. Can be used as a function decorator or context manager. Increments the counter when an exception of the given type is raised up out of the code. """ self._raise_if_not_observable() return ExceptionCounter(self, exception) def _child_samples(self): return ( ('_total', {}, self._value.get()), ('_created', {}, self._created), ) class Gauge(MetricWrapperBase): """Gauge metric, to report instantaneous values. Examples of Gauges include: - Inprogress requests - Number of items in a queue - Free memory - Total memory - Temperature Gauges can go both up and down. from prometheus_client import Gauge g = Gauge('my_inprogress_requests', 'Description of gauge') g.inc() # Increment by 1 g.dec(10) # Decrement by given value g.set(4.2) # Set to a given value There are utilities for common use cases: g.set_to_current_time() # Set to current unixtime # Increment when entered, decrement when exited. @g.track_inprogress() def f(): pass with g.track_inprogress(): pass A Gauge can also take its value from a callback: d = Gauge('data_objects', 'Number of objects') my_dict = {} d.set_function(lambda: len(my_dict)) """ _type = 'gauge' _MULTIPROC_MODES = frozenset(('min', 'max', 'livesum', 'liveall', 'all')) def __init__(self, name, documentation, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY, labelvalues=None, multiprocess_mode='all', ): self._multiprocess_mode = multiprocess_mode if multiprocess_mode not in self._MULTIPROC_MODES: raise ValueError('Invalid multiprocess mode: ' + multiprocess_mode) super(Gauge, self).__init__( name=name, documentation=documentation, labelnames=labelnames, namespace=namespace, subsystem=subsystem, unit=unit, registry=registry, labelvalues=labelvalues, ) self._kwargs['multiprocess_mode'] = self._multiprocess_mode def _metric_init(self): self._value = values.ValueClass( self._type, self._name, self._name, self._labelnames, self._labelvalues, multiprocess_mode=self._multiprocess_mode ) def inc(self, amount=1): """Increment gauge by the given amount.""" self._value.inc(amount) def dec(self, amount=1): """Decrement gauge by the given amount.""" self._value.inc(-amount) def set(self, value): """Set gauge to the given value.""" self._value.set(float(value)) def set_to_current_time(self): """Set gauge to the current unixtime.""" self.set(time.time()) def track_inprogress(self): """Track inprogress blocks of code or functions. Can be used as a function decorator or context manager. Increments the gauge when the code is entered, and decrements when it is exited. """ self._raise_if_not_observable() return InprogressTracker(self) def time(self): """Time a block of code or function, and set the duration in seconds. Can be used as a function decorator or context manager. """ self._raise_if_not_observable() return Timer(self.set) def set_function(self, f): """Call the provided function to return the Gauge value. The function must return a float, and may be called from multiple threads. All other methods of the Gauge become NOOPs. """ def samples(self): return (('', {}, float(f())),) self._child_samples = create_bound_method(samples, self) def _child_samples(self): return (('', {}, self._value.get()),) class Summary(MetricWrapperBase): """A Summary tracks the size and number of events. Example use cases for Summaries: - Response latency - Request size Example for a Summary: from prometheus_client import Summary s = Summary('request_size_bytes', 'Request size (bytes)') s.observe(512) # Observe 512 (bytes) Example for a Summary using time: from prometheus_client import Summary REQUEST_TIME = Summary('response_latency_seconds', 'Response latency (seconds)') @REQUEST_TIME.time() def create_response(request): '''A dummy function''' time.sleep(1) Example for using the same Summary object as a context manager: with REQUEST_TIME.time(): pass # Logic to be timed """ _type = 'summary' _reserved_labelnames = ['quantile'] def _metric_init(self): self._count = values.ValueClass(self._type, self._name, self._name + '_count', self._labelnames, self._labelvalues) self._sum = values.ValueClass(self._type, self._name, self._name + '_sum', self._labelnames, self._labelvalues) self._created = time.time() def observe(self, amount): """Observe the given amount.""" self._count.inc(1) self._sum.inc(amount) def time(self): """Time a block of code or function, and observe the duration in seconds. Can be used as a function decorator or context manager. """ self._raise_if_not_observable() return Timer(self.observe) def _child_samples(self): return ( ('_count', {}, self._count.get()), ('_sum', {}, self._sum.get()), ('_created', {}, self._created)) class Histogram(MetricWrapperBase): """A Histogram tracks the size and number of events in buckets. You can use Histograms for aggregatable calculation of quantiles. Example use cases: - Response latency - Request size Example for a Histogram: from prometheus_client import Histogram h = Histogram('request_size_bytes', 'Request size (bytes)') h.observe(512) # Observe 512 (bytes) Example for a Histogram using time: from prometheus_client import Histogram REQUEST_TIME = Histogram('response_latency_seconds', 'Response latency (seconds)') @REQUEST_TIME.time() def create_response(request): '''A dummy function''' time.sleep(1) Example of using the same Histogram object as a context manager: with REQUEST_TIME.time(): pass # Logic to be timed The default buckets are intended to cover a typical web/rpc request from milliseconds to seconds. They can be overridden by passing `buckets` keyword argument to `Histogram`. """ _type = 'histogram' _reserved_labelnames = ['le'] DEFAULT_BUCKETS = (.005, .01, .025, .05, .075, .1, .25, .5, .75, 1.0, 2.5, 5.0, 7.5, 10.0, INF) def __init__(self, name, documentation, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY, labelvalues=None, buckets=DEFAULT_BUCKETS, ): self._prepare_buckets(buckets) super(Histogram, self).__init__( name=name, documentation=documentation, labelnames=labelnames, namespace=namespace, subsystem=subsystem, unit=unit, registry=registry, labelvalues=labelvalues, ) self._kwargs['buckets'] = buckets def _prepare_buckets(self, buckets): buckets = [float(b) for b in buckets] if buckets != sorted(buckets): # This is probably an error on the part of the user, # so raise rather than sorting for them. raise ValueError('Buckets not in sorted order') if buckets and buckets[-1] != INF: buckets.append(INF) if len(buckets) < 2: raise ValueError('Must have at least two buckets') self._upper_bounds = buckets def _metric_init(self): self._buckets = [] self._created = time.time() bucket_labelnames = self._labelnames + ('le',) self._sum = values.ValueClass(self._type, self._name, self._name + '_sum', self._labelnames, self._labelvalues) for b in self._upper_bounds: self._buckets.append(values.ValueClass( self._type, self._name, self._name + '_bucket', bucket_labelnames, self._labelvalues + (floatToGoString(b),)) ) def observe(self, amount): """Observe the given amount.""" self._sum.inc(amount) for i, bound in enumerate(self._upper_bounds): if amount <= bound: self._buckets[i].inc(1) break def time(self): """Time a block of code or function, and observe the duration in seconds. Can be used as a function decorator or context manager. """ return Timer(self.observe) def _child_samples(self): samples = [] acc = 0 for i, bound in enumerate(self._upper_bounds): acc += self._buckets[i].get() samples.append(('_bucket', {'le': floatToGoString(bound)}, acc)) samples.append(('_count', {}, acc)) if self._upper_bounds[0] >= 0: samples.append(('_sum', {}, self._sum.get())) samples.append(('_created', {}, self._created)) return tuple(samples) class Info(MetricWrapperBase): """Info metric, key-value pairs. Examples of Info include: - Build information - Version information - Potential target metadata Example usage: from prometheus_client import Info i = Info('my_build', 'Description of info') i.info({'version': '1.2.3', 'buildhost': 'foo@bar'}) Info metrics do not work in multiprocess mode. """ _type = 'info' def _metric_init(self): self._labelname_set = set(self._labelnames) self._lock = Lock() self._value = {} def info(self, val): """Set info metric.""" if self._labelname_set.intersection(val.keys()): raise ValueError('Overlapping labels for Info metric, metric: %s child: %s' % ( self._labelnames, val)) with self._lock: self._value = dict(val) def _child_samples(self): with self._lock: return (('_info', self._value, 1.0,),) class Enum(MetricWrapperBase): """Enum metric, which of a set of states is true. Example usage: from prometheus_client import Enum e = Enum('task_state', 'Description of enum', states=['starting', 'running', 'stopped']) e.state('running') The first listed state will be the default. Enum metrics do not work in multiprocess mode. """ _type = 'stateset' def __init__(self, name, documentation, labelnames=(), namespace='', subsystem='', unit='', registry=REGISTRY, labelvalues=None, states=None, ): super(Enum, self).__init__( name=name, documentation=documentation, labelnames=labelnames, namespace=namespace, subsystem=subsystem, unit=unit, registry=registry, labelvalues=labelvalues, ) if name in labelnames: raise ValueError('Overlapping labels for Enum metric: %s' % (name,)) if not states: raise ValueError('No states provided for Enum metric: %s' % (name,)) self._kwargs['states'] = self._states = states def _metric_init(self): self._value = 0 self._lock = Lock() def state(self, state): """Set enum metric state.""" self._raise_if_not_observable() with self._lock: self._value = self._states.index(state) def _child_samples(self): with self._lock: return [ ('', {self._name: s}, 1 if i == self._value else 0,) for i, s in enumerate(self._states) ]	21.66 KB	2026-01-20 12:58
📄 metrics_core.py import re from .samples import Sample METRIC_TYPES = ( 'counter', 'gauge', 'summary', 'histogram', 'gaugehistogram', 'unknown', 'info', 'stateset', ) METRIC_NAME_RE = re.compile(r'^[a-zA-Z_:][a-zA-Z0-9_:]$') METRIC_LABEL_NAME_RE = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]$') RESERVED_METRIC_LABEL_NAME_RE = re.compile(r'^__.$') class Metric(object): """A single metric family and its samples. This is intended only for internal use by the instrumentation client. Custom collectors should use GaugeMetricFamily, CounterMetricFamily and SummaryMetricFamily instead. """ def __init__(self, name, documentation, typ, unit=''): if unit and not name.endswith("_" + unit): name += "_" + unit if not METRIC_NAME_RE.match(name): raise ValueError('Invalid metric name: ' + name) self.name = name self.documentation = documentation self.unit = unit if typ == 'untyped': typ = 'unknown' if typ not in METRIC_TYPES: raise ValueError('Invalid metric type: ' + typ) self.type = typ self.samples = [] def add_sample(self, name, labels, value, timestamp=None, exemplar=None): """Add a sample to the metric. Internal-only, do not use.""" self.samples.append(Sample(name, labels, value, timestamp, exemplar)) def __eq__(self, other): return (isinstance(other, Metric) and self.name == other.name and self.documentation == other.documentation and self.type == other.type and self.unit == other.unit and self.samples == other.samples) def __repr__(self): return "Metric(%s, %s, %s, %s, %s)" % ( self.name, self.documentation, self.type, self.unit, self.samples, ) class UnknownMetricFamily(Metric): """A single unknown metric and its samples. For use by custom collectors. """ def __init__(self, name, documentation, value=None, labels=None, unit=''): Metric.__init__(self, name, documentation, 'unknown', unit) if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels, value, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values value: The value of the metric. """ self.samples.append(Sample(self.name, dict(zip(self._labelnames, labels)), value, timestamp)) # For backward compatibility. UntypedMetricFamily = UnknownMetricFamily class CounterMetricFamily(Metric): """A single counter and its samples. For use by custom collectors. """ def __init__(self, name, documentation, value=None, labels=None, created=None, unit=''): # Glue code for pre-OpenMetrics metrics. if name.endswith('_total'): name = name[:-6] Metric.__init__(self, name, documentation, 'counter', unit) if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value, created) def add_metric(self, labels, value, created=None, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values value: The value of the metric created: Optional unix timestamp the child was created at. """ self.samples.append(Sample(self.name + '_total', dict(zip(self._labelnames, labels)), value, timestamp)) if created is not None: self.samples.append(Sample(self.name + '_created', dict(zip(self._labelnames, labels)), created, timestamp)) class GaugeMetricFamily(Metric): """A single gauge and its samples. For use by custom collectors. """ def __init__(self, name, documentation, value=None, labels=None, unit=''): Metric.__init__(self, name, documentation, 'gauge', unit) if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels, value, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values value: A float """ self.samples.append(Sample(self.name, dict(zip(self._labelnames, labels)), value, timestamp)) class SummaryMetricFamily(Metric): """A single summary and its samples. For use by custom collectors. """ def __init__(self, name, documentation, count_value=None, sum_value=None, labels=None, unit=''): Metric.__init__(self, name, documentation, 'summary', unit) if (sum_value is None) != (count_value is None): raise ValueError('count_value and sum_value must be provided together.') if labels is not None and count_value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if count_value is not None: self.add_metric([], count_value, sum_value) def add_metric(self, labels, count_value, sum_value, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values count_value: The count value of the metric. sum_value: The sum value of the metric. """ self.samples.append(Sample(self.name + '_count', dict(zip(self._labelnames, labels)), count_value, timestamp)) self.samples.append(Sample(self.name + '_sum', dict(zip(self._labelnames, labels)), sum_value, timestamp)) class HistogramMetricFamily(Metric): """A single histogram and its samples. For use by custom collectors. """ def __init__(self, name, documentation, buckets=None, sum_value=None, labels=None, unit=''): Metric.__init__(self, name, documentation, 'histogram', unit) if sum_value is not None and buckets is None: raise ValueError('sum value cannot be provided without buckets.') if labels is not None and buckets is not None: raise ValueError('Can only specify at most one of buckets and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if buckets is not None: self.add_metric([], buckets, sum_value) def add_metric(self, labels, buckets, sum_value, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values buckets: A list of lists. Each inner list can be a pair of bucket name and value, or a triple of bucket name, value, and exemplar. The buckets must be sorted, and +Inf present. sum_value: The sum value of the metric. """ for b in buckets: bucket, value = b[:2] exemplar = None if len(b) == 3: exemplar = b[2] self.samples.append(Sample( self.name + '_bucket', dict(list(zip(self._labelnames, labels)) + [('le', bucket)]), value, timestamp, exemplar, )) # +Inf is last and provides the count value. self.samples.append( Sample(self.name + '_count', dict(zip(self._labelnames, labels)), buckets[-1][1], timestamp)) # Don't iunclude sum if there's negative buckets. if float(buckets[0][0]) >= 0 and sum_value is not None: self.samples.append( Sample(self.name + '_sum', dict(zip(self._labelnames, labels)), sum_value, timestamp)) class GaugeHistogramMetricFamily(Metric): """A single gauge histogram and its samples. For use by custom collectors. """ def __init__(self, name, documentation, buckets=None, gsum_value=None, labels=None, unit=''): Metric.__init__(self, name, documentation, 'gaugehistogram', unit) if labels is not None and buckets is not None: raise ValueError('Can only specify at most one of buckets and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if buckets is not None: self.add_metric([], buckets, gsum_value) def add_metric(self, labels, buckets, gsum_value, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values buckets: A list of pairs of bucket names and values. The buckets must be sorted, and +Inf present. gsum_value: The sum value of the metric. """ for bucket, value in buckets: self.samples.append(Sample( self.name + '_bucket', dict(list(zip(self._labelnames, labels)) + [('le', bucket)]), value, timestamp)) # +Inf is last and provides the count value. self.samples.extend([ Sample(self.name + '_gcount', dict(zip(self._labelnames, labels)), buckets[-1][1], timestamp), Sample(self.name + '_gsum', dict(zip(self._labelnames, labels)), gsum_value, timestamp), ]) class InfoMetricFamily(Metric): """A single info and its samples. For use by custom collectors. """ def __init__(self, name, documentation, value=None, labels=None): Metric.__init__(self, name, documentation, 'info') if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels, value, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values value: A dict of labels """ self.samples.append(Sample( self.name + '_info', dict(dict(zip(self._labelnames, labels)), *value), 1, timestamp, )) class StateSetMetricFamily(Metric): """A single stateset and its samples. For use by custom collectors. """ def __init__(self, name, documentation, value=None, labels=None): Metric.__init__(self, name, documentation, 'stateset') if labels is not None and value is not None: raise ValueError('Can only specify at most one of value and labels.') if labels is None: labels = [] self._labelnames = tuple(labels) if value is not None: self.add_metric([], value) def add_metric(self, labels, value, timestamp=None): """Add a metric to the metric family. Args: labels: A list of label values value: A dict of string state names to booleans """ labels = tuple(labels) for state, enabled in sorted(value.items()): v = (1 if enabled else 0) self.samples.append(Sample( self.name, dict(zip(self._labelnames + (self.name,), labels + (state,))), v, timestamp, ))	11.6 KB	2026-01-20 12:58
📄 mmap_dict.py import json import mmap import os import struct _INITIAL_MMAP_SIZE = 1 << 20 _pack_integer_func = struct.Struct(b'i').pack _pack_double_func = struct.Struct(b'd').pack _unpack_integer = struct.Struct(b'i').unpack_from _unpack_double = struct.Struct(b'd').unpack_from # struct.pack_into has atomicity issues because it will temporarily write 0 into # the mmap, resulting in false reads to 0 when experiencing a lot of writes. # Using direct assignment solves this issue. def _pack_double(data, pos, value): data[pos:pos + 8] = _pack_double_func(value) def _pack_integer(data, pos, value): data[pos:pos + 4] = _pack_integer_func(value) def _read_all_values(data, used=0): """Yield (key, value, pos). No locking is performed.""" if used <= 0: # If not valid `used` value is passed in, read it from the file. used = _unpack_integer(data, 0)[0] pos = 8 while pos < used: encoded_len = _unpack_integer(data, pos)[0] # check we are not reading beyond bounds if encoded_len + pos > used: raise RuntimeError('Read beyond file size detected, file is corrupted.') pos += 4 encoded_key = data[pos : pos + encoded_len] padded_len = encoded_len + (8 - (encoded_len + 4) % 8) pos += padded_len value = _unpack_double(data, pos)[0] yield encoded_key.decode('utf-8'), value, pos pos += 8 class MmapedDict(object): """A dict of doubles, backed by an mmapped file. The file starts with a 4 byte int, indicating how much of it is used. Then 4 bytes of padding. There's then a number of entries, consisting of a 4 byte int which is the size of the next field, a utf-8 encoded string key, padding to a 8 byte alignment, and then a 8 byte float which is the value. Not thread safe. """ def __init__(self, filename, read_mode=False): self._f = open(filename, 'rb' if read_mode else 'a+b') self._fname = filename capacity = os.fstat(self._f.fileno()).st_size if capacity == 0: self._f.truncate(_INITIAL_MMAP_SIZE) capacity = _INITIAL_MMAP_SIZE self._capacity = capacity self._m = mmap.mmap(self._f.fileno(), self._capacity, access=mmap.ACCESS_READ if read_mode else mmap.ACCESS_WRITE) self._positions = {} self._used = _unpack_integer(self._m, 0)[0] if self._used == 0: self._used = 8 _pack_integer(self._m, 0, self._used) else: if not read_mode: for key, _, pos in self._read_all_values(): self._positions[key] = pos @staticmethod def read_all_values_from_file(filename): with open(filename, 'rb') as infp: # Read the first block of data, including the first 4 bytes which tell us # how much of the file (which is preallocated to _INITIAL_MMAP_SIZE bytes) is occupied. data = infp.read(mmap.PAGESIZE) used = _unpack_integer(data, 0)[0] if used > len(data): # Then read in the rest, if needed. data += infp.read(used - len(data)) return _read_all_values(data, used) def _init_value(self, key): """Initialize a value. Lock must be held by caller.""" encoded = key.encode('utf-8') # Pad to be 8-byte aligned. padded = encoded + (b' ' * (8 - (len(encoded) + 4) % 8)) value = struct.pack('i{0}sd'.format(len(padded)).encode(), len(encoded), padded, 0.0) while self._used + len(value) > self._capacity: self._capacity *= 2 self._f.truncate(self._capacity) self._m = mmap.mmap(self._f.fileno(), self._capacity) self._m[self._used:self._used + len(value)] = value # Update how much space we've used. self._used += len(value) _pack_integer(self._m, 0, self._used) self._positions[key] = self._used - 8 def _read_all_values(self): """Yield (key, value, pos). No locking is performed.""" return _read_all_values(data=self._m, used=self._used) def read_all_values(self): """Yield (key, value). No locking is performed.""" for k, v, _ in self._read_all_values(): yield k, v def read_value(self, key): if key not in self._positions: self._init_value(key) pos = self._positions[key] # We assume that reading from an 8 byte aligned value is atomic return _unpack_double(self._m, pos)[0] def write_value(self, key, value): if key not in self._positions: self._init_value(key) pos = self._positions[key] # We assume that writing to an 8 byte aligned value is atomic _pack_double(self._m, pos, value) def close(self): if self._f: self._m.close() self._m = None self._f.close() self._f = None def mmap_key(metric_name, name, labelnames, labelvalues): """Format a key for use in the mmap file.""" # ensure labels are in consistent order for identity labels = dict(zip(labelnames, labelvalues)) return json.dumps([metric_name, name, labels], sort_keys=True)	5.14 KB	2026-01-20 12:58
📄 multiprocess.py from __future__ import unicode_literals from collections import defaultdict import glob import json import os from .metrics_core import Metric from .mmap_dict import MmapedDict from .samples import Sample from .utils import floatToGoString try: # Python3 FileNotFoundError except NameError: # Python >= 2.5 FileNotFoundError = IOError MP_METRIC_HELP = 'Multiprocess metric' class MultiProcessCollector(object): """Collector for files for multi-process mode.""" def __init__(self, registry, path=None): if path is None: path = os.environ.get('prometheus_multiproc_dir') if not path or not os.path.isdir(path): raise ValueError('env prometheus_multiproc_dir is not set or not a directory') self._path = path if registry: registry.register(self) @staticmethod def merge(files, accumulate=True): """Merge metrics from given mmap files. By default, histograms are accumulated, as per prometheus wire format. But if writing the merged data back to mmap files, use accumulate=False to avoid compound accumulation. """ metrics = MultiProcessCollector._read_metrics(files) return MultiProcessCollector._accumulate_metrics(metrics, accumulate) @staticmethod def _read_metrics(files): metrics = {} key_cache = {} def _parse_key(key): val = key_cache.get(key) if not val: metric_name, name, labels = json.loads(key) labels_key = tuple(sorted(labels.items())) val = key_cache[key] = (metric_name, name, labels, labels_key) return val for f in files: parts = os.path.basename(f).split('_') typ = parts[0] try: file_values = MmapedDict.read_all_values_from_file(f) except FileNotFoundError: if typ == 'gauge' and parts[1] in ('liveall', 'livesum'): # Those files can disappear between the glob of collect # and now (via a mark_process_dead call) so don't fail if # the file is missing continue raise for key, value, pos in file_values: metric_name, name, labels, labels_key = _parse_key(key) metric = metrics.get(metric_name) if metric is None: metric = Metric(metric_name, MP_METRIC_HELP, typ) metrics[metric_name] = metric if typ == 'gauge': pid = parts[2][:-3] metric._multiprocess_mode = parts[1] metric.add_sample(name, labels_key + (('pid', pid),), value) else: # The duplicates and labels are fixed in the next for. metric.add_sample(name, labels_key, value) return metrics @staticmethod def _accumulate_metrics(metrics, accumulate): for metric in metrics.values(): samples = defaultdict(float) buckets = defaultdict(lambda: defaultdict(float)) samples_setdefault = samples.setdefault for s in metric.samples: name, labels, value, timestamp, exemplar = s if metric.type == 'gauge': without_pid_key = (name, tuple([l for l in labels if l[0] != 'pid'])) if metric._multiprocess_mode == 'min': current = samples_setdefault(without_pid_key, value) if value < current: samples[without_pid_key] = value elif metric._multiprocess_mode == 'max': current = samples_setdefault(without_pid_key, value) if value > current: samples[without_pid_key] = value elif metric._multiprocess_mode == 'livesum': samples[without_pid_key] += value else: # all/liveall samples[(name, labels)] = value elif metric.type == 'histogram': # A for loop with early exit is faster than a genexpr # or a listcomp that ends up building unnecessary things for l in labels: if l[0] == 'le': bucket_value = float(l[1]) # _bucket without_le = tuple(l for l in labels if l[0] != 'le') buckets[without_le][bucket_value] += value break else: # did not find the `le` key # _sum/_count samples[(name, labels)] += value else: # Counter and Summary. samples[(name, labels)] += value # Accumulate bucket values. if metric.type == 'histogram': for labels, values in buckets.items(): acc = 0.0 for bucket, value in sorted(values.items()): sample_key = ( metric.name + '_bucket', labels + (('le', floatToGoString(bucket)),), ) if accumulate: acc += value samples[sample_key] = acc else: samples[sample_key] = value if accumulate: samples[(metric.name + '_count', labels)] = acc # Convert to correct sample format. metric.samples = [Sample(name_, dict(labels), value) for (name_, labels), value in samples.items()] return metrics.values() def collect(self): files = glob.glob(os.path.join(self._path, '*.db')) return self.merge(files, accumulate=True) def mark_process_dead(pid, path=None): """Do bookkeeping for when one process dies in a multi-process setup.""" if path is None: path = os.environ.get('prometheus_multiproc_dir') for f in glob.glob(os.path.join(path, 'gauge_livesum_{0}.db'.format(pid))): os.remove(f) for f in glob.glob(os.path.join(path, 'gauge_liveall_{0}.db'.format(pid))): os.remove(f)	6.32 KB	2026-01-20 12:58
📁 openmetrics	-	2026-03-21 15:45
📄 parser.py from __future__ import unicode_literals import re from .metrics_core import Metric from .samples import Sample try: import StringIO except ImportError: # Python 3 import io as StringIO def text_string_to_metric_families(text): """Parse Prometheus text format from a unicode string. See text_fd_to_metric_families. """ for metric_family in text_fd_to_metric_families(StringIO.StringIO(text)): yield metric_family ESCAPE_SEQUENCES = { '\\\\': '\\', '\\n': '\n', '\\"': '"', } def replace_escape_sequence(match): return ESCAPE_SEQUENCES[match.group(0)] HELP_ESCAPING_RE = re.compile(r'\\[\\n]') ESCAPING_RE = re.compile(r'\\[\\n"]') def _replace_help_escaping(s): return HELP_ESCAPING_RE.sub(replace_escape_sequence, s) def _replace_escaping(s): return ESCAPING_RE.sub(replace_escape_sequence, s) def _is_character_escaped(s, charpos): num_bslashes = 0 while (charpos > num_bslashes and s[charpos - 1 - num_bslashes] == '\\'): num_bslashes += 1 return num_bslashes % 2 == 1 def _parse_labels(labels_string): labels = {} # Return if we don't have valid labels if "=" not in labels_string: return labels escaping = False if "\\" in labels_string: escaping = True # Copy original labels sub_labels = labels_string try: # Process one label at a time while sub_labels: # The label name is before the equal value_start = sub_labels.index("=") label_name = sub_labels[:value_start] sub_labels = sub_labels[value_start + 1:].lstrip() # Find the first quote after the equal quote_start = sub_labels.index('"') + 1 value_substr = sub_labels[quote_start:] # Find the last unescaped quote i = 0 while i < len(value_substr): i = value_substr.index('"', i) if not _is_character_escaped(value_substr, i): break i += 1 # The label value is between the first and last quote quote_end = i + 1 label_value = sub_labels[quote_start:quote_end] # Replace escaping if needed if escaping: label_value = _replace_escaping(label_value) labels[label_name.strip()] = label_value # Remove the processed label from the sub-slice for next iteration sub_labels = sub_labels[quote_end + 1:] next_comma = sub_labels.find(",") + 1 sub_labels = sub_labels[next_comma:].lstrip() return labels except ValueError: raise ValueError("Invalid labels: %s" % labels_string) # If we have multiple values only consider the first def _parse_value_and_timestamp(s): s = s.lstrip() separator = " " if separator not in s: separator = "\t" values = [value.strip() for value in s.split(separator) if value.strip()] if not values: return float(s), None value = float(values[0]) timestamp = (float(values[-1])/1000) if len(values) > 1 else None return value, timestamp def _parse_sample(text): # Detect the labels in the text try: label_start, label_end = text.index("{"), text.rindex("}") # The name is before the labels name = text[:label_start].strip() # We ignore the starting curly brace label = text[label_start + 1:label_end] # The value is after the label end (ignoring curly brace and space) value, timestamp = _parse_value_and_timestamp(text[label_end + 2:]) return Sample(name, _parse_labels(label), value, timestamp) # We don't have labels except ValueError: # Detect what separator is used separator = " " if separator not in text: separator = "\t" name_end = text.index(separator) name = text[:name_end] # The value is after the name value, timestamp = _parse_value_and_timestamp(text[name_end:]) return Sample(name, {}, value, timestamp) def text_fd_to_metric_families(fd): """Parse Prometheus text format from a file descriptor. This is a laxer parser than the main Go parser, so successful parsing does not imply that the parsed text meets the specification. Yields Metric's. """ name = '' documentation = '' typ = 'untyped' samples = [] allowed_names = [] def build_metric(name, documentation, typ, samples): # Munge counters into OpenMetrics representation # used internally. if typ == 'counter': if name.endswith('_total'): name = name[:-6] else: new_samples = [] for s in samples: new_samples.append(Sample(s[0] + '_total', *s[1:])) samples = new_samples metric = Metric(name, documentation, typ) metric.samples = samples return metric for line in fd: line = line.strip() if line.startswith('#'): parts = line.split(None, 3) if len(parts) < 2: continue if parts[1] == 'HELP': if parts[2] != name: if name != '': yield build_metric(name, documentation, typ, samples) # New metric name = parts[2] typ = 'untyped' samples = [] allowed_names = [parts[2]] if len(parts) == 4: documentation = _replace_help_escaping(parts[3]) else: documentation = '' elif parts[1] == 'TYPE': if parts[2] != name: if name != '': yield build_metric(name, documentation, typ, samples) # New metric name = parts[2] documentation = '' samples = [] typ = parts[3] allowed_names = { 'counter': [''], 'gauge': [''], 'summary': ['_count', '_sum', ''], 'histogram': ['_count', '_sum', '_bucket'], }.get(typ, ['']) allowed_names = [name + n for n in allowed_names] else: # Ignore other comment tokens pass elif line == '': # Ignore blank lines pass else: sample = _parse_sample(line) if sample.name not in allowed_names: if name != '': yield build_metric(name, documentation, typ, samples) # New metric, yield immediately as untyped singleton name = '' documentation = '' typ = 'untyped' samples = [] allowed_names = [] yield build_metric(sample[0], documentation, typ, [sample]) else: samples.append(sample) if name != '': yield build_metric(name, documentation, typ, samples)	7.09 KB	2026-01-20 12:58
📄 platform_collector.py # -*- coding: utf-8 from __future__ import unicode_literals import platform as pf from .metrics_core import GaugeMetricFamily from .registry import REGISTRY class PlatformCollector(object): """Collector for python platform information""" def __init__(self, registry=REGISTRY, platform=None): self._platform = pf if platform is None else platform info = self._info() system = self._platform.system() if system == "Java": info.update(self._java()) self._metrics = [ self._add_metric("python_info", "Python platform information", info) ] if registry: registry.register(self) def collect(self): return self._metrics @staticmethod def _add_metric(name, documentation, data): labels = data.keys() values = [data[k] for k in labels] g = GaugeMetricFamily(name, documentation, labels=labels) g.add_metric(values, 1) return g def _info(self): major, minor, patchlevel = self._platform.python_version_tuple() return { "version": self._platform.python_version(), "implementation": self._platform.python_implementation(), "major": major, "minor": minor, "patchlevel": patchlevel } def _java(self): java_version, _, vminfo, osinfo = self._platform.java_ver() vm_name, vm_release, vm_vendor = vminfo return { "jvm_version": java_version, "jvm_release": vm_release, "jvm_vendor": vm_vendor, "jvm_name": vm_name } PLATFORM_COLLECTOR = PlatformCollector() """PlatformCollector in default Registry REGISTRY"""	1.69 KB	2026-01-20 12:58
📄 process_collector.py from __future__ import unicode_literals import os from .metrics_core import CounterMetricFamily, GaugeMetricFamily from .registry import REGISTRY try: import resource _PAGESIZE = resource.getpagesize() except ImportError: # Not Unix _PAGESIZE = 4096 class ProcessCollector(object): """Collector for Standard Exports such as cpu and memory.""" def __init__(self, namespace='', pid=lambda: 'self', proc='/proc', registry=REGISTRY): self._namespace = namespace self._pid = pid self._proc = proc if namespace: self._prefix = namespace + '_process_' else: self._prefix = 'process_' self._ticks = 100.0 try: self._ticks = os.sysconf('SC_CLK_TCK') except (ValueError, TypeError, AttributeError, OSError): pass self._pagesize = _PAGESIZE # This is used to test if we can access /proc. self._btime = 0 try: self._btime = self._boot_time() except IOError: pass if registry: registry.register(self) def _boot_time(self): with open(os.path.join(self._proc, 'stat'), 'rb') as stat: for line in stat: if line.startswith(b'btime '): return float(line.split()[1]) def collect(self): if not self._btime: return [] pid = os.path.join(self._proc, str(self._pid()).strip()) result = [] try: with open(os.path.join(pid, 'stat'), 'rb') as stat: parts = (stat.read().split(b')')[-1].split()) vmem = GaugeMetricFamily(self._prefix + 'virtual_memory_bytes', 'Virtual memory size in bytes.', value=float(parts[20])) rss = GaugeMetricFamily(self._prefix + 'resident_memory_bytes', 'Resident memory size in bytes.', value=float(parts[21]) * self._pagesize) start_time_secs = float(parts[19]) / self._ticks start_time = GaugeMetricFamily(self._prefix + 'start_time_seconds', 'Start time of the process since unix epoch in seconds.', value=start_time_secs + self._btime) utime = float(parts[11]) / self._ticks stime = float(parts[12]) / self._ticks cpu = CounterMetricFamily(self._prefix + 'cpu_seconds_total', 'Total user and system CPU time spent in seconds.', value=utime + stime) result.extend([vmem, rss, start_time, cpu]) except IOError: pass try: with open(os.path.join(pid, 'limits'), 'rb') as limits: for line in limits: if line.startswith(b'Max open file'): max_fds = GaugeMetricFamily(self._prefix + 'max_fds', 'Maximum number of open file descriptors.', value=float(line.split()[3])) break open_fds = GaugeMetricFamily(self._prefix + 'open_fds', 'Number of open file descriptors.', len(os.listdir(os.path.join(pid, 'fd')))) result.extend([open_fds, max_fds]) except (IOError, OSError): pass return result PROCESS_COLLECTOR = ProcessCollector() """Default ProcessCollector in default Registry REGISTRY."""	3.57 KB	2026-01-20 12:58
📄 registry.py import copy from threading import Lock from .metrics_core import Metric class CollectorRegistry(object): """Metric collector registry. Collectors must have a no-argument method 'collect' that returns a list of Metric objects. The returned metrics should be consistent with the Prometheus exposition formats. """ def __init__(self, auto_describe=False, target_info=None): self._collector_to_names = {} self._names_to_collectors = {} self._auto_describe = auto_describe self._lock = Lock() self._target_info = {} self.set_target_info(target_info) def register(self, collector): """Add a collector to the registry.""" with self._lock: names = self._get_names(collector) duplicates = set(self._names_to_collectors).intersection(names) if duplicates: raise ValueError( 'Duplicated timeseries in CollectorRegistry: {0}'.format( duplicates)) for name in names: self._names_to_collectors[name] = collector self._collector_to_names[collector] = names def unregister(self, collector): """Remove a collector from the registry.""" with self._lock: for name in self._collector_to_names[collector]: del self._names_to_collectors[name] del self._collector_to_names[collector] def _get_names(self, collector): """Get names of timeseries the collector produces.""" desc_func = None # If there's a describe function, use it. try: desc_func = collector.describe except AttributeError: pass # Otherwise, if auto describe is enabled use the collect function. if not desc_func and self._auto_describe: desc_func = collector.collect if not desc_func: return [] result = [] type_suffixes = { 'counter': ['_total', '_created'], 'summary': ['', '_sum', '_count', '_created'], 'histogram': ['_bucket', '_sum', '_count', '_created'], 'gaugehistogram': ['_bucket', '_gsum', '_gcount'], 'info': ['_info'], } for metric in desc_func(): for suffix in type_suffixes.get(metric.type, ['']): result.append(metric.name + suffix) return result def collect(self): """Yields metrics from the collectors in the registry.""" collectors = None ti = None with self._lock: collectors = copy.copy(self._collector_to_names) if self._target_info: ti = self._target_info_metric() if ti: yield ti for collector in collectors: for metric in collector.collect(): yield metric def restricted_registry(self, names): """Returns object that only collects some metrics. Returns an object which upon collect() will return only samples with the given names. Intended usage is: generate_latest(REGISTRY.restricted_registry(['a_timeseries'])) Experimental.""" names = set(names) collectors = set() metrics = [] with self._lock: if 'target_info' in names and self._target_info: metrics.append(self._target_info_metric()) names.remove('target_info') for name in names: if name in self._names_to_collectors: collectors.add(self._names_to_collectors[name]) for collector in collectors: for metric in collector.collect(): samples = [s for s in metric.samples if s[0] in names] if samples: m = Metric(metric.name, metric.documentation, metric.type) m.samples = samples metrics.append(m) class RestrictedRegistry(object): def collect(self): return metrics return RestrictedRegistry() def set_target_info(self, labels): with self._lock: if labels: if not self._target_info and 'target_info' in self._names_to_collectors: raise ValueError('CollectorRegistry already contains a target_info metric') self._names_to_collectors['target_info'] = None elif self._target_info: self._names_to_collectors.pop('target_info', None) self._target_info = labels def get_target_info(self): with self._lock: return self._target_info def _target_info_metric(self): m = Metric('target', 'Target metadata', 'info') m.add_sample('target_info', self._target_info, 1) return m def get_sample_value(self, name, labels=None): """Returns the sample value, or None if not found. This is inefficient, and intended only for use in unittests. """ if labels is None: labels = {} for metric in self.collect(): for s in metric.samples: if s.name == name and s.labels == labels: return s.value return None REGISTRY = CollectorRegistry(auto_describe=True)	5.23 KB	2026-01-20 12:58
📄 samples.py from collections import namedtuple class Timestamp(object): """A nanosecond-resolution timestamp.""" def __init__(self, sec, nsec): if nsec < 0 or nsec >= 1e9: raise ValueError("Invalid value for nanoseconds in Timestamp: {0}".format(nsec)) if sec < 0: nsec = -nsec self.sec = int(sec) self.nsec = int(nsec) def __str__(self): return "{0}.{1:09d}".format(self.sec, self.nsec) def __repr__(self): return "Timestamp({0}, {1})".format(self.sec, self.nsec) def __float__(self): return float(self.sec) + float(self.nsec) / 1e9 def __eq__(self, other): return type(self) == type(other) and self.sec == other.sec and self.nsec == other.nsec def __ne__(self, other): return not self == other def __gt__(self, other): return self.sec > other.sec or self.nsec > other.nsec # Timestamp and exemplar are optional. # Value can be an int or a float. # Timestamp can be a float containing a unixtime in seconds, # a Timestamp object, or None. # Exemplar can be an Exemplar object, or None. Sample = namedtuple('Sample', ['name', 'labels', 'value', 'timestamp', 'exemplar']) Sample.__new__.__defaults__ = (None, None) Exemplar = namedtuple('Exemplar', ['labels', 'value', 'timestamp']) Exemplar.__new__.__defaults__ = (None,)	1.33 KB	2026-01-20 12:58
📁 twisted	-	2026-03-21 15:45
📄 utils.py import math INF = float("inf") MINUS_INF = float("-inf") NaN = float("NaN") def floatToGoString(d): d = float(d) if d == INF: return '+Inf' elif d == MINUS_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 = '{0}.{1}{2}'.format(s[0], s[1:dot], s[dot + 1:]).rstrip('0.') return '{0}e+0{1}'.format(mantissa, dot - 1) return s	621 B	2026-01-20 12:58
📄 values.py from __future__ import unicode_literals import os from threading import Lock from .mmap_dict import mmap_key, MmapedDict class MutexValue(object): """A float protected by a mutex.""" _multiprocess = False def __init__(self, typ, metric_name, name, labelnames, labelvalues, kwargs): self._value = 0.0 self._lock = Lock() def inc(self, amount): with self._lock: self._value += amount def set(self, value): with self._lock: self._value = value def get(self): with self._lock: return self._value def MultiProcessValue(process_identifier=os.getpid): """Returns a MmapedValue class based on a process_identifier function. The 'process_identifier' function MUST comply with this simple rule: when called in simultaneously running processes it MUST return distinct values. Using a different function than the default 'os.getpid' is at your own risk. """ files = {} values = [] pid = {'value': process_identifier()} # Use a single global lock when in multi-processing mode # as we presume this means there is no threading going on. # This avoids the need to also have mutexes in __MmapDict. lock = Lock() class MmapedValue(object): """A float protected by a mutex backed by a per-process mmaped file.""" _multiprocess = True def __init__(self, typ, metric_name, name, labelnames, labelvalues, multiprocess_mode='', kwargs): self._params = typ, metric_name, name, labelnames, labelvalues, multiprocess_mode with lock: self.__check_for_pid_change() self.__reset() values.append(self) def __reset(self): typ, metric_name, name, labelnames, labelvalues, multiprocess_mode = self._params if typ == 'gauge': file_prefix = typ + '_' + multiprocess_mode else: file_prefix = typ if file_prefix not in files: filename = os.path.join( os.environ['prometheus_multiproc_dir'], '{0}_{1}.db'.format(file_prefix, pid['value'])) files[file_prefix] = MmapedDict(filename) self._file = files[file_prefix] self._key = mmap_key(metric_name, name, labelnames, labelvalues) self._value = self._file.read_value(self._key) def __check_for_pid_change(self): actual_pid = process_identifier() if pid['value'] != actual_pid: pid['value'] = actual_pid # There has been a fork(), reset all the values. for f in files.values(): f.close() files.clear() for value in values: value.__reset() def inc(self, amount): with lock: self.__check_for_pid_change() self._value += amount self._file.write_value(self._key, self._value) def set(self, value): with lock: self.__check_for_pid_change() self._value = value self._file.write_value(self._key, self._value) def get(self): with lock: self.__check_for_pid_change() return self._value return MmapedValue def get_value_class(): # Should we enable multi-process mode? # This needs to be chosen before the first metric is constructed, # and as that may be in some arbitrary library the user/admin has # no control over we use an environment variable. if 'prometheus_multiproc_dir' in os.environ: return MultiProcessValue() else: return MutexValue ValueClass = get_value_class()	3.75 KB	2026-01-20 12:58

Edit File:

Terminal