sebs package

sebs.benchmark module 

Module for handling benchmarks in the Serverless Benchmarking Suite (SeBS).

This module provides classes for benchmark configuration, code packaging, and execution. It handles the preparation of code packages with dependencies for deployment to various serverless platforms, including caching mechanisms to avoid redundant builds.

class sebs.benchmark.Benchmark(benchmark: str, deployment_name: str, config: ExperimentConfig, system_config: SeBSConfig, output_dir: str, cache_client: Cache, docker_client: docker.client.DockerClient, system_variant_suffix: str | None = None, verbose: bool = False)[source]

Bases: LoggingBase

Creates code package representing a benchmark with all code and assets.

This class handles building, packaging, and deploying benchmark code for serverless platforms. This includes copying source files, adding deployment-specific wrappers, adding deployment-specific dependencies, and installing application dependencies within Docker images corresponding to the target cloud deployment. Code packages are cached.

The behavior of this class, particularly the build method, depends on the state of the SeBS cache:

If no cache entry exists for the benchmark (for the current language, deployment, etc.), a new code package is built.
If a cache entry exists, the hash of the benchmark’s source directory is computed and compared with the hash of cached package. If they differ, or if an update is forced, the package is rebuilt.
Otherwise (cache entry exists and hash matches), the cached code package is used.

benchmark: Name of the benchmark

benchmark_path: Path to the benchmark directory

benchmark_config: Configuration for the benchmark

code_package: Dictionary with code package information

functions: Dictionary of functions for this benchmark

code_location: Location of the code package

is_cached: Whether the benchmark is cached

is_cached_valid: Whether the cached benchmark is valid

code_size: Size of the code package in bytes

container_uri: URI of the container for container deployments

language: Programming language for the benchmark

language_name: Name of the programming language

language_version: Version of the programming language

has_input_processed: Whether input processing has been performed

uses_storage: Whether the benchmark uses cloud storage

uses_nosql: Whether the benchmark uses NoSQL databases

architecture: CPU architecture of the deployment target

system_variant: Deployment variant selected for the target provider

add_benchmark_data(output_dir: str) → None[source]

Add benchmark-specific data and assets to output directory.

Executes benchmark initialization scripts (init.sh) if present in the benchmark directory. These scripts typically download or generate additional data files required by the benchmark.

Parameters:: output_dir – Directory where benchmark data should be added

add_deployment_files(output_dir: str) → None[source]

Add deployment-specific wrapper files to output directory.

Copies platform-specific wrapper files (handlers, adapters) that integrate the benchmark code with the target FaaS platform’s execution environment.

Files are sourced from benchmarks/wrappers/{deployment_name}/{language_name}/.

Parameters:: output_dir – Directory where deployment files should be added

add_deployment_package(output_dir: str) → None[source]

Add deployment packages based on programming language.

Delegates to language-specific package addition methods to include platform-specific dependencies in the deployment package.

Parameters:: output_dir – Directory where deployment packages should be added
Raises:: NotImplementedError – If the language is not supported

add_deployment_package_cpp(output_dir: str) → None[source]

Generates CMakeLists.txt file for C++ benchmark.

The CMake file contains multiple steps: * Basic definition of benchmark target. * Packaging instructions for AWS. * Linking dependencies required by the benchmark. * Linking AWS SDK and Hiredis.

Parameters:: output_dir – Benchmark directory

add_deployment_package_java(output_dir: str)[source]

Extend benchmark’s pom.xml with system-specific packages. All Java dependencies for each platform are defined in systems.json.

Parameters:: output_dir – benchmark directory containing pom.xml to modify
Raises:: ValueError – when benchmark’s pom.xml is missing placeholder

add_deployment_package_nodejs(output_dir: str) → None[source]

Add Node.js deployment packages to package.json.

Modifies the package.json file to include platform-specific Node.js dependencies required for deployment. Handles versioned package.json files (e.g., package.json.12).

Parameters:: output_dir – Directory containing the package.json file to modify

add_deployment_package_python(output_dir: str) → None[source]

Add Python deployment packages to requirements file.

Appends platform-specific Python packages and benchmark module dependencies to the requirements.txt file for the deployment.

Handles versioned requirements files (e.g., requirements.txt.3.8).

Parameters:: output_dir – Directory containing the requirements file to modify

property architecture: str

Get the CPU architecture of the deployment target.

Returns:: Architecture name (e.g., ‘x86_64’, ‘arm64’)
Return type:: str

property benchmark: str

Get the benchmark name.

Returns:: Name of the benchmark
Return type:: str

property benchmark_config: BenchmarkConfig

Get the benchmark configuration.

Returns:: Configuration for the benchmark
Return type:: BenchmarkConfig

property benchmark_path: str

Get the path to the benchmark directory.

Returns:: Path to the benchmark directory
Return type:: str

build(package_build_step: Callable[[str, Language, str, str, str, bool], Tuple[str, float]], container_client: DockerContainer | None, container_build_step: Callable[[str, Language, str, str, str, bool], Tuple[str, float]] | None) → Tuple[bool, str | None, SystemVariant, str | None][source]

Build the complete benchmark deployment package.

Orchestrates the entire build process for a benchmark, including: - Code copying and dependency installation - Adding benchmark data and deployment-specific files - Running platform-specific build and packaging steps

(e.g., zipping, creating container image).

Cache validation and reuse if possible
Cache updates after successful build

Parameters:

package_build_step – Platform-specific build function for code package
container_client – Docker client for building container images (if container deployment)
container_build_step – Platform-specific build function for container deployments

Returns:

bool: Whether a new build was performed (False if cached)
str: Path to the built code package
SystemVariant: Selected deployment variant
str: Container URI (empty string if not container deployment)

Return type:

Tuple containing

builder_image_name() → Tuple[str, str][source]

Image names of builder Docker images for preparing benchmarks.

Returns two image names for fallback behavior: - Current version image (tagged with current SeBS version) - Previous version image (tagged with previous major SeBS version)

This allows new SeBS versions to use images from the previous stable version without requiring a complete rebuild of all images.

Returns:: Tuple of (previous_version_image_name, current_version_image_name).

property code_location: str | None

Get the absolute path to the prepared code package. If cached, it points to the location within the SeBS cache directory. Otherwise, it points to the build output directory.

Returns:: Path to the code package
Return type:: str

property code_package: Dict[str, Any]

Get the cached code package information, if available. This typically includes ‘location’ (relative to cache_dir), ‘hash’, and ‘size’.

Returns:: Dictionary with code package information
Return type:: Dict[str, Any]

code_package_is_archive() → bool[source]

Check if the code package is an archive file.

Determines whether the code package is stored as an archive file (ZIP) rather than a directory structure.

Returns:: True if package is a ZIP archive, False if it’s a directory
Return type:: bool

code_package_modify(filename: str, data: bytes) → None[source]

Updates a specific file within the code package without rebuilding the entire package. Currently only supports ZIP archive packages. This is used in experiments that modify the size of input package.

Does not support resizing containers or Azure deployments (non-ZIP).

Parameters:

filename – Name of the file to modify within the package
data – New content for the file as bytes

Raises:

NotImplementedError – If the code package is not a ZIP archive

code_package_recompute_size() → float[source]

Recalculate the size of the code package file.

Updates the internal size tracking after modifications to the code package file.

Returns:: Updated package size in bytes
Return type:: float

property code_size: int

Get the size of the code package in bytes.

Returns:: Size in bytes
Return type:: int

property container_uri: str

Get the URI of the container for container deployments.

Returns:: Container URI
Return type:: str
Raises:: AssertionError – If container URI is None

copy_code(output_dir: str) → None[source]

Copy benchmark source code to output directory.

Copies language-specific source files and dependency files from the benchmark directory to the output directory for deployment preparation. Handles Python requirements files, Node.js package.json files, and Java projects.

Parameters:: output_dir – Destination directory for copied files

static directory_size(directory: str) → float[source]

Calculate total size of all files in a directory.

Recursively calculates the total size in bytes of all files within the specified directory and its subdirectories.

Parameters:: directory – Path to the directory to measure
Returns:: Total size in bytes of all files in the directory
Return type:: int

format_maven_dependency(group_artifact: str, version: str) → str[source]

Helper method to format Java system dependencies. Dependencies in system.json are in “group:artifact”: version format; this function converts them to proper Maven <dependency> blocks.

Parameters:

group_artifact – name of library to add to benchmark
version – library version

Returns:

XML-formatted block inserted into pom.xml

property functions: Dict[str, Any]

Get the cached information about deployed functions associated with this benchmark for the current deployment, keyed by function name.

Returns:: Dictionary of functions
Return type:: Dict[str, Any]

property has_input_processed: bool

Check if input processing has been performed.

Returns:: True if processed, False otherwise
Return type:: bool

property hash: str

Get the hash of the benchmark code.

Computes an MD5 hash of the benchmark directory to determine if the code has changed since the last build.

Returns:: MD5 hash as a hexadecimal string
Return type:: str

static hash_directory(directory: str, deployment: str, language: Language, variant: str = 'default')[source]

Compute MD5 hash of an entire directory.

Calculates a hash of the benchmark source code by combining hashes of all relevant files. This includes language-specific files, deployment wrappers, and shared files like shell scripts and JSON configuration.

Parameters:

directory – Path to the directory to hash
deployment – Name of the deployment platform
language – Programming language name

Returns:

MD5 hash as a hexadecimal string

Return type:

str

install_dependencies(output_dir: str) → None[source]

Install benchmark dependencies using Docker.

Uses Docker containers to install language-specific dependencies (pip packages for Python, npm packages for Node.js) in an environment matching the target deployment platform. Pulls a pre-built Docker image specific to the deployment, language, and runtime version. Mounts the output directory into the container and runs an installer script (/sebs/installer.sh) within the container.

Tries current SeBS version image first, falls back to previous major version image if the current version image is not available. This allows new SeBS versions to use images from the previous stable version without requiring a complete rebuild.

Parameters:

output_dir – Directory containing the code package to build

Raises:

RuntimeError – If Docker image pull fails
docker.errors.ContainerError – If dependency installation fails

property is_cached: bool

Check if the benchmark is cached.

Returns:: True if cached, False otherwise
Return type:: bool

property is_cached_valid: bool

True if a cached code package exists and its hash matches the current benchmark source code hash.

Returns:: True if valid, False otherwise
Return type:: bool

property language: Language

Get the programming language for the benchmark.

Returns:: Programming language
Return type:: Language

property language_name: str

Get the name of the programming language, e.g., “python”.

Returns:: Name of the language
Return type:: str

property language_variant: str

Get the language runtime variant.

Returns:: The runtime variant (e.g., “default”, “pypy”, “bun”)
Return type:: str

property language_version: str

Get the version of the programming language, e.g. “3.8”.

Returns:: Version of the language
Return type:: str

prepare_input(system_resources: SystemResources, size: str, replace_existing: bool = False) → Dict[str, str][source]

Prepare benchmark input data and allocate cloud resources.

Locates the benchmark’s input generator module (input.py), determines storage requirements (object storage buckets, NoSQL tables), and invokes the generate_input function from the module to create and upload input data. Handles the setup of cloud storage buckets and NoSQL databases required by the benchmark. Updates the cache with storage details after successful preparation.

Parameters:

system_resources – Cloud system resources manager
size – Benchmark workload size (‘small’, ‘medium’, ‘large’)
replace_existing – Whether to replace existing input data

Returns:

Input configuration for the benchmark function

Return type:

Dict[str, str]

query_cache() → None[source]

Query the cache for existing benchmark code packages and functions.

Checks if there’s a cached code package or container for this benchmark and deployment combination. Updates the cache status fields based on whether the cache exists and if it’s still valid (hash matches).

recalculate_code_size() → float[source]

Recalculate and update the code package size.

Measures the current size of the output directory and updates the internal code size tracking.

Returns:: Updated code package size in bytes
Return type:: int

serialize() → dict[source]

Serialize the benchmark to a dictionary.

Returns:: Dictionary containing size and hash of the benchmark code
Return type:: dict

property system_variant: SystemVariant: Return the selected deployment variant for this benchmark.

property system_variant_suffix: str | None: Return the selected deployment variant for this benchmark.

static typename() → str[source]

Get the type name of this class.

Returns:: The type name
Return type:: str

property uses_nosql: bool

Check if the benchmark uses NoSQL databases.

Returns:: True if using NoSQL, False otherwise
Return type:: bool

property uses_storage: bool

Check if the benchmark uses cloud storage.

Returns:: True if using storage, False otherwise
Return type:: bool

validate_output(input_config: dict, output: dict, storage: PersistentStorage | None = None) → str | None[source]

Validate benchmark output against expected values.

Delegates to the benchmark’s input module validate_output function if it is defined. Passes the optional storage client through when the module’s validator declares a storage parameter. Logs a warning and returns an error message when no validation function is available.

Parameters:

input_config – The input configuration used to invoke the benchmark
output – The output returned by the benchmark function handler
storage – Optional persistent storage client for download-based checks

Returns:

None if validation passes, or a string describing the failure reason

class sebs.benchmark.BenchmarkConfig(timeout: int, memory: int, language_specs: List[LanguageSpec], modules: List[BenchmarkModule], cpp_dependencies: List[CppDependencies] | None = None)[source]

Bases: object

Configuration for a benchmark in the Serverless Benchmarking Suite.

This class stores the configuration parameters for a benchmark, including timeout, memory allocation, supported languages, and included modules.

timeout: Maximum execution time in seconds

memory: Memory allocation in MB

languages: List of supported programming languages

modules: List of benchmark modules/features required

static deserialize(json_object: dict) → BenchmarkConfig[source]

Create a BenchmarkConfig instance from a JSON object.

Parameters:: json_object – Dictionary containing benchmark configuration
Returns:: A new instance with the deserialized data
Return type:: BenchmarkConfig

property language_specs: List[LanguageSpec]

Get the list of language specifications with their variants.

Returns:: Language specifications for this benchmark
Return type:: List[LanguageSpec]

property languages: List[Language]

Get the list of supported programming languages.

Returns:: Supported programming languages
Return type:: List[Language]

property memory: int

Get the memory allocation in MB.

Returns:: The memory allocation
Return type:: int

property modules: List[BenchmarkModule]

Get the list of benchmark modules/features required.

Returns:: Required benchmark modules
Return type:: List[BenchmarkModule]

supported_variants(language: Language) → List[str][source]: Return the list of variants supported for the given language, or [] if the language has no implementation in this benchmark.

supports(language: Language, variant: str) → bool[source]: Return True when language + variant combination is declared in config.json.

property timeout: int

Get the maximum execution time in seconds.

Returns:: The timeout value
Return type:: int

class sebs.benchmark.BenchmarkModuleInterface[source]

Bases: object

Interface definition for benchmark input modules. Useful for static type hinting with mypy and documentation.

This class defines the interface that benchmark input modules must implement to provide input data generation, storage allocation, and NoSQL database setup for benchmarks.

All methods are static as they operate on benchmark data rather than instance state. Benchmark modules are dynamically loaded from the input.py file in each benchmark directory.

abstractmethod static allocate_nosql() → Dict[str, Dict[str, str]][source]

Define NoSQL table requirements for the benchmark.

Returns:

{

‘table_name’: {: ‘primary_key’: ‘key_field_name’, ‘secondary_key’: ‘optional_secondary_key_name’

}

Return type:

Dict containing table definitions with primary and secondary keys

abstractmethod static buckets_count() → Tuple[int, int][source]

Get the number of storage buckets required by the benchmark.

Returns:: Number of (input_buckets, output_buckets) needed
Return type:: Tuple[int, int]

abstractmethod static generate_input(data_dir: str | None, size: str, benchmarks_bucket: str | None, input_paths: List[str], output_paths: List[str], upload_func: Callable[[int, str, str], None] | None, nosql_func: Callable[[str, str, dict, Tuple[str, str], Tuple[str, str] | None], None] | None) → Dict[str, str][source]

Generate benchmark input data and configuration.

Creates the input data files and configuration needed for benchmark execution, uploading data to cloud storage and NoSQL databases as needed.

Parameters:

data_dir – Directory containing benchmark data files
size – Benchmark workload size (‘small’, ‘medium’, ‘large’)
benchmarks_bucket – Name of the cloud storage bucket for data
input_paths – List of input data paths in cloud storage
output_paths – List of output data paths in cloud storage
upload_func – Function for uploading files to cloud storage
nosql_func – Function for writing data to NoSQL databases

Returns:

Input configuration dictionary for the benchmark

Return type:

Dict[str, str]

static validate_output(data_dir: str | None, input_config: dict, output: dict, language: str, storage: PersistentStorage | None) → str | None[source]

Validate benchmark output against expected values.

Checks that the benchmark function’s output is correct for the given input. This optional method can be implemented in each benchmark’s input.py to enable output validation during regression testing.

Parameters:

data_dir – Directory containing benchmark data files (if exists)
input_config – The input configuration used to invoke the benchmark
output – The output returned by the benchmark function handler
language – Benchmark implementation language (e.g., ‘python’, ‘nodejs’)
storage – Storage interface for downloading output files if needed for validation

Returns:

None if validation passes, or a string describing the failure reason

class sebs.benchmark.LanguageSpec(language: Language, variants: List[str])[source]

Bases: object

Represents a language with its supported variants for a benchmark.

Parses the config language settings, supports both the legacy format (e.g. “python”) and the new dict format:

{“language”: “nodejs”, “variants”: [“default”, “bun”, “llrt”]}

The legacy format is treated as having just the “default” variant.

static deserialize(val) → LanguageSpec[source]

Deserialize a language specification from config.

Parameters:: val – Either a string (legacy format) or dict with language and variants
Returns:: Deserialized language specification
Return type:: LanguageSpec

property language: Language

Get the programming language.

Returns:: The programming language
Return type:: Language

serialize() → dict[source]

Serialize the language specification to a dictionary.

Returns:: Dictionary with language and variants keys
Return type:: dict

property variants: List[str]

Get the list of supported runtime variants.

Returns:: List of variant names (e.g., [“default”, “pypy”])
Return type:: List[str]

sebs.benchmark.load_benchmark_input(benchmark_path: str) → BenchmarkModuleInterface[source]

Dynamically load the input module for a benchmark.

Loads the input.py file from the benchmark directory and returns it as a module interface for generating benchmark input data.

Parameters:

benchmark_path – Path to the benchmark directory containing input.py

Returns:

Loaded input module with benchmark-specific: input generation functions

Return type:

BenchmarkModuleInterface

Raises:

FileNotFoundError – If input.py is not found in the benchmark directory
ImportError – If the input module cannot be loaded

sebs.cache module 

Caching system for SeBS (Serverless Benchmarking Suite).

This module provides comprehensive caching functionality for the SeBS framework, including configuration caching, code package management, function deployment tracking, and storage resource management.

The Cache class manages persistent storage of benchmark configurations, compiled code packages, Docker containers, deployed functions, and cloud resource configurations to optimize repeated benchmark executions and deployments.

This class is essential for efficient benchmarking - we avoid regenerating cloud resources, and we do not have to keep querying them every time we start the benchmark. This is particularly important for cloud platforms like Azure, where queries require CLI tool running in a container and can take long time to resolve.

Example

Basic cache usage:: cache = Cache(“/path/to/cache”, docker_client) config = cache.get_benchmark_config(“aws”, “110.dynamic-html”) cache.add_code_package(“aws”, benchmark_instance)

class sebs.cache.Cache(cache_dir: str, docker_client: DockerClient)[source]

Bases: LoggingBase

Persistent caching system for SeBS benchmark configurations and deployments.

This class provides comprehensive caching functionality for SeBS benchmarks, including configuration management, code package storage, function tracking, and cloud resource management. It uses a file-based cache system with thread-safe operations.

cached_config

In-memory cache of cloud configurations.

Type:: Dict[str, Any]

config_updated

Flag indicating if configuration needs to be saved.

Type:: bool

cache_dir

Absolute path to the cache directory.

Type:: str

ignore_functions

Flag to skip function caching operations.

Type:: bool

ignore_storage

Flag to skip storage resource caching.

Type:: bool

docker_client

Docker client for container operations.

Type:: docker.DockerClient

add_code_package(deployment_name: str, code_package: Benchmark) → None[source]

Add a new code package to the cache.

Copies the code package (directory or zip file) into the cache structure. Records metadata (hash, size, location, timestamps, image details if container) in the benchmark’s config.json within the cache. Handles both package and container deployments.

Parameters:

deployment_name (str) – Name of the deployment platform.
code_package (Benchmark) – The benchmark code package to cache.

Raises:

RuntimeError – If cached application already exists for the deployment.

add_function(deployment_name: str, language_name: str, code_package: Benchmark, function: Function) → None[source]

Add new function to cache.

Caches a deployed function configuration for a benchmark. Links the function to its corresponding code package.

Parameters:

deployment_name (str) – Name of the deployment platform.
language_name (str) – Programming language name.
code_package (Benchmark) – The benchmark code package.
function (Function) – The deployed function to cache.

Raises:

RuntimeError – If code package doesn’t exist in cache.

static code_cache_keys(code_package: Benchmark) → Tuple[List[str], List[str]][source]: Add language and system variant suffixes to the package cache key so differing build artifacts do not conflict in cache.

get_all_functions(deployment: str) → Dict[str, Any][source]

Get all cached function configurations for a given deployment.

Iterates all benchmarks and languages in the cache to collect every function deployed to the specified platform.

Parameters:: deployment (str) – Deployment platform name
Returns:: Mapping of function name to function configuration, aggregated across benchmarks and languages.

get_benchmark_config(deployment: str, benchmark: str) → Dict[str, Any] | None[source]

Access cached configuration of a benchmark.

Parameters:

deployment (str) – Deployment platform (‘aws’, ‘azure’, ‘gcp’, ‘openwhisk’, ‘local’).
benchmark (str) – Benchmark name (e.g., ‘110.dynamic-html’).

Returns:

Benchmark configuration or None if not found.

Return type:

Optional[Dict[str, Any]]

get_code_package(deployment_name: str, code_package: Benchmark) → Dict[str, Any] | None[source]

Access cached version of benchmark code package.

Parameters:

deployment (str) – Deployment platform name.
code_package (Benchmark) – Benchmark package.

Returns:

Code package configuration or None if not found.

Return type:

Optional[Dict[str, Any]]

get_config(cloud: str) → Dict[str, Any] | None[source]

Get cached configuration for a specific cloud provider.

Parameters:: cloud (str) – Cloud provider name (e.g., ‘aws’, ‘azure’, ‘gcp’).
Returns:: The cached configuration or None if not found.
Return type:: Optional[Dict[str, Any]]

get_config_key(keys: List[str]) → Any | None[source]

Return the value at a nested key path in the cached configuration. Does not throw an error if the key path does not exist.

Parameters:: keys – key path needed to access the config value
Returns:: The value at the specified key path, or None if not found.

get_container(deployment_name: str, code_package: Benchmark) → Dict[str, Any] | None[source]

Access cached container configuration for a benchmark.

Parameters:

deployment (str) – Deployment platform name.
code_package (Benchmark) – Benchmark package.

Returns:

Container configuration or None if not found.

Return type:

Optional[Dict[str, Any]]

get_functions(deployment: str, benchmark: str, language: str) → Dict[str, Any] | None[source]

Get cached function configurations for a benchmark.

Parameters:

deployment (str) – Deployment platform name.
benchmark (str) – Benchmark name.
language (str) – Programming language.

Returns:

Function configurations or None if not found.

Return type:

Optional[Dict[str, Any]]

get_nosql_config(deployment: str, benchmark: str) → Dict[str, Any] | None[source]

Access cached NoSQL configuration of a benchmark.

Parameters:

deployment (str) – Deployment platform name.
benchmark (str) – Benchmark name.

Returns:

NoSQL configuration or None if not found.

Return type:

Optional[Dict[str, Any]]

get_nosql_configs(deployment: str) → Dict[str, Any][source]

Access cached NoSQL configuration for all benchmarks.

Iterates all benchmark directories in the cache and merges their NoSQL table configurations for the given deployment into a single dict.

Parameters:: deployment (str) – Deployment platform name.
Returns:: NoSQL configurations across all benchmarks

get_storage_config(deployment: str, benchmark: str) → Dict[str, Any] | None[source]

Access cached storage configuration of a benchmark.

Parameters:

deployment (str) – Deployment platform name.
benchmark (str) – Benchmark name.

Returns:

Storage configuration or None if not found.

Return type:

Optional[Dict[str, Any]]

invalidate_all_container_uris(deployment: str) → None[source]

Set image-uri to None for all cached containers of a deployment.

Walks all benchmark directories and clears the image-uri field for every container entry under the given deployment. This forces a re-push to the registry on next use without invalidating the rest of the cached state.

This function is used primarily after cleaning up cloud resources.

Parameters:: deployment (str) – Deployment platform name.

load_config() → None[source]

Load cached cloud configurations from disk.

Reads configuration files for all supported cloud platforms from the cache directory and loads them into memory.

lock() → None[source]: Acquire the cache lock for thread-safe operations.

remove_config_key(keys: List[str])[source]

Removes a configuration entry nested within cache dictiariony. Used after deleting a specific cloud resource.

Does not throw an error if the key path does not exist.

Parameters:: keys – key path needed to access the config value

remove_function(deployment: str, benchmark: str, language: str, function_name: str)[source]

Remove a function entry from all benchmark cache configs.

Parameters:: function_name – function for removal

remove_nosql(deployment: str)[source]

Remove nosql config entries across all benchmarks for a deployment.

Parameters:: deployment – cloud platform name

remove_storage(deployment: str)[source]

Remove storage config entries across all benchmarks for a deployment.

Parameters:: deployment – cloud platform name

shutdown() → None[source]

Save cached configurations to disk if they were updated.

Writes all updated cloud configurations back to their respective JSON files in the cache directory.

static typename() → str[source]

Get the typename for this cache.

Returns:: The cache type name.
Return type:: str

unlock() → None[source]: Release the cache lock.

update_code_package(deployment_name: str, code_package: Benchmark) → None[source]

Update an existing code package in the cache.

Copies the new code package version over the old one. Updates metadata (hash, size, modification timestamp, image details if container) in the benchmark’s config.json. If the cached package doesn’t exist, adds it as a new package.

Parameters:

deployment_name (str) – Name of the deployment platform.
code_package (Benchmark) – The benchmark code package to update.

update_config(val: Any, keys: List[str]) → None[source]

Update configuration values at nested key path.

Updates cached configuration by setting a value at the specified nested key path. Sets the config_updated flag to ensure changes are persisted to disk.

Parameters:

val (Any) – New value to store.
keys (List[str]) – Array of consecutive keys for multi-level dictionary.

update_container_uri(deployment_name: str, code_package: Benchmark, uri: str) → None[source]

Update the image-uri for a specific cached container entry.

Used when the image is cached locally, but needs to be pushed to the registry to be accessible for cloud deployment.

Parameters:

deployment_name (str) – Deployment platform name.
code_package (Benchmark) – Benchmark package identifying the cache entry.
uri (str) – New image URI to store.

update_function(function: Function) → None[source]

Update an existing function in the cache.

Updates cached function configuration with new metadata. Searches across all deployments and languages to find the function by name.

Parameters:: function (Function) – The function with updated configuration.
Raises:: RuntimeError – If function’s code package doesn’t exist in cache.

update_nosql(deployment: str, benchmark: str, config: Dict[str, Any]) → None[source]

Update cached NoSQL configuration for a benchmark.

Parameters:

deployment (str) – Deployment platform name.
benchmark (str) – Benchmark name.
config (Dict[str, Any]) – NoSQL configuration to cache.

update_storage(deployment: str, benchmark: str, config: Dict[str, Any]) → None[source]

Update cached storage configuration for a benchmark.

Parameters:

deployment (str) – Deployment platform name.
benchmark (str) – Benchmark name.
config (Dict[str, Any]) – Storage configuration to cache.

sebs.cache.keys_exist(obj: Dict, keys: List[Any]) → bool[source]

Find if a nested object exists in a dictionary.

example: {key1: {key2: {key3: value}}} for [key1, key2, key3] -> True

Parameters:

obj – dictionary
keys – dynamic list of nested keys

Returns:

true if the nested object exists

sebs.cache.keys_get(obj: Dict, keys: List[Any]) → Any[source]

Find if a nested object exists in a dictionary.

example: {key1: {key2: {key3: value}}} for [key1, key2, key3] -> True

Parameters:

obj – dictionary
keys – dynamic list of nested keys

Returns:

true if the nested object exists

sebs.cache.update(d: Dict[str, Any], u: Mapping[str, Any]) → Dict[str, Any][source]

Recursively update nested dictionary with another dictionary.

This function performs deep merge of two dictionaries, merging nested dictionary values rather than replacing them entirely.

Parameters:

d (Dict[str, Any]) – The target dictionary to update.
u (Mapping[str, Any]) – The source dictionary with updates.

Returns:

The updated dictionary.

Return type:

Dict[str, Any]

sebs.cache.update_dict(cfg: Dict[str, Any], val: Any, keys: List[str]) → None[source]

Update dictionary value at nested key path.

Updates a nested dictionary by setting a value at a path specified by a list of keys. Creates intermediate dictionaries as needed.

Parameters:

cfg (Dict[str, Any]) – The dictionary to update.
val (Any) – The value to set at the key path.
keys (List[str]) – List of keys forming the path to the target location.

sebs.config module 

Configuration management for SeBS (Serverless Benchmarking Suite).

This module provides configuration management functionality for the SeBS framework, including system configuration loading, Docker image management, and deployment setting retrieval from the systems.json configuration file.

The SeBSConfig class serves as the central configuration manager that provides access to platform-specific settings, language configurations, and deployment options across different cloud providers and local deployments.

class sebs.config.SeBSConfig[source]

Bases: object

Central configuration manager for SeBS framework.

This class manages all configuration settings for the SeBS benchmarking suite, including system configurations, Docker settings, deployment options, and platform-specific parameters. It loads configuration from systems.json and provides convenient access methods for various configuration aspects.

_system_config

The loaded system configuration from systems.json.

Type:: Dict

_image_tag_prefix

Custom prefix for Docker image tags.

Type:: str

benchmark_base_images(deployment_name: str, language_name: str, architecture: str) → Dict[str, str][source]

Get base Docker images for benchmarks on a specific platform.

Parameters:

deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
language_name (str) – Programming language name (e.g., ‘python’, ‘nodejs’).
architecture (str) – Target architecture (e.g., ‘x64’, ‘arm64’).

Returns:

Dictionary mapping language versions to base image names.

Return type:

Dict[str, str]

benchmark_image_name(system: str, benchmark: str, language_name: str, language_version: str, architecture: str, registry: str | None = None, repository: str | None = None) → str[source]

Generate full Docker image name for a benchmark.

Parameters:

system (str) – Deployment system name (e.g., ‘aws’, ‘azure’).
benchmark (str) – Benchmark name (e.g., ‘110.dynamic-html’).
language_name (str) – Programming language name (e.g., ‘python’).
language_version (str) – Language version (e.g., ‘3.8’).
architecture (str) – Target architecture (e.g., ‘x64’).
registry (Optional[str]) – Docker registry URL. If None, uses default repository.

Returns:

Complete Docker image name including registry and tag.

Return type:

str

benchmark_image_tag(system: str, benchmark: str, language_name: str, language_version: str, architecture: str) → str[source]

Generate Docker image tag for a benchmark container.

Creates a standardized tag format that includes system, benchmark, language, version, architecture, optional prefix, and SeBS version.

Format: function.{system}.{benchmark}.{language_name}-{language_version}-: {architecture}[-{image_prefix}]-{sebs_version}

Parameters:

system (str) – Deployment system name (e.g., ‘aws’, ‘azure’).
benchmark (str) – Benchmark name (e.g., ‘110.dynamic-html’).
language_name (str) – Programming language name (e.g., ‘python’).
language_version (str) – Language version (e.g., ‘3.8’).
architecture (str) – Target architecture (e.g., ‘x64’).

Returns:

Generated Docker image tag.

Return type:

str

default_system_variant(deployment_name: str) → str[source]

Return the default deployment variant for a platform.

The default is the first declared variant in systems.json.

Parameters:: deployment_name – Name of the deployment platform.
Returns:: Default deployment variant for the platform.

deployment_files(deployment_name: str, language_name: str) → List[str][source]

Get deployment files list for a specific deployment and language.

Parameters:

deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
language_name (str) – Programming language name (e.g., ‘python’, ‘nodejs’).

Returns:

List of required deployment files.

Return type:

List[str]

deployment_module_packages(deployment_name: str, language_name: str) → Dict[str, str][source]

Get deployment module packages for a specific deployment and language, e.g., packages specific to object or NoSQL storage.

Parameters:

deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
language_name (str) – Programming language name (e.g., ‘python’, ‘nodejs’).

Returns:

Dictionary mapping module package names to their versions.

Return type:

Dict[str, str]

deployment_packages(deployment_name: str, language_name: str) → Dict[str, str][source]

Get deployment packages for a specific deployment and language. These are packages added by SeBS to the benchmark’s list of dependencies.

Parameters:

deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
language_name (str) – Programming language name (e.g., ‘python’, ‘nodejs’).

Returns:

Dictionary mapping package names to their versions.

Return type:

Dict[str, str]

docker_image_name(system: str, image_type: str, language_name: str | None = None, language_version: str | None = None) → str[source]

Generate standardized Docker image name for infrastructure images.

Creates a standardized name format for Docker images used in the SeBS infrastructure (build, run, manage, function, dependencies images).

Format: {repo}:{type}.{system}[.{lang}[.{version}]]-{sebs_version}

Parameters:

system (str) – Deployment system name (e.g., ‘aws’, ‘azure’, ‘local’).
image_type (str) – Type of image (e.g., ‘build’, ‘run’, ‘manage’, ‘function’).
language_name (Optional[str]) – Programming language name (e.g., ‘python’, ‘nodejs’).
language_version (Optional[str]) – Language version (e.g., ‘3.9’, ‘16’).

Returns:

Complete Docker image name with repository and tag.

Return type:

str

docker_image_types(deployment_name: str, language_name: str) → List[str][source]

Get available Docker image types for a deployment and language.

Parameters:

deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
language_name (str) – Programming language name (e.g., ‘python’, ‘nodejs’).

Returns:

List of available Docker image types.

Return type:

List[str]

docker_repository() → str[source]

Get the Docker repository name from configuration.

Returns:: The Docker repository name configured in systems.json.
Return type:: str

property image_tag_prefix: str

Get the current Docker image tag prefix.

Returns:: The current image tag prefix.
Return type:: str

previous_version() → str[source]

Get the previous major SeBS framework version.

This is used as a fallback version for Docker images that haven’t been rebuilt for the current version. It allows new SeBS versions to use images from the previous stable version without requiring a full rebuild.

Returns:: The previous major version string, or ‘unknown’ if not configured.
Return type:: str

supported_architecture(deployment_name: str) → List[str][source]

Get supported architectures for a deployment platform.

Parameters:: deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
Returns:: List of supported architectures (e.g., [‘x64’, ‘arm64’]).
Return type:: List[str]

supported_language_versions(deployment_name: str, language_name: str, architecture: str) → List[str][source]

Get supported language versions for a deployment, language, and architecture.

Parameters:

deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
language_name (str) – Programming language name (e.g., ‘python’, ‘nodejs’).
architecture (str) – Target architecture (e.g., ‘x64’, ‘arm64’).

Returns:

List of supported language versions.

Return type:

List[str]

supported_system_variants(deployment_name: str) → List[str][source]: Return the supported deployment variants for a platform.

supported_variants(deployment_name: str, language_name: str) → List[str][source]

Return the list of language variants supported by a deployment platform.

Falls back to ["default"] when the key is absent so that platforms that haven’t declared their variant support still work with the default variant.

username(deployment_name: str, language_name: str) → str[source]

Get the username for a specific deployment and language configuration.

Parameters:

deployment_name (str) – Name of the deployment platform (e.g., ‘aws’, ‘azure’).
language_name (str) – Programming language name (e.g., ‘python’, ‘nodejs’).

Returns:

The username configured for the deployment and language combination.

Return type:

str

version() → str[source]

Get the SeBS framework version.

Returns:: The SeBS version string, or ‘unknown’ if not configured.
Return type:: str

sebs.sebs module 

Main SeBS (Serverless Benchmarking Suite) client implementation.

This module provides the main interface for SeBS: - Deployment client creation for different platforms (AWS, Azure, GCP, OpenWhisk, local) - Benchmark execution and configuration - Experiment setup and execution - Storage access (object storage and NoSQL) - Caching and Docker management - Logging and output handling

The SeBS client is the central point of interaction for both the CLI and programmatic use.

class sebs.sebs.SeBS(cache_dir: str, output_dir: str, verbose: bool = False, logging_filename: str | None = None)[source]

Bases: LoggingBase

Main client for the Serverless Benchmarking Suite.

cache_client: Client for managing cached artifacts (code packages, etc.)

docker_client: Docker client for container operations

output_dir: Directory for storing output files and logs

verbose: Whether to enable verbose logging

logging_filename: Default log file name

config: Global SeBS configuration

property cache_client: Cache

Get the cache client.

Returns:: Cache client for managing cached artifacts

property config: SeBSConfig

Get the global SeBS configuration.

Returns:: Global configuration object

property docker_client: DockerClient

Get the Docker client.

Returns:: Docker client for container operations

generate_logging_handlers(logging_filename: str | None = None) → LoggingHandlers[source]

Generate logging handlers for a specific file.

This method creates or retrieves cached logging handlers for a given filename. If no filename is provided, the default logging filename is used.

Parameters:: logging_filename – Optional filename for logs, defaults to self.logging_filename
Returns:: LoggingHandlers configured for the specified file

get_benchmark(name: str, deployment: System, config: Config, logging_filename: str | None = None) → Benchmark[source]

Get a benchmark implementation for a specific benchmark.

This method creates and returns a benchmark implementation for the specified benchmark name. It configures the benchmark with the appropriate deployment, configuration, and resources.

Parameters:

name – Name of the benchmark to create (e.g., “210.thumbnailer”)
deployment – FaaS system deployment client
config – Experiment configuration
logging_filename – Optional filename for logs

Returns:

An initialized benchmark implementation

get_deployment(config: dict, logging_filename: str | None = None, deployment_config: Config | None = None) → System[source]

Get a deployment client for a specific cloud platform.

This method creates and returns a deployment client for the specified cloud platform. It validates that the requested platform and configuration are supported, and initializes the client with the appropriate resources.

The method dynamically imports the necessary modules for each platform based on what’s available in the environment, determined by has_platform().

Parameters:

config – Configuration dictionary with deployment and experiment settings
logging_filename – Optional filename for logs
deployment_config – Optional pre-configured deployment config

Returns:

An initialized FaaS system deployment client

Raises:

RuntimeError – If the requested deployment is not supported or if the configuration is invalid (unsupported architecture, deployment type, etc.)

get_deployment_config(config: dict, logging_filename: str | None = None) → Config[source]

Create a deployment configuration from a dictionary.

This method deserializes a deployment configuration from a dictionary, setting up logging handlers and connecting it to the cache client.

Parameters:

config – Configuration dictionary
logging_filename – Optional filename for logs

Returns:

A deserialized deployment configuration object

get_experiment(experiment_type: str, config: dict, logging_filename: str | None = None) → Experiment[source]

Get an experiment implementation for a specific experiment type.

This method creates and returns an experiment implementation for the specified experiment type. It validates that the requested experiment type is supported and initializes the experiment with the appropriate configuration.

Parameters:

experiment_type – Type of experiment to create (e.g., “perf-cost”)
config – Configuration dictionary
logging_filename – Optional filename for logs

Returns:

An initialized experiment implementation

Raises:

RuntimeError – If the requested experiment type is not supported

get_experiment_config(config: dict) → Config[source]

Create an experiment configuration from a dictionary.

This method deserializes an experiment configuration from a dictionary. The experiment configuration contains settings specific to the experiment being run, such as the number of iterations, timeout, etc.

Parameters:: config – Configuration dictionary
Returns:: A deserialized experiment configuration object

static get_nosql_config_implementation(storage_type: NoSQLStorage)[source]

Get a NoSQL configuration implementation for a specific storage type.

This method returns the class for a NoSQL configuration implementation for the specified storage type.

Parameters:: storage_type – Type of NoSQL storage to get configuration for
Returns:: NoSQL configuration implementation class
Raises:: AssertionError – If the requested storage type is not supported

static get_nosql_implementation(storage_type: NoSQLStorage) → Type[NoSQLStorage][source]

Get a NoSQL storage implementation for a specific storage type.

This method returns the class for a NoSQL storage implementation for the specified storage type.

Parameters:: storage_type – Type of NoSQL storage to get implementation for
Returns:: NoSQL storage implementation class
Raises:: AssertionError – If the requested storage type is not supported

static get_storage_config_implementation(storage_type: Storage)[source]

Get a storage configuration implementation for a specific storage type.

This method returns the class for a storage configuration implementation for the specified storage type.

Parameters:: storage_type – Type of storage to get configuration for
Returns:: Storage configuration implementation class
Raises:: AssertionError – If the requested storage type is not supported

static get_storage_implementation(storage_type: Storage) → Type[PersistentStorage][source]

Get a storage implementation for a specific storage type.

This method returns the class for a persistent storage implementation for the specified storage type.

Parameters:: storage_type – Type of storage to get implementation for
Returns:: Storage implementation class
Raises:: AssertionError – If the requested storage type is not supported

ignore_cache()[source]

Configure the cache to only store code packages.

After calling this method, the cache will only store code packages and won’t update or use cached functions and storage. This is useful when you want to ensure that functions are redeployed and storage is recreated, but still want to reuse code packages.

property logging_filename: str | None

Get the default logging filename.

Returns:: Default logging filename or None if not set

property output_dir: str

Get the output directory.

Returns:: Path to the output directory

shutdown()[source]

Shutdown the SeBS client and release resources.

This method shuts down the cache client and releases any resources that need to be cleaned up when the client is no longer needed. It is automatically called when using the client as a context manager.

property verbose: bool

Get the verbose flag.

Returns:: Whether verbose logging is enabled

sebs.statistics module 

Statistical analysis utilities for benchmark experiments.

This module provides functions for computing basic statistics and confidence intervals on benchmark experiment results. It includes both parametric (Student’s t-distribution) and non-parametric (Le Boudec) methods for computing confidence intervals.

class sebs.statistics.BasicStats(mean, median, std, cv)

Bases: tuple

cv: Alias for field number 3

mean: Alias for field number 0

median: Alias for field number 1

std: Alias for field number 2

sebs.statistics.basic_stats(times: List[float]) → BasicStats[source]

Compute basic statistics for a list of measurement times.

This function computes the mean, median, standard deviation, and coefficient of variation for a list of measurement times.

Parameters:: times – List of measurement times
Returns:: A BasicStats named tuple with the computed statistics

sebs.statistics.ci_le_boudec(alpha: float, times: List[float]) → Tuple[float, float][source]

Compute non-parametric confidence interval using Le Boudec’s method.

It requires a sufficient number of samples but it is a non-parametric method that does not assume that data follows the normal distribution.

Reference:: J.-Y. Le Boudec, “Performance Evaluation of Computer and Communication Systems”, 2010.

Parameters:

alpha – Confidence level (e.g., 0.95 for 95% confidence)
times – List of measurement times

Returns:

A tuple (lower, upper) representing the confidence interval

Raises:

AssertionError – If an unsupported confidence level is provided

sebs.statistics.ci_tstudents(alpha: float, times: List[float]) → Tuple[float, float][source]

Compute parametric confidence interval using Student’s t-distribution.

This is a parametric method that assumes the data follows a normal distribution.

Parameters:

alpha – Confidence level (e.g., 0.95 for 95% confidence)
times – List of measurement times

Returns:

A tuple (lower, upper) representing the confidence interval

sebs.statistics.print_stats(logger: Logger, experiments: Result)[source]

Print statistics of a selected experiment result. Used by the main CLI driver for pretty printing.

Parameters:

logger – Logger to user
experiments – Experiment to process

sebs.types module 

sebs.utils module 

Utility functions and classes for the Serverless Benchmarking Suite (SeBs).

This module provides common utilities used throughout the framework, including: - File system operations and path management - Process execution and command handling - JSON serialization and data manipulation - Logging configuration and utilities - Platform detection functions

class sebs.utils.ColoredWrapper(prefix, logger, verbose=True, propagte=False)[source]

Bases: object

Wrapper for logging with colored console output.

This class provides formatted, colorized logging output for better readability in terminal environments. It optionally propagates messages to the standard Python logger.

SUCCESS: Green color code for success messages

STATUS: Blue color code for status/info messages

WARNING: Yellow color code for warnings

ERROR: Red color code for errors

BOLD: Bold text formatting code

END: Code to reset text formatting

BOLD = '\x1b[1m'

END = '\x1b[0m'

ERROR = '\x1b[91m'

STATUS = '\x1b[94m'

SUCCESS = '\x1b[92m'

WARNING = '\x1b[93m'

critical(message)[source]

Log a critical error message.

Parameters:: message – The message to log

debug(message)[source]

Log a debug message.

Parameters:: message – The message to log

error(message)[source]

Log an error message.

Parameters:: message – The message to log

info(message)[source]

Log an informational message.

Parameters:: message – The message to log

set_filter(filter: SensitiveDataFilter)[source]

Set custom data filter.

Parameters:: filter

warning(message)[source]

Log a warning message.

Parameters:: message – The message to log

class sebs.utils.JSONSerializer(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: JSONEncoder

Custom JSON encoder for objects with serialize method.

This encoder handles objects by: 1. Using their serialize() method if available 2. Converting dictionaries to strings 3. Using vars() to get object attributes 4. Falling back to string representation

default(o)[source]

Custom serialization for objects.

Parameters:: o – Object to serialize
Returns:: JSON serializable representation of the object

class sebs.utils.LoggingBase[source]

Bases: object

Base class providing consistent logging functionality across the framework.

This class sets up a logger with a unique identifier and provides methods for logging at different levels with consistent formatting. It supports both console output with color coding and optional file logging.

log_name: Unique identifier for this logger

logging: ColoredWrapper for formatted console output

REDACTION_FILTER: SensitiveDataFilter | None = None

classmethod enable_filtering() → None[source]: Enable sensitive data filtering for all loggers.

property logging: ColoredWrapper

Get the colored logging wrapper.

Returns:: The logging wrapper for this instance
Return type:: ColoredWrapper

property logging_handlers: LoggingHandlers

Get the logging handlers configuration.

Returns:: The current handlers configuration
Return type:: LoggingHandlers

classmethod set_filtering_resource_id(resource_id: str, cloud_id: str | None = None) → None[source]

Add resource ID and cloud user IDs to logging filtering.

Parameters:

resource_id – SeBS cloud ID
cloud_id – cloud-specific user ID (e.g., AWS account ID or GCP project name)

class sebs.utils.LoggingHandlers(verbose: bool = False, filename: str | None = None)[source]

Bases: object

Configures and manages logging handlers.

This class sets up handlers for logging to files and tracks verbosity settings for use with ColoredWrapper.

handler: FileHandler for logging to a file

verbosity: Whether to include debug-level messages

class sebs.utils.SensitiveDataFilter[source]

Bases: Filter

Logging filter that removes function URLs and resource ID from output.

_DEFAULT_URL_PATTERNS

List of patterns for URLs of functions.

Type:: tuple[str, …]

REDACTED: String to replace with.

REDACTED = '[REDACTED]'

Redacts serverless endpoint URLs and configurable resource IDs from logs. Resource IDs can be added/removed at runtime as deployments happen.

This allows hiding exact resource names in the cloud from publically visible logs, e.g., in CI workers.

filter(record: LogRecord) → bool[source]

Apply redaction to logging messages.

Parameters:: record – logging record
Returns:: always true

filter_string(msg: str) → str[source]

Apply redaction to custom messages.

Parameters:: msg – message
Returns:: message with data redacted

set_resource_id(resource_id: str, cloud_id: str | None = None) → None[source]

Set filtering for a specific resource ID.

The function is idempotent - we can only set the resource ID once. It is also thread-safe, so multiple threads doing multithreading can call it many times and we initialize only once. :param resource_id:

sebs.utils.append_nested_dict(cfg: dict, keys: List[str], value: dict | None) → None[source]

Append a dictionary to a nested location in another dictionary.

Parameters:

cfg – Dictionary to update
keys – List of keys forming a path to the value
value – Dictionary to append (skipped if None or empty)

sebs.utils.catch_interrupt() → None[source]

Set up a signal handler to catch interrupt signals (Ctrl+C).

Prints a stack trace and exits when an interrupt is received. This helps with debugging by showing the execution context at the time of the interruption.

sebs.utils.configure_logging() → None[source]

Configure global logging settings.

Reduces noise from third-party libraries by setting their log levels to ERROR. This ensures that only important messages from these libraries are shown.

sebs.utils.create_output(directory: str, preserve_dir: bool, verbose: bool) → str[source]

Create or clean an output directory for benchmark results.

Parameters:

directory – Path to create
preserve_dir – Whether to preserve existing directory
verbose – Verbosity level for logging

Returns:

Absolute path to the output directory

Return type:

str

sebs.utils.ensure_benchmarks_data(logger: ColoredWrapper) → Path[source]

Ensure benchmarks-data exists, cloning if necessary.

For local installation, we use submodule to ensure that benchmarks-data is initialized. For package installation, we clone benchmarks-data to a home directory if it doesn’t exist.

Returns:: Path to benchmarks-data directory
Raises:: RuntimeError – If cloning fails

sebs.utils.execute(cmd, shell=False, cwd=None) → str[source]

Execute a shell command and capture its output, handling errors.

Parameters:

cmd – Command to execute (string)
shell – Whether to use shell execution (enables wildcards, pipes, etc.)
cwd – Working directory for command execution

Returns:

Command output as string

Return type:

str

Raises:

RuntimeError – If command execution fails

sebs.utils.find(name: str, path: str) → str | None[source]

Find a directory with the given name in the specified path.

Parameters:

name – Directory name to find
path – Path to search in

Returns:

Path to the found directory, or None if not found

Return type:

str

sebs.utils.find_benchmark(benchmark: str, path: str) → str | None[source]

Locate directory corresponding to a benchmark in the repository.

Searches for a benchmark directory in either the benchmarks or benchmarks-data directories.

Parameters:

benchmark – Benchmark name
path – Path for lookup, relative to repository (usually ‘benchmarks’ or ‘benchmarks-data’)

Returns:

Path to benchmark directory, or None if not found

Return type:

str

sebs.utils.get_benchmarks_data_path() → Path[source]

Get path to benchmarks-data directory.

Returns:

./benchmarks-data/ - For package install: ~/.sebs/benchmarks-data/

Return type:

For git clone

sebs.utils.get_project_root() → Path[source]

Get project root directory. This points to directory where everything is located.

Returns:

repository root - For package install: main installation path

Return type:

For git clone

sebs.utils.get_resource_path(*path_parts: str) → Path[source]

Get path to a resource (config, benchmarks, dockerfiles, tools). Resolves to path within git repository (outside of sebs) or in the installed package.

Parameters:: *path_parts – Path components (e.g., “config”, “systems.json”)
Returns:: Path to the resource

sebs.utils.global_logging() → None[source]

Set up basic global logging configuration.

Configures the root logger with a standard format, timestamp, and INFO level. This provides a baseline for all logging in the application.

sebs.utils.has_platform(name: str) → bool[source]

Check if a specific platform is enabled and supported.

We check if the required libraries for a specific platform are available.

Parameters:: name – Platform name to check
Returns:: True if platform is enabled, False otherwise
Return type:: bool

sebs.utils.is_linux() → bool[source]

Check if the system is Linux and not Windows Subsystem for Linux.

Returns:: True if native Linux, False otherwise
Return type:: bool

sebs.utils.serialize(obj) → str[source]

Serialize an object to a JSON string. Applies serialize method when defined by the object.

Parameters:: obj – Object to serialize
Returns:: JSON string representation of the object
Return type:: str

sebs.utils.update_nested_dict(cfg: dict, keys: List[str], value: str | None) → None[source]

Update a nested dictionary with a value at the specified key path.

Parameters:

cfg – Dictionary to update
keys – List of keys forming a path to the value
value – Value to set (skipped if None)

sebs.version module 

Main SeBS version information.

Module contents 

SeBS

sebs package

Subpackages 

Submodules 

sebs.benchmark module 

sebs.cache module 

sebs.config module 

sebs.sebs module 

sebs.statistics module 

sebs.types module 

sebs.utils module 

sebs.version module 

Module contents 