如何在Prometheus中用Counter统计动态变化的唯一URI请求数

Got it, let's walk through exactly how to solve this problem with Python's collections.Counter—it's tailor-made for this kind of dynamic, unbounded counting scenario where URIs can pop up out of nowhere and repeat.

Core Idea #

Counter is a subclass of Python's dictionary built specifically for counting hashable objects. The best part? You don't need to predefine all possible URIs (no constant tags required!). It automatically handles new keys (URIs) by initializing their count to 0 the first time you reference them, then increments as needed.

Step 1: Initialize the Counter #

First, import and set up your counter:

from collections import Counter

# Initialize an empty counter to track URI requests
uri_request_counter = Counter()

Step 2: Count Dynamic Requests #

Every time you receive a request, grab its URI and update the counter. This works whether the URI has been seen before or is brand new:

# Example: Simulate a stream of dynamic requests
dynamic_requests = [
    "/foo", "/foo", "/bar", "/foo",
    "/pooh", "/bar", "/pooh", "/pooh",
    "/new-uri", "/new-uri"  # New URI that wasn't anticipated
]

for uri in dynamic_requests:
    # Increment count for this URI—Counter handles new URIs automatically
    uri_request_counter[uri] += 1

# Check the results
print(uri_request_counter)
# Output: Counter({'/foo': 3, '/pooh': 3, '/bar': 2, '/new-uri': 2})

Step 3: Real-World HTTP Service Example #

If you're integrating this into an actual HTTP service (like Flask), you can hook into request handling to update the counter in real time. No need to predefine routes or URI constants—new routes will be counted automatically as they're hit:

from flask import Flask, request
from collections import Counter
import threading  # For thread safety in multi-threaded servers

app = Flask(__name__)
uri_counter = Counter()
# Add a lock if your server uses multiple threads (most do!)
counter_lock = threading.Lock()

# Run this before every request to count the URI
@app.before_request
def track_uri_requests():
    current_uri = request.path
    # Use the lock to prevent race conditions when updating the counter
    with counter_lock:
        uri_counter[current_uri] += 1

# Example routes (add more anytime—counter will track them)
@app.route("/foo")
def foo_endpoint():
    return "Response from /foo"

@app.route("/bar")
def bar_endpoint():
    return "Response from /bar"

# Add a new URI later—no changes needed to the counter logic!
@app.route("/pooh")
def pooh_endpoint():
    return "Response from /pooh"

# Endpoint to view current stats
@app.route("/request-stats")
def get_stats():
    with counter_lock:
        # Return the counter as a dictionary for easy JSON serialization
        return dict(uri_counter)

if __name__ == "__main__":
    app.run(threaded=True)

Useful Bonus Operations #

• Get the most frequent URIs: uri_counter.most_common(3) (returns top 3 URIs with counts)
• Reset a specific URI's count: uri_counter["/foo"] = 0
• Remove a URI from the counter: del uri_counter["/bar"]

Why This Works #

Since Counter acts like a dictionary, it dynamically adds new keys (URIs) as they're encountered. No upfront configuration or constant tags are required—perfect for your scenario where URIs change over time and new ones are added regularly.

内容的提问来源于stack exchange，提问作者bharat nc