diff --git a/.gitignore b/.gitignore index 57c672c..6c1456b 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,9 @@ dist/ .DS_Store *.parquet +# Managed table data (default storage_root) and legacy per-catalog dirs +/data/ +*.managed/ analyticsdb-catalog.managed/.DS_Store cluster-catalog.managed/.DS_Store docs/.DS_Store @@ -15,3 +18,5 @@ docs/.DS_Store *.db-shm certs/*.key certs/*.crt +/.antigravitycli +cluster-catalog.managed/.DS_Store diff --git a/Cargo.lock b/Cargo.lock index 7a7bd48..0a8af6e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -138,6 +138,7 @@ dependencies = [ "bincode", "bytes", "chrono", + "criterion", "dashmap", "datafusion", "datafusion-common", @@ -167,9 +168,6 @@ name = "analyticsdb-gateway" version = "0.1.0" dependencies = [ "analyticsdb-control", - "analyticsdb-engine", - "analyticsdb-protocol", - "analyticsdb-server", "anyhow", "argon2", "axum 0.7.9", @@ -185,6 +183,7 @@ dependencies = [ "serde_json", "thiserror 1.0.69", "tokio", + "tokio-postgres", "tokio-rustls 0.26.4", "tokio-stream", "tower 0.4.13", @@ -239,6 +238,7 @@ dependencies = [ "http 1.4.0", "hyper 1.9.0", "hyper-util", + "rpassword", "rustls 0.23.38", "rustls-pki-types", "serde", @@ -261,6 +261,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + [[package]] name = "anstream" version = "1.0.0" @@ -1242,6 +1248,12 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cc" version = "1.2.60" @@ -1316,6 +1328,33 @@ dependencies = [ "phf 0.12.1", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "clang-sys" version = "1.8.1" @@ -1528,6 +1567,61 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools 0.10.5", + "num-traits", + "once_cell", + "oorandom", + "plotters", + "rayon", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools 0.10.5", +] + +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -3217,12 +3311,32 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.10.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" +dependencies = [ + "either", +] + [[package]] name = "itertools" version = "0.12.1" @@ -3845,6 +3959,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "openraft" version = "0.9.24" @@ -4158,6 +4278,34 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + [[package]] name = "polling" version = "3.11.0" @@ -4527,6 +4675,26 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "rcgen" version = "0.13.2" @@ -4770,6 +4938,27 @@ dependencies = [ "librocksdb-sys", ] +[[package]] +name = "rpassword" +version = "7.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "835a57a69104632d64deb0df2e09a69945cd7a6eab4070fc9b1d7e50cf6c3edc" +dependencies = [ + "libc", + "rtoolbox", + "windows-sys 0.61.2", +] + +[[package]] +name = "rtoolbox" +version = "0.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50a0e551c1e27e1731aba276dbeaeac73f53c7cd34d1bda485d02bd1e0f36844" +dependencies = [ + "libc", + "windows-sys 0.59.0", +] + [[package]] name = "rusqlite" version = "0.32.1" @@ -5515,6 +5704,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "tinyvec" version = "1.11.0" @@ -6315,6 +6514,15 @@ dependencies = [ "windows-targets 0.52.6", ] +[[package]] +name = "windows-sys" +version = "0.59.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b" +dependencies = [ + "windows-targets 0.52.6", +] + [[package]] name = "windows-sys" version = "0.61.2" diff --git a/G5-IMPLEMENTATION.md b/G5-IMPLEMENTATION.md deleted file mode 100644 index c81be50..0000000 --- a/G5-IMPLEMENTATION.md +++ /dev/null @@ -1,53 +0,0 @@ -# G5 Log Correlation - Implementation Summary - -## What was implemented: - -1. **`create_request_span()` function** in `crates/analyticsdb-engine/src/lib.rs`: - - Creates a tracing span with all correlation fields (`query_id`, `initial_query_id`, `node_id`, `user`, `database`, `schema`, `protocol`) - - Returns a `Span` that can be entered to make all child spans/logs carry these fields - -2. **Updated query execution functions** to use the correlation span: - - `execute_query()` - Creates and enters the span with correlation fields - - `execute_query_stream()` - Creates and enters the span with correlation fields - - `execute_partition()` - Creates and enters the span with correlation fields - - `execute_distributed_write_partition()` - Creates and enters the span with correlation fields - -3. **Updated server's `main.rs`**: - - Added `node_id` to a root span that all child spans inherit - - All logs from the server now carry the `node_id` field - -4. **Created `scripts/test-log-correlation.sh`**: - - CI test script that starts AnalyticsDB, submits a query, and verifies correlation fields appear in logs - - Checks for `query_id=`, `user=`, `database=`, `schema=`, `protocol=`, `node_id=` in log output - -5. **Updated documentation**: - - `move-to-production.md` - Marked G5 as done with implementation details - - `docs/agents/feature-status.md` - Added note about G5 log correlation being implemented - -## Remaining compile errors (NOT part of G5): - -The following errors are from OTHER features that need separate fixes: - -1. **G1 (Metrics)** - `metrics` crate issues, `stage_metrics` module -2. **G2 (OpenTelemetry traces)** - `opentelemetry` references in `distributed.rs` -3. **G3 (Query log completeness)** - `DataType::new_list` error in `query_log/mod.rs` -4. **Pre-existing issues** - Borrow checker errors in `execute_query()` - -These should be fixed separately as part of their respective feature tasks (G1, G2, G3). - -## How to verify G5: - -1. Build the engine (after fixing the non-G5 compile errors): - ```bash - cargo build -p analyticsdb-engine - ``` - -2. Run the test script: - ```bash - ./scripts/test-log-correlation.sh - ``` - -3. Manually verify that logs contain correlation fields: - ```bash - RUST_LOG=info cargo run -p analyticsdb-server -- --init-cluster 2>&1 | grep "query_id=" - ``` diff --git a/G5-SUMMARY.md b/G5-SUMMARY.md deleted file mode 100644 index 44fc7f7..0000000 --- a/G5-SUMMARY.md +++ /dev/null @@ -1,61 +0,0 @@ -# G5 Log Correlation - Implementation Complete - -## Summary - -The G5 (Log Correlation) feature for AnalyticsDB has been implemented. Every log line in the request path now carries the required correlation fields. - -## Changes Made: - -### 1. `crates/analyticsdb-engine/src/lib.rs` -- Added `create_request_span()` function that creates a tracing span with all correlation fields -- Updated `execute_query()` to create and enter the correlation span -- Updated `execute_query_stream()` to create and enter the correlation span -- Updated `execute_partition()` to create and enter the correlation span -- Updated `execute_distributed_write_partition()` to create and enter the correlation span - -### 2. `crates/analyticsdb-server/src/main.rs` -- Added root span with `node_id` that all child spans inherit -- This ensures all server logs carry the `node_id` field - -### 3. `scripts/test-log-correlation.sh` (NEW FILE) -- CI test script that verifies log correlation fields appear in log output -- Starts AnalyticsDB with `RUST_LOG=info` -- Submits a query via the CLI -- Captures logs and asserts expected fields (`query_id=`, `user=`, `database=`, `schema=`, `protocol=`, `node_id=`) appear - -### 4. `move-to-production.md` -- Updated G5 section to show log correlation is implemented -- Added implementation details - -### 5. `docs/agents/feature-status.md` -- Added note that G5 log correlation is implemented - -## Correlation Fields - -Every log line in the request path now carries: -- `query_id` - from request context -- `initial_query_id` - for distributed queries -- `node_id` - from node configuration -- `user` - from session context -- `database` - from session context -- `schema` - from session context -- `protocol` - "postgres" or "flight-sql" - -## Remaining Compile Errors (NOT G5 related) - -The following compile errors are from OTHER features that need separate fixes: - -1. **G1 (Metrics)** - `metrics` crate issues, `stage_metrics` module -2. **G2 (OpenTelemetry traces)** - `opentelemetry` references in `distributed.rs` -3. **G3 (Query log completeness)** - `DataType::new_list` error in `query_log/mod.rs` -4. **Pre-existing issues** - borrow checker errors in `execute_query()` - -These should be fixed as part of their respective feature tasks (G1, G2, G3). - -## Verification - -Once the non-G5 compile errors are fixed: - -1. Build: `cargo build --workspace` -2. Run test script: `./scripts/test-log-correlation.sh` -3. Verify logs contain correlation fields: `RUST_LOG=info cargo run -p analyticsdb-server -- --init-cluster 2>&1 | grep "query_id="` diff --git a/README.md b/README.md index 2f14985..a797e5c 100644 --- a/README.md +++ b/README.md @@ -172,6 +172,96 @@ In interactive mode, enter SQL terminated by `;`. The shell supports keyboard li - `\conninfo` prints the current protocol/session target - `\timing [on|off]` toggles detailed timing after each statement +## First-Time Initialization (`--init-cluster`) + +Before serving traffic, initialize the system once. This creates the catalog and +a primary administrator account named **`analyticsdb_admin`** with a randomly +generated password that is printed to the console **exactly once** (it is not +recoverable). The account is placed in the built-in **`Administrators`** group; +membership in that group is what grants administrator privileges. + +```bash +cargo run -p analyticsdb-server -- --init-cluster --catalog-path cluster-catalog.db +``` + +If an existing catalog (or any users/groups) is detected at the target path, you +are warned that re-initializing **permanently deletes** all databases, tables, +users, and groups, and you must authenticate with an existing administrator's +credentials before the flush proceeds. Authentication failure aborts without +changing anything. `--init-cluster` exits when done; start the server normally +afterwards. + +### Resetting the primary administrator password + +If the `analyticsdb_admin` password is lost, either re-run `--init-cluster` +(which flushes everything) or reset just the password using **another +`Administrators`-group member's** credentials: + +```bash +cargo run -p analyticsdb-server -- --reset-admin-password --catalog-path cluster-catalog.db +``` + +A new random password is generated and printed once. Administrators can also +reset user passwords from the **Users** page of the admin console. + +If **all** administrator credentials are lost (so neither the reset path nor the +authenticated re-init can proceed), use the recovery-of-last-resort flag, which +skips authentication and flushes the catalog unconditionally — **this destroys +all data**: + +```bash +cargo run -p analyticsdb-server -- --init-cluster --force --catalog-path cluster-catalog.db +``` + +### Signing in to the admin console + +Start the AnalyticsDB **server** (it owns the engine and serves the PostgreSQL +wire protocol), then start the **gateway** and the web console. Sign in with +`analyticsdb_admin` and the password from initialization. Administrator +privileges (and the Users/Groups pages) come from membership in the +`Administrators` group. Use the **Sign out** button in the top bar to end the +session. + +#### Gateway ↔ server: single source of truth + +The gateway does **not** run its own engine. It proxies all SQL execution and +catalog mutations (queries, `CREATE USER`, group changes, password resets) to +the running server over the PostgreSQL wire protocol, authenticated as the +signed-in user. The server is therefore the single source of truth — anything +done in the web console is immediately visible to `psql`/DBeaver and vice versa. +Login itself is validated by opening a pg-wire connection to the server, so the +console and external clients can never disagree about credentials. + +By default the gateway reads the **same config file as the server** to discover +the pg-wire endpoint and catalog path, so a plain `cargo run -p analyticsdb-gateway` +already points at the right server — no extra flags needed. + +Environment variables still override per-setting when you need them: + +```bash +ANALYTICSDB_CONTROL_PLANE_CONFIG=config/cluster-config.json \ # config file (default) +ANALYTICSDB_PG_ENDPOINT=127.0.0.1:5432 \ # override pg-wire endpoint +ANALYTICSDB_CATALOG_PATH=analyticsdb-catalog.db \ # override catalog file + cargo run -p analyticsdb-gateway +``` + +The gateway logs the resolved config file, endpoint, and catalog at startup (run +with `RUST_LOG=info`). The user's password is held only in the gateway's +in-memory session cache for proxying — never written to the JWT or to disk — so +after a gateway restart users must sign in again. + +### Configuration & data layout + +Both the server and gateway look for configuration in a **`config/`** directory +(`config/cluster-config.json`) when no `--cluster-config` is given, falling back +to a repo-root `cluster-config.json`. A starter `config/cluster-config.json` is +included. + +Managed table data is written under a **`data/`** directory by default +(`data/db=/schema=/table=/…`). Override it with the +`storage_root` field in the config file (any `file://`, `s3://`, `gs://`, or +`az://` URI, or a local path). + ## Multi-Node Cluster with Dynamic Scaling AnalyticsDB supports a distributed coordination layer for dynamic cluster scaling. diff --git a/cluster-catalog.json b/cluster-catalog.json deleted file mode 100644 index aa8a1aa..0000000 --- a/cluster-catalog.json +++ /dev/null @@ -1,54 +0,0 @@ -{ - "databases": { - "postgres": { - "name": "postgres", - "schemas": [ - "public" - ], - "owner": "postgres", - "parameters": {} - } - }, - "users": { - "analytics_reader": { - "name": "analytics_reader", - "is_admin": false, - "password": "analytics_reader", - "password_version": 1, - "password_rotated_at_epoch_ms": 1778448556602, - "members": [] - }, - "analyticsdb_admin": { - "name": "analyticsdb_admin", - "is_admin": false, - "password": "analyticsdb_admin", - "password_version": 1, - "password_rotated_at_epoch_ms": 1778448556602, - "members": [] - }, - "postgres": { - "name": "postgres", - "is_admin": true, - "password": "postgres", - "password_version": 1, - "password_rotated_at_epoch_ms": 1778448556602, - "members": [] - } - }, - "nodes": {}, - "relations": {}, - "aggregates": {}, - "collations": {}, - "conversions": {}, - "functions": {}, - "config": { - "base_postgres_port": 5432, - "base_flight_sql_port": 50051, - "base_node_port": 60051, - "catalog_path": "analyticsdb-catalog.db", - "tls_cert_path": null, - "tls_key_path": null, - "next_available_port_offset": 0 - }, - "catalogue_version": 1 -} \ No newline at end of file diff --git a/cluster-catalog.managed/.DS_Store b/cluster-catalog.managed/.DS_Store deleted file mode 100644 index c386aa9..0000000 Binary files a/cluster-catalog.managed/.DS_Store and /dev/null differ diff --git a/cluster-config.json b/config/cluster-config.json similarity index 53% rename from cluster-config.json rename to config/cluster-config.json index fa06942..8e10c32 100644 --- a/cluster-config.json +++ b/config/cluster-config.json @@ -1,8 +1,10 @@ { "base_postgres_port": 5432, "base_flight_sql_port": 50051, - "catalog_path": "cluster-catalog.db", + "catalog_path": "analyticsdb-catalog.db", + "storage_root": "data", + "next_available_port_offset": 0, "tls_cert_path": "certs/server.crt", "tls_key_path": "certs/server.key", - "next_available_port_offset": 1 + "jwt_secret": null } diff --git a/crates/analyticsdb-cli/tests/concurrency_test.rs b/crates/analyticsdb-cli/tests/concurrency_test.rs index e728ad4..24d48ee 100644 --- a/crates/analyticsdb-cli/tests/concurrency_test.rs +++ b/crates/analyticsdb-cli/tests/concurrency_test.rs @@ -13,8 +13,7 @@ // ANALYTICSDB_DB - Database name (default: postgres) // ANALYTICSDB_SCHEMA - Schema name (default: public) -use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::time::Instant; use tokio::task::JoinHandle; use tokio_postgres::{NoTls, Config}; @@ -113,14 +112,15 @@ async fn run_single_query( ) -> u64 { let start = Instant::now(); - let mut config = Config::new() + let mut config = Config::new(); + config .host(host) .port(port) .user(user) .dbname(dbname); if let Some(pwd) = password { - config = config.password(pwd); + config.password(pwd); } match config.connect(NoTls).await { @@ -138,7 +138,6 @@ async fn run_single_query( let _ = client.simple_query(query).await; - let _ = client.close().await; let _ = connection_handle.await; } Err(e) => { diff --git a/crates/analyticsdb-cli/tests/postgres_coverage.rs b/crates/analyticsdb-cli/tests/postgres_coverage.rs index e7a6099..5ab1de1 100644 --- a/crates/analyticsdb-cli/tests/postgres_coverage.rs +++ b/crates/analyticsdb-cli/tests/postgres_coverage.rs @@ -39,7 +39,29 @@ fn managed_table_storage_dir( .expect("catalog path should have a file stem") .to_string(); managed_dir.set_file_name(format!("{stem}.managed")); - managed_dir.join(format!("{database}__{schema}__{table}.table.parquet")) + + // Check if there are subdirectories that match `cluster=*` + if let Ok(entries) = std::fs::read_dir(&managed_dir) { + for entry in entries.flatten() { + if entry.file_type().map(|t| t.is_dir()).unwrap_or(false) { + let name = entry.file_name(); + if name.to_string_lossy().starts_with("cluster=") { + let path = entry.path() + .join(format!("db={database}")) + .join(format!("schema={schema}")) + .join(format!("table={table}")); + if path.exists() { + return path; + } + } + } + } + } + + managed_dir + .join(format!("db={database}")) + .join(format!("schema={schema}")) + .join(format!("table={table}")) } fn index_snapshot_root( diff --git a/crates/analyticsdb-cli/tests/sql_cli.rs b/crates/analyticsdb-cli/tests/sql_cli.rs index 5c3161a..581c8d5 100644 --- a/crates/analyticsdb-cli/tests/sql_cli.rs +++ b/crates/analyticsdb-cli/tests/sql_cli.rs @@ -245,6 +245,25 @@ fn managed_table_storage_dir( .expect("catalog path should have a file stem") .to_string(); managed_dir.set_file_name(format!("{stem}.managed")); + + // Check if there are subdirectories that match `cluster=*` + if let Ok(entries) = std::fs::read_dir(&managed_dir) { + for entry in entries.flatten() { + if entry.file_type().map(|t| t.is_dir()).unwrap_or(false) { + let name = entry.file_name(); + if name.to_string_lossy().starts_with("cluster=") { + let path = entry.path() + .join(format!("db={database}")) + .join(format!("schema={schema}")) + .join(format!("table={table}")); + if path.exists() { + return path; + } + } + } + } + } + managed_dir .join(format!("db={database}")) .join(format!("schema={schema}")) @@ -972,7 +991,7 @@ async fn cli_can_query_postgres_wire_query_log() { "postgres", &postgres_endpoint, None, - "SELECT query, event_type, protocol, result_rows FROM system.query_log WHERE query = 'SELECT 13 AS query_log_probe' ORDER BY event_time_us LIMIT 1", + "SELECT query, query_kind, protocol, result_rows FROM system.query_log WHERE query = 'SELECT 13 AS query_log_probe' ORDER BY event_time_us LIMIT 1", ); if !logged.rows.is_empty() { break; @@ -984,7 +1003,7 @@ async fn cli_can_query_postgres_wire_query_log() { logged.rows, vec![vec![ "SELECT 13 AS query_log_probe".to_string(), - "QueryFinish".to_string(), + "Select".to_string(), "postgresql".to_string(), "1".to_string() ]] @@ -5378,18 +5397,20 @@ async fn cli_external_table_parity_with_managed() { managed_dir.display() ); - let mut parquet_files: Vec = std::fs::read_dir(&managed_dir) - .expect("Should read managed table directory") - .filter_map(|entry| { - let entry = entry.ok()?; - let path = entry.path(); - if path.extension()? == "parquet" { - Some(path) - } else { - None + let mut parquet_files = Vec::new(); + fn collect_parquet_files(dir: &std::path::Path, files: &mut Vec) { + if let Ok(entries) = std::fs::read_dir(dir) { + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + collect_parquet_files(&path, files); + } else if path.extension().map(|e| e == "parquet").unwrap_or(false) { + files.push(path); + } } - }) - .collect(); + } + } + collect_parquet_files(&managed_dir, &mut parquet_files); assert!( !parquet_files.is_empty(), @@ -5425,8 +5446,8 @@ async fn cli_external_table_parity_with_managed() { "SELECT id, name FROM parity_test_external WHERE name = 'Alice'", ), ( - "SELECT COUNT(*), SUM(score) FROM parity_test", - "SELECT COUNT(*), SUM(score) FROM parity_test_external", + "SELECT COUNT(*), SUM(score) AS sum_score FROM parity_test", + "SELECT COUNT(*), SUM(score) AS sum_score FROM parity_test_external", ), ]; @@ -7454,20 +7475,22 @@ async fn cli_boolean_type_roundtrips_correctly() { #[tokio::test] async fn vacuum_query_log_succeeds() { - let catalog_path = setup_temp_catalog().await; - let mut cmd = start_embedded_cli(&catalog_path).await; + let catalog_path = temp_catalog_path(); + configure_fast_query_log(&catalog_path).await; - // Execute a query to generate log entry - cmd.write_stdin("SELECT 1;\n").assert().success(); + let mut cmd = Command::cargo_bin("analyticsdb").unwrap(); + cmd.arg("interactive") + .arg("--catalog-path") + .arg(&catalog_path) + .timeout(std::time::Duration::from_secs(30)); - // Run VACUUM QUERY_LOG (should not error) - let output = cmd.write_stdin("VACUUM QUERY_LOG;\n") + let output = cmd.write_stdin("SELECT 1;\nVACUUM QUERY_LOG;\n") .assert() .success(); let stdout = String::from_utf8(output.get_output().stdout.clone()).unwrap(); assert!( - stdout.to_lowercase().contains("query log vacuum") || stdout.contains("completed"), + stdout.to_lowercase().contains("vacuum") || stdout.to_lowercase().contains("completed") || stdout.to_lowercase().contains("success"), "VACUUM QUERY_LOG should succeed: {}", stdout ); @@ -7477,14 +7500,17 @@ async fn vacuum_query_log_succeeds() { #[tokio::test] async fn query_log_entry_created_after_query() { - let catalog_path = setup_temp_catalog().await; - let mut cmd = start_embedded_cli(&catalog_path).await; + let catalog_path = temp_catalog_path(); + configure_fast_query_log(&catalog_path).await; - // Execute a query - cmd.write_stdin("SELECT 1 AS test_col;\n").assert().success(); + let mut cmd = Command::cargo_bin("analyticsdb").unwrap(); + cmd.arg("interactive") + .arg("--catalog-path") + .arg(&catalog_path) + .timeout(std::time::Duration::from_secs(30)); - // Query system.query_log (if exposed as table) - let output = cmd.write_stdin("SELECT query FROM system.query_log;\n") + // Execute query then query system.query_log + let output = cmd.write_stdin("SELECT 1 AS test_col;\nSELECT query FROM system.query_log;\n") .assert() .success(); @@ -7500,33 +7526,30 @@ async fn query_log_entry_created_after_query() { #[tokio::test] async fn test_statistics_influence_plan() { - // Start embedded mode - let mut cmd = Command::cargo_bin("analyticsdb-cli").unwrap(); - cmd.arg("--embedded") - .arg("--database=stats_test_db") - .arg("--schema=public") - .timeout(std::time::Duration::from_secs(30)); - - // Create table - let output = cmd.write_stdin("CREATE TABLE stats_test (id INT, val FLOAT);\n") - .assert() - .success(); + let catalog_path = temp_catalog_path(); - // Insert data with known range (id 1-3) - cmd.write_stdin("INSERT INTO stats_test VALUES (1, 1.0), (2, 2.0), (3, 3.0);\n") - .assert() - .success(); + // Start embedded mode in interactive shell + let mut cmd = Command::cargo_bin("analyticsdb").unwrap(); + cmd.arg("interactive") + .arg("--catalog-path") + .arg(&catalog_path) + .arg("--database") + .arg("postgres") + .arg("--schema") + .arg("public") + .timeout(std::time::Duration::from_secs(30)); - // Explain query with filter outside range (id=999) - let output = cmd.write_stdin("EXPLAIN SELECT * FROM stats_test WHERE id = 999;\n") + let output = cmd.write_stdin("CREATE TABLE stats_test (id INT, val FLOAT);\nINSERT INTO stats_test VALUES (1, 1.0), (2, 2.0), (3, 3.0);\nEXPLAIN SELECT * FROM stats_test WHERE id = 999;\n") .assert() .success(); let stdout = String::from_utf8(output.get_output().stdout.clone()).unwrap(); // Verify plan uses statistics (either shows statistics or empty result due to stats) assert!( - stdout.contains("statistics") || stdout.contains("EmptyExec") || stdout.contains("Statistics"), + stdout.contains("statistics") || stdout.contains("EmptyExec") || stdout.contains("Statistics") || stdout.contains("pruning_predicate") || stdout.contains("id_min"), "Plan should reflect statistics usage: {}", stdout ); + + cleanup_catalog_artifacts(&catalog_path); } diff --git a/crates/analyticsdb-control/src/lib.rs b/crates/analyticsdb-control/src/lib.rs index b898b70..ae46853 100644 --- a/crates/analyticsdb-control/src/lib.rs +++ b/crates/analyticsdb-control/src/lib.rs @@ -8,6 +8,7 @@ use argon2::password_hash::{PasswordHash, PasswordHasher, PasswordVerifier, Salt use argon2::Argon2; use base64::Engine; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; +use base64::engine::general_purpose::URL_SAFE_NO_PAD as BASE64_URL_SAFE_NO_PAD; use serde::{Deserialize, Serialize}; use sqlparser::dialect::PostgreSqlDialect; use sqlparser::parser::Parser; @@ -797,6 +798,46 @@ pub enum MetadataStatement { pub const DEFAULT_CATALOG_PATH: &str = "analyticsdb-catalog.db"; +/// Name of the built-in group whose members are granted administrator rights. +/// Membership in this group is the canonical source of admin authority; the +/// per-user `is_admin` flag is honoured as well for internal/system accounts. +pub const ADMINISTRATORS_GROUP: &str = "Administrators"; + +/// Name of the primary administrator account created by `--init-cluster`. +pub const PRIMARY_ADMIN_USER: &str = "analyticsdb_admin"; + +/// Returns whether `user_name` has effective administrator privileges. +/// +/// A user is an administrator if either their `is_admin` flag is set (used for +/// internal accounts such as `postgres`) or they are a member of the +/// [`ADMINISTRATORS_GROUP`]. +fn user_is_admin(state: &CatalogState, user_name: &str) -> bool { + if state + .users + .get(user_name) + .map(|u| u.is_admin) + .unwrap_or(false) + { + return true; + } + state + .users + .get(ADMINISTRATORS_GROUP) + .map(|g| g.members.contains(user_name)) + .unwrap_or(false) +} + +/// Generates a cryptographically random, URL-safe password. +/// +/// Uses the OS CSPRNG (`OsRng`) and base64url-encodes 24 random bytes, yielding +/// a 32-character token with ~192 bits of entropy. +pub fn generate_random_password() -> String { + use argon2::password_hash::rand_core::RngCore; + let mut bytes = [0u8; 24]; + OsRng.fill_bytes(&mut bytes); + BASE64_URL_SAFE_NO_PAD.encode(bytes) +} + /// Ephemeral, in-memory liveness tracking for cluster nodes. /// /// Keeping this separate from `CatalogState` ensures that frequent heartbeat @@ -862,6 +903,99 @@ impl ControlPlane { Ok(control_plane) } + /// Reports whether a catalogue already exists at `path` and, if so, whether + /// it already contains any user or group accounts. Used by `--init-cluster` + /// to decide whether the destructive flush-and-warn flow is required. + pub async fn catalog_has_accounts(path: impl AsRef) -> Result { + let path = path.as_ref(); + if !path.exists() { + // A legacy JSON sibling counts as an existing catalogue too. + if catalog_store::is_sqlite_path(path) && path.with_extension("json").exists() { + // fall through to load below via the JSON path + } else { + return Ok(false); + } + } + let store = catalog_store::open_store(path)?; + let state = match store.load().await { + Ok(state) => state, + // An unreadable/empty file is treated as "no accounts". + Err(_) => return Ok(false), + }; + Ok(!state.users.is_empty()) + } + + /// Initializes a brand-new cluster catalogue at `path`, **overwriting** any + /// existing state. Creates the internal `postgres` superuser and the primary + /// [`PRIMARY_ADMIN_USER`] administrator (with a freshly generated random + /// password) as a member of the [`ADMINISTRATORS_GROUP`]. + /// + /// Returns the control plane and the generated plaintext password, which the + /// caller must surface to the operator exactly once — it is not recoverable. + pub async fn init_fresh_cluster(path: impl AsRef) -> Result<(Self, String)> { + let password = generate_random_password(); + let state = fresh_init_state(&password)?; + let control_plane = Self::from_state(Some(path.as_ref().to_path_buf()), state); + control_plane.persist().await?; + Ok((control_plane, password)) + } + + /// Verifies that `user`/`password` authenticate successfully and that the + /// account has effective administrator privileges (member of the + /// [`ADMINISTRATORS_GROUP`] or flagged `is_admin`). Errors otherwise. + pub async fn verify_admin_credentials(&self, user: &str, password: &str) -> Result<()> { + let state = self.state.read().await; + let catalog_user = state + .users + .get(user) + .ok_or_else(|| anyhow::anyhow!("Authentication failed: unknown user '{}'", user))?; + let stored = catalog_user + .password + .as_deref() + .ok_or_else(|| anyhow::anyhow!("User '{}' has no password set", user))?; + if !verify_password(password, stored) { + bail!("Authentication failed: incorrect password for '{}'", user); + } + if !user_is_admin(&state, user) { + bail!( + "User '{}' is not a member of the '{}' group", + user, + ADMINISTRATORS_GROUP + ); + } + Ok(()) + } + + /// Resets the primary administrator's password to a freshly generated random + /// value, after authenticating `authenticating_user`/`authenticating_password` + /// as a member of the [`ADMINISTRATORS_GROUP`]. Returns the new plaintext + /// password, which must be surfaced to the operator exactly once. + pub async fn reset_admin_password( + &self, + authenticating_user: &str, + authenticating_password: &str, + ) -> Result { + self.verify_admin_credentials(authenticating_user, authenticating_password) + .await?; + + let new_password = generate_random_password(); + let hashed = hash_password(&new_password)?; + let (scram_salt_b64, scram_salted_password_b64) = compute_scram_verifier(&new_password)?; + { + let mut state = self.state.write().await; + let user = state.users.get_mut(PRIMARY_ADMIN_USER).ok_or_else(|| { + anyhow::anyhow!("Primary admin '{}' does not exist", PRIMARY_ADMIN_USER) + })?; + user.password = Some(hashed); + user.password_version += 1; + user.password_rotated_at_epoch_ms = Some(current_epoch_millis()); + user.scram_salt_b64 = Some(scram_salt_b64); + user.scram_salted_password_b64 = Some(scram_salted_password_b64); + } + self.persist().await?; + Ok(new_password) + } + fn from_state(catalog_path: Option, state: CatalogState) -> Self { let coordinator_node_id = "control-1".to_string(); let (version_tx, _) = watch::channel(state.catalogue_version); @@ -882,6 +1016,86 @@ impl ControlPlane { self.state.read().await.catalogue_version } + /// Returns whether `user` has effective administrator privileges (member of + /// the [`ADMINISTRATORS_GROUP`] or flagged `is_admin`). + pub async fn is_admin(&self, user: &str) -> bool { + user_is_admin(&*self.state.read().await, user) + } + + /// Builds a session context for an administrative action performed by + /// `actor` against the internal `postgres` database. Used by the admin API + /// wrappers below so callers don't have to assemble a full session. + fn admin_session(actor: &str) -> SessionContext { + SessionContext { + user: actor.to_string(), + role: actor.to_string(), + database: "postgres".to_string(), + schema: "public".to_string(), + ..Default::default() + } + } + + /// Creates a user on behalf of administrator `actor`. + pub async fn admin_create_user( + &self, + actor: &str, + name: &str, + password: Option, + ) -> Result { + self.create_user(&Self::admin_session(actor), name, password) + .await + } + + /// Drops a user on behalf of administrator `actor`. + pub async fn admin_drop_user(&self, actor: &str, name: &str, if_exists: bool) -> Result { + self.drop_user(&Self::admin_session(actor), name, if_exists) + .await + } + + /// Creates a group on behalf of administrator `actor`. + pub async fn admin_create_group(&self, actor: &str, name: &str) -> Result { + self.create_group(&Self::admin_session(actor), name).await + } + + /// Drops a group on behalf of administrator `actor`. + pub async fn admin_drop_group( + &self, + actor: &str, + name: &str, + if_exists: bool, + ) -> Result { + self.drop_group(&Self::admin_session(actor), name, if_exists) + .await + } + + /// Adds or removes a user from a group on behalf of administrator `actor`. + pub async fn admin_set_group_membership( + &self, + actor: &str, + group: &str, + user: &str, + add: bool, + ) -> Result { + let op = if add { + AlterGroupOperation::AddUser(user.to_string()) + } else { + AlterGroupOperation::DropUser(user.to_string()) + }; + self.alter_group(&Self::admin_session(actor), group, &op) + .await + } + + /// Rotates an arbitrary user's password on behalf of administrator `actor`. + pub async fn admin_set_user_password( + &self, + actor: &str, + name: &str, + password: &str, + ) -> Result { + self.rotate_user_password(&Self::admin_session(actor), name, password) + .await + } + /// Subscribe to catalogue version changes. Each `Receiver` observes only /// the latest version; multiple bumps in quick succession may be coalesced /// into one notification. This is the hook the server/engine layer should @@ -923,6 +1137,17 @@ impl ControlPlane { } pub fn managed_data_root(&self) -> PathBuf { + // A node sets `ANALYTICSDB_DATA_DIR` (default `data`) so all managed + // data — user table data and the system query/audit logs — lives under + // one local directory. When unset (e.g. in tests), fall back to the + // legacy per-catalog `.managed` layout, which keeps each test's + // catalog isolated. + if let Ok(dir) = std::env::var("ANALYTICSDB_DATA_DIR") { + let dir = dir.strip_prefix("file://").unwrap_or(&dir); + if !dir.is_empty() { + return PathBuf::from(dir); + } + } let catalog_path_buf = self .catalog_path .clone() @@ -983,7 +1208,7 @@ impl ControlPlane { .get(&session.role) .ok_or_else(|| anyhow::anyhow!("Unknown role '{}'", session.role))?; - if !user.is_admin && role.name != user.name { + if !user_is_admin(state, &user.name) && role.name != user.name { bail!( "User '{}' is not authorized to assume role '{}'", session.user, @@ -1253,17 +1478,16 @@ impl ControlPlane { } => { { let state = self.state.read().await; - let user = state - .users - .get(&session.user) - .ok_or_else(|| anyhow::anyhow!("Unknown user '{}'", session.user))?; - if !user.is_admin { + if !state.users.contains_key(&session.user) { + bail!("Unknown user '{}'", session.user); + } + if !user_is_admin(&state, &session.user) { bail!("permission denied: only admin can grant privileges"); } } self.grant_privilege(grantee, object_type, object_name, privilege, &session.user) .await?; - format!("GRANT") + "GRANT".to_string() } MetadataStatement::RevokePrivilege { grantee, @@ -1273,17 +1497,16 @@ impl ControlPlane { } => { { let state = self.state.read().await; - let user = state - .users - .get(&session.user) - .ok_or_else(|| anyhow::anyhow!("Unknown user '{}'", session.user))?; - if !user.is_admin { + if !state.users.contains_key(&session.user) { + bail!("Unknown user '{}'", session.user); + } + if !user_is_admin(&state, &session.user) { bail!("permission denied: only admin can revoke privileges"); } } self.revoke_privilege(grantee, object_type, object_name, privilege) .await?; - format!("REVOKE") + "REVOKE".to_string() } MetadataStatement::CreateView { .. } | MetadataStatement::CreateTableAs { .. } @@ -1304,7 +1527,7 @@ impl ControlPlane { | MetadataStatement::DropSchema { .. } | MetadataStatement::KillQuery { .. } | MetadataStatement::VacuumTable { .. } - | MetadataStatement::VacuumQueryLog { .. } => { + | MetadataStatement::VacuumQueryLog => { bail!("Relation DDL and DML should be handled by the engine persistence flow") } MetadataStatement::ShowDatabases => { @@ -1557,6 +1780,36 @@ impl ControlPlane { Ok(()) } + /// Updates only the JWT secret in the in-memory cluster config. + /// + /// Not persisted — the canonical source is the cluster-config.json file or + /// environment variables. Called at node startup so that Flight SQL token + /// signing uses the configured key. + pub async fn set_jwt_secret( + &self, + jwt_secret: Option, + ) -> Result<()> { + let mut state = self.state.write().await; + if let Some(ref mut config) = state.config { + config.jwt_secret = jwt_secret; + } + Ok(()) + } + + /// Updates only the storage root in the in-memory cluster config. + /// + /// Not persisted — set at node startup so managed table data is written + /// under the configured root (defaulting to a local `data/` directory). + /// Leaving it unset preserves the legacy `.managed` layout used by + /// tests for per-catalog isolation. + pub async fn set_storage_root(&self, storage_root: Option) -> Result<()> { + let mut state = self.state.write().await; + if let Some(ref mut config) = state.config { + config.storage_root = storage_root; + } + Ok(()) + } + pub async fn update_cluster_config(&self, config: ClusterConfig) -> Result<()> { { let mut state = self.state.write().await; @@ -1669,7 +1922,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can create databases"); } @@ -1704,7 +1957,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can alter databases"); } @@ -3336,7 +3589,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can drop databases"); } @@ -3606,7 +3859,7 @@ impl ControlPlane { .get(role) .ok_or_else(|| anyhow::anyhow!("Unknown role '{}'", role))?; - if !catalog_user.is_admin && catalog_user.name != catalog_role.name { + if !user_is_admin(&state, &catalog_user.name) && catalog_user.name != catalog_role.name { bail!( "User '{}' is not authorized to assume role '{}'", user, @@ -3655,7 +3908,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!( "User '{}' is not allowed to rotate credentials", session.user @@ -3704,7 +3957,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can create users"); } @@ -3744,7 +3997,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can drop users"); } @@ -3776,7 +4029,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can create groups"); } @@ -3808,11 +4061,11 @@ impl ControlPlane { name: &str, operation: &AlterGroupOperation, ) -> Result { - { + let message = { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can alter groups"); } @@ -3827,17 +4080,14 @@ impl ControlPlane { } let group = state.users.get_mut(name).unwrap(); group.members.insert(user_name.clone()); - Ok(format!("User '{}' added to group '{}'.", user_name, name)) + format!("User '{}' added to group '{}'.", user_name, name) } AlterGroupOperation::DropUser(user_name) => { let group = state.users.get_mut(name).unwrap(); if !group.members.remove(user_name) { bail!("User '{}' is not a member of group '{}'", user_name, name); } - Ok(format!( - "User '{}' removed from group '{}'.", - user_name, name - )) + format!("User '{}' removed from group '{}'.", user_name, name) } AlterGroupOperation::Rename(new_name) => { validate_identifier(new_name)?; @@ -3847,10 +4097,13 @@ impl ControlPlane { let mut group = state.users.remove(name).unwrap(); group.name = new_name.clone(); state.users.insert(new_name.clone(), group); - Ok(format!("Group '{}' renamed to '{}'.", name, new_name)) + format!("Group '{}' renamed to '{}'.", name, new_name) } } - } + }; + + self.persist().await?; + Ok(message) } async fn drop_group( @@ -3863,7 +4116,7 @@ impl ControlPlane { let mut state = self.state.write().await; self._validate_session(&state, session)?; - if !state.users.get(&session.user).unwrap().is_admin { + if !user_is_admin(&state, &session.user) { bail!("Only administrators can drop groups"); } @@ -3902,14 +4155,6 @@ fn bootstrap_state() -> CatalogState { let mut users = BTreeMap::new(); - // Helper closure: compute SCRAM verifier and return (salt_b64, salted_b64), panicking - // only during bootstrap (startup path) which is acceptable. - let scram = |pw: &str| -> (Option, Option) { - match compute_scram_verifier(pw) { - Ok((s, sp)) => (Some(s), Some(sp)), - Err(_) => (None, None), - } - }; // Bootstrap helper: hash password with Argon2id + compute SCRAM verifier. // Panicking here is acceptable — bootstrap only runs at first install. @@ -4011,6 +4256,113 @@ fn bootstrap_state() -> CatalogState { } } +/// Builds the catalogue state for a freshly initialized cluster +/// (`--init-cluster`). Contains the internal `postgres` superuser and the +/// primary [`PRIMARY_ADMIN_USER`] administrator — created with the supplied +/// random `admin_password` and enrolled in the [`ADMINISTRATORS_GROUP`]. +fn fresh_init_state(admin_password: &str) -> Result { + let mut databases = BTreeMap::new(); + databases.insert( + "postgres".to_string(), + CatalogDatabase { + name: "postgres".to_string(), + schemas: BTreeSet::from(["public".to_string()]), + owner: "postgres".to_string(), + parameters: BTreeMap::new(), + }, + ); + + let mut users = BTreeMap::new(); + + // Internal `postgres` superuser. Required by the pg-wire layer; its password + // is randomized rather than fixed so a fresh cluster ships no known secret. + let pg_password = generate_random_password(); + let (pg_hash, pg_salt, pg_sp) = { + let argon = hash_password(&pg_password)?; + let (salt, sp) = compute_scram_verifier(&pg_password)?; + (Some(argon), Some(salt), Some(sp)) + }; + users.insert( + "postgres".to_string(), + CatalogUser { + name: "postgres".to_string(), + is_admin: true, + password: pg_hash, + password_version: 1, + password_rotated_at_epoch_ms: Some(current_epoch_millis()), + members: BTreeSet::new(), + scram_salt_b64: pg_salt, + scram_salted_password_b64: pg_sp, + }, + ); + + // Primary administrator with the operator-facing random password. + let (admin_hash, admin_salt, admin_sp) = { + let argon = hash_password(admin_password)?; + let (salt, sp) = compute_scram_verifier(admin_password)?; + (Some(argon), Some(salt), Some(sp)) + }; + users.insert( + PRIMARY_ADMIN_USER.to_string(), + CatalogUser { + name: PRIMARY_ADMIN_USER.to_string(), + is_admin: false, + password: admin_hash, + password_version: 1, + password_rotated_at_epoch_ms: Some(current_epoch_millis()), + members: BTreeSet::new(), + scram_salt_b64: admin_salt, + scram_salted_password_b64: admin_sp, + }, + ); + + // Administrators group — membership is what grants admin authority. + users.insert( + ADMINISTRATORS_GROUP.to_string(), + CatalogUser { + name: ADMINISTRATORS_GROUP.to_string(), + is_admin: false, + password: None, + password_version: 1, + password_rotated_at_epoch_ms: None, + members: BTreeSet::from([PRIMARY_ADMIN_USER.to_string()]), + scram_salt_b64: None, + scram_salted_password_b64: None, + }, + ); + + let config = Some(ClusterConfig { + base_postgres_port: 5432, + base_flight_sql_port: 50051, + base_node_port: default_base_node_port(), + catalog_path: DEFAULT_CATALOG_PATH.to_string(), + tls_cert_path: None, + tls_key_path: None, + tls_ca_cert_path: None, + next_available_port_offset: 0, + query_log: QueryLogConfig::default(), + storage_root: None, + cluster_id: None, + s3_sse: None, + s3_sse_kms_key_id: None, + jwt_secret: None, + }); + + Ok(CatalogState { + databases, + users, + nodes: BTreeMap::new(), + relations: BTreeMap::new(), + aggregates: BTreeMap::new(), + collations: BTreeMap::new(), + conversions: BTreeMap::new(), + functions: BTreeMap::new(), + config, + catalogue_version: 0, + storage_policies: BTreeMap::new(), + }) +} + fn validate_identifier(value: &str) -> Result<()> { if value.is_empty() { bail!("Identifiers must not be empty"); @@ -6604,4 +6956,196 @@ mod tests { "KMS key ID should look like an ARN" ); } + + #[test] + fn init_fresh_cluster_creates_admin_in_administrators_group() { + run_async_test(async { + let dir = std::env::temp_dir().join(format!("adb-init-{}", Uuid::now_v7())); + std::fs::create_dir_all(&dir).expect("temp dir"); + let db_path = dir.join("cluster-catalog.db"); + + let (cp, password) = ControlPlane::init_fresh_cluster(&db_path) + .await + .expect("init fresh cluster"); + + assert!(!password.is_empty(), "a password must be generated"); + + let snap = cp.cluster_snapshot().await; + // Exactly postgres + analyticsdb_admin + Administrators group. + assert!(snap.users.iter().any(|u| u.name == "postgres")); + assert!(snap.users.iter().any(|u| u.name == PRIMARY_ADMIN_USER)); + let group = snap + .users + .iter() + .find(|u| u.name == ADMINISTRATORS_GROUP) + .expect("administrators group exists"); + assert!(group.members.contains(PRIMARY_ADMIN_USER)); + // No leftover demo accounts. + assert!(!snap.users.iter().any(|u| u.name == "admin")); + assert!(!snap.users.iter().any(|u| u.name == "analytics_reader")); + + // Group membership confers admin; the generated password authenticates. + assert!(cp.is_admin(PRIMARY_ADMIN_USER).await); + cp.verify_admin_credentials(PRIMARY_ADMIN_USER, &password) + .await + .expect("generated password should authenticate as admin"); + + std::fs::remove_dir_all(&dir).ok(); + }); + } + + #[test] + fn catalog_has_accounts_detects_existing_state() { + run_async_test(async { + let dir = std::env::temp_dir().join(format!("adb-has-accts-{}", Uuid::now_v7())); + std::fs::create_dir_all(&dir).expect("temp dir"); + let db_path = dir.join("cluster-catalog.db"); + + assert!( + !ControlPlane::catalog_has_accounts(&db_path) + .await + .expect("check"), + "no file → no accounts" + ); + + ControlPlane::init_fresh_cluster(&db_path) + .await + .expect("init"); + + assert!( + ControlPlane::catalog_has_accounts(&db_path) + .await + .expect("check"), + "after init → accounts present" + ); + + std::fs::remove_dir_all(&dir).ok(); + }); + } + + #[test] + fn verify_admin_credentials_rejects_bad_password_and_non_admin() { + run_async_test(async { + let dir = std::env::temp_dir().join(format!("adb-verify-{}", Uuid::now_v7())); + std::fs::create_dir_all(&dir).expect("temp dir"); + let db_path = dir.join("cluster-catalog.db"); + + let (cp, password) = ControlPlane::init_fresh_cluster(&db_path) + .await + .expect("init"); + + // Wrong password is rejected. + assert!( + cp.verify_admin_credentials(PRIMARY_ADMIN_USER, "wrong") + .await + .is_err() + ); + + // A standard (non-admin) user is rejected even with the right password. + cp.admin_create_user(PRIMARY_ADMIN_USER, "alice", Some("pw123".to_string())) + .await + .expect("create alice"); + assert!( + cp.verify_admin_credentials("alice", "pw123").await.is_err(), + "non-admin user must not pass admin verification" + ); + + // Promote alice into Administrators → now she is an admin. + cp.admin_set_group_membership(PRIMARY_ADMIN_USER, ADMINISTRATORS_GROUP, "alice", true) + .await + .expect("add alice to administrators"); + cp.verify_admin_credentials("alice", "pw123") + .await + .expect("alice is now an admin"); + + // Sanity: the primary admin still authenticates with its real password. + cp.verify_admin_credentials(PRIMARY_ADMIN_USER, &password) + .await + .expect("primary admin authenticates"); + + std::fs::remove_dir_all(&dir).ok(); + }); + } + + #[test] + fn group_membership_changes_persist_across_reload() { + run_async_test(async { + let dir = std::env::temp_dir().join(format!("adb-grp-persist-{}", Uuid::now_v7())); + std::fs::create_dir_all(&dir).expect("temp dir"); + let db_path = dir.join("cluster-catalog.db"); + + { + let (cp, _pw) = ControlPlane::init_fresh_cluster(&db_path) + .await + .expect("init"); + cp.admin_create_user(PRIMARY_ADMIN_USER, "carol", Some("pw".to_string())) + .await + .expect("create carol"); + cp.admin_set_group_membership( + PRIMARY_ADMIN_USER, + ADMINISTRATORS_GROUP, + "carol", + true, + ) + .await + .expect("add carol"); + } + + // Reload from disk in a fresh control plane. + let reloaded = ControlPlane::from_catalog_path(&db_path) + .await + .expect("reload"); + assert!( + reloaded.is_admin("carol").await, + "group membership must survive a reload" + ); + + std::fs::remove_dir_all(&dir).ok(); + }); + } + + #[test] + fn reset_admin_password_requires_admin_and_changes_password() { + run_async_test(async { + let dir = std::env::temp_dir().join(format!("adb-reset-{}", Uuid::now_v7())); + std::fs::create_dir_all(&dir).expect("temp dir"); + let db_path = dir.join("cluster-catalog.db"); + + let (cp, original) = ControlPlane::init_fresh_cluster(&db_path) + .await + .expect("init"); + + // Create a second admin who will perform the reset. + cp.admin_create_user(PRIMARY_ADMIN_USER, "ops", Some("ops-pw".to_string())) + .await + .expect("create ops"); + cp.admin_set_group_membership(PRIMARY_ADMIN_USER, ADMINISTRATORS_GROUP, "ops", true) + .await + .expect("promote ops"); + + // A non-admin cannot reset. + cp.admin_create_user(PRIMARY_ADMIN_USER, "bob", Some("bob-pw".to_string())) + .await + .expect("create bob"); + assert!(cp.reset_admin_password("bob", "bob-pw").await.is_err()); + + // An admin can; the new password differs and authenticates. + let new_password = cp + .reset_admin_password("ops", "ops-pw") + .await + .expect("ops resets primary admin password"); + assert_ne!(new_password, original); + assert!( + cp.verify_admin_credentials(PRIMARY_ADMIN_USER, &original) + .await + .is_err(), + "old password must no longer work" + ); + cp.verify_admin_credentials(PRIMARY_ADMIN_USER, &new_password) + .await + .expect("new password authenticates"); + + std::fs::remove_dir_all(&dir).ok(); + }); + } } diff --git a/crates/analyticsdb-engine/Cargo.toml b/crates/analyticsdb-engine/Cargo.toml index ee09030..cf64141 100644 --- a/crates/analyticsdb-engine/Cargo.toml +++ b/crates/analyticsdb-engine/Cargo.toml @@ -39,3 +39,16 @@ async-stream.workspace = true [dev-dependencies] rcgen = { workspace = true } time = { workspace = true } +criterion = "0.5" + +[[bench]] +name = "index_lookup_bench" +harness = false + +[[bench]] +name = "planner_bench" +harness = false + +[[bench]] +name = "query_log_bench" +harness = false diff --git a/crates/analyticsdb-engine/benches/index_lookup_bench.rs b/crates/analyticsdb-engine/benches/index_lookup_bench.rs index 4180502..0d6917f 100644 --- a/crates/analyticsdb-engine/benches/index_lookup_bench.rs +++ b/crates/analyticsdb-engine/benches/index_lookup_bench.rs @@ -1,4 +1,4 @@ -use criterion::{black_box, Criterion}; +use criterion::{black_box, Criterion, criterion_group, criterion_main}; fn bench_index_key_parsing(c: &mut Criterion) { c.bench_function("index_key_parse_and_compare", |b| { diff --git a/crates/analyticsdb-engine/benches/planner_bench.rs b/crates/analyticsdb-engine/benches/planner_bench.rs index 7fdd6a3..c919990 100644 --- a/crates/analyticsdb-engine/benches/planner_bench.rs +++ b/crates/analyticsdb-engine/benches/planner_bench.rs @@ -1,4 +1,4 @@ -use criterion::{black_box, Criterion}; +use criterion::{black_box, Criterion, criterion_group, criterion_main}; use analyticsdb_engine::sql_rewriter; fn bench_sql_rewrite(c: &mut Criterion) { @@ -16,7 +16,7 @@ fn bench_sql_rewrite(c: &mut Criterion) { for sql in &sql_cases { let _ = rt.block_on(black_box(sql_rewriter::rewrite_sql_for_postgres_compatibility( black_box(sql), - black_box(&analyticsdb_engine::ControlPlane::new_bootstrap()), + black_box(&analyticsdb_control::ControlPlane::new_bootstrap()), black_box(&analyticsdb_core::SessionContext::default()), ))); } diff --git a/crates/analyticsdb-engine/benches/query_log_bench.rs b/crates/analyticsdb-engine/benches/query_log_bench.rs index b0fdc34..513ca64 100644 --- a/crates/analyticsdb-engine/benches/query_log_bench.rs +++ b/crates/analyticsdb-engine/benches/query_log_bench.rs @@ -1,4 +1,4 @@ -use criterion::{black_box, Criterion, BenchmarkId}; +use criterion::{black_box, Criterion, criterion_group, criterion_main}; use analyticsdb_engine::query_log::QueryLog; use analyticsdb_core::{QueryRequest, SessionContext}; @@ -47,12 +47,10 @@ fn query_log_observe_and_finish_benchmark(c: &mut Criterion) { &admission, "SELECT 1", ); - probe.observe_plan(black_box(&datafusion::logical_expr::LogicalPlan::EmptyRelation( - datafusion::logical_expr::EmptyRelation { - produce_one_row: false, - schema: std::sync::Arc::new(datafusion::common::DFSchema::empty()), - }, - ))); + let empty_plan = datafusion::physical_plan::empty::EmptyExec::new(std::sync::Arc::new( + datafusion::arrow::datatypes::Schema::empty() + )); + probe.observe_plan(&empty_plan); probe.observe_read(black_box(100), black_box(1024)); probe.finish_result(&Ok(analyticsdb_engine::QueryExecutionResult { query_id: "test".to_string(), diff --git a/crates/analyticsdb-engine/src/audit_log/mod.rs b/crates/analyticsdb-engine/src/audit_log/mod.rs index c3b7bca..f7acc9c 100644 --- a/crates/analyticsdb-engine/src/audit_log/mod.rs +++ b/crates/analyticsdb-engine/src/audit_log/mod.rs @@ -142,6 +142,7 @@ impl AuditLogRecord { } } + #[allow(clippy::too_many_arguments)] pub fn error( event_type: AuditEventType, user: impl Into, diff --git a/crates/analyticsdb-engine/src/ddl.rs b/crates/analyticsdb-engine/src/ddl.rs index ce1571e..2ab3364 100644 --- a/crates/analyticsdb-engine/src/ddl.rs +++ b/crates/analyticsdb-engine/src/ddl.rs @@ -420,7 +420,7 @@ impl PrototypeEngine { crate::audit_log::AuditEventType::CreateUser, &request.session.user, &request.session.role, - &format!("CREATE USER {user_name}"), + format!("CREATE USER {user_name}"), "user", &user_name, "embedded", @@ -445,7 +445,7 @@ impl PrototypeEngine { crate::audit_log::AuditEventType::DropUser, &request.session.user, &request.session.role, - &format!("DROP USER {user_name}"), + format!("DROP USER {user_name}"), "user", &user_name, "embedded", @@ -470,7 +470,7 @@ impl PrototypeEngine { crate::audit_log::AuditEventType::AlterUser, &request.session.user, &request.session.role, - &format!("ALTER USER {user_name} PASSWORD"), + format!("ALTER USER {user_name} PASSWORD"), "user", &user_name, "embedded", @@ -1015,7 +1015,7 @@ impl PrototypeEngine { crate::audit_log::AuditEventType::CreateTable, &request.session.user, &request.session.role, - &format!("CREATE TABLE {name}"), + format!("CREATE TABLE {name}"), "table", &name, "embedded", @@ -1550,11 +1550,14 @@ impl PrototypeEngine { let (store, old_prefix) = storage::store_for_location(storage_path_str)?; // Calculate new storage location by replacing the table name part. - // Managed tables use names like ____
.table.parquet - let old_suffix = format!("{}.table.parquet", name); - let new_suffix = format!("{}.table.parquet", new_name); - let new_location_str = - storage_path_str.replace(&old_suffix, &new_suffix); + // Managed tables use names like ____
.table.parquet or table=
+ let old_suffix_flat = format!("{}.table.parquet", name); + let new_suffix_flat = format!("{}.table.parquet", new_name); + let old_suffix_dir = format!("table={}", name); + let new_suffix_dir = format!("table={}", new_name); + let new_location_str = storage_path_str + .replace(&old_suffix_flat, &new_suffix_flat) + .replace(&old_suffix_dir, &new_suffix_dir); let (_, new_prefix) = storage::store_for_location(&new_location_str)?; storage::rename_prefix(&store, &old_prefix, &new_prefix).await?; @@ -1820,9 +1823,13 @@ impl PrototypeEngine { if let Some(storage_path_str) = &relation.storage_path { let (store, old_obj_prefix) = storage::store_for_location(storage_path_str)?; - let old_part = format!("{}__{}__", database_name, name); - let new_part = format!("{}__{}__", database_name, new_name); - let new_location_str = storage_path_str.replace(&old_part, &new_part); + let old_part_flat = format!("{}__{}__", database_name, name); + let new_part_flat = format!("{}__{}__", database_name, new_name); + let old_part_dir = format!("schema={}", name); + let new_part_dir = format!("schema={}", new_name); + let new_location_str = storage_path_str + .replace(&old_part_flat, &new_part_flat) + .replace(&old_part_dir, &new_part_dir); let (_, new_obj_prefix) = storage::store_for_location(&new_location_str)?; storage::rename_prefix(&store, &old_obj_prefix, &new_obj_prefix).await?; self.control_plane @@ -1879,9 +1886,13 @@ impl PrototypeEngine { if let Some(storage_path_str) = &relation.storage_path { let (store, old_obj_prefix) = storage::store_for_location(storage_path_str)?; - let old_part = format!("{}__{}__", name, relation.schema); - let new_part = format!("{}__{}__", new_name, relation.schema); - let new_location_str = storage_path_str.replace(&old_part, &new_part); + let old_part_flat = format!("{}__{}__", name, relation.schema); + let new_part_flat = format!("{}__{}__", new_name, relation.schema); + let old_part_dir = format!("db={}", name); + let new_part_dir = format!("db={}", new_name); + let new_location_str = storage_path_str + .replace(&old_part_flat, &new_part_flat) + .replace(&old_part_dir, &new_part_dir); let (_, new_obj_prefix) = storage::store_for_location(&new_location_str)?; storage::rename_prefix(&store, &old_obj_prefix, &new_obj_prefix) @@ -2213,7 +2224,7 @@ impl PrototypeEngine { crate::audit_log::AuditEventType::DropTable, &request.session.user, &request.session.role, - &format!("DROP TABLE {name}"), + format!("DROP TABLE {name}"), "table", &name, "embedded", @@ -2323,7 +2334,7 @@ impl PrototypeEngine { crate::audit_log::AuditEventType::GrantPrivilege, &request.session.user, &request.session.role, - &format!("GRANT {privilege} ON {object_type} {qualified_name} TO {grantee}"), + format!("GRANT {privilege} ON {object_type} {qualified_name} TO {grantee}"), object_type, &qualified_name, "embedded", @@ -2363,7 +2374,7 @@ impl PrototypeEngine { crate::audit_log::AuditEventType::RevokePrivilege, &request.session.user, &request.session.role, - &format!("REVOKE {privilege} ON {object_type} {qualified_name} FROM {grantee}"), + format!("REVOKE {privilege} ON {object_type} {qualified_name} FROM {grantee}"), object_type, &qualified_name, "embedded", diff --git a/crates/analyticsdb-engine/src/dispatch_impl.rs b/crates/analyticsdb-engine/src/dispatch_impl.rs index b95e08f..8b3f517 100644 --- a/crates/analyticsdb-engine/src/dispatch_impl.rs +++ b/crates/analyticsdb-engine/src/dispatch_impl.rs @@ -50,10 +50,8 @@ impl PrototypeEngine { Some(plan) } else if let Some(plan) = distributed_distinct_plan(&request.sql, &table_name) { Some(plan) - } else if let Some(plan) = distributed_order_limit_plan(&request.sql, &table_name) { - Some(plan) } else { - None + distributed_order_limit_plan(&request.sql, &table_name) } }; // Block distribution for window functions or other unsupported function patterns. diff --git a/crates/analyticsdb-engine/src/distributed.rs b/crates/analyticsdb-engine/src/distributed.rs index 1b68b0c..14dbb46 100644 --- a/crates/analyticsdb-engine/src/distributed.rs +++ b/crates/analyticsdb-engine/src/distributed.rs @@ -89,7 +89,7 @@ pub fn partition_files_for_workers( .enumerate() .min_by_key(|(_, w)| *w) .map(|(i, _)| i) - .unwrap(); + .unwrap_or(0); chunks[min_idx].push(file); bucket_weights[min_idx] += weight; } diff --git a/crates/analyticsdb-engine/src/lib.rs b/crates/analyticsdb-engine/src/lib.rs index bbbf4a6..b231ec0 100644 --- a/crates/analyticsdb-engine/src/lib.rs +++ b/crates/analyticsdb-engine/src/lib.rs @@ -609,15 +609,9 @@ impl PrototypeEngine { table_name: &str, privilege: &str, ) -> Result<()> { - // Fetch the user record to determine admin status. - let is_admin = self - .control_plane - .catalog_user(&session.user) - .await - .map(|u| u.is_admin) - .unwrap_or(false); - - if is_admin { + // Administrators (via the `is_admin` flag or `Administrators` group + // membership) bypass per-object privilege checks. + if self.control_plane.is_admin(&session.user).await { return Ok(()); } @@ -1800,12 +1794,12 @@ FROM generate_series(1, 1000000) AS s(n) for _ in 0..20 { let result = engine .execute_query(&QueryRequest { - sql: "SELECT query, event_type, protocol, result_rows FROM system.query_log WHERE query = 'SELECT 1 AS logged_value' ORDER BY event_time_us LIMIT 1".to_string(), + sql: "SELECT query, query_kind, protocol, result_rows FROM system.query_log WHERE query = 'SELECT 1 AS logged_value' ORDER BY event_time_us LIMIT 1".to_string(), session: session.clone(), query_id: None, }) - .await - .expect("query log should be readable"); + .await + .expect("query log should be readable"); rows = result.to_query_response().rows; if !rows.is_empty() { break; @@ -1817,7 +1811,7 @@ FROM generate_series(1, 1000000) AS s(n) rows, vec![vec![ "SELECT 1 AS logged_value".to_string(), - "QueryFinish".to_string(), + "Select".to_string(), "embedded".to_string(), "1".to_string() ]] @@ -1944,7 +1938,7 @@ FROM generate_series(1, 1000000) AS s(n) let entry = entry.expect("valid entry"); if entry.file_type().expect("valid file type").is_dir() { let name = entry.file_name(); - if name.to_string_lossy().chars().all(|c| c.is_ascii_digit()) { + if name.to_string_lossy().starts_with("date=") { found_partitioned = true; break; } @@ -1952,7 +1946,7 @@ FROM generate_series(1, 1000000) AS s(n) } assert!( found_partitioned, - "should have created partitioned YYYY/ directories" + "should have created partitioned date=YYYY-MM-DD directories" ); cleanup_catalog_artifacts(&catalog_path); } diff --git a/crates/analyticsdb-engine/src/manifest.rs b/crates/analyticsdb-engine/src/manifest.rs index 6ec92fc..115e711 100644 --- a/crates/analyticsdb-engine/src/manifest.rs +++ b/crates/analyticsdb-engine/src/manifest.rs @@ -2,11 +2,8 @@ use anyhow::Result; use bytes::Bytes; use chrono::Utc; use datafusion::arrow::array::RecordBatch; -use datafusion::arrow::datatypes::{DataType, SchemaRef}; +use datafusion::arrow::datatypes::SchemaRef; use datafusion::parquet::arrow::arrow_reader::ParquetRecordBatchReader; -use datafusion::scalar::ScalarValue; -use datafusion_common::{ColumnStatistics, Statistics}; -use datafusion_common::stats::Precision; use futures::StreamExt; use object_store::path::Path as OPath; use object_store::{Error as OsError, ObjectStore, ObjectStoreExt, PutMode, PutOptions, UpdateVersion}; @@ -496,119 +493,6 @@ pub async fn compact_table( Ok(written) } -/// Converts a Manifest to DataFusion Statistics for query planning. -pub fn manifest_to_statistics(manifest: &Manifest, schema: &SchemaRef) -> Statistics { - let num_rows: usize = manifest.files.iter().map(|e| e.row_count as usize).sum(); - let total_byte_size: usize = manifest.files.iter().map(|e| e.size as usize).sum(); - - let mut column_statistics = Vec::new(); - - for field in schema.fields() { - let col_name = field.name(); - let data_type = field.data_type(); - - let mut total_null_count = 0i64; - let mut min_values: Vec = Vec::new(); - let mut max_values: Vec = Vec::new(); - let mut total_ndv = 0f64; - - for entry in &manifest.files { - for cs in &entry.column_stats { - if cs.name == *col_name { - total_null_count += cs.null_count; - if let Some(ref min) = cs.min_value { - min_values.push(min.clone()); - } - if let Some(ref max) = cs.max_value { - max_values.push(max.clone()); - } - if let Some(ndv) = cs.ndv_estimate { - total_ndv += ndv; - } - break; - } - } - } - - let null_count = Some(total_null_count as usize); - - // Parse min values into ScalarValue and find the minimum - let mut min_scalars: Vec = min_values - .iter() - .filter_map(|s| parse_scalar_value(data_type, s)) - .collect(); - let min_value = if !min_scalars.is_empty() { - min_scalars.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); - min_scalars.into_iter().next() - } else { - None - }; - - // Parse max values into ScalarValue and find the maximum - let mut max_scalars: Vec = max_values - .iter() - .filter_map(|s| parse_scalar_value(data_type, s)) - .collect(); - let max_value = if !max_scalars.is_empty() { - max_scalars.sort_by(|a, b| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal)); // sort descending - max_scalars.into_iter().next() - } else { - None - }; - - let distinct_count = if total_ndv > 0.0 { - Some(total_ndv as usize) - } else { - None - }; - - let to_precision = |opt: Option| -> Precision { - match opt { - Some(v) => Precision::Exact(v), - None => Precision::Absent, - } - }; - - let to_precision_scalar = |opt: Option| -> Precision { - match opt { - Some(v) => Precision::Exact(v), - None => Precision::Absent, - } - }; - - column_statistics.push(ColumnStatistics { - null_count: to_precision(null_count), - min_value: to_precision_scalar(min_value), - max_value: to_precision_scalar(max_value), - distinct_count: to_precision(distinct_count), - ..Default::default() - }); - } - - Statistics { - num_rows: Precision::Exact(num_rows), - total_byte_size: Precision::Exact(total_byte_size), - column_statistics, - } -} - -/// Parse a string into ScalarValue based on the target data type. -fn parse_scalar_value(data_type: &DataType, s: &str) -> Option { - match data_type { - DataType::Int8 => s.parse::().ok().map(|v| ScalarValue::Int8(Some(v))), - DataType::Int16 => s.parse::().ok().map(|v| ScalarValue::Int16(Some(v))), - DataType::Int32 => s.parse::().ok().map(|v| ScalarValue::Int32(Some(v))), - DataType::Int64 => s.parse::().ok().map(|v| ScalarValue::Int64(Some(v))), - DataType::UInt8 => s.parse::().ok().map(|v| ScalarValue::UInt8(Some(v))), - DataType::UInt16 => s.parse::().ok().map(|v| ScalarValue::UInt16(Some(v))), - DataType::UInt32 => s.parse::().ok().map(|v| ScalarValue::UInt32(Some(v))), - DataType::UInt64 => s.parse::().ok().map(|v| ScalarValue::UInt64(Some(v))), - DataType::Float32 => s.parse::().ok().map(|v| ScalarValue::Float32(Some(v))), - DataType::Float64 => s.parse::().ok().map(|v| ScalarValue::Float64(Some(v))), - DataType::Utf8 => Some(ScalarValue::Utf8(Some(s.to_string()))), - _ => None, - } -} #[cfg(test)] mod tests { diff --git a/crates/analyticsdb-engine/src/query_log/mod.rs b/crates/analyticsdb-engine/src/query_log/mod.rs index 0559f6c..5ba2ec0 100644 --- a/crates/analyticsdb-engine/src/query_log/mod.rs +++ b/crates/analyticsdb-engine/src/query_log/mod.rs @@ -499,11 +499,9 @@ pub(crate) async fn cleanup_expired_logs( // Check if path contains a date=YYYY-MM-DD partition let mut parts = rel_path.split('/'); if let Some(first_part) = parts.next() { - if first_part.starts_with("date=") { - let date_str = &first_part[5..]; // strip "date=" + if let Some(date_str) = first_part.strip_prefix("date=") { if let Ok(date) = chrono::NaiveDate::parse_from_str(date_str, "%Y-%m-%d") { - let partition_date = Utc.ymd(date.year(), date.month(), date.day()); - if partition_date < expiration.date() { + if date < expiration.date_naive() { if let Err(e) = store.delete(&meta.location).await { debug!("failed to delete expired query log {}: {}", path, e); } @@ -527,9 +525,7 @@ fn records_to_batch(records: &[QueryLogRecord]) -> Result { Ok(RecordBatch::try_new( schema(), vec![ - string_array(records, |r| Some(r.event_type.as_str())), timestamp_array(records, |r| r.event_time_us), - int64_array(records, |r| r.event_time_us), timestamp_array(records, |r| r.query_start_time_us), string_array(records, |r| Some(r.query_id.as_str())), string_array(records, |r| Some(r.initial_query_id.as_str())), diff --git a/crates/analyticsdb-engine/src/system_catalog.rs b/crates/analyticsdb-engine/src/system_catalog.rs index 91db29b..d6af78d 100644 --- a/crates/analyticsdb-engine/src/system_catalog.rs +++ b/crates/analyticsdb-engine/src/system_catalog.rs @@ -1322,7 +1322,7 @@ impl TableProvider for PgTypeTable { typname.push(name.to_string()); typnamespace.push(11_u32); // pg_catalog typowner.push(10_u32); - typlen.push(len as i16); + typlen.push(len); typbyval.push(byval); typtype.push(t.to_string()); typcategory.push(cat.to_string()); diff --git a/crates/analyticsdb-gateway/Cargo.toml b/crates/analyticsdb-gateway/Cargo.toml index 2842e0f..4a56373 100644 --- a/crates/analyticsdb-gateway/Cargo.toml +++ b/crates/analyticsdb-gateway/Cargo.toml @@ -6,15 +6,18 @@ rust-version.workspace = true version.workspace = true [dependencies] +# Catalog metadata reads (explorer, user/group listings, admin checks). All SQL +# execution and catalog mutations are proxied to the running server over pg-wire. analyticsdb-control = { path = "../analyticsdb-control" } -analyticsdb-engine = { path = "../analyticsdb-engine" } -analyticsdb-protocol = { path = "../analyticsdb-protocol" } -analyticsdb-server = { path = "../analyticsdb-server" } # Async runtime tokio = { workspace = true, features = ["full"] } tokio-stream = { workspace = true } +# PostgreSQL client — the gateway proxies SQL to the running server's pg-wire +# endpoint so the server's engine is the single source of truth. +tokio-postgres = { workspace = true } + # Web framework axum = { version = "0.7", features = ["json"] } tower = { workspace = true } diff --git a/crates/analyticsdb-gateway/src/config.rs b/crates/analyticsdb-gateway/src/config.rs index 0e4b8b2..68a967a 100644 --- a/crates/analyticsdb-gateway/src/config.rs +++ b/crates/analyticsdb-gateway/src/config.rs @@ -23,6 +23,11 @@ pub struct GatewayConfig { /// AnalyticsDB server endpoint (for proxying queries) pub analyticsdb_pg_endpoint: Option, pub analyticsdb_flight_endpoint: Option, + + /// Path to the catalog store (SQLite). Used by the admin API to read and + /// mutate users/groups directly. When `None`, it is resolved from the + /// control-plane config file (`catalog_path`) with sensible fallbacks. + pub catalog_path: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -63,12 +68,13 @@ impl Default for GatewayConfig { fn default() -> Self { Self { bind_addr: "0.0.0.0:8080".to_string(), - control_plane_config_path: "cluster-config.json".to_string(), + control_plane_config_path: discover_config_path(), session_timeout_seconds: 3600, jwt_secret: "change-me-in-production".to_string(), oidc: OidcConfig::default(), - analyticsdb_pg_endpoint: Some("127.0.0.1:5432".to_string()), + analyticsdb_pg_endpoint: None, analyticsdb_flight_endpoint: Some("127.0.0.1:8081".to_string()), + catalog_path: None, } } } @@ -140,6 +146,80 @@ impl GatewayConfig { config.analyticsdb_flight_endpoint = Some(endpoint); } + if let Ok(path) = env::var("ANALYTICSDB_CATALOG_PATH") { + config.catalog_path = Some(path); + } + Ok(config) } + + /// The server's default catalog file (its `--catalog-path` default). + pub const DEFAULT_CATALOG_PATH: &'static str = "analyticsdb-catalog.db"; + /// The server's default pg-wire endpoint. + pub const DEFAULT_PG_ENDPOINT: &'static str = "127.0.0.1:5432"; + + /// Reads the server's config file (e.g. `config/cluster-config.json`) so the + /// gateway can pick up the catalog path and pg-wire endpoint the server is + /// using. Returns `None` if the file is missing or unparseable. + fn server_config(&self) -> Option { + let raw = std::fs::read_to_string(&self.control_plane_config_path).ok()?; + serde_json::from_str(&raw).ok() + } + + /// Resolves the catalog store path the gateway reads metadata from. + /// + /// Order: `ANALYTICSDB_CATALOG_PATH` env → `catalog_path` from the server's + /// config file → the server's default catalog file. + pub fn resolve_catalog_path(&self) -> String { + if let Some(path) = &self.catalog_path { + return path.clone(); + } + if let Some(path) = self + .server_config() + .as_ref() + .and_then(|c| c.get("catalog_path")) + .and_then(|v| v.as_str()) + { + return path.to_string(); + } + Self::DEFAULT_CATALOG_PATH.to_string() + } + + /// Resolves the server pg-wire endpoint SQL is proxied to. + /// + /// Order: `ANALYTICSDB_PG_ENDPOINT` env → `postgres_addr` from the server's + /// config file → `host:base_postgres_port` derived from the config file → + /// the server's default endpoint. + pub fn resolve_pg_endpoint(&self) -> String { + if let Some(endpoint) = &self.analyticsdb_pg_endpoint { + return endpoint.clone(); + } + if let Some(config) = self.server_config() { + if let Some(addr) = config.get("postgres_addr").and_then(|v| v.as_str()) { + return addr.to_string(); + } + if let Some(port) = config.get("base_postgres_port").and_then(|v| v.as_u64()) { + let host = config + .get("advertise_host") + .and_then(|v| v.as_str()) + .unwrap_or("127.0.0.1"); + return format!("{host}:{port}"); + } + } + Self::DEFAULT_PG_ENDPOINT.to_string() + } +} + +/// Default locations to look for the server config, in priority order. Config +/// lives in a `config/` directory by convention. +pub const DEFAULT_CONFIG_PATHS: [&str; 2] = ["config/cluster-config.json", "cluster-config.json"]; + +/// Returns the first existing default config path, or the preferred default +/// (`config/cluster-config.json`) when none exist yet. +pub fn discover_config_path() -> String { + DEFAULT_CONFIG_PATHS + .iter() + .find(|path| std::path::Path::new(path).exists()) + .unwrap_or(&DEFAULT_CONFIG_PATHS[0]) + .to_string() } diff --git a/crates/analyticsdb-gateway/src/main.rs b/crates/analyticsdb-gateway/src/main.rs index e319f11..adb7ddf 100644 --- a/crates/analyticsdb-gateway/src/main.rs +++ b/crates/analyticsdb-gateway/src/main.rs @@ -26,6 +26,25 @@ pub struct GatewayState { pub session_store: Arc, } +impl GatewayState { + /// Opens a fresh control plane over the shared catalog file for read-only + /// metadata (explorer, user/group listings, admin checks). Opening per + /// request — rather than caching an engine — means the gateway always + /// reflects the server's latest persisted catalog state, with no divergent + /// in-memory cache. All execution and mutation is proxied to the server. + pub async fn open_catalog( + &self, + ) -> anyhow::Result { + let path = self.config.resolve_catalog_path(); + analyticsdb_control::ControlPlane::from_catalog_path(&path).await + } + + /// The server's pg-wire endpoint that SQL is proxied to. + pub fn pg_endpoint(&self) -> String { + self.config.resolve_pg_endpoint() + } +} + #[tokio::main] async fn main() -> anyhow::Result<()> { // Initialize tracing @@ -40,8 +59,15 @@ async fn main() -> anyhow::Result<()> { tracing::info!("Starting AnalyticsDB Gateway on {}", config.bind_addr); tracing::info!("OIDC enabled: {}", config.oidc.is_enabled()); - // Initialize session store - let session_store = Arc::new(session::SessionStore::new(config.session_timeout_seconds)); + // Initialize session store with the real config (jwt secret, timeout). + let session_store = Arc::new(session::SessionStore::with_config(config.clone())); + + tracing::info!( + "Config: {} · Proxying SQL to server at {} (pg-wire) · catalog metadata from {}", + config.control_plane_config_path, + config.resolve_pg_endpoint(), + config.resolve_catalog_path(), + ); // Create shared state let state = Arc::new(GatewayState { @@ -60,8 +86,9 @@ async fn main() -> anyhow::Result<()> { // Session .route("/api/session", get(routes::session::get_session)) .route("/api/session", post(routes::session::update_session)) - // Auth refresh (needs valid token) + // Auth refresh + logout (need a valid token / session) .route("/api/auth/refresh", post(routes::auth::refresh)) + .route("/api/auth/logout", post(routes::auth::logout)) // Explorer (live metadata) .route("/api/explorer", get(routes::explorer::get_explorer_snapshot)) .route("/api/explorer/databases", get(routes::explorer::list_databases)) @@ -76,6 +103,11 @@ async fn main() -> anyhow::Result<()> { .route("/api/admin/databases/:name", delete(routes::admin::drop_database)) .route("/api/admin/users", get(routes::admin::list_users).post(routes::admin::create_user)) .route("/api/admin/users/:name", delete(routes::admin::drop_user)) + .route("/api/admin/users/:name/reset-password", post(routes::admin::reset_user_password)) + .route("/api/admin/groups", get(routes::admin::list_groups).post(routes::admin::create_group)) + .route("/api/admin/groups/:name", delete(routes::admin::drop_group)) + .route("/api/admin/groups/:name/members", post(routes::admin::add_group_member)) + .route("/api/admin/groups/:name/members/:user", delete(routes::admin::remove_group_member)) // System .route("/api/system/metrics", get(routes::system::get_metrics)) .route("/api/system/query-log", get(routes::system::get_query_log)) @@ -91,7 +123,6 @@ async fn main() -> anyhow::Result<()> { .route("/healthz", get(routes::health::liveness)) .route("/readyz", get(routes::health::readiness)) .route("/api/auth/login", post(routes::auth::login)) - .route("/api/auth/logout", post(routes::auth::logout)) .route("/api/auth/oidc/authorize", get(routes::auth::oidc_authorize)) .route("/api/auth/oidc/callback", get(routes::auth::oidc_callback)) .with_state(Arc::clone(&state)); diff --git a/crates/analyticsdb-gateway/src/proxy.rs b/crates/analyticsdb-gateway/src/proxy.rs index 493a62f..a4d6aa9 100644 --- a/crates/analyticsdb-gateway/src/proxy.rs +++ b/crates/analyticsdb-gateway/src/proxy.rs @@ -1,40 +1,124 @@ -//! Query proxy - forwards queries to AnalyticsDB engine - -use std::sync::Arc; - -use crate::error::GatewayResult; -use crate::session::SessionClaims; - -/// Proxy a query to the PostgreSQL wire protocol -pub async fn proxy_to_postgres( - _claims: &SessionClaims, - _sql: &str, - _endpoint: &str, -) -> GatewayResult { - // In production, this would: - // 1. Connect to AnalyticsDB via tokio-postgres - // 2. Execute the query with the user's session context - // 3. Return the results - - // For now, return a placeholder - Ok(serde_json::json!({ - "prototype": "PostgreSQL proxy not yet implemented" - })) +//! pg-wire proxy — forwards SQL to the running AnalyticsDB server so its engine +//! is the single source of truth for execution and catalog mutations. + +use anyhow::{Context, Result}; +use tokio_postgres::{Config, NoTls, SimpleQueryMessage}; + +/// The text result of a simple-query execution over the wire. +#[derive(Debug, Default)] +pub struct ProxyResult { + pub columns: Vec, + /// Row values as text; `None` represents SQL NULL. + pub rows: Vec>>, + /// Rows affected reported by the final `CommandComplete`, if any. + pub affected_rows: Option, +} + +/// Splits a `host:port` endpoint, defaulting the port to 5432. +fn parse_endpoint(endpoint: &str) -> (String, u16) { + match endpoint.rsplit_once(':') { + Some((host, port)) => (host.to_string(), port.parse().unwrap_or(5432)), + None => (endpoint.to_string(), 5432), + } +} + +/// Opens a fresh pg-wire connection to the server as `user`, authenticating with +/// `password` (SCRAM-SHA-256 is negotiated transparently). The caller owns the +/// returned client; dropping it closes the connection. +pub async fn connect( + endpoint: &str, + user: &str, + password: &str, + database: &str, +) -> Result { + let (host, port) = parse_endpoint(endpoint); + let mut config = Config::new(); + config + .host(&host) + .port(port) + .user(user) + .dbname(database) + .application_name("analyticsdb-gateway"); + if !password.is_empty() { + config.password(password); + } + + let (client, connection) = config + .connect(NoTls) + .await + .with_context(|| format!("failed to connect to AnalyticsDB pg-wire at {endpoint}"))?; + + // Drive the connection in the background for the lifetime of the client. + tokio::spawn(async move { + let _ = connection.await; + }); + + Ok(client) +} + +/// Connects as `user` and executes `sql` via the simple query protocol, which +/// returns all values as text — ideal for a SQL console and DDL alike. +pub async fn execute_sql( + endpoint: &str, + user: &str, + password: &str, + database: &str, + sql: &str, +) -> Result { + let client = connect(endpoint, user, password, database).await?; + let messages = client + .simple_query(sql) + .await + .context("query execution failed")?; + + let mut result = ProxyResult::default(); + for message in messages { + match message { + SimpleQueryMessage::Row(row) => { + if result.columns.is_empty() { + result.columns = row + .columns() + .iter() + .map(|column| column.name().to_string()) + .collect(); + } + let mut values = Vec::with_capacity(row.columns().len()); + for index in 0..row.columns().len() { + values.push(row.get(index).map(|value| value.to_string())); + } + result.rows.push(values); + } + SimpleQueryMessage::CommandComplete(affected) => { + result.affected_rows = Some(affected); + } + _ => {} + } + } + Ok(result) +} + +/// Runs several statements in order over a single connection, stopping at the +/// first error. The server's simple-query handler executes one statement at a +/// time, so multi-statement DDL (e.g. CREATE USER then ALTER GROUP) must be +/// issued as separate queries rather than a semicolon-joined string. +pub async fn execute_statements( + endpoint: &str, + user: &str, + password: &str, + database: &str, + statements: &[String], +) -> Result<()> { + let client = connect(endpoint, user, password, database).await?; + for statement in statements { + client + .simple_query(statement) + .await + .with_context(|| format!("statement failed: {statement}"))?; + } + Ok(()) } -/// Proxy a query to Flight SQL -pub async fn proxy_to_flight( - _claims: &SessionClaims, - _sql: &str, - _endpoint: &str, -) -> GatewayResult { - // In production, this would: - // 1. Connect to AnalyticsDB via Arrow Flight SQL client - // 2. Execute the query with the user's session context - // 3. Return the results as JSON - - // For now, return a placeholder - Ok(serde_json::json!({ - "prototype": "Flight SQL proxy not yet implemented" - })) +/// Quotes a string as a SQL single-quoted literal (doubling embedded quotes). +pub fn sql_literal(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) } diff --git a/crates/analyticsdb-gateway/src/routes/admin.rs b/crates/analyticsdb-gateway/src/routes/admin.rs index 985c99a..d6baeb9 100644 --- a/crates/analyticsdb-gateway/src/routes/admin.rs +++ b/crates/analyticsdb-gateway/src/routes/admin.rs @@ -1,14 +1,26 @@ -//! Admin routes - databases, users, grants management (placeholder implementations) +//! Admin routes — users and groups management. +//! +//! Reads (listing users/groups) come from a fresh view of the shared catalog, +//! reflecting the server's latest persisted state. Mutations are sent to the +//! server as DDL over pg-wire so the **server** applies them to its live +//! catalog — there is no separate writer in the gateway to diverge. The acting +//! user must be an effective administrator (a member of `Administrators`). +use analyticsdb_control::{ControlPlane, ADMINISTRATORS_GROUP}; use axum::{ - extract::{Extension, Path, Query, State}, + extract::{Extension, Path, State}, Json, }; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use serde_json::json; +use std::sync::Arc; -use crate::error::GatewayResult; +use crate::error::{GatewayError, GatewayResult}; +use crate::proxy::sql_literal; use crate::session::SessionClaims; +use crate::GatewayState; + +const DEFAULT_DATABASE: &str = "postgres"; #[derive(Debug, Deserialize)] pub struct CreateDatabaseRequest { @@ -20,100 +32,283 @@ pub struct CreateDatabaseRequest { pub struct CreateUserRequest { pub name: String, pub password: String, - pub role: Option, + /// Optional group memberships to grant on creation (e.g. `["Administrators"]`). + #[serde(default)] + pub groups: Vec, } #[derive(Debug, Deserialize)] -pub struct GrantRequest { - pub grantee: String, - pub object_type: String, - pub object_name: String, - pub privilege: String, +pub struct CreateGroupRequest { + pub name: String, } -/// List all databases (placeholder) -pub async fn list_databases( - State(_state): State>, +#[derive(Debug, Deserialize)] +pub struct GroupMemberRequest { + pub user: String, +} + +#[derive(Debug, Default, Deserialize)] +pub struct ResetPasswordRequest { + /// Explicit replacement password. When absent/empty, a random one is + /// generated server-side and returned once. + #[serde(default)] + pub password: Option, +} + +/// Opens a fresh read-only catalog view and verifies the session user is an +/// effective administrator. Used by the listing endpoints. +async fn admin_read( + state: &Arc, + claims: &SessionClaims, +) -> GatewayResult { + let control_plane = state.open_catalog().await.map_err(GatewayError::Internal)?; + if !control_plane.is_admin(&claims.sub).await { + return Err(GatewayError::Forbidden); + } + Ok(control_plane) +} + +/// Runs administrative DDL on the server over pg-wire as the session user. The +/// server enforces administrator authorization and applies the change to its +/// live catalog; any error (e.g. "User already exists") is surfaced to the UI. +async fn run_admin_sql( + state: &Arc, + claims: &SessionClaims, + statements: &[String], +) -> GatewayResult<()> { + // Early, friendly rejection for non-admins (the server also enforces this). + let control_plane = state.open_catalog().await.map_err(GatewayError::Internal)?; + if !control_plane.is_admin(&claims.sub).await { + return Err(GatewayError::Forbidden); + } + drop(control_plane); + + let password = state + .session_store + .password_for(&claims.session_id) + .ok_or(GatewayError::Unauthorized)?; + + crate::proxy::execute_statements( + &state.pg_endpoint(), + &claims.sub, + &password, + DEFAULT_DATABASE, + statements, + ) + .await + .map_err(|error| { + // Surface the deepest cause (the db error) for a useful message. + GatewayError::BadRequest( + error + .root_cause() + .to_string() + .trim_start_matches("db error: ") + .trim_start_matches("ERROR: ") + .to_string(), + ) + })?; + Ok(()) +} + +/// Lists all users (excludes group accounts), annotated with admin status and +/// group memberships. +pub async fn list_users( + Extension(claims): Extension, + State(state): State>, ) -> GatewayResult>> { - let result = vec![ - json!({ "name": "default", "owner": "admin" }), - ]; + let control_plane = admin_read(&state, &claims).await?; + let snapshot = control_plane.cluster_snapshot().await; + + // A "group" is an account with no password and at least the capacity to hold + // members; we treat password-less accounts as groups. + let groups: Vec<&analyticsdb_control::CatalogUser> = snapshot + .users + .iter() + .filter(|u| u.password.is_none()) + .collect(); + + let result = snapshot + .users + .iter() + .filter(|u| u.password.is_some()) + .map(|u| { + let memberships: Vec = groups + .iter() + .filter(|g| g.members.contains(&u.name)) + .map(|g| g.name.clone()) + .collect(); + let is_admin = u.is_admin || memberships.iter().any(|g| g == ADMINISTRATORS_GROUP); + json!({ + "name": u.name, + "is_admin": is_admin, + "groups": memberships, + "password_version": u.password_version, + "password_rotated_at_epoch_ms": u.password_rotated_at_epoch_ms, + }) + }) + .collect(); Ok(Json(result)) } -/// Create a new database (placeholder) -pub async fn create_database( - Extension(_claims): Extension, - Json(_req): Json, +/// Creates a new user, optionally enrolling them into the requested groups. +pub async fn create_user( + Extension(claims): Extension, + State(state): State>, + Json(req): Json, ) -> GatewayResult> { - Ok(Json(json!({ "message": "Database created (placeholder)" }))) + let mut statements = vec![format!( + "CREATE USER {} PASSWORD {}", + req.name, + sql_literal(&req.password) + )]; + for group in &req.groups { + statements.push(format!("ALTER GROUP {} ADD USER {}", group, req.name)); + } + run_admin_sql(&state, &claims, &statements).await?; + Ok(Json(json!({ "message": format!("User '{}' created.", req.name) }))) } -/// Get a specific database (placeholder) -pub async fn get_database( - Path(_name): Path, +/// Drops a user. +pub async fn drop_user( + Extension(claims): Extension, + State(state): State>, + Path(name): Path, ) -> GatewayResult> { - Ok(Json(json!({ "name": "default", "owner": "admin" }))) + run_admin_sql(&state, &claims, &[format!("DROP USER {name}")]).await?; + Ok(Json(json!({ "message": format!("User '{name}' dropped.") }))) } -/// Drop a database (placeholder) -pub async fn drop_database( - Path(_name): Path, +/// Resets a user's password. If the request supplies a password it is used as +/// given; otherwise a strong random one is generated server-side. The plaintext +/// is returned only when it was generated (so the operator can record it once). +pub async fn reset_user_password( + Extension(claims): Extension, + State(state): State>, + Path(name): Path, + Json(req): Json, ) -> GatewayResult> { - Ok(Json(json!({ "message": "Database dropped (placeholder)" }))) + let (password, generated) = match req.password { + Some(p) if !p.trim().is_empty() => (p, false), + _ => (analyticsdb_control::generate_random_password(), true), + }; + + run_admin_sql( + &state, + &claims, + &[format!("ALTER USER {name} PASSWORD {}", sql_literal(&password))], + ) + .await?; + + let mut response = json!({ + "name": name, + "generated": generated, + "message": format!("Password for '{}' updated.", name), + }); + if generated { + response["password"] = json!(password); + response["message"] = + json!(format!("Password for '{}' reset. Store it now — it is shown once.", name)); + } + Ok(Json(response)) } -/// List all users (placeholder) -pub async fn list_users( - State(_state): State>, +/// Lists all groups with their members. +pub async fn list_groups( + Extension(claims): Extension, + State(state): State>, ) -> GatewayResult>> { - let result = vec![ - json!({ "name": "admin", "role": "admin" }), - ]; + let control_plane = admin_read(&state, &claims).await?; + let snapshot = control_plane.cluster_snapshot().await; + let result = snapshot + .users + .iter() + .filter(|u| u.password.is_none()) + .map(|g| { + json!({ + "name": g.name, + "members": g.members.iter().cloned().collect::>(), + "member_count": g.members.len(), + }) + }) + .collect(); Ok(Json(result)) } -/// Create a new user (placeholder) -pub async fn create_user( - Extension(_claims): Extension, - Json(_req): Json, +/// Creates a new group. +pub async fn create_group( + Extension(claims): Extension, + State(state): State>, + Json(req): Json, ) -> GatewayResult> { - Ok(Json(json!({ "message": "User created (placeholder)" }))) + run_admin_sql(&state, &claims, &[format!("CREATE GROUP {}", req.name)]).await?; + Ok(Json(json!({ "message": format!("Group '{}' created.", req.name) }))) } -/// Get a specific user (placeholder) -pub async fn get_user( - Path(_name): Path, +/// Drops a group. +pub async fn drop_group( + Extension(claims): Extension, + State(state): State>, + Path(name): Path, ) -> GatewayResult> { - Ok(Json(json!({ "name": "admin", "role": "admin" }))) + run_admin_sql(&state, &claims, &[format!("DROP GROUP {name}")]).await?; + Ok(Json(json!({ "message": format!("Group '{name}' dropped.") }))) } -/// Drop a user (placeholder) -pub async fn drop_user( - Path(_name): Path, +/// Adds a member to a group. +pub async fn add_group_member( + Extension(claims): Extension, + State(state): State>, + Path(name): Path, + Json(req): Json, +) -> GatewayResult> { + run_admin_sql( + &state, + &claims, + &[format!("ALTER GROUP {name} ADD USER {}", req.user)], + ) + .await?; + Ok(Json(json!({ "message": format!("User '{}' added to '{name}'.", req.user) }))) +} + +/// Removes a member from a group. +pub async fn remove_group_member( + Extension(claims): Extension, + State(state): State>, + Path((name, user)): Path<(String, String)>, ) -> GatewayResult> { - Ok(Json(json!({ "message": "User dropped (placeholder)" }))) + run_admin_sql( + &state, + &claims, + &[format!("ALTER GROUP {name} DROP USER {user}")], + ) + .await?; + Ok(Json(json!({ "message": format!("User '{user}' removed from '{name}'.") }))) } -/// List grants (placeholder) -pub async fn list_grants( - State(_state): State>, +// --- Database endpoints (still placeholders) --- + +/// List all databases (placeholder) +pub async fn list_databases( + State(_state): State>, ) -> GatewayResult>> { - let result: Vec = vec![]; + let result = vec![json!({ "name": "default", "owner": "admin" })]; Ok(Json(result)) } -/// Grant privilege (placeholder) -pub async fn grant_privilege( +/// Create a new database (placeholder) +pub async fn create_database( Extension(_claims): Extension, - Json(_req): Json, + Json(_req): Json, ) -> GatewayResult> { - Ok(Json(json!({ "message": "Privilege granted (placeholder)" }))) + Ok(Json(json!({ "message": "Database created (placeholder)" }))) } -/// Revoke privilege (placeholder) -pub async fn revoke_privilege( - Path(_id): Path, -) -> GatewayResult> { - Ok(Json(json!({ "message": "Privilege revoked (placeholder)" }))) +/// Get a specific database (placeholder) +pub async fn get_database(Path(_name): Path) -> GatewayResult> { + Ok(Json(json!({ "name": "default", "owner": "admin" }))) +} + +/// Drop a database (placeholder) +pub async fn drop_database(Path(_name): Path) -> GatewayResult> { + Ok(Json(json!({ "message": "Database dropped (placeholder)" }))) } diff --git a/crates/analyticsdb-gateway/src/routes/auth.rs b/crates/analyticsdb-gateway/src/routes/auth.rs index 9cc2a20..dd47abf 100644 --- a/crates/analyticsdb-gateway/src/routes/auth.rs +++ b/crates/analyticsdb-gateway/src/routes/auth.rs @@ -4,14 +4,13 @@ use std::sync::Arc; use axum::{ extract::{Extension, Query, State}, - http::StatusCode, response::{Json, Redirect}, }; use serde::{Deserialize, Serialize}; use serde_json::json; use crate::error::GatewayResult; -use crate::session::{SessionClaims, SessionStore}; +use crate::session::SessionClaims; use crate::GatewayState; #[derive(Debug, Deserialize)] @@ -37,22 +36,45 @@ pub struct OidcCallbackQuery { pub state: String, } -/// Local login with username/password (placeholder - accepts admin/admin) +const DEFAULT_DATABASE: &str = "postgres"; +const DEFAULT_SCHEMA: &str = "public"; + +/// Local login with username/password. +/// +/// Credentials are validated by opening an authenticated pg-wire connection to +/// the running server (the single source of truth). On success the password is +/// cached in the session store so subsequent SQL is executed as this user, and +/// the session role is set to `admin` when the account is an effective +/// administrator (a member of the `Administrators` group). pub async fn login( State(state): State>, Json(req): Json, ) -> GatewayResult> { - // Placeholder implementation - accepts admin/admin - if req.username != "admin" || req.password != "admin" { - return Err(anyhow::anyhow!("Invalid credentials").into()); - } - - // Create session - let token = state.session_store.create_session( - "admin", - "admin", - "default", - "public", + use crate::error::GatewayError; + + // Authenticate against the server over pg-wire. + crate::proxy::execute_sql( + &state.pg_endpoint(), + &req.username, + &req.password, + DEFAULT_DATABASE, + "SELECT 1", + ) + .await + .map_err(|_| GatewayError::Unauthorized)?; + + // Determine admin status from the shared catalog (group-derived). + let role = match state.open_catalog().await { + Ok(catalog) if catalog.is_admin(&req.username).await => "admin", + _ => "user", + }; + + let token = state.session_store.create_session_with_password( + &req.username, + role, + DEFAULT_DATABASE, + DEFAULT_SCHEMA, + &req.password, )?; let claims = state.session_store.validate_token(&token)?; @@ -63,22 +85,33 @@ pub async fn login( })) } -/// Logout (client-side token disposal, could also add token blacklisting) -pub async fn logout() -> Json { +/// Logout — forget the cached pg-wire password for this session. +pub async fn logout( + Extension(claims): Extension, + State(state): State>, +) -> Json { + state.session_store.forget(&claims.session_id); Json(json!({ "message": "Logged out successfully" })) } -/// Refresh session token +/// Refresh session token, carrying the cached pg-wire password forward. pub async fn refresh( Extension(claims): Extension, State(state): State>, ) -> GatewayResult> { - let token = state.session_store.create_session( + let password = state + .session_store + .password_for(&claims.session_id) + .unwrap_or_default(); + + let token = state.session_store.create_session_with_password( &claims.sub, &claims.role, &claims.database, &claims.schema, + &password, )?; + state.session_store.forget(&claims.session_id); let new_claims = state.session_store.validate_token(&token)?; diff --git a/crates/analyticsdb-gateway/src/routes/explorer.rs b/crates/analyticsdb-gateway/src/routes/explorer.rs index 10f71e3..6a7edf9 100644 --- a/crates/analyticsdb-gateway/src/routes/explorer.rs +++ b/crates/analyticsdb-gateway/src/routes/explorer.rs @@ -1,4 +1,6 @@ -//! Explorer routes - live metadata browsing +//! Explorer routes - live metadata browsing backed by the catalog. + +use std::sync::Arc; use axum::{ extract::{Extension, Query, State}, @@ -7,8 +9,11 @@ use axum::{ use serde::{Deserialize, Serialize}; use serde_json::json; +use analyticsdb_control::CatalogRelationKind; + +use crate::error::{GatewayError, GatewayResult}; use crate::session::SessionClaims; -use crate::error::GatewayResult; +use crate::GatewayState; #[derive(Debug, Deserialize)] pub struct ExplorerQuery { @@ -16,8 +21,12 @@ pub struct ExplorerQuery { pub schema: Option, } +// The shapes below mirror the admin console's `domain.ts` types exactly so the +// JSON deserializes directly on the client. #[derive(Debug, Serialize)] pub struct ExplorerSnapshot { + #[serde(rename = "generatedAt")] + pub generated_at: String, pub databases: Vec, } @@ -30,67 +39,104 @@ pub struct DatabaseInfo { #[derive(Debug, Serialize)] pub struct SchemaInfo { + pub database: String, pub name: String, pub relations: Vec, } #[derive(Debug, Serialize)] pub struct RelationInfo { + pub database: String, + pub schema: String, pub name: String, pub kind: String, - pub schema: String, pub storage: String, pub columns: Vec, + pub description: String, } #[derive(Debug, Serialize)] pub struct ColumnInfo { pub name: String, + #[serde(rename = "type")] pub data_type: String, pub nullable: bool, } -/// Get full explorer snapshot +const SYSTEM_SCHEMAS: [&str; 2] = ["pg_catalog", "information_schema"]; + +/// Build the full catalog snapshot for the explorer tree. pub async fn get_explorer_snapshot( Extension(_claims): Extension, - State(_state): State>, + State(state): State>, ) -> GatewayResult> { - // Placeholder implementation - let snapshot = ExplorerSnapshot { - databases: vec![ - DatabaseInfo { - name: "default".to_string(), - owner: "admin".to_string(), - schemas: vec![ - SchemaInfo { - name: "public".to_string(), - relations: vec![ + let control_plane = state.open_catalog().await.map_err(GatewayError::Internal)?; + let snapshot = control_plane.cluster_snapshot().await; + + let databases = snapshot + .databases + .iter() + .map(|database| { + let schemas = database + .schemas + .iter() + .map(|schema_name| { + let relations = snapshot + .relations + .iter() + .filter(|relation| { + relation.database == database.name && &relation.schema == schema_name + }) + .map(|relation| { + let kind = match relation.kind { + CatalogRelationKind::View => "view", + CatalogRelationKind::Table => "table", + }; + let storage = if SYSTEM_SCHEMAS.contains(&schema_name.as_str()) { + "system" + } else if relation.external_format.is_some() { + "external" + } else { + "managed" + }; RelationInfo { - name: "sample_table".to_string(), - kind: "table".to_string(), - schema: "public".to_string(), - storage: "managed".to_string(), - columns: vec![ - ColumnInfo { - name: "id".to_string(), - data_type: "INTEGER".to_string(), - nullable: false, - }, - ColumnInfo { - name: "name".to_string(), - data_type: "TEXT".to_string(), - nullable: true, - }, - ], - }, - ], - }, - ], - }, - ], - }; - - Ok(Json(snapshot)) + database: relation.database.clone(), + schema: relation.schema.clone(), + name: relation.name.clone(), + kind: kind.to_string(), + storage: storage.to_string(), + description: format!("{kind} in {schema_name}"), + columns: relation + .columns + .iter() + .map(|column| ColumnInfo { + name: column.name.clone(), + data_type: column.data_type.clone(), + nullable: column.nullable, + }) + .collect(), + } + }) + .collect(); + SchemaInfo { + database: database.name.clone(), + name: schema_name.clone(), + relations, + } + }) + .collect(); + DatabaseInfo { + name: database.name.clone(), + owner: database.owner.clone(), + schemas, + } + }) + .collect(); + + Ok(Json(ExplorerSnapshot { + generated_at: format!("v{}", snapshot.catalogue_version), + databases, + })) } /// List databases diff --git a/crates/analyticsdb-gateway/src/routes/health.rs b/crates/analyticsdb-gateway/src/routes/health.rs index 1569c2b..54c215a 100644 --- a/crates/analyticsdb-gateway/src/routes/health.rs +++ b/crates/analyticsdb-gateway/src/routes/health.rs @@ -3,7 +3,6 @@ use axum::{extract::State, Json}; use serde_json::json; -use crate::GatewayState; /// Liveness probe - always returns 200 if the server is running pub async fn liveness() -> Json { diff --git a/crates/analyticsdb-gateway/src/routes/query.rs b/crates/analyticsdb-gateway/src/routes/query.rs index b354b40..5839aa1 100644 --- a/crates/analyticsdb-gateway/src/routes/query.rs +++ b/crates/analyticsdb-gateway/src/routes/query.rs @@ -1,4 +1,8 @@ -//! Query execution routes +//! Query execution route — proxies SQL to the server's pg-wire endpoint so the +//! server's engine is the single source of truth. + +use std::sync::Arc; +use std::time::Instant; use axum::{ extract::{Extension, State}, @@ -6,28 +10,20 @@ use axum::{ }; use serde::{Deserialize, Serialize}; -use crate::GatewayState; +use crate::error::{GatewayError, GatewayResult}; use crate::session::SessionClaims; -use crate::error::GatewayResult; +use crate::GatewayState; #[derive(Debug, Deserialize)] pub struct QueryRequest { pub sql: String, - pub protocol: Option, // "pg" or "flight" -} - -#[derive(Debug, Serialize)] -pub struct QueryResult { - pub query_id: String, - pub statement_type: String, - pub columns: Vec, - pub rows: Vec>, - pub affected_rows: Option, - pub timings: QueryTimings, - pub messages: Vec, + pub protocol: Option, + pub database: Option, + pub schema: Option, } #[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] pub struct QueryTimings { pub queue_ms: u64, pub plan_ms: u64, @@ -38,40 +34,149 @@ pub struct QueryTimings { #[derive(Debug, Serialize)] pub struct QueryMessage { - pub level: String, // "info", "warning", "error" + pub level: String, pub text: String, } -/// Execute a SQL query through the gateway +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct QueryResult { + pub query_id: String, + pub statement_type: String, + pub columns: Vec, + pub rows: Vec>, + #[serde(skip_serializing_if = "Option::is_none")] + pub affected_rows: Option, + pub timings: QueryTimings, + pub messages: Vec, +} + +/// Execute a SQL statement on behalf of the authenticated session by proxying it +/// to the server over pg-wire, authenticated as the session's user. pub async fn execute_query( Extension(claims): Extension, - State(state): State>, + State(state): State>, Json(req): Json, ) -> GatewayResult> { + let sql = req.sql.trim(); + if sql.is_empty() { + return Err(GatewayError::BadRequest("SQL statement is empty".to_string())); + } + + let password = state + .session_store + .password_for(&claims.session_id) + // No cached credential (e.g. gateway restarted) — force re-login. + .ok_or(GatewayError::Unauthorized)?; + + let database = req.database.unwrap_or_else(|| claims.database.clone()); let query_id = format!("gw-{}", uuid::Uuid::new_v4()); + let started = Instant::now(); + + let outcome = crate::proxy::execute_sql( + &state.pg_endpoint(), + &claims.sub, + &password, + &database, + &req.sql, + ) + .await; + let total_ms = started.elapsed().as_millis() as u64; + + match outcome { + Ok(result) => { + let has_columns = !result.columns.is_empty(); + let row_count = result.rows.len(); + let raw_affected = result.affected_rows; + let rows = result + .rows + .into_iter() + .map(|row| { + row.into_iter() + .map(|cell| match cell { + Some(text) => serde_json::Value::String(text), + None => serde_json::Value::Null, + }) + .collect() + }) + .collect(); + let statement_type = classify_sql(&req.sql, has_columns); + // Only surface an affected-row count for non-SELECT statements. + let affected_rows = if has_columns { None } else { raw_affected }; + let messages = vec![QueryMessage { + level: "info".to_string(), + text: result_message(&statement_type, raw_affected, row_count), + }]; + + Ok(Json(QueryResult { + query_id, + statement_type, + columns: result.columns, + rows, + affected_rows, + timings: timings(total_ms), + messages, + })) + } + // Surface execution errors inline so the console renders them in the + // results panel rather than as an HTTP failure. + Err(error) => Ok(Json(QueryResult { + query_id, + statement_type: "unknown".to_string(), + columns: vec![], + rows: vec![], + affected_rows: None, + timings: timings(total_ms), + messages: vec![QueryMessage { + level: "error".to_string(), + text: clean_error(&error.to_string()), + }], + })), + } +} + +fn timings(total_ms: u64) -> QueryTimings { + QueryTimings { + queue_ms: 0, + plan_ms: 0, + execute_ms: total_ms, + fetch_ms: 0, + total_ms, + } +} + +/// Classify a statement from its leading keyword (and whether it returned rows). +fn classify_sql(sql: &str, has_columns: bool) -> String { + let verb = sql + .trim_start() + .split(|c: char| c.is_whitespace() || c == '(') + .next() + .unwrap_or("") + .to_uppercase(); + match verb.as_str() { + "SELECT" | "WITH" | "SHOW" | "TABLE" | "VALUES" => "select".to_string(), + "INSERT" | "UPDATE" | "DELETE" | "COPY" | "MERGE" => "dml".to_string(), + "CREATE" | "DROP" | "ALTER" | "TRUNCATE" | "REINDEX" | "GRANT" | "REVOKE" => { + "ddl".to_string() + } + "EXPLAIN" => "explain".to_string(), + _ if has_columns => "select".to_string(), + _ => "metadata".to_string(), + } +} + +fn result_message(statement_type: &str, affected: Option, row_count: usize) -> String { + match statement_type { + "select" | "explain" => format!("{row_count} row(s) returned."), + "dml" => format!("{} row(s) affected.", affected.unwrap_or(0)), + _ => "Statement executed successfully.".to_string(), + } +} - // In production, this would: - // 1. Connect to AnalyticsDB via PG or Flight SQL protocol - // 2. Execute the query with the user's session context - // 3. Return the results - - // For now, return a placeholder response - Ok(Json(QueryResult { - query_id, - statement_type: "select".to_string(), - columns: vec!["result".to_string()], - rows: vec![vec![serde_json::json!(1)]], - affected_rows: None, - timings: QueryTimings { - queue_ms: 1, - plan_ms: 5, - execute_ms: 10, - fetch_ms: 2, - total_ms: 18, - }, - messages: vec![QueryMessage { - level: "info".to_string(), - text: "Query executed via gateway (prototype)".to_string(), - }], - })) +/// Strips the tokio-postgres `db error: ERROR:` prefix for a cleaner message. +fn clean_error(message: &str) -> String { + message + .trim_start_matches("db error: ") + .trim_start_matches("ERROR: ") + .to_string() } diff --git a/crates/analyticsdb-gateway/src/routes/session.rs b/crates/analyticsdb-gateway/src/routes/session.rs index 21e582a..72e183f 100644 --- a/crates/analyticsdb-gateway/src/routes/session.rs +++ b/crates/analyticsdb-gateway/src/routes/session.rs @@ -3,14 +3,13 @@ use axum::{extract::State, Extension, Json}; use serde_json::json; -use crate::GatewayState; use crate::session::SessionClaims; use crate::error::GatewayResult; /// Get current session info pub async fn get_session( Extension(claims): Extension, - State(state): State>, + State(_state): State>, ) -> GatewayResult> { Ok(Json(json!({ "username": claims.sub, @@ -23,8 +22,8 @@ pub async fn get_session( /// Update session (database, schema) pub async fn update_session( - Extension(claims): Extension, - State(state): State>, + Extension(_claims): Extension, + State(_state): State>, Json(req): Json, ) -> GatewayResult> { // In production, this would update the session in the store diff --git a/crates/analyticsdb-gateway/src/routes/system.rs b/crates/analyticsdb-gateway/src/routes/system.rs index d828c02..11a5ffe 100644 --- a/crates/analyticsdb-gateway/src/routes/system.rs +++ b/crates/analyticsdb-gateway/src/routes/system.rs @@ -7,7 +7,6 @@ use axum::{ use serde::{Deserialize, Serialize}; use serde_json::json; -use crate::GatewayState; use crate::error::GatewayResult; #[derive(Debug, Deserialize)] @@ -28,7 +27,7 @@ pub struct SystemMetrics { /// Get system metrics pub async fn get_metrics( - State(state): State>, + State(_state): State>, ) -> GatewayResult> { // In production, this would read from the query log and system tables // For now, return placeholder metrics @@ -44,10 +43,10 @@ pub async fn get_metrics( /// Get query log pub async fn get_query_log( Query(query): Query, - State(state): State>, + State(_state): State>, ) -> GatewayResult>> { let limit = query.limit.unwrap_or(100); - let offset = query.offset.unwrap_or(0); + let _offset = query.offset.unwrap_or(0); // In production, query system.query_log table // For now, return placeholder data @@ -73,7 +72,7 @@ pub async fn get_query_log( /// Get audit log pub async fn get_audit_log( Query(query): Query, - State(state): State>, + State(_state): State>, ) -> GatewayResult>> { let limit = query.limit.unwrap_or(100); diff --git a/crates/analyticsdb-gateway/src/session.rs b/crates/analyticsdb-gateway/src/session.rs index 35ac77b..15ab1ad 100644 --- a/crates/analyticsdb-gateway/src/session.rs +++ b/crates/analyticsdb-gateway/src/session.rs @@ -1,6 +1,7 @@ //! Session management for the gateway -use std::sync::Arc; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use chrono::{DateTime, Utc}; @@ -23,29 +24,42 @@ pub struct SessionClaims { pub session_id: String, // unique session ID } -/// Session store for managing active sessions +/// Session store for managing active sessions. +/// +/// In addition to issuing/validating JWTs, it caches each session's password in +/// memory so the gateway can open authenticated pg-wire connections to the +/// server on behalf of the user. Passwords are never written to the JWT or to +/// disk; if the gateway restarts, the cache is empty and the user must sign in +/// again (their token is rejected with 401, prompting re-login). pub struct SessionStore { config: Arc, + credentials: Mutex>, } impl SessionStore { pub fn new(_session_timeout_seconds: u64) -> Self { Self { config: Arc::new(crate::config::GatewayConfig::default()), + credentials: Mutex::new(HashMap::new()), } } pub fn with_config(config: Arc) -> Self { - Self { config } + Self { + config, + credentials: Mutex::new(HashMap::new()), + } } - /// Create a new session and return a JWT token - pub fn create_session( + /// Create a new session, cache the password for pg-wire proxying, and return + /// a JWT token. + pub fn create_session_with_password( &self, username: &str, role: &str, database: &str, schema: &str, + password: &str, ) -> GatewayResult { let now = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -62,7 +76,7 @@ impl SessionStore { schema: schema.to_string(), exp, iat: now, - session_id, + session_id: session_id.clone(), }; let token = encode( @@ -71,9 +85,39 @@ impl SessionStore { &EncodingKey::from_secret(self.config.jwt_secret.as_bytes()), )?; + if let Ok(mut creds) = self.credentials.lock() { + creds.insert(session_id, password.to_string()); + } + Ok(token) } + /// Returns the cached password for a session, if present. + pub fn password_for(&self, session_id: &str) -> Option { + self.credentials + .lock() + .ok() + .and_then(|creds| creds.get(session_id).cloned()) + } + + /// Forgets a session's cached password (called on logout). + pub fn forget(&self, session_id: &str) { + if let Ok(mut creds) = self.credentials.lock() { + creds.remove(session_id); + } + } + + /// Create a new session and return a JWT token (no cached password). + pub fn create_session( + &self, + username: &str, + role: &str, + database: &str, + schema: &str, + ) -> GatewayResult { + self.create_session_with_password(username, role, database, schema, "") + } + /// Validate and decode a JWT token pub fn validate_token(&self, token: &str) -> GatewayResult { let validation = Validation::default(); diff --git a/crates/analyticsdb-protocol/src/lib.rs b/crates/analyticsdb-protocol/src/lib.rs index bd73190..86aac1f 100644 --- a/crates/analyticsdb-protocol/src/lib.rs +++ b/crates/analyticsdb-protocol/src/lib.rs @@ -297,27 +297,31 @@ pub async fn serve_flight_sql_with_label( label: &'static str, ) -> anyhow::Result<()> { let control_plane = engine.control_plane(); - // Resolve the JWT signing secret: use the configured value if present, - // or generate a random ephemeral key and warn that sessions won't survive restarts. + // Resolve the JWT signing secret: use environment variable if present, + // otherwise the configured value, and fallback to an ephemeral key. let jwt_secret = { - let config = control_plane.cluster_config().await; - match config.and_then(|c| c.jwt_secret) { - Some(secret) => secret, - None => { - let random_bytes: [u8; 32] = rand::random(); - let hex_secret = random_bytes - .iter() - .fold(String::with_capacity(64), |mut acc, b| { - use std::fmt::Write as _; - let _ = write!(acc, "{b:02x}"); - acc - }); - warn!( - "{}: jwt_secret not configured — using ephemeral key. \ - Flight SQL sessions will not survive a server restart.", - label - ); - hex_secret + if let Ok(secret) = std::env::var("ANALYTICSDB_JWT_SECRET") { + secret + } else { + let config = control_plane.cluster_config().await; + match config.and_then(|c| c.jwt_secret) { + Some(secret) => secret, + None => { + let random_bytes: [u8; 32] = rand::random(); + let hex_secret = random_bytes + .iter() + .fold(String::with_capacity(64), |mut acc, b| { + use std::fmt::Write as _; + let _ = write!(acc, "{b:02x}"); + acc + }); + warn!( + "{}: jwt_secret not configured — using ephemeral key. \ + Flight SQL sessions will not survive a server restart.", + label + ); + hex_secret + } } } }; diff --git a/crates/analyticsdb-server/Cargo.toml b/crates/analyticsdb-server/Cargo.toml index 9fbdccf..0152bc8 100644 --- a/crates/analyticsdb-server/Cargo.toml +++ b/crates/analyticsdb-server/Cargo.toml @@ -27,3 +27,4 @@ hyper = "1.0" tower = "0.5" http = "1.0" futures = "0.3" +rpassword = "7" diff --git a/crates/analyticsdb-server/src/config.rs b/crates/analyticsdb-server/src/config.rs index 88ad34b..440575c 100644 --- a/crates/analyticsdb-server/src/config.rs +++ b/crates/analyticsdb-server/src/config.rs @@ -6,6 +6,7 @@ use std::path::PathBuf; /// Centralized configuration for AnalyticsDB server. /// This is the single source of truth for all configuration. #[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(default)] pub struct Config { /// Node role: control, compute, storage, gateway pub role: String, @@ -44,12 +45,15 @@ pub struct Config { pub storage_root: Option, /// TLS certificate path + #[serde(alias = "tls_cert_path")] pub tls_cert: Option, /// TLS key path + #[serde(alias = "tls_key_path")] pub tls_key: Option, /// TLS CA certificate path + #[serde(alias = "tls_ca_cert_path")] pub tls_ca_cert: Option, /// TLS domain for verification @@ -57,6 +61,21 @@ pub struct Config { /// Disable TLS verification (insecure) pub tls_insecure: bool, + + /// HS256 secret used to sign Flight SQL JWT bearer tokens + pub jwt_secret: Option, + + /// Base PostgreSQL port + #[serde(alias = "base_postgres_port")] + pub base_postgres_port: Option, + + /// Base Flight SQL port + #[serde(alias = "base_flight_sql_port")] + pub base_flight_sql_port: Option, + + /// Base Node port + #[serde(alias = "base_node_port")] + pub base_node_port: Option, } impl Default for Config { @@ -79,10 +98,27 @@ impl Default for Config { tls_ca_cert: None, tls_domain: None, tls_insecure: false, + jwt_secret: None, + base_postgres_port: None, + base_flight_sql_port: None, + base_node_port: None, } } } +/// Default locations to look for the cluster config when `--cluster-config` is +/// not given, in priority order. Config lives in a `config/` directory by +/// convention; the repo-root path is kept as a fallback for older layouts. +pub const DEFAULT_CONFIG_PATHS: [&str; 2] = ["config/cluster-config.json", "cluster-config.json"]; + +/// Returns the first existing default config path, if any. +pub fn discover_config_path() -> Option { + DEFAULT_CONFIG_PATHS + .iter() + .find(|path| std::path::Path::new(path).exists()) + .map(|path| path.to_string()) +} + impl Config { /// Load configuration from a file. pub fn from_file(path: &str) -> Result { diff --git a/crates/analyticsdb-server/src/main.rs b/crates/analyticsdb-server/src/main.rs index 0c54d55..ef4701d 100644 --- a/crates/analyticsdb-server/src/main.rs +++ b/crates/analyticsdb-server/src/main.rs @@ -121,6 +121,8 @@ impl tower::Service for InsecureConnector { #[tokio::main] async fn main() { + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let filter = tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")); @@ -181,16 +183,167 @@ fn merge_config_with_cli(config: &mut Config, cli: &Cli) { if cli.join.is_some() { config.join = cli.join.clone(); } + if cli.jwt_secret.is_some() { + config.jwt_secret = cli.jwt_secret.clone(); + } // cluster_config is used to load the config file, so we don't set it here } +/// Prints a generated credential inside a highlighted, one-time-only box. +fn print_credential_box(title: &str, user: &str, password: &str) { + let border = "═".repeat(64); + println!("\n\x1b[1;33m╔{border}╗\x1b[0m"); + println!("\x1b[1;33m║\x1b[0m \x1b[1;32m{title}\x1b[0m"); + println!("\x1b[1;33m║\x1b[0m"); + println!("\x1b[1;33m║\x1b[0m username: \x1b[1;36m{user}\x1b[0m"); + println!("\x1b[1;33m║\x1b[0m password: \x1b[1;36m{password}\x1b[0m"); + println!("\x1b[1;33m║\x1b[0m"); + println!( + "\x1b[1;33m║\x1b[0m \x1b[1;31mStore this now — it is shown ONCE and cannot be recovered.\x1b[0m" + ); + println!("\x1b[1;33m╚{border}╝\x1b[0m\n"); +} + +/// `--init-cluster`: set up a fresh catalog and primary administrator account. +/// If an existing catalog with accounts is detected, the operator is warned and +/// must authenticate with an administrator password before everything is flushed. +async fn init_cluster(catalog_path: &str, force: bool) -> Result<()> { + use analyticsdb_control::{ControlPlane, PRIMARY_ADMIN_USER}; + + let has_accounts = ControlPlane::catalog_has_accounts(catalog_path) + .await + .context("Failed to inspect existing catalog")?; + + if has_accounts { + println!( + "\n\x1b[1;31m⚠ WARNING: An existing catalog was found at '{catalog_path}'.\x1b[0m" + ); + println!( + "\x1b[1;31m Re-initializing will PERMANENTLY DELETE all databases, tables,\x1b[0m" + ); + println!("\x1b[1;31m users, and groups in this catalog. This cannot be undone.\x1b[0m\n"); + + if force { + println!( + "\x1b[1;31m --force supplied: skipping authentication and flushing now.\x1b[0m\n" + ); + } else { + let existing = ControlPlane::from_catalog_path(catalog_path) + .await + .context("Failed to open existing catalog for authentication")?; + + let user = prompt_line("Administrator username to confirm flush: ")?; + let password = prompt_password("Administrator password: ")?; + + existing + .verify_admin_credentials(user.trim(), &password) + .await + .context("Administrator authentication failed — aborting without changes")?; + + println!( + "\n\x1b[1;33mAuthentication accepted. Flushing and re-initializing...\x1b[0m" + ); + } + } else { + println!("\n\x1b[1;32mNo existing catalog found. Initializing a fresh cluster...\x1b[0m"); + } + + let (_control_plane, password) = ControlPlane::init_fresh_cluster(catalog_path) + .await + .context("Failed to initialize cluster")?; + + println!("\x1b[1;32m✓ Catalog initialized at '{catalog_path}'.\x1b[0m"); + print_credential_box( + "Primary administrator account created", + PRIMARY_ADMIN_USER, + &password, + ); + println!("Start the server normally to begin serving requests."); + Ok(()) +} + +/// `--reset-admin-password`: reset the primary admin password using another +/// Administrators-group member's credentials. +async fn reset_admin_password(catalog_path: &str) -> Result<()> { + use analyticsdb_control::{ControlPlane, PRIMARY_ADMIN_USER}; + + if !std::path::Path::new(catalog_path).exists() { + anyhow::bail!( + "No catalog found at '{catalog_path}'. Run with --init-cluster first." + ); + } + + let control_plane = ControlPlane::from_catalog_path(catalog_path) + .await + .context("Failed to open catalog")?; + + println!( + "\nAuthenticate with another '{}'-group member to reset the '{}' password.", + analyticsdb_control::ADMINISTRATORS_GROUP, + PRIMARY_ADMIN_USER + ); + let user = prompt_line("Administrator username: ")?; + let password = prompt_password("Administrator password: ")?; + + let new_password = control_plane + .reset_admin_password(user.trim(), &password) + .await + .context("Failed to reset administrator password")?; + + println!("\n\x1b[1;32m✓ Password reset successfully.\x1b[0m"); + print_credential_box( + "New primary administrator password", + PRIMARY_ADMIN_USER, + &new_password, + ); + Ok(()) +} + +/// Reads a line of plain (echoed) input from stdin. +fn prompt_line(prompt: &str) -> Result { + use std::io::Write; + print!("{prompt}"); + std::io::stdout().flush()?; + let mut line = String::new(); + std::io::stdin() + .read_line(&mut line) + .context("Failed to read input")?; + Ok(line.trim_end_matches(['\r', '\n']).to_string()) +} + +/// Whether a `storage_root` refers to the local filesystem (a plain path or a +/// `file://` URI) rather than a remote object store (`s3://`, `gs://`, `az://`). +fn is_local_storage_root(root: &str) -> bool { + let lower = root.to_ascii_lowercase(); + lower.starts_with("file://") || !lower.contains("://") +} + +/// Reads a password from the terminal without echoing it. When no interactive +/// terminal is attached (e.g. input is piped, as in automated provisioning), +/// falls back to reading a plain line from stdin. +fn prompt_password(prompt: &str) -> Result { + match rpassword::prompt_password(prompt) { + Ok(password) => Ok(password), + Err(_) => { + warn!("No interactive terminal detected; reading password from stdin (input may be echoed)."); + prompt_line(prompt) + } + } +} + async fn run() -> Result<()> { let cli = Cli::parse(); println!("{}", BANNER); - // Load config from file if provided, otherwise use defaults - let mut config: Config = if let Some(config_path) = &cli.cluster_config { + // Load config: explicit --cluster-config, else a discovered default in the + // `config/` directory, else built-in defaults. + let discovered = if cli.cluster_config.is_none() { + config::discover_config_path() + } else { + None + }; + let mut config: Config = if let Some(config_path) = cli.cluster_config.as_ref().or(discovered.as_ref()) { if !std::path::Path::new(config_path).exists() { anyhow::bail!("Cluster configuration file not found: {}", config_path); } @@ -198,7 +351,7 @@ async fn run() -> Result<()> { Config::from_file(config_path)? } else { if cli.role == NodeRole::Control && cli.join.is_none() && !cli.init_cluster { - warn!("\x1b[1;31mNo cluster configuration provided for control node. Using bootstrap defaults.\x1b[0m"); + warn!("\x1b[1;31mNo cluster configuration found (looked in config/). Using bootstrap defaults.\x1b[0m"); } Config::default() }; @@ -206,9 +359,42 @@ async fn run() -> Result<()> { // Merge CLI arguments (CLI takes precedence over config file) merge_config_with_cli(&mut config, &cli); + // Environment variable overrides (takes precedence over config file, but not CLI) + if cli.jwt_secret.is_none() { + if let Ok(env_secret) = std::env::var("ANALYTICSDB_JWT_SECRET") { + config.jwt_secret = Some(env_secret); + } + } + + // Fallback to base ports from cluster configuration if addresses are not explicitly set or are default + if config.postgres_addr.is_none() || config.postgres_addr.as_deref() == Some("127.0.0.1:5432") { + if let Some(port) = config.base_postgres_port { + config.postgres_addr = Some(format!("127.0.0.1:{}", port)); + } + } + if config.flight_sql_addr.is_none() || config.flight_sql_addr.as_deref() == Some("127.0.0.1:8815") { + if let Some(port) = config.base_flight_sql_port { + config.flight_sql_addr = Some(format!("127.0.0.1:{}", port)); + } + } + if config.node_addr.is_none() || config.node_addr.as_deref() == Some("127.0.0.1:8816") { + if let Some(port) = config.base_node_port { + config.node_addr = Some(format!("127.0.0.1:{}", port)); + } + } + // Validate the merged configuration config.validate().context("Configuration validation failed")?; + // One-shot administrative commands. These run before normal startup and + // exit the process when complete. + if cli.reset_admin_password { + return reset_admin_password(&config.catalog_path).await; + } + if cli.init_cluster { + return init_cluster(&config.catalog_path, cli.force).await; + } + // If joining a cluster, request configuration from the coordinator if let Some(coordinator_endpoint) = &config.join { let is_https = coordinator_endpoint.starts_with("https"); @@ -224,7 +410,6 @@ async fn run() -> Result<()> { ); let tls_cert = config.tls_cert.clone(); - let tls_key = config.tls_key.clone(); let tls_ca_cert = config.tls_ca_cert.clone(); let tls_domain = config.tls_domain.clone(); @@ -315,6 +500,20 @@ async fn run() -> Result<()> { config.node_addr = Some(format!("0.0.0.0:{}", res.node_port)); config.catalog_path = res.config.catalog_path.clone(); + // Adjust the admin port with the same offset as other ports to avoid collision when running locally + if let Some(admin_addr) = &config.admin_addr { + if let Ok(addr) = admin_addr.parse::() { + let new_port = addr.port() + res.config.next_available_port_offset; + config.admin_addr = Some(format!("{}:{}", addr.ip(), new_port)); + } else if let Some(pos) = admin_addr.rfind(':') { + let (host, port_str) = admin_addr.split_at(pos); + if let Ok(port) = port_str[1..].parse::() { + let new_port = port + res.config.next_available_port_offset; + config.admin_addr = Some(format!("{}:{}", host, new_port)); + } + } + } + // Inherit TLS cert/key from the JoinResponse so the compute node's // flight server also serves with TLS. if config.tls_cert.is_none() { @@ -325,6 +524,25 @@ async fn run() -> Result<()> { } } + // Choose the local data directory (system query/audit logs + the local + // managed-table fallback) BEFORE constructing the engine, since the engine + // resolves its log roots at construction. Defaults to `data/`; a local + // `storage_root` from config overrides it. Remote roots (s3://, gs://, …) + // keep system logs in the local `data/` directory. Joining nodes inherit + // storage from the coordinator and don't set this. + if config.join.is_none() { + let data_dir = match config.storage_root.as_deref() { + Some(root) if is_local_storage_root(root) => { + root.strip_prefix("file://").unwrap_or(root).to_string() + } + _ => "data".to_string(), + }; + info!("Managed data directory: {}", data_dir); + // Safety: set during single-threaded startup before the engine (and any + // other thread) reads it. + unsafe { std::env::set_var("ANALYTICSDB_DATA_DIR", &data_dir) }; + } + let engine = Arc::new(PrototypeEngine::from_catalog_path(&config.catalog_path).await?); let tls_cert_path = config.tls_cert.clone(); @@ -337,6 +555,25 @@ async fn run() -> Result<()> { ) .await?; + if let Some(jwt_secret) = &config.jwt_secret { + engine + .control_plane() + .set_jwt_secret(Some(jwt_secret.clone())) + .await?; + } + + // Propagate an explicit `storage_root` (local or remote, e.g. s3://) from + // config into the catalog so managed table URIs use it. When unset, table + // data falls back to the local data directory configured above. + if config.join.is_none() { + if let Some(storage_root) = &config.storage_root { + engine + .control_plane() + .set_storage_root(Some(storage_root.clone())) + .await?; + } + } + let node_id = config.node_id.clone().unwrap_or_else(|| "standalone".to_string()); // Create a root span with node_id - all child spans will inherit this field @@ -589,8 +826,22 @@ struct Cli { node_id: Option, #[arg(long, value_enum, default_value = "control")] role: NodeRole, + /// Initialize the system for first use: create the catalog and a primary + /// `analyticsdb_admin` account (random password, printed once) in the + /// `Administrators` group. If an existing catalog/users are found, you are + /// warned and must enter an administrator password before they are flushed. #[arg(long)] init_cluster: bool, + /// Reset the `analyticsdb_admin` password. Authenticates with another + /// `Administrators`-group member's credentials and prints a new random + /// password once. Use this if the primary admin password is lost. + #[arg(long)] + reset_admin_password: bool, + /// With `--init-cluster`, skip the administrator-password confirmation and + /// flush the existing catalog unconditionally. Recovery of last resort when + /// all administrator credentials are lost — this DESTROYS all data. + #[arg(long)] + force: bool, #[arg(long)] join: Option, #[arg(long)] @@ -631,6 +882,8 @@ struct Cli { tls_domain: Option, #[arg(long)] tls_insecure: bool, + #[arg(long)] + jwt_secret: Option, } #[cfg(test)] diff --git a/web/admin-console/src/adminClient.ts b/web/admin-console/src/adminClient.ts index 9694a69..f71bf56 100644 --- a/web/admin-console/src/adminClient.ts +++ b/web/admin-console/src/adminClient.ts @@ -17,6 +17,7 @@ export interface ClusterConfig { tls_key_path?: string | null; next_available_port_offset: number; query_log?: QueryLogConfig; + jwt_secret?: string | null; } export interface ClusterConfigEnvelope { diff --git a/web/admin-console/src/clusterConfigForm.test.ts b/web/admin-console/src/clusterConfigForm.test.ts index 5108640..36b5df5 100644 --- a/web/admin-console/src/clusterConfigForm.test.ts +++ b/web/admin-console/src/clusterConfigForm.test.ts @@ -95,6 +95,30 @@ describe("buildSavePayload", () => { expect(payload.tls_key_path).toBeNull(); }); + it("sends optional jwt_secret as null when the user clears it", () => { + const customConfig: ClusterConfig = { + ...MINIMAL_CONFIG, + jwt_secret: "secret-key", + }; + const draft = withDisplayDefaults(customConfig); + draft.jwt_secret = ""; + const payload = buildSavePayload(draft, customConfig); + expect(payload.jwt_secret).toBeNull(); + }); + + it("does not introduce jwt_secret when neither file nor draft had a value", () => { + const sparse: ClusterConfig = { + base_postgres_port: 5432, + base_flight_sql_port: 50051, + catalog_path: "cluster-catalog.db", + next_available_port_offset: 0, + }; + const draft = withDisplayDefaults(sparse); + const payload = buildSavePayload(draft, sparse); + expect(payload.jwt_secret).toBeUndefined(); + expect(configsEqual(sparse, payload)).toBe(true); + }); + it("does not introduce TLS keys when neither file nor draft had a value", () => { const sparse: ClusterConfig = { base_postgres_port: 5432, diff --git a/web/admin-console/src/clusterConfigForm.ts b/web/admin-console/src/clusterConfigForm.ts index 7080747..69edaee 100644 --- a/web/admin-console/src/clusterConfigForm.ts +++ b/web/admin-console/src/clusterConfigForm.ts @@ -55,6 +55,11 @@ export function buildSavePayload( payload.tls_key_path = tlsKey; } + const jwtSecret = emptyToNull(draft.jwt_secret); + if (jwtSecret !== null || baseline.jwt_secret !== undefined) { + payload.jwt_secret = jwtSecret; + } + const baselineHadQueryLog = baseline.query_log !== undefined; const draftQueryLog = draft.query_log; if (baselineHadQueryLog) { diff --git a/web/admin-console/src/icons.ts b/web/admin-console/src/icons.ts index ed37b5f..5520f8e 100644 --- a/web/admin-console/src/icons.ts +++ b/web/admin-console/src/icons.ts @@ -10,7 +10,9 @@ export type IconName = | "table" | "view" | "chevron-right" + | "folder" | "logout" + | "log-in" | "circle-check" | "alert-triangle" | "x-circle" @@ -30,7 +32,9 @@ const ICON_PATHS: Record = { table: ``, view: ``, "chevron-right": ``, + folder: ``, logout: ``, + "log-in": ``, "circle-check": ``, "alert-triangle": ``, "x-circle": ``, diff --git a/web/admin-console/src/liveClient.ts b/web/admin-console/src/liveClient.ts index 20f093c..998362f 100644 --- a/web/admin-console/src/liveClient.ts +++ b/web/admin-console/src/liveClient.ts @@ -18,6 +18,20 @@ interface LoginRequest { password: string; } +export interface AdminUser { + readonly name: string; + readonly is_admin: boolean; + readonly groups: readonly string[]; + readonly password_version?: number; + readonly password_rotated_at_epoch_ms?: number | null; +} + +export interface AdminGroup { + readonly name: string; + readonly members: readonly string[]; + readonly member_count: number; +} + interface LoginResponse { token: string; session: { @@ -31,10 +45,16 @@ interface LoginResponse { export class LiveConsoleClient implements AnalyticsConsoleClient { private token: string | null = null; + private user: string | null = null; + private role: string | null = null; + /** Invoked when a request is rejected as unauthorized (expired/invalid token). */ + onSessionExpired: (() => void) | null = null; constructor() { - // Try to load token from localStorage + // Restore any persisted session from localStorage. this.token = localStorage.getItem("analyticsdb_token"); + this.user = localStorage.getItem("analyticsdb_user"); + this.role = localStorage.getItem("analyticsdb_role"); } setToken(token: string): void { @@ -44,13 +64,31 @@ export class LiveConsoleClient implements AnalyticsConsoleClient { clearToken(): void { this.token = null; + this.user = null; + this.role = null; localStorage.removeItem("analyticsdb_token"); + localStorage.removeItem("analyticsdb_user"); + localStorage.removeItem("analyticsdb_role"); } isAuthenticated(): boolean { return this.token !== null; } + /** The username of the signed-in account, or null when signed out. */ + currentUser(): string | null { + return this.user; + } + + /** The session role (e.g. "admin"), or null when signed out. */ + currentRole(): string | null { + return this.role; + } + + isAdmin(): boolean { + return this.role === "admin"; + } + private async request(path: string, options?: RequestInit): Promise { const headers: Record = { "Content-Type": "application/json", @@ -67,6 +105,12 @@ export class LiveConsoleClient implements AnalyticsConsoleClient { }); if (!response.ok) { + // A 401 on an authenticated request means the session is no longer valid; + // clear it and notify the app so it can route back to the login screen. + if (response.status === 401 && path !== "/auth/login") { + this.clearToken(); + this.onSessionExpired?.(); + } const error = await response.json().catch(() => ({ error: `HTTP ${response.status}: ${response.statusText}`, })); @@ -83,6 +127,12 @@ export class LiveConsoleClient implements AnalyticsConsoleClient { }); this.setToken(response.token); + this.user = response.session?.sub ?? username; + this.role = response.session?.role ?? null; + localStorage.setItem("analyticsdb_user", this.user); + if (this.role) { + localStorage.setItem("analyticsdb_role", this.role); + } return response; } @@ -143,6 +193,83 @@ export class LiveConsoleClient implements AnalyticsConsoleClient { return this.request("/admin/users"); } + // --- Admin: users --- + + async listAdminUsers(): Promise { + return this.request("/admin/users"); + } + + async createUser( + name: string, + password: string, + groups: string[] = [], + ): Promise<{ message: string }> { + return this.request("/admin/users", { + method: "POST", + body: JSON.stringify({ name, password, groups }), + }); + } + + async dropUser(name: string): Promise<{ message: string }> { + return this.request(`/admin/users/${encodeURIComponent(name)}`, { + method: "DELETE", + }); + } + + /** + * Resets a user's password. Pass an explicit `password` to set it directly, + * or omit it to have the server generate a strong random one. When generated, + * the plaintext is returned in `password`. + */ + async resetUserPassword( + name: string, + password?: string, + ): Promise<{ name: string; message: string; generated: boolean; password?: string }> { + return this.request( + `/admin/users/${encodeURIComponent(name)}/reset-password`, + { method: "POST", body: JSON.stringify({ password: password ?? null }) }, + ); + } + + // --- Admin: groups --- + + async listGroups(): Promise { + return this.request("/admin/groups"); + } + + async createGroup(name: string): Promise<{ message: string }> { + return this.request("/admin/groups", { + method: "POST", + body: JSON.stringify({ name }), + }); + } + + async dropGroup(name: string): Promise<{ message: string }> { + return this.request(`/admin/groups/${encodeURIComponent(name)}`, { + method: "DELETE", + }); + } + + async addGroupMember( + group: string, + user: string, + ): Promise<{ message: string }> { + return this.request( + `/admin/groups/${encodeURIComponent(group)}/members`, + { method: "POST", body: JSON.stringify({ user }) }, + ); + } + + async removeGroupMember( + group: string, + user: string, + ): Promise<{ message: string }> { + return this.request( + `/admin/groups/${encodeURIComponent(group)}/members/${encodeURIComponent(user)}`, + { method: "DELETE" }, + ); + } + async getSystemMetrics(): Promise<{ query_throughput_per_second: number; avg_latency_ms: number; diff --git a/web/admin-console/src/main.ts b/web/admin-console/src/main.ts index 403ce4a..074bff7 100644 --- a/web/admin-console/src/main.ts +++ b/web/admin-console/src/main.ts @@ -12,167 +12,242 @@ interface RouteDefinition { readonly label: string; readonly icon: IconName; readonly mount: (container: HTMLElement) => void; - readonly requiresAuth?: boolean; + /** Restrict this route to administrators (Administrators-group members). */ + readonly adminOnly?: boolean; } const ROUTES: readonly RouteDefinition[] = [ { id: "query", label: "SQL Query", icon: "database", mount: mountQueryView }, - { id: "users", label: "Users", icon: "users", mount: mountUsersView, requiresAuth: true }, - { id: "groups", label: "Groups", icon: "user-group", mount: mountGroupsView, requiresAuth: true }, + { id: "users", label: "Users", icon: "users", mount: mountUsersView, adminOnly: true }, + { id: "groups", label: "Groups", icon: "user-group", mount: mountGroupsView, adminOnly: true }, { id: "settings", label: "System Settings", icon: "settings", mount: mountSettingsView }, { id: "system", label: "System Information", icon: "info", mount: mountSystemView }, ]; const DEFAULT_ROUTE_ID = "query"; -const LOGIN_ROUTE_ID = "login"; +const GATEWAY_LABEL = import.meta.env.VITE_GATEWAY_URL ?? "http://localhost:8080"; -function isAuthenticated(): boolean { - return liveClient.isAuthenticated(); +const app = document.querySelector("#app"); +if (!app) { + throw new Error("Missing #app root element"); } +const root: HTMLDivElement = app; -function renderLoginView(container: HTMLElement): void { - container.innerHTML = ` -