birb_latency_reporter/reporter.py

import os
import time
import requests
from apscheduler.schedulers.blocking import BlockingScheduler
from typing import List, Tuple


def run_report_task(hostname: str, test_urls: list, report_urls: list):
    print('.', end='', flush=True)
    for testpoint_name, testpoint_url in test_urls:

        start_time = time.time()
        try:
            r = requests.get(testpoint_url, timeout=10)
        except requests.exceptions.Timeout:
            print(testpoint_name, testpoint_url, "Timed out")
            continue
        else:
            latency_s = time.time() - start_time

        if r.status_code != 204:
            print(testpoint_name, testpoint_url, "Wrong response:", r.status_code)
            continue

        report = {
            "client": hostname,
            "site": testpoint_name,
            "measurements": {
                "latency": latency_s * 1000
            }
        }

        for report_name, report_url in report_urls:
            r = requests.post(report_url, json=report)

            if r.status_code != 201:
                print(report_name, report_url, "Wrong response:", r.status_code)
                continue


def parse_url_list(url_list: str) -> List[Tuple[str, str]]:
    parsed_urls = []
    for pair in url_list.split(' '):
        parsed_urls.append(
            tuple(pair.split(';', 1))
        )
    return parsed_urls


def main():
    test_urls = parse_url_list(os.environ.get("TEST_URLS", ""))
    report_urls = parse_url_list(os.environ.get("REPORT_URLS", ""))
    frequency = float(os.environ.get("FREQUENCY", 15))
    hostname = os.environ['HOSTNAME']

    if (not test_urls) or (not report_urls):
        print("No urls defined")
        return

    print("Testing the following endpoints:")
    for testpoint_name, testpoint_url in test_urls:
        print(testpoint_name, testpoint_url)

    print("Reporting to the following endpoints:")
    for report_name, report_url in report_urls:
        print(report_name, report_url)

    scheduler = BlockingScheduler()
    scheduler.add_job(lambda: run_report_task(hostname, test_urls, report_urls), trigger='interval', seconds=frequency)
    scheduler.start()


if __name__ == '__main__':
    main()