#!/usr/bin/python3
import argparse
import json
import logging
import time
from datetime import datetime
from http import HTTPStatus
from typing import Callable, Dict
import requests
def retry(fn, **kwargs):
response = fn(**kwargs)
if response.status_code == HTTPStatus.SERVICE_UNAVAILABLE:
for attempt in range(1, 11):
time.sleep(0.1 * attempt)
logging.info(f"Trying to execute request to {kwargs['url']} attempt {attempt}")
response = fn(**kwargs)
if response.status_code != HTTPStatus.SERVICE_UNAVAILABLE:
break
return response
def make_json_error(code: int, text: str):
return {
"error_code": code,
"error": text
}
def execute_request(request: Callable, path, output=False, token=None, **kwargs):
cookies = {}
headers = {}
if token:
cookies.update({"ses6": token})
headers.update({"x-xsrf-token": token})
logging.info(f"Executing request to {path}")
response = retry(request, url=path, cookies=cookies, headers=headers, **kwargs)
content = response.content.decode("utf-8")
if response.status_code not in (HTTPStatus.OK, HTTPStatus.CREATED):
logging.error(f"Return code {response.status_code}: {content}")
return make_json_error(response.status_code, content)
if output:
try:
result = json.loads(content)
except json.JSONDecodeError as e:
logging.error(f"Error while parsing response: {content}")
result = make_json_error(HTTPStatus.INTERNAL_SERVER_ERROR.value, e.msg)
return result
return None
class API:
def __init__(self, url, email, password):
self.url = url
self.email = email
self.password = password
self.token = None
def get(self, path, params=None):
return execute_request(requests.get, f"{self.url}{path}", output=True, token=self.token, params=params)
def post(self, path, data = {}):
return execute_request(requests.post, f"{self.url}{path}", output=True, token=self.token, json=data)
def delete(self, path):
return execute_request(requests.delete, f"{self.url}{path}", token=self.token)
def __auth(self):
result = self.post("/auth/v4/public/token", {"email": self.email, "password": self.password})
if result is not None and result.get("error"):
raise RuntimeError(result["error"])
self.token = result["token"]
def __enter__(self):
self.__auth()
return self
def __exit__(self, e_type, e_value, e_traceback):
if self.token:
self.delete(f"/auth/v4/token/{self.token}")
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Check VM bandwidth")
parser.add_argument("--url", type=str, metavar="url", help=f"Platform url in format \"https://<your_domain>\"", required=True)
parser.add_argument("--email", type=str, metavar="email", help="Platform administrator email", required=True)
parser.add_argument("--password", type=str, metavar="password", help="Platform administrator password", required=True)
parser.add_argument("--vmid", type=int, metavar="vmid", help="Id of the VM", required=True)
parser.add_argument("--threshold-gib", type=int, metavar="threshold_gib", help="VM bandwidth threshold, in GiB", default=1024)
parser.add_argument("--limit-mbitps", type=int, metavar="limit_mbitps", help="VM bandwidth limit after reaching threshold, in Mbit/s", default=10)
return parser.parse_args()
def main():
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
args = parse_args()
print(args)
now = datetime.now().strftime('%H:%M_%Y%m%d')
first_day = datetime.today().replace(day=1, hour=0, minute=0).strftime('%H:%M_%Y%m%d')
logging.info(f"Checking bandwidth from {first_day} to {now} for VM {args.vmid}")
with API(args.url, args.email, args.password) as api:
summary_bytes = 0
rx = api.get(f"/vm/v3/host/{args.vmid}/metrics", params = {
"target": "net_rx_summary",
"from": first_day,
"until": now,
"output": "single",
"interval": "1month"
})
logging.debug(rx)
if rx is not None and isinstance(rx, Dict) and rx.get("error"):
raise RuntimeError(rx["error"])
if rx:
datapoints = rx[0].get("datapoints", [])
for point in datapoints:
if point[0]:
summary_bytes += point[0]
tx = api.get(f"/vm/v3/host/{args.vmid}/metrics", params = {
"target": "net_tx_summary",
"from": first_day,
"until": now,
"output": "single",
"interval": "1month"
})
if tx is not None and isinstance(tx, Dict) and tx.get("error"):
raise RuntimeError(tx["error"])
if tx:
datapoints = tx[0].get("datapoints", [])
for point in datapoints:
if point[0]:
summary_bytes += point[0]
if summary_bytes / 2**30 >= args.threshold_gib:
logging.info(f"Threshold {args.threshold_gib} GiB reached, limiting speed to {args.limit_mbitps}")
result = api.post(f"/vm/v3/host/{args.vmid}/resource", data = {
"net_in_mbitps": args.limit_mbitps,
"net_out_mbitps": args.limit_mbitps
})
logging.info(result)
if __name__ == '__main__':
main()
En
Es