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.

There's not enough information to go off here, what does the instrumentation look like? — brian-brazil
Like so: github.com/trallnag/prometheus-flask-instrumentator/blob/master/… I'm creating a single metric inside a method and then instrument the Flask app using decorators. Nothing new but I was missing stuff like grouping of status codes from other similar packages — trallnag

trallnag trallnag · Accepted Answer · 2020-06-20T17:25:12

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

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

1 Answers