Base Class for Infrastructure and Service Management
ObjSubstrate is an abstract base class that defines a standardized interface for interacting with various types of infrastructure "substrates." It is the foundation for creating controllers that manage virtual machines, containers, and even supervised processes in a unified way.
The primary goal of ObjSubstrate is to provide a consistent API for common lifecycle operations, regardless of the underlying technology (e.g., Docker, LXC, Proxmox, Libvirt/KVM, Supervisor). This allows for interchangeable and technology-agnostic management of application components and services.
The class establishes a contract for what a "substrate controller" should be able to do. Any class inheriting from ObjSubstrate must implement a set of abstract methods for managing "instances." An instance can be:
The following methods must be implemented by any concrete subclass:
| Method | Description |
|---|---|
connect() |
Establishes and verifies a connection to the substrate's control plane (e.g., a daemon or API endpoint). |
list_instances() |
Returns a list of all available instances, typically including their name, ID, and status. |
get_instance(name) |
Retrieves detailed information about a single, specific instance. |
create_instance(config) |
Creates a new instance based on a provided configuration dictionary. |
start_instance(name) |
Starts a stopped or inactive instance. |
stop_instance(name, force) |
Stops a running instance, with an option to do so forcefully. |
delete_instance(name, force) |
Removes an instance permanently. |
execute_command(name, command) |
Executes a command inside a running instance (if supported by the substrate). |
| Method | Description |
|---|---|
health_check() |
Check substrate health, version, resource usage, and issues. |
get_metrics(name) |
Get real-time instance metrics (CPU, memory, disk, network). |
ObjSubstrate provides concrete implementations for common operations:
| Method | Description |
|---|---|
restart_instance(name, force) |
Restarts an instance by calling stop_instance() followed by start_instance(). Emits events. |
is_connected() |
Checks whether the client has a valid connection to the substrate. |
ensure_connected() |
Smart reconnection with periodic health checks. |
get_substrate_metrics() |
Get metrics for all substrate operations (counts, durations, errors). |
get_logs(name, lines, follow) |
Get instance logs (optional implementation). |
create_from_template(template) |
Create instance from InstanceTemplate object. |
| Method | Description |
|---|---|
set_resource_limits(name, limits) |
Set CPU/memory/disk limits for an instance. |
get_resource_limits(name) |
Query current resource limits. |
create_snapshot(name, snapshot_name) |
Create a snapshot of an instance. |
restore_snapshot(name, snapshot_name) |
Restore instance from snapshot. |
list_snapshots(name) |
List available snapshots for an instance. |
delete_snapshot(name, snapshot_name) |
Delete a specific snapshot. |
clone_instance(source, destination) |
Clone an instance to a new name. |
attach_network(name, network) |
Attach instance to a network. |
detach_network(name, network_name) |
Detach instance from a network. |
list_networks() |
List available networks. |
ObjSubstrate has been enhanced with:
class ObjSubstrate(ObjData.ObjData, CachedSubstrate, ABC):
"""
Enhanced with:
- Retry logic
- Caching
- Metrics collection
- Event system
- Resource management
- Network management
- Snapshot/clone support
- Template system
"""
def __init__(self, db_connection: Any = 0):
ObjData.ObjData.__init__(self, db_connection)
CachedSubstrate.__init__(self)
self.client = None
self.connection_timeout = 30
self.operation_timeout = 300
self.retry_attempts = 3
self.last_health_check = None
self.health_check_interval = 60
self.metrics = SubstrateMetrics()
self.event_handler = EventHandler()
@dataclass
class ResourceLimits:
cpu_cores: Optional[float] = None
memory_mb: Optional[int] = None
disk_gb: Optional[int] = None
network_mbps: Optional[int] = None
@dataclass
class ResourceUsage:
cpu_percent: float = 0.0
memory_mb: int = 0
disk_gb: float = 0.0
network_rx_mbps: float = 0.0
network_tx_mbps: float = 0.0
@dataclass
class NetworkConfig:
network_name: str
ip_address: Optional[str] = None
gateway: Optional[str] = None
dns_servers: Optional[List[str]] = None
@dataclass
class InstanceTemplate:
name: str
base_image: str
resources: Optional[ResourceLimits] = None
networks: Optional[List[NetworkConfig]] = None
environment: Optional[Dict[str, str]] = None
volumes: Optional[List[Dict[str, str]]] = None
metadata: Optional[Dict[str, str]] = None
You should not instantiate ObjSubstrate directly. Instead, create a subclass that implements the abstract methods for a specific technology.
Example Subclass Structure:
from ObjSubstrate import ObjSubstrate
class ObjMyTechnology(ObjSubstrate):
def __init__(self):
super().__init__()
self.connect()
def connect(self) -> bool:
# Implementation for connecting to MyTechnology
...
def list_instances(self) -> List[Dict[str, Any]]:
# Implementation for listing instances
...
def health_check(self) -> Dict[str, Any]:
# Implementation for health checking
...
def get_metrics(self, name: str) -> Optional[ResourceUsage]:
# Implementation for getting metrics
...
# ... implement all other abstract methods
from ObjSubstrate import InstanceTemplate, ResourceLimits, NetworkConfig
template = InstanceTemplate(
name="web-server-01",
base_image="ubuntu:22.04",
resources=ResourceLimits(cpu_cores=2.0, memory_mb=4096, disk_gb=50),
networks=[NetworkConfig(network_name="web-net", ip_address="10.0.1.10")],
environment={"ENV": "production"},
metadata={"owner": "devops", "project": "webapp"}
)
# 📌 Create from template
substrate.create_from_template(template)
from ObjSubstrate import InstanceEvent
def on_instance_started(event):
print(f"Instance {event.instance_name} started at {event.timestamp}")
substrate.event_handler.subscribe(InstanceEvent.STARTED, on_instance_started)
substrate.start_instance("my-vm") # Event will be emitted
metrics = substrate.get_metrics("my-vm")
if metrics:
print(f"CPU: {metrics.cpu_percent}%")
print(f"Memory: {metrics.memory_mb}MB")
print(f"Disk: {metrics.disk_gb}GB")
print(f"Network RX: {metrics.network_rx_mbps}MB/s")
print(f"Network TX: {metrics.network_tx_mbps}MB/s")
health = substrate.health_check()
print(f"Healthy: {health['healthy']}")
print(f"Version: {health['version']}")
if not health["healthy"]:
print(f"Issues: {health['issues']}")
from SubstrateConfig import SubstrateConfig
config = SubstrateConfig("substrate_config.yaml")
registry_host = config.get('docker', 'registry.host', '10.0.0.69')
memory_default = config.get('docker', 'default_resources.memory_mb', 512)
from ObjSubstrate import retry_on_failure
class ObjMySubstrate(ObjSubstrate):
@retry_on_failure(retries=5, delay=2.0, backoff=2.0)
def start_instance(self, name: str) -> bool:
# This will automatically retry up to 5 times
# with exponential backoff (2s, 4s, 8s, 16s, 32s)
...
| Feature | ObjSubstrate | ObjProxmox | ObjDocker | ObjLxc | ObjLibvirt | ObjSupervisor |
|---|---|---|---|---|---|---|
| Base Infrastructure | ✅ | ✅ | ⚠️ | ⚠️ | ⚠️ | ⚠️ |
| connect() | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| list_instances() | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| get_instance() | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| create_instance() | ✅ | ✅ | ⚠️ | ✅ | ✅ | ❌ |
| start_instance() | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| stop_instance() | ✅ | ✅ | ✅ | ✅ | ✅ | ✅ |
| delete_instance() | ✅ | ✅ | ✅ | ✅ | ✅ | ❌ |
| execute_command() | ✅ | ✅ | ✅ | ✅ | ❌ | N/A |
| health_check() | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ |
| get_metrics() | ✅ | ✅ | ❌ | ❌ | ❌ | ❌ |
| Retry Logic | ✅ | ✅ | ⚠️ | ⚠️ | ⚠️ | ⚠️ |
| Caching | ✅ | ✅ | ⚠️ | ⚠️ | ⚠️ | ⚠️ |
| Metrics | ✅ | ✅ | ⚠️ | ⚠️ | ⚠️ | ⚠️ |
| Events | ✅ | ✅ | ⚠️ | ⚠️ | ⚠️ | ⚠️ |
Legend:
class InstanceEvent(Enum):
CREATED = "created"
STARTED = "started"
STOPPED = "stopped"
DELETED = "deleted"
FAILED = "failed"
RESTARTED = "restarted"
SNAPSHOT_CREATED = "snapshot_created"
SNAPSHOT_RESTORED = "snapshot_restored"
CLONED = "cloned"
Configuration is managed through SubstrateConfig class and substrate_config.yaml:
docker:
registry:
host: "10.0.0.69"
port: 9443
default_resources:
memory_mb: 512
cpu_cores: 1.0
proxmox:
url: "https://proxmox.local:8006"
user: "root@pam"
verify_ssl: false
lxc:
default_profile: "default"
default_image: "ubuntu/22.04"
libvirt:
uri: "qemu:///system"
supervisor:
url: "http://localhost:9001/RPC2"
global:
enable_metrics: true
enable_events: true
enable_caching: true
Test the enhanced features:
# 📌 Test Proxmox
dev-env/bin/python factory.deploy/ObjProxmox.py list
# ❤️ Test health check
dev-env/bin/python -c "
from factory.deploy.ObjProxmox import ObjProxmox
px = ObjProxmox()
health = px.health_check()
print(health)
"
# 📊 Test metrics
dev-env/bin/python -c "
from factory.deploy.ObjProxmox import ObjProxmox
px = ObjProxmox()
metrics = px.get_metrics('vm-name')
if metrics:
print(f'CPU: {metrics.cpu_percent}%')
print(f'Memory: {metrics.memory_mb}MB')
"
factory.deploy/SubstrateConfig.py - Unified configuration managementsubstrate_config.yaml - Example configuration filefactory.deploy/ObjDocker.py - Docker substrate implementationfactory.deploy/ObjLxc.py - LXC/LXD substrate implementationfactory.deploy/ObjProxmox.py - Proxmox substrate implementation (fully enhanced)factory.deploy/ObjLibvirt.py - Libvirt/KVM substrate implementationfactory.deploy/ObjSupervisor.py - Supervisor substrate implementationTo complete the remaining substrates:
health_check() to ObjDocker, ObjLxc, ObjLibvirt, ObjSupervisorget_metrics() to all substrates