ungleich-tools/opennebula-vm-etcd/put-vm-info-into-etcd.py

98 lines
2.8 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import json
from enum import IntEnum
from xmlrpc.client import ServerProxy as RPCClient
from xmltodict import parse
from config import config, etcd_client
# Constants
ALL_VM_STATES = -1
START_ID = -1 # First id whatever it is
END_ID = -1 # Last id whatever it is
def put_under_list(obj):
if not isinstance(obj, list):
return [obj]
return obj
class VMState(IntEnum):
INIT = 0
PENDING = 1
HOLD = 2
ACTIVE = 3
STOPPED = 4
SUSPENDED = 5
DONE = 6
FAILED = 7
POWEROFF = 8
UNDEPLOYED = 9
CLONING = 10
CLONING_FAILURE = 11
class VmFilterFlag(IntEnum):
UIDUserResources = 0 # UID Users Resources
UserAndItsGroupsResources = -1 # Resources belonging to the user and any of his groups
AllResources = -2 # All resources
UserResources = -3 # Resources belonging to the user
UserPrimaryGroupResources = -4 # Resources belonging to the users primary group
class VM:
def __init__(self, vm: dict):
self.id = vm.get('ID', None)
self.owner = {
'id': vm.get('UID', None),
'name': vm.get('UNAME', None),
'gname': vm.get('GNAME', None)
}
self.name = vm.get('NAME', None)
self.status = vm.get('STATE', None)
if self.status:
self.status = VMState(int(self.status)).name.lower()
template = vm['TEMPLATE']
self.disk = put_under_list(template.get('DISK', []))
self.graphics = template.get('GRAPHICS', {})
self.memory = template.get('MEMORY', None)
self.nic = put_under_list(template.get('NIC', []))
self.vcpu = template.get('VCPU', None)
self.host = {
'name': ((vm.get('HISTORY_RECORDS', {}) or {}).get('HISTORY', {}) or {}).get('HOSTNAME', None),
'id': ((vm.get('HISTORY_RECORDS', {}) or {}).get('HISTORY', {}) or {}).get('HID', None),
}
self.snapshots = put_under_list(vm.get('SNAPSHOTS', []))
def get_data(self):
return {
attr: getattr(self, attr)
for attr in dir(self)
if not attr.startswith('__') and not callable(getattr(self, attr))
}
def main():
with RPCClient('https://opennebula.ungleich.ch:2634/RPC2') as rpc_client:
success, response, *_ = rpc_client.one.vmpool.infoextended(
config['oca']['client_secrets'], VmFilterFlag.AllResources.value, START_ID, END_ID, ALL_VM_STATES
)
if success:
vms = json.loads(json.dumps(parse(response)))['VM_POOL']['VM']
for i, vm in enumerate(vms):
vm_id = vm['ID']
etcd_client.put(f'/opennebula/vm/{vm_id}', vm)
parsed_vm = VM(vm)
etcd_client.put(f'/opennebula/parsed_vm/{parsed_vm.id}', parsed_vm.get_data())
else:
print(response)
if __name__ == "__main__":
main()