Virtualization & Containers — Hypervisors, Docker & Kubernetes Pod Isolation

class VirtualMachine:
    def __init__(self, name, vcpus=1, memory_mb=1024, disk_gb=20):
        self.name = name
        self.vcpus = vcpus
        self.memory_mb = memory_mb
        self.disk_gb = disk_gb
        self.state = 'powered_off'

    def power_on(self, hypervisor_type):
        if hypervisor_type == 'Type 1':
            print(f'[VM] {self.name}: Booting via hardware passthrough')
        else:
            print(f'[VM] {self.name}: Booting via host OS emulation')
        self.state = 'running'
        print(f'[VM] {self.name}: Assigned {self.vcpus} vCPUs, '
              f'{self.memory_mb}MB RAM, {self.disk_gb}GB disk')

    def __repr__(self):
        return f'VM({self.name}, {self.vcpus}vCPU, {self.memory_mb}MB)'

class Hypervisor:
    def __init__(self, hv_type='Type 1'):
        self.hv_type = hv_type
        self.vms = []

    def create_vm(self, name, **kwargs):
        vm = VirtualMachine(name, **kwargs)
        self.vms.append(vm)
        return vm

    def start_all(self):
        for vm in self.vms:
            vm.power_on(self.hv_type)

    def resources_used(self):
        total_vcpus = sum(vm.vcpus for vm in self.vms)
        total_mem = sum(vm.memory_mb for vm in self.vms)
        return total_vcpus, total_mem

# Type 1: ESXi on bare metal
esxi = Hypervisor('Type 1')
web_vm = esxi.create_vm('web-server', vcpus=2, memory_mb=4096)
db_vm = esxi.create_vm('database', vcpus=4, memory_mb=8192)
esxi.start_all()
cpu, mem = esxi.resources_used()
print(f'\nTotal: {cpu} vCPUs, {mem}MB RAM allocated')

import os

class ContainerNamespace:
    def __init__(self, name):
        self.name = name
        self.pids = set()
        self.network = {
            'interfaces': ['lo', 'eth0'],
            'ip': None,
            'routes': [],
        }
        self.mounts = {'/': 'rootfs', '/proc': 'proc'}
        self.hostname = name

    def add_process(self, pid):
        self.pids.add(pid)

    def __repr__(self):
        return (f'Namespace[{self.name}]: '
                f'pids={len(self.pids)}, '
                f'ip={self.network["ip"]}, '
                f'hostname={self.hostname}')

# Simulate creating namespaced containers
containers = [
    ContainerNamespace('web-app'),
    ContainerNamespace('api-server'),
    ContainerNamespace('background-worker'),
]

for i, ctn in enumerate(containers):
    ctn.add_process(1000 + i)
    ctn.network['ip'] = f'10.0.{i}.2'
    print(ctn)

class CGroup:
    def __init__(self, name, cpu_shares=1024, memory_limit_mb=512):
        self.name = name
        self.cpu_shares = cpu_shares
        self.memory_limit_mb = memory_limit_mb
        self.current_memory_mb = 0
        self.cpu_usage_percent = 0

    def set_limit(self, resource, limit):
        if resource == 'memory':
            self.memory_limit_mb = limit
        elif resource == 'cpu':
            self.cpu_shares = limit

    def account_memory(self, mb):
        self.current_memory_mb += mb
        if self.current_memory_mb > self.memory_limit_mb:
            print(f'[OOM] Container {self.name} exceeded '
                  f'memory limit ({self.memory_limit_mb}MB)!')
            return False
        return True

    def __repr__(self):
        return (f'cgroup[{self.name}]: '
                f'cpu={self.cpu_shares}, '
                f'mem={self.current_memory_mb}/{self.memory_limit_mb}MB')

# Simulate cgroup limits
web_cgroup = CGroup('web', cpu_shares=2048, memory_limit_mb=1024)
for i in range(5):
    ok = web_cgroup.account_memory(256)
    print(f'Allocated 256MB: {ok} — {web_cgroup.current_memory_mb}MB used')
    if not ok:
        break

import time

class OCIProcess:
    """Simulate an OCI-compliant container process"""

    def __init__(self, pid, rootfs, command):
        self.pid = pid
        self.rootfs = rootfs
        self.command = command
        self.state = 'created'

    def start(self):
        print(f'[runc] Creating container (pid={self.pid})')
        print(f'[runc] rootfs: {self.rootfs}')
        print(f'[runc] Namespaces: pid,net,mnt,uts,ipc')
        print(f'[runc] cgroups: cpu=1024, memory=512MB')
        self.state = 'running'
        print(f'[runc] Started PID {self.pid}: {self.command}')
        return True

    def stop(self, timeout=10):
        print(f'[runc] Sending SIGTERM to PID {self.pid}')
        self.state = 'stopped'
        print(f'[runc] Container stopped')

class Containerd:
    def __init__(self):
        self.containers = {}

    def create_container(self, image, command, container_id):
        print(f'[containerd] Pulling image: {image}')
        time.sleep(0.1)
        print(f'[containerd] Creating container instance: {container_id}')
        pid = hash(container_id) & 0xFFFF
        proc = OCIProcess(pid, f'/var/lib/containerd/{container_id}', command)
        self.containers[container_id] = proc
        return proc

class DockerEngine:
    def __init__(self):
        self.containerd = Containerd()

    def run(self, image, command, name=None):
        container_id = name or f'container_{int(time.time())}'
        print(f'[docker] docker run {image} {command}')
        proc = self.containerd.create_container(image, command, container_id)
        proc.start()
        return proc

engine = DockerEngine()
container = engine.run('nginx:alpine', 'nginx -g "daemon off;"', name='web')
print()
container.stop()

class Pod:
    def __init__(self, name, namespace='default'):
        self.name = name
        self.namespace = namespace
        self.containers = []
        self.pod_ip = None
        self.status = 'Pending'

    def add_container(self, name, image, ports=None, resources=None):
        container = {
            'name': name,
            'image': image,
            'ports': ports or [],
            'resources': resources or {'cpu': '100m', 'memory': '128Mi'},
        }
        self.containers.append(container)
        return container

    def assign_ip(self, cni_plugin):
        ip = cni_plugin.allocate_ip(self)
        self.pod_ip = ip
        return ip

    def start(self):
        print(f'[K8s] Pod {self.name}/{self.namespace} scheduled to node-1')
        print(f'[K8s] Pod IP: {self.pod_ip}')
        for c in self.containers:
            print(f'[K8s]   Container {c["name"]}: {c["image"]}')
            print(f'[K8s]   Ports: {c["ports"]}')
            print(f'[K8s]   Limits: {c["resources"]}')
        self.status = 'Running'
        print(f'[K8s] Pod status: {self.status}')

class CNIPlugin:
    def __init__(self):
        self.allocated = set()

    def allocate_ip(self, pod):
        base = 10
        while base in self.allocated:
            base += 1
        ip = f'10.42.{base // 256}.{base % 256}'
        self.allocated.add(base)
        return ip

cni = CNIPlugin()
pod = Pod('web-app', 'production')
pod.add_container('nginx', 'nginx:1.25',
                  ports=[{'containerPort': 80}],
                  resources={'cpu': '500m', 'memory': '256Mi'})
pod.add_container('sidecar', 'fluentd:v1.16',
                  resources={'cpu': '100m', 'memory': '128Mi'})
pod.assign_ip(cni)
pod.start()