sebs.gcp package

Submodules

sebs.gcp.cli module

Google Cloud CLI integration for SeBS.

This module provides a Docker-based Google Cloud CLI interface. Currently, we use it mostly to allocate and manage Datastore accounts. There’s no API or Python library for that.

Classes:

GCloudCLI: Docker-based gcloud CLI interface for GCP operations

Example

Using the gcloud CLI interface:

cli = GCloudCLI(credentials, system_config, docker_client) cli.login(project_name) result = cli.execute(“gcloud functions list”) cli.shutdown()

class sebs.gcp.cli.GCloudCLI(credentials: GCPCredentials, system_config: SeBSConfig, docker_client: DockerClient)[source]

Bases: LoggingBase

Docker-based Google Cloud CLI interface.

Provides a containerized environment for executing gcloud commands with proper authentication and project configuration. Uses a Docker container with the gcloud CLI pre-installed and configured.

docker_instance

Running Docker container with gcloud CLI

execute(cmd: str) → bytes[source]

Execute a command in the gcloud CLI container.

Parameters:

cmd – Command string to execute in the container

Returns:

Command output as bytes

Raises:

RuntimeError – If the command fails (non-zero exit code)

login(project_name: str) → None[source]

Authenticate gcloud CLI and set the active project.

Authenticates using the mounted credentials file (/credentials.json in the container) and then sets the active Google Cloud project. Automatically confirms any prompts that may appear during project setup. .. important:

- `gcloud init` is not used as it requires browser-based authentication.
  Instead, we authenticate as a service account.
- Setting the project might show warnings about Cloud Resource Manager API
  permissions, which are generally not needed for SeBS operations.
Parameters:

project_name – GCP project ID to set as active

Note

Uses service account authentication instead of browser-based auth. May show warnings about Cloud Resource Manager API permissions.

shutdown() → None[source]

Shutdown the gcloud CLI Docker container.

Stops and removes the Docker container used for gcloud operations.

static typename() → str[source]

Get the type name for this CLI implementation.

Returns:

Type name string for GCP CLI

sebs.gcp.config module

Configuration classes for Google Cloud Platform (GCP) integration.

This module provides configuration classes for GCP, including credentials management, resource allocation, and cloud region configuration. It handles authentication through service account JSON files and manages project-specific settings required for Cloud Functions deployment and execution.

Classes:

GCPCredentials: Handles authentication and project identification GCPResources: Manages allocated cloud resources GCPConfig: Main configuration container for GCP deployment

Example

Basic GCP configuration setup:

credentials = GCPCredentials(“/path/to/service-account.json”) resources = GCPResources() config = GCPConfig(credentials, resources)

class sebs.gcp.config.GCPConfig(credentials: GCPCredentials, resources: GCPResources)[source]

Bases: Config

Main configuration class for Google Cloud Platform deployment.

Combines credentials and resources into a complete configuration for GCP serverless function deployment. Manages cloud region settings, authentication, and resource allocation for the benchmarking suite.

This class handles serialization/deserialization for cache persistence and provides validation for configuration consistency across sessions.

_project_name

GCP project identifier

Type:

str

_region

GCP region for resource deployment

Type:

str

_credentials

GCP authentication credentials

_resources

Allocated GCP resources

property credentials: GCPCredentials

Get the GCP credentials instance.

Returns:

GCP authentication credentials

property deployment_config: GCPConfiguration

Get the deployment configuration for GCP workloads.

Returns:

GCPConfiguration instance containing workload deployment settings

static deserialize(config: Dict, cache: Cache, handlers: LoggingHandlers) → Config[source]

Deserialize GCP configuration from dictionary and cache.

Loads complete GCP configuration including credentials and resources. Validates consistency between cached and provided configuration values, updating cache with new user-provided values when they differ.

Parameters:

  • config – Configuration dictionary with GCP settings

  • cache – Cache instance for storing/retrieving configuration

  • handlers – Logging handlers for status reporting

Returns:

Initialized GCPConfig instance

static initialize(cfg: Config, dct: Dict) → None[source]

Initialize GCP configuration from dictionary.

Parameters:

  • cfg – Config instance to initialize (will be cast to GCPConfig)

  • dct – Dictionary containing configuration values including region

property project_name: str

Get the GCP project name from credentials.

Returns:

GCP project identifier string

property region: str

Get the GCP region for resource deployment.

Returns:

GCP region identifier (e.g., ‘us-central1’)

property resources: GCPResources

Get the GCP resources instance.

Returns:

GCP resource allocation settings

serialize() → Dict[source]

Serialize configuration to dictionary for cache storage.

Returns:

Dictionary containing complete GCP configuration including name, region, credentials, and resources

update_cache(cache: Cache) → None[source]

Update cache with current configuration values.

Updates region, credentials, and resources in the cache.

Parameters:

cache – Cache instance to update with configuration data

class sebs.gcp.config.GCPConfiguration[source]

Bases: object

User-provided configuration of workloads on the GCP.

Currently, this class primarily inherits functionality from the base Resources class, as we do not need more GCP-specific resources beyond standard storage buckets.

Inherits all attributes from the base Resources class
property container_config: GCPContainerConfig

Get the Cloud Run container configuration.

Returns:

Cloud Run container deployment configuration object.

property function_gen1_config: GCPFunctionGen1Config

Get the Gen1 deployment configuration.

Returns:

Gen1 deployment configuration object.

property function_gen2_config: GCPFunctionGen2Config

Get the Gen2 deployment configuration.

Returns:

Gen2 deployment configuration object.

static initialize(config: GCPConfiguration, dct: Dict) → GCPConfiguration[source]

Populate a configuration object from serialized data.

Parameters:

  • config – Configuration object to populate.

  • dct – Serialized deployment configuration.

Returns:

The populated configuration object.

serialize() → Dict[source]

Serialize resources to dictionary for cache storage.

Returns:

Dictionary representation of resources for cache storage

class sebs.gcp.config.GCPContainerConfig(environment: str, vcpus: int, gcp_concurrency: int, worker_concurrency: int, worker_threads: int, min_instances: int, max_instances: int, cpu_boost: bool, cpu_throttle: bool)[source]

Bases: GCPFunctionGen2Config

Configuration for Cloud Run container deployments.

static deserialize(dct: Dict) → GCPContainerConfig[source]

Deserialize a container configuration.

Parameters:

dct – Serialized container configuration.

Returns:

Deserialized container configuration object.

serialize() → Dict[source]

Serialize the container configuration.

Returns:

Dictionary containing the serialized container settings.

class sebs.gcp.config.GCPCredentials(gcp_credentials: str)[source]

Bases: Credentials

Credentials manager for Google Cloud Platform authentication.

Handles authentication to GCP services using service account JSON files. Automatically extracts project ID from credentials and manages environment variable setup for Google Cloud SDK authentication.

The class supports multiple credential sources in priority order: 1. User-provided credentials file path 2. GOOGLE_APPLICATION_CREDENTIALS environment variable 3. GCP_SECRET_APPLICATION_CREDENTIALS environment variable

_gcp_credentials

Path to the service account JSON file

_project_id

GCP project ID extracted from credentials

static deserialize(config: Dict, cache: Cache, handlers: LoggingHandlers) → Credentials[source]

Deserialize GCP credentials from configuration and cache.

Loads credentials from multiple sources in priority order: 1. User-provided config with credentials-json path 2. GOOGLE_APPLICATION_CREDENTIALS environment variable 3. GCP_SECRET_APPLICATION_CREDENTIALS environment variable

Sets the GOOGLE_APPLICATION_CREDENTIALS environment variable if credentials are loaded from SeBS config or SeBS-specific environment variables.

Parameters:

  • config – Configuration dictionary potentially containing credentials

  • cache – Cache instance for storing/retrieving credentials

  • handlers – Logging handlers for error reporting

Returns:

Initialized GCPCredentials instance

Raises:

RuntimeError – If no valid credentials are found or if project ID mismatch occurs between cache and new credentials

property gcp_credentials: str

Get the path to the GCP service account credentials file.

Returns:

Path to the service account JSON file

static initialize(gcp_credentials: str) → GCPCredentials[source]

Create a new GCPCredentials instance.

Parameters:

gcp_credentials – Path to the GCP service account JSON file

Returns:

A new GCPCredentials instance

property project_name: str

Get the GCP project ID from the credentials.

Returns:

The GCP project ID string

serialize() → Dict[source]

Serialize credentials to dictionary for cache storage.

Only stores the project_id, as the path to credentials might change or be environment-dependent. It also avoids any potential security issues.

Returns:

Dictionary containing project_id for cache storage

update_cache(cache: Cache) → None[source]

Update the cache with current GCP project id.

Parameters:

cache – Cache instance to update with project ID

class sebs.gcp.config.GCPFunctionGen1Config(min_instances: int = 0, max_instances: int = 20)[source]

Bases: object

Configuration for Cloud Functions Gen1 deployments.

static deserialize(dct: Dict) → GCPFunctionGen1Config[source]

Deserialize a Gen1 configuration.

Parameters:

dct – Serialized Gen1 configuration.

Returns:

Deserialized Gen1 configuration object.

serialize() → Dict[source]

Serialize the Gen1 configuration.

Returns:

Dictionary containing the serialized Gen1 settings.

class sebs.gcp.config.GCPFunctionGen2Config(vcpus: int, gcp_concurrency: int, worker_concurrency: int, worker_threads: int, min_instances: int, max_instances: int, cpu_boost: bool, cpu_throttle: bool)[source]

Bases: object

Configuration for Cloud Functions Gen2 deployments.

static deserialize(dct: Dict) → GCPFunctionGen2Config[source]

Deserialize a Gen2 configuration.

Parameters:

dct – Serialized Gen2 configuration.

Returns:

Deserialized Gen2 configuration object.

serialize() → Dict[source]

Serialize the Gen2 configuration.

Returns:

Dictionary containing the serialized Gen2 settings.

class sebs.gcp.config.GCPResources[source]

Bases: Resources

Resource manager for serverless resources on Google Cloud Platform.

Currently, this class primarily inherits functionality from the base Resources class, as we do not need more GCP-specific resources beyond standard storage buckets.

Inherits all attributes from the base Resources class
check_container_repository_exists(config: Config, ar_client)[source]

Check whether the configured container repository exists.

property container_repository: str

Get the Artifact Registry repository name for containers.

create_container_repository(ar_client, parent)[source]

Create the Artifact Registry repository for benchmark containers.

static deserialize(config: Dict, cache: Cache, handlers: LoggingHandlers) → Resources[source]

Deserialize GCP resources from configuration and cache.

Loads resources from cache if available, otherwise initializes from user configuration or creates empty resource set.

Parameters:

  • config – Configuration dictionary potentially containing resources

  • cache – Cache instance for storing/retrieving resources

  • handlers – Logging handlers for status reporting

Returns:

Initialized GCPResources instance

get_container_repository(config: Config, ar_client)[source]

Get or create the Artifact Registry repository for container images.

static initialize(res: Resources, dct: Dict) → GCPResources[source]

Initialize GCP resources from a dictionary configuration.

Parameters:

  • res – Base Resources instance to initialize

  • dct – Dictionary containing resource configuration

Returns:

Initialized GCPResources instance

serialize() → Dict[source]

Serialize resources to dictionary for cache storage.

Returns:

Dictionary representation of resources for cache storage

update_cache(cache: Cache) → None[source]

Update the cache with current resource information.

Parameters:

cache – Cache instance to update with resource data

sebs.gcp.datastore module

Google Cloud Datastore/Firestore implementation for SeBS NoSQL storage.

This module provides NoSQL database functionality using Google Cloud Firestore in Datastore mode. It manages database allocation, table creation, and data operations for benchmarks requiring NoSQL storage capabilities. To create databases, we use the gcloud CLI instance since there is no API that we could access directly.

Classes:

BenchmarkResources: Resource configuration for benchmark databases Datastore: NoSQL storage implementation using Google Cloud Firestore

Example

Using Datastore for benchmark NoSQL operations:

datastore = Datastore(cli_instance, cache, resources, region) table_name = datastore.create_table(“benchmark-name”, “user-data”, “user_id”) datastore.write_to_table(“benchmark-name”, table_name, data, primary_key, secondary_key)

class sebs.gcp.datastore.BenchmarkResources(database: str, kinds: List[str], database_client: Client | None = None)[source]

Bases: object

Resource configuration for a benchmark’s Datastore database.

Tracks the allocated database name, table kinds, and client instance for a specific benchmark’s NoSQL storage requirements.

database

Name of the Firestore database in Datastore mode

Type:

str

kinds

List of entity kinds (table equivalents) in the database

Type:

List[str]

database_client

Optional Datastore client instance (allocated dynamically)

Type:

google.cloud.datastore.client.Client | None

database: str
database_client: Client | None = None
static deserialize(config: Dict) → BenchmarkResources[source]

Deserialize benchmark resources from cached configuration.

Parameters:

config – Dictionary containing cached resource configuration

Returns:

BenchmarkResources instance with database and kinds

kinds: List[str]
serialize() → Dict[source]

Serialize benchmark resources for cache storage.

Returns:

Dictionary containing database name and kinds list

class sebs.gcp.datastore.Datastore(cli_instance: GCloudCLI, cache_client: Cache, resources: Resources, region: str)[source]

Bases: NoSQLStorage

Google Cloud Firestore/Datastore implementation for NoSQL storage.

Provides NoSQL database functionality using Google Cloud Firestore in Datastore mode. Manages database allocation, entity kind creation, and data operations for benchmarks requiring NoSQL capabilities.

_cli_instance

gcloud CLI interface for database management

_region

GCP region for database allocation

_benchmark_resources

Mapping of benchmarks to their database resources

benchmark_database(benchmark: str) → str[source]

Get the database name for a benchmark.

Parameters:

benchmark – Name of the benchmark

Returns:

Database name for the benchmark’s NoSQL resources

clear_table(name: str) → str[source]

Clear all entities from a table.

Parameters:

name – Name of the table to clear

Returns:

Table name

Raises:

NotImplementedError – This method is not yet implemented

create_table(benchmark: str, name: str, primary_key: str, _: str | None = None) → str[source]

Create a new entity kind (table) in Datastore.

Creates a new Firestore database in Datastore mode if needed using gloud CLI. Datastore kinds are schemaless and created implicitly when an entity of that kind is first written. This method primarily ensures the database exists and registers the kind name for the benchmark. The primary_key is noted but not directly used to create schema for the kind itself, as Datastore is schemaless.

Parameters:

  • benchmark – Name of the benchmark

  • name – Name of the entity kind (table) to create

  • primary_key – Primary key field name

  • _ – Unused parameter for compatibility

Returns:

Name of the created entity kind

Raises:

RuntimeError – If database operations fail

static deployment_name() → str[source]

Get the deployment name for this NoSQL storage implementation.

Returns:

Deployment name string ‘gcp’

get_tables(benchmark: str) → Dict[str, str][source]

Get table name mappings for a benchmark.

GCP Datastore requires no table mappings as the entity kind name is the same as the benchmark table name.

Parameters:

benchmark – Name of the benchmark

Returns:

Empty dictionary (no mappings needed for GCP)

remove_table(name: str) → str[source]

Remove a table from the database.

Parameters:

name – Name of the table to remove

Returns:

Table name

Raises:

NotImplementedError – This method is not yet implemented

retrieve_cache(benchmark: str) → bool[source]

Retrieve benchmark resources from cache.

Parameters:

benchmark – Name of the benchmark to retrieve resources for

Returns:

True if resources were found in cache, False otherwise

static typename() → str[source]

Get the type name for this NoSQL storage implementation.

Returns:

Type name string for GCP Datastore

update_cache(benchmark: str) → None[source]

Update cache with current benchmark resources.

Parameters:

benchmark – Name of the benchmark to cache resources for

write_to_table(benchmark: str, table: str, data: Dict, primary_key: Tuple[str, str], secondary_key: Tuple[str, str] | None = None) → None[source]

Write data to a Datastore entity kind (table).

Parameters:

  • benchmark – Name of the benchmark

  • table – Name of the table (entity kind)

  • data – Dictionary of data to write

  • primary_key – Primary key tuple (name, value)

  • secondary_key – Secondary key tuple (name, value) - required for GCP

Raises:

AssertionError – If secondary_key is None (required for GCP)

sebs.gcp.function module

Google Cloud Platform function implementation for SeBS.

This module provides the GCPFunction class that represents a Cloud Function deployed on Google Cloud Platform. It handles function metadata, serialization, deserialization, and bucket management for code deployment.

Classes:

GCPFunction: Represents a deployed Google Cloud Function with GCP-specific features

Example

Creating a GCP function instance:

config = FunctionConfig(memory=256, timeout=60, runtime=”python39”) function = GCPFunction(“my-function”, “benchmark-name”, “hash123”, config)

class sebs.gcp.function.FunctionDeploymentType(*values)[source]

Bases: str, Enum

Enumeration of deployment methods on GCP.

  • FUNCTION_GEN1: Original Google Cloud Functions.

  • FUNCTION_GEN2: Google Cloud Functions gen2, based on Cloud Run.

  • CONTAINER: Google Cloud Run containers.

CONTAINER = 'container'
FUNCTION_GEN1 = 'function-gen1'
FUNCTION_GEN2 = 'function-gen2'
static deserialize(val: str) → FunctionDeploymentType[source]

Deserialize a string value to a FunctionDeploymentEngine enum.

Parameters:

val – String value to convert to enum

Returns:

Corresponding enum value

Return type:

FunctionDeploymentEngine

Raises:

Exception – If the value doesn’t match any enum member

property is_container: bool

Return whether the deployment type is container-based.

Returns:

True if the deployment uses Cloud Run containers, otherwise False.

class sebs.gcp.function.GCPFunction(name: str, benchmark: str, code_package_hash: str, cfg: FunctionConfig, deployment_type: FunctionDeploymentType, deployment_config: GCPFunctionGen1Config | GCPFunctionGen2Config | GCPContainerConfig, bucket: str | None = None, container_uri: str | None = None)[source]

Bases: Function

Represents a Google Cloud Function with GCP-specific functionality.

Extends the base Function class with GCP-specific features like bucket management for code storage and GCP-specific serialization/deserialization.

bucket

Cloud Storage bucket name containing the function’s code

code_bucket(benchmark: str, storage_client: GCPStorage) → str[source]

Get or create the Cloud Storage bucket for function code.

Returns the bucket name where the function’s code is stored, creating a deployment bucket if none is assigned.

Parameters:

  • benchmark – Benchmark name (unused but kept for compatibility)

  • storage_client – GCP storage client for bucket operations

Returns:

Cloud Storage bucket name containing function code

container_uri() → str | None[source]

Return the container image URI if this function uses one.

Returns:

Container image URI, or None for non-container deployments.

property deployment_config: GCPFunctionGen1Config | GCPFunctionGen2Config | GCPContainerConfig

Get the deployment-specific configuration for this function.

Returns:

Deployment-specific configuration object.

property deployment_type: FunctionDeploymentType

Get the deployment type for this function.

Returns:

Deployment type enum for the function.

static deserialize(cached_config: Dict) → GCPFunction[source]

Deserialize function from cached configuration.

Reconstructs a GCPFunction instance from cached data including triggers and configuration. Handles both Library and HTTP triggers.

Parameters:

cached_config – Dictionary containing cached function configuration

Returns:

Reconstructed GCPFunction instance with triggers

Raises:

AssertionError – If an unknown trigger type is encountered

serialize() → Dict[source]

Serialize function to dictionary for cache storage. Adds code bucket in cloud storage.

Returns:

Dictionary containing function state including bucket information

set_container_uri(container_uri: str | None) → None[source]

Update the container image URI for this function.

Parameters:

container_uri – Container image URI to store, or None.

static typename() → str[source]

Get the type name for this function implementation.

Returns:

Type name string for GCP functions

sebs.gcp.gcp module

Google Cloud Platform (GCP) serverless system implementation.

This module provides the main GCP implementation with function deployment, management, monitoring, and resource allocation. It integrates with Google Cloud Functions, Cloud Storage, Cloud Monitoring, and Cloud Logging.

The module handles: - Function creation, updating, and lifecycle management - Code packaging and deployment to Cloud Functions - HTTP and library trigger management - Performance metrics collection via Cloud Monitoring - Execution logs retrieval via Cloud Logging - Cold start enforcement for benchmarking - Storage bucket management for code deployment

Classes:

GCP: Main system class implementing the FaaS System interface

Example

Basic GCP system initialization:

config = GCPConfig(credentials, resources) gcp_system = GCP(system_config, config, cache, docker_client, logging_handlers) gcp_system.initialize()

class sebs.gcp.gcp.CloudFunctionGen1Strategy(storage: GCPStorage, config: GCPConfig, logging_handlers: ColoredWrapper)[source]

Bases: DeploymentStrategy

Deployment strategy for Google Cloud Functions Gen1.

allow_public_access(project_name: str, location: str, func_name: str) → None[source]

Set IAM policy for public access on Cloud Function Gen1.

create(func_name: str, code_package: Benchmark, function_cfg: FunctionConfig, envs: Dict, container_uri: str | None) → None[source]

Create a Cloud Function Gen1.

create_trigger(func_name: str) → str[source]

Create HTTP trigger and return the invoke URL for Cloud Function.

Parameters:

func_name – Function name

Returns:

HTTP trigger URL

delete_function(func_name: str) → None[source]

Delete a Google Cloud Function.

Parameters:

func_name – Name of the function to delete

download_execution_metrics(function_name: str, start_time: int, end_time: int, requests: Dict) → None[source]

Download execution times for Cloud Functions Gen1 from Cloud Logging.

download_metrics(function_name: str, start_time: int, end_time: int, metrics: Dict) → None[source]

Download monitoring metrics for Cloud Functions Gen1. Use metrics to find estimated values for maximum memory used, active instances and network traffic. https://cloud.google.com/monitoring/api/metrics_gcp#gcp-cloudfunctions

function_exists(project_name: str, location: str, func_name: str) → Any[source]

Check whether the Cloud Function exists.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • func_name – Function name.

Returns:

True if the function exists, otherwise False.

generate_runtime_envs() → Dict[source]

Return runtime environment variables for Gen1 deployments.

Returns:

Empty dictionary because Gen1 does not require runtime overrides.

static get_full_function_name(project_name: str, location: str, func_name: str) → str[source]

Build the fully qualified Cloud Functions resource name.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • func_name – Function name.

Returns:

Fully qualified Cloud Functions resource name.

is_deployed(func_name: str, versionId: int = -1) → Tuple[bool, int][source]

Check if Cloud Function is deployed and optionally verify its version.

Parameters:

  • func_name – Name of the function to check

  • versionId – Optional specific version ID to verify (-1 to check any)

Returns:

Tuple of (is_deployed, current_version_id)

update_code(function: GCPFunction, code_package: Benchmark, envs: Dict, container_uri: str | None) → None[source]

Update Cloud Function Gen1 code.

update_config(function: GCPFunction, envs: Dict) → int[source]

Update Cloud Function Gen1 configuration.

update_envs(full_function_name: str, envs: Dict) → Dict[source]

Merge new environment variables with existing Cloud Function environment.

Parameters:

  • full_function_name – Fully qualified function name

  • envs – New environment variables to add/update

Returns:

Merged environment variables dictionary

wait_for_deployment(func_name: str) → None[source]

Wait for Cloud Function Gen1 deployment via build polling.

class sebs.gcp.gcp.CloudFunctionGen2Strategy(config: GCPConfig, logging_handlers: ColoredWrapper)[source]

Bases: DeploymentStrategy

Deployment strategy for Google Cloud Functions Gen2 package deployments.

allow_public_access(project_name: str, location: str, func_name: str) → None[source]

Grant public invocation access to the underlying Cloud Run service.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • func_name – Function name whose backing service should be public.

create(func_name: str, code_package: Benchmark, function_cfg: FunctionConfig, envs: Dict, container_uri: str | None) → None[source]

Create a new Cloud Functions Gen2 deployment.

Parameters:

  • func_name – Function name to create.

  • code_package – Benchmark package with code to deploy.

  • function_cfg – Function configuration.

  • envs – Environment variables for the function service.

  • container_uri – Unused for package deployments.

create_trigger(func_name: str) → str[source]

Return the HTTPS trigger URL for a Gen2 function.

Parameters:

func_name – Function name.

Returns:

Public invoke URL for the function.

delete_function(func_name: str) → None[source]

Delete a Cloud Functions Gen2 deployment.

Parameters:

func_name – Function name to delete.

download_execution_metrics(function_name: str, start_time: int, end_time: int, requests: Dict) → None[source]

Download execution timings for a Gen2 function from request logs.

Parameters:

  • function_name – Function name whose backing service is queried.

  • start_time – Start timestamp for log collection.

  • end_time – End timestamp for log collection.

  • requests – Invocation results keyed by request ID.

download_metrics(function_name: str, start_time: int, end_time: int, metrics: Dict) → None[source]

Download monitoring metrics for a Gen2 function.

Parameters:

  • function_name – Function name whose backing service is queried.

  • start_time – Start timestamp for metric collection.

  • end_time – End timestamp for metric collection.

  • metrics – Dictionary to populate with monitoring samples.

function_exists(project_name: str, location: str, func_name: str) → Any[source]

Check whether the Cloud Functions Gen2 resource exists.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • func_name – Function name.

Returns:

True if the function exists, otherwise False.

generate_runtime_envs() → Dict[source]

Return runtime environment overrides for Gen2 package deployments.

Returns:

Environment variables controlling Gunicorn worker settings.

static get_full_function_name(project_name: str, location: str, func_name: str) → str[source]

Build the fully qualified Cloud Functions Gen2 resource name.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • func_name – Function name.

Returns:

Fully qualified Cloud Functions Gen2 resource name.

is_deployed(func_name: str, versionId: int = -1) → Tuple[bool, int][source]

Check whether a Gen2 function is deployed and ready.

Parameters:

  • func_name – Function name to inspect.

  • versionId – Unused for Gen2 deployments.

Returns:

Tuple of readiness flag and placeholder version value.

update_code(function: GCPFunction, code_package: Benchmark, envs: Dict, container_uri: str | None) → None[source]

Update the code of an existing Cloud Functions Gen2 deployment.

Parameters:

  • function – Existing deployed function.

  • code_package – New benchmark package to upload.

  • envs – Environment variables for the updated service.

  • container_uri – Unused for package deployments.

update_config(function: GCPFunction, envs: Dict) → int[source]

Update configuration of an existing Cloud Functions Gen2 deployment.

Parameters:

  • function – Deployed function to update.

  • envs – Full environment variable map to apply.

Returns:

Placeholder version value for interface compatibility.

update_envs(full_function_name: str, envs: Dict) → Dict[source]

Merge new environment variables with existing Gen2 service variables.

Parameters:

  • full_function_name – Fully qualified function name.

  • envs – New environment variables to add or override.

Returns:

Merged environment variables dictionary.

wait_for_deployment(func_name: str) → None[source]

Wait for the active create or patch operation to complete.

Parameters:

func_name – Function name being deployed.

class sebs.gcp.gcp.CloudRunMetricsHelper[source]

Bases: object

Helpers shared by Cloud Run services and Cloud Functions gen2.

static download_execution_metrics(logging: ColoredWrapper, project_name: str, service_name: str, start_time: int, end_time: int, requests: Dict, label: str) → None[source]

Download execution times from Cloud Run request logs.

static download_metrics(project_name: str, service_name: str, start_time: int, end_time: int, metrics: Dict) → None[source]

Download Cloud Run monitoring metrics.

static extract_latency_us(entry) → int | None[source]

Extract request latency from a Cloud Run log entry.

Parameters:

entry – Log entry to inspect.

Returns:

Request latency in microseconds, or None if unavailable.

static extract_trace_id(entry) → str | None[source]

Extract the trace ID from a Cloud Run log entry.

Parameters:

entry – Log entry to inspect.

Returns:

Trace ID if present, otherwise None.

class sebs.gcp.gcp.DeploymentStrategy(*args, **kwargs)[source]

Bases: Protocol

Protocol defining the interface for GCP deployment strategies.

Different deployment types (Cloud Function Gen1, Cloud Run, etc.) implement this protocol to handle their specific deployment, update, and management logic.

TRANSIENT_HTTP_CODES: frozenset[int] = frozenset({429, 503})
allow_public_access(project_name: str, location: str, func_name: str) → None[source]

Set IAM policy for public access.

Parameters:

  • func_name – Function/service name

  • full_resource_name – Full GCP resource name

create(func_name: str, code_package: Benchmark, function_cfg: FunctionConfig, envs: Dict, container_uri: str | None) → None[source]

Create function/service without waiting for deployment to complete.

Parameters:

  • func_name – Name for the function/service

  • code_package – Benchmark package with code

  • function_cfg – Function configuration (memory, timeout, etc.)

  • envs – Environment variables

  • container_uri – Container image URI (for container deployments)

create_trigger(func_name: str) → str[source]

Create HTTP trigger and return the invoke URL.

Parameters:

func_name – Function/service name

Returns:

HTTP trigger URL

delete_function(func_name: str) → None[source]

Delete the function/service.

Parameters:

func_name – Function/service name to delete

download_execution_metrics(function_name: str, start_time: int, end_time: int, requests: Dict) → None[source]

Populate provider execution times for completed invocations.

Parameters:

  • function_name – Function or service name.

  • start_time – Start timestamp for metric collection.

  • end_time – End timestamp for metric collection.

  • requests – Invocation results keyed by request ID.

download_metrics(function_name: str, start_time: int, end_time: int, metrics: Dict) → None[source]

Populate metrics with data from cloud monitoring.

Parameters:

  • function_name – Function or service name.

  • start_time – Start timestamp for metric collection.

  • end_time – End timestamp for metric collection.

  • metrics – Dictionary mapping metric names to found values.

function_exists(project_name: str, location: str, func_name: str) → Any[source]

Check whether the function or service exists.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • func_name – Function or service name.

Returns:

True if the resource exists, otherwise False.

generate_runtime_envs() → Dict[source]

Generate deployment-runtime environment variables.

static get_full_function_name(project_name: str, location: str, func_name: str) → str[source]

Generate the fully qualified function name for GCP API calls.

Parameters:

  • project_name – GCP project ID

  • location – GCP region/location

  • func_name – Function name

Returns:

Fully qualified function name in GCP format

is_deployed(func_name: str, versionId: int = -1) → Tuple[bool, int][source]

Check if function/service is deployed.

Parameters:

  • func_name – Function/service name

  • versionId – Optional specific version ID to verify (-1 to check any)

Returns:

Tuple of (is_deployed, current_version_id)

update_code(function: GCPFunction, code_package: Benchmark, envs: Dict, container_uri: str | None) → None[source]

Update function/service code without waiting for deployment to complete.

Parameters:

  • function – Existing function instance

  • code_package – New benchmark package

  • envs – Environment variables

  • container_uri – Container image URI (for container deployments)

update_config(function: GCPFunction, envs: Dict) → int[source]

Update function/service configuration (memory, timeout, env vars).

Parameters:

  • function – Function instance to update

  • envs – Environment variables

Returns:

Version number after update

update_envs(full_function_name: str, envs: Dict) → Dict[source]

Merge new environment variables with existing ones.

Parameters:

  • full_function_name – Full GCP resource name

  • envs – New environment variables to add/update

Returns:

Merged environment variables dictionary

wait_for_deployment(func_name: str) → None[source]

Wait for deployment to complete (build polling for Gen1, operation wait for Run).

Parameters:

func_name – Name of the function/service to wait for

class sebs.gcp.gcp.GCP(system_config: SeBSConfig, config: GCPConfig, cache_client: Cache, docker_client: DockerClient, logging_handlers: LoggingHandlers)[source]

Bases: System

Google Cloud Platform serverless system implementation.

Provides complete integration with Google Cloud Functions including deployment, monitoring, logging, and resource management. Handles code packaging, function lifecycle management, trigger creation, and performance metrics collection.

_config

GCP-specific configuration including credentials and region

function_client

Google Cloud Functions API client

cold_start_counter

Counter for enforcing cold starts in benchmarking

logging_handlers

Logging configuration for status reporting

cached_function(function: Function) → None[source]

Configure a cached function instance for use.

Sets up library triggers for functions loaded from cache, ensuring they have the proper deployment client and logging configuration.

Parameters:

function – Cached function instance to configure

can_reuse_cached_function(cached_function: Function, benchmark: Benchmark) → str | None[source]

Check whether a cached GCP function matches the requested deployment mode.

Parameters:

  • cached_function – Cached function selected from SeBS cache.

  • benchmark – Benchmark requesting the function.

Returns:

if the cached function does not fit the requested deployment type.

Return type:

str

property config: GCPConfig

Get the GCP configuration instance.

Returns:

GCP configuration with credentials and region settings

property container_client: GCRContainer | None

Get the GCP-specific container manager that uses Artifact Registry.

Returns:

Container manager instance.

create_function(code_package: Benchmark, func_name: str, system_variant: SystemVariant, container_uri: str | None) → GCPFunction[source]

Create a new GCP Cloud Function or update existing one.

Deploys a benchmark as a Cloud Function, handling code upload to Cloud Storage, function creation with proper configuration, and IAM policy setup for unauthenticated invocations (HTTP triggers). If the function already exists, updates it instead.

Parameters:

  • code_package – Benchmark package with code and configuration

  • func_name – Name for the Cloud Function

  • system_variant – Selected deployment variant

  • container_uri – Container image URI (unused for GCP)

Returns:

GCPFunction instance representing the deployed function

Raises:

  • NotImplementedError – If the deployment variant is unsupported

  • RuntimeError – If function creation or IAM configuration fails

create_trigger(function: Function, trigger_type: TriggerType) → Trigger[source]

Create a trigger for the given function.

Creates HTTP triggers for Cloud Functions, waiting for function deployment to complete before extracting the trigger URL. Only HTTP triggers are supported here; Library triggers are added by default during function creation.

Parameters:

  • function – Function instance to create trigger for

  • trigger_type – Type of trigger to create (only HTTP supported)

Returns:

Created trigger instance with URL and configuration

Raises:

RuntimeError – If trigger type is not supported

default_function_name(code_package: Benchmark, resources: Resources | None = None) → str[source]

Generate a default function name for the given benchmark.

Creates a standardized function name using resource ID, benchmark name, language, and version information. Formats the name according to GCP Cloud Functions naming requirements.

Parameters:

  • code_package – Benchmark package containing metadata

  • resources – Optional resource configuration for ID generation

Returns:

Formatted function name suitable for GCP Cloud Functions

delete_function(func_name: str, function: Dict) → None[source]

Delete a Google Cloud Function or Cloud Run service.

Parameters:

func_name – Name of the function/service to delete

download_metrics(function_name: str, start_time: int, end_time: int, requests: Dict, metrics: Dict) → None[source]

Download execution metrics and logs from GCP monitoring services.

Retrieves function execution times from Cloud Logging and performance metrics from Cloud Monitoring. Processes logs to extract execution times and collects metrics like memory usage and network egress.

Parameters:

  • function_name – Name of the function to collect metrics for

  • start_time – Start timestamp for metric collection (Unix timestamp)

  • end_time – End timestamp for metric collection (Unix timestamp)

  • requests – Dictionary of requests keyed by execution ID

  • metrics – Dictionary to populate with collected metrics

enforce_cold_start(functions: List[Function], code_package: Benchmark) → None[source]

Enforce cold starts for multiple functions simultaneously.

Updates all provided functions to force cold starts and waits for all deployments to complete before returning.

Parameters:

  • functions – List of functions to enforce cold starts on

  • code_package – Benchmark package for configuration

static format_function_name(func_name: str) → str[source]

Format function name according to GCP Cloud Functions requirements.

Converts function names to comply with GCP naming rules by replacing hyphens and dots with underscores. GCP functions must begin with a letter and can only contain letters, numbers, and underscores.

Parameters:

func_name – Raw function name to format

Returns:

GCP-compliant function name

static function_type() → Type[Function][source]

Get the function class type for this platform.

Returns:

GCPFunction class type

get_function_client()[source]

Get the Cloud Functions v1 API client.

Returns:

Google Cloud Functions v1 API client

get_functions(code_package: Benchmark, function_names: List[str]) → List[Function][source]

Retrieve multiple function instances and ensure they are deployed.

Gets function instances for the provided names and waits for all functions to be in ACTIVE deployment state.

Parameters:

  • code_package – Benchmark package for function creation

  • function_names – List of function names to retrieve

Returns:

List of deployed function instances

get_run_client()[source]

Get the Cloud Run v2 API client.

Returns:

Google Cloud Run v2 API client

static helper_zip(base_directory: str, path: str, archive: ZipFile) → None[source]

Recursively add files and directories to a zip archive.

Helper method for recursive_zip that handles directory traversal and adds files with relative paths to the archive.

Parameters:

  • base_directory – Base directory path for relative path calculation

  • path – Current path being processed (file or directory)

  • archive – ZipFile object to add files to

initialize(config: Dict[str, str] = {}, resource_prefix: str | None = None, quiet: bool = False) → None[source]

Initialize the GCP system for function deployment and management.

Sets up the Cloud Functions API client and initializes system resources including storage buckets and other required infrastructure. After this call, the GCP system should be ready to allocate functions, manage storage, and invoke functions.

Parameters:

  • config – Additional system-specific configuration parameters

  • resource_prefix – Optional prefix for resource naming to avoid conflicts

is_configuration_changed(cached_function: Function, benchmark: Benchmark) → bool[source]

Override the default implementation.

In addition to checking for timeout and language runtime - the shared config - we also check if the GCP-specific config has changed.

Parameters:

  • cached_function – Previously cached function configuration

  • benchmark – Current benchmark configuration to compare against

Returns:

True if configuration has changed and function needs updating

is_deployed(function: Function, versionId: int = -1) → Tuple[bool, int][source]

Check if a function is deployed and optionally verify its version. :param func_name: Name of the function to check :param versionId: Optional specific version ID to verify (-1 to check any)

Returns:

Tuple of (is_deployed, current_version_id)

static name() → str[source]

Get the platform name identifier.

Returns:

Platform name string ‘gcp’

package_code(directory: str, language: Language, language_version: str, architecture: str, benchmark: str, is_cached: bool) → Tuple[str, float][source]

Package benchmark code for GCP Cloud Functions deployment.

Transforms the benchmark code directory structure to meet GCP Cloud Functions requirements. Creates a zip archive with the appropriate handler file naming and directory structure for the specified language runtime.

The packaging process: 1. Creates a ‘function’ subdirectory for benchmark sources 2. Renames handler files to GCP-required names (handler.py -> main.py) 3. Creates a zip archive for deployment 4. Restores original file structure

Parameters:

  • directory – Path to the benchmark code directory

  • language – Programming language (python, nodejs)

  • language_version – Language version (e.g., ‘3.8’, ‘14’)

  • architecture – Target architecture (x86_64, arm64)

  • benchmark – Benchmark name for archive naming

  • is_cached – Whether this package is from cache

Returns:

Tuple of (archive_path, archive_size_bytes)

static recursive_zip(directory: str, archname: str) → bool[source]

Create a zip archive of a directory with relative paths.

Creates a compressed zip archive of the specified directory, preserving the relative directory structure. Uses maximum compression level.

Parameters:

  • directory – Absolute path to the directory to be zipped

  • archname – Path where the zip file should be created

Returns:

True if archiving was successful

shutdown() → None[source]

Shutdown the GCP system and clean up resources.

Performs cleanup of system resources and calls parent shutdown method.

system_variant_suffix(system_variant: SystemVariant) → str | None[source]

Return a provider-local system variant suffix for GCP package variants.

Parameters:

system_variant – Selected deployment variant.

Returns:

Short suffix for GCP package variants, otherwise None.

static typename() → str[source]

Get the platform type name for display.

Returns:

Platform type string ‘GCP’

update_function(function: Function, code_package: Benchmark, system_variant: SystemVariant, container_uri: str | None) → None[source]

Update an existing Cloud Function with new code and configuration.

Uploads new code package to Cloud Storage and patches the existing function with updated runtime, memory, timeout, and environment variables. Waits for deployment to complete before returning.

Parameters:

  • function – Existing function instance to update

  • code_package – New benchmark package with updated code

  • system_variant – Selected deployment variant

  • container_uri – Container image URI (unused)

Raises:

  • NotImplementedError – If the deployment variant is unsupported

  • RuntimeError – If function update fails after maximum retries

update_function_configuration(function: Function, code_package: Benchmark, env_variables: Dict = {}) → int[source]

Update function configuration including memory, timeout, and environment.

Updates the Cloud Function’s memory allocation, timeout, and environment variables without changing the code. Waits for deployment to complete.

Parameters:

  • function – Function instance to update

  • code_package – Benchmark package with configuration requirements

  • env_variables – Additional environment variables to set

  • container_uri – Container image URI (for container deployments)

Returns:

Version ID of the updated function

Raises:

RuntimeError – If configuration update fails after maximum retries

class sebs.gcp.gcp.RunContainerStrategy(config: GCPConfig, logging_handlers: ColoredWrapper)[source]

Bases: DeploymentStrategy

Deployment strategy for Google Cloud Run containers (Gen1).

allow_public_access(project_name: str, location: str, func_name: str) → None[source]

Set IAM policy for public access on Cloud Run.

create(func_name: str, code_package: Benchmark, function_cfg: FunctionConfig, envs: Dict, container_uri: str | None) → None[source]

Create a Cloud Run service.

create_trigger(func_name: str) → str[source]

Create HTTP trigger and return the invoke URL for Cloud Run service.

Parameters:

func_name – Service name

Returns:

HTTP trigger URL

delete_function(func_name: str) → None[source]

Delete a Cloud Run service.

Parameters:

func_name – Name of the service to delete

download_execution_metrics(function_name: str, start_time: int, end_time: int, requests: Dict) → None[source]

Download execution times for Cloud Run from request logs.

download_metrics(function_name: str, start_time: int, end_time: int, metrics: Dict) → None[source]

Download monitoring metrics for a Cloud Run service.

Parameters:

  • function_name – Name of the deployed service.

  • start_time – Start timestamp for metric collection.

  • end_time – End timestamp for metric collection.

  • metrics – Dictionary to populate with monitoring samples.

function_exists(project_name: str, location: str, func_name: str) → Any[source]

Check whether the Cloud Run service exists.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • func_name – Cloud Run service name.

Returns:

True if the service exists, otherwise False.

generate_runtime_envs() → Dict[source]

Return runtime environment variables for Cloud Run deployments.

Returns:

Runtime environment variables for Gunicorn worker configuration.

static get_full_function_name(project_name: str, location: str, service_name: str) → str[source]

Build the fully qualified Cloud Run service resource name.

Parameters:

  • project_name – GCP project ID.

  • location – GCP region/location.

  • service_name – Cloud Run service name.

Returns:

Fully qualified Cloud Run service resource name.

is_deployed(func_name: str, versionId: int = -1) → Tuple[bool, int][source]

Check if Cloud Run service is deployed.

Parameters:

  • func_name – Name of the service to check

  • versionId – Ignored for Cloud Run (always returns 0)

Returns:

Tuple of (is_deployed, 0)

update_code(function: GCPFunction, code_package: Benchmark, envs: Dict, container_uri: str | None) → None[source]

Update Cloud Run service code.

update_config(function: GCPFunction, envs: Dict) → int[source]

Update Cloud Run service configuration.

update_envs(full_function_name: str, envs: Dict) → Dict[source]

Merge new environment variables with existing Cloud Run service environment.

Parameters:

  • full_function_name – Fully qualified service name

  • envs – New environment variables to add/update

Returns:

Merged environment variables dictionary

wait_for_deployment(func_name: str) → None[source]

Wait for Cloud Run deployment via operation wait.

sebs.gcp.resources module

System resource management for Google Cloud Platform.

This module provides the GCPSystemResources class that manages all GCP resources required for serverless benchmarking, including storage, NoSQL databases, and CLI tools.

Classes:

GCPSystemResources: Main resource manager for GCP services

Example

Creating and using GCP system resources:

resources = GCPSystemResources(system_config, gcp_config, cache, docker_client, handlers) storage = resources.get_storage(replace_existing=False) datastore = resources.get_nosql_storage()

class sebs.gcp.resources.GCPSystemResources(system_config: SeBSConfig, config: GCPConfig, cache_client: Cache, docker_client: DockerClient, logger_handlers: LoggingHandlers)[source]

Bases: SystemResources

System resource manager for Google Cloud Platform services.

Handles resource initialization, configuration, and cleanup.

_storage

Cloud Storage instance for object storage

_nosql_storage

Datastore instance for NoSQL operations

_cli_instance

gcloud CLI interface for administrative operations

_system_config

SeBS system configuration

_logging_handlers

Logging configuration

property cli_instance: GCloudCLI

Get or create the gcloud CLI instance.

Provides access to a containerized gcloud CLI for administrative operations. Initializes and authenticates the CLI if needed.

Returns:

Authenticated gcloud CLI instance

property config: GCPConfig

Get the GCP configuration instance.

Returns:

GCP configuration with credentials and settings

get_nosql_storage() → Datastore[source]

Get or create the Datastore instance for NoSQL operations.

Provides access to Google Cloud Firestore/Datastore for NoSQL database operations required by benchmarks.

Returns:

Initialized Datastore instance

get_storage(replace_existing: bool | None = None) → GCPStorage[source]

Get or create the Cloud Storage instance.

Provides access to Google Cloud Storage for persistent object storage. Creates the storage instance if it doesn’t exist, or updates the replace_existing setting if provided.

Parameters:

replace_existing – Whether to replace existing benchmark input data

Returns:

Initialized GCP storage instance

initialize_cli(cli: GCloudCLI) → None[source]

Initialize with an existing CLI instance.

Uses a pre-configured CLI instance instead of creating a new one.

Parameters:

cli – Pre-configured gcloud CLI instance

shutdown() → None[source]

Shutdown system resources and clean up.

Stops the gcloud CLI container if it was created by this instance.

static typename() → str[source]

Get the type name for this resource manager.

Returns:

Type name string for GCP system resources

sebs.gcp.storage module

Google Cloud Storage implementation for SeBS.

This module provides the GCPStorage class that implements object storage operations using Google Cloud Storage. It handles bucket management, file uploads/downloads, and storage resource allocation for benchmarks and deployment artifacts.

Classes:

GCPStorage: Google Cloud Storage implementation with bucket and blob management

Example

Using GCP storage for benchmark files:

storage = GCPStorage(region, cache, resources, replace_existing=False) bucket = storage.add_benchmark_bucket(“my-benchmark”) storage.upload(bucket, “/path/to/file.zip”, “benchmark-code.zip”)

class sebs.gcp.storage.GCPStorage(region: str, cache_client: Cache, resources: Resources, replace_existing: bool)[source]

Bases: PersistentStorage

Google Cloud Storage implementation providing persistent storage.

Handles bucket creation, file operations, and storage resource management for benchmarks, deployment artifacts, and experiment outputs.

client

Google Cloud Storage client instance

cached

Whether storage operations use cached data

clean_bucket(bucket: str) → None[source]

Clean all objects from a Cloud Storage bucket.

Parameters:

bucket – Name of the bucket to clean

Raises:

NotImplementedError – This method is not yet implemented

correct_name(name: str) → str[source]

Correct bucket name to meet GCP naming requirements. Currently it does nothing - no special requirements on GCP.

Parameters:

name – Original bucket name

Returns:

Corrected bucket name (no changes needed for GCP)

static deployment_name() → str[source]

Get the deployment name for this storage implementation.

Returns:

Deployment name string ‘gcp’

download(bucket_name: str, key: str, filepath: str) → None[source]

Download a file from Cloud Storage.

Parameters:

  • bucket_name – Name of the storage bucket

  • key – Object key/path in the bucket

  • filepath – Local file path to save the downloaded file

exists_bucket(bucket_name: str) → bool[source]

Check if a Cloud Storage bucket exists.

Handles exceptions.Forbidden which can occur if the bucket exists but is not accessible by the current credentials (treated as not existing for SeBS purposes).

Parameters:

bucket_name – Name of the bucket to check

Returns:

True if bucket exists and is accessible, False otherwise

list_bucket(bucket_name: str, prefix: str = '') → List[str][source]

List objects in a Cloud Storage bucket with optional prefix filter.

Parameters:

  • bucket_name – Name of the bucket to list

  • prefix – Optional prefix to filter objects

Returns:

List of object names in the bucket matching the prefix

list_buckets(bucket_name: str | None = None) → List[str][source]

List Cloud Storage buckets, optionally filtered by name.

Parameters:

bucket_name – Optional bucket name filter

Returns:

List of bucket names, filtered if bucket_name is provided

remove_bucket(bucket_name: str) → None[source]

Remove a Cloud Storage bucket.

Parameters:

bucket_name – Name of the bucket to remove

property replace_existing: bool

Flag indicating whether to replace existing files in buckets.

static typename() → str[source]

Get the type name for this storage implementation.

Returns:

Type name string for GCP storage

upload(bucket_name: str, filepath: str, key: str) → None[source]

Upload a file to Cloud Storage.

Parameters:

  • bucket_name – Name of the storage bucket

  • filepath – Local file path to upload

  • key – Object key/path in the bucket for the uploaded file

uploader_func(path_idx: int, key: str, filepath: str) → None[source]

Upload function for batch operations with caching support.

Uploads a file to the appropriate benchmark bucket, respecting cache settings and replace_existing configuration.

This is primarily used by benchmarks to upload input data.

Parameters:

  • path_idx – Index of the input path prefix

  • key – Object key for the uploaded file

  • filepath – Local file path to upload

sebs.gcp.triggers module

Trigger implementations for Google Cloud Functions.

This module provides trigger classes for invoking Cloud Functions through different mechanisms including direct library calls and HTTP requests. Supports both synchronous and asynchronous invocation patterns.

Classes:

LibraryTrigger: Direct Cloud Functions API invocation trigger HTTPTrigger: HTTP endpoint invocation trigger

Example

Using a library trigger for direct invocation:

trigger = LibraryTrigger(“my-function”, gcp_client) result = trigger.sync_invoke({“input”: “data”})

Using an HTTP trigger:

trigger = HTTPTrigger(”https://region-project.cloudfunctions.net/my-function”) result = trigger.sync_invoke({“input”: “data”})

class sebs.gcp.triggers.HTTPTrigger(url: str)[source]

Bases: Trigger

HTTP endpoint trigger for Cloud Functions invocation.

Invokes Cloud Functions through their HTTP endpoints, supporting both synchronous and asynchronous execution patterns using HTTP requests.

url

HTTP endpoint URL for the Cloud Function

async_invoke(payload: Dict) → Future[source]

Asynchronously invoke the Cloud Function via HTTP.

Parameters:

payload – Input data to send to the function

Returns:

Future object for the async HTTP invocation

static deserialize(obj: Dict) → Trigger[source]

Deserialize trigger from cached configuration.

Parameters:

obj – Dictionary containing serialized trigger data

Returns:

Reconstructed HTTPTrigger instance

serialize() → Dict[source]

Serialize trigger to dictionary for cache storage.

Returns:

Dictionary containing trigger type and URL

sync_invoke(payload: Dict) → ExecutionResult[source]

Synchronously invoke the Cloud Function via HTTP.

Parameters:

payload – Input data to send to the function

Returns:

ExecutionResult from the HTTP invocation

static trigger_type() → TriggerType[source]

Get the trigger type for this implementation.

Returns:

HTTP trigger type enum value

static typename() → str[source]

Get the type name for this trigger implementation.

Returns:

Type name string for HTTP triggers

class sebs.gcp.triggers.LibraryTrigger(fname: str, deployment_client: GCP | None = None, deployment_type: FunctionDeploymentType | None = None)[source]

Bases: Trigger

Direct Cloud Functions API trigger for synchronous invocation.

Uses the Google Cloud Functions API to invoke functions directly through the cloud functions client. Provides precise execution timing and error handling. Waits for function deployment before invocation.

name

Function name to invoke

_deployment_client

GCP client for API operations

_deployment_type

Type of deployment (gen1 function or container)

async_invoke(payload: Dict) → Future[source]

Asynchronously invoke the Cloud Function.

Uses a ThreadPoolExecutor to run sync_invoke in the background.

Parameters:

payload – Input data to send to the function

Returns:

Future object for the async invocation

property deployment_client: GCP

Get the GCP deployment client.

Returns:

GCP client instance for API operations

Raises:

AssertionError – If deployment client is not set

property deployment_type: FunctionDeploymentType

Get the deployment type associated with this trigger.

Returns:

Deployment type currently associated with the trigger.

static deserialize(obj: Dict) → Trigger[source]

Deserialize trigger from cached configuration.

Parameters:

obj – Dictionary containing serialized trigger data

Returns:

Reconstructed LibraryTrigger instance

serialize() → Dict[source]

Serialize trigger to dictionary for cache storage.

Returns:

Dictionary containing trigger type, name, and deployment type

sync_invoke(payload: Dict) → ExecutionResult[source]

Synchronously invoke the Cloud Function using the API.

Waits for function deployment, then invokes via appropriate API based on deployment type: - FUNCTION_GEN1: Cloud Functions v1 API - CONTAINER: Cloud Run service via HTTP - FUNCTION_GEN2: Not yet supported

Parameters:

payload – Input data to send to the function

Returns:

ExecutionResult with timing, output, and error information

Raises:

NotImplementedError – If deployment type is not supported

static trigger_type() → TriggerType[source]

Get the trigger type for this implementation.

Returns:

Library trigger type enum value

static typename() → str[source]

Get the type name for this trigger implementation.

Returns:

Type name string for library triggers

sebs.gcp.triggers.normalize_request_id(res: ExecutionResult) → None[source]

Normalize the request identifier to the trace ID format used by GCP logs.

Parameters:

res – Execution result object to update in place.

Module contents

Google Cloud Platform (GCP) integration for SeBS.

This package provides comprehensive Google Cloud Platform support, including Cloud Functions deployment, Cloud Storage for object storage, Firestore/Datastore for NoSQL operations, and Cloud Monitoring for performance metrics collection.

The package includes: - Function deployment and management via Cloud Functions API - Object storage through Google Cloud Storage buckets - NoSQL database operations using Firestore in Datastore mode - Performance monitoring via Cloud Monitoring and Cloud Logging - Docker-based gcloud CLI integration for administrative operations - Credential and resource management

Modules:

gcp: Main GCP system implementation config: Configuration and credential management storage: Cloud Storage integration function: Cloud Function representation triggers: Function invocation triggers datastore: Firestore/Datastore NoSQL implementation resources: System resource management cli: gcloud CLI integration

Example

Basic GCP system setup:

from sebs.gcp import GCP, GCPConfig

# Configure GCP with credentials config = GCPConfig.deserialize(config_dict, cache, handlers)

# Initialize GCP system gcp_system = GCP(system_config, config, cache, docker_client, handlers) gcp_system.initialize()

# Deploy a function function = gcp_system.create_function(benchmark, “my-function”, False, “”)