Function generate_latest

prometheus_client/exposition.py:265–340

Generates the exposition format using the basic Prometheus text format. Params: registry: Collector to export data from. escaping: Escaping scheme used for metric and label names. Returns: UTF-8 encoded string containing the metrics in text format.

(registry: Collector = REGISTRY, escaping: str = openmetrics.UNDERSCORES)

def generate_latest(registry: Collector = REGISTRY, escaping: str = openmetrics.UNDERSCORES) -> bytes:
    """
    Generates the exposition format using the basic Prometheus text format.

    Params:
        registry: Collector to export data from.
        escaping: Escaping scheme used for metric and label names.

    Returns: UTF-8 encoded string containing the metrics in text format.
    """

    def sample_line(samples):
        if samples.labels:
            labelstr = '{0}'.format(','.join(
                # Label values always support UTF-8
                ['{}="{}"'.format(
                    openmetrics.escape_label_name(k, escaping), openmetrics._escape(v, openmetrics.ALLOWUTF8, False))
                    for k, v in sorted(samples.labels.items())]))
        else:
            labelstr = ''
        timestamp = ''
        if samples.timestamp is not None:
            # Convert to milliseconds.
            timestamp = f' {int(float(samples.timestamp) * 1000):d}'
        if escaping != openmetrics.ALLOWUTF8 or openmetrics._is_valid_legacy_metric_name(samples.name):
            if labelstr:
                labelstr = '{{{0}}}'.format(labelstr)
            return f'{openmetrics.escape_metric_name(samples.name, escaping)}{labelstr} {floatToGoString(samples.value)}{timestamp}\n'
        maybe_comma = ''
        if labelstr:
            maybe_comma = ','
        return f'{{{openmetrics.escape_metric_name(samples.name, escaping)}{maybe_comma}{labelstr}}} {floatToGoString(samples.value)}{timestamp}\n'

    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 {} {}\n'.format(
                openmetrics.escape_metric_name(mname, escaping), metric.documentation.replace('\\', r'\\').replace('\n', r'\n')))
            output.append(f'# TYPE {openmetrics.escape_metric_name(mname, escaping)} {mtype}\n')

            om_samples: Dict[str, List[str]] = {}