prometheus/client_python: How to assign new registry without restart?

Question

I have written a small script that instruments a Flask application and I want to write unit tests where every test can write requests against a mock-up Flask app and test metrics without having to deal with metrics/requests from previous test methods like so:

def test_grouped_codes():
    app = create_app()
    instrument(app)
    # test stuff

But I am not able to "reset" the registry and so I get the error "Duplicated timeseries in CollectorRegistry" all the time.

How can I reset the registry (or set it to an empty registry) of the Prometheus Python client library during run-time?

Among other things I have tried the following, but it does not work:

def create_app():
    app = Flask(__name__)
    registry = CollectorRegistry()  # Create new registry.
    prometheus_client.REGISTRY = registry  # Try to override global registry.
    prometheus_client.registry.REGISTRY = registry  # Try to override global registry.

    @app.route("/")
    def home():
        return "Hello World!"

    # More functions ...

    @app.route("/metrics")
    @FlaskInstrumentator.do_not_track()
    def metrics():
        data = generate_latest(registry)
        headers = {
            "Content-Type": CONTENT_TYPE_LATEST,
            "Content-Length": str(len(data))}
        return data, 200, headers

    return app

---

I have found the following qa on stack overflow here. @brian-brazil recommends to declare the metrics on the module-level, but then I would have to hard-code label names which I wanted to avoid. Some use handler, the others method or path so I want to keep that customizable.

Answer 1

Okay thanks to the hint in this answer by @Xitrum "deregister method" I found a solution:

collectors = list(REGISTRY._collector_to_names.keys())
for collector in collectors:
    REGISTRY.unregister(collector)

Now all tests can start with their own registry:

def test_metrics_endpoint_availability():
    app = create_app()
    FlaskInstrumentator(app).instrument()
    client = app.test_client()

    response = client.get("/")
    response = client.get("/metrics")
    
    # Test stuff
    

def test_grouped_status_codes():
    app = create_app()
    FlaskInstrumentator(app).instrument()
    client = app.test_client()

    client.get("/does_not_exist")  # Should be ignored.
    client.get("/does_not_exist")  # Should be ignored.
    client.post("/")
    client.post("/")
    
    # Test stuff

---

Edit 2023

Here is a helper function I use for unit tests atm:

from prometheus_client import REGISTRY

def reset_prom_collectors() -> None:
    """Resets collectors in the default Prometheus registry.

    Modifies the `REGISTRY` registry. Supposed to be called at the beginning
    of individual test functions. Else registry is reused across test functions
    and so we can run into errors like duplicate metrics or unexpected values
    for metrics.
    """

    # Unregister all collectors.
    collectors = list(REGISTRY._collector_to_names.keys())
    print(f"before unregister collectors={collectors}")
    for collector in collectors:
        REGISTRY.unregister(collector)
    print(f"after unregister collectors={list(REGISTRY._collector_to_names.keys())}")

    # Import default collectors.
    from prometheus_client import gc_collector, platform_collector, process_collector

    # Re-register default collectors.
    process_collector.ProcessCollector()
    platform_collector.PlatformCollector()
    gc_collector.GCCollector()

    print(f"after re-register collectors={list(REGISTRY._collector_to_names.keys())}")