Ceph Storage Cluster Hexagonal Architecture

Hexagonal + Ceph

Ceph Storage Cluster Hexagonal Architecture Ports Adapters Pattern Domain Business Logic Storage Port Ceph Adapter S3 Local Testing Mock Production

Domain and Ports

# === Hexagonal Architecture — Storage Domain ===

# from abc import ABC, abstractmethod
# from dataclasses import dataclass
# from typing import Optional, List
# from datetime import datetime
#
# # --- Value Objects ---
# @dataclass(frozen=True)
# class StorageObject:
#     key: str
#     data: bytes
#     content_type: str
#     size: int
#     checksum: str
#     created_at: datetime
#
# @dataclass(frozen=True)
# class StorageMetadata:
#     key: str
#     size: int
#     content_type: str
#     last_modified: datetime
#
# # --- Port (Interface) ---
# class StoragePort(ABC):
#     @abstractmethod
#     def store(self, key: str, data: bytes, content_type: str) -> StorageObject:
#         pass
#
#     @abstractmethod
#     def retrieve(self, key: str) -> Optional[StorageObject]:
#         pass
#
#     @abstractmethod
#     def delete(self, key: str) -> bool:
#         pass
#
#     @abstractmethod
#     def list_objects(self, prefix: str = "") -> List[StorageMetadata]:
#         pass
#
#     @abstractmethod
#     def exists(self, key: str) -> bool:
#         pass
#
# # --- Domain Service ---
# class StorageService:
#     def __init__(self, storage: StoragePort, max_size: int = 100_000_000):
#         self._storage = storage
#         self._max_size = max_size
#
#     def upload(self, key: str, data: bytes, content_type: str) -> StorageObject:
#         if len(data) > self._max_size:
#             raise ValueError(f"File too large: {len(data)} > {self._max_size}")
#         if not key or "/" not in key:
#             raise ValueError("Key must include bucket prefix: bucket/filename")
#         return self._storage.store(key, data, content_type)
#
#     def download(self, key: str) -> Optional[StorageObject]:
#         return self._storage.retrieve(key)
#
#     def remove(self, key: str) -> bool:
#         if not self._storage.exists(key):
#             raise FileNotFoundError(f"Object not found: {key}")
#         return self._storage.delete(key)

from dataclasses import dataclass

@dataclass
class PortMethod:
    method: str
    params: str
    returns: str
    description: str

methods = [
    PortMethod("store", "key, data, content_type", "StorageObject", "Upload object to storage"),
    PortMethod("retrieve", "key", "Optional[StorageObject]", "Download object by key"),
    PortMethod("delete", "key", "bool", "Remove object from storage"),
    PortMethod("list_objects", "prefix", "List[StorageMetadata]", "List objects by prefix"),
    PortMethod("exists", "key", "bool", "Check if object exists"),
    PortMethod("get_metadata", "key", "StorageMetadata", "Get object metadata only"),
]

print("=== StoragePort Interface ===")
for m in methods:
    print(f"  {m.method}({m.params}) -> {m.returns}")
    print(f"    Description: {m.description}")

Adapters

# === Storage Adapters ===

# Ceph RADOS Adapter
# import rados
# class CephRadosAdapter(StoragePort):
#     def __init__(self, conf_file="/etc/ceph/ceph.conf", pool="data"):
#         self.cluster = rados.Rados(conffile=conf_file)
#         self.cluster.connect()
#         self.ioctx = self.cluster.open_ioctx(pool)
#
#     def store(self, key, data, content_type):
#         self.ioctx.write_full(key, data)
#         self.ioctx.set_xattr(key, "content-type", content_type.encode())
#         return StorageObject(key=key, data=data, ...)
#
#     def retrieve(self, key):
#         try:
#             size, _ = self.ioctx.stat(key)
#             data = self.ioctx.read(key, size)
#             ct = self.ioctx.get_xattr(key, "content-type").decode()
#             return StorageObject(key=key, data=data, content_type=ct, ...)
#         except rados.ObjectNotFound:
#             return None

# S3 Adapter (Ceph RGW or AWS S3)
# import boto3
# class S3Adapter(StoragePort):
#     def __init__(self, endpoint, access_key, secret_key, bucket):
#         self.s3 = boto3.client("s3",
#             endpoint_url=endpoint,
#             aws_access_key_id=access_key,
#             aws_secret_access_key=secret_key)
#         self.bucket = bucket
#
#     def store(self, key, data, content_type):
#         self.s3.put_object(Bucket=self.bucket, Key=key,
#             Body=data, ContentType=content_type)
#         return StorageObject(...)

# In-Memory Adapter (for testing)
# class InMemoryAdapter(StoragePort):
#     def __init__(self):
#         self._store = {}
#
#     def store(self, key, data, content_type):
#         obj = StorageObject(key=key, data=data, content_type=content_type, ...)
#         self._store[key] = obj
#         return obj
#
#     def retrieve(self, key):
#         return self._store.get(key)

@dataclass
class AdapterComparison:
    adapter: str
    backend: str
    performance: str
    use_case: str
    config: str

adapters = [
    AdapterComparison("CephRadosAdapter", "Ceph RADOS", "Very high (native)", "Production high-perf", "ceph.conf + pool name"),
    AdapterComparison("CephS3Adapter", "Ceph RGW (S3)", "High", "Production S3-compatible", "endpoint + keys + bucket"),
    AdapterComparison("AWSS3Adapter", "AWS S3", "High (network)", "Cloud production", "region + keys + bucket"),
    AdapterComparison("LocalAdapter", "Local filesystem", "Fast (disk)", "Development", "base_path directory"),
    AdapterComparison("InMemoryAdapter", "Python dict", "Very fast", "Unit testing", "None"),
    AdapterComparison("MinioAdapter", "MinIO S3", "High", "Self-hosted S3", "endpoint + keys + bucket"),
]

print("\n=== Adapter Comparison ===")
for a in adapters:
    print(f"  [{a.adapter}] Backend: {a.backend}")
    print(f"    Perf: {a.performance} | Use: {a.use_case}")
    print(f"    Config: {a.config}")

Testing Strategy

# === Testing Hexagonal Storage ===

# Unit Test — Domain with Mock
# def test_upload_validates_size():
#     storage = InMemoryAdapter()
#     service = StorageService(storage, max_size=1000)
#     with pytest.raises(ValueError):
#         service.upload("bucket/file.txt", b"x" * 1001, "text/plain")
#
# def test_upload_validates_key():
#     storage = InMemoryAdapter()
#     service = StorageService(storage)
#     with pytest.raises(ValueError):
#         service.upload("no-bucket", b"data", "text/plain")
#
# def test_upload_stores_object():
#     storage = InMemoryAdapter()
#     service = StorageService(storage)
#     result = service.upload("bucket/test.txt", b"hello", "text/plain")
#     assert result.key == "bucket/test.txt"
#     assert storage.exists("bucket/test.txt")

# Integration Test — Real Ceph
# @pytest.fixture
# def ceph_adapter():
#     adapter = CephS3Adapter(
#         endpoint="http://ceph-rgw:7480",
#         access_key="test", secret_key="test", bucket="test-bucket")
#     yield adapter
#     # Cleanup
#
# def test_ceph_store_retrieve(ceph_adapter):
#     ceph_adapter.store("test/file.txt", b"hello", "text/plain")
#     obj = ceph_adapter.retrieve("test/file.txt")
#     assert obj.data == b"hello"

@dataclass
class TestType:
    test_type: str
    adapter_used: str
    speed: str
    coverage: str
    ci_friendly: bool

test_types = [
    TestType("Unit Test (Domain)", "InMemoryAdapter", "< 1ms", "Business logic", True),
    TestType("Unit Test (Adapter)", "InMemoryAdapter", "< 1ms", "Adapter logic", True),
    TestType("Contract Test", "All Adapters", "Varies", "Port compliance", True),
    TestType("Integration Test", "CephS3Adapter", "100-500ms", "Real Ceph access", False),
    TestType("E2E Test", "Production Adapter", "1-5s", "Full flow", False),
    TestType("Performance Test", "CephRadosAdapter", "Minutes", "Throughput latency", False),
]

print("Testing Strategy:")
for t in test_types:
    ci = "CI-friendly" if t.ci_friendly else "Requires infra"
    print(f"  [{t.test_type}] Adapter: {t.adapter_used}")
    print(f"    Speed: {t.speed} | Coverage: {t.coverage} | {ci}")

เคล็ดลับ

• Port First: ออกแบบ Port ก่อน แล้วค่อยสร้าง Adapter
• InMemory: ใช้ InMemoryAdapter สำหรับ Unit Test เร็วมาก
• Config: เปลี่ยน Adapter ผ่าน Config ไม่แก้ Code
• Contract: ทดสอบ Contract ทุก Adapter ต้อง Pass เหมือนกัน
• Domain Pure: Domain ไม่ Import External Library เด็ดขาด

Hexagonal Architecture คืออะไร

Ports Adapters Pattern แยก Domain จาก External Port Interface Adapter Implementation เปลี่ยน Adapter ไม่กระทบ Domain ทดสอบง่าย Mock

สรุป

Ceph Storage Cluster Hexagonal Architecture Ports Adapters Domain StoragePort CephAdapter S3Adapter InMemoryAdapter Testing Contract Integration Production