birbnetes/birb_latency_reporter
birbnetes/birb_latency_reporter
4
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Added support to new format

Browse Source
This commit is contained in:
Pünkösd Marcell 2021-12-02 23:57:33 +01:00
parent e33177a07c
commit 0c791452bf
1 changed files with 28 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
reporter.py
View File

@ -1,52 +1,68 @@
import os
import time
import requests
from urllib.parse import urljoin
from apscheduler.schedulers.blocking import BlockingScheduler
from typing import List, Tuple
def run_report_task(hostname: str, test_urls: list, report_urls: list):
for test_url in test_urls:
for testpoint_name, testpoint_url in test_urls:
start_time = time.time()
try:
r = requests.get(test_url, timeout=10)
r = requests.get(testpoint_url, timeout=10)
except requests.exceptions.Timeout:
print(test_url, "Timed out")
print(testpoint_name, testpoint_url, "Timed out")
continue
else:
latency_s = time.time() - start_time
if r.status_code != 204:
print(test_url, "Wrong response")
print(testpoint_name, testpoint_url, "Wrong response:", r.status_code)
continue
report = {
"client": hostname,
"cloud": test_url,
"site": testpoint_name,
"measurements": {
"latency": latency_s * 1000
}
}
for report_url in report_urls:
for report_name, report_url in report_urls:
r = requests.post(report_url, json=report)
if r.status_code != 201:
print(report_url, "Wrong response")
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 = os.environ.get("TEST_URLS", "").split(';')
report_urls = os.environ.get("REPORT_URLS", "").split(';')
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
hostname = os.environ['HOSTNAME']
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)