matt/cosmoserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

first cosmoserver commit

Browse Source
This commit is contained in:
Matt
2026-03-21 21:20:00 -07:00
parent 7c29cbdab5
commit 324eaff135
14 changed files with 1019 additions and 423 deletions
Download Patch File Download Diff File Expand all files Collapse all files

325
files/api/Components.py
View File

@ -2,80 +2,89 @@
import subprocess
import json
import time
import weakref
import base64, hashlib
from typing import Dict, Any, List
from Cosmos_Settings import *
# Global Class Vars
global_max_length = 500
debug_output = False
null_result = [
"",
"null",
None,
[],
"Unknown",
"To Be Filled By O.E.M."
]
# import the component descriptor
try:
with open("component_descriptors.json", encoding="utf-8") as f:
with open("descriptors.json", encoding="utf-8") as f:
component_class_tree: List[Dict] = json.load(f)
except FileNotFoundError as exc:
raise RuntimeError("Descriptor file not found") from exc
component_types = [{"name": entry["name"], "multi_check": entry["multi_check"] == "True"} for entry in component_class_tree]
component_types = []
for entry in component_class_tree:
if entry["name"] != "System":
component_types.append({"name": entry["name"], "multi_check": entry["multi_check"] == "True"})
#################################################################
#################################################################
# Component Class
#################################################################
#################################################################
class Component:
############################################################
# instantiate new component
# this_device is set when the component has multiple instances
############################################################
def __init__(self, name: str, comp_type: str, this_device="None", is_virtual = "True"):
def __init__(self, name: str, comp_type: str, parent_system, this_device=None):
# begin init
self.name = name
self.type = comp_type
self.parent_system = weakref.ref(parent_system)
# this variable is set when the device can have multiples
# it indicates that the commands in the descriptor might need templating
self.this_device = this_device
self.is_virtual = is_virtual
print(f"This device - {self.this_device}")
self.is_virtual = parent_system.is_virtual()
self.cpu_arch = parent_system.get_system_arch()
if self.this_device is None:
log_data(log_output = f"This device - {self.name}", log_level = "log_output")
else:
log_data(log_output = f"This device - {self.this_device}", log_level = "log_output")
# build the component descriptor dictionary
for component in component_class_tree:
if component["name"] == self.type:
COMPONENT_DESCRIPTORS = component
descriptor = COMPONENT_DESCRIPTORS
self._descriptor = descriptor
if descriptor is None:
raise ValueError(
f"Component type '{comp_type}' is not defined in the "
f"component descriptor tree."
)
self._descriptor = self._parse_descriptor()
# store static properties
self.multi_check = self.is_multi()
self.virt_ignore = self._descriptor.get('virt_ignore', [])
self.multi_metrics = self._descriptor.get('multi_metrics', [])
#if 'precheck' in self._descriptor:
# precheck_command = self._descriptor.get('precheck', [])
# precheck_value = int(run_command(precheck_command, zero_only = True))
# if precheck_value == 0:
# raise ValueError(f"No devices of type {self.type}")
self.arch_check = self._descriptor.get('arch_check', [])
if self.is_virtual:
self.virt_ignore = []
# initialize properties
self._properties: Dict[str, str | list[str]] = {}
for key, command in descriptor.get('properties', {}).items():
return_string = True
if key in self.multi_metrics:
return_string = False
if self.this_device != "None":
# this means this component type is a multi and the commands need templating for each device
formatted_command = command.format(this_device=self.this_device)
self._properties[key] = run_command(formatted_command, zero_only = return_string)
else:
self._properties[key] = run_command(command, zero_only = return_string)
print(self._properties[key])
# build the description string
self._description_template: str | None = descriptor.get("description")
self._process_properties()
# build the description string, requires the properties first
self._description_template: str | None = self._descriptor.get("description")
self.description = self._description_template.format(**self._properties)
# initialize metrics
self._metrics: Dict[str, str] = {}
self.update_metrics()
def __str__(self):
self_string = (f"Component name: {self.name}, type: {self.type} - "
f"{self.description}")
@ -92,13 +101,21 @@ class Component:
def update_metrics(self):
for key, command in self._descriptor.get('metrics', {}).items():
if self.this_device != "None":
formatted_command = command.format(this_device=self.this_device)
this_metric = run_command(formatted_command, True)
if this_metric is not None:
self._metrics[key] = this_metric
else:
self._metrics[key] = run_command(command, zero_only = True)
log_data(log_output = f"Key: {key} - Command: {command}", log_level = "noisy_test")
formatted_command = command
if self.arch_check is not None:
arch_variance = self._descriptor.get('arch_variance', {})
if key in arch_variance:
if self.cpu_arch in formatted_command:
formatted_command = command[self.cpu_arch]
else:
formatted_command = f"echo Missing {self.cpu_arch} command"
if self.this_device is not None:
formatted_command = formatted_command.format(this_device=self.this_device)
if formatted_command is not None:
result = run_command(formatted_command, zero_only = True)
if result not in null_result:
self._metrics[key] = result
def get_property(self, type = None):
if type == None:
@ -112,8 +129,52 @@ class Component:
return component_type["multi_check"]
return False
# return descriptor for this device type
def _parse_descriptor(self):
for component in component_class_tree:
if component["name"] == self.type:
COMPONENT_DESCRIPTORS = component
descriptor = COMPONENT_DESCRIPTORS
if descriptor is None:
raise ValueError(
f"Component type '{comp_type}' is not defined in the "
f"component descriptor tree."
)
return descriptor
# iterate over all properties to process descriptor
def _process_properties(self):
for key, command in self._descriptor.get('properties', {}).items():
return_string = True
if key in self.multi_metrics:
return_string = False
formatted_command = self._parse_command(key, command, return_string)
log_data(log_output = f"Property {key} - command: {formatted_command}", log_level = "debug_output")
result = run_command(formatted_command, zero_only = return_string)
if result not in null_result:
self._properties[key] = result
# helper function to parse command key
def _parse_command(self, key: str, command: str | list[str], return_string = True):
result_command = command
log_data(log_output = f"_parse_command - {command}", log_level = "debug_output")
if self.arch_check:
# since the keys are stored with the arch variable this can be concise
arch_variance = self._descriptor.get('arch_variance', {})
if key in arch_variance:
if self.cpu_arch in result_command:
log_data(log_output = f"arch_variance - {key} - {result_command}", log_level = "debug_output")
result_command = result_command[self.cpu_arch]
else:
result_command = f"echo Missing {self.cpu_arch} command"
if self.this_device is not None:
# template the key if the component type can have multiples
result_command = command.format(this_device=self.this_device)
log_data(log_output = f"result - {result_command}", log_level = "debug_output")
return result_command
########################################################
# redis data functions
# keyed data functions
########################################################
def get_properties_keys(self, component = None):
@ -230,9 +291,10 @@ class Component:
}
return result
############################################################
############################################################
# System Class
# this is a big one...
############################################################
############################################################
class System:
@ -241,25 +303,18 @@ class System:
# system variable declarations
########################################################
static_key_variables = [
{"name": "Hostname", "command": "hostname"},
{"name": "Virtual Machine", "command": 'echo $( [ "$(systemd-detect-virt)" = none ] && echo False || echo True )', "req_check": "False"},
{"name": "CPU Architecture:", "command": "lscpu --json | jq -r '.lscpu[] | select(.field==\"Architecture:\") | .data'"},
{"name": "OS Kernel", "command": "uname -r"},
{"name": "OS Name", "command": "cat /etc/os-release | grep PRETTY | cut -d\\\" -f2"},
{"name": "Manufacturer", "command": "sudo dmidecode --type 1 | grep Manufacturer: | cut -d: -f2 | sed -e 's/^[ \\t]*//'"},
{"name": "Product Name", "command": "sudo dmidecode --type 2 | grep 'Product Name:' | cut -d: -f2 | sed -e 's/^[ \\t]*//'"},
{"name": "Serial Number", "command": "sudo dmidecode --type 2 | grep 'Serial Number: '| cut -d: -f2 | sed -e 's/^[ \\t]*//'"},
]
dynamic_key_variables = [
{"name": "System Uptime", "command": "uptime -p"},
{"name": "Current Date", "command": "date '+%D %r'"},
]
virt_ignore = [
"Product Name",
"Serial Number"
]
for component in component_class_tree:
if component["name"] == "System":
SYSTEM_DESCRIPTOR = component
descriptor = SYSTEM_DESCRIPTOR
if descriptor is None:
raise ValueError(
f"Component type 'System' is not defined in the "
f"component descriptor tree."
)
static_key_variables = descriptor["static_key_variables"]
dynamic_key_variables = descriptor["dynamic_key_variables"]
virt_ignore = descriptor["virt_ignore"]
########################################################
# instantiate new system
@ -268,8 +323,9 @@ class System:
def __init__(self, name: str):
# the system needs a name
self.name = name
if debug_output:
print(f"System initializing, name {self.name}")
log_data(log_output = f"System initializing, name {self.name}", log_level = "debug_output")
self.uuid = run_command(cmd = "cat /etc/machine-id", zero_only = True)
self.short_id = self.short_uuid(self.uuid)
# system contains an array of component objects
self.components = []
self.component_class_tree = component_class_tree
@ -277,6 +333,7 @@ class System:
self._properties: Dict[str, str] = {}
self._metrics: Dict[str, str] = {}
self._virt_string = run_command('systemd-detect-virt', zero_only = True, req_check = False)
self._virt_ignore = self.virt_ignore
if self._virt_string == "none":
self._virt_ignore = []
@ -285,36 +342,56 @@ class System:
# load static keys
for static_key in self.static_key_variables:
if static_key["name"] not in self._virt_ignore:
command = static_key["command"]
if "req_check" in static_key:
result = run_command(command, zero_only = True, req_check = static_key["req_check"])
else:
result = run_command(command, zero_only = True)
if debug_output:
print(f'Static key [{static_key["name"]}] - command [{command}] - output [{result}]')
self._properties[static_key["name"]] = result
self.process_property(static_key = static_key)
# initialize live keys
self.update_live_keys()
# initialze components
self.load_components()
for component in component_types:
self.create_component(component)
def __str__(self):
components_str = "\n".join(f" - {c}" for c in self.components)
return f"System hostname: {self.name}\nComponent Count: {self.get_component_count()}\n{components_str}"
def __repr__(self):
self_string = f"Cosmostat Client {self.short_id}"
def short_uuid(self, value: str, length=8):
hasher = hashlib.md5()
hasher.update(value.encode('utf-8'))
full_hex = hasher.hexdigest()
return full_hex[:length]
########################################################
# critical class functions
########################################################
# process static keys
def process_property(self, static_key):
command = static_key["command"]
if "arch_check" in static_key:
arch_string = run_command("lscpu --json | jq -r '.lscpu[] | select(.field==\"Architecture:\") | .data'", zero_only = True)
if arch_string in command:
command = command[arch_string]
else:
command = f"echo Missing {arch_string} command"
if "req_check" in static_key:
result = run_command(command, zero_only = True, req_check = static_key["req_check"])
else:
result = run_command(command, zero_only = True)
log_data(log_output = f'Static key [{static_key["name"]}] - command [{command}] - output [{result}]', log_level = "debug_output")
if result not in null_result:
self._properties[static_key["name"]] = result
# update only system dynamic keys
def update_live_keys(self):
for live_key in self.dynamic_key_variables:
if live_key['command'] is not None:
command = live_key['command']
result = run_command(command, zero_only = True)
self._metrics[live_key['name']] = result
if debug_output:
print(f'Command {live_key["name"]} - [{command}] Result - [{result}]')
if result not in null_result:
self._metrics[live_key['name']] = result
log_data(log_output = f'Command {live_key["name"]} - [{command}] Result - [{result}]', log_level = "noisy_test")
# update all dynamic keys, including components
def update_system_state(self):
@ -322,33 +399,25 @@ class System:
for component in self.components:
component.update_metrics()
# check for components
def load_components(self):
for component in component_types:
component_name = component["name"]
multi_check = component["multi_check"]
# if multi, note that the command in device_list creates the list of things to pipe into this_device
if multi_check:
print(f"Creating one component of type {component_name} for each one found")
component_type_device_list = get_device_list(component_name)
component_id = 0
for this_device in component_type_device_list:
this_component_ID = component_type_device_list.index(this_device)
this_component_name = f"{component_name} {this_component_ID}"
print(f"{this_component_name} - {component_name} - {this_device}")
new_component = Component(name = this_component_name, comp_type = component_name, this_device = this_device)
self.add_components(new_component)
else:
if debug_output:
print(f'Creating component {component["name"]}')
self.add_components(Component(name = component_name, comp_type = component_name, is_virtual = self.is_virtual()))
# Add a component to the system
def add_components(self, component: Component,):
if debug_output:
print(f"Component description: {component.description}")
self.components.append(component)
# component creation helper
def create_component(self, component):
component_name = component["name"]
multi_check = component["multi_check"]
# if multi, note that the command in device_list creates the list of things to pipe into this_device
if multi_check:
log_data(log_output = f"Creating one component of type {component_name} for each one found", log_level = "log_output")
component_type_device_list = get_device_list(component_name)
component_id = 0
for this_device in component_type_device_list:
this_component_ID = component_type_device_list.index(this_device)
this_component_name = f"{component_name} {this_component_ID}"
log_data(log_output = f"{this_component_name} - {component_name} - {this_device}", log_level = "debug_output")
new_component = Component(name = this_component_name, comp_type = component_name, this_device = this_device, parent_system = self)
self.components.append(new_component)
else:
log_data(log_output = f'Creating component {component["name"]}', log_level = "debug_output")
new_component = Component(name = component_name, comp_type = component_name, parent_system = self)
self.components.append(new_component)
########################################################
# helper class functions
@ -380,7 +449,7 @@ class System:
def is_virtual(self):
vm_check = self.get_property('Virtual Machine')
print(f'vm_check: {vm_check}')
log_data(log_output = f'vm_check: {vm_check}', log_level = "debug_output")
return vm_check
def check_system_timer(self):
@ -390,6 +459,9 @@ class System:
def get_component_class_tree(self):
return self.component_class_tree
def get_system_arch(self):
return self.get_property("CPU Architecture")
########################################################
# static metrics redis data functions
########################################################
@ -493,43 +565,6 @@ class System:
"Metric": f"component_count: {self.get_component_count()}"
})
return result
# straggler functions, might cut them
# return both static and dynamic data
def get_sysvars_summary_keys(self):
result = []
for name, value in self._properties.items():
thisvar = {
"name": "System Class Property",
"type": name,
"value": value
}
result.append(thisvar)
for name, value in self._metrics.items():
thisvar = {
"name": "System Class Metric",
"type": name,
"value": value
}
result.append(thisvar)
return result
def get_component_strings(self, component_type: type = None):
if component_type is None:
result = []
for component in self.components:
result.append(component.description)
return result
else:
result = []
for component in self.components:
if component.type == component_type:
result.append(component.description)
if component.is_multi():
return result
else:
return result[0]
############################################################
# Non-class Helper Functions
@ -557,10 +592,12 @@ def get_device_list(device_type_name: str):
precheck_command = component["precheck"]
precheck_value_output = run_command(precheck_command, zero_only = True)
precheck_value = int(precheck_value_output)
print(f"Precheck found - {precheck_command} - {precheck_value}")
log_data(log_output = f"Precheck found - {precheck_command} - {precheck_value}", log_level = "log_output")
if component["name"] == device_type_name and precheck_value != 0:
device_list_command = component["device_list"]
device_list_result = run_command(device_list_command)
result = device_list_result
return result