106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
"""Proxmox Client."""
|
|
|
|
import requests
|
|
from prometheus_client import Counter
|
|
|
|
from prometheuspvesd.config import SingleConfig
|
|
from prometheuspvesd.exception import APIError
|
|
from prometheuspvesd.logger import SingleLog
|
|
from prometheuspvesd.model import HostList
|
|
from prometheuspvesd.utils import to_bool
|
|
|
|
try:
|
|
from proxmoxer import ProxmoxAPI
|
|
|
|
HAS_PROXMOXER = True
|
|
except ImportError:
|
|
HAS_PROXMOXER = False
|
|
|
|
PVE_REQUEST_COUNT_TOTAL = Counter("pve_sd_requests_total", "Total count of requests to PVE API")
|
|
PVE_REQUEST_COUNT_ERROR_TOTAL = Counter(
|
|
"pve_sd_requests_error_total", "Total count of failed requests to PVE API"
|
|
)
|
|
|
|
|
|
class ProxmoxClient:
|
|
"""Proxmox API Client."""
|
|
|
|
def __init__(self):
|
|
if not HAS_PROXMOXER:
|
|
self.log.sysexit_with_message(
|
|
"The Proxmox VE Prometheus SD requires proxmoxer: "
|
|
"https://pypi.org/project/proxmoxer/"
|
|
)
|
|
|
|
self.config = SingleConfig()
|
|
self.log = SingleLog()
|
|
self.logger = SingleLog().logger
|
|
self.client = self._auth()
|
|
self.logger.debug("Successfully authenticated")
|
|
self.host_list = HostList()
|
|
|
|
def _auth(self):
|
|
try:
|
|
self.logger.debug(
|
|
"Trying to authenticate against {} as user {}".format(
|
|
self.config.config["pve"]["server"], self.config.config["pve"]["user"]
|
|
)
|
|
)
|
|
|
|
if self.config.config["pve"]["token_name"]:
|
|
self.logger.debug("Using token login")
|
|
return ProxmoxAPI(
|
|
self.config.config["pve"]["server"],
|
|
user=self.config.config["pve"]["user"],
|
|
token_name=self.config.config["pve"]["token_name"],
|
|
token_value=self.config.config["pve"]["token_value"],
|
|
verify_ssl=to_bool(self.config.config["pve"]["verify_ssl"]),
|
|
timeout=self.config.config["pve"]["auth_timeout"],
|
|
)
|
|
|
|
return ProxmoxAPI(
|
|
self.config.config["pve"]["server"],
|
|
user=self.config.config["pve"]["user"],
|
|
password=self.config.config["pve"]["password"],
|
|
verify_ssl=to_bool(self.config.config["pve"]["verify_ssl"]),
|
|
timeout=self.config.config["pve"]["auth_timeout"],
|
|
)
|
|
except requests.RequestException as e:
|
|
PVE_REQUEST_COUNT_ERROR_TOTAL.inc()
|
|
raise APIError(str(e)) from e
|
|
|
|
def _do_request(self, *args):
|
|
PVE_REQUEST_COUNT_TOTAL.inc()
|
|
try:
|
|
# create a new tuple containing nodes and unpack it again for client.get
|
|
return self.client.get(*("nodes", *args))
|
|
except requests.RequestException as e:
|
|
PVE_REQUEST_COUNT_ERROR_TOTAL.inc()
|
|
raise APIError(str(e)) from e
|
|
|
|
def get_nodes(self):
|
|
self.logger.debug("fetching all nodes")
|
|
return self._do_request()
|
|
|
|
def get_all_vms(self, pve_node):
|
|
self.logger.debug(f"fetching all vms on node {pve_node}")
|
|
return self._do_request(pve_node, "qemu")
|
|
|
|
def get_all_containers(self, pve_node):
|
|
self.logger.debug(f"fetching all containers on node {pve_node}")
|
|
return self._do_request(pve_node, "lxc")
|
|
|
|
def get_instance_config(self, pve_node, pve_type, vmid):
|
|
self.logger.debug(f"fetching instance config for {vmid} on {pve_node}")
|
|
return self._do_request(pve_node, pve_type, vmid, "config")
|
|
|
|
def get_agent_info(self, pve_node, pve_type, vmid):
|
|
self.logger.debug(f"fetching agent info for {vmid} on {pve_node}")
|
|
return self._do_request(pve_node, pve_type, vmid, "agent", "info")["result"]
|
|
|
|
def get_network_interfaces(self, pve_node, vmid):
|
|
self.logger.debug(f"fetching network interfaces for {vmid} on {pve_node}")
|
|
return self._do_request(pve_node, "qemu", vmid, "agent", "network-get-interfaces")[
|
|
"result"
|
|
]
|