import os
import subprocess
import shutil
import resource
import time
import psutil
import yaml
from collections import defaultdict
class Container:
def __init__(self, name, memory_limit_mb, cpu_limit=None):
self.name = name
self.memory_limit_mb = memory_limit_mb
self.cpu_limit = cpu_limit
self.container_dir = f"./{self.name}"
self.running = False
self.logs = []
self.create_container_directory()
def create_container_directory(self):
os.makedirs(self.container_dir, exist_ok=True)
print(f"Container '{self.name}' created.")
def allocate_memory(self):
resource.setrlimit(resource.RLIMIT_AS, (self.memory_limit_mb * 1024 * 1024,
resource.RLIM_INFINITY))
if self.cpu_limit:
print(f"CPU limit set to {self.cpu_limit} cores.")
# Placeholder for actual CPU limiting implementation
print(f"Allocated {self.memory_limit_mb}MB memory to the container.")
def write_code(self, code_content):
code_file_path = os.path.join(self.container_dir, "code.py")
with open(code_file_path, "w") as code_file:
code_file.write(code_content)
print(f"Code written to {code_file_path}.")
def run(self):
if self.running:
print(f"Container '{self.name}' is already running.")
return
code_file = os.path.join(self.container_dir, "code.py")
if os.path.exists(code_file):
print(f"Running code inside container: {self.name}")
result = subprocess.run(["python3", code_file], capture_output=True, text=True)
self.logs.append(result.stdout)
print("Output from container:")
print(result.stdout)
self.running = True
else:
print("No code file found to execute.")
def stop(self):
print(f"Stopping container '{self.name}'...")
self.running = False
def status(self):
return f"Container '{self.name}' is {'running' if self.running else 'stopped'}."
def cleanup(self):
shutil.rmtree(self.container_dir)
print(f"Container '{self.name}' deleted.")
def get_logs(self):
return "\n".join(self.logs)
class Network:
def __init__(self):
self.container_links = defaultdict(list)
def connect(self, container1, container2):
self.container_links[container1].append(container2)
self.container_links[container2].append(container1)
self.container_links[container2].append(container1)
print(f"Connected {container1} and {container2}.")
def show_connections(self):
for container, connections in self.container_links.items():
print(f"{container} is connected to: {', '.join(connections)}")
class Volume:
def __init__(self, name):
self.name = name
self.volume_dir = f"./volumes/{self.name}"
os.makedirs(self.volume_dir, exist_ok=True)
def write_data(self, filename, data):
with open(os.path.join(self.volume_dir, filename), 'w') as f:
f.write(data)
def read_data(self, filename):
with open(os.path.join(self.volume_dir, filename), 'r') as f:
return f.read()
class ContainerManager:
def __init__(self):
self.containers = {}
self.network = Network()
self.volumes = {}
def create_container(self, name, memory_limit_mb, cpu_limit=None):
if name in self.containers:
print(f"Container '{name}' already exists.")
return
container = Container(name, memory_limit_mb, cpu_limit)
self.containers[name] = container
return container
def run_container(self, name):
if name in self.containers:
self.containers[name].run()
else:
print(f"Container '{name}' not found.")
def stop_container(self, name):
if name in self.containers:
self.containers[name].stop()
else:
print(f"Container '{name}' not found.")
def show_status(self):
for name, container in self.containers.items():
print(container.status())
def show_logs(self, name):
if name in self.containers:
logs = self.containers[name].get_logs()
print(f"Logs for container '{name}':\n{logs}")
else:
print(f"Container '{name}' not found.")
def create_volume(self, name):
if name in self.volumes:
print(f"Volume '{name}' already exists.")
return
volume = Volume(name)
self.volumes[name] = volume
print(f"Volume '{name}' created.")
def connect_containers(self, name1, name2):
if name1 in self.containers and name2 in self.containers:
self.network.connect(name1, name2)
else:
print("One or both containers not found.")
def load_configuration(self, config_file):
with open(config_file, 'r') as f:
config = yaml.safe_load(f)
for container in config.get('containers', []):
self.create_container(container['name'], container['memory'], container.get('cpu'))
# Code execution could be managed here
def generate_code():
return """
print("Hello from the container!")
for i in range(5):
print(f"Line {i + 1}: This is a line from the container code.")
"""
if __name__ == "__main__":
manager = ContainerManager()
# Tạo container với giới hạn bộ nhớ 256MB và 1 core CPU
container_name = "test_container"
container_memory_limit = 256 # MB
container_cpu_limit = 1 # core
container = manager.create_container(container_name, container_memory_limit,
container_cpu_limit)
# Cấp phát bộ nhớ cho container
container.allocate_memory()
# Ghi mã vào container
code = generate_code()
container.write_code(code)
# Chạy mã trong container
manager.run_container(container_name)
# Kiểm tra trạng thái container
manager.show_status()
# Hiển thị nhật ký container
manager.show_logs(container_name)
# Dừng container
manager.stop_container(container_name)
# Tạo volume
manager.create_volume("data_volume")
# Kết nối container
manager.create_container("another_container", 256, 1)
manager.connect_containers(container_name, "another_container")
# Hiển thị kết nối mạng
manager.network.show_connections()
# Dọn dẹp container
container.cleanup()
