prometheus exporter async python http trace #python #prometheus #aiohttp #tracing #blackbox #synthetics #asyncio #scrape #probe

aiohttp_trace_exporter.py

import logging
import time
import asyncio
import aiohttp
from prometheus_client import start_http_server, Counter, Histogram

# Prometheus Metrics
h = Histogram('requests_latency', 'latency histogram', ["url", "method"])
c = Counter("requests_total", "Total count of requests", ["code", "url"])

# Trace Configuration for aiohttp
trace_config = aiohttp.TraceConfig()

pool_sem = asyncio.Semaphore(2)


async def on_request_start(session, trace_config_ctx, params):
    trace_config_ctx.start = asyncio.get_event_loop().time()


async def on_request_end(session, trace_config_ctx, params):
    elapsed = asyncio.get_event_loop().time() - trace_config_ctx.start
    h.labels(url=params.url, method=params.method).observe(elapsed)
    logging.debug(f"Request took {elapsed}")

trace_config.on_request_start.append(on_request_start)
trace_config.on_request_end.append(on_request_end)


async def get_url(client, url):
    with await pool_sem:
        try:
            async with client.get(url) as response:
                resp = await response.read()
                c.labels(code=response.status, url=url).inc(1)
                return resp, response.status
        except Exception as e:
            logging.debug(f"Exception: {e}")


async def get(code):
    async with aiohttp.ClientSession(loop=loop, trace_configs=[trace_config]) as client:
        url = f"https://httpbin.org/status/{code}"
        resp, status = await get_url(client, url)
        logging.debug("Done: {code}".format(code=code))
        return resp, status


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG, handlers=[logging.StreamHandler()],
                        format='%(asctime)s %(levelname)s %(processName)s %(threadName)s %(message)s')

    start_http_server(8080)
    loop = asyncio.get_event_loop()
    while True:
        futures = [
            get(code) for code in {
                200: 1,
                202: 2,
                204: 3,
                301: 4,
                400: 5,
                401: 6,
                410: 7,
                500: 8,
            }
        ]
        result = loop.run_until_complete(asyncio.gather(*futures))
        logging.debug(result)
        time.sleep(10)