diff --git a/sebs/cache.py b/sebs/cache.py index 098aa6eac..ea2f4b47b 100644 --- a/sebs/cache.py +++ b/sebs/cache.py @@ -158,16 +158,25 @@ class Cache(LoggingBase): _lock_registry_guard = threading.Lock() _lock_registry: Dict[str, threading.RLock] = {} - def __init__(self, cache_dir: str, docker_client: docker.DockerClient) -> None: + # Cloud platforms that own a per-cloud cache file (`.json`). + SUPPORTED_CLOUDS: List[str] = ["azure", "aws", "gcp", "openwhisk", "local"] + + def __init__(self, cache_dir: str, docker_client: Optional[docker.DockerClient] = None) -> None: """Initialize the Cache with directory and Docker client. Sets up the cache directory structure and loads existing configurations. Creates the cache directory if it doesn't exist, otherwise loads existing cached configurations. + The Docker client is optional so that read-only consumers (for example, + the cache inspection command) can open a cache without a running Docker + daemon. It is only required for operations that copy container images, + such as `add_code_package`. + Args: cache_dir (str): Path to the cache directory. - docker_client (docker.DockerClient): Docker client for container operations. + docker_client (Optional[docker.DockerClient]): Docker client for + container operations. May be None for read-only usage. """ super().__init__() self.cached_config: Dict[str, Any] = {} @@ -199,6 +208,24 @@ def _cache_dir_lock(cls, cache_dir: str) -> threading.RLock: cls._lock_registry[cache_dir] = threading.RLock() return cls._lock_registry[cache_dir] + def _require_docker_client(self) -> docker.DockerClient: + """Return the Docker client, raising if it is unavailable. + + The Docker client is optional to allow read-only consumers to open a + cache without a running Docker daemon. Operations that copy container + images require it, so they call this helper to fail loudly (and, unlike + an ``assert``, reliably under ``python -O``) when it is missing. + + Returns: + docker.DockerClient: The configured Docker client. + + Raises: + RuntimeError: If no Docker client was provided to the cache. + """ + if self.docker_client is None: + raise RuntimeError("A Docker client is required to cache container images.") + return self.docker_client + @staticmethod def _write_json_atomic(path: str, data: Any) -> None: """Atomically replace a JSON file after fully writing it to a temp file.""" @@ -240,7 +267,7 @@ def load_config(self) -> None: the cache directory and loads them into memory. """ with self._lock: - for cloud in ["azure", "aws", "gcp", "openwhisk", "local"]: + for cloud in self.SUPPORTED_CLOUDS: cloud_config_file = os.path.join(self.cache_dir, "{}.json".format(cloud)) if os.path.exists(cloud_config_file): with open(cloud_config_file, "r") as f: @@ -288,7 +315,7 @@ def shutdown(self) -> None: """ if self.config_updated: with self._lock: - for cloud in ["azure", "aws", "gcp", "openwhisk", "local"]: + for cloud in self.SUPPORTED_CLOUDS: if cloud in self.cached_config: cloud_config_file = os.path.join(self.cache_dir, "{}.json".format(cloud)) self.logging.info("Update cached config {}".format(cloud_config_file)) @@ -498,6 +525,169 @@ def get_all_functions(self, deployment: str) -> Dict[str, Any]: return result + def get_deployed_benchmarks(self, deployment: Optional[str] = None) -> List[Dict[str, Any]]: + """Collect every deployed benchmark entry recorded in the cache. + + Walks all per-benchmark `config.json` files and flattens them into a + list of rows describing what is deployed where. Each row corresponds to + a single (benchmark, platform, language) combination and reports the + deployed functions, packaging type, and cached storage/nosql tables. + + This is a purely read-only view intended for inspection tooling; it does + not query any cloud provider and does not require a Docker client. + + Args: + deployment (Optional[str]): Restrict the result to a single platform + (e.g. 'aws'). When None, entries for all platforms are returned. + + Returns: + List[Dict[str, Any]]: One dictionary per (benchmark, platform, + language) combination. Keys: 'benchmark', 'platform', 'language', + 'packaging', 'functions' (list of function names), + 'function_details' (per-function name/hash/triggers), 'triggers' + (sorted list of trigger types across all functions), 'storage' + (list of bucket names), and 'nosql' (list of table names). + """ + + def summarize_functions(functions: Dict[str, Any]) -> List[Dict[str, Any]]: + """Reduce a cached `functions` mapping to per-function summaries. + + Args: + functions (Dict[str, Any]): Mapping of function name to the + cached function configuration. + + Returns: + List[Dict[str, Any]]: Summaries with 'name', 'hash', and the + sorted list of trigger types for each function. + """ + summaries: List[Dict[str, Any]] = [] + for name, cfg in functions.items(): + triggers = cfg.get("triggers", []) if isinstance(cfg, dict) else [] + trigger_types = sorted( + {str(t["type"]) for t in triggers if isinstance(t, dict) and t.get("type")} + ) + summaries.append( + { + "name": name, + "hash": cfg.get("hash") if isinstance(cfg, dict) else None, + "triggers": trigger_types, + } + ) + return summaries + + rows: List[Dict[str, Any]] = [] + + if not os.path.exists(self.cache_dir): + return rows + + with self._lock: + for entry in sorted(os.listdir(self.cache_dir)): + config_path = os.path.join(self.cache_dir, entry, "config.json") + if not os.path.exists(config_path): + continue + + with open(config_path, "r") as fp: + config = json.load(fp) + + for platform, dep_cfg in config.items(): + if deployment is not None and platform != deployment: + continue + if not isinstance(dep_cfg, dict): + continue + + for language, lang_cfg in dep_cfg.items(): + # Skip resource-level keys (storage/nosql live under the + # deployment too) - languages always map to a dict with + # a 'functions'/'code_package'/'containers' shape. + if not isinstance(lang_cfg, dict): + continue + if language in ("storage", "nosql"): + continue + + functions = lang_cfg.get("functions") or {} + if "containers" in lang_cfg and lang_cfg["containers"]: + packaging = "container" + elif "code_package" in lang_cfg and lang_cfg["code_package"]: + packaging = "code_package" + else: + packaging = "unknown" + + func_summaries = summarize_functions(functions) + all_triggers = sorted({t for f in func_summaries for t in f["triggers"]}) + + storage_cfg = dep_cfg.get("storage") or {} + nosql_cfg = dep_cfg.get("nosql") or {} + + rows.append( + { + "benchmark": entry, + "platform": platform, + "language": language, + "packaging": packaging, + "functions": [f["name"] for f in func_summaries], + "function_details": func_summaries, + "triggers": all_triggers, + "storage": sorted(storage_cfg.keys()), + "nosql": sorted(nosql_cfg.keys()), + } + ) + + return rows + + def get_allocated_resources( + self, deployment: Optional[str] = None + ) -> Dict[str, Dict[str, Any]]: + """Collect the allocated cloud resources recorded per platform. + + Reads the per-cloud cache files (`aws.json`, `local.json`, ...) and + extracts the resource block that SeBS persists for reuse: the + `resources_id` namespace, allocated storage buckets, and (for local + deployments) allocated container ports. + + Like `get_deployed_benchmarks`, this is a read-only view that does not + contact any cloud provider. + + Args: + deployment (Optional[str]): Restrict the result to a single platform. + When None, all platforms present in the cache are returned. + + Returns: + Dict[str, Dict[str, Any]]: Mapping of platform name to its resource + summary. Each summary contains 'resources_id' (Optional[str]), + 'region' (Optional[str]), 'storage_buckets' (Dict[str, Any]), and + 'allocated_ports' (List[int]). + """ + result: Dict[str, Dict[str, Any]] = {} + + clouds = [deployment] if deployment is not None else self.SUPPORTED_CLOUDS + + with self._lock: + for cloud in clouds: + cloud_cfg = self.cached_config.get(cloud) + if not isinstance(cloud_cfg, dict): + continue + + resources = cloud_cfg.get("resources") + if not isinstance(resources, dict): + resources = {} + + buckets = resources.get("storage_buckets") + if not isinstance(buckets, dict): + buckets = {} + + ports = resources.get("allocated_ports") + if not isinstance(ports, list): + ports = [] + + result[cloud] = { + "resources_id": resources.get("resources_id"), + "region": cloud_cfg.get("region"), + "storage_buckets": buckets, + "allocated_ports": ports, + } + + return result + def get_storage_config(self, deployment: str, benchmark: str) -> Optional[Dict[str, Any]]: """Access cached storage configuration of a benchmark. @@ -783,6 +973,11 @@ def add_code_package( RuntimeError: If cached application already exists for the deployment. """ with self._lock: + # Fail before mutating the cache (creating directories, copying + # code) when a container deployment has no Docker client available. + if code_package.system_variant.is_container: + self._require_docker_client() + benchmark_dir = os.path.join(self.cache_dir, code_package.benchmark) os.makedirs(benchmark_dir, exist_ok=True) @@ -829,7 +1024,7 @@ def add_code_package( "functions": {}, } if code_package.system_variant.is_container: - image = self.docker_client.images.get(code_package.container_uri) + image = self._require_docker_client().images.get(code_package.container_uri) language_config["image-uri"] = code_package.container_uri language_config["image-id"] = image.id @@ -891,6 +1086,11 @@ def update_code_package( code_package (Benchmark): The benchmark code package to update. """ with self._lock: + # Fail before mutating the cache (deleting/copying code) when a + # container deployment has no Docker client available. + if code_package.system_variant.is_container: + self._require_docker_client() + benchmark_dir = os.path.join(self.cache_dir, code_package.benchmark) # Check if cache directory for this deployment exist @@ -952,7 +1152,7 @@ def update_code_package( cached_config["size"] = code_package.code_size if code_package.system_variant.is_container: - image = self.docker_client.images.get(code_package.container_uri) + image = self._require_docker_client().images.get(code_package.container_uri) cached_config["image-id"] = image.id cached_config["image-uri"] = code_package.container_uri diff --git a/sebs/cli.py b/sebs/cli.py index ec7bab57e..4e28c284b 100755 --- a/sebs/cli.py +++ b/sebs/cli.py @@ -20,6 +20,7 @@ import sebs from sebs import SeBS +from sebs.cache import Cache from sebs.sebs_types import Storage as StorageTypes from sebs.sebs_types import NoSQLStorage as NoSQLStorageTypes from sebs.regression import regression_suite @@ -844,6 +845,108 @@ def resources(): pass +@resources.command("inspect") +@click.option( + "--cache", + default=os.path.join(os.path.curdir, "cache"), + type=click.Path(exists=True, file_okay=False, readable=True), + help="Location of the experiments cache to inspect.", +) +@click.option( + "--deployment", + default=None, + type=click.Choice(["azure", "aws", "gcp", "local", "openwhisk"]), + help="Restrict the view to a single platform.", +) +@click.option( + "--output", + "output_format", + default="table", + type=click.Choice(["table", "json"]), + help="Render a rich table (default) or emit machine-readable JSON.", +) +def resources_inspect(cache, deployment, output_format): + """Inspect the local SeBS cache and show what is deployed. + + This is a read-only view built directly from the on-disk cache. It does not + contact any cloud provider and does not require credentials or a running + Docker daemon. It reports which benchmarks are deployed (platform, language, + packaging, functions, triggers) and what resources are allocated per + platform (resources_id, storage buckets, and locally allocated ports). + """ + cache_client = Cache(cache) + + benchmarks = cache_client.get_deployed_benchmarks(deployment) + allocated = cache_client.get_allocated_resources(deployment) + + if output_format == "json": + click.echo( + json.dumps( + {"deployed_benchmarks": benchmarks, "allocated_resources": allocated}, + indent=2, + ) + ) + return + + from rich.console import Console + from rich.table import Table + + console = Console() + + bench_table = Table(title="Deployed benchmarks", show_lines=False) + bench_table.add_column("Benchmark", style="cyan", no_wrap=True) + bench_table.add_column("Platform", style="green") + bench_table.add_column("Language") + bench_table.add_column("Packaging") + bench_table.add_column("Functions", justify="right") + bench_table.add_column("Triggers") + + for row in benchmarks: + bench_table.add_row( + row["benchmark"], + row["platform"], + row["language"], + row["packaging"], + str(len(row["functions"])), + ", ".join(row["triggers"]) or "-", + ) + + if benchmarks: + console.print(bench_table) + else: + console.print("[yellow]No deployed benchmarks found in cache.[/yellow]") + + res_table = Table(title="Allocated resources", show_lines=False) + res_table.add_column("Platform", style="green", no_wrap=True) + res_table.add_column("Resources ID", style="cyan") + res_table.add_column("Region") + res_table.add_column("Storage buckets", justify="right") + res_table.add_column("Allocated ports") + + has_resources = False + for platform, res in allocated.items(): + if ( + not res.get("resources_id") + and not res.get("storage_buckets") + and not res.get("allocated_ports") + ): + continue + has_resources = True + ports = res.get("allocated_ports") or [] + res_table.add_row( + platform, + res.get("resources_id") or "-", + res.get("region") or "-", + str(len(res.get("storage_buckets") or {})), + ", ".join(str(p) for p in ports) or "-", + ) + + if has_resources: + console.print(res_table) + else: + console.print("[yellow]No allocated resources found in cache.[/yellow]") + + @resources.command("list") @click.argument("resource", type=click.Choice(["buckets", "resource-groups"])) @common_params diff --git a/tests/test_cache_inspect.py b/tests/test_cache_inspect.py new file mode 100644 index 000000000..09dd3a137 --- /dev/null +++ b/tests/test_cache_inspect.py @@ -0,0 +1,138 @@ +#!/usr/bin/env python3 +# Copyright 2020-2025 ETH Zurich and the SeBS authors. All rights reserved. +"""Unit tests for the read-only cache inspection API. + +These tests build a synthetic cache directory on disk (per-cloud `.json` +files plus per-benchmark `config.json` files) and verify that +`Cache.get_deployed_benchmarks` and `Cache.get_allocated_resources` flatten the +model correctly. They run fully offline and require no Docker daemon. +""" + +import json +import os +import tempfile +import unittest + +from sebs.cache import Cache + + +def _write_json(path: str, data: dict) -> None: + """Write a dictionary as JSON to a path, creating parent directories. + + Args: + path: Destination file path. + data: JSON-serializable dictionary to write. + """ + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w") as fp: + json.dump(data, fp) + + +class CacheInspectTests(unittest.TestCase): + """Verify the read-only inspection helpers on a synthetic cache.""" + + def setUp(self) -> None: + """Create a temporary cache directory with representative content.""" + self._tmp = tempfile.TemporaryDirectory() + self.cache_dir = self._tmp.name + + # Per-cloud resource files. + _write_json( + os.path.join(self.cache_dir, "aws.json"), + { + "region": "us-east-1", + "resources": { + "resources_id": "abc123", + "storage_buckets": { + "benchmarks": "sebs-benchmarks-abc123", + "experiments": "sebs-experiments-abc123", + }, + }, + }, + ) + _write_json( + os.path.join(self.cache_dir, "local.json"), + { + "resources": { + "resources_id": "local42", + "allocated_ports": [9000, 9001], + }, + }, + ) + + # Per-benchmark config with a deployed AWS python function. + _write_json( + os.path.join(self.cache_dir, "110.dynamic-html", "config.json"), + { + "aws": { + "python": { + "code_package": {"3.9": {"x64": {"location": "code"}}}, + "containers": {}, + "functions": { + "sebs-abc123-110.dynamic-html-python-3.9-x64": { + "name": "sebs-abc123-110.dynamic-html-python-3.9-x64", + "hash": "deadbeef", + "triggers": [{"type": "library"}, {"type": "http"}], + } + }, + }, + "storage": {"sebs-benchmarks-abc123": {}}, + } + }, + ) + + def tearDown(self) -> None: + """Remove the temporary cache directory.""" + self._tmp.cleanup() + + def test_deployed_benchmarks(self) -> None: + """Deployed benchmark rows expose platform, packaging, and triggers.""" + cache = Cache(self.cache_dir) + rows = cache.get_deployed_benchmarks() + + self.assertEqual(len(rows), 1) + row = rows[0] + self.assertEqual(row["benchmark"], "110.dynamic-html") + self.assertEqual(row["platform"], "aws") + self.assertEqual(row["language"], "python") + self.assertEqual(row["packaging"], "code_package") + self.assertEqual(len(row["functions"]), 1) + self.assertEqual(row["triggers"], ["http", "library"]) + self.assertEqual(row["storage"], ["sebs-benchmarks-abc123"]) + + def test_deployed_benchmarks_filter(self) -> None: + """Filtering by a platform with no entries yields an empty list.""" + cache = Cache(self.cache_dir) + self.assertEqual(cache.get_deployed_benchmarks("gcp"), []) + + def test_allocated_resources(self) -> None: + """Allocated resources expose ids, buckets, region, and ports.""" + cache = Cache(self.cache_dir) + allocated = cache.get_allocated_resources() + + self.assertEqual(allocated["aws"]["resources_id"], "abc123") + self.assertEqual(allocated["aws"]["region"], "us-east-1") + self.assertEqual(len(allocated["aws"]["storage_buckets"]), 2) + self.assertEqual(allocated["aws"]["allocated_ports"], []) + + self.assertEqual(allocated["local"]["resources_id"], "local42") + self.assertEqual(allocated["local"]["allocated_ports"], [9000, 9001]) + + def test_allocated_resources_filter(self) -> None: + """Filtering allocated resources returns only the requested platform.""" + cache = Cache(self.cache_dir) + allocated = cache.get_allocated_resources("aws") + self.assertEqual(list(allocated.keys()), ["aws"]) + + def test_allocated_resources_ports_only(self) -> None: + """A local entry with only allocated ports is still reported.""" + cache = Cache(self.cache_dir) + allocated = cache.get_allocated_resources("local") + local = allocated["local"] + # local42 has no buckets, only ports - it must remain visible. + self.assertEqual(local["storage_buckets"], {}) + self.assertEqual(local["allocated_ports"], [9000, 9001]) + + +if __name__ == "__main__": + unittest.main()