feat(docker): add support check for Docker's --storage-opt option

- Introduced a static method to verify if the Docker storage driver supports the --storage-opt size= option.
- Enhanced resource argument handling in DockerEnvironment to conditionally include storage options based on the support check.
- Added caching for the support check result to optimize performance across instances.
This commit is contained in:
teknium1
2026-02-26 01:15:56 -08:00
parent 8bd82119be
commit 240f33a06f

View File

@@ -31,6 +31,9 @@ _SECURITY_ARGS = [
]
_storage_opt_ok: Optional[bool] = None # cached result across instances
class DockerEnvironment(BaseEnvironment):
"""Hardened Docker container execution with resource limits and persistence.
@@ -70,7 +73,7 @@ class DockerEnvironment(BaseEnvironment):
resource_args.extend(["--cpus", str(cpu)])
if memory > 0:
resource_args.extend(["--memory", f"{memory}m"])
if disk > 0 and sys.platform != "darwin":
if disk > 0 and sys.platform != "darwin" and self._storage_opt_supported():
resource_args.extend(["--storage-opt", f"size={disk}m"])
if not network:
resource_args.append("--network=none")
@@ -110,6 +113,45 @@ class DockerEnvironment(BaseEnvironment):
)
self._container_id = self._inner.container_id
@staticmethod
def _storage_opt_supported() -> bool:
"""Check if Docker's storage driver supports --storage-opt size=.
Only overlay2 on XFS with pquota supports per-container disk quotas.
Ubuntu (and most distros) default to ext4, where this flag errors out.
"""
global _storage_opt_ok
if _storage_opt_ok is not None:
return _storage_opt_ok
try:
result = subprocess.run(
["docker", "info", "--format", "{{.Driver}}"],
capture_output=True, text=True, timeout=10,
)
driver = result.stdout.strip().lower()
if driver != "overlay2":
_storage_opt_ok = False
return False
# overlay2 only supports storage-opt on XFS with pquota.
# Probe by attempting a dry-ish run — the fastest reliable check.
probe = subprocess.run(
["docker", "create", "--storage-opt", "size=1m", "hello-world"],
capture_output=True, text=True, timeout=15,
)
if probe.returncode == 0:
# Clean up the created container
container_id = probe.stdout.strip()
if container_id:
subprocess.run(["docker", "rm", container_id],
capture_output=True, timeout=5)
_storage_opt_ok = True
else:
_storage_opt_ok = False
except Exception:
_storage_opt_ok = False
logger.debug("Docker --storage-opt support: %s", _storage_opt_ok)
return _storage_opt_ok
def execute(self, command: str, cwd: str = "", *,
timeout: int | None = None,
stdin_data: str | None = None) -> dict: