diff --git a/docs/dev_workflow_guide.md b/docs/dev_workflow_guide.md index b06ef1c..d1379db 100644 --- a/docs/dev_workflow_guide.md +++ b/docs/dev_workflow_guide.md @@ -20,58 +20,56 @@ Its job is to expose the REST API, run the discrete-event simulation, talk to th ``` fastsim-backend/ -├── Dockerfile -├── docker_fs/ # docker-compose for dev & prod -│ ├── docker-compose.dev.yml -│ └── docker-compose.prod.yml -├── scripts/ # helper bash scripts (lint, dev-startup, …) -│ ├── init-docker-dev.sh +├── example/ # examples of working simulations +│ ├── data +├── scripts/ # helper bash scripts (lint, dev-startup, …) │ └── quality-check.sh -├── alembic/ # DB migrations (versions/ contains revision files) -│ ├── env.py -│ └── versions/ -├── documentation/ # project vision & low-level docs -│ └── backend_documentation/ -│ └── … -├── tests/ # unit & integration tests +├── docs/ # project vision & low-level docs +│ └── fastsim-documentation/ +├── tests/ # unit & integration tests │ ├── unit/ │ └── integration/ -├── src/ # **application code lives here** +├── src/ # application code lives here │ └── app/ -│ ├── api/ # FastAPI routers & endpoint handlers -│ ├── config/ # Pydantic Settings + constants -│ ├── db/ # SQLAlchemy base, sessions, initial seed utilities -│ ├── metrics/ # helpers to compute/aggregate simulation KPIs -│ ├── resources/ # SimPy resource registry (CPU/RAM containers, etc.) -│ ├── runtime/ # simulation core -│ │ ├── rqs_state.py # RequestState & Hop -│ │ └── actors/ # SimPy “actors”: Edge, Server, Client, RqsGenerator -│ ├── samplers/ # stochastic samplers (Gaussian-Poisson, etc.) -│ ├── schemas/ # Pydantic input/output models -│ ├── main.py # FastAPI application factory / ASGI entry-point -│ └── simulation_run.py # CLI utility to run a sim outside of HTTP layer +│ ├── config/ # Pydantic Settings + constants +│ ├── metrics/ # logic to compute/aggregate simulation KPIs +│ ├── resources/ # SimPy resource registry (CPU/RAM containers, etc.) +│ ├── runtime/ # simulation core +│ │ ├── rqs_state.py # RequestState & Hop +│ │ ├── simulation_runner.py # logic to initialize the whole simulation + | └── actors/ # SimPy “actors”: Edge, Server, Client, RqsGenerator +│ ├── samplers/ # stochastic samplers (Gaussian-Poisson, etc.) +│ ├── schemas/ # Pydantic input/output models ├── poetry.lock ├── pyproject.toml └── README.md ``` +### **What each top-level directory in `src/app` does** + +| Directory | Purpose | +| ---------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| **`config/`** | Centralised configuration layer. Contains Pydantic `BaseSettings` classes for reading environment variables and constants/enums used across the simulation engine. | +| **`metrics/`** | Post-processing and analytics. Aggregates raw simulation traces into KPIs such as latency percentiles, throughput, resource utilisation, and other performance metrics. | +| **`resources/`** | Runtime resource registry for simulated hardware components (e.g., SimPy `Container`s for CPU and RAM). Decouples resource management from actor behaviour. | +| **`runtime/`** | Core simulation engine. Orchestrates SimPy execution, maintains request state, and wires together simulation components. Includes: | +| | - **`rqs_state.py`** — Defines `RequestState` and `Hop` for tracking request lifecycle. | +| | - **`simulation_runner.py`** — Entry point for initialising and running simulations. | +| | - **`actors/`** — SimPy actor classes representing system components (`RqsGenerator`, `Client`, `Server`, `Edge`) and their behaviour. | +| **`samplers/`** | Random-variable samplers for stochastic simulation. Supports Poisson, Normal, and mixed distributions for modelling inter-arrival times and service steps. | +| **`schemas/`** | Pydantic models for input/output validation and serialisation. Includes scenario definitions, topology graphs, simulation settings, and results payloads. | -#### What each top-level directory in `src/app` does +--- -| Directory | Purpose | -| ----------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **`api/`** | Defines the public HTTP surface. Each module holds a router with path operations and dependency wiring. | -| **`config/`** | Centralised configuration: `settings.py` (Pydantic `BaseSettings`) reads env vars; `constants.py` stores enums and global literals. | -| **`db/`** | Persistence layer. Contains the SQLAlchemy base class, the session factory, and a thin wrapper that seeds or resets the database (Alembic migration scripts live at project root). | -| **`metrics/`** | Post-processing helpers that turn raw simulation traces into aggregated KPIs (latency percentiles, cost per request, utilisation curves, …). | -| **`resources/`** | A tiny run-time registry mapping every simulated server to its SimPy `Container`s (CPU, RAM). Keeps resource management separate from actor logic. | -| **`runtime/`** | The heart of the simulator. `rqs_state.py` holds the mutable `RequestState`; sub-package **`actors/`** contains each SimPy process class (Generator, Edge, Server, Client). | -| **`samplers/`** | Probability-distribution utilities that generate inter-arrival and service-time samples—used by the actors during simulation. | -| **`schemas/`** | All Pydantic models for validation and (de)serialisation: request DTOs, topology definitions, simulation settings, outputs. | -| **`main.py`** | Creates and returns the FastAPI app; imported by Uvicorn/Gunicorn. | -| **`simulation_run.py`** | Convenience script to launch a simulation offline (e.g. inside tests or CLI). | +### **Other Top-Level Directories** -Everything under `src/` is import-safe thanks to Poetry’s `packages = [{ include = "app" }]` entry in `pyproject.toml`. +| Directory | Purpose | +| -------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- | +| **`example/`** | Ready-to-run simulation scenarios and example configurations. Includes `data/` with YAML definitions and scripts to demonstrate engine usage. | +| **`scripts/`** | Utility shell scripts for development workflow, linting, formatting, and local startup (`quality-check.sh`, etc.). | +| **`docs/`** | Project documentation. Contains both high-level vision documents and low-level technical references (`fastsim-documentation/`). | +| **`tests/`** | Automated test suite, split into **unit** and **integration** tests to verify correctness of both individual components and end-to-end scenarios. | +--- ## 3. Branching Strategy: Git Flow @@ -182,21 +180,11 @@ We will start to describe the CI part related to push and PR in the develop bran * **Full Suite (push to `develop`)** *Runs in a few minutes; includes real services and Docker.* - - * All steps from the Quick Suite - * PostgreSQL service container started via `services:` - * Alembic migrations applied to the test database + * Full test suite, including `@pytest.mark.integration` tests - * Multi-stage Docker build of the backend image - * Smoke test: container started with Uvicorn → `curl /health` + -### 4.1.3 Key Implementation Details -* **Service containers** – PostgreSQL 17 is spun up in CI with a health-check to ensure migrations run against a live instance. -* **Test markers** – integration tests are isolated with `@pytest.mark.integration`, enabling selective execution. -* **Caching** – Poetry’s download cache is restored to cut installation time; Docker layer cache is reused between builds. -* **Smoke test logic** – after the image is built, CI launches it in detached mode, polls the `/health` endpoint, prints logs, and stops the container. The job fails if the endpoint is unreachable. -* **Secrets management** – database credentials and registry tokens are stored in GitHub Secrets and injected as environment variables only at runtime. diff --git a/docs/fastsim-docs/requests_generator.md b/docs/fastsim-docs/requests_generator.md index 95190fa..a15e241 100644 --- a/docs/fastsim-docs/requests_generator.md +++ b/docs/fastsim-docs/requests_generator.md @@ -8,43 +8,151 @@ This document describes the design of the **requests generator**, which models a Following the FastSim philosophy, we accept a small set of input parameters to drive a “what-if” analysis in a pre-production environment. These inputs let you explore reliability and cost implications under different traffic scenarios. -**Inputs** +## **Inputs** -1. **Average concurrent users** – expected number of users (or sessions) simultaneously hitting the endpoint. -2. **Average requests per minute per user** – average number of requests each user issues per minute. -3. **Simulation time** – total duration of the simulation, in seconds. +1. **Average Concurrent Users (`avg_active_users`)** + Expected number of simultaneous active users (or sessions) interacting with the system. -**Output** -A continuous sequence of timestamps (seconds) marking individual request arrivals. + * Modeled as a random variable (`RVConfig`). + * Allowed distributions: **Poisson** or **Normal**. + +2. **Average Requests per Minute per User (`avg_request_per_minute_per_user`)** + Average request rate per user, expressed in requests per minute. + + * Modeled as a random variable (`RVConfig`). + * **Must** use the **Poisson** distribution. + +3. **User Sampling Window (`user_sampling_window`)** + Time interval (in seconds) over which active users are resampled. + + * Constrained between `MIN_USER_SAMPLING_WINDOW` and `MAX_USER_SAMPLING_WINDOW`. + * Defaults to `USER_SAMPLING_WINDOW`. --- -## Model Assumptions +## **Model Assumptions** -* *Concurrent users* and *requests per minute per user* are **random variables**. -* *Simulation time* is **deterministic**. +* **Random variables**: -We model: + * *Concurrent users* and *requests per minute per user* are independent random variables. + * Each is configured via the `RVConfig` model, which specifies: -* **Requests per minute per user** as Poisson($\lambda_r$). -* **Concurrent users** as either Poisson($\lambda_u$) or truncated Normal. -* **The variables are independent** + * **mean** (mandatory, must be numeric and positive), + * **distribution** (default: Poisson), + * **variance** (optional; defaults to `mean` for Normal and Log-Normal distributions). -```python -from pydantic import BaseModel -from typing import Literal +* **Supported joint sampling cases**: + + * Poisson (users) × Poisson (requests) + * Normal (users) × Poisson (requests) + + Other combinations are currently unsupported. + +* **Variance handling**: + + * If the distribution is **Normal** or **Log-Normal** and `variance` is not provided, it is automatically set to the `mean`. + +--- + +## **Validation Rules** +* `avg_request_per_minute_per_user`: + + * **Must** be Poisson-distributed. + * Validation enforces this constraint. + +* `avg_active_users`: + + * Must be either Poisson or Normal. + * Validation enforces this constraint. + +* `mean` in `RVConfig`: + + * Must be a positive number (int or float). + * Automatically coerced to `float`. + +```python class RVConfig(BaseModel): - """Configure a random-variable parameter.""" + """class to configure random variables""" + mean: float - distribution: Literal["poisson", "normal", "gaussian"] = "poisson" - variance: float | None = None # required only for normal/gaussian + distribution: Distribution = Distribution.POISSON + variance: float | None = None + + @field_validator("mean", mode="before") + def ensure_mean_is_numeric_and_positive( + cls, # noqa: N805 + v: float, + ) -> float: + """Ensure `mean` is numeric, then coerce to float.""" + err_msg = "mean must be a number (int or float)" + if not isinstance(v, (float, int)): + raise ValueError(err_msg) # noqa: TRY004 + + return float(v) + + @model_validator(mode="after") # type: ignore[arg-type] + def default_variance(cls, model: "RVConfig") -> "RVConfig": # noqa: N805 + """Set variance = mean when distribution require and variance is missing.""" + needs_variance: set[Distribution] = { + Distribution.NORMAL, + Distribution.LOG_NORMAL, + } + + if model.variance is None and model.distribution in needs_variance: + model.variance = model.mean + return model + class RqsGeneratorInput(BaseModel): - """Define simulation inputs.""" + """Define the expected variables for the simulation""" + + id: str + type: SystemNodes = SystemNodes.GENERATOR avg_active_users: RVConfig avg_request_per_minute_per_user: RVConfig - total_simulation_time: int | None = None + + user_sampling_window: int = Field( + default=TimeDefaults.USER_SAMPLING_WINDOW, + ge=TimeDefaults.MIN_USER_SAMPLING_WINDOW, + le=TimeDefaults.MAX_USER_SAMPLING_WINDOW, + description=( + "Sampling window in seconds " + f"({TimeDefaults.MIN_USER_SAMPLING_WINDOW}-" + f"{TimeDefaults.MAX_USER_SAMPLING_WINDOW})." + ), + ) + + @field_validator("avg_request_per_minute_per_user", mode="after") + def ensure_avg_request_is_poisson( + cls, # noqa: N805 + v: RVConfig, + ) -> RVConfig: + """ + Force the distribution for the rqs generator to be poisson + at the moment we have a joint sampler just for the poisson-poisson + and gaussian-poisson case + """ + if v.distribution != Distribution.POISSON: + msg = "At the moment the variable avg request must be Poisson" + raise ValueError(msg) + return v + + @field_validator("avg_active_users", mode="after") + def ensure_avg_user_is_poisson_or_gaussian( + cls, # noqa: N805 + v: RVConfig, + ) -> RVConfig: + """ + Force the distribution for the rqs generator to be poisson + at the moment we have a joint sampler just for the poisson-poisson + and gaussian-poisson case + """ + if v.distribution not in {Distribution.POISSON, Distribution.NORMAL}: + msg = "At the moment the variable active user must be Poisson or Gaussian" + raise ValueError(msg) + return v + ``` --- diff --git a/docs/fastsim-docs/simulation_input.md b/docs/fastsim-docs/simulation_input.md index aa27d58..ad89247 100644 --- a/docs/fastsim-docs/simulation_input.md +++ b/docs/fastsim-docs/simulation_input.md @@ -1,228 +1,349 @@ -Of course. Here is the complete documentation, translated into English, based on the new Pydantic schemas. +# FastSim — Simulation Input Schema (v2) ------ +This document describes the **complete input contract** used by FastSim to run a simulation, the **design rationale** behind it, and the **guarantees** provided by the validation layer. It closes with an **end-to-end example** (YAML) you can drop into the project and run as-is. -### **FastSim — Simulation Input Schema** +The entry point is the Pydantic model: -The `SimulationPayload` is the single, self-contained contract that defines an entire simulation run. Its architecture is guided by a core philosophy: to achieve maximum control over input data through robust, upfront validation. To implement this, we extensively leverage Pydantic's powerful validation capabilities and Python's `Enum` classes. This approach creates a strictly-typed and self-consistent schema that guarantees any configuration is validated *before* the simulation engine starts. +```python +class SimulationPayload(BaseModel): + """Full input structure to perform a simulation""" + rqs_input: RqsGeneratorInput + topology_graph: TopologyGraph + sim_settings: SimulationSettings +``` + +Everything the engine needs is captured by these three components: + +* **`rqs_input`** — the workload model (how traffic is generated). +* **`topology_graph`** — the system under test, described as a directed graph (nodes & edges). +* **`sim_settings`** — global simulation controls and which metrics to collect. + +--- + +## Why this shape? (Rationale) + +### 1) **Separation of concerns** + +* **Workload** (traffic intensity & arrival process) is independent from the **topology** (architecture under test) and from **simulation control** (duration & metrics). +* This lets you reuse the same topology with different workloads, or vice versa, without touching unrelated parts. + +### 2) **Validation-first, fail-fast** + +* All inputs are **typed** and **validated** with Pydantic before the engine starts. +* Validation catches type errors, inconsistent references, and illegal combinations (e.g., an I/O step with a CPU metric). +* When a payload parses successfully, the engine can run without defensive checks scattered in runtime code. + +### 3) **Small-to-large composition** + +* The smallest unit is a **`Step`** (one resource-bound operation). +* Steps compose into an **`Endpoint`** (an ordered workflow). +* Endpoints live on a **`Server`** node with finite resources. +* Nodes and **Edges** form a **`TopologyGraph`**. +* A disciplined set of **Enums** (no magic strings) ensure a closed vocabulary. + +--- + +## 1. Workload: `RqsGeneratorInput` + +### Purpose + +Defines the traffic generator that produces request arrivals. + +```python +class RqsGeneratorInput(BaseModel): + id: str + type: SystemNodes = SystemNodes.GENERATOR + avg_active_users: RVConfig + avg_request_per_minute_per_user: RVConfig + user_sampling_window: int = Field( ... ) # seconds +``` + +### Random variables (`RVConfig`) + +```python +class RVConfig(BaseModel): + mean: float + distribution: Distribution = Distribution.POISSON + variance: float | None = None +``` + +#### Validation & guarantees -This contract brings together three distinct but interconnected layers of configuration into one cohesive structure: +* **`mean` is numeric** + `@field_validator("mean", mode="before")` coerces to `float` and rejects non-numeric values. +* **Auto variance** for Normal/LogNormal + `@model_validator(mode="after")` sets `variance = mean` if missing and the distribution is `NORMAL` or `LOG_NORMAL`. +* **Distribution constraints** on workload: -1. **`rqs_input` (`RqsGeneratorInput`)**: Defines the **workload profile**—how many users are active and how frequently they generate requests—and acts as the **source node** in our system graph. -2. **`topology_graph` (`TopologyGraph`)**: Describes the **system's architecture**—its components, resources, and the network connections between them, represented as a directed graph. -3. **`sim_settings` (`SimulationSettings`)**: Configures **global simulation parameters**, such as total runtime and which metrics to collect. + * `avg_request_per_minute_per_user` **must be Poisson** (engine currently optimised for Poisson arrivals). + * `avg_active_users` **must be Poisson or Normal**. + * Enforced via `@field_validator(..., mode="after")` with clear error messages. -This layered design decouples the *what* (the system topology) from the *how* (the traffic pattern and simulation control), allowing for modular and reusable configurations. Adherence to our validation-first philosophy means every payload is rigorously parsed against this schema. By using a controlled vocabulary of `Enums` and the power of Pydantic, we guarantee that any malformed or logically inconsistent input is rejected upfront with clear, actionable errors, ensuring the simulation engine operates only on perfectly valid data. +#### Why these constraints? ------ +* They reflect the current joint-sampling logic in the generator: **Poisson–Poisson** and **Normal–Poisson** are implemented and tested. Additional combos can be enabled later without changing the public contract. -## **1. The System Graph (`topology_graph` and `rqs_input`)** +--- -At the core of FastSim is the representation of the system as a **directed graph**. The **nodes** represent the architectural components (like servers, clients, and the traffic generator itself), while the **edges** represent the directed network connections between them. This graph-based approach allows for flexible and realistic modeling of request flows through distributed systems. +## 2. System Graph: `TopologyGraph` -### **Design Philosophy: A "Micro-to-Macro" Approach** +### Purpose -The schema is built on a compositional, "micro-to-macro" principle. We start by defining the smallest indivisible units of work (`Step`) and progressively assemble them into larger, more complex structures (`Endpoint`, `Server`, and finally the `TopologyGraph`). +Defines the architecture under test as a **directed graph**: nodes are components (client, server, optional load balancer), edges are network links with latency models. -This layered approach provides several key advantages: +```python +class TopologyGraph(BaseModel): + nodes: TopologyNodes + edges: list[Edge] +``` - * **Modularity and Reusability:** Core operations are defined once as `Steps` and can be reused across multiple `Endpoints`. - * **Local Reasoning, Global Safety:** Each model is responsible for its own internal consistency (e.g., a `Step` ensures its metric is valid for its kind). Parent models then enforce the integrity of the connections *between* these components (e.g., the `TopologyGraph` ensures all `Edges` connect to valid `Nodes`). - * **Guaranteed Robustness:** By catching all structural and referential errors before the simulation begins, this approach embodies the "fail-fast" principle. It guarantees that the SimPy engine operates on a valid, self-consistent model. +### Nodes -### **A Controlled Vocabulary: Topology Constants** +```python +class TopologyNodes(BaseModel): + servers: list[Server] + client: Client + load_balancer: LoadBalancer | None = None +``` -The schema's robustness is founded on a controlled vocabulary defined by Python `Enum` classes. Instead of error-prone "magic strings," the schema uses these enums to define the finite set of legal values for categories like operation kinds, metrics, and node types. +#### `Client` -| Enum | Purpose | -| :------------------------- | :------------------------------------------------------------------------ | -| **`EndpointStepCPU`, `EndpointStepRAM`, `EndpointStepIO`** | Defines the exhaustive list of valid `kind` values for a `Step`. | -| **`StepOperation`** | Specifies the legal dictionary keys within a `Step`'s `step_operation`. | -| **`SystemNodes`** | Enumerate the allowed `type` for nodes (e.g., `"server"`, `"client"`, `"generator"`). | -| **`SystemEdges`** | Enumerate the allowed categories for connections between nodes. | +```python +class Client(BaseModel): + id: str + type: SystemNodes = SystemNodes.CLIENT +``` ------ +* **Validator**: `type` must equal `SystemNodes.CLIENT`. -### **Schema Hierarchy and In-Depth Validation** +#### `ServerResources` -Here we break down each component of the topology, highlighting the specific Pydantic validators that enforce its correctness. +```python +class ServerResources(BaseModel): + cpu_cores: PositiveInt = Field(...) + db_connection_pool: PositiveInt | None = Field(...) + ram_mb: PositiveInt = Field(...) +``` -#### **Random Variable Schema (`RVConfig`)** +* Maps directly to SimPy containers (CPU tokens, RAM capacity, etc.). +* Bounds enforced via `Field(ge=..., ...)`. -At the core of both the traffic generator and network latencies is `RVConfig`, a schema for defining stochastic variables. This allows critical parameters to be modeled not as fixed numbers, but as draws from a probability distribution. +#### `Step` (the atomic unit) -| Check | Pydantic Hook | Rule & Rationale | -| :---------------------------- | :---------------------------------------- | :---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **Numeric `mean` Enforcement** | `@field_validator("mean", mode="before")` | Intercepts the `mean` field and ensures the provided value is an `int` or `float`, rejecting invalid types. This guarantees a valid numeric type for all downstream logic. | -| **Valid `distribution` Name** | `Distribution` (`StrEnum`) type hint | Pydantic automatically ensures that the `distribution` field's value must be one of the predefined members (e.g., `"poisson"`, `"normal"`). Any typo or unsupported value results in an immediate validation error. | -| **Intelligent `variance` Defaulting** | `@model_validator(mode="after")` | Enforces a crucial business rule: if `distribution` is `"normal"` or `"log_normal"` **and** `variance` is not provided, the schema automatically sets `variance = mean`. This provides a safe, logical default. | +```python +class Step(BaseModel): + kind: EndpointStepIO | EndpointStepCPU | EndpointStepRAM + step_operation: dict[StepOperation, PositiveFloat | PositiveInt] +``` -#### **1. `Step`: The Atomic Unit of Work** +**Key validator (coherence):** -A `Step` represents a single, indivisible operation. +```python +@model_validator(mode="after") +def ensure_coherence_type_operation(cls, model: "Step") -> "Step": + # exactly one operation key, and it must match the step kind +``` -| Validation Check | Pydantic Hook | Rule & Rationale | -| :------------------------------- | :--------------------- | :---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **Coherence of `kind` and `step_operation`** | `@model_validator` | **Rule:** The `step_operation` dictionary must contain *exactly one* entry, and its key (`StepOperation`) must be the correct metric for the `Step`'s `kind`. \\ **Rationale:** This is the most critical validation on a `Step`. It prevents illogical pairings like a RAM allocation step being measured in `cpu_time`. It ensures every step has a clear, unambiguous impact on a single system resource. | -| **Positive Metric Values** | `PositiveFloat` / `PositiveInt` | **Rule:** All numeric values in `step_operation` must be greater than zero. \\ **Rationale:** It is physically impossible to spend negative or zero time on an operation. This ensures that only plausible resource requests enter the system. | +* If `kind` is CPU → the only key must be `CPU_TIME`. +* If `kind` is RAM → only `NECESSARY_RAM`. +* If `kind` is I/O → only `IO_WAITING_TIME`. +* Values must be positive (`PositiveFloat/PositiveInt`). -#### **2. `Endpoint`: Composing Workflows** +This guarantees every step is **unambiguous** and **physically meaningful**. -An `Endpoint` defines a complete operation (e.g., an API call like `/predict`) as an ordered sequence of `Steps`. +#### `Endpoint` -| Validation Check | Pydantic Hook | Rule & Rationale | -| :-------------------- | :--------------------------------- | :---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **Consistent Naming** | `@field_validator("endpoint_name")` | **Rule:** Automatically converts the `endpoint_name` to lowercase. \\ **Rationale:** This enforces a canonical representation, eliminating ambiguity from inconsistent capitalization (e.g., treating `/predict` and `/Predict` as the same). | +```python +class Endpoint(BaseModel): + endpoint_name: str + steps: list[Step] -#### **3. System Nodes: `Server`, `Client`, and `RqsGeneratorInput`** + @field_validator("endpoint_name", mode="before") + def name_to_lower(cls, v: str) -> str: + return v.lower() +``` -These models define the macro-components of your architecture where work is performed, resources are located, and requests originate. +* Canonical lowercase naming avoids duplicates differing only by case. -| Validation Check | Pydantic Hook | Rule & Rationale | -| :-------------------------------- | :---------------------------------------- | :-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **Standardized Node `type`** | `@field_validator("type")` | **Rule:** The `type` field must strictly match the expected `SystemNodes` enum member (e.g., a `Server` object must have `type: "server"`). \\ **Rationale:** This enforces a strict contract: a `Server` object is always and only a server, preventing object state confusion. | -| **Unique Node IDs** | `@model_validator` in `TopologyNodes` | **Rule:** All `id` fields across all `Server` nodes, the `Client` node, and the `RqsGeneratorInput` node must be unique. \\ **Rationale:** This is fundamental to creating a valid graph. Node IDs are the primary keys. If two nodes shared the same ID, any `Edge` pointing to that ID would be ambiguous. | -| **Workload Distribution Constraints** | `@field_validator` in `RqsGeneratorInput` | **Rule:** The `avg_request_per_minute_per_user` field must use a Poisson distribution. The `avg_active_users` field must use a Poisson or Normal distribution. \\ **Rationale:** This is a current restriction of the simulation engine, which has a joint sampler optimized only for these combinations. This validator ensures that only supported configurations are accepted. | +#### `Server` -#### **4. `Edge`: Connecting the Components** +```python +class Server(BaseModel): + id: str + type: SystemNodes = SystemNodes.SERVER + server_resources: ServerResources + endpoints: list[Endpoint] +``` -An `Edge` represents a directed network link between two nodes. +* **Validator**: `type` must equal `SystemNodes.SERVER`. -| Validation Check | Pydantic Hook | Rule & Rationale | -| :---------------- | :----------------- | :---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **No Self-Loops** | `@model_validator` | **Rule:** An edge's `source` ID cannot be the same as its `target` ID. \\ **Rationale:** A network call from a service to itself is a logical anti-pattern in a system topology. Such an operation should be modeled as an internal process (i.e., another `Step`), not a network hop. | -| **Unique Edge IDs** | `@model_validator` in `TopologyGraph` | **Rule:** All `id` fields of the `Edge`s must be unique. \\ **Rationale:** Ensures that every network connection is uniquely identifiable, which is useful for logging and debugging. | +#### `LoadBalancer` (optional) -#### **5. `TopologyGraph`: The Complete System** +```python +class LoadBalancer(BaseModel): + id: str + type: SystemNodes = SystemNodes.LOAD_BALANCER + algorithms: LbAlgorithmsName = LbAlgorithmsName.ROUND_ROBIN + server_covered: set[str] = Field(default_factory=set) +``` + +### Edges + +```python +class Edge(BaseModel): + id: str + source: str + target: str + latency: RVConfig + probability: float = Field(1.0, ge=0.0, le=1.0) + edge_type: SystemEdges = SystemEdges.NETWORK_CONNECTION + dropout_rate: float = Field(...) +``` -This is the root model that aggregates all `nodes` and `edges` and performs the final, most critical validation: ensuring referential integrity. +#### Validation & guarantees -| Validation Check | Pydantic Hook | Rule & Rationale | -| :---------------------- | :----------------- | :-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| **Referential Integrity** | `@model_validator` | **Rule:** Every `edge.source` and `edge.target` ID must correspond to an actual node ID defined in the topology. \\ **Rationale:** This is the capstone validation that guarantees the structural integrity of the entire system graph. It prevents "dangling edges"—connections that point to non-existent nodes—ensuring the described system is a complete and validly connected graph. | +* **Latency sanity** + `@field_validator("latency", mode="after")` ensures `mean > 0` and `variance >= 0` (if provided). Error messages mention the **edge id** for clarity. +* **No self-loops** + `@model_validator(mode="after")` rejects `source == target`. +* **Unique edge IDs** + `TopologyGraph.unique_ids` enforces uniqueness across `edges`. +* **Referential integrity** + `TopologyGraph.edge_refs_valid` ensures: ------ - -## **2. Global Simulation Control (`SimulationSettings`)** - -This final component configures the simulation's execution parameters and, critically, determines what data is collected. - -#### **Payload Structure (`SimulationSettings`)** - -| Field | Type | Purpose & Validation | -| :----------------------- | :------------------------- | :------------------------------------------------------------------------------------------------------ | -| `total_simulation_time` | `int` | The total simulation horizon in seconds. Must be `>=` a defined minimum (e.g., 1800s). | -| `enabled_sample_metrics` | `set[SampledMetricName]` | A set of metrics to be sampled at fixed intervals, creating a time-series (e.g., `"ready_queue_len"`, `"ram_in_use"`). | -| `enabled_event_metrics` | `set[EventMetricName]` | A set of metrics recorded only when specific events occur (e.g., `"rqs_latency"`). | - -### **Design Rationale: Pre-validated, On-Demand Metrics** - -The design of the `settings`, particularly the `enabled_*_metrics` fields, is centered on **user-driven selectivity** and **ironclad validation**. - -1. **Selectivity:** Data collection has a performance cost. By allowing the user to explicitly select only the metrics they need, we make the simulator more efficient and versatile. - -2. **Ironclad Validation:** Simply allowing users to provide a list of strings is risky. Our schema mitigates this risk by validating every metric name provided by the user against the canonical `Enum` definitions (`SampledMetricName`, `EventMetricName`). If a user provides a misspelled or invalid metric name (e.g., `"request_latncy"`), Pydantic immediately rejects the entire payload *before* the simulation engine runs. - -This guarantees that the simulation engine can safely initialize all necessary data collection structures at the start of the run, completely eliminating an entire class of potential `KeyError` exceptions at runtime. - ------ - -## **End-to-End Example (`SimulationPayload`)** - -The following JSON object shows how these components combine into a single `SimulationPayload` for a minimal client-server setup, updated according to the new schema. - -```jsonc -{ - // Defines the workload profile as a generator node. - "rqs_input": { - "id": "mobile_user_generator", - "type": "generator", - "avg_active_users": { - "mean": 50, - "distribution": "poisson" - }, - "avg_request_per_minute_per_user": { - "mean": 5.0, - "distribution": "poisson" - }, - "user_sampling_window": 60 - }, - // Describes the system's architecture as a graph. - "topology_graph": { - "nodes": { - "client": { - "id": "entry_point_client", - "type": "client" - }, - "servers": [ - { - "id": "api_server", - "type": "server", - "server_resources": { - "cpu_cores": 4, - "ram_mb": 4096, - "db_connection_pool": 10 - }, - "endpoints": [ - { - "endpoint_name": "/predict", - "steps": [ - { - "kind": "initial_parsing", - "step_operation": { "cpu_time": 0.005 } - }, - { - "kind": "io_db_query", - "step_operation": { "io_waiting_time": 0.050 } - } - ] - } - ] - } - ] - }, - "edges": [ - { - "id": "client_to_generator", - "source": "entry_point_client", - "target": "mobile_user_generator", - "latency": { - "distribution": "log_normal", - "mean": 0.001, - "variance": 0.0001 - } - }, - { - "id": "generator_to_api", - "source": "mobile_user_generator", - "target": "api_server", - "latency": { - "distribution": "log_normal", - "mean": 0.04, - "variance": 0.01 - }, - "probability": 1.0, - "dropout_rate": 0.0 - } - ] - }, - // Configures the simulation run and metric collection. - "sim_settings": { - "total_simulation_time": 3600, - "enabled_sample_metrics": [ - "ready_queue_len", - "ram_in_use", - "core_busy" - ], - "enabled_event_metrics": [ - "rqs_latency" - ] - } -} + * Every `target` is a declared node ID. + * **External sources** (e.g., the generator id) are allowed, but **may not** appear as a `target` anywhere. +* **Load balancer integrity** (if present) + `TopologyGraph.valid_load_balancer` enforces: + + * `server_covered ⊆ {server ids}`. + * For every covered server there exists an **outgoing edge from the LB** to that server. + +These checks make the graph **closed**, **consistent**, and **wirable** without surprises at runtime. + +--- + +## 3. Simulation Control: `SimulationSettings` + +```python +class SimulationSettings(BaseModel): + total_simulation_time: int = Field(..., ge=TimeDefaults.MIN_SIMULATION_TIME) + enabled_sample_metrics: set[SampledMetricName] = Field(default_factory=...) + enabled_event_metrics: set[EventMetricName] = Field(default_factory=...) + sample_period_s: float = Field(..., ge=SamplePeriods.MINIMUM_TIME, le=SamplePeriods.MAXIMUM_TIME) ``` -### **Key Takeaways** +### What it controls + +* **Clock** — `total_simulation_time` (seconds). +* **Sampling cadence** — `sample_period_s` for time-series metrics. +* **What to collect** — two sets of enums: + + * `enabled_sample_metrics`: time-series KPIs (e.g., ready queue length, RAM in use, edge concurrency). + * `enabled_event_metrics`: per-event KPIs (e.g., request clocks/latency). + +### Why Enums matter here + +Letting users pass strings like `"ram_in_use"` is error-prone. By using **`SampledMetricName`** and **`EventMetricName`** enums, the settings are **validated upfront**, so the runtime collector knows exactly which lists to allocate and fill. No hidden `KeyError`s halfway through a run. + +--- + +## What these validations buy you + +* **Type safety** (no accidental strings where enums are expected). +* **Physical realism** (no zero/negative times or RAM). +* **Graph integrity** (no dangling edges or self-loops). +* **Operational clarity** (every step has exactly one effect). +* **Better errors** (validators point to the exact field/entity at fault). + +Together, they make the model **predictable** for the simulation engine and **pleasant** to debug. + +--- + +## End-to-End Example (YAML) + +This is a complete, valid payload you can load with `SimulationRunner.from_yaml(...)`. + +```yaml +# ─────────────────────────────────────────────────────────────── +# FastSim scenario: generator → client → server → client +# ─────────────────────────────────────────────────────────────── + +rqs_input: + id: rqs-1 + # avg_active_users can be POISSON or NORMAL; mean is required. + avg_active_users: + mean: 100 + distribution: poisson + # must be POISSON (engine constraint) + avg_request_per_minute_per_user: + mean: 20 + distribution: poisson + user_sampling_window: 60 # seconds + +topology_graph: + nodes: + client: + id: client-1 + type: client + servers: + - id: srv-1 + type: server + server_resources: + cpu_cores: 1 + ram_mb: 2048 + # db_connection_pool: 50 # optional + endpoints: + - endpoint_name: /predict + steps: + - kind: ram + step_operation: { necessary_ram: 100 } + - kind: initial_parsing # CPU step (enum in your code) + step_operation: { cpu_time: 0.001 } + - kind: io_wait # I/O step + step_operation: { io_waiting_time: 0.100 } + + edges: + - id: gen-to-client + source: rqs-1 # external source OK + target: client-1 + latency: { mean: 0.003, distribution: exponential } + + - id: client-to-server + source: client-1 + target: srv-1 + latency: { mean: 0.003, distribution: exponential } + + - id: server-to-client + source: srv-1 + target: client-1 + latency: { mean: 0.003, distribution: exponential } + +sim_settings: + total_simulation_time: 500 + sample_period_s: 0.05 + enabled_sample_metrics: + - ready_queue_len + - event_loop_io_sleep + - ram_in_use + - edge_concurrent_connection + enabled_event_metrics: + - rqs_clock +``` + + Notes: +> + * `kind` uses the **EndpointStep** enums you’ve defined (e.g., `ram`, `initial_parsing`, `io_wait`). + * The coherence validator ensures that each `kind` uses the correct `step_operation` key and **exactly one** entry. + * The **edge** constraints guarantee a clean, connected, and sensible graph. + +--- + +## Summary + +* The **payload** is small but expressive: workload, topology, and settings. +* The **validators** are doing real work: they make illegal states unrepresentable. +* The **enums** keep the contract tight and maintainable. +* Together, they let you move fast **without** breaking the simulation engine. - * **Single Source of Truth**: `Enum` classes centralize all valid string literals, providing robust, type-safe validation across the entire schema. - * **Layered Validation**: The `Constants → Component Schemas → SimulationPayload` hierarchy ensures that only well-formed and self-consistent configurations reach the simulation engine. - * **Separation of Concerns**: The three top-level keys (`rqs_input`, `topology_graph`, `sim_settings`) clearly separate the workload, the system architecture, and simulation control, making configurations easier to read, write, and reuse. \ No newline at end of file +If you extend the engine (new distributions, step kinds, metrics), you can **keep the same contract** and enrich the enums & validators to preserve the same guarantees. diff --git a/docs/fastsim-docs/simulation_runner.md b/docs/fastsim-docs/simulation_runner.md new file mode 100644 index 0000000..8153e13 --- /dev/null +++ b/docs/fastsim-docs/simulation_runner.md @@ -0,0 +1,236 @@ + +# **Simulation Runner — Technical Documentation** + +## **Overview** + +The `SimulationRunner` is the **orchestrator** of the FastSim engine. +Its main responsibility is to: + +1. **Build** simulation actors from a structured input (`SimulationPayload`). +2. **Wire** actors together via `EdgeRuntime` connections. +3. **Start** all simulation processes in a SimPy environment. +4. **Run** the simulation clock for the configured duration. +5. **Collect and return** performance metrics through a `ResultsAnalyzer` instance. + +This design separates **topology definition** (data models) from **execution logic** (runtime actors), ensuring clarity, testability, and future extensibility. + +--- + +## **High-Level Flow** + +```mermaid +flowchart TD + subgraph Input + A["SimulationPayload (Pydantic)"] + end + + subgraph Build_Phase + B1[Build RqsGeneratorRuntime] + B2[Build ClientRuntime] + B3[Build ServerRuntimes] + B4["Build LoadBalancerRuntime (optional)"] + end + + subgraph Wire_Phase + C["Build EdgeRuntimes + assign target boxes"] + end + + subgraph Start_Phase + D1[Start all Startable actors] + D2[Start SampledMetricCollector] + end + + subgraph Run_Phase + E["SimPy env.run(until=total_simulation_time)"] + end + + subgraph Output + F[ResultsAnalyzer] + end + + A --> B1 & B2 & B3 & B4 + B1 & B2 & B3 & B4 --> C + C --> D1 & D2 + D1 & D2 --> E + E --> F +``` + +--- + +## **Component Responsibilities** + +### **SimulationRunner** + +* **Inputs:** + + * `env`: a `simpy.Environment` controlling discrete-event simulation time. + * `simulation_input`: a `SimulationPayload` describing topology, request generation parameters, and simulation settings. + +* **Responsibilities:** + + * Build **all runtime actors** (`RqsGeneratorRuntime`, `ClientRuntime`, `ServerRuntime`, `LoadBalancerRuntime`). + * Instantiate **EdgeRuntime** objects to connect actors. + * Start processes and the metric collector. + * Advance the simulation clock. + * Package results into a `ResultsAnalyzer`. + +--- + +### **Actors** + +All runtime actors implement the `Startable` protocol (i.e., they expose a `.start()` method returning a `simpy.Process`). + +| Actor | Responsibility | +| ----------------------- | -------------------------------------------------------------------------------------------- | +| **RqsGeneratorRuntime** | Produces incoming requests according to stochastic models (Poisson, Gaussian-Poisson, etc.). | +| **ClientRuntime** | Consumes responses, tracks completion, and stores latency samples. | +| **ServerRuntime** | Processes incoming requests, interacts with CPU/RAM containers, measures processing times. | +| **LoadBalancerRuntime** | Distributes incoming requests across available servers according to configured policy. | +| **EdgeRuntime** | Models the connection between two nodes (latency, bandwidth, loss). | + +--- + +### **ResourcesRuntime** + +* Registry mapping server IDs to their SimPy `Container` resources (CPU, RAM). +* Keeps resource allocation/consumption logic decoupled from server logic. + +--- + +### **Metrics** + +* **SampledMetricCollector**: Periodically snapshots runtime metrics (queue sizes, RAM usage, connection counts). +* **ResultsAnalyzer**: Consumes raw metrics and computes aggregated KPIs (latency distribution, throughput series, etc.). + +--- + +## **Execution Timeline** + +```mermaid +sequenceDiagram + participant Runner as SimulationRunner + participant RqsGen as RqsGeneratorRuntime + participant Client as ClientRuntime + participant Server as ServerRuntime + participant LB as LoadBalancerRuntime + participant Edge as EdgeRuntime + participant Metrics as SampledMetricCollector + + Runner->>RqsGen: Build from input + Runner->>Client: Build from input + Runner->>Server: Build each from input + Runner->>LB: Build if present + Runner->>Edge: Build + assign target_box + Runner->>RqsGen: Set out_edge + Runner->>Client: Set out_edge + Runner->>Server: Set out_edge + Runner->>LB: Append out_edges + Runner->>Metrics: Start collector + Runner->>RqsGen: start() + Runner->>Client: start() + Runner->>Server: start() + Runner->>LB: start() + Runner->>Runner: env.run(until=total_simulation_time) + Runner->>Metrics: Gather results + Runner->>Runner: Return ResultsAnalyzer +``` + +--- + +## **Detailed Build & Wire Steps** + +### 1️⃣ **Build Phase** + +* **`_build_rqs_generator()`**: Creates a single `RqsGeneratorRuntime` for now; architecture allows for multiple (future CDN scenarios). +* **`_build_client()`**: Instantiates the single client node; stored in a dict for future multi-client extensions. +* **`_build_servers()`**: Creates one `ServerRuntime` per configured server. Pulls CPU/RAM resources from `ResourcesRuntime`. +* **`_build_load_balancer()`**: Optional; created only if present in the topology. + +--- + +### 2️⃣ **Wire Phase** + +* Merges all runtime actor dictionaries into a `all_nodes` map. +* For each `Edge` in the topology: + + * Looks up **target** object and assigns the correct inbox (`simpy.Store`). + * Creates an `EdgeRuntime` and assigns it as `out_edge` (or appends to `out_edges` for LBs). + +--- + +### 3️⃣ **Start Phase** + +* Uses `itertools.chain` to lazily iterate over all runtime actors in the correct deterministic order. +* Casts to `Iterable[Startable]` to make Mypy type-checking explicit. +* Starts `SampledMetricCollector` to record periodic metrics. + +--- + +### 4️⃣ **Run Phase** + +* Advances SimPy’s event loop until `total_simulation_time` from the simulation settings. +* Returns a `ResultsAnalyzer` for downstream reporting and plotting. + +--- + +## **Extensibility Hooks** + +* **Multiple Generators / Clients**: Dictionaries keyed by node ID already prepared. +* **CDN or Multi-tier Architectures**: Easily extendable via new actor types + wiring rules. +* **Different LB Policies**: Swap `LoadBalancerRuntime` strategy without touching the runner. +* **Metric Expansion**: `SampledMetricCollector` can be extended to capture additional KPIs. + +--- + +## **Architecture Diagram** + +``` + ┌───────────────────────┐ + │ SimulationPayload │ + │ (input topology + cfg) │ + └─────────┬─────────────┘ + │ + ▼ + ┌───────────────────────┐ + │ SimulationRunner │ + └─────────┬─────────────┘ + │ build actors + ▼ + ┌─────────────────────────────────────────────────┐ + │ Runtime Actors (Startable) │ + │ ┌──────────────────┐ ┌──────────────────────┐ │ + │ │ RqsGenerator │→│ ClientRuntime │ │ + │ └──────────────────┘ └──────────────────────┘ │ + │ ↓ edges ↑ edges │ + │ ┌──────────────────┐ ┌──────────────────────┐ │ + │ │ ServerRuntime(s) │←→│ LoadBalancerRuntime │ │ + │ └──────────────────┘ └──────────────────────┘ │ + └─────────────────────────────────────────────────┘ + │ + ▼ + ┌────────────────────────────┐ + │ SampledMetricCollector │ + └──────────────┬─────────────┘ + ▼ + ┌────────────────┐ + │ ResultsAnalyzer │ + └────────────────┘ +``` + +--- + +## Architectural rationale + +✅ **Separation of concerns** — Topology definition, resource allocation, runtime behaviour, and metric processing are decoupled. + +✅ **Extensible** — Adding new node types or connection logic requires minimal changes. + +✅ **Testable** — Each phase can be tested in isolation (unit + integration). + +✅ **Deterministic order** — Startup sequence guarantees reproducibility. + +✅ **Scalable** — Supports larger topologies by design. + +--- + + diff --git a/example/data/single_server.yml b/example/data/single_server.yml new file mode 100644 index 0000000..c6e2ce6 --- /dev/null +++ b/example/data/single_server.yml @@ -0,0 +1,56 @@ +# ─────────────────────────────────────────────────────────────── +# FastSim scenario: generator ➜ client ➜ server ➜ client +# ─────────────────────────────────────────────────────────────── + +# 1. Traffic generator (light load) +rqs_input: + id: rqs-1 + avg_active_users: { mean: 100 } + avg_request_per_minute_per_user: { mean: 20 } + user_sampling_window: 60 + +# 2. Topology +topology_graph: + nodes: + client: { id: client-1 } + servers: + - id: srv-1 + server_resources: { cpu_cores: 1, ram_mb: 2048 } + endpoints: + - endpoint_name: ep-1 + probability: 1.0 + steps: + - kind: initial_parsing + step_operation: { cpu_time: 0.001 } + - kind: ram + step_operation: { necessary_ram: 100} + - kind: io_wait + step_operation: { io_waiting_time: 0.1 } + + edges: + - id: gen-to-client + source: rqs-1 + target: client-1 + latency: { mean: 0.003, distribution: exponential } + + - id: client-to-server + source: client-1 + target: srv-1 + latency: { mean: 0.003, distribution: exponential } + + - id: server-to-client + source: srv-1 + target: client-1 + latency: { mean: 0.003, distribution: exponential } + +# 3. Simulation settings +sim_settings: + total_simulation_time: 500 + sample_period_s: 0.05 + enabled_sample_metrics: + - ready_queue_len + - event_loop_io_sleep + - ram_in_use + - edge_concurrent_connection + enabled_event_metrics: + - rqs_clock diff --git a/example/single_server_yaml.py b/example/single_server_yaml.py new file mode 100644 index 0000000..3057a52 --- /dev/null +++ b/example/single_server_yaml.py @@ -0,0 +1,273 @@ +#!/usr/bin/env python3 +""" +Run a FastSim scenario from a YAML file and print diagnostics. + +What it does: +- Loads the simulation payload from YAML via `SimulationRunner.from_yaml`. +- Runs the simulation. +- Prints latency stats, 1s-bucket throughput, and a preview of sampled metrics. +- Saves four plots (latency histogram, throughput, server queues, RAM). +- Performs sanity checks (expected vs observed) with simple queueing heuristics. + +Usage: + python src/app/example/run_from_yaml.py \ + --yaml src/app/example/data/single_server.yml +""" + +from __future__ import annotations + +from argparse import ArgumentParser +from pathlib import Path +from typing import Dict, Iterable, List, Mapping, Tuple + +import matplotlib.pyplot as plt +import numpy as np +import simpy +from app.config.constants import ( + EndpointStepCPU, + EndpointStepIO, + EndpointStepRAM, + LatencyKey, + StepOperation, +) +from app.metrics.analyzer import ResultsAnalyzer +from app.runtime.simulation_runner import SimulationRunner + + +# ───────────────────────────────────────────────────────────── +# Pretty printers +# ───────────────────────────────────────────────────────────── +def print_latency_stats(res: ResultsAnalyzer) -> None: + """Print latency statistics returned by the analyzer.""" + stats: Mapping[LatencyKey, float] = res.get_latency_stats() + print("\n════════ LATENCY STATS ════════") + if not stats: + print("(empty)") + return + + # Keep deterministic ordering for readability. + order: List[LatencyKey] = [ + LatencyKey.TOTAL_REQUESTS, + LatencyKey.MEAN, + LatencyKey.MEDIAN, + LatencyKey.STD_DEV, + LatencyKey.P95, + LatencyKey.P99, + LatencyKey.MIN, + LatencyKey.MAX, + ] + for key in order: + if key in stats: + print(f"{key.name:<20} = {stats[key]:.6f}") + + +def print_throughput(res: ResultsAnalyzer) -> None: + """Print 1-second throughput buckets.""" + timestamps, rps = res.get_throughput_series() + print("\n════════ THROUGHPUT (req/sec) ════════") + if not timestamps: + print("(empty)") + return + + for t, rate in zip(timestamps, rps): + print(f"t={t:4.1f}s → {rate:6.2f} rps") + + +def print_sampled_preview(res: ResultsAnalyzer) -> None: + """Print first 5 samples of each sampled metric series.""" + sampled: Dict[str, Dict[str, List[float]]] = res.get_sampled_metrics() + print("\n════════ SAMPLED METRICS ════════") + if not sampled: + print("(empty)") + return + + for metric, series in sampled.items(): + print(f"\n📈 {metric}:") + for entity, vals in series.items(): + head = list(vals[:5]) if vals else [] + print(f" - {entity}: len={len(vals)}, first={head}") + + +# ───────────────────────────────────────────────────────────── +# Plotting +# ───────────────────────────────────────────────────────────── +def save_all_plots(res: ResultsAnalyzer, out_path: Path) -> None: + """Generate the 2x2 plot figure and save it to `out_path`.""" + fig, axes = plt.subplots(2, 2, figsize=(12, 8)) + res.plot_latency_distribution(axes[0, 0]) + res.plot_throughput(axes[0, 1]) + res.plot_server_queues(axes[1, 0]) + res.plot_ram_usage(axes[1, 1]) + fig.tight_layout() + fig.savefig(out_path) + print(f"\n🖼️ Plots saved to: {out_path}") + + +# ───────────────────────────────────────────────────────────── +# Sanity checks (expected vs observed) +# ───────────────────────────────────────────────────────────── +REL_TOL = 0.30 # 30% tolerance for rough sanity checks + + +def _tick(label: str, expected: float, observed: float) -> None: + """Print a ✓ or ⚠ depending on relative error vs `REL_TOL`.""" + if expected == 0.0: + delta_pct = 0.0 + icon = "•" + else: + delta = abs(observed - expected) / abs(expected) + delta_pct = delta * 100.0 + icon = "✓" if delta <= REL_TOL else "⚠" + print(f"{icon} {label:<28} expected≈{expected:.3f} observed={observed:.3f} Δ={delta_pct:.1f}%") + + +def _endpoint_totals(runner: SimulationRunner) -> Tuple[float, float, float]: + """ + Return (CPU_seconds, IO_seconds, RAM_MB) of the first endpoint on the first server. + + This keeps the check simple. If you use multiple endpoints weighted by probability, + extend this function to compute a probability-weighted average. + """ + servers = runner.simulation_input.topology_graph.nodes.servers + if not servers or not servers[0].endpoints: + return (0.0, 0.0, 0.0) + + ep = servers[0].endpoints[0] + cpu_s = 0.0 + io_s = 0.0 + ram_mb = 0.0 + + for step in ep.steps: + if isinstance(step.kind, EndpointStepCPU): + cpu_s += float(step.step_operation[StepOperation.CPU_TIME]) + elif isinstance(step.kind, EndpointStepIO): + io_s += float(step.step_operation[StepOperation.IO_WAITING_TIME]) + elif isinstance(step.kind, EndpointStepRAM): + ram_mb += float(step.step_operation[StepOperation.NECESSARY_RAM]) + + return (cpu_s, io_s, ram_mb) + + +def _edges_mean_latency(runner: SimulationRunner) -> float: + """Sum of edge mean latencies across the graph (simple additive approximation).""" + return float(sum(e.latency.mean for e in runner.simulation_input.topology_graph.edges)) + + +def _mean(series: Iterable[float]) -> float: + """Numerically stable mean for a generic float iterable.""" + arr = np.asarray(list(series), dtype=float) + return float(np.mean(arr)) if arr.size else 0.0 + + +def run_sanity_checks(runner: SimulationRunner, res: ResultsAnalyzer) -> None: + """ + Compare expected vs observed metrics using back-of-the-envelope approximations. + + Approximations used: + - Throughput ≈ λ = users * RPM / 60 + - Mean latency ≈ CPU_s + IO_s + NET_s (ignores queueing inside the server) + - Mean RAM ≈ λ * T_srv * RAM_per_request (Little’s law approximation) + - Mean I/O queue length ≈ λ * IO_s + - Edge concurrency ≈ λ * edge_mean_latency + """ + print("\n════════ SANITY CHECKS (expected vs observed) ════════") + + # Arrival rate λ (requests per second) + w = runner.simulation_input.rqs_input + lam_rps = float(w.avg_active_users.mean) * float(w.avg_request_per_minute_per_user.mean) / 60.0 + + # Endpoint sums + cpu_s, io_s, ram_mb = _endpoint_totals(runner) + net_s = _edges_mean_latency(runner) + t_srv = cpu_s + io_s + latency_expected = cpu_s + io_s + net_s + + # Observed latency, throughput + stats = res.get_latency_stats() + latency_observed = float(stats.get(LatencyKey.MEAN, 0.0)) + _, rps_series = res.get_throughput_series() + rps_observed = _mean(rps_series) + + # Observed RAM and queues + sampled = res.get_sampled_metrics() + ram_series = sampled.get("ram_in_use", {}) + ram_means = [_mean(vals) for vals in ram_series.values()] + ram_observed = float(sum(ram_means)) if ram_means else 0.0 + + ready_series = sampled.get("ready_queue_len", {}) + ioq_series = sampled.get("event_loop_io_sleep", {}) + ready_observed = _mean([_mean(v) for v in ready_series.values()]) if ready_series else 0.0 + ioq_observed = _mean([_mean(v) for v in ioq_series.values()]) if ioq_series else 0.0 + + # Expected quantities (very rough) + rps_expected = lam_rps + ram_expected = lam_rps * t_srv * ram_mb + ioq_expected = lam_rps * io_s + + _tick("Mean throughput (rps)", rps_expected, rps_observed) + _tick("Mean latency (s)", latency_expected, latency_observed) + _tick("Mean RAM (MB)", ram_expected, ram_observed) + _tick("Mean I/O queue", ioq_expected, ioq_observed) + + # Edge concurrency + edge_conc = sampled.get("edge_concurrent_connection", {}) + if edge_conc: + print("\n— Edge concurrency —") + edge_means: Dict[str, float] = {eid: _mean(vals) for eid, vals in edge_conc.items()} + for e in runner.simulation_input.topology_graph.edges: + exp = lam_rps * float(e.latency.mean) + obs = edge_means.get(e.id, 0.0) + _tick(f"edge {e.id}", exp, obs) + + # Extra diagnostics + print("\n— Diagnostics —") + print( + "λ={:.3f} rps | CPU_s={:.3f} IO_s={:.3f} NET_s≈{:.3f} | RAM/req={:.1f} MB" + .format(lam_rps, cpu_s, io_s, net_s, ram_mb) + ) + print("T_srv={:.3f}s → RAM_expected≈λ*T_srv*RAM = {:.1f} MB".format(t_srv, ram_expected)) + + +# ───────────────────────────────────────────────────────────── +# Main +# ───────────────────────────────────────────────────────────── +def main() -> None: + """Entry-point: parse args, run simulation, print/plot, sanity-check.""" + parser = ArgumentParser(description="Run FastSim from YAML and print outputs + sanity checks.") + parser.add_argument( + "--yaml", + type=Path, + default=Path(__file__).parent / "data" /"single_server.yml", + help="Path to the simulation YAML file.", + ) + parser.add_argument( + "--out", + type=Path, + default=Path(__file__).parent / "single_server_yml.png", + help="Path to the output image (plots).", + ) + args = parser.parse_args() + + yaml_path: Path = args.yaml + if not yaml_path.exists(): + raise FileNotFoundError(f"YAML not found: {yaml_path}") + + # Build runner from YAML and execute + env = simpy.Environment() + runner = SimulationRunner.from_yaml(env=env, yaml_path=yaml_path) + results: ResultsAnalyzer = runner.run() + + # Prints + print_latency_stats(results) + print_throughput(results) + print_sampled_preview(results) + + # Sanity checks + run_sanity_checks(runner, results) + + # Plots + save_all_plots(results, args.out) + + +if __name__ == "__main__": + main() diff --git a/example/single_server_yml.png b/example/single_server_yml.png new file mode 100644 index 0000000..89cca6d Binary files /dev/null and b/example/single_server_yml.png differ diff --git a/src/app/config/constants.py b/src/app/config/constants.py index 07e383c..0aa5f9d 100644 --- a/src/app/config/constants.py +++ b/src/app/config/constants.py @@ -1,15 +1,15 @@ """ Application-wide constants and configuration values. -This module groups all the *static* enumerations used by the FastSim backend +This module groups all the static enumerations used by the FastSim backend so that: -* JSON / YAML payloads can be strictly validated with Pydantic. -* Front-end and simulation engine share a single source of truth. -* Ruff, mypy and IDEs can leverage the strong typing provided by Enum classes. + JSON / YAML payloads can be strictly validated with Pydantic. + Front-end and simulation engine share a single source of truth. + Ruff, mypy and IDEs can leverage the strong typing provided by Enum classes. -**IMPORTANT:** Changing any enum *value* is a breaking-change for every -stored configuration file. Add new members whenever possible instead of +IMPORTANT: Changing any enum value is a breaking-change for every +stored configuration file. Add new members whenever possible instead of renaming existing ones. """ @@ -22,27 +22,27 @@ class TimeDefaults(IntEnum): """ - Default time-related constants (expressed in **seconds**). + Default time-related constants (expressed in seconds). - These values are used when the user omits an explicit parameter. They also + These values are used when the user omits an explicit parameter. They also serve as lower / upper bounds for validation for the requests generator. """ - MIN_TO_SEC = 60 # 1 minute → 60 s - USER_SAMPLING_WINDOW = 60 # keep U(t) constant for 60 s - SIMULATION_TIME = 3_600 # run 1 h if user gives no value - MIN_SIMULATION_TIME = 1_800 # enforce at least 30 min - MIN_USER_SAMPLING_WINDOW = 1 # 1 s minimum - MAX_USER_SAMPLING_WINDOW = 120 # 2 min maximum + MIN_TO_SEC = 60 # 1 minute → 60 s + USER_SAMPLING_WINDOW = 60 # every 60 seconds sample the number of active user + SIMULATION_TIME = 3_600 # run 1 h if user gives no value + MIN_SIMULATION_TIME = 5 # 5 seconds give a broad spectrum + MIN_USER_SAMPLING_WINDOW = 1 # 1 s minimum + MAX_USER_SAMPLING_WINDOW = 120 # 2 min maximum class Distribution(StrEnum): """ - Probability distributions accepted by :class:`~app.schemas.RVConfig`. + Probability distributions accepted by app.schemas.RVConfig. - The *string value* is exactly the identifier that must appear in JSON + The string value is exactly the identifier that must appear in JSON payloads. The simulation engine will map each name to the corresponding - random sampler (e.g. ``numpy.random.poisson``). + random sampler (e.g.numpy.random.poisson). """ POISSON = "poisson" @@ -55,43 +55,19 @@ class Distribution(StrEnum): # CONSTANTS FOR ENDPOINT STEP DEFINITION (REQUEST-HANDLER) # ====================================================================== -# The JSON received by the API for an endpoint step is expected to look like: -# -# { -# "endpoint_name": "/predict", -# "kind": "io_llm", -# "metrics": { -# "cpu_time": 0.150, -# "necessary_ram": 256 -# } -# } -# -# The Enum classes below guarantee that only valid *kind* and *metric* keys -# are accepted by the Pydantic schema. - - class EndpointStepIO(StrEnum): """ - I/O-bound operation categories that can occur inside an endpoint *step*. - - .. list-table:: - :header-rows: 1 - - * - Constant - - Meaning (executed by coroutine) - * - ``TASK_SPAWN`` - - Spawns an additional ``asyncio.Task`` and returns immediately. - * - ``LLM`` - - Performs a remote Large-Language-Model inference call. - * - ``WAIT`` - - Passive, *non-blocking* wait for I/O completion; no new task spawned. - * - ``DB`` - - Round-trip to a relational / NoSQL database. - * - ``CACHE`` - - Access to a local or distributed cache layer. - - The *value* of each member (``"io_llm"``, ``"io_db"``, …) is the exact - identifier expected in external JSON. + I/O-bound operation categories that can occur inside an endpoint step. + - TASK_SPAWN + Spawns an additional ``asyncio.Task`` and returns immediately. + - LLM + Performs a remote Large-Language-Model inference call. + - WAIT + Passive, *non-blocking* wait for I/O completion; no new task spawned. + - DB + Round-trip to a relational / NoSQL database. + - CACHE + Access to a local or distributed cache layer. """ TASK_SPAWN = "io_task_spawn" @@ -126,11 +102,11 @@ class EndpointStepRAM(StrEnum): class StepOperation(StrEnum): """ - Keys used inside the ``metrics`` dictionary of a *step*. + Keys used inside the metrics dictionary of a step. - * ``CPU_TIME`` - Service time (seconds) during which the coroutine occupies + CPU_TIME - Service time (seconds) during which the coroutine occupies the CPU / GIL. - * ``NECESSARY_RAM`` - Peak memory (MB) required by the step. + NECESSARY_RAM - Peak memory (MB) required by the step. """ CPU_TIME = "cpu_time" @@ -191,7 +167,7 @@ class SystemNodes(StrEnum): class SystemEdges(StrEnum): """ - Edge categories connecting different :class:`SystemNodes`. + Edge categories connecting different class SystemNodes. Currently only network links are modeled; new types (IPC queue, message bus, stream) can be added without impacting existing payloads. @@ -205,7 +181,7 @@ class SystemEdges(StrEnum): class SampledMetricName(StrEnum): """ - define the metrics sampled every fixed amount of + Define the metrics sampled every fixed amount of time to create a time series """ @@ -218,7 +194,7 @@ class SampledMetricName(StrEnum): class SamplePeriods(float, Enum): """ - defining the value of the sample periods for the metrics for which + Defining the value of the sample periods for the metrics for which we have to extract a time series """ @@ -232,12 +208,12 @@ class SamplePeriods(float, Enum): class EventMetricName(StrEnum): """ - define the metrics triggered by event with no + Define the metrics triggered by event with no time series """ # Mandatory - RQS_CLOCK = "rqs_clock" + RQS_CLOCK = "rqs_clock" # useful to collect starting and finishing time of rqs # Not mandatory LLM_COST = "llm_cost" diff --git a/src/app/metrics/analyzer.py b/src/app/metrics/analyzer.py index 762e5c6..e1aa50e 100644 --- a/src/app/metrics/analyzer.py +++ b/src/app/metrics/analyzer.py @@ -51,11 +51,10 @@ def __init__( ) -> None: """ Args: - client: Client runtime object, containing RqsClock entries. - servers: List of server runtime objects. - edges: List of edge runtime objects. - settings: Original simulation settings. - + client: Client runtime object, containing RqsClock entries. + servers: List of server runtime objects. + edges: List of edge runtime objects. + settings: Original simulation settings. """ self._client = client self._servers = servers diff --git a/src/app/metrics/collector.py b/src/app/metrics/collector.py index 17339fa..d2158f5 100644 --- a/src/app/metrics/collector.py +++ b/src/app/metrics/collector.py @@ -14,7 +14,6 @@ # way we optimize the initialization of various objects reducing # the global overhead - class SampledMetricCollector: """class to define a centralized object to collect sampled metrics""" @@ -26,7 +25,14 @@ def __init__( env: simpy.Environment, sim_settings: SimulationSettings, ) -> None: - """Docstring to complete""" + """ + Args: + edges (list[EdgeRuntime]): list of the class EdgeRuntime + servers (list[ServerRuntime]): list of server of the class ServerRuntime + env (simpy.Environment): environment for the simulation + sim_settings (SimulationSettings): general settings for the simulation + + """ self.edges = edges self.servers = servers self.sim_settings = sim_settings @@ -41,8 +47,6 @@ def __init__( self._ready_key = SampledMetricName.READY_QUEUE_LEN - env.process(self._build_time_series()) - def _build_time_series(self) -> Generator[simpy.Event, None, None]: """Function to build time series for enabled metrics""" while True: @@ -61,6 +65,8 @@ def _build_time_series(self) -> Generator[simpy.Event, None, None]: server.enabled_metrics[self._io_key].append(server.io_queue_len) server.enabled_metrics[self._ready_key].append(server.ready_queue_len) + + def start(self) -> simpy.Process: """Definition of the process to collect sampled metrics""" return self.env.process(self._build_time_series()) diff --git a/src/app/runtime/actors/client.py b/src/app/runtime/actors/client.py index 42b0c3f..2f2da4b 100644 --- a/src/app/runtime/actors/client.py +++ b/src/app/runtime/actors/client.py @@ -44,6 +44,7 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]: """Updtate the state before passing it to another node""" assert self.out_edge is not None while True: + state: RequestState = yield self.client_box.get() # type: ignore[assignment] state.record_hop( @@ -58,7 +59,7 @@ def _forwarder(self) -> Generator[simpy.Event, None, None]: # would be equal to two would mean that the state # went through the mandatory path to be generated # rqs generator and client registration - if len(state.history) > 2: + if len(state.history) > 3: state.finish_time = self.env.now clock_data = RqsClock( start=state.initial_time, diff --git a/src/app/runtime/actors/server.py b/src/app/runtime/actors/server.py index 9051d99..d791bbb 100644 --- a/src/app/runtime/actors/server.py +++ b/src/app/runtime/actors/server.py @@ -141,6 +141,7 @@ def _handle_request( # noqa: PLR0915, PLR0912, C901 yield self.server_resources[ServerResourceName.RAM.value].get(total_ram) self._ram_in_use += total_ram + # Initial conditions of the server a rqs a priori is not in any queue # and it does not occupy a core until it started to be elaborated core_locked = False @@ -175,7 +176,7 @@ def _handle_request( # noqa: PLR0915, PLR0912, C901 for step in selected_endpoint.steps: - if isinstance(step.kind, EndpointStepCPU): + if step.kind in EndpointStepCPU: # with the boolean we avoid redundant operation of asking # the core multiple time on a given step # for example if we have two consecutive cpu bound step @@ -200,8 +201,9 @@ def _handle_request( # noqa: PLR0915, PLR0912, C901 # Execute the step giving back the control to the simpy env yield self.env.timeout(cpu_time) - - elif isinstance(step.kind, EndpointStepIO): + # since the object is of an Enum class we check if the step.kind + # is one member of enum + elif step.kind in EndpointStepIO: io_time = step.step_operation[StepOperation.IO_WAITING_TIME] # Same here with the boolean if we have multiple I/O steps # we release the core just the first time if the previous step diff --git a/src/app/samplers/common_helpers.py b/src/app/samplers/common_helpers.py index e565a20..d594538 100644 --- a/src/app/samplers/common_helpers.py +++ b/src/app/samplers/common_helpers.py @@ -47,30 +47,43 @@ def exponential_variable_generator( return float(rng.exponential(mean)) def general_sampler(random_variable: RVConfig, rng: np.random.Generator) -> float: - """Sample a number according to the distribution described in `random_variable`.""" - dist = random_variable.distribution - mean = random_variable.mean + """ + Draw one sample from the distribution described by *random_variable*. + + Only **Normal** and **Log-normal** require an explicit ``variance``. + For **Uniform**, **Poisson** and **Exponential** the mean is enough. + """ + dist = random_variable.distribution + mean = random_variable.mean + var = random_variable.variance match dist: + # ── No extra parameters needed ────────────────────────────────── case Distribution.UNIFORM: - - assert random_variable.variance is None + # Variance is meaningless for an ad-hoc uniform [0, 1) helper. + assert var is None return uniform_variable_generator(rng) - case _: + case Distribution.POISSON: + # λ == mean ; numpy returns ints → cast to float for consistency + assert var is None + return float(poisson_variable_generator(mean, rng)) - variance = random_variable.variance - assert variance is not None - - match dist: - case Distribution.NORMAL: - return truncated_gaussian_generator(mean, variance, rng) - case Distribution.LOG_NORMAL: - return lognormal_variable_generator(mean, variance, rng) - case Distribution.POISSON: - return float(poisson_variable_generator(mean, rng)) - case Distribution.EXPONENTIAL: - return exponential_variable_generator(mean, rng) - case _: - msg = f"Unsupported distribution: {dist}" - raise ValueError(msg) + case Distribution.EXPONENTIAL: + # β (scale) == mean ; nothing else required + assert var is None + return exponential_variable_generator(mean, rng) + + # ── Distributions that *do* need a variance parameter ─────────── + case Distribution.NORMAL: + assert var is not None + return truncated_gaussian_generator(mean, var, rng) + + case Distribution.LOG_NORMAL: + assert var is not None + return lognormal_variable_generator(mean, var, rng) + + # ── Anything else is unsupported ──────────────────────────────── + case _: + msg = f"Unsupported distribution: {dist}" + raise ValueError(msg) diff --git a/src/app/samplers/poisson_poisson.py b/src/app/samplers/poisson_poisson.py index 7f17364..725faa5 100644 --- a/src/app/samplers/poisson_poisson.py +++ b/src/app/samplers/poisson_poisson.py @@ -28,7 +28,7 @@ def poisson_poisson_sampling( Algorithm --------- - 1. Every *sampling_window_s* seconds, draw + 1. Every sampling_window_s seconds, draw U ~ Poisson(mean_concurrent_user). 2. Compute the aggregate rate Λ = U * (mean_req_per_minute_per_user / 60) [req/s]. diff --git a/src/app/schemas/random_variables_config.py b/src/app/schemas/random_variables_config.py index b09ed13..e15d509 100644 --- a/src/app/schemas/random_variables_config.py +++ b/src/app/schemas/random_variables_config.py @@ -13,7 +13,7 @@ class RVConfig(BaseModel): variance: float | None = None @field_validator("mean", mode="before") - def ensure_mean_is_numeric( + def ensure_mean_is_numeric_and_positive( cls, # noqa: N805 v: float, ) -> float: @@ -21,6 +21,7 @@ def ensure_mean_is_numeric( err_msg = "mean must be a number (int or float)" if not isinstance(v, (float, int)): raise ValueError(err_msg) # noqa: TRY004 + return float(v) @model_validator(mode="after") # type: ignore[arg-type] diff --git a/src/app/schemas/simulation_output.py b/src/app/schemas/simulation_output.py deleted file mode 100644 index 26d1adb..0000000 --- a/src/app/schemas/simulation_output.py +++ /dev/null @@ -1,13 +0,0 @@ -"""Define the output of the simulation""" - -from pydantic import BaseModel - - -class SimulationOutput(BaseModel): - """Define the output of the simulation""" - - total_requests: dict[str, int | float] - # TO DEFINE - metric_2: str - #...... - metric_n: str diff --git a/src/app/schemas/system_topology/endpoint.py b/src/app/schemas/system_topology/endpoint.py index 8933dab..aa40e9e 100644 --- a/src/app/schemas/system_topology/endpoint.py +++ b/src/app/schemas/system_topology/endpoint.py @@ -53,8 +53,10 @@ def ensure_coherence_type_operation( raise ValueError(msg) # Coherence CPU bound operation and operation - if isinstance(model.kind, EndpointStepCPU): - if operation_keys != {StepOperation.CPU_TIME}: + if ( + isinstance(model.kind, EndpointStepCPU) + and operation_keys != {StepOperation.CPU_TIME} + ): msg = ( "The operation to quantify a CPU BOUND step" f"must be {StepOperation.CPU_TIME}" @@ -62,8 +64,10 @@ def ensure_coherence_type_operation( raise ValueError(msg) # Coherence RAM operation and operation - elif isinstance(model.kind, EndpointStepRAM): - if operation_keys != {StepOperation.NECESSARY_RAM}: + if ( + isinstance(model.kind, EndpointStepRAM) + and operation_keys != {StepOperation.NECESSARY_RAM} + ): msg = ( "The operation to quantify a RAM step" f"must be {StepOperation.NECESSARY_RAM}" @@ -71,11 +75,12 @@ def ensure_coherence_type_operation( raise ValueError(msg) # Coherence I/O operation and operation - elif operation_keys != {StepOperation.IO_WAITING_TIME}: - msg = ( - "The operation to quantify an I/O step" - f"must be {StepOperation.IO_WAITING_TIME}" - ) + if ( + isinstance(model.kind, EndpointStepIO) + and operation_keys != {StepOperation.IO_WAITING_TIME} + ): + + msg = f"An I/O step must use {StepOperation.IO_WAITING_TIME}" raise ValueError(msg) return model diff --git a/src/app/schemas/system_topology/full_system_topology.py b/src/app/schemas/system_topology/full_system_topology.py index 6b49b21..ddd4c5f 100644 --- a/src/app/schemas/system_topology/full_system_topology.py +++ b/src/app/schemas/system_topology/full_system_topology.py @@ -16,6 +16,7 @@ field_validator, model_validator, ) +from pydantic_core.core_schema import ValidationInfo from app.config.constants import ( LbAlgorithmsName, @@ -204,6 +205,37 @@ class Edge(BaseModel): ), ) + # The idea to put here the control about variance and mean about the edges + # latencies and not in RVConfig is to provide a better error handling + # providing a direct indication of the edge with the error + # The idea to put here the control about variance and mean about the edges + # latencies and not in RVConfig is to provide a better error handling + # providing a direct indication of the edge with the error + @field_validator("latency", mode="after") + def ensure_latency_is_non_negative( + cls, # noqa: N805 + v: RVConfig, + info: ValidationInfo, + ) -> RVConfig: + """Ensures that the latency's mean and variance are positive.""" + mean = v.mean + variance = v.variance + + # We can get the edge ID from the validation context for a better error message + edge_id = info.data.get("id", "unknown") + + if mean <= 0: + msg = f"The mean latency of the edge '{edge_id}' must be positive" + raise ValueError(msg) + if variance is not None and variance < 0: # Variance can be zero + msg = ( + f"The variance of the latency of the edge {edge_id}" + "must be positive" + ) + raise ValueError(msg) + return v + + @model_validator(mode="after") # type: ignore[arg-type] def check_src_trgt_different(cls, model: "Edge") -> "Edge": # noqa: N805 """Ensure source is different from target""" @@ -241,33 +273,55 @@ def unique_ids( return model - @model_validator(mode="after") # type: ignore[arg-type] - def edge_refs_valid(cls, model: "TopologyGraph") -> "TopologyGraph": # noqa: N805 + @model_validator(mode="after") # type: ignore[arg-type] + def edge_refs_valid( + cls, # noqa: N805 + model: "TopologyGraph", + ) -> "TopologyGraph": """ - Ensure that **every** edge points to valid nodes. - - The validator is executed *after* the entire ``TopologyGraph`` model has - been built, so all servers and the client already exist in ``m.nodes``. + Validate that the graph is self-consistent. - Steps - ----- - 1. Build the set ``valid_ids`` containing: - * all ``Server.id`` values, **plus** - * the single ``Client.id``. - 2. Iterate through each ``Edge`` in ``m.edges`` and raise - :class:`ValueError` if either ``edge.source`` or ``edge.target`` is - **not** present in ``valid_ids``. - - Returning the (unchanged) model signals that the integrity check passed. + * All targets must be nodes declared in ``m.nodes``. + * External IDs are allowed as sources (entry points, generator) but + they must never appear as a target anywhere else. """ - valid_ids = {s.id for s in model.nodes.servers} | {model.nodes.client.id} + # ------------------------------------------------------------------ + # 1. Collect declared node IDs (servers, client, optional LB) + # ------------------------------------------------------------------ + node_ids: set[str] = {srv.id for srv in model.nodes.servers} + node_ids.add(model.nodes.client.id) if model.nodes.load_balancer is not None: - valid_ids.add(model.nodes.load_balancer.id) - - for e in model.edges: - if e.source not in valid_ids or e.target not in valid_ids: - msg = f"Edge {e.source}->{e.target} references unknown node" + node_ids.add(model.nodes.load_balancer.id) + + # ------------------------------------------------------------------ + # 2. Scan every edge once + # ------------------------------------------------------------------ + external_sources: set[str] = set() + + for edge in model.edges: + # ── Rule 1: target must be a declared node + if edge.target not in node_ids: + msg = ( + f"Edge {edge.source}->{edge.target} references " + f"unknown target node '{edge.target}'." + ) raise ValueError(msg) + + # Collect any source that is not a declared node + if edge.source not in node_ids: + external_sources.add(edge.source) + + # ------------------------------------------------------------------ + # 3. Ensure external sources never appear as targets elsewhere + # ------------------------------------------------------------------ + forbidden_targets = external_sources & {e.target for e in model.edges} + if forbidden_targets: + msg = ( + "External IDs cannot be used as targets as well:" + f"{sorted(forbidden_targets)}" + ) + raise ValueError(msg) + return model @model_validator(mode="after") # type: ignore[arg-type] diff --git a/tests/conftest.py b/tests/conftest.py index 09a80d4..39c8b40 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -6,6 +6,7 @@ from numpy.random import default_rng from app.config.constants import ( + Distribution, EventMetricName, SampledMetricName, SamplePeriods, @@ -17,6 +18,7 @@ from app.schemas.simulation_settings_input import SimulationSettings from app.schemas.system_topology.full_system_topology import ( Client, + Edge, TopologyGraph, TopologyNodes, ) @@ -109,15 +111,23 @@ def rqs_input() -> RqsGeneratorInput: @pytest.fixture def topology_minimal() -> TopologyGraph: """ - A valid topology containing a single client and **no** servers or edges. + A valid *tiny* topology: one generator ➜ one client. - Suitable for low-level tests that do not need to traverse the server - layer or network graph. + The single edge has a negligible latency; its only purpose is to give the + generator a valid ``out_edge`` so that the runtime can start. """ client = Client(id="client-1") - nodes = TopologyNodes(servers=[], client=client) - return TopologyGraph(nodes=nodes, edges=[]) + # Stub edge: generator id comes from rqs_input fixture (“rqs-1”) + edge = Edge( + id="gen-to-client", + source="rqs-1", + target="client-1", + latency=RVConfig(mean=0.001, distribution=Distribution.POISSON), + ) + + nodes = TopologyNodes(servers=[], client=client) + return TopologyGraph(nodes=nodes, edges=[edge]) # --------------------------------------------------------------------------- # # Complete simulation payload # diff --git a/tests/integration/minimal/conftest.py b/tests/integration/minimal/conftest.py index 1c9f81e..434f9f0 100644 --- a/tests/integration/minimal/conftest.py +++ b/tests/integration/minimal/conftest.py @@ -21,12 +21,7 @@ if TYPE_CHECKING: from app.schemas.full_simulation_input import SimulationPayload - from app.schemas.full_simulation_input import ( - SimulationPayload as _Payload, # noqa: F401 - ) - from app.schemas.rqs_generator_input import ( - RqsGeneratorInput as _RqsIn, # noqa: F401 - ) + # ────────────────────────────────────────────────────────────────────────────── # 0-traffic generator (shadows the project-wide fixture) diff --git a/tests/integration/payload/data/invalid/missing_field.yml b/tests/integration/payload/data/invalid/missing_field.yml new file mode 100644 index 0000000..c74102d --- /dev/null +++ b/tests/integration/payload/data/invalid/missing_field.yml @@ -0,0 +1,17 @@ +rqs_input: + id: gen-1 + avg_active_users: { mean: 1 } + avg_request_per_minute_per_user: { mean: 10 } + +topology_graph: + nodes: + client: { id: cli } + servers: + - id: srv-1 + endpoints: + - endpoint_name: ep + steps: + - { kind: cpu_parse, step_operation: { cpu_time: 0.001 } } + + edges: [] +sim_settings: { total_simulation_time: 10 } diff --git a/tests/integration/payload/data/invalid/negative_latency.yml b/tests/integration/payload/data/invalid/negative_latency.yml new file mode 100644 index 0000000..f69fb60 --- /dev/null +++ b/tests/integration/payload/data/invalid/negative_latency.yml @@ -0,0 +1,15 @@ +rqs_input: + id: gen-1 + avg_active_users: { mean: 1 } + avg_request_per_minute_per_user: { mean: 10 } + +topology_graph: + nodes: + client: { id: cli } + servers: [] + edges: + - id: bad-lat + source: gen-1 + target: cli + latency: { mean: -0.001 } +sim_settings: { total_simulation_time: 5 } diff --git a/tests/integration/payload/data/invalid/wrong_enum.yml b/tests/integration/payload/data/invalid/wrong_enum.yml new file mode 100644 index 0000000..58a1c50 --- /dev/null +++ b/tests/integration/payload/data/invalid/wrong_enum.yml @@ -0,0 +1,13 @@ +rqs_input: + id: gen-1 + avg_active_users: { mean: 1 } + avg_request_per_minute_per_user: + mean: 10 + distribution: gamma # not valid enum + +topology_graph: + nodes: + client: { id: cli } + servers: [] + edges: [] +sim_settings: { total_simulation_time: 5 } diff --git a/tests/integration/payload/test_payload_invalid.py b/tests/integration/payload/test_payload_invalid.py new file mode 100644 index 0000000..09249c1 --- /dev/null +++ b/tests/integration/payload/test_payload_invalid.py @@ -0,0 +1,19 @@ +"""test to verify validation on invalid yml""" + +from pathlib import Path + +import pytest +import yaml +from pydantic import ValidationError + +from app.schemas.full_simulation_input import SimulationPayload + +DATA_DIR = Path(__file__).parent / "data" / "invalid" +YMLS = sorted(DATA_DIR.glob("*.yml")) + +@pytest.mark.integration +@pytest.mark.parametrize("yaml_path", YMLS, ids=lambda p: p.stem) +def test_invalid_payloads_raise(yaml_path: Path) -> None : + raw = yaml.safe_load(yaml_path.read_text()) + with pytest.raises(ValidationError): + SimulationPayload.model_validate(raw) diff --git a/tests/integration/single_server/conftest.py b/tests/integration/single_server/conftest.py new file mode 100644 index 0000000..38f575a --- /dev/null +++ b/tests/integration/single_server/conftest.py @@ -0,0 +1,47 @@ +""" +Fixtures for the *single-server* integration scenario: + +generator ──edge──> server ──edge──> client + +The topology is stored as a YAML file (`tests/data/single_server.yml`) so +tests remain declarative and we avoid duplicating Pydantic wiring logic. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +import pytest +import simpy + +if TYPE_CHECKING: # heavy imports only when type-checking + from app.runtime.simulation_runner import SimulationRunner + + +# --------------------------------------------------------------------------- # +# Shared SimPy environment (function-scope so every test starts fresh) # +# --------------------------------------------------------------------------- # +@pytest.fixture +def env() -> simpy.Environment: + """Return an empty ``simpy.Environment`` for each test.""" + return simpy.Environment() + + +# --------------------------------------------------------------------------- # +# Build a SimulationRunner from the YAML scenario # +# --------------------------------------------------------------------------- # +@pytest.fixture +def runner(env: simpy.Environment) -> SimulationRunner: + """ + Load *single_server.yml* through the public constructor + :pymeth:`SimulationRunner.from_yaml`. + """ + # import deferred to avoid ruff TC001 + from app.runtime.simulation_runner import SimulationRunner # noqa: PLC0415 + + yaml_path: Path = ( + Path(__file__).parent / "data" / "single_server.yml" + ) + + return SimulationRunner.from_yaml(env=env, yaml_path=yaml_path) diff --git a/tests/integration/single_server/data/single_server.yml b/tests/integration/single_server/data/single_server.yml new file mode 100644 index 0000000..4072f48 --- /dev/null +++ b/tests/integration/single_server/data/single_server.yml @@ -0,0 +1,54 @@ +# ─────────────────────────────────────────────────────────────── +# FastSim scenario: generator ➜ client ➜ server ➜ client +# ─────────────────────────────────────────────────────────────── + +# 1. Traffic generator (light load) +rqs_input: + id: rqs-1 + avg_active_users: { mean: 5 } + avg_request_per_minute_per_user: { mean: 40 } + user_sampling_window: 60 + +# 2. Topology +topology_graph: + nodes: + client: { id: client-1 } + servers: + - id: srv-1 + server_resources: { cpu_cores: 2, ram_mb: 2048 } + endpoints: + - endpoint_name: ep-1 + probability: 1.0 + steps: + - kind: initial_parsing + step_operation: { cpu_time: 0.001 } + - kind: io_wait + step_operation: { io_waiting_time: 0.002 } + + edges: + - id: gen-to-client + source: rqs-1 + target: client-1 + latency: { mean: 0.003, distribution: exponential } + + - id: client-to-server + source: client-1 + target: srv-1 + latency: { mean: 0.003, distribution: exponential } + + - id: server-to-client + source: srv-1 + target: client-1 + latency: { mean: 0.003, distribution: exponential } + +# 3. Simulation settings +sim_settings: + total_simulation_time: 50 + sample_period_s: 0.01 + enabled_sample_metrics: + - ready_queue_len + - event_loop_io_sleep + - ram_in_use + - edge_concurrent_connection + enabled_event_metrics: + - rqs_clock diff --git a/tests/integration/single_server/test_single_server.py b/tests/integration/single_server/test_single_server.py new file mode 100644 index 0000000..498a7a1 --- /dev/null +++ b/tests/integration/single_server/test_single_server.py @@ -0,0 +1,52 @@ +""" +End-to-end verification of a *functional* topology (1 generator, 1 server). + +Assertions cover: + +* non-zero latency stats, +* throughput series length > 0, +* presence of sampled metrics for both edge & server. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest + +from app.config.constants import LatencyKey, SampledMetricName + +if TYPE_CHECKING: # only needed for type-checking + from app.metrics.analyzer import ResultsAnalyzer + from app.runtime.simulation_runner import SimulationRunner + + +# --------------------------------------------------------------------------- # +# Tests # +# --------------------------------------------------------------------------- # +@pytest.mark.integration +def test_single_server_happy_path(runner: SimulationRunner) -> None: + """Run the simulation and ensure that *something* was processed.""" + results: ResultsAnalyzer = runner.run() + + # ── Latency stats must exist ─────────────────────────────────────────── + stats = results.get_latency_stats() + assert stats, "Expected non-empty latency statistics." + assert stats[LatencyKey.TOTAL_REQUESTS] > 0 + assert stats[LatencyKey.MEAN] > 0.0 + + # ── Throughput series must have at least one bucket > 0 ─────────────── + ts, rps = results.get_throughput_series() + assert len(ts) == len(rps) > 0 + assert any(val > 0 for val in rps) + + # ── Sampled metrics must include *one* server and *one* edge ─────────── + sampled = results.get_sampled_metrics() + + # Server RAM & queues + assert SampledMetricName.RAM_IN_USE in sampled + assert sampled[SampledMetricName.RAM_IN_USE], "Server RAM time-series missing." + + # Edge concurrent-connection metric + assert SampledMetricName.EDGE_CONCURRENT_CONNECTION in sampled + assert sampled[SampledMetricName.EDGE_CONCURRENT_CONNECTION], "Edge metric missing." diff --git a/tests/unit/runtime/actors/test_client.py b/tests/unit/runtime/actors/test_client.py index bc17471..7ee194e 100644 --- a/tests/unit/runtime/actors/test_client.py +++ b/tests/unit/runtime/actors/test_client.py @@ -84,6 +84,7 @@ def test_inbound_is_completed() -> None: req = RequestState(id=2, initial_time=0.0) req.record_hop(SystemNodes.GENERATOR, "gen-1", env.now) req.record_hop(SystemEdges.NETWORK_CONNECTION, "edge-X", env.now) + req.record_hop(SystemNodes.CLIENT, "cli-1", env.now) # simulate return inbox.put(req) env.run() diff --git a/tests/unit/runtime/actors/test_edge.py b/tests/unit/runtime/actors/test_edge.py index 37ad296..644b078 100644 --- a/tests/unit/runtime/actors/test_edge.py +++ b/tests/unit/runtime/actors/test_edge.py @@ -82,7 +82,7 @@ def _make_edge( id="edge-1", source="src", target="dst", - latency=RVConfig(mean=0.0, variance=1.0, distribution="normal"), + latency=RVConfig(mean=1.0, variance=1.0, distribution="normal"), dropout_rate=dropout_rate, ) diff --git a/tests/unit/runtime/test_simulation_runner.py b/tests/unit/runtime/test_simulation_runner.py new file mode 100644 index 0000000..d785e3d --- /dev/null +++ b/tests/unit/runtime/test_simulation_runner.py @@ -0,0 +1,159 @@ +"""Unit-tests for :pyclass:`app.runtime.simulation_runner.SimulationRunner`. + +Purpose +------- +Validate each private builder in isolation and run a minimal end-to-end +execution without relying on the full integration scenarios. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import patch + +import pytest +import simpy +import yaml + +from app.metrics.analyzer import ResultsAnalyzer +from app.runtime.simulation_runner import SimulationRunner + +if TYPE_CHECKING: + from pathlib import Path + + from app.runtime.actors.client import ClientRuntime + from app.runtime.actors.rqs_generator import RqsGeneratorRuntime + from app.schemas.full_simulation_input import SimulationPayload + + +# --------------------------------------------------------------------------- # +# Fixtures # +# --------------------------------------------------------------------------- # +@pytest.fixture +def env() -> simpy.Environment: + """Return a fresh SimPy environment for every unit test.""" + return simpy.Environment() + + +@pytest.fixture +def runner( + env: simpy.Environment, + payload_base: SimulationPayload, +) -> SimulationRunner: + """Factory producing an **un-started** SimulationRunner.""" + return SimulationRunner(env=env, simulation_input=payload_base) + + +# --------------------------------------------------------------------------- # +# Builder-level tests # +# --------------------------------------------------------------------------- # +def test_build_rqs_generator_populates_dict(runner: SimulationRunner) -> None: + """_build_rqs_generator() must register one generator runtime.""" + runner._build_rqs_generator() # noqa: SLF001 + assert len(runner._rqs_runtime) == 1 # noqa: SLF001 + gen_rt: RqsGeneratorRuntime = next( + iter(runner._rqs_runtime.values()), # noqa: SLF001 + ) + assert gen_rt.rqs_generator_data.id == runner.rqs_generator.id + + +def test_build_client_populates_dict(runner: SimulationRunner) -> None: + """_build_client() must register exactly one client runtime.""" + runner._build_client() # noqa: SLF001 + assert len(runner._client_runtime) == 1 # noqa: SLF001 + cli_rt: ClientRuntime = next( + iter(runner._client_runtime.values()), # noqa: SLF001 + ) + assert cli_rt.client_config.id == runner.client.id + assert cli_rt.out_edge is None + + +def test_build_servers_keeps_empty_with_minimal_topology( + runner: SimulationRunner, +) -> None: + """Zero servers in the payload → dict stays empty.""" + runner._build_servers() # noqa: SLF001 + assert runner._servers_runtime == {} # noqa: SLF001 + + +def test_build_load_balancer_noop_when_absent( + runner: SimulationRunner, +) -> None: + """No LB in the payload → builder leaves dict empty.""" + runner._build_load_balancer() # noqa: SLF001 + assert runner._lb_runtime == {} # noqa: SLF001 + + +# --------------------------------------------------------------------------- # +# Edges builder # +# --------------------------------------------------------------------------- # +def test_build_edges_with_stub_edge(runner: SimulationRunner) -> None: + """ + `_build_edges()` must register exactly one `EdgeRuntime`, corresponding + to the single stub edge (generator → client) present in the minimal + topology fixture. + """ + runner._build_rqs_generator() # noqa: SLF001 + runner._build_client() # noqa: SLF001 + runner._build_edges() # noqa: SLF001 + assert len(runner._edges_runtime) == 1 # noqa: SLF001 + + +# --------------------------------------------------------------------------- # +# End-to-end “mini” run # +# --------------------------------------------------------------------------- # +def test_run_returns_results_analyzer(runner: SimulationRunner) -> None: + """ + `.run()` must complete even though the client is a sink node. We patch + `_build_client` to assign a no-op edge to avoid assertions. + """ + + class _NoOpEdge: + """Edge stub that silently discards transported states.""" + + def transport(self) -> None: + return + + def patched_build_client(self: SimulationRunner) -> None: + # Call the original builder + SimulationRunner._build_client(self) # noqa: SLF001 + cli_rt = next(iter(self._client_runtime.values())) + cli_rt.out_edge = _NoOpEdge() # type: ignore[assignment] + + with patch.object(runner, "_build_client", patched_build_client.__get__(runner)): + results: ResultsAnalyzer = runner.run() + + assert isinstance(results, ResultsAnalyzer) + assert ( + pytest.approx(runner.env.now) + == runner.simulation_settings.total_simulation_time + ) + + +# --------------------------------------------------------------------------- # +# from_yaml utility # +# --------------------------------------------------------------------------- # +def test_from_yaml_minimal(tmp_path: Path, env: simpy.Environment) -> None: + """from_yaml() parses YAML, validates via Pydantic and returns a runner.""" + yml_payload = { + "rqs_input": { + "id": "gen-yaml", + "avg_active_users": {"mean": 1}, + "avg_request_per_minute_per_user": {"mean": 2}, + "user_sampling_window": 10, + }, + "topology_graph": { + "nodes": {"client": {"id": "cli-yaml"}, "servers": []}, + "edges": [], + }, + "sim_settings": {"total_simulation_time": 5}, + } + + yml_path: Path = tmp_path / "scenario.yml" + yml_path.write_text(yaml.safe_dump(yml_payload)) + + runner = SimulationRunner.from_yaml(env=env, yaml_path=yml_path) + + assert isinstance(runner, SimulationRunner) + assert runner.rqs_generator.id == "gen-yaml" + assert runner.client.id == "cli-yaml" diff --git a/tests/unit/samplers/test_sampler_helper.py b/tests/unit/samplers/test_sampler_helper.py index 7e615d0..ff44c3b 100644 --- a/tests/unit/samplers/test_sampler_helper.py +++ b/tests/unit/samplers/test_sampler_helper.py @@ -176,7 +176,7 @@ def test_general_sampler_normal_path() -> None: def test_general_sampler_poisson_path() -> None: """Poisson branch returns the dummy's preset integer as *float*.""" dummy = cast("np.random.Generator", DummyRNG(poisson_value=4)) - cfg = RVConfig(mean=5.0, variance=5.0, distribution=Distribution.POISSON) + cfg = RVConfig(mean=5.0, distribution=Distribution.POISSON) result = general_sampler(cfg, dummy) assert isinstance(result, float) assert result == 4.0 @@ -192,5 +192,5 @@ def test_general_sampler_lognormal_path() -> None: def test_general_sampler_exponential_path() -> None: """Exponential branch produces a strictly positive float.""" rng = np.random.default_rng(7) - cfg = RVConfig(mean=1.5, variance=1.5, distribution=Distribution.EXPONENTIAL) + cfg = RVConfig(mean=1.5, distribution=Distribution.EXPONENTIAL) assert general_sampler(cfg, rng) > 0.0