Hi all.
After a very huge amount of time spent on looking for the solution on how to monitor Proxmox properly, light me an idea bulb - why not to try create a new monitoring solution? The problem is that I don’t have any knowledge and experience in programming. So I tried to get a bit help from GPT Chat, and here is whole idea, may be someone will succeed with realization.
Proxmox have its own API:
Documentation - Proxmox VE API - Proxmox VE
API Viewer - Proxmox VE API Documentation
The best & easiest way to connect to Proxmox API (as far as i see) is to use a username configured with read-only privileges + using an API token. The second option is to use pvesh command, but I not tried to do that.
I succeeded to connect and get data using command_line platform and curl command.
Here is example of the sensor:
- platform: command_line
command: 'curl -k -H "Authorization: PVEAPIToken=user@pve!blah-blah-blah=1234a56b-c7d8-9012-345d-67ef890g123h" https://proxmox_ip:8006/api2/json/nodes/pvepc/disks/lvm'
name: PVE API LVM
value_template: '{{ value_json }}'
scan_interval: 20
command_timeout: 30
Here is the output for state of this sensor
`{'data': {'children': [{'free': 17175674880, 'size': 511570870272, 'leaf': 0, 'children': [{'name': '/dev/sdb3', 'leaf': 1, 'size': 511570870272, 'free': 17175674880}], 'name': 'pve', 'lvcount': 7}], 'leaf': 0}}`
Because I’m noob in programming, I not succeeded to parse that response.
As i said before, I tried to get help from AI GPT Chat. And that what it suggests:
import json
import requests
class ProxmoxSensor(Entity):
def __init__(self, name, auth_data,endpoint):
self._name = name
self._auth_data = auth_data
self._state = None
self._endpoint=endpoint
@property
def name(self):
return self._name
@property
def state(self):
return self._state
def update(self):
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {self._auth_data["token"]}'
}
response = requests.get(f'{self._auth_data["address"]}/api2/json/{self._endpoint}', headers=headers)
response.raise_for_status()
data = json.loads(response.text)['data']
self._state = self._get_value(data)
def _get_value(self, data):
"""This method should be implemented in the subclasses to extract the relevant value from the API response"""
pass
class ProxmoxHDDSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/storage'
super().__init__('Proxmox HDD Usage', auth_data,self._endpoint)
def _get_value(self, data):
total_size = 0
used_size = 0
for storage in data:
if storage['type'] == 'disk':
total_size += storage
used_size += storage['used']
percent_used = used_size / total_size * 100
return round(percent_used, 2)
class ProxmoxRAMSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/memory'
super().__init__('Proxmox RAM Usage', auth_data,self._endpoint)
def _get_value(self, data):
total_ram = data['total']
used_ram = data['used']
percent_used = used_ram / total_ram * 100
return round(percent_used, 2)
class ProxmoxCPUSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/cpu'
super().__init__('Proxmox CPU Usage', auth_data,self._endpoint)
def _get_value(self, data):
percent_used = data['load_avg']
return percent_used
class ProxmoxCPUTempSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/hardware'
super().__init__('Proxmox CPU Temperature', auth_data,self._endpoint)
def _get_value(self, data):
for entry in data['entries']:
if entry['key'] == 'cpu_temp':
return entry['value']
def authenticate(address,node_name,username,api_token):
headers = {
'Content-Type': 'application/json',
}
data = {
'username': username,
'password': api_token,
}
response = requests.post(f'{address}/api2/json/access/ticket', headers=headers, json=data)
response.raise_for_status()
auth_data = json.loads(response.text)['data']
auth_data.update({"node":node_name,"address":address})
return auth_data
def setup_platform(hass, config, add_entities, discovery_info=None):
address = input('Enter the Proxmox VE address: ')
node_name = input('Enter the Proxmox VE node name: ')
username = input('Enter your Proxmox VE username: ')
api_token = input('Enter your Proxmox VE API Token: ')
auth_data = authenticate(address,node_name,username,api_token)
add_entities([
ProxmoxHDDSensor(auth_data),
ProxmoxRAMSensor(auth_data),
ProxmoxCPUSensor(auth_data),
ProxmoxCPUTempSensor(auth_data),
])
And example of __init__.py file:
import json
import requests
from homeassistant.const import CONF_NAME
from homeassistant.helpers.entity import Entity
class ProxmoxSensor(Entities):
def __init__(self, name, auth_data,endpoint):
self._name = name
self._state = None
self._auth_data = auth_data
self._endpoint = endpoint
@property
def name(self):
"""Return the name of the sensor."""
return self._name
@property
def state(self):
"""Return the state of the sensor."""
return self._state
def update(self):
"""Fetch new state data for the sensor.
This is the only method that should fetch new data for Home Assistant.
"""
response = requests.get(
f'{self._auth_data["address"]}/api2/json/{self._endpoint}',
headers={
'Content-Type': 'application/json',
'Authorization': f'PVEAPIToken={self._auth_data["ticket"]}',
},
)
response.raise_for_status()
data = json.loads(response.text)['data']
self._state = self._get_value(data)
def _get_value(self, data):
"""Return the value for the sensor.
Subclasses should override this method to return the appropriate value.
"""
raise NotImplementedError
class ProxmoxHDDSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/storage'
super().__init__('Proxmox HDD Usage', auth_data,self._endpoint)
def _get_value(self, data):
total_size = 0
used_size = 0
for storage in data['storage']:
total_size += storage['total']
used_size += storage['used']
percent_used = used_size / total_size * 100
return round(percent_used, 2)
class ProxmoxRAMSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/memory'
super().__init__('Proxmox RAM Usage', auth_data,self._endpoint)
def _get_value(self, data):
total_ram = data['total']
used_ram = data['used']
percent_used = used_ram / total_ram * 100
return round(percent_used, 2)
class ProxmoxCPUSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/status'
super().__init__('Proxmox CPU Usage', auth_data,self._endpoint)
def _get_value(self, data):
return data['cpu']
class ProxmoxCPUTempSensor(ProxmoxSensor):
def __init__(self, auth_data):
self._endpoint = f'nodes/{auth_data["node"]}/status'
super().__init__('Proxmox CPU Temperature', auth_data,self._endpoint)
def _get_value(self, data):
return data['temperature']
def setup_platform(hass, config, add_entities, discovery_info=None):
"""Set up the Proxmox VE sensor."""
address = input('Enter your Proxmox VE address: ')
node_name = input('Enter your Proxmox VE node name: ')
username = input('Enter your Proxmox VE username: ')
api_token = input('Enter your Proxmox VE API Token: ')
auth_data = authenticate(address,node_name,username,api_token)
add_entities([
ProxmoxHDDSensor(auth_data),
ProxmoxRAMSensor(auth_data),
ProxmoxCPUSensor(auth_data),
ProxmoxCPUTempSensor(auth_data),
])
def authenticate(address,node_name,username,api_token):
data = {
'username':username,
'api_token': api_token
}
response = requests.post(f'{address}/api2/json/access/token', data=data)
response.raise_for_status()
auth_data = json.loads(response.text)['data']
auth_data['address'] = address
auth_data['node'] = node_name
return auth_data
So I hope that it will help someone to build a new add-on for monitoring Proxmox instances