diff --git a/CHANGELOG.md b/CHANGELOG.md index c4eff13..6c594a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -18,7 +18,7 @@ Analyses that previously returned wrong numbers without any error. If you rely o - Seeded Monte Carlo runs were not reproducible across identical scenario rebuilds: link ids carried a random uuid suffix whose sort order changed per build, remapping seeded draws onto different parallel links. Link ids are now a deterministic per-(source, target) sequence (`A|B|0`), and `add_link` raises instead of silently overwriting when an id collides (possible when node names contain `|`) or when the same link is added twice - A risk group sharing its name with a node or link excluded the wrong entity: the node was excluded and the group's members were left up. Failed entities are now classified by rule scope - `expand_groups` expanded differently depending on whether a failure came from an entity rule or a `risk_group` rule, and never reached members of nested groups. Expansion is now transitive and identical for both rule kinds -- **BREAKING**: `MaxFlowResult.min_cut` (`max_flow_detailed` with `include_min_cut=True`) returns a true minimum cut whose capacity equals the max flow, instead of all saturated edges. Saturated-edge analysis remains available via `sensitivity()` +- **BREAKING**: `MaxFlowResult.min_cut` (`max_flow_detailed` with `include_min_cut=True`) returns a true minimum cut whose capacity equals the max flow, instead of all saturated edges. The capacities match for the default max-flow configuration (`PROPORTIONAL` placement, `require_capacity=True`, `shortest_path=False`); the other configurations are placement models rather than max-flow computations. Saturated-edge analysis remains available via `sensitivity()` - Monte Carlo with `seed=None` now falls back to the failure policy's own seed; a seeded policy previously produced one identical failure pattern on every iteration - `group_mode: per_group` now matches its documented semantics: `combine` creates one demand per source group, `pairwise` pairs nodes within each same-label group, and volume splits evenly across groups. A skipped group's share is not redistributed - `k_shortest_paths` between multi-node groups merges results across all source/sink pairs instead of returning paths for the single best pair, and breaks ties structurally so results no longer depend on `PYTHONHASHSEED`; `shortest_paths` ordering is likewise deterministic @@ -79,13 +79,14 @@ Other fixes. ### Added +- Demands can be pinned to explicit routes with `static_paths`, modelling MPLS-style LSPs: one flow per route, and a route broken by a failure carries nothing instead of rerouting. A route is a list of node names or link ids (link ids pick a specific one of several parallel links). Requires `mode: pairwise` with selectors matching exactly one source and one target - `AnalysisContext.sensitivity_with_flow`: max flow and edge sensitivity per group pair in a single pass, used by `sensitivity_analysis` and the sensitivity Monte Carlo path (results unchanged) - `AnalysisContext.build_node_mask`/`build_edge_mask` as public methods, for analysis functions calling Core primitives directly - `Scenario.run(step_hook=...)`: a callable returning a context manager entered around each step, used by the CLI `--profile` path - `FailurePolicy.apply_failures_typed`, returning scope-typed failure sets (`apply_failures` remains as a merged-list wrapper), plus `prepare_weights`, `build_risk_group_index`, and the `prepared_rg_index` parameter on `apply_failures` for reuse across Monte Carlo iterations - `build_demand_placement_inputs`, exported from `ngraph.analysis`; `demand_placement_analysis` accepts precomputed `expansion=` and `resolved_ids=` - `TrafficDemand.to_dict()`, `Mode.from_string` in `ngraph.types`, and `link_path_key` in `ngraph.model.selectors` (the canonical "source|target" key used when path-matching links) -- Minimum `netgraph-core` raised to 0.7.0, the API family this release targets; the previous 0.3.0 floor predated APIs now in use +- Minimum `netgraph-core` raised to 0.8.0, which introduces the `FlowPolicy.set_static_paths` and `PredDAG.from_edges` APIs that pinned routes are built on, and whose max-flow completion phase makes `max_flow` return a true maximum (values can increase against 0.7.x) ### Removed diff --git a/docs/reference/api-full.md b/docs/reference/api-full.md index e08a475..33ce544 100644 --- a/docs/reference/api-full.md +++ b/docs/reference/api-full.md @@ -12,9 +12,9 @@ Quick links: - [CLI Reference](cli.md) - [DSL Reference](dsl.md) -Generated from source code on: August 21, 2026 at 00:49 UTC +Generated from source code on: August 24, 2026 at 01:58 UTC -Modules auto-discovered: 53 +Modules auto-discovered: 54 --- @@ -488,7 +488,33 @@ Attributes: Traffic demand specification. Defines `TrafficDemand`, a user-facing specification used by demand expansion -and placement. Routing behavior is selected via an optional `FlowPolicyPreset`. +and placement. Routing behavior is selected via an optional `FlowPolicyPreset`, +or pinned to explicit routes with `StaticPath`. + +### StaticPath + +One explicit route a demand can be pinned to (an MPLS-style LSP). + +Give exactly one of `nodes` or `links`: + +- `nodes`: the node names the route visits, source first and target last. + + Each consecutive pair must be adjacent. When several parallel links + connect a pair, the cheapest is used (ties broken by link id); name the + link explicitly to choose a different one. + +- `links`: the link ids the route traverses, in order. Unambiguous when + + parallel links exist. A link may be traversed in either direction. + +Attributes: + nodes: Node names along the route, or empty when `links` is given. + links: Link ids along the route, or empty when `nodes` is given. + +**Attributes:** + +- `nodes` (Tuple) = () +- `links` (Tuple) = () ### TrafficDemand @@ -503,6 +529,10 @@ Attributes: group_mode: How grouped nodes produce demands ("flatten", "per_group", "group_pairwise"). flow_policy: Policy preset for routing. + static_paths: Explicit routes to pin this demand to. When set, the + demand is placed only on these routes: one flow per route, and a + route broken by a failure carries nothing rather than rerouting. + Requires selectors matching exactly one source and one target. attrs: Arbitrary user metadata. id: Unique identifier. Auto-generated if empty. @@ -515,6 +545,7 @@ Attributes: - `mode` (str) = combine - `group_mode` (str) = flatten - `flow_policy` (Union) +- `static_paths` (Tuple) = () - `attrs` (Dict) = {} - `id` (str) @@ -888,7 +919,7 @@ Enumerates common flow policy presets for traffic routing. These presets map to specific combinations of path algorithms, flow placement strategies, and edge selection modes provided by NetGraph-Core. -### create_flow_policy(algorithms: 'netgraph_core.Algorithms', graph: 'netgraph_core.Graph', preset: 'FlowPolicyPreset', node_mask=None, edge_mask=None) -> 'netgraph_core.FlowPolicy' +### create_flow_policy(algorithms: 'netgraph_core.Algorithms', graph: 'netgraph_core.Graph', preset: 'FlowPolicyPreset', node_mask=None, edge_mask=None, static_path_count: 'Optional[int]' = None) -> 'netgraph_core.FlowPolicy' Create a FlowPolicy instance from a preset configuration. @@ -899,6 +930,9 @@ Args: flow-count bounds to apply. node_mask: Optional numpy bool array for node exclusions (True = include). edge_mask: Optional numpy bool array for edge exclusions (True = include). + static_path_count: Number of routes the caller will pin with + `FlowPolicy.set_static_paths`. Sets the flow count to match, since + a pinned policy creates one flow per route and never grows. Returns: netgraph_core.FlowPolicy: Configured policy instance. @@ -2580,7 +2614,9 @@ Captures total flow, cost distribution, and optionally min-cut edges. Attributes: total_flow: Maximum flow value achieved. - cost_distribution: Mapping of path cost to flow volume placed at that cost. + cost_distribution: Mapping of cost to flow volume placed at that cost. + Completion-phase entries are marginal costs, so a key need not match + any single path's cost. min_cut: Edges forming a minimum cut (None if not computed). **Attributes:** @@ -2922,6 +2958,8 @@ Attributes: volume: Traffic volume to place. priority: Priority class (lower is higher priority). policy_preset: FlowPolicy configuration preset. + static_paths: Routes this demand is pinned to, empty when it is + routed by the policy. **Attributes:** @@ -2930,6 +2968,7 @@ Attributes: - `volume` (float) - `priority` (int) - `policy_preset` (FlowPolicyPreset) +- `static_paths` (Tuple[StaticPath, ...]) = () ### expand_demands(network: 'Network', traffic_demands: 'List[TrafficDemand]', default_policy_preset: 'FlowPolicyPreset' = ) -> 'DemandExpansion' @@ -2957,9 +2996,11 @@ Returns: DemandExpansion with demands and augmentations. Raises: - ValueError: If no demands could be expanded, or if two demands share - an id (pseudo node names embed the id, so duplicates would merge - distinct demands' attachment edges into one endpoint). + ValueError: If no demands could be expanded, if two demands share an + id (pseudo node names embed the id, so duplicates would merge + distinct demands' attachment edges into one endpoint), or if a + demand with `static_paths` does not resolve to exactly one + source/target pair. --- @@ -3262,6 +3303,39 @@ Raises: --- +## ngraph.analysis.static_paths + +Resolution of explicit routes into Core path bundles. + +Turns the `StaticPath` entries on a demand into the `PredDAG` bundles that +`FlowPolicy.set_static_paths` pins traffic to. A bundle is a single simple +path: one edge per hop, so a route that names adjacent nodes with parallel +links between them picks one of those links (see `StaticPath`). + +### build_static_path_bundles(ctx: "'AnalysisContext'", paths: 'Sequence[StaticPath]', src_name: 'str', dst_name: 'str') -> 'List[netgraph_core.PredDAG]' + +Build the Core path bundles a demand is pinned to. + +Args: + ctx: Context holding the built graph; routes resolve against it. + paths: Routes to pin, in the order flows should be created. + src_name: Node every route must start at. + dst_name: Node every route must end at. + +Returns: + One `PredDAG` per route, in the given order. Results are cached per + context, so repeated calls during a Monte Carlo run resolve once. + +Raises: + ValueError: If an endpoint is absent from the context graph, if a + route names an unknown node or a disabled/unknown link, has a hop + whose nodes are joined only by disabled links, has a hop with + no link, traverses a link that does not leave the node it has + reached, does not run from `src_name` to `dst_name`, or revisits + a node (a pinned route must be a simple path). + +--- + ## ngraph.lib.nx NetworkX graph conversion utilities. diff --git a/docs/reference/api.md b/docs/reference/api.md index 3149710..ce9943e 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -229,7 +229,7 @@ print(summary.cost_distribution) # Dict[float, float] mapping cost to flow volu - `analyze(network, *, source=None, sink=None, mode=Mode.COMBINE)` - Create analysis context - `ctx.max_flow(source, sink, *, mode, shortest_path, require_capacity, flow_placement, excluded_nodes, excluded_links)` - Maximum flow -- `ctx.max_flow_detailed(..., include_min_cut=False)` - Maximum flow with cost distribution and optional min-cut; the min-cut is a true minimum cut (its capacity equals the max flow), not the set of saturated edges +- `ctx.max_flow_detailed(..., include_min_cut=False)` - Maximum flow with cost distribution and optional min-cut; the min-cut is a true minimum cut (its capacity equals the max flow under the default `PROPORTIONAL` placement with `require_capacity=True` and `shortest_path=False`), not the set of saturated edges - `ctx.sensitivity(...)` - Identify critical edges and their impact on flow - `ctx.sensitivity_with_flow(...)` - Compute max flow and edge sensitivity together per group pair in a single pass (used by the sensitivity Monte Carlo hot path) - `ctx.shortest_path_cost(source, sink, *, mode, edge_select=ALL_MIN_COST, excluded_nodes, excluded_links)` - Shortest path cost diff --git a/docs/reference/design.md b/docs/reference/design.md index ba3b5ef..9e560ba 100644 --- a/docs/reference/design.md +++ b/docs/reference/design.md @@ -491,11 +491,11 @@ See "Routing Semantics: IP/IGP vs SDN/TE" section for detailed explanation. The residual network is maintained via `FlowState`, which tracks per-edge flow and computes residual capacities on demand. For each edge u→v: - Forward residual capacity: `capacity(u,v) - flow(u,v)` -- Reverse residual capacity: `flow(u,v)` (used for residual reachability when computing the min-cut and reachable set) +- Reverse residual capacity: `flow(u,v)` (traversed to return previously placed flow during the completion phase, and for residual reachability when computing the min-cut and reachable set) SPF operates over the residual graph by requesting edges with `require_capacity=true`, which filters to edges with positive residual capacity. The `FlowState` provides a residual capacity view without graph mutation. -Note: Reverse residual arcs are distinct from physical reverse edges added via `add_reverse=True` during graph construction. Physical reverse edges model bidirectional links with independent capacity; reverse residual arcs are bookkeeping over a single edge's flow. The augmenting SPF search traverses forward residual edges only — placed flow is never cancelled across tiers; reverse residual arcs are traversed only for reachability when computing the min-cut and reachable set, while Dinic-style reverse edges allow redistribution within a single tier's placement. +Note: Reverse residual arcs are distinct from physical reverse edges added via `add_reverse=True` during graph construction. Physical reverse edges model bidirectional links with independent capacity; reverse residual arcs are bookkeeping over a single edge's flow. The cost-tier SPF loop traverses forward residual edges only, so it cannot cancel an earlier placement and may stop below the true maximum. A completion phase then runs BFS augmentation over the full residual graph, traversing arcs backwards to return previously placed flow, which makes the result a true maximum flow whose min-cut matches it. The completion phase applies only to max-flow semantics — `PROPORTIONAL` placement with `require_capacity=True` and `shortest_path=False`; `EQUAL_BALANCED` (ECMP admission) and `require_capacity=False` (fixed-cost IP routing) are placement models rather than max-flow computations and keep the tier-loop result. Dinic-style reverse edges additionally allow redistribution within a single tier's placement. The core loop finds augmenting paths using the cost-aware SPF described above: @@ -521,7 +521,7 @@ After the loop, the C++ algorithm computes a FlowSummary which includes: - min_cut: the list of edges that are saturated and go from reachable to non-reachable (these form the minimum cut) -- cost_distribution: flow volume placed at each path cost tier. Core returns parallel arrays (`costs`, `flows`); AnalysisContext converts these to the `Dict[Cost, Flow]` mapping in `MaxFlowResult.cost_distribution`. +- cost_distribution: flow volume keyed by cost. Tier-loop entries are the cost of the shortest-path DAG the flow was placed on; completion-phase entries are marginal costs (the augmenting path's forward edge costs minus the cost of the flow it cancels), so a key need not correspond to any traversable path. Core returns parallel arrays (`costs`, `flows`); AnalysisContext converts these to the `Dict[Cost, Flow]` mapping in `MaxFlowResult.cost_distribution`. The summary is returned along with the total flow value. @@ -688,6 +688,22 @@ function MAX_FLOW(graph, S, T, placement=PROPORTIONAL, require_capacity=True, if shortest_path: # Single augmentation pass (IP/IGP mode) break + # Completion phase: max-flow semantics only. The tier loop above walks + # forward residual edges, so it can stop below the true maximum. + if placement == PROPORTIONAL and require_capacity and not shortest_path: + while True: + # BFS over the full residual graph, including reverse arcs that + # return previously placed flow + path = BFS_AUGMENTING_PATH(flow_state.residual_view(), S, T) + if path is None: + break + placed = flow_state.augment(path) + if placed < kMinFlow: + break + total_flow += placed + # Marginal cost: forward edge costs minus the cost of cancelled flow + cost_distribution[marginal_cost(path)] += placed + # Compute min-cut, reachability, cost distribution min_cut = flow_state.compute_min_cut(S, node_mask, edge_mask) @@ -703,7 +719,7 @@ function MAX_FLOW(graph, S, T, placement=PROPORTIONAL, require_capacity=True, The flow tolerance constant `kMinFlow` (1/4096 ≈ 2.4e-4) determines when flow placement is considered negligible and iteration terminates. -Each augmentation phase performs one SPF \(O((V+E) \log V)\) and one placement pass over the tier's predecessor DAG. For EQUAL_BALANCED the placement is a single topological pass \(O(V+E)\); for PROPORTIONAL it is a complete Dinic max-flow over the tier DAG (repeated BFS level construction, level-restricted blocking-flow DFS, and a group rebuild from the updated residual), worst case \(O(V^2 E)\). Placed flow is never removed from an edge, so each phase permanently saturates at least one edge before the next SPF runs, bounding the number of phases by \(O(E)\); with PROPORTIONAL placement the tier's path cost also strictly increases between phases, so phases are further bounded by the number of distinct path-cost values. The resulting loose worst-case bound is \(O(E \cdot (V^2 E + (V+E) \log V))\). +Each augmentation phase performs one SPF \(O((V+E) \log V)\) and one placement pass over the tier's predecessor DAG. For EQUAL_BALANCED the placement is a single topological pass \(O(V+E)\); for PROPORTIONAL it is a complete Dinic max-flow over the tier DAG (repeated BFS level construction, level-restricted blocking-flow DFS, and a group rebuild from the updated residual), worst case \(O(V^2 E)\). The tier loop never removes placed flow, so each phase permanently saturates at least one edge before the next SPF runs, bounding the number of phases by \(O(E)\); with PROPORTIONAL placement the tier's path cost also strictly increases between phases, so phases are further bounded by the number of distinct path-cost values. The resulting loose worst-case bound is \(O(E \cdot (V^2 E + (V+E) \log V))\). The completion phase that follows is Edmonds-Karp at \(O(V E^2)\), which this bound dominates. Practical performance is significantly better than these worst-case bounds: iteration stops as soon as the residual network disconnects source from sink, the phase count in practice equals the small number of cost tiers actually used, and the `kMinFlow` threshold additionally caps the number of phases at \(F / k_{MinFlow}\) for total flow \(F\). @@ -775,7 +791,7 @@ NetGraph's design includes several features that differentiate it from tradition - Configurable flow placement: Proportional (WCMP-style, capacity-weighted) and Equal-Balanced (ECMP-style, uniform) splitting across parallel equal-cost edges -- Cost-aware augmentation: Prefer cheapest capacity first via successive shortest paths. Does not re-route previously placed flow. +- Cost-aware augmentation: Prefer cheapest capacity first via successive shortest paths. The cost-tier loop does not re-route previously placed flow; the max-flow completion phase may cancel earlier placements to reach the true maximum. - Deterministic simulation with seeding: Random aspects (e.g., failure sampling) are controlled by explicit seeds that propagate through steps. Runs are reproducible given the same scenario and seed. diff --git a/docs/reference/dsl.md b/docs/reference/dsl.md index 19fb5b4..6f61579 100644 --- a/docs/reference/dsl.md +++ b/docs/reference/dsl.md @@ -1249,11 +1249,67 @@ demands: | `mode` | string | Node pairing mode: `combine` or `pairwise` (default: `combine`) | | `group_mode` | string | How grouped nodes produce demands (default: `flatten`) | | `flow_policy` | string or integer | Routing policy preset name (case-insensitive, or its integer value); inline policy mappings fail schema validation at scenario load | +| `static_paths` | array | Explicit routes to pin the demand to; see below | | `attrs` | object | Arbitrary metadata | | `expand` | object | Variable expansion block | Each demand receives an auto-generated unique `id`; an explicit `id` key is not accepted in scenario YAML. Duplicate ids can arise only when demands are constructed programmatically, and demand expansion rejects them with `ValueError`. +### Static Paths + +`static_paths` pins a demand to routes you choose, instead of letting the flow +policy pick them. This models MPLS-style LSPs: one flow per route, and a route +broken by a failure carries nothing rather than rerouting. + +```yaml +demands: + default: + - source: "^A$" + target: "^C$" + volume: 10 + mode: pairwise + flow_policy: SHORTEST_PATHS_WCMP + static_paths: + - ["A", "B", "C"] # node names, source first + - links: ["A|D|0", "D|C|0"] # or link ids, in traversal order +``` + +Each route is either a list of node names or a mapping with `nodes` or +`links`. Name nodes for readability; name links when parallel links connect +the same pair and you need a specific one. + +A route is a strict explicit route: every hop is one link. Where parallel +links connect a pair, a node hop takes the cheapest enabled one (ties broken +by link id), so the route carries that single link's capacity and fails when +that link fails — not when the whole bundle does. To model an LSP per parallel +link, list one route per link using the `links` form. Disabled links are never +chosen for a node hop, and naming one in the `links` form is an error. + +Semantics: + +- One flow per route, created in the order listed. How volume divides depends + on the preset: proportional presets such as `SHORTEST_PATHS_WCMP` fill the + routes in the order you list them, so a volume smaller than the first + route's bottleneck never reaches the second; equal-balanced presets give + every surviving route the same share, so total placement is limited by the + smallest surviving route. Route order is therefore significant under + proportional presets. +- A route whose nodes or links are excluded carries nothing. Traffic does not + move to another route, which is what makes this different from ordinary + routing. +- The policy neither adds routes nor reoptimizes, so preset cost ceilings and + LSP counts do not apply. + +Two demands pinned between the same source, target, and priority are rejected, +because their flows would collide. Put every route on one demand, or separate +the demands by priority. + +Because routes run between two concrete nodes, a demand using `static_paths` +must set `mode: pairwise` and use selectors matching exactly one source and +one target; anything else raises `ValueError` during expansion. Every route +must start at the source and end at the target, visit adjacent nodes, and not +revisit a node. + ### Selector Fields The `source` and `target` fields accept either: diff --git a/ngraph/analysis/demand.py b/ngraph/analysis/demand.py index 3fe2702..9666133 100644 --- a/ngraph/analysis/demand.py +++ b/ngraph/analysis/demand.py @@ -8,11 +8,11 @@ from __future__ import annotations from dataclasses import dataclass, replace -from typing import Dict, List +from typing import Dict, List, Tuple from ngraph.analysis.context import LARGE_CAPACITY, AugmentationEdge from ngraph.dsl.selectors import normalize_selector -from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.demand.spec import StaticPath, TrafficDemand from ngraph.model.flow.policy_config import FlowPolicyPreset from ngraph.model.network import Network, Node from ngraph.model.selectors import select_nodes @@ -31,6 +31,8 @@ class ExpandedDemand: volume: Traffic volume to place. priority: Priority class (lower is higher priority). policy_preset: FlowPolicy configuration preset. + static_paths: Routes this demand is pinned to, empty when it is + routed by the policy. """ src_name: str @@ -38,6 +40,7 @@ class ExpandedDemand: volume: float priority: int policy_preset: FlowPolicyPreset + static_paths: Tuple[StaticPath, ...] = () @dataclass @@ -144,6 +147,7 @@ def _expand_pairwise( volume=volume_per_pair, priority=td.priority, policy_preset=policy_preset, + static_paths=td.static_paths, ) for src, dst in pairs ] @@ -294,9 +298,11 @@ def expand_demands( DemandExpansion with demands and augmentations. Raises: - ValueError: If no demands could be expanded, or if two demands share - an id (pseudo node names embed the id, so duplicates would merge - distinct demands' attachment edges into one endpoint). + ValueError: If no demands could be expanded, if two demands share an + id (pseudo node names embed the id, so duplicates would merge + distinct demands' attachment edges into one endpoint), or if a + demand with `static_paths` does not resolve to exactly one + source/target pair. """ seen_ids: set[str] = set() for td in traffic_demands: @@ -320,6 +326,11 @@ def expand_demands( dst_groups = select_nodes(network, tgt_sel, default_active_only=True) if not src_groups or not dst_groups: + if td.static_paths: + raise ValueError( + f"Demand '{td.id}' sets static_paths but its selectors match " + "no active source or target node" + ) continue policy_preset = td.flow_policy or default_policy_preset @@ -329,6 +340,24 @@ def expand_demands( td, src_groups, dst_groups, policy_preset ) + if td.static_paths: + # Routes are pinned between two concrete nodes, so the demand has + # to name exactly one pair. Combine mode routes through pseudo + # endpoints, which no operator-supplied route can start from. + if td.mode == "combine": + raise ValueError( + f"Demand '{td.id}' sets static_paths, which pins traffic to " + "routes between two nodes; use mode 'pairwise' instead of " + "'combine'" + ) + if len(demands) != 1: + raise ValueError( + f"Demand '{td.id}' sets static_paths but its selectors " + f"expand to {len(demands)} source/target pairs; static " + "paths require selectors matching exactly one source and " + "one target" + ) + all_demands.extend(demands) all_augmentations.extend(augmentations) diff --git a/ngraph/analysis/functions.py b/ngraph/analysis/functions.py index 87dea1b..eea1055 100644 --- a/ngraph/analysis/functions.py +++ b/ngraph/analysis/functions.py @@ -26,12 +26,43 @@ from ngraph.analysis.demand import DemandExpansion, expand_demands from ngraph.analysis.placement import place_demands from ngraph.model.demand.builder import coerce_flow_policy -from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.demand.spec import StaticPath, TrafficDemand from ngraph.model.flow.policy_config import FlowPolicyPreset from ngraph.results.flow import FlowEntry, FlowIterationResult, FlowSummary from ngraph.types.base import FlowPlacement, Mode +def _static_paths_from_config(raw: Any) -> tuple[StaticPath, ...]: + """Rebuild pinned routes from their serialized form. + + Accepts every form the scenario DSL accepts, plus `StaticPath` objects + unchanged, so demand configs built in Python behave like ones produced by + `TrafficDemand.to_dict()`. + + Raises: + ValueError: If an entry is not a StaticPath, a list of node names, or + a mapping with exactly one of 'nodes' or 'links'. + """ + if not raw: + return () + paths = [] + for entry in raw: + if isinstance(entry, StaticPath): + paths.append(entry) + elif isinstance(entry, (list, tuple)): + paths.append(StaticPath(nodes=tuple(entry))) + elif isinstance(entry, dict) and set(entry) == {"nodes"}: + paths.append(StaticPath(nodes=tuple(entry["nodes"]))) + elif isinstance(entry, dict) and set(entry) == {"links"}: + paths.append(StaticPath(links=tuple(entry["links"]))) + else: + raise ValueError( + f"Invalid static path {entry!r}: expected a list of node names " + "or a mapping with exactly one of 'nodes' or 'links'" + ) + return tuple(paths) + + def _reconstruct_traffic_demands( demands_config: list[dict[str, Any]], ) -> list[TrafficDemand]: @@ -49,7 +80,7 @@ def _reconstruct_traffic_demands( Args: demands_config: List of demand configurations with fields: source, target, volume, mode, group_mode, flow_policy, - priority, attrs. + priority, static_paths, attrs. Returns: List of TrafficDemand objects with stable IDs. @@ -67,6 +98,7 @@ def _reconstruct_traffic_demands( group_mode=config.get("group_mode", "flatten"), flow_policy=coerce_flow_policy(config.get("flow_policy")), priority=config.get("priority", 0), + static_paths=_static_paths_from_config(config.get("static_paths")), attrs=config.get("attrs") or {}, ) ) diff --git a/ngraph/analysis/placement.py b/ngraph/analysis/placement.py index 81f1f8e..16f1b89 100644 --- a/ngraph/analysis/placement.py +++ b/ngraph/analysis/placement.py @@ -8,6 +8,8 @@ import netgraph_core import numpy as np +from ngraph.analysis.static_paths import build_static_path_bundles +from ngraph.model.demand.spec import StaticPath from ngraph.model.flow.policy_config import FlowPolicyPreset, create_flow_policy if TYPE_CHECKING: @@ -189,7 +191,7 @@ def place_demands( ): total_demand += volume - if demand.policy_preset in CACHEABLE_PRESETS: + if demand.policy_preset in CACHEABLE_PRESETS and not demand.static_paths: placed, cost_dist, used_edges, flow_idx_counter = _place_cached( src_id, dst_id, @@ -208,11 +210,21 @@ def place_demands( else: triple = (src_id, dst_id, demand.priority) if triple in policy_triples: + same_pair = ( + f"source '{demand.src_name}', destination " + f"'{demand.dst_name}', priority {demand.priority}" + ) + if demand.static_paths: + raise ValueError( + f"Two demands pinned to static paths share {same_pair}. " + "Their flow ids would collide and corrupt placement. " + "List every route on a single demand, or give the " + "demands distinct priorities." + ) raise ValueError( - f"Duplicate policy-based demand for source '{demand.src_name}', " - f"destination '{demand.dst_name}', priority {demand.priority}: " - "flow ids would collide and corrupt placement. Merge the " - "demand volumes or use distinct priorities." + f"Duplicate policy-based demand for {same_pair}: flow ids " + "would collide and corrupt placement. Merge the demand " + "volumes or use distinct priorities." ) policy_triples.add(triple) placed, cost_dist, used_edges = _place_with_policy( @@ -227,6 +239,9 @@ def place_demands( edge_mask, include_cost_distribution, include_used_edges, + static_paths=demand.static_paths, + src_name=demand.src_name, + dst_name=demand.dst_name, ) total_placed += placed @@ -382,15 +397,27 @@ def _place_with_policy( edge_mask: np.ndarray, include_cost_distribution: bool, include_used_edges: bool, + static_paths: Sequence[StaticPath] = (), + src_name: str = "", + dst_name: str = "", ) -> tuple[float, dict[float, float], set[str]]: - """Place single demand using FlowPolicy (for non-cacheable presets).""" + """Place a single demand using FlowPolicy. + + Used for non-cacheable presets and for any demand pinned to explicit + routes. With `static_paths` the policy is pinned to those routes: one flow + per route, and a route broken by the masks carries nothing. + """ policy = create_flow_policy( ctx.algorithms, ctx.handle, preset, node_mask=node_mask, edge_mask=edge_mask, + static_path_count=len(static_paths) or None, ) + if static_paths: + bundles = build_static_path_bundles(ctx, static_paths, src_name, dst_name) + policy.set_static_paths(src_id, dst_id, bundles) placed, _ = policy.place_demand(flow_graph, src_id, dst_id, priority, volume) cost_dist: dict[float, float] = {} diff --git a/ngraph/analysis/static_paths.py b/ngraph/analysis/static_paths.py new file mode 100644 index 0000000..b27db7f --- /dev/null +++ b/ngraph/analysis/static_paths.py @@ -0,0 +1,213 @@ +"""Resolution of explicit routes into Core path bundles. + +Turns the `StaticPath` entries on a demand into the `PredDAG` bundles that +`FlowPolicy.set_static_paths` pins traffic to. A bundle is a single simple +path: one edge per hop, so a route that names adjacent nodes with parallel +links between them picks one of those links (see `StaticPath`). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Tuple + +import netgraph_core + +from ngraph.model.demand.spec import StaticPath + +if TYPE_CHECKING: + from ngraph.analysis.context import AnalysisContext + +__all__ = ["build_static_path_bundles"] + +# Bundles depend only on the static graph, not on the per-iteration masks +# (Core prunes them against the masks itself), so they are resolved once per +# context and reused across Monte Carlo iterations and MSD probes. The cache +# lives on the context so it dies with it. +_CACHE_ATTR = "_static_path_bundle_cache" + + +def _disabled_edge_ids(ctx: "AnalysisContext") -> frozenset: + """Core edge ids belonging to administratively disabled links.""" + link_edges = ctx.link_id_to_edge_indices + return frozenset( + int(edge_id) + for link_id in ctx.disabled_link_ids + for edge_id in link_edges.get(link_id, ()) + ) + + +def _hop_edge( + src_id: int, dst_id: int, adjacency, cost, ext_ids, disabled_edges +) -> int: + """Return the edge id to use for one node-to-node hop. + + Picks the cheapest enabled edge from src_id to dst_id, breaking ties by + external edge id so the choice is stable across identical scenario + builds. Disabled links are skipped: pinning a route to one would take the + route down for the whole run even when an enabled parallel link exists. + Walks the source's adjacency row, so cost is proportional to that node's + degree rather than to the size of the graph. + """ + row, col, adj_edge = adjacency + best = -1 + best_key: Tuple[int, int] = (0, 0) + for i in range(int(row[src_id]), int(row[src_id + 1])): + if int(col[i]) != dst_id: + continue + edge_id = int(adj_edge[i]) + if edge_id in disabled_edges: + continue + key = (int(cost[edge_id]), int(ext_ids[edge_id])) + if best < 0 or key < best_key: + best, best_key = edge_id, key + return best + + +def _edges_from_nodes(ctx: "AnalysisContext", path: StaticPath) -> List[int]: + """Resolve a node-sequence route to one Core edge per hop.""" + graph = ctx.multidigraph + adjacency = ( + graph.row_offsets_view(), + graph.col_indices_view(), + graph.adj_edge_index_view(), + ) + cost = graph.cost_view() + ext_ids = graph.ext_edge_ids_view() + disabled_edges = _disabled_edge_ids(ctx) + + edges: List[int] = [] + for u, v in zip(path.nodes, path.nodes[1:], strict=False): + for name in (u, v): + if name not in ctx.node_mapper.node_id_of: + raise ValueError( + f"Static path names unknown node {name!r}; " + f"route was {list(path.nodes)}" + ) + u_id = ctx.node_mapper.to_id(u) + v_id = ctx.node_mapper.to_id(v) + edge_id = _hop_edge(u_id, v_id, adjacency, cost, ext_ids, disabled_edges) + if edge_id < 0: + raise ValueError( + f"Static path hop {u!r} -> {v!r} has no enabled link; " + f"route was {list(path.nodes)}" + ) + edges.append(edge_id) + return edges + + +def _edges_from_links( + ctx: "AnalysisContext", path: StaticPath, src_id: int +) -> List[int]: + """Resolve a link-id route to Core edges, following the traversal order.""" + graph = ctx.multidigraph + edge_src = graph.edge_src_view() + edge_dst = graph.edge_dst_view() + link_edges = ctx.link_id_to_edge_indices + + current = src_id + edges: List[int] = [] + for link_id in path.links: + candidates = link_edges.get(link_id) + if not candidates: + raise ValueError( + f"Static path names unknown link {link_id!r}; " + f"route was {list(path.links)}" + ) + # A link has a forward and (for bidirectional links) a reverse edge; + # pick whichever leaves the node the route has reached. + if link_id in ctx.disabled_link_ids: + raise ValueError( + f"Static path names disabled link {link_id!r}; a route pinned to " + "a disabled link can never carry traffic" + ) + chosen = next( + (int(e) for e in candidates if int(edge_src[int(e)]) == current), None + ) + if chosen is None: + reached = ctx.node_mapper.to_name(current) + raise ValueError( + f"Static path link {link_id!r} does not leave node {reached!r}; " + f"route was {list(path.links)}" + ) + edges.append(chosen) + current = int(edge_dst[chosen]) + return edges + + +def build_static_path_bundles( + ctx: "AnalysisContext", + paths: Sequence[StaticPath], + src_name: str, + dst_name: str, +) -> List[netgraph_core.PredDAG]: + """Build the Core path bundles a demand is pinned to. + + Args: + ctx: Context holding the built graph; routes resolve against it. + paths: Routes to pin, in the order flows should be created. + src_name: Node every route must start at. + dst_name: Node every route must end at. + + Returns: + One `PredDAG` per route, in the given order. Results are cached per + context, so repeated calls during a Monte Carlo run resolve once. + + Raises: + ValueError: If an endpoint is absent from the context graph, if a + route names an unknown node or a disabled/unknown link, has a hop + whose nodes are joined only by disabled links, has a hop with + no link, traverses a link that does not leave the node it has + reached, does not run from `src_name` to `dst_name`, or revisits + a node (a pinned route must be a simple path). + """ + cache_key = (src_name, dst_name, tuple(paths)) + per_ctx: Optional[Dict[tuple, List[netgraph_core.PredDAG]]] = getattr( + ctx, _CACHE_ATTR, None + ) + if per_ctx is None: + per_ctx = {} + setattr(ctx, _CACHE_ATTR, per_ctx) + cached = per_ctx.get(cache_key) + if cached is not None: + return cached + + graph = ctx.multidigraph + edge_src = graph.edge_src_view() + edge_dst = graph.edge_dst_view() + for name in (src_name, dst_name): + if name not in ctx.node_mapper.node_id_of: + raise ValueError( + f"Demand endpoint {name!r} is not in the analysis context graph, " + "so its static paths cannot be resolved" + ) + src_id = ctx.node_mapper.to_id(src_name) + dst_id = ctx.node_mapper.to_id(dst_name) + + bundles: List[netgraph_core.PredDAG] = [] + for path in paths: + if path.nodes: + if path.nodes[0] != src_name or path.nodes[-1] != dst_name: + raise ValueError( + f"Static path must run from {src_name!r} to {dst_name!r}, " + f"but runs from {path.nodes[0]!r} to {path.nodes[-1]!r}" + ) + edges = _edges_from_nodes(ctx, path) + else: + edges = _edges_from_links(ctx, path, src_id) + + if int(edge_src[edges[0]]) != src_id or int(edge_dst[edges[-1]]) != dst_id: + start = ctx.node_mapper.to_name(int(edge_src[edges[0]])) + end = ctx.node_mapper.to_name(int(edge_dst[edges[-1]])) + raise ValueError( + f"Static path must run from {src_name!r} to {dst_name!r}, " + f"but runs from {start!r} to {end!r}" + ) + + try: + bundles.append(netgraph_core.PredDAG.from_edges(graph, edges)) + except ValueError as exc: + route = list(path.nodes or path.links) + raise ValueError(f"Invalid static path {route}: {exc}") from exc + + per_ctx[cache_key] = bundles + return bundles diff --git a/ngraph/model/demand/builder.py b/ngraph/model/demand/builder.py index 2a9594c..f2e2c02 100644 --- a/ngraph/model/demand/builder.py +++ b/ngraph/model/demand/builder.py @@ -5,11 +5,11 @@ from __future__ import annotations -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from ngraph.dsl.expansion import ExpansionSpec, expand_block from ngraph.model.demand.matrix import DemandSet -from ngraph.model.demand.spec import TrafficDemand +from ngraph.model.demand.spec import StaticPath, TrafficDemand from ngraph.model.flow.policy_config import FlowPolicyPreset from ngraph.utils.yaml_utils import normalize_yaml_dict_keys @@ -92,9 +92,59 @@ def _build_demand(d: Dict[str, Any], set_name: str) -> TrafficDemand: if "flow_policy" in d: td_kwargs["flow_policy"] = coerce_flow_policy(d["flow_policy"]) + if "static_paths" in d: + td_kwargs["static_paths"] = _build_static_paths(d["static_paths"], set_name) + return TrafficDemand(**td_kwargs) +def _build_static_paths(raw: Any, set_name: str) -> Tuple[StaticPath, ...]: + """Build the pinned routes of one demand from its YAML form. + + Each entry is either a list of node names or a mapping with `nodes` or + `links`. + + Raises: + ValueError: If the block is not a non-empty list, or an entry is not + one of the accepted forms. + """ + if not isinstance(raw, list) or not raw: + raise ValueError( + f"'static_paths' in set '{set_name}' must be a non-empty list of routes" + ) + + def _names(values: Any, kind: str) -> Tuple[str, ...]: + if not isinstance(values, (list, tuple)) or not all( + isinstance(v, str) for v in values + ): + # Variable expansion substitutes native types, so a ${var} bound to + # a list reaches here despite the schema constraining route entries. + raise ValueError( + f"'static_paths' in set '{set_name}': {kind} must all be " + f"strings, got {values!r}" + ) + return tuple(values) + + paths: list[StaticPath] = [] + for entry in raw: + if isinstance(entry, list): + paths.append(StaticPath(nodes=_names(entry, "node names"))) + continue + if isinstance(entry, dict): + keys = set(entry) + if keys == {"nodes"}: + paths.append(StaticPath(nodes=_names(entry["nodes"], "node names"))) + continue + if keys == {"links"}: + paths.append(StaticPath(links=_names(entry["links"], "link ids"))) + continue + raise ValueError( + f"Each entry of 'static_paths' in set '{set_name}' must be a list " + "of node names, or a mapping with exactly one of 'nodes' or 'links'" + ) + return tuple(paths) + + def coerce_flow_policy(value: Any) -> Optional[FlowPolicyPreset]: """Return a FlowPolicyPreset from various user-friendly forms. diff --git a/ngraph/model/demand/spec.py b/ngraph/model/demand/spec.py index 246ccaa..2f3d5a5 100644 --- a/ngraph/model/demand/spec.py +++ b/ngraph/model/demand/spec.py @@ -1,11 +1,12 @@ """Traffic demand specification. Defines `TrafficDemand`, a user-facing specification used by demand expansion -and placement. Routing behavior is selected via an optional `FlowPolicyPreset`. +and placement. Routing behavior is selected via an optional `FlowPolicyPreset`, +or pinned to explicit routes with `StaticPath`. """ from dataclasses import dataclass, field -from typing import Any, Dict, Optional, Union +from typing import Any, Dict, Optional, Tuple, Union from ngraph.model.flow.policy_config import FlowPolicyPreset, serialize_policy_preset from ngraph.types.base import Mode @@ -16,6 +17,34 @@ _VALID_GROUP_MODES = ("flatten", "per_group", "group_pairwise") +@dataclass(frozen=True) +class StaticPath: + """One explicit route a demand can be pinned to (an MPLS-style LSP). + + Give exactly one of `nodes` or `links`: + + - `nodes`: the node names the route visits, source first and target last. + Each consecutive pair must be adjacent. When several parallel links + connect a pair, the cheapest is used (ties broken by link id); name the + link explicitly to choose a different one. + - `links`: the link ids the route traverses, in order. Unambiguous when + parallel links exist. A link may be traversed in either direction. + + Attributes: + nodes: Node names along the route, or empty when `links` is given. + links: Link ids along the route, or empty when `nodes` is given. + """ + + nodes: Tuple[str, ...] = () + links: Tuple[str, ...] = () + + def __post_init__(self) -> None: + if bool(self.nodes) == bool(self.links): + raise ValueError("StaticPath requires exactly one of 'nodes' or 'links'") + if self.nodes and len(self.nodes) < 2: + raise ValueError("StaticPath 'nodes' needs at least a source and a target") + + @dataclass class TrafficDemand: """Traffic demand specification using unified selectors. @@ -29,6 +58,10 @@ class TrafficDemand: group_mode: How grouped nodes produce demands ("flatten", "per_group", "group_pairwise"). flow_policy: Policy preset for routing. + static_paths: Explicit routes to pin this demand to. When set, the + demand is placed only on these routes: one flow per route, and a + route broken by a failure carries nothing rather than rerouting. + Requires selectors matching exactly one source and one target. attrs: Arbitrary user metadata. id: Unique identifier. Auto-generated if empty. """ @@ -40,6 +73,7 @@ class TrafficDemand: mode: str = "combine" group_mode: str = "flatten" flow_policy: Optional[FlowPolicyPreset] = None + static_paths: Tuple[StaticPath, ...] = () attrs: Dict[str, Any] = field(default_factory=dict) id: str = "" @@ -55,6 +89,12 @@ def __post_init__(self) -> None: f"Unknown demand group_mode '{self.group_mode}'. " f"Expected one of: {', '.join(_VALID_GROUP_MODES)}" ) + for path in self.static_paths: + if not isinstance(path, StaticPath): + raise ValueError( + f"static_paths entries must be StaticPath objects, got " + f"{type(path).__name__}" + ) if not self.id: # Build a stable identifier from source/target src_key = self.source if isinstance(self.source, str) else str(self.source) @@ -77,5 +117,9 @@ def to_dict(self) -> Dict[str, Any]: "mode": self.mode, "group_mode": self.group_mode, "flow_policy": serialize_policy_preset(self.flow_policy), + "static_paths": [ + {"nodes": list(p.nodes)} if p.nodes else {"links": list(p.links)} + for p in self.static_paths + ], "attrs": dict(self.attrs), } diff --git a/ngraph/model/flow/policy_config.py b/ngraph/model/flow/policy_config.py index d1d98e3..a32069d 100644 --- a/ngraph/model/flow/policy_config.py +++ b/ngraph/model/flow/policy_config.py @@ -82,6 +82,7 @@ def create_flow_policy( preset: FlowPolicyPreset, node_mask=None, edge_mask=None, + static_path_count: Optional[int] = None, ) -> netgraph_core.FlowPolicy: """Create a FlowPolicy instance from a preset configuration. @@ -92,6 +93,9 @@ def create_flow_policy( flow-count bounds to apply. node_mask: Optional numpy bool array for node exclusions (True = include). edge_mask: Optional numpy bool array for edge exclusions (True = include). + static_path_count: Number of routes the caller will pin with + `FlowPolicy.set_static_paths`. Sets the flow count to match, since + a pinned policy creates one flow per route and never grows. Returns: netgraph_core.FlowPolicy: Configured policy instance. @@ -105,6 +109,20 @@ def create_flow_policy( >>> graph = algs.build_graph(strict_multidigraph) >>> policy = create_flow_policy(algs, graph, FlowPolicyPreset.SHORTEST_PATHS_ECMP) """ + + def _build(config: netgraph_core.FlowPolicyConfig) -> netgraph_core.FlowPolicy: + if static_path_count is not None: + # A pinned policy creates exactly one flow per route, so the flow + # bounds must match; Core rejects a mismatch. Cost ceilings and + # reoptimization are inert once paths are pinned. + config.min_flow_count = 1 + config.max_flow_count = static_path_count + config.reoptimize_flows_on_each_placement = False + config.shortest_path = False + return netgraph_core.FlowPolicy( + algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask + ) + if preset == FlowPolicyPreset.SHORTEST_PATHS_ECMP: # Hop-by-hop equal-cost balanced routing (similar to IP forwarding with ECMP) config = netgraph_core.FlowPolicyConfig() @@ -117,9 +135,7 @@ def create_flow_policy( ) config.min_flow_count = 1 config.max_flow_count = 1 - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.SHORTEST_PATHS_WCMP: # Hop-by-hop weighted ECMP (WCMP) over equal-cost paths (proportional split) @@ -133,9 +149,7 @@ def create_flow_policy( ) config.min_flow_count = 1 config.max_flow_count = 1 - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.TE_WCMP_UNLIM: # Traffic engineering with WCMP (proportional split) and capacity-aware selection @@ -149,9 +163,7 @@ def create_flow_policy( ) config.min_flow_count = 1 # max_flow_count defaults to None (unlimited) - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.TE_ECMP_UP_TO_256_LSP: # TE with up to 256 LSPs using ECMP flow placement @@ -168,9 +180,7 @@ def create_flow_policy( config.min_flow_count = 1 config.max_flow_count = 256 config.reoptimize_flows_on_each_placement = True - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) elif preset == FlowPolicyPreset.TE_ECMP_16_LSP: # TE with exactly 16 LSPs using ECMP flow placement @@ -187,9 +197,7 @@ def create_flow_policy( config.min_flow_count = 16 config.max_flow_count = 16 config.reoptimize_flows_on_each_placement = True - return netgraph_core.FlowPolicy( - algorithms, graph, config, node_mask=node_mask, edge_mask=edge_mask - ) + return _build(config) else: raise ValueError(f"Unknown flow policy preset: {preset}") diff --git a/ngraph/schemas/scenario.json b/ngraph/schemas/scenario.json index 5ff8a62..3ce5515 100644 --- a/ngraph/schemas/scenario.json +++ b/ngraph/schemas/scenario.json @@ -31,7 +31,9 @@ } } }, - "required": ["name"], + "required": [ + "name" + ], "additionalProperties": false } ] @@ -674,6 +676,55 @@ "attrs": { "type": "object", "description": "Additional demand attributes" + }, + "static_paths": { + "type": "array", + "description": "Explicit routes to pin this demand to (one flow per route, no reroute on failure). Requires selectors matching exactly one source and one target.", + "minItems": 1, + "items": { + "oneOf": [ + { + "type": "array", + "description": "Node names along the route, source first and target last", + "minItems": 2, + "items": { + "type": "string" + } + }, + { + "type": "object", + "properties": { + "nodes": { + "type": "array", + "minItems": 2, + "items": { + "type": "string" + } + } + }, + "required": [ + "nodes" + ], + "additionalProperties": false + }, + { + "type": "object", + "properties": { + "links": { + "type": "array", + "minItems": 1, + "items": { + "type": "string" + } + } + }, + "required": [ + "links" + ], + "additionalProperties": false + } + ] + } } }, "required": [ diff --git a/ngraph/types/dto.py b/ngraph/types/dto.py index 25f9f49..1ee38a0 100644 --- a/ngraph/types/dto.py +++ b/ngraph/types/dto.py @@ -38,7 +38,9 @@ class MaxFlowResult: Attributes: total_flow: Maximum flow value achieved. - cost_distribution: Mapping of path cost to flow volume placed at that cost. + cost_distribution: Mapping of cost to flow volume placed at that cost. + Completion-phase entries are marginal costs, so a key need not match + any single path's cost. min_cut: Edges forming a minimum cut (None if not computed). """ diff --git a/pyproject.toml b/pyproject.toml index 71faf55..0999c0a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,7 +35,7 @@ dependencies = [ "pyyaml>=6.0", "pandas>=2.0", "jsonschema>=4.0", - "netgraph-core>=0.7.0", + "netgraph-core>=0.8.0", ] [project.urls] diff --git a/tests/analysis/test_static_paths.py b/tests/analysis/test_static_paths.py new file mode 100644 index 0000000..90838ca --- /dev/null +++ b/tests/analysis/test_static_paths.py @@ -0,0 +1,407 @@ +"""Demands pinned to explicit routes (MPLS-style LSPs). + +Covers the property that distinguishes a pinned route from ordinary routing: +a route broken by a failure carries nothing instead of rerouting. +""" + +from __future__ import annotations + +import netgraph_core +import pytest + +from ngraph.analysis import analyze +from ngraph.analysis.demand import expand_demands +from ngraph.analysis.placement import place_demands +from ngraph.analysis.static_paths import build_static_path_bundles +from ngraph.model.demand.spec import StaticPath, TrafficDemand +from ngraph.model.flow.policy_config import FlowPolicyPreset +from ngraph.model.network import Link, Network, Node + +SHORT = StaticPath(nodes=("A", "B", "C")) +LONG = StaticPath(nodes=("A", "D", "C")) + + +def _network() -> Network: + """A->B->C carries 6 at cost 1/hop; A->D->C carries 4 at cost 9/hop.""" + net = Network() + for name in ("A", "B", "D", "C"): + net.add_node(Node(name)) + net.add_link(Link("A", "B", capacity=6.0, cost=1)) + net.add_link(Link("B", "C", capacity=6.0, cost=1)) + net.add_link(Link("A", "D", capacity=4.0, cost=9)) + net.add_link(Link("D", "C", capacity=4.0, cost=9)) + return net + + +def _place( + paths, + *, + excluded_links=None, + volume: float = 10.0, + preset: FlowPolicyPreset = FlowPolicyPreset.SHORTEST_PATHS_WCMP, + net: Network | None = None, +): + net = net or _network() + demand = TrafficDemand( + source="^A$", + target="^C$", + volume=volume, + mode="pairwise", + flow_policy=preset, + static_paths=tuple(paths), + id="d", + ) + expansion = expand_demands(net, [demand]) + ctx = analyze(net) + flow_graph = netgraph_core.FlowGraph(ctx.multidigraph) + return place_demands( + expansion.demands, + [d.volume for d in expansion.demands], + flow_graph, + ctx, + ctx.build_node_mask(None), + ctx.build_edge_mask(excluded_links), + collect_entries=True, + include_used_edges=True, + ) + + +class TestPlacement: + def test_uses_every_pinned_route(self): + result = _place([SHORT, LONG]) + assert result.summary.total_placed == pytest.approx(10.0) + assert result.entries[0].used_edges == { + "A|B|0:fwd", + "B|C|0:fwd", + "A|D|0:fwd", + "D|C|0:fwd", + } + + def test_pins_traffic_off_the_cheapest_route(self): + """Ordinary routing would send everything down the cost-2 path.""" + result = _place([LONG]) + assert result.summary.total_placed == pytest.approx(4.0) + assert result.entries[0].used_edges == {"A|D|0:fwd", "D|C|0:fwd"} + + def test_broken_route_carries_nothing_and_does_not_reroute(self): + result = _place([SHORT], excluded_links={"B|C|0"}) + assert result.summary.total_placed == pytest.approx(0.0) + + def test_surviving_routes_still_carry_their_share(self): + result = _place([SHORT, LONG], excluded_links={"B|C|0"}) + assert result.summary.total_placed == pytest.approx(4.0) + + def test_link_form_selects_a_specific_parallel_link(self): + net = _network() + net.add_link(Link("A", "B", capacity=1.0, cost=50)) # A|B|1 + cheap = _place([StaticPath(links=("A|B|0", "B|C|0"))], net=net) + pricey = _place([StaticPath(links=("A|B|1", "B|C|0"))], net=_rebuild(net)) + assert cheap.summary.total_placed == pytest.approx(6.0) + assert pricey.summary.total_placed == pytest.approx(1.0) + + def test_node_form_prefers_the_cheapest_parallel_link(self): + net = _network() + net.add_link(Link("A", "B", capacity=1.0, cost=50)) + result = _place([SHORT], net=net) + assert result.summary.total_placed == pytest.approx(6.0) + + +def _rebuild(source: Network) -> Network: + """Copy a network so each placement starts from clean capacity.""" + net = Network() + for name in source.nodes: + net.add_node(Node(name)) + for link in source.links.values(): + net.add_link( + Link(link.source, link.target, capacity=link.capacity, cost=link.cost) + ) + return net + + +class TestResolution: + def _ctx(self): + return analyze(_network()) + + def test_rejects_non_adjacent_hop(self): + with pytest.raises(ValueError, match="has no enabled link"): + build_static_path_bundles( + self._ctx(), [StaticPath(nodes=("A", "C"))], "A", "C" + ) + + def test_rejects_unknown_node(self): + with pytest.raises(ValueError, match="unknown node"): + build_static_path_bundles( + self._ctx(), [StaticPath(nodes=("A", "B", "Z", "C"))], "A", "C" + ) + + def test_rejects_unknown_link(self): + with pytest.raises(ValueError, match="unknown link"): + build_static_path_bundles( + self._ctx(), [StaticPath(links=("nope",))], "A", "C" + ) + + def test_rejects_link_not_leaving_the_reached_node(self): + with pytest.raises(ValueError, match="does not leave node"): + build_static_path_bundles( + self._ctx(), [StaticPath(links=("B|C|0",))], "A", "C" + ) + + def test_rejects_route_with_wrong_endpoints(self): + with pytest.raises(ValueError, match="must run from"): + build_static_path_bundles( + self._ctx(), [StaticPath(nodes=("A", "B"))], "A", "C" + ) + + def test_rejects_route_revisiting_a_node(self): + net = _network() + net.add_link(Link("B", "A", capacity=1.0, cost=1)) + ctx = analyze(net) + with pytest.raises(ValueError, match="simple path"): + build_static_path_bundles( + ctx, [StaticPath(nodes=("A", "B", "A", "B", "C"))], "A", "C" + ) + + +class TestSpec: + def test_requires_exactly_one_form(self): + with pytest.raises(ValueError, match="exactly one"): + StaticPath() + with pytest.raises(ValueError, match="exactly one"): + StaticPath(nodes=("A", "B"), links=("A|B|0",)) + + def test_requires_two_nodes(self): + with pytest.raises(ValueError, match="source and a target"): + StaticPath(nodes=("A",)) + + def test_round_trips_through_to_dict(self): + demand = TrafficDemand( + source="^A$", + target="^C$", + static_paths=(SHORT, StaticPath(links=("A|B|0",))), + ) + assert demand.to_dict()["static_paths"] == [ + {"nodes": ["A", "B", "C"]}, + {"links": ["A|B|0"]}, + ] + + +class TestExpansionConstraints: + def test_rejects_combine_mode(self): + net = _network() + demand = TrafficDemand( + source="^A$", + target="^C$", + volume=1.0, + mode="combine", + static_paths=(SHORT,), + id="d", + ) + with pytest.raises(ValueError, match="use mode 'pairwise'"): + expand_demands(net, [demand]) + + def test_rejects_selectors_matching_several_pairs(self): + net = _network() + net.add_node(Node("A2")) + net.add_link(Link("A2", "B", capacity=1.0, cost=1)) + demand = TrafficDemand( + source="^A", + target="^C$", + volume=1.0, + mode="pairwise", + static_paths=(SHORT,), + id="d", + ) + with pytest.raises(ValueError, match="exactly one source and one target"): + expand_demands(net, [demand]) + + +class TestScenarioYaml: + """The YAML surface: both spellings, and schema rejection of bad forms.""" + + _NETWORK = """ +network: + nodes: + A: {} + B: {} + D: {} + C: {} + links: + - source: A + target: B + capacity: 6 + cost: 1 + - source: B + target: C + capacity: 6 + cost: 1 + - source: A + target: D + capacity: 4 + cost: 9 + - source: D + target: C + capacity: 4 + cost: 9 +""" + + def _scenario(self, static_paths_block: str) -> str: + return ( + self._NETWORK + + """ +demands: + default: + - source: "^A$" + target: "^C$" + volume: 10 + mode: pairwise + flow_policy: SHORTEST_PATHS_WCMP + static_paths: +""" + + static_paths_block + ) + + def test_node_and_link_forms_both_load_and_place(self): + from ngraph.scenario import Scenario + + scenario = Scenario.from_yaml( + self._scenario( + ' - ["A", "B", "C"]\n - links: ["A|D|0", "D|C|0"]\n' + ) + ) + demand = scenario.demand_set.get_set("default")[0] + assert [p.nodes or p.links for p in demand.static_paths] == [ + ("A", "B", "C"), + ("A|D|0", "D|C|0"), + ] + + def test_mapping_form_with_nodes_key(self): + from ngraph.scenario import Scenario + + scenario = Scenario.from_yaml( + self._scenario(' - nodes: ["A", "B", "C"]\n') + ) + assert scenario.demand_set.get_set("default")[0].static_paths[0].nodes == ( + "A", + "B", + "C", + ) + + def test_schema_rejects_empty_and_malformed_routes(self): + import jsonschema + + from ngraph.scenario import Scenario + + with pytest.raises(jsonschema.ValidationError): + Scenario.from_yaml(self._scenario(" []\n")) + with pytest.raises(jsonschema.ValidationError): + Scenario.from_yaml( + self._scenario(' - {nodes: ["A", "B"], links: ["A|B|0"]}\n') + ) + with pytest.raises(jsonschema.ValidationError): + Scenario.from_yaml(self._scenario(' - ["A"]\n')) + + +class TestDisabledLinks: + """A route must never be pinned to an administratively disabled link.""" + + def _network_with_disabled_cheap_link(self) -> Network: + net = Network() + for name in ("A", "B", "C"): + net.add_node(Node(name)) + net.add_link(Link("A", "B", capacity=6.0, cost=1)) # cheapest, disabled + net.add_link(Link("A", "B", capacity=7.0, cost=2)) # enabled alternative + net.add_link(Link("B", "C", capacity=100.0, cost=1)) + net.links["A|B|0"].disabled = True + return net + + def test_node_hop_skips_disabled_link(self): + result = _place( + [StaticPath(nodes=("A", "B", "C"))], + net=self._network_with_disabled_cheap_link(), + ) + assert result.summary.total_placed == pytest.approx(7.0) + assert "A|B|1:fwd" in result.entries[0].used_edges + + def test_naming_a_disabled_link_is_an_error(self): + ctx = analyze(self._network_with_disabled_cheap_link()) + with pytest.raises(ValueError, match="disabled link"): + build_static_path_bundles( + ctx, [StaticPath(links=("A|B|0", "B|C|0"))], "A", "C" + ) + + def test_hop_with_only_disabled_links_is_an_error(self): + net = self._network_with_disabled_cheap_link() + net.links["A|B|1"].disabled = True + with pytest.raises(ValueError, match="has no enabled link"): + build_static_path_bundles( + analyze(net), [StaticPath(nodes=("A", "B", "C"))], "A", "C" + ) + + +class TestInputForms: + """Every route form the DSL accepts must work programmatically too.""" + + def test_config_round_trip_accepts_list_and_mapping_forms(self): + from ngraph.analysis.functions import _static_paths_from_config + + assert _static_paths_from_config([["A", "B", "C"]]) == ( + StaticPath(nodes=("A", "B", "C")), + ) + assert _static_paths_from_config([{"links": ["A|B|0"]}]) == ( + StaticPath(links=("A|B|0",)), + ) + assert _static_paths_from_config([StaticPath(nodes=("A", "B"))]) == ( + StaticPath(nodes=("A", "B")), + ) + + @pytest.mark.parametrize( + "bad", + [[{"path": ["A", "B"]}], ["not-a-route"], [{"nodes": ["A"], "links": ["x"]}]], + ) + def test_config_round_trip_rejects_other_shapes(self, bad): + from ngraph.analysis.functions import _static_paths_from_config + + with pytest.raises(ValueError, match="Invalid static path"): + _static_paths_from_config(bad) + + def test_traffic_demand_rejects_non_staticpath_entries(self): + with pytest.raises(ValueError, match="must be StaticPath objects"): + TrafficDemand( + source="^A$", target="^B$", static_paths=[{"nodes": ["A", "B"]}] + ) + + def test_yaml_rejects_non_string_hops(self): + from ngraph.model.demand.builder import _build_static_paths + + with pytest.raises(ValueError, match="must all be strings"): + _build_static_paths([["A", ["B", "B"], "C"]], "default") + + +class TestPinnedDemandIsNeverSilentlyDropped: + def test_selector_matching_nothing_raises(self): + net = _network() + demand = TrafficDemand( + source="^A$", + target="^NOPE$", + volume=5.0, + mode="pairwise", + static_paths=(SHORT,), + id="d", + ) + with pytest.raises(ValueError, match="match no active source or target"): + expand_demands(net, [demand]) + + +class TestBundleCaching: + """Bundles are resolved once per context but must still track the masks.""" + + def test_repeated_placements_track_changing_failures(self): + outcomes = [] + for excluded in (None, {"B|C|0"}, None, {"A|D|0"}, {"B|C|0", "A|D|0"}, None): + outcomes.append( + round( + _place([SHORT, LONG], excluded_links=excluded).summary.total_placed, + 3, + ) + ) + assert outcomes == [10.0, 4.0, 10.0, 6.0, 0.0, 10.0]