Repository: prometheus/client_python
Pull request #1102 (2025-04-08 21:29:40)

UTF-8 Content Negotiation
Part of https://github.com/prometheus/client_python/issues/1013

The remaining piece is configuration to force escaping even if a scraper does not request it

ywwg: > Though I agree with you, this will be a breaking change as other programs use CONTENT_TYPE_LATEST. Perhaps we should update this to version=1.0.0 instead?

we do still need a 0.0.4 constant as the fallback when all else fails.  What I could do is update LATEST to be 1.0.0 and then create the new PLAIN constant for 0.0.4

Patch:
diff --git a/prometheus_client/__init__.py b/prometheus_client/__init__.py
index 84a7ba8..221ad27 100644
--- a/prometheus_client/__init__.py
+++ b/prometheus_client/__init__.py
@@ -5,9 +5,10 @@ from . import (
     process_collector, registry,
 )
 from .exposition import (
-    CONTENT_TYPE_LATEST, delete_from_gateway, generate_latest,
-    instance_ip_grouping_key, make_asgi_app, make_wsgi_app, MetricsHandler,
-    push_to_gateway, pushadd_to_gateway, start_http_server, start_wsgi_server,
+    CONTENT_TYPE_LATEST, CONTENT_TYPE_PLAIN_0_0_4, CONTENT_TYPE_PLAIN_1_0_0,
+    delete_from_gateway, generate_latest, instance_ip_grouping_key,
+    make_asgi_app, make_wsgi_app, MetricsHandler, push_to_gateway,
+    pushadd_to_gateway, start_http_server, start_wsgi_server,
     write_to_textfile,
 )
 from .gc_collector import GC_COLLECTOR, GCCollector
@@ -33,6 +34,8 @@ __all__ = (
     'enable_created_metrics',
     'disable_created_metrics',
     'CONTENT_TYPE_LATEST',
+    'CONTENT_TYPE_PLAIN_0_0_4',
+    'CONTENT_TYPE_PLAIN_1_0_0',
     'generate_latest',
     'MetricsHandler',
     'make_wsgi_app',
diff --git a/prometheus_client/exposition.py b/prometheus_client/exposition.py
index 0bc3632..8c84ffb 100644
--- a/prometheus_client/exposition.py
+++ b/prometheus_client/exposition.py
@@ -1,5 +1,6 @@
 import base64
 from contextlib import closing
+from functools import partial
 import gzip
 from http.server import BaseHTTPRequestHandler
 import os
@@ -17,13 +18,16 @@ from urllib.request import (
 )
 from wsgiref.simple_server import make_server, WSGIRequestHandler, WSGIServer
 
+from packaging.version import Version
+
 from .openmetrics import exposition as openmetrics
 from .registry import CollectorRegistry, REGISTRY
 from .utils import floatToGoString
-from .validation import _is_valid_legacy_metric_name
 
 __all__ = (
     'CONTENT_TYPE_LATEST',
+    'CONTENT_TYPE_PLAIN_0_0_4',
+    'CONTENT_TYPE_PLAIN_1_0_0',
     'delete_from_gateway',
     'generate_latest',
     'instance_ip_grouping_key',
@@ -37,8 +41,13 @@ __all__ = (
     'write_to_textfile',
 )
 
-CONTENT_TYPE_LATEST = 'text/plain; version=0.0.4; charset=utf-8'
-"""Content type of the latest text format"""
+CONTENT_TYPE_PLAIN_0_0_4 = 'text/plain; version=0.0.4; charset=utf-8'
+"""Content type of the compatibility format"""
+
+CONTENT_TYPE_PLAIN_1_0_0 = 'text/plain; version=1.0.0; charset=utf-8'
+"""Content type of the latest format"""
+
+CONTENT_TYPE_LATEST = CONTENT_TYPE_PLAIN_1_0_0
 
 
 class _PrometheusRedirectHandler(HTTPRedirectHandler):
@@ -245,14 +254,23 @@ def start_wsgi_server(
 start_http_server = start_wsgi_server
 
 
-def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes:
-    """Returns the metrics from the registry in latest text format as a string."""
+def generate_latest(registry: CollectorRegistry = REGISTRY, escaping: str = openmetrics.UNDERSCORES) -> bytes:
+    """
+    Generates the exposition format using the basic Prometheus text format.
+
+    Params:
+        registry: CollectorRegistry 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), openmetrics._escape(v))
+                    openmetrics.escape_label_name(k, escaping), openmetrics._escape(v, openmetrics.ALLOWUTF8, False))
                     for k, v in sorted(samples.labels.items())]))
         else:
             labelstr = ''
@@ -260,14 +278,14 @@ def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes:
         if samples.timestamp is not None:
             # Convert to milliseconds.
             timestamp = f' {int(float(samples.timestamp) * 1000):d}'
-        if _is_valid_legacy_metric_name(samples.name):
+        if escaping != openmetrics.ALLOWUTF8 or openmetrics._is_valid_legacy_metric_name(samples.name):
             if labelstr:
                 labelstr = '{{{0}}}'.format(labelstr)
-            return f'{samples.name}{labelstr} {floatToGoString(samples.value)}{timestamp}\n'
+            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)}{maybe_comma}{labelstr}}} {floatToGoString(samples.value)}{timestamp}\n'
+        return f'{{{openmetrics.escape_metric_name(samples.name, escaping)}{maybe_comma}{labelstr}}} {floatToGoString(samples.value)}{timestamp}\n'
 
     output = []
     for metric in registry.collect():
@@ -290,8 +308,8 @@ def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes:
                 mtype = 'untyped'
 
             output.append('# HELP {} {}\n'.format(
-                openmetrics.escape_metric_name(mname), metric.documentation.replace('\\', r'\\').replace('\n', r'\n')))
-            output.append(f'# TYPE {openmetrics.escape_metric_name(mname)} {mtype}\n')
+                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]] = {}
             for s in metric.samples:
@@ -307,20 +325,79 @@ def generate_latest(registry: CollectorRegistry = REGISTRY) -> bytes:
             raise
 
         for suffix, lines in sorted(om_samples.items()):
-            output.append('# HELP {} {}\n'.format(openmetrics.escape_metric_name(metric.name + suffix),
+            output.append('# HELP {} {}\n'.format(openmetrics.escape_metric_name(metric.name + suffix, escaping),
                                                   metric.documentation.replace('\\', r'\\').replace('\n', r'\n')))
-            output.append(f'# TYPE {openmetrics.escape_metric_name(metric.name + suffix)} gauge\n')
+            output.append(f'# TYPE {openmetrics.escape_metric_name(metric.name + suffix, escaping)} gauge\n')
             output.extend(lines)
     return ''.join(output).encode('utf-8')
 
 
 def choose_encoder(accept_header: str) -> Tuple[Callable[[CollectorRegistry], bytes], str]:
+    # Python client library accepts a narrower range of content-types than
+    # Prometheus does.
     accept_header = accept_header or ''
+    escaping = openmetrics.UNDERSCORES
     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
+            toks = accepted.split(';')
+            version = _get_version(toks)
+            escaping = _get_escaping(toks)
+            # Only return an escaping header if we have a good version and
+            # mimetype.
+            if not version:
+                return (partial(openmetrics.generate_latest, escaping=openmetrics.UNDERSCORES), openmetrics.CONTENT_TYPE_LATEST)
+            if version and Version(version) >= Version('1.0.0'):
+                return (partial(openmetrics.generate_latest, escaping=escaping),
+                        openmetrics.CONTENT_TYPE_LATEST + '; escaping=' + str(escaping))
+        elif accepted.split(';')[0].strip() == 'text/plain':
+            toks = accepted.split(';')
+            version = _get_version(toks)
+            escaping = _get_escaping(toks)
+            # Only return an escaping header if we have a good version and
+            # mimetype.
+            if version and Version(version) >= Version('1.0.0'):
+                return (partial(generate_latest, escaping=escaping),
+                        CONTENT_TYPE_LATEST + '; escaping=' + str(escaping))
+    return generate_latest, CONTENT_TYPE_PLAIN_0_0_4
+
+
+def _get_version(accept_header: List[str]) -> str:
+    """Return the version tag from the Accept header.
+
+    If no version is specified, returns empty string."""
+
+    for tok in accept_header:
+        if '=' not in tok:
+            continue
+        key, value = tok.strip().split('=', 1)
+        if key == 'version':
+            return value
+    return ""
+
+
+def _get_escaping(accept_header: List[str]) -> str:
+    """Return the escaping scheme from the Accept header.
+
+    If no escaping scheme is specified or the scheme is not one of the allowed
+    strings, defaults to UNDERSCORES."""
+
+    for tok in accept_header:
+        if '=' not in tok:
+            continue
+        key, value = tok.strip().split('=', 1)
+        if key != 'escaping':
+            continue
+        if value == openmetrics.ALLOWUTF8:
+            return openmetrics.ALLOWUTF8
+        elif value == openmetrics.UNDERSCORES:
+            return openmetrics.UNDERSCORES
+        elif value == openmetrics.DOTS:
+            return openmetrics.DOTS
+        elif value == openmetrics.VALUES:
+            return openmetrics.VALUES
+        else:
+            return openmetrics.UNDERSCORES
+    return openmetrics.UNDERSCORES
 
 
 def gzip_accepted(accept_encoding_header: str) -> bool:
@@ -369,7 +446,7 @@ class MetricsHandler(BaseHTTPRequestHandler):
         return MyMetricsHandler
 
 
-def write_to_textfile(path: str, registry: CollectorRegistry) -> None:
+def write_to_textfile(path: str, registry: CollectorRegistry, escaping: str = openmetrics.ALLOWUTF8) -> None:
     """Write metrics to the given path.
 
     This is intended for use with the Node exporter textfile collector.
@@ -377,7 +454,7 @@ def write_to_textfile(path: str, registry: CollectorRegistry) -> None:
     tmppath = f'{path}.{os.getpid()}.{threading.current_thread().ident}'
     try:
         with open(tmppath, 'wb') as f:
-            f.write(generate_latest(registry))
+            f.write(generate_latest(registry, escaping))
 
         # rename(2) is atomic but fails on Windows if the destination file exists
         if os.name == 'nt':
@@ -645,7 +722,7 @@ def _use_gateway(
 
     handler(
         url=url, method=method, timeout=timeout,
-        headers=[('Content-Type', CONTENT_TYPE_LATEST)], data=data,
+        headers=[('Content-Type', CONTENT_TYPE_PLAIN_0_0_4)], data=data,
     )()
 
 
diff --git a/prometheus_client/openmetrics/exposition.py b/prometheus_client/openmetrics/exposition.py
index 8460060..a89acda 100644
--- a/prometheus_client/openmetrics/exposition.py
+++ b/prometheus_client/openmetrics/exposition.py
@@ -1,5 +1,8 @@
 #!/usr/bin/env python
 
+from io import StringIO
+from sys import maxunicode
+from typing import Callable
 
 from ..utils import floatToGoString
 from ..validation import (
@@ -8,6 +11,13 @@ from ..validation import (
 
 CONTENT_TYPE_LATEST = 'application/openmetrics-text; version=1.0.0; charset=utf-8'
 """Content type of the latest OpenMetrics text format"""
+ESCAPING_HEADER_TAG = 'escaping'
+
+
+ALLOWUTF8 = 'allow-utf-8'
+UNDERSCORES = 'underscores'
+DOTS = 'dots'
+VALUES = 'values'
 
 
 def _is_valid_exemplar_metric(metric, sample):
@@ -20,34 +30,35 @@ def _is_valid_exemplar_metric(metric, sample):
     return False
 
 
-def generate_latest(registry):
+def generate_latest(registry, escaping=UNDERSCORES):
     '''Returns the metrics from the registry in latest text format as a string.'''
     output = []
     for metric in registry.collect():
         try:
             mname = metric.name
             output.append('# HELP {} {}\n'.format(
-                escape_metric_name(mname), _escape(metric.documentation)))
-            output.append(f'# TYPE {escape_metric_name(mname)} {metric.type}\n')
+                escape_metric_name(mname, escaping), _escape(metric.documentation, ALLOWUTF8, _is_legacy_labelname_rune)))
+            output.append(f'# TYPE {escape_metric_name(mname, escaping)} {metric.type}\n')
             if metric.unit:
-                output.append(f'# UNIT {escape_metric_name(mname)} {metric.unit}\n')
+                output.append(f'# UNIT {escape_metric_name(mname, escaping)} {metric.unit}\n')
             for s in metric.samples:
-                if not _is_valid_legacy_metric_name(s.name):
-                    labelstr = escape_metric_name(s.name)
+                if escaping == ALLOWUTF8 and not _is_valid_legacy_metric_name(s.name):
+                    labelstr = escape_metric_name(s.name, escaping)
                     if s.labels:
                         labelstr += ', '
                 else:
                     labelstr = ''
-                
+
                 if s.labels:
                     items = sorted(s.labels.items())
+                    # Label values always support UTF-8
                     labelstr += ','.join(
                         ['{}="{}"'.format(
-                            escape_label_name(k), _escape(v))
+                            escape_label_name(k, escaping), _escape(v, ALLOWUTF8, _is_legacy_labelname_rune))
                             for k, v in items])
                 if labelstr:
                     labelstr = "{" + labelstr + "}"
-                    
+
                 if s.exemplar:
                     if not _is_valid_exemplar_metric(metric, s):
                         raise ValueError(f"Metric {metric.name} has exemplars, but is not a histogram bucket or counter")
@@ -71,9 +82,9 @@ def generate_latest(registry):
                 timestamp = ''
                 if s.timestamp is not None:
                     timestamp = f' {s.timestamp}'
-                if _is_valid_legacy_metric_name(s.name):
+                if (escaping != ALLOWUTF8) or _is_valid_legacy_metric_name(s.name):
                     output.append('{}{} {}{}{}\n'.format(
-                        s.name,
+                        _escape(s.name, escaping, _is_legacy_labelname_rune),
                         labelstr,
                         floatToGoString(s.value),
                         timestamp,
@@ -94,24 +105,118 @@ def generate_latest(registry):
     return ''.join(output).encode('utf-8')
 
 
-def escape_metric_name(s: str) -> str:
+def escape_metric_name(s: str, escaping: str = UNDERSCORES) -> str:
     """Escapes the metric name and puts it in quotes iff the name does not
     conform to the legacy Prometheus character set.
     """
-    if _is_valid_legacy_metric_name(s):
+    if len(s) == 0:
         return s
-    return '"{}"'.format(_escape(s))
+    if escaping == ALLOWUTF8:
+        if not _is_valid_legacy_metric_name(s):
+            return '"{}"'.format(_escape(s, escaping, _is_legacy_metric_rune))
+        return _escape(s, escaping, _is_legacy_metric_rune)
+    elif escaping == UNDERSCORES:
+        if _is_valid_legacy_metric_name(s):
+            return s
+        return _escape(s, escaping, _is_legacy_metric_rune)
+    elif escaping == DOTS:
+        return _escape(s, escaping, _is_legacy_metric_rune)
+    elif escaping == VALUES:
+        if _is_valid_legacy_metric_name(s):
+            return s
+        return _escape(s, escaping, _is_legacy_metric_rune)
+    return s
 
 
-def escape_label_name(s: str) -> str:
+def escape_label_name(s: str, escaping: str = UNDERSCORES) -> str:
     """Escapes the label name and puts it in quotes iff the name does not
     conform to the legacy Prometheus character set.
     """
-    if _is_valid_legacy_labelname(s):
+    if len(s) == 0:
         return s
-    return '"{}"'.format(_escape(s))
+    if escaping == ALLOWUTF8:
+        if not _is_valid_legacy_labelname(s):
+            return '"{}"'.format(_escape(s, escaping, _is_legacy_labelname_rune))
+        return _escape(s, escaping, _is_legacy_labelname_rune)
+    elif escaping == UNDERSCORES:
+        if _is_valid_legacy_labelname(s):
+            return s
+        return _escape(s, escaping, _is_legacy_labelname_rune)
+    elif escaping == DOTS:
+        return _escape(s, escaping, _is_legacy_labelname_rune)
+    elif escaping == VALUES:
+        if _is_valid_legacy_labelname(s):
+            return s
+        return _escape(s, escaping, _is_legacy_labelname_rune)
+    return s
+
+
+def _escape(s: str, escaping: str, valid_rune_fn: Callable[[str, int], bool]) -> str:
+    """Performs backslash escaping on backslash, newline, and double-quote characters.
+
+    valid_rune_fn takes the input character and its index in the containing string."""
+    if escaping == ALLOWUTF8:
+        return s.replace('\\', r'\\').replace('\n', r'\n').replace('"', r'\"')
+    elif escaping == UNDERSCORES:
+        escaped = StringIO()
+        for i, b in enumerate(s):
+            if valid_rune_fn(b, i):
+                escaped.write(b)
+            else:
+                escaped.write('_')
+        return escaped.getvalue()
+    elif escaping == DOTS:
+        escaped = StringIO()
+        for i, b in enumerate(s):
+            if b == '_':
+                escaped.write('__')
+            elif b == '.':
+                escaped.write('_dot_')
+            elif valid_rune_fn(b, i):
+                escaped.write(b)
+            else:
+                escaped.write('__')
+        return escaped.getvalue()
+    elif escaping == VALUES:
+        escaped = StringIO()
+        escaped.write("U__")
+        for i, b in enumerate(s):
+            if b == '_':
+                escaped.write("__")
+            elif valid_rune_fn(b, i):
+                escaped.write(b)
+            elif not _is_valid_utf8(b):
+                escaped.write("_FFFD_")
+            else:
+                escaped.write('_')
+                escaped.write(format(ord(b), 'x'))
+                escaped.write('_')
+        return escaped.getvalue()
+    return s
+
 
+def _is_legacy_metric_rune(b: str, i: int) -> bool:
+    return _is_legacy_labelname_rune(b, i) or b == ':'
 
-def _escape(s: str) -> str:
-    """Performs backslash escaping on backslash, newline, and double-quote characters."""
-    return s.replace('\\', r'\\').replace('\n', r'\n').replace('"', r'\"')
+
+def _is_legacy_labelname_rune(b: str, i: int) -> bool:
+    if len(b) != 1:
+        raise ValueError("Input 'b' must be a single character.")
+    return (
+        ('a' <= b <= 'z')
+        or ('A' <= b <= 'Z')
+        or (b == '_')
+        or ('0' <= b <= '9' and i > 0)
+    )
+
+
+_SURROGATE_MIN = 0xD800
+_SURROGATE_MAX = 0xDFFF
+
+
+def _is_valid_utf8(s: str) -> bool:
+    if 0 <= ord(s) < _SURROGATE_MIN:
+        return True
+    if _SURROGATE_MAX < ord(s) <= maxunicode:
+        return True
+    return False
diff --git a/prometheus_client/registry.py b/prometheus_client/registry.py
index 694e4bd..8de4ce9 100644
--- a/prometheus_client/registry.py
+++ b/prometheus_client/registry.py
@@ -103,7 +103,7 @@ class CollectorRegistry(Collector):
         only samples with the given names.
 
         Intended usage is:
-            generate_latest(REGISTRY.restricted_registry(['a_timeseries']))
+            generate_latest(REGISTRY.restricted_registry(['a_timeseries']), escaping)
 
         Experimental."""
         names = set(names)
diff --git a/prometheus_client/validation.py b/prometheus_client/validation.py
index bf19fc7..7ada5d8 100644
--- a/prometheus_client/validation.py
+++ b/prometheus_client/validation.py
@@ -51,6 +51,8 @@ def _validate_metric_name(name: str) -> None:
 
 def _is_valid_legacy_metric_name(name: str) -> bool:
     """Returns true if the provided metric name conforms to the legacy validation scheme."""
+    if len(name) == 0:
+        return False
     return METRIC_NAME_RE.match(name) is not None
 
 
@@ -94,6 +96,8 @@ def _validate_labelname(l):
 
 def _is_valid_legacy_labelname(l: str) -> bool:
     """Returns true if the provided label name conforms to the legacy validation scheme."""
+    if len(l) == 0:
+        return False
     if METRIC_LABEL_NAME_RE.match(l) is None:
         return False
     return RESERVED_METRIC_LABEL_NAME_RE.match(l) is None
diff --git a/tools/simple_client.py b/tools/simple_client.py
new file mode 100755
index 0000000..0ccefb7
--- /dev/null
+++ b/tools/simple_client.py
@@ -0,0 +1,28 @@
+# A simple client that serves random gauges. 
+# usage: uvicorn tools.simple_client:app --reload
+
+from fastapi import FastAPI
+from fastapi.responses import RedirectResponse
+from prometheus_client.asgi import make_asgi_app
+from prometheus_client.core import GaugeMetricFamily, REGISTRY
+import random
+
+
+class CustomCollector:
+    def collect(self):
+        g = GaugeMetricFamily('my.random.utf8.metric', 'Random value', labels=['label.1'])
+        g.add_metric(['value.1'], random.random())
+        g.add_metric(['value.2'], random.random())
+        yield g
+
+
+app = FastAPI()
+
+
+@app.get("/")
+async def root():
+    return RedirectResponse(url="/metrics")
+
+
+REGISTRY.register(CustomCollector())
+app.mount("/metrics", make_asgi_app(REGISTRY))


Test patch:
diff --git a/tests/openmetrics/test_exposition.py b/tests/openmetrics/test_exposition.py
index 124e55e..9f79064 100644
--- a/tests/openmetrics/test_exposition.py
+++ b/tests/openmetrics/test_exposition.py
@@ -1,13 +1,18 @@
 import time
 import unittest
 
+import pytest
+
 from prometheus_client import (
     CollectorRegistry, Counter, Enum, Gauge, Histogram, Info, Metric, Summary,
 )
 from prometheus_client.core import (
     Exemplar, GaugeHistogramMetricFamily, Timestamp,
 )
-from prometheus_client.openmetrics.exposition import generate_latest
+from prometheus_client.openmetrics.exposition import (
+    ALLOWUTF8, DOTS, escape_label_name, escape_metric_name, generate_latest,
+    UNDERSCORES, VALUES,
+)
 
 
 class TestGenerateText(unittest.TestCase):
@@ -33,12 +38,22 @@ class TestGenerateText(unittest.TestCase):
         c.inc()
         self.assertEqual(b'# HELP cc A counter\n# TYPE cc counter\ncc_total 1.0\ncc_created 123.456\n# EOF\n',
                          generate_latest(self.registry))
-        
+
     def test_counter_utf8(self):
         c = Counter('cc.with.dots', 'A counter', registry=self.registry)
         c.inc()
         self.assertEqual(b'# HELP "cc.with.dots" A counter\n# TYPE "cc.with.dots" counter\n{"cc.with.dots_total"} 1.0\n{"cc.with.dots_created"} 123.456\n# EOF\n',
-                         generate_latest(self.registry))
+                         generate_latest(self.registry, ALLOWUTF8))
+
+    def test_counter_utf8_escaped_underscores(self):
+        c = Counter('utf8.cc', 'A counter', registry=self.registry)
+        c.inc()
+        assert b"""# HELP utf8_cc A counter
+# TYPE utf8_cc counter
+utf8_cc_total 1.0
+utf8_cc_created 123.456
+# EOF
+""" == generate_latest(self.registry, UNDERSCORES)
 
     def test_counter_total(self):
         c = Counter('cc_total', 'A counter', registry=self.registry)
@@ -282,5 +297,147 @@ ts{foo="f"} 0.0 123.000000456
 """, generate_latest(self.registry))
 
 
+@pytest.mark.parametrize("scenario", [
+    {
+        "name": "empty string",
+        "input": "",
+        "expectedUnderscores": "",
+        "expectedDots": "",
+        "expectedValue": "",
+    },
+    {
+        "name": "legacy valid metric name",
+        "input": "no:escaping_required",
+        "expectedUnderscores": "no:escaping_required",
+        "expectedDots": "no:escaping__required",
+        "expectedValue": "no:escaping_required",
+    },
+    {
+        "name": "metric name with dots",
+        "input": "mysystem.prod.west.cpu.load",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load",
+    },
+    {
+        "name": "metric name with dots and underscore",
+        "input": "mysystem.prod.west.cpu.load_total",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load_total",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load__total",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load__total",
+    },
+    {
+        "name": "metric name with dots and colon",
+        "input": "http.status:sum",
+        "expectedUnderscores": "http_status:sum",
+        "expectedDots": "http_dot_status:sum",
+        "expectedValue": "U__http_2e_status:sum",
+    },
+    {
+        "name": "metric name with spaces and emoji",
+        "input": "label with 😱",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__1f631_",
+    },
+    {
+        "name": "metric name with unicode characters > 0x100",
+        "input": "花火",
+        "expectedUnderscores": "__",
+        "expectedDots": "____",
+        "expectedValue": "U___82b1__706b_",
+    },
+    {
+        "name": "metric name with spaces and edge-case value",
+        "input": "label with \u0100",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__100_",
+    },
+])
+def test_escape_metric_name(scenario):
+    input = scenario["input"]
+
+    got = escape_metric_name(input, UNDERSCORES)
+    assert got == scenario["expectedUnderscores"], f"[{scenario['name']}] Underscore escaping failed"
+
+    got = escape_metric_name(input, DOTS)
+    assert got == scenario["expectedDots"], f"[{scenario['name']}] Dots escaping failed"
+
+    got = escape_metric_name(input, VALUES)
+    assert got == scenario["expectedValue"], f"[{scenario['name']}] Value encoding failed"
+
+
+@pytest.mark.parametrize("scenario", [
+    {
+        "name": "empty string",
+        "input": "",
+        "expectedUnderscores": "",
+        "expectedDots": "",
+        "expectedValue": "",
+    },
+    {
+        "name": "legacy valid label name",
+        "input": "no_escaping_required",
+        "expectedUnderscores": "no_escaping_required",
+        "expectedDots": "no__escaping__required",
+        "expectedValue": "no_escaping_required",
+    },
+    {
+        "name": "label name with dots",
+        "input": "mysystem.prod.west.cpu.load",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load",
+    },
+    {
+        "name": "label name with dots and underscore",
+        "input": "mysystem.prod.west.cpu.load_total",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load_total",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load__total",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load__total",
+    },
+    {
+        "name": "label name with dots and colon",
+        "input": "http.status:sum",
+        "expectedUnderscores": "http_status_sum",
+        "expectedDots": "http_dot_status__sum",
+        "expectedValue": "U__http_2e_status_3a_sum",
+    },
+    {
+        "name": "label name with spaces and emoji",
+        "input": "label with 😱",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__1f631_",
+    },
+    {
+        "name": "label name with unicode characters > 0x100",
+        "input": "花火",
+        "expectedUnderscores": "__",
+        "expectedDots": "____",
+        "expectedValue": "U___82b1__706b_",
+    },
+    {
+        "name": "label name with spaces and edge-case value",
+        "input": "label with \u0100",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__100_",
+    },
+])
+def test_escape_label_name(scenario):
+    input = scenario["input"]
+
+    got = escape_label_name(input, UNDERSCORES)
+    assert got == scenario["expectedUnderscores"], f"[{scenario['name']}] Underscore escaping failed"
+
+    got = escape_label_name(input, DOTS)
+    assert got == scenario["expectedDots"], f"[{scenario['name']}] Dots escaping failed"
+
+    got = escape_label_name(input, VALUES)
+    assert got == scenario["expectedValue"], f"[{scenario['name']}] Value encoding failed"
+
+
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/test_asgi.py b/tests/test_asgi.py
index 78e2419..eaa195d 100644
--- a/tests/test_asgi.py
+++ b/tests/test_asgi.py
@@ -2,7 +2,7 @@ import gzip
 from unittest import skipUnless, TestCase
 
 from prometheus_client import CollectorRegistry, Counter
-from prometheus_client.exposition import CONTENT_TYPE_LATEST
+from prometheus_client.exposition import CONTENT_TYPE_PLAIN_0_0_4
 
 try:
     # Python >3.5 only
@@ -104,7 +104,7 @@ class ASGITest(TestCase):
         # Headers
         num_of_headers = 2 if compressed else 1
         self.assertEqual(len(response_start['headers']), num_of_headers)
-        self.assertIn((b"Content-Type", CONTENT_TYPE_LATEST.encode('utf8')), response_start['headers'])
+        self.assertIn((b"Content-Type", CONTENT_TYPE_PLAIN_0_0_4.encode('utf8')), response_start['headers'])
         if compressed:
             self.assertIn((b"Content-Encoding", b"gzip"), response_start['headers'])
         # Body
@@ -176,7 +176,7 @@ class ASGITest(TestCase):
         """Response content type is application/openmetrics-text when appropriate Accept header is in request"""
         app = make_asgi_app(self.registry)
         self.seed_app(app)
-        self.scope["headers"] = [(b"Accept", b"application/openmetrics-text")]
+        self.scope["headers"] = [(b"Accept", b"application/openmetrics-text; version=1.0.0")]
         self.send_input({"type": "http.request", "body": b""})
 
         content_type = self.get_response_header_value('Content-Type').split(";")[0]
diff --git a/tests/test_exposition.py b/tests/test_exposition.py
index 2a3f08c..3dd5e37 100644
--- a/tests/test_exposition.py
+++ b/tests/test_exposition.py
@@ -7,9 +7,10 @@ import unittest
 import pytest
 
 from prometheus_client import (
-    CollectorRegistry, CONTENT_TYPE_LATEST, core, Counter, delete_from_gateway,
-    Enum, Gauge, generate_latest, Histogram, Info, instance_ip_grouping_key,
-    Metric, push_to_gateway, pushadd_to_gateway, Summary,
+    CollectorRegistry, CONTENT_TYPE_LATEST, CONTENT_TYPE_PLAIN_0_0_4,
+    CONTENT_TYPE_PLAIN_1_0_0, core, Counter, delete_from_gateway, Enum, Gauge,
+    generate_latest, Histogram, Info, instance_ip_grouping_key, Metric,
+    push_to_gateway, pushadd_to_gateway, Summary,
 )
 from prometheus_client.core import GaugeHistogramMetricFamily, Timestamp
 from prometheus_client.exposition import (
@@ -46,8 +47,8 @@ cc_total 1.0
 # HELP cc_created A counter
 # TYPE cc_created gauge
 cc_created 123.456
-""", generate_latest(self.registry))
-        
+""", generate_latest(self.registry, openmetrics.ALLOWUTF8))
+
     def test_counter_utf8(self):
         c = Counter('utf8.cc', 'A counter', registry=self.registry)
         c.inc()
@@ -57,7 +58,18 @@ cc_created 123.456
 # HELP "utf8.cc_created" A counter
 # TYPE "utf8.cc_created" gauge
 {"utf8.cc_created"} 123.456
-""", generate_latest(self.registry))
+""", generate_latest(self.registry, openmetrics.ALLOWUTF8))
+
+    def test_counter_utf8_escaped_underscores(self):
+        c = Counter('utf8.cc', 'A counter', registry=self.registry)
+        c.inc()
+        assert b"""# HELP utf8_cc_total A counter
+# TYPE utf8_cc_total counter
+utf8_cc_total 1.0
+# HELP utf8_cc_created A counter
+# TYPE utf8_cc_created gauge
+utf8_cc_created 123.456
+""" == generate_latest(self.registry, openmetrics.UNDERSCORES)
 
     def test_counter_name_unit_append(self):
         c = Counter('requests', 'Request counter', unit="total", registry=self.registry)
@@ -264,70 +276,70 @@ class TestPushGateway(unittest.TestCase):
         push_to_gateway(self.address, "my_job", self.registry)
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_push_schemeless_url(self):
         push_to_gateway(self.address.replace('http://', ''), "my_job", self.registry)
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_push_with_groupingkey(self):
         push_to_gateway(self.address, "my_job", self.registry, {'a': 9})
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job/a/9')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_push_with_groupingkey_empty_label(self):
         push_to_gateway(self.address, "my_job", self.registry, {'a': ''})
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job/a@base64/=')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_push_with_complex_groupingkey(self):
         push_to_gateway(self.address, "my_job", self.registry, {'a': 9, 'b': 'a/ z'})
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job/a/9/b@base64/YS8geg==')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_push_with_complex_job(self):
         push_to_gateway(self.address, "my/job", self.registry)
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job@base64/bXkvam9i')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_pushadd(self):
         pushadd_to_gateway(self.address, "my_job", self.registry)
         self.assertEqual(self.requests[0][0].command, 'POST')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_pushadd_with_groupingkey(self):
         pushadd_to_gateway(self.address, "my_job", self.registry, {'a': 9})
         self.assertEqual(self.requests[0][0].command, 'POST')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job/a/9')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_delete(self):
         delete_from_gateway(self.address, "my_job")
         self.assertEqual(self.requests[0][0].command, 'DELETE')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'')
 
     def test_delete_with_groupingkey(self):
         delete_from_gateway(self.address, "my_job", {'a': 9})
         self.assertEqual(self.requests[0][0].command, 'DELETE')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job/a/9')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'')
 
     def test_push_with_handler(self):
@@ -340,7 +352,7 @@ class TestPushGateway(unittest.TestCase):
         push_to_gateway(self.address, "my_job", self.registry, handler=my_test_handler)
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][0].headers.get('x-test-header'), 'foobar')
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
@@ -351,7 +363,7 @@ class TestPushGateway(unittest.TestCase):
         push_to_gateway(self.address, "my_job_with_basic_auth", self.registry, handler=my_auth_handler)
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job_with_basic_auth')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_push_with_tls_auth_handler(self):
@@ -362,7 +374,7 @@ class TestPushGateway(unittest.TestCase):
         push_to_gateway(self.address, "my_job_with_tls_auth", self.registry, handler=my_auth_handler)
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job_with_tls_auth')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
     def test_push_with_redirect_handler(self):
@@ -372,7 +384,7 @@ class TestPushGateway(unittest.TestCase):
         push_to_gateway(self.address, "my_job_with_redirect", self.registry, handler=my_redirect_handler)
         self.assertEqual(self.requests[0][0].command, 'PUT')
         self.assertEqual(self.requests[0][0].path, '/metrics/job/my_job_with_redirect')
-        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_LATEST)
+        self.assertEqual(self.requests[0][0].headers.get('content-type'), CONTENT_TYPE_PLAIN_0_0_4)
         self.assertEqual(self.requests[0][1], b'# HELP g help\n# TYPE g gauge\ng 0.0\n')
 
         # ensure the redirect preserved request settings from the initial request.
@@ -423,7 +435,7 @@ class Collector:
 
 def _expect_metric_exception(registry, expected_error):
     try:
-        generate_latest(registry)
+        generate_latest(registry, openmetrics.ALLOWUTF8)
     except expected_error as exception:
         assert isinstance(exception.args[-1], core.Metric)
         # Got a valid error as expected, return quietly
@@ -484,10 +496,251 @@ def test_histogram_metric_families(MetricFamily, registry, buckets, sum_value, e
     _expect_metric_exception(registry, error)
 
 
-def test_choose_encoder():
-    assert choose_encoder(None) == (generate_latest, CONTENT_TYPE_LATEST)
-    assert choose_encoder(CONTENT_TYPE_LATEST) == (generate_latest, CONTENT_TYPE_LATEST)
-    assert choose_encoder(openmetrics.CONTENT_TYPE_LATEST) == (openmetrics.generate_latest, openmetrics.CONTENT_TYPE_LATEST)
+class TestChooseEncoder(unittest.TestCase):
+    def setUp(self):
+        self.registry = CollectorRegistry()
+        c = Counter('dotted.counter', 'A counter', registry=self.registry)
+        c.inc()
+
+    def custom_collector(self, metric_family):
+        class CustomCollector:
+            def collect(self):
+                return [metric_family]
+
+        self.registry.register(CustomCollector())
+
+    def assert_is_escaped(self, exp):
+        self.assertRegex(exp, r'.*\ndotted_counter_total 1.0\n.*')
+
+    def assert_is_utf8(self, exp):
+        self.assertRegex(exp, r'.*\n{"dotted.counter_total"} 1.0\n.*')
+
+    def assert_is_prom(self, exp):
+        self.assertNotRegex(exp, r'# EOF')
+
+    def assert_is_openmetrics(self, exp):
+        self.assertRegex(exp, r'# EOF')
+
+    def test_default_encoder(self):
+        generator, content_type = choose_encoder(None)
+        assert content_type == CONTENT_TYPE_PLAIN_0_0_4
+        exp = generator(self.registry).decode('utf-8')
+        self.assert_is_escaped(exp)
+        self.assert_is_prom(exp)
+
+    def test_plain_encoder(self):
+        generator, content_type = choose_encoder(CONTENT_TYPE_PLAIN_0_0_4)
+        assert content_type == CONTENT_TYPE_PLAIN_0_0_4
+        exp = generator(self.registry).decode('utf-8')
+        self.assert_is_escaped(exp)
+        self.assert_is_prom(exp)
+
+    def test_openmetrics_latest(self):
+        generator, content_type = choose_encoder(openmetrics.CONTENT_TYPE_LATEST)
+        assert content_type == 'application/openmetrics-text; version=1.0.0; charset=utf-8; escaping=underscores'
+        exp = generator(self.registry).decode('utf-8')
+        self.assert_is_escaped(exp)
+        self.assert_is_openmetrics(exp)
+
+    def test_openmetrics_utf8(self):
+        generator, content_type = choose_encoder(openmetrics.CONTENT_TYPE_LATEST + '; escaping=allow-utf-8')
+        assert content_type == openmetrics.CONTENT_TYPE_LATEST + '; escaping=allow-utf-8'
+        exp = generator(self.registry).decode('utf-8')
+        self.assert_is_utf8(exp)
+        self.assert_is_openmetrics(exp)
+
+    def test_openmetrics_dots_escaping(self):
+        generator, content_type = choose_encoder(openmetrics.CONTENT_TYPE_LATEST + '; escaping=dots')
+        assert content_type == openmetrics.CONTENT_TYPE_LATEST + '; escaping=dots'
+        exp = generator(self.registry).decode('utf-8')
+        self.assertRegex(exp, r'.*\ndotted_dot_counter__total 1.0\n.*')
+        self.assert_is_openmetrics(exp)
+
+    def test_prom_latest(self):
+        generator, content_type = choose_encoder(CONTENT_TYPE_LATEST)
+        assert content_type == CONTENT_TYPE_PLAIN_1_0_0 + '; escaping=underscores'
+        exp = generator(self.registry).decode('utf-8')
+        self.assert_is_escaped(exp)
+        self.assert_is_prom(exp)
+
+    def test_prom_plain_1_0_0(self):
+        generator, content_type = choose_encoder(CONTENT_TYPE_PLAIN_1_0_0)
+        assert content_type == CONTENT_TYPE_PLAIN_1_0_0 + '; escaping=underscores'
+        exp = generator(self.registry).decode('utf-8')
+        self.assert_is_escaped(exp)
+        self.assert_is_prom(exp)
+
+    def test_prom_utf8(self):
+        generator, content_type = choose_encoder(CONTENT_TYPE_PLAIN_1_0_0 + '; escaping=allow-utf-8')
+        assert content_type == CONTENT_TYPE_PLAIN_1_0_0 + '; escaping=allow-utf-8'
+        exp = generator(self.registry).decode('utf-8')
+        self.assert_is_utf8(exp)
+        self.assert_is_prom(exp)
+
+    def test_prom_dots_escaping(self):
+        generator, content_type = choose_encoder(CONTENT_TYPE_PLAIN_1_0_0 + '; escaping=dots')
+        assert content_type == CONTENT_TYPE_PLAIN_1_0_0 + '; escaping=dots'
+        exp = generator(self.registry).decode('utf-8')
+        self.assertRegex(exp, r'.*\ndotted_dot_counter__total 1.0\n.*')
+        self.assert_is_prom(exp)
+
+    def test_openmetrics_no_version(self):
+        generator, content_type = choose_encoder('application/openmetrics-text; charset=utf-8; escaping=allow-utf-8')
+        assert content_type == 'application/openmetrics-text; version=1.0.0; charset=utf-8'
+        exp = generator(self.registry).decode('utf-8')
+        # No version -- allow-utf-8 rejected.
+        self.assert_is_escaped(exp)
+        self.assert_is_openmetrics(exp)
+
+    def test_prom_no_version(self):
+        generator, content_type = choose_encoder('text/plain; charset=utf-8; escaping=allow-utf-8')
+        assert content_type == 'text/plain; version=0.0.4; charset=utf-8'
+        exp = generator(self.registry).decode('utf-8')
+        # No version -- allow-utf-8 rejected.
+        self.assert_is_escaped(exp)
+        self.assert_is_prom(exp)
+
+
+@pytest.mark.parametrize("scenario", [
+    {
+        "name": "empty string",
+        "input": "",
+        "expectedUnderscores": "",
+        "expectedDots": "",
+        "expectedValue": "",
+    },
+    {
+        "name": "legacy valid metric name",
+        "input": "no:escaping_required",
+        "expectedUnderscores": "no:escaping_required",
+        "expectedDots": "no:escaping__required",
+        "expectedValue": "no:escaping_required",
+    },
+    {
+        "name": "metric name with dots",
+        "input": "mysystem.prod.west.cpu.load",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load",
+    },
+    {
+        "name": "metric name with dots and underscore",
+        "input": "mysystem.prod.west.cpu.load_total",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load_total",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load__total",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load__total",
+    },
+    {
+        "name": "metric name with dots and colon",
+        "input": "http.status:sum",
+        "expectedUnderscores": "http_status:sum",
+        "expectedDots": "http_dot_status:sum",
+        "expectedValue": "U__http_2e_status:sum",
+    },
+    {
+        "name": "metric name with spaces and emoji",
+        "input": "label with 😱",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__1f631_",
+    },
+    {
+        "name": "metric name with unicode characters > 0x100",
+        "input": "花火",
+        "expectedUnderscores": "__",
+        "expectedDots": "____",
+        "expectedValue": "U___82b1__706b_",
+    },
+    {
+        "name": "metric name with spaces and edge-case value",
+        "input": "label with \u0100",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__100_",
+    },
+])
+def test_escape_metric_name(scenario):
+    input = scenario["input"]
+
+    got = openmetrics.escape_metric_name(input, openmetrics.UNDERSCORES)
+    assert got == scenario["expectedUnderscores"], f"[{scenario['name']}] Underscore escaping failed"
+
+    got = openmetrics.escape_metric_name(input, openmetrics.DOTS)
+    assert got == scenario["expectedDots"], f"[{scenario['name']}] Dots escaping failed"
+
+    got = openmetrics.escape_metric_name(input, openmetrics.VALUES)
+    assert got == scenario["expectedValue"], f"[{scenario['name']}] Value encoding failed"
+
+
+@pytest.mark.parametrize("scenario", [
+    {
+        "name": "empty string",
+        "input": "",
+        "expectedUnderscores": "",
+        "expectedDots": "",
+        "expectedValue": "",
+    },
+    {
+        "name": "legacy valid label name",
+        "input": "no_escaping_required",
+        "expectedUnderscores": "no_escaping_required",
+        "expectedDots": "no__escaping__required",
+        "expectedValue": "no_escaping_required",
+    },
+    {
+        "name": "label name with dots",
+        "input": "mysystem.prod.west.cpu.load",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load",
+    },
+    {
+        "name": "label name with dots and underscore",
+        "input": "mysystem.prod.west.cpu.load_total",
+        "expectedUnderscores": "mysystem_prod_west_cpu_load_total",
+        "expectedDots": "mysystem_dot_prod_dot_west_dot_cpu_dot_load__total",
+        "expectedValue": "U__mysystem_2e_prod_2e_west_2e_cpu_2e_load__total",
+    },
+    {
+        "name": "label name with dots and colon",
+        "input": "http.status:sum",
+        "expectedUnderscores": "http_status_sum",
+        "expectedDots": "http_dot_status__sum",
+        "expectedValue": "U__http_2e_status_3a_sum",
+    },
+    {
+        "name": "label name with spaces and emoji",
+        "input": "label with 😱",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__1f631_",
+    },
+    {
+        "name": "label name with unicode characters > 0x100",
+        "input": "花火",
+        "expectedUnderscores": "__",
+        "expectedDots": "____",
+        "expectedValue": "U___82b1__706b_",
+    },
+    {
+        "name": "label name with spaces and edge-case value",
+        "input": "label with \u0100",
+        "expectedUnderscores": "label_with__",
+        "expectedDots": "label__with____",
+        "expectedValue": "U__label_20_with_20__100_",
+    },
+])
+def test_escape_label_name(scenario):
+    input = scenario["input"]
+
+    got = openmetrics.escape_label_name(input, openmetrics.UNDERSCORES)
+    assert got == scenario["expectedUnderscores"], f"[{scenario['name']}] Underscore escaping failed"
+
+    got = openmetrics.escape_label_name(input, openmetrics.DOTS)
+    assert got == scenario["expectedDots"], f"[{scenario['name']}] Dots escaping failed"
+
+    got = openmetrics.escape_label_name(input, openmetrics.VALUES)
+    assert got == scenario["expectedValue"], f"[{scenario['name']}] Value encoding failed"
 
 
 if __name__ == '__main__':
diff --git a/tests/test_parser.py b/tests/test_parser.py
index e18a878..66cb5ec 100644
--- a/tests/test_parser.py
+++ b/tests/test_parser.py
@@ -6,6 +6,7 @@ from prometheus_client.core import (
     HistogramMetricFamily, Metric, Sample, SummaryMetricFamily,
 )
 from prometheus_client.exposition import generate_latest
+from prometheus_client.openmetrics.exposition import ALLOWUTF8
 from prometheus_client.parser import text_string_to_metric_families
 
 
@@ -367,7 +368,7 @@ prometheus_local_storage_chunk_ops_total{type="unpin"} 32662.0
 
         registry = CollectorRegistry()
         registry.register(TextCollector())
-        self.assertEqual(text.encode('utf-8'), generate_latest(registry))
+        self.assertEqual(text.encode('utf-8'), generate_latest(registry, ALLOWUTF8))
 
 
 if __name__ == '__main__':
diff --git a/tests/test_twisted.py b/tests/test_twisted.py
index e63c903..730e56e 100644
--- a/tests/test_twisted.py
+++ b/tests/test_twisted.py
@@ -1,6 +1,7 @@
 from unittest import skipUnless
 
 from prometheus_client import CollectorRegistry, Counter, generate_latest
+from prometheus_client.openmetrics.exposition import ALLOWUTF8
 
 try:
     from warnings import filterwarnings
@@ -47,6 +48,6 @@ class MetricsResourceTest(TestCase):
                        "with a transport that does not have an abortConnection method")
 
         d.addCallback(readBody)
-        d.addCallback(self.assertEqual, generate_latest(self.registry))
+        d.addCallback(self.assertEqual, generate_latest(self.registry, ALLOWUTF8))
 
         return d
diff --git a/tests/test_wsgi.py b/tests/test_wsgi.py
index 2ecfd72..eb2d056 100644
--- a/tests/test_wsgi.py
+++ b/tests/test_wsgi.py
@@ -3,7 +3,7 @@ from unittest import TestCase
 from wsgiref.util import setup_testing_defaults
 
 from prometheus_client import CollectorRegistry, Counter, make_wsgi_app
-from prometheus_client.exposition import _bake_output, CONTENT_TYPE_LATEST
+from prometheus_client.exposition import _bake_output, CONTENT_TYPE_PLAIN_0_0_4
 
 
 class WSGITest(TestCase):
@@ -35,7 +35,7 @@ class WSGITest(TestCase):
         # Headers
         num_of_headers = 2 if compressed else 1
         self.assertEqual(len(self.captured_headers), num_of_headers)
-        self.assertIn(("Content-Type", CONTENT_TYPE_LATEST), self.captured_headers)
+        self.assertIn(("Content-Type", CONTENT_TYPE_PLAIN_0_0_4), self.captured_headers)
         if compressed:
             self.assertIn(("Content-Encoding", "gzip"), self.captured_headers)
         # Body