feat(placement): add the sie-placement tune KEDA applier

packages/sie_placement/README.md

@@ -3,8 +3,9 @@
 Versioned documents for a measured-only placement planner: the service level
 each served identity must meet, the measurements a placement may be chosen
 from, the prices those placements cost, the demand they serve, and the plan
-that results. The package uses only the Python standard library. It is a
-workspace package and is not published to PyPI.
+that results. The package uses only the Python standard library, except the
+KEDA applier, whose `keda` extra adds the SIE Python SDK. It is a workspace
+package and is not published to PyPI.
 
 A served identity is a model id or a `model:profile` pair, the routing key a
 request resolves to. A measurement cell is a machine profile together with the
@@ -133,6 +134,156 @@ start: read back as `current`, the settings climb one step per tick under
 steady demand until the lane serves the demand or reaches the top of its
 envelope.
 
+## KEDA applier
+
+`sie-placement tune` keeps each plan lane's warm floor inside its envelope on a
+`sie-cluster` deployment that autoscales workers with KEDA. It needs the `keda`
+extra.
+
+```bash
+sie-placement tune --plan plan.json \
+  --prometheus-url http://prometheus:9090 \
+  --gateway-url http://sie-gateway:8080 \
+  --api-key-file /run/secrets/gateway-client-token
+```
+
+Every tick, 60 seconds apart by default or once with `--once`, it reads the
+four lane series below from Prometheus, reads each lane's floor-carrier pool
+from the gateway with `SIEClient.get_pool`, and calls `step` with the carrier's
+floor as `current`. It prints one JSON audit row per lane to standard output. Without
+`--apply` it runs in shadow mode: it decides and records and writes nothing.
+With `--apply` it upserts every lane's carrier on every tick through
+`SIEClient.create_pool`, which renews the carrier's lease and sets its floor to
+the step's floor. Each write uses its own SDK client and closes it after the
+write, so the SDK's background lease renewal never runs: a carrier's lease is
+renewed only by a tick that writes it. A lane that has no carrier and whose
+floor stays at 0 gets none.
+
+Under KEDA only the floor moves per lane. The scale-down window and the queue
+threshold are chart-wide settings, and there is no buffer to tune, so the
+applier passes `Settings(floor=n)` and ignores the envelope's buffer and window
+ranges.
+
+### Floor carriers
+
+A floor carrier is an API logical pool named
+`placement-floor-<queuePool>-<machineProfile>-<bundle>`, with `queue_pool` and
+`bundle` set to the lane's, `gpus` set to `{<machineProfile>: 0}`, the
+gateway's default lease, no GPU caps, no pinned models, and
+`minimum_worker_count` set to the step's floor. The gateway keeps each lane at
+the largest floor among the logical pools on it, so the carrier can raise the
+lane above its chart floor and never lowers another pool's floor. The applier
+writes only its carriers: a pool that has a carrier's name and any other spec is
+reported and left unchanged. A carrier carries no pins, because every floor
+change is a spec change, which resets the pool's worker assignment; pinned
+models stay in the chart values.
+
+A tick that cannot read a lane's carrier, or finds a pool with its name that is
+not its carrier, does not write it. A carrier that no tick writes, because those
+reads keep failing or the applier has stopped, lapses 1200 seconds after its
+last write by default, the gateway removes it, and the lane returns to its
+chart floor. On a lane whose chart `minReplicas` is 0, a floor of 1 or more
+activates the lane, which then holds at least the chart's fallback floor.
+
+### Series
+
+| Series | Query | Step input |
+| --- | --- | --- |
+| Queue depth | `max by (pool, machine_profile, bundle) (sie_gateway_lane_queue_depth{} and on(producer_instance, collector_generation, pool, machine_profile, bundle) (abs(time() - sie_gateway_lane_queue_snapshot_timestamp_seconds{}) < 20))` | `running` |
+| Pending demand | `max by (pool, machine_profile, bundle) (sie_gateway_pending_demand{} and on(producer_instance, collector_generation) (abs(time() - sie_gateway_capacity_snapshot_timestamp_seconds{}) < 20))` | `backlog` |
+| Warm floor | `max by (pool, machine_profile, bundle) (sie_gateway_pool_warm_floor{} and on(producer_instance, collector_generation) (abs(time() - sie_gateway_capacity_snapshot_timestamp_seconds{}) < 20))` | Reported only |
+| Running workers | `count by (lane) (max by (lane, producer_instance) (sie_worker_gpu_slots{state="ready"}) > 0)` | `runners` |
+
+`--series-selector` adds label matchers to every selector, timestamps included,
+for example `namespace="sie"` when one Prometheus scrapes several releases.
+Lane labels match the plan's lane tokens after trimming and lowering them, as
+the chart names lanes.
+
+The collector keeps a producer's last point for minutes after the producer
+stops reporting, so each gateway series keeps only the producers whose snapshot
+is at most 20 seconds old, as the chart's KEDA triggers do: the lane's own
+queue snapshot for queue depth, and the producer's capacity snapshot for
+pending demand and warm floor. A lane whose gateway values are all stale reads
+as missing. The worker series has no snapshot timestamp, so a worker that
+stopped reporting still counts among the runners until the collector drops it.
+
+Pending demand is a 0 or 1 flag per lane, not a count. The gateway sets it
+while a request that found no ready capacity is recent, while any request's
+durable publish to the lane awaits its acknowledgement, and for 120 seconds
+after a failed dispatch. The attainment is `unmeasured`, so a lane is under
+pressure whenever the flag is set, and idle when neither the flag nor queued or
+in-flight work is reported. Because a publish in flight sets the flag, any
+traffic can read as pressure even when the lane has capacity to spare, and a
+lane that serves steady traffic seldom reads idle, so its floor can climb
+towards `floorMax` while traffic lasts and falls one step per idle tick
+afterwards. Run the applier in shadow mode first to see how often each lane
+reads as under pressure. A lane with no
+ready worker has 0 runners. A lane whose queue depth, pending demand or worker
+series is missing is held: the step is given a busy lane with nothing waiting,
+so it only brings the carrier's floor inside the envelope.
+
+### Audit rows
+
+Each row carries `at`, `mode` (`shadow` or `apply`), the lane's `lane`,
+`queuePool`, `machineProfile` and `bundle`, the `carrier` pool name, the
+`telemetry` read for the lane (`pendingDemand` is the 0 or 1 flag),
+`currentFloor`, the step's `action` and `floor`, whether the write was
+`applied`, `laneFloorAboveEnvelope` and `error`.
+`laneFloorAboveEnvelope` is true when the lane's warm floor exceeds `floorMax`.
+The applier does not lower another pool's floor, so a lane held above its
+envelope by another pool stays reported until that pool changes.
+
+### Refusals
+
+The command exits with status 2 before it calls Prometheus or the gateway when:
+
+- a lane's `floorMax` exceeds `--max-pool-floor`, which must match the gateway's
+  `SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT` (default 4);
+- the plan has more lanes than `--max-pools`, which must match the gateway's
+  `SIE_GATEWAY_MAX_POOLS` (default 64);
+- a lane's `queuePool`, machine profile or bundle is not 1 to 63 letters,
+  digits, `_` or `-`, two lanes need the same carrier, or a carrier name is
+  longer than the gateway's 128 characters;
+- the `sie-cluster` chart cannot run the plan: a lane on the `default` queue
+  pool pins an identity, whose pins no static queue pool carries, or an
+  identity is `bridged`, `snapshot-cold` or `scheduled-warm`, warmths the chart
+  does not realize;
+- a token file or `SIE_ADMIN_TOKEN` holds anything but one run of printable
+  ASCII characters without spaces after trimming;
+- `--prometheus-url` or `--gateway-url` is not a valid absolute http or https
+  URL with a host, embeds credentials, carries a query or a fragment, or has an
+  invalid port. The refusal does not repeat the URL.
+
+A failed query, pool read or pool write fails only its tick, which is recorded,
+and the next tick runs as usual.
+
+With `--once`, a tick whose query, pool read or pool write failed exits with
+status 1.
+
+### Credentials
+
+| Call | Permission | Source |
+| --- | --- | --- |
+| Prometheus `/api/v1/query` | Read access to the query API | None sent |
+| `GET /v1/pools/{name}` | A gateway client token, when gateway token auth is on | `--api-key-file`, or `SIE_API_KEY` when `SIE_BASE_URL` names the same origin, as the SDK resolves it |
+| `POST /v1/pools`, which also renews the lease | The gateway admin token | `--admin-token-file`, or `SIE_ADMIN_TOKEN`, read only with `--apply` |
+
+The gateway admin token authorizes every gateway admin route, not only pools.
+The applier reads it only with `--apply`, sends it only on pool writes to
+`--gateway-url`, never takes it as a command-line value and never prints or
+logs it. Without it, `--apply` sends its writes with no credential, which a
+gateway with token auth refuses. The client token is never sent on writes. A
+refused token is named by its source, never by its value. A gateway call that
+gets an error reply is reported by its HTTP status alone, and so is a reply
+that is neither a success nor an error, such as a redirect; a success reply the
+SDK cannot parse is reported as an unexpected response. No text a server
+returns reaches an audit row, so an intermediary that quotes the request's
+`Authorization` header in its error body cannot put the token there. A failed
+connection is reported by the SDK's connection error message, sent before any
+header, and any other transport failure by its exception type alone. A message
+the SDK raises before any reply is cut to 512 characters, with the token the
+client holds replaced by `[redacted]`.
+
 ## Tests
 
 ```bash

packages/sie_placement/pyproject.toml

@@ -6,6 +6,15 @@ requires-python = ">=3.12,<3.13"
 license = { text = "Apache-2.0" }
 dependencies = []
 
+[project.optional-dependencies]
+keda = [
+    "httpx>=0.28,<1",
+    "sie-sdk>=0.11",
+]
+
+[project.scripts]
+sie-placement = "sie_placement.cli:main"
+
 [build-system]
 requires = ["hatchling"]
 build-backend = "hatchling.build"

packages/sie_placement/src/sie_placement/cli.py

@@ -0,0 +1,183 @@
+"""Command-line entry point of the placement tools."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import os
+import re
+import sys
+import time
+from collections.abc import Callable, Sequence
+from pathlib import Path
+
+from sie_placement import keda
+from sie_placement.schema import Plan
+
+ADMIN_TOKEN_ENV = "SIE_ADMIN_TOKEN"  # noqa: S105 - an environment variable name
+SDK_API_KEY_ENV = "SIE_API_KEY"
+DEFAULT_INTERVAL_S = 60.0
+EXIT_FAILED = 1
+EXIT_REFUSED = 2
+
+_TOKEN = re.compile(r"[\x21-\x7e]+")
+
+
+class _UsageError(ValueError):
+    pass
+
+
+def _positive_float(text: str) -> float:
+    value = float(text)
+    if not math.isfinite(value) or value <= 0:
+        raise argparse.ArgumentTypeError(f"must be greater than 0, got {text!r}")
+    return value
+
+
+def _integer_at_least(minimum: int) -> Callable[[str], int]:
+    def parse(text: str) -> int:
+        value = int(text)
+        if value < minimum:
+            raise argparse.ArgumentTypeError(f"must be at least {minimum}, got {text!r}")
+        return value
+
+    return parse
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="sie-placement", description="Placement planner tools.")
+    commands = parser.add_subparsers(dest="command", required=True)
+    tune = commands.add_parser(
+        "tune",
+        help="keep each plan lane's warm floor inside its envelope through gateway floor-carrier pools",
+        description=(
+            "Read each plan lane's series from Prometheus, decide one tuner step per lane and, with --apply, "
+            "upsert one floor-carrier pool per lane through the gateway pool API. Without --apply nothing is written."
+        ),
+    )
+    tune.add_argument("--plan", required=True, type=Path, help="placement-plan document")
+    tune.add_argument("--prometheus-url", required=True, help="Prometheus base URL that serves /api/v1/query")
+    tune.add_argument("--gateway-url", required=True, help="gateway base URL")
+    tune.add_argument(
+        "--series-selector",
+        default="",
+        help="label matchers added to every series selector, for example 'namespace=\"sie\"'",
+    )
+    tune.add_argument("--apply", action="store_true", help="write the floor-carrier pools")
+    tune.add_argument("--once", action="store_true", help="run one tick and exit")
+    tune.add_argument(
+        "--interval-s", type=_positive_float, default=DEFAULT_INTERVAL_S, help="seconds between ticks (default 60)"
+    )
+    tune.add_argument(
+        "--max-pool-floor",
+        type=_integer_at_least(0),
+        default=keda.DEFAULT_MAX_POOL_FLOOR,
+        help="the gateway's SIE_GATEWAY_POOL_MAX_MINIMUM_WORKER_COUNT (default 4)",
+    )
+    tune.add_argument(
+        "--max-pools",
+        type=_integer_at_least(1),
+        default=keda.DEFAULT_MAX_POOLS,
+        help="the gateway's SIE_GATEWAY_MAX_POOLS (default 64)",
+    )
+    tune.add_argument("--api-key-file", type=Path, help="file holding a gateway client token for pool reads")
+    tune.add_argument(
+        "--admin-token-file",
+        type=Path,
+        help=f"file holding the gateway admin token for pool writes; read only with --apply (default ${ADMIN_TOKEN_ENV})",
+    )
+    return parser
+
+
+def _token(value: str, source: str, label: str) -> str:
+    if _TOKEN.fullmatch(value) is None:
+        raise _UsageError(f"the {label} in {source} must be one run of printable ASCII characters without spaces")
+    return value
+
+
+def _secret(path: Path | None, env_name: str | None, label: str) -> str | None:
+    if path is not None:
+        try:
+            value = path.read_text(encoding="utf-8").strip()
+        except (OSError, UnicodeError) as error:
+            reason = error.strerror if isinstance(error, OSError) else "it is not UTF-8 text"
+            raise _UsageError(f"cannot read the {label} file {str(path)!r}: {reason}") from None
+        if not value:
+            raise _UsageError(f"the {label} file {str(path)!r} is empty")
+        return _token(value, f"the file {str(path)!r}", label)
+    if env_name is None:
+        return None
+    value = os.environ.get(env_name, "").strip()
+    return _token(value, f"${env_name}", label) if value else None
+
+
+def _load_plan(path: Path) -> Plan:
+    try:
+        with path.open(encoding="utf-8") as handle:
+            document = json.load(handle)
+    except OSError as error:
+        raise _UsageError(f"cannot read the plan {str(path)!r}: {error.strerror}") from None
+    except ValueError as error:
+        raise _UsageError(f"the plan {str(path)!r} is not JSON: {error}") from None
+    return Plan.from_dict(document)
+
+
+def _emit(result: keda.Tick) -> None:
+    for row in result.audit_rows():
+        print(json.dumps(row), flush=True)
+    for error in result.series_errors:
+        print(f"sie-placement tune: Prometheus series {error}", file=sys.stderr, flush=True)
+
+
+def _tune(args: argparse.Namespace) -> int:
+    try:
+        carriers = keda.carriers(_load_plan(args.plan), max_pool_floor=args.max_pool_floor, max_pools=args.max_pools)
+        keda.series_queries(args.series_selector)
+        prometheus = keda.PrometheusHttp(args.prometheus_url)
+        keda.require_base_url("gateway", args.gateway_url)
+        read_token = _secret(args.api_key_file, None, "gateway client token")
+        if read_token is None:
+            _secret(None, SDK_API_KEY_ENV, "gateway client token")
+        admin_token = _secret(args.admin_token_file, ADMIN_TOKEN_ENV, "gateway admin token") if args.apply else None
+    except ValueError as error:
+        print(f"sie-placement tune: {error}", file=sys.stderr)
+        return EXIT_REFUSED
+    try:
+        from sie_placement import pools  # noqa: PLC0415
+    except ImportError:
+        print("sie-placement tune needs the sie-placement keda extra, which installs the SIE SDK", file=sys.stderr)
+        return EXIT_REFUSED
+    if args.apply and admin_token is None:
+        print(
+            f"sie-placement tune: no gateway admin token in --admin-token-file or ${ADMIN_TOKEN_ENV}; "
+            "pool writes carry no credential",
+            file=sys.stderr,
+        )
+    try:
+        reader = pools.connect(args.gateway_url, read_token)
+        writer = pools.connect(args.gateway_url, admin_token or "") if args.apply else None
+    except ValueError as error:
+        print(f"sie-placement tune: {error}", file=sys.stderr)
+        return EXIT_REFUSED
+    try:
+        while True:
+            started = time.monotonic()
+            result = keda.tick(
+                carriers, prometheus=prometheus, reader=reader, writer=writer, selector=args.series_selector
+            )
+            _emit(result)
+            if args.once:
+                return EXIT_FAILED if result.failed else 0
+            time.sleep(max(0.0, args.interval_s - (time.monotonic() - started)))
+    except KeyboardInterrupt:
+        return 0
+    finally:
+        reader.close()
+        if writer is not None:
+            writer.close()
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    return _tune(args)