diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml index 9eb3f05..90b5a3e 100644 --- a/.github/workflows/docker.yml +++ b/.github/workflows/docker.yml @@ -1,3 +1,5 @@ +# Builds the archive router, which is on the arrowsquid branch: run it there. +# It stays on main because GitHub offers workflow_dispatch only for workflows on the default branch. name: docker on: workflow_dispatch: # manually run diff --git a/Cargo.lock b/Cargo.lock index 262b600..0136f42 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -281,333 +281,6 @@ version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" -[[package]] -name = "aws-config" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "56a636c44c77fa18bdba56126a34d30cfe5538fe88f7d34988fa731fee143ddd" -dependencies = [ - "aws-http", - "aws-sdk-sso", - "aws-sdk-sts", - "aws-smithy-async", - "aws-smithy-client", - "aws-smithy-http", - "aws-smithy-http-tower", - "aws-smithy-json", - "aws-smithy-types", - "aws-types", - "bytes 1.12.1", - "hex", - "http 0.2.12", - "hyper 0.14.32", - "ring 0.16.20", - "time", - "tokio 1.53.1", - "tower", - "tracing", - "zeroize", -] - -[[package]] -name = "aws-endpoint" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ca8f374874f6459aaa88dc861d7f5d834ca1ff97668eae190e97266b5f6c3fb" -dependencies = [ - "aws-smithy-http", - "aws-smithy-types", - "aws-types", - "http 0.2.12", - "regex", - "tracing", -] - -[[package]] -name = "aws-http" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "78d41e19e779b73463f5f0c21b3aacc995f4ba783ab13a7ae9f5dfb159a551b4" -dependencies = [ - "aws-smithy-http", - "aws-smithy-types", - "aws-types", - "bytes 1.12.1", - "http 0.2.12", - "http-body 0.4.6", - "lazy_static", - "percent-encoding", - "pin-project-lite", - "tracing", -] - -[[package]] -name = "aws-sdk-s3" -version = "0.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9f08665c8e03aca8cb092ef01e617436ebfa977fddc1240e1b062488ab5d48a" -dependencies = [ - "aws-endpoint", - "aws-http", - "aws-sig-auth", - "aws-sigv4", - "aws-smithy-async", - "aws-smithy-checksums", - "aws-smithy-client", - "aws-smithy-eventstream", - "aws-smithy-http", - "aws-smithy-http-tower", - "aws-smithy-types", - "aws-smithy-xml", - "aws-types", - "bytes 1.12.1", - "bytes-utils", - "http 0.2.12", - "http-body 0.4.6", - "tokio-stream", - "tower", - "tracing", -] - -[[package]] -name = "aws-sdk-sso" -version = "0.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "86dcb1cb71aa8763b327542ead410424515cff0cde5b753eedd2917e09c63734" -dependencies = [ - "aws-endpoint", - "aws-http", - "aws-sig-auth", - "aws-smithy-async", - "aws-smithy-client", - "aws-smithy-http", - "aws-smithy-http-tower", - "aws-smithy-json", - "aws-smithy-types", - "aws-types", - "bytes 1.12.1", - "http 0.2.12", - "tokio-stream", - "tower", -] - -[[package]] -name = "aws-sdk-sts" -version = "0.21.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fdfcf584297c666f6b472d5368a78de3bc714b6e0a53d7fbf76c3e347c292ab1" -dependencies = [ - "aws-endpoint", - "aws-http", - "aws-sig-auth", - "aws-smithy-async", - "aws-smithy-client", - "aws-smithy-http", - "aws-smithy-http-tower", - "aws-smithy-query", - "aws-smithy-types", - "aws-smithy-xml", - "aws-types", - "bytes 1.12.1", - "http 0.2.12", - "tower", -] - -[[package]] -name = "aws-sig-auth" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12cbe7b2be9e185c1fbce27fc9c41c66b195b32d89aa099f98768d9544221308" -dependencies = [ - "aws-sigv4", - "aws-smithy-eventstream", - "aws-smithy-http", - "aws-types", - "http 0.2.12", - "tracing", -] - -[[package]] -name = "aws-sigv4" -version = "0.51.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c0b2658d2cb66dbf02f0e8dee80810ef1e0ca3530ede463e0ef994c301087d1" -dependencies = [ - "aws-smithy-eventstream", - "aws-smithy-http", - "bytes 1.12.1", - "form_urlencoded", - "hex", - "http 0.2.12", - "once_cell", - "percent-encoding", - "regex", - "ring 0.16.20", - "time", - "tracing", -] - -[[package]] -name = "aws-smithy-async" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b3442b4c5d3fc39891a2e5e625735fba6b24694887d49c6518460fde98247a9" -dependencies = [ - "futures-util", - "pin-project-lite", - "tokio 1.53.1", - "tokio-stream", -] - -[[package]] -name = "aws-smithy-checksums" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cc227e36e346f45298288359f37123e1a92628d1cec6b11b5eb335553278bd9e" -dependencies = [ - "aws-smithy-http", - "aws-smithy-types", - "bytes 1.12.1", - "crc32c", - "crc32fast", - "hex", - "http 0.2.12", - "http-body 0.4.6", - "md-5", - "pin-project-lite", - "sha1", - "sha2", - "tracing", -] - -[[package]] -name = "aws-smithy-client" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff28d553714f8f54cd921227934fc13a536a1c03f106e56b362fd57e16d450ad" -dependencies = [ - "aws-smithy-async", - "aws-smithy-http", - "aws-smithy-http-tower", - "aws-smithy-types", - "bytes 1.12.1", - "fastrand 1.9.0", - "http 0.2.12", - "http-body 0.4.6", - "hyper 0.14.32", - "hyper-rustls 0.23.2", - "lazy_static", - "pin-project-lite", - "tokio 1.53.1", - "tower", - "tracing", -] - -[[package]] -name = "aws-smithy-eventstream" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7ea0df7161ce65b5c8ca6eb709a1a907376fa18226976e41c748ce02ccccf24" -dependencies = [ - "aws-smithy-types", - "bytes 1.12.1", - "crc32fast", -] - -[[package]] -name = "aws-smithy-http" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bf58ed4fefa61dbf038e5421a521cbc2c448ef69deff0ab1d915d8a10eda5664" -dependencies = [ - "aws-smithy-eventstream", - "aws-smithy-types", - "bytes 1.12.1", - "bytes-utils", - "futures-core", - "http 0.2.12", - "http-body 0.4.6", - "hyper 0.14.32", - "once_cell", - "percent-encoding", - "pin-project-lite", - "pin-utils", - "tokio 1.53.1", - "tokio-util", - "tracing", -] - -[[package]] -name = "aws-smithy-http-tower" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "20c96d7bd35e7cf96aca1134b2f81b1b59ffe493f7c6539c051791cbbf7a42d3" -dependencies = [ - "aws-smithy-http", - "bytes 1.12.1", - "http 0.2.12", - "http-body 0.4.6", - "pin-project-lite", - "tower", - "tracing", -] - -[[package]] -name = "aws-smithy-json" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8324ba98c8a94187723cc16c37aefa09504646ee65c3d2c3af495bab5ea701b" -dependencies = [ - "aws-smithy-types", -] - -[[package]] -name = "aws-smithy-query" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "83834ed2ff69ea6f6657baf205267dc2c0abe940703503a3e5d60ce23be3d306" -dependencies = [ - "aws-smithy-types", - "urlencoding", -] - -[[package]] -name = "aws-smithy-types" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b02e06ea63498c43bc0217ea4d16605d4e58d85c12fc23f6572ff6d0a840c61" -dependencies = [ - "itoa", - "num-integer", - "ryu", - "time", -] - -[[package]] -name = "aws-smithy-xml" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "246e9f83dd1fdf5d347fa30ae4ad30a9d1d42ce4cd74a93d94afa874646f94cd" -dependencies = [ - "xmlparser", -] - -[[package]] -name = "aws-types" -version = "0.51.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05701d32da168b44f7ee63147781aed8723e792cc131cb9b18363b5393f17f70" -dependencies = [ - "aws-smithy-async", - "aws-smithy-client", - "aws-smithy-http", - "aws-smithy-types", - "http 0.2.12", - "rustc_version 0.4.1", - "tracing", - "zeroize", -] - [[package]] name = "axum" version = "0.6.20" @@ -657,18 +330,6 @@ dependencies = [ "tower-service", ] -[[package]] -name = "axum-macros" -version = "0.3.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cdca6a10ecad987bda04e95606ef85a5417dcaac1a78455242d72e031e2b6b62" -dependencies = [ - "heck 0.4.1", - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "base-x" version = "0.2.11" @@ -879,16 +540,6 @@ dependencies = [ "serde", ] -[[package]] -name = "bytes-utils" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35" -dependencies = [ - "bytes 1.12.1", - "either", -] - [[package]] name = "bzip2" version = "0.4.4" @@ -1288,15 +939,6 @@ dependencies = [ "libc", ] -[[package]] -name = "crc32c" -version = "0.6.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a47af21622d091a8f0fb295b88bc886ac74efcc613efc19f5d0b21de5c89e47" -dependencies = [ - "rustc_version 0.4.1", -] - [[package]] name = "crc32fast" version = "1.5.2" @@ -2177,15 +1819,6 @@ dependencies = [ "siphasher", ] -[[package]] -name = "fastrand" -version = "1.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be" -dependencies = [ - "instant", -] - [[package]] name = "fastrand" version = "2.5.0" @@ -2860,21 +2493,6 @@ dependencies = [ "want", ] -[[package]] -name = "hyper-rustls" -version = "0.23.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1788965e61b367cd03a62950836d5cd41560c3577d90e40e0819373194d1661c" -dependencies = [ - "http 0.2.12", - "hyper 0.14.32", - "log", - "rustls 0.20.9", - "rustls-native-certs", - "tokio 1.53.1", - "tokio-rustls 0.23.4", -] - [[package]] name = "hyper-rustls" version = "0.24.2" @@ -2886,7 +2504,7 @@ dependencies = [ "hyper 0.14.32", "rustls 0.21.12", "tokio 1.53.1", - "tokio-rustls 0.24.1", + "tokio-rustls", ] [[package]] @@ -4147,10 +3765,10 @@ dependencies = [ "libc", "log", "openssl", - "openssl-probe 0.2.1", + "openssl-probe", "openssl-sys", "schannel", - "security-framework 3.7.0", + "security-framework", "security-framework-sys", "tempfile", ] @@ -4413,12 +4031,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "openssl-probe" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e" - [[package]] name = "openssl-probe" version = "0.2.1" @@ -4684,12 +4296,6 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" -[[package]] -name = "pin-utils" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" - [[package]] name = "ping" version = "0.5.2" @@ -4884,46 +4490,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "procfs" -version = "0.16.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "731e0d9356b0c25f16f33b5be79b1c57b562f141ebfcdb0ad8ac2c13a24293b4" -dependencies = [ - "bitflags 2.13.2", - "hex", - "lazy_static", - "procfs-core", - "rustix 0.38.44", -] - -[[package]] -name = "procfs-core" -version = "0.16.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d3554923a69f4ce04c4a754260c338f505ce22642d3830e049a399fc2059a29" -dependencies = [ - "bitflags 2.13.2", - "hex", -] - -[[package]] -name = "prometheus" -version = "0.13.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" -dependencies = [ - "cfg-if 1.0.5", - "fnv", - "lazy_static", - "libc", - "memchr", - "parking_lot 0.12.5", - "procfs", - "protobuf", - "thiserror 1.0.69", -] - [[package]] name = "prometheus-client" version = "0.23.1" @@ -5061,12 +4627,6 @@ dependencies = [ "prost 0.12.6", ] -[[package]] -name = "protobuf" -version = "2.28.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "106dd99e98437432fed6519dedecfade6a06a73bb7b2a1e019fdd2bee5778d94" - [[package]] name = "quick-protobuf" version = "0.8.1" @@ -5397,7 +4957,7 @@ dependencies = [ "http 0.2.12", "http-body 0.4.6", "hyper 0.14.32", - "hyper-rustls 0.24.2", + "hyper-rustls", "hyper-tls", "ipnet", "js-sys", @@ -5416,7 +4976,7 @@ dependencies = [ "system-configuration 0.5.1", "tokio 1.53.1", "tokio-native-tls", - "tokio-rustls 0.24.1", + "tokio-rustls", "tower-service", "url", "wasm-bindgen", @@ -5502,39 +5062,6 @@ dependencies = [ "syn 1.0.109", ] -[[package]] -name = "router" -version = "0.2.1" -dependencies = [ - "anyhow", - "aws-config", - "aws-sdk-s3", - "axum", - "axum-macros", - "clap", - "lazy_static", - "libc", - "prometheus", - "router-controller", - "sqd-messages", - "tokio 1.53.1", - "tracing", - "tracing-subscriber", - "url", -] - -[[package]] -name = "router-controller" -version = "0.1.0" -dependencies = [ - "base64 0.21.7", - "log", - "parking_lot 0.12.5", - "rand 0.8.8", - "serde", - "sqd-messages", -] - [[package]] name = "rtnetlink" version = "0.20.0" @@ -5635,18 +5162,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "rustls" -version = "0.20.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b80e3dec595989ea8510028f30c408a4630db12c9cbb8de34203b89d6577e99" -dependencies = [ - "log", - "ring 0.16.20", - "sct", - "webpki", -] - [[package]] name = "rustls" version = "0.21.12" @@ -5673,18 +5188,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "rustls-native-certs" -version = "0.6.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9aace74cb666635c918e9c12bc0d348266037aa8eb599b5cba565709a8dff00" -dependencies = [ - "openssl-probe 0.1.6", - "rustls-pemfile", - "schannel", - "security-framework 2.11.1", -] - [[package]] name = "rustls-pemfile" version = "1.0.4" @@ -5865,19 +5368,6 @@ dependencies = [ "zeroize", ] -[[package]] -name = "security-framework" -version = "2.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "897b2245f0b511c87893af39b033e5ca9cce68824c4d7e7630b5a1d339658d02" -dependencies = [ - "bitflags 2.13.2", - "core-foundation 0.9.4", - "core-foundation-sys", - "libc", - "security-framework-sys", -] - [[package]] name = "security-framework" version = "3.7.0" @@ -6510,7 +6000,7 @@ version = "3.27.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ - "fastrand 2.5.0", + "fastrand", "getrandom 0.4.3", "once_cell", "rustix 1.1.5", @@ -6806,17 +6296,6 @@ dependencies = [ "tokio-sync", ] -[[package]] -name = "tokio-rustls" -version = "0.23.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c43ee83903113e03984cb9e5cebe6c04a5116269e900e3ddba8f068a62adda59" -dependencies = [ - "rustls 0.20.9", - "tokio 1.53.1", - "webpki", -] - [[package]] name = "tokio-rustls" version = "0.24.1" @@ -6901,7 +6380,7 @@ dependencies = [ "log", "rustls 0.21.12", "tokio 1.53.1", - "tokio-rustls 0.24.1", + "tokio-rustls", "tungstenite", "webpki-roots", ] @@ -7252,12 +6731,6 @@ dependencies = [ "serde", ] -[[package]] -name = "urlencoding" -version = "2.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" - [[package]] name = "utf-8" version = "0.7.6" @@ -7440,16 +6913,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "webpki" -version = "0.22.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ed63aea5ce73d0ff405984102c42de94fc55a6b75765d621c65262469b3c9b53" -dependencies = [ - "ring 0.17.14", - "untrusted 0.9.0", -] - [[package]] name = "webpki-roots" version = "0.25.4" @@ -7916,12 +7379,6 @@ version = "0.8.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e450f9b2ed1dff33c94c12589a87338689467b9c4f5d8a5710bd09a847d2c8a7" -[[package]] -name = "xmlparser" -version = "0.13.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" - [[package]] name = "xmltree" version = "0.10.3" diff --git a/Dockerfile b/Dockerfile index 0fa47b7..0a2e109 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,17 +1,3 @@ -FROM rust:1.98-bookworm AS archive-router-builder -RUN apt-get update && apt-get install protobuf-compiler -y -WORKDIR /archive-router -COPY ./ . -RUN rm -r crates/logs-collector -RUN --mount=type=ssh cargo build --release - -FROM debian:bullseye-slim AS archive-router -RUN apt-get update && apt-get install ca-certificates -y -WORKDIR /archive-router -COPY --from=archive-router-builder /archive-router/target/release/router ./router -ENTRYPOINT ["/archive-router/router"] -EXPOSE 3000 - FROM lukemathwalker/cargo-chef:latest-rust-1.98-slim-bookworm AS chef WORKDIR /app diff --git a/README.md b/README.md index 18a088d..38da5c0 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # SQD Network components -Rust crates for SQD Network nodes: the archive router and the services that collect pings and logs from workers, gateways, and portals. +Rust crates for SQD Network nodes: the services that collect pings and logs from workers, gateways, and portals. SQD is an open data platform for Web3. SQD Network is a decentralized data lake and query engine run by worker, gateway, and scheduler nodes. This repository holds the shared and standalone Rust components used by those nodes. For the network protocol and node implementations, see [subsquid/sqd-network](https://github.com/subsquid/sqd-network). @@ -10,8 +10,6 @@ This is a Cargo workspace. The crates under `crates/` are: | Crate | Type | Description | |---|---|---| -| `router` | binary | Archive router. Reads dataset definitions from S3, assigns data ranges to workers on a scheduling interval, and exposes an HTTP API (port 3000) so clients can find which worker holds a given range. | -| `router-controller` | library | Scheduling and assignment logic used by `router`. | | `logs-collector` | binary | Collects query-execution logs from workers and gateways over the network transport and stores them in ClickHouse. | | `portal-logs-collector` | binary | Collects logs from SQD Portal instances and stores them in ClickHouse. | | `pings-collector` | binary | Collects pings from workers and stores them in ClickHouse. | @@ -19,6 +17,8 @@ This is a Cargo workspace. The crates under `crates/` are: The collector binaries connect to the network over [`sqd-network-transport`](https://github.com/subsquid/sqd-network) (libp2p / QUIC) and persist to ClickHouse. +The archive router (`router`, `router-controller`) is on the `arrowsquid` branch. + ## Build Requires the Rust toolchain pinned in `rust-toolchain` (channel 1.83) and `protobuf-compiler` for building the protocol message dependencies. @@ -27,15 +27,15 @@ Requires the Rust toolchain pinned in `rust-toolchain` (channel 1.83) and `proto # Build the whole workspace cargo build --release --workspace -# Build a single crate, for example the router -cargo build --release -p router +# Build a single crate, for example the pings collector +cargo build --release -p pings-collector ``` Binaries are written to `target/release/`. ### Docker -The `Dockerfile` defines separate build targets for the router and each collector. For example: +The `Dockerfile` defines a build target for each collector: ```bash docker build --target logs-collector -t logs-collector . @@ -45,18 +45,7 @@ docker build --target portal-logs-collector -t portal-logs-collector . ## Usage -Each binary is configured with command-line flags and environment variables (run with `--help` for the full list). For example, the router takes one or more S3 datasets and a set of managed workers: - -```bash -router \ - --dataset eth-mainnet=s3://my-bucket/eth-mainnet \ - --worker \ - --replication 2 \ - --scheduling-unit 50 \ - --scheduling-interval 300 -``` - -The collector binaries read their P2P listen address from `P2P_LISTEN_ADDRS` and their ClickHouse connection details from flags and environment variables. +Each binary is configured with command-line flags and environment variables (run with `--help` for the full list). The collector binaries read their P2P listen address from `P2P_LISTEN_ADDRS` and their ClickHouse connection details from flags and environment variables. ## Scripts diff --git a/crates/router-controller/Cargo.toml b/crates/router-controller/Cargo.toml deleted file mode 100644 index b7e0cb0..0000000 --- a/crates/router-controller/Cargo.toml +++ /dev/null @@ -1,13 +0,0 @@ -[package] -name = "router-controller" -version = "0.1.0" -edition = "2021" - -[dependencies] -base64 = "0.21" -log = "0.4" -parking_lot = "0.12" -rand = "0.8" -serde = "1" - -sqd-messages = { workspace = true } # FIXME: Remove this dependency diff --git a/crates/router-controller/src/atom.rs b/crates/router-controller/src/atom.rs deleted file mode 100644 index 31affd1..0000000 --- a/crates/router-controller/src/atom.rs +++ /dev/null @@ -1,60 +0,0 @@ -// -// Attempt to develop analog of -// https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicReference.html -// -use parking_lot::RwLock; -use std::fmt::{Debug, Formatter}; -use std::ops::{Deref, DerefMut}; -use std::sync::Arc; - -pub struct Atom { - inner: RwLock>, -} - -unsafe impl Send for Atom {} -unsafe impl Sync for Atom {} - -impl Atom { - pub fn new(val: Arc) -> Self { - Atom { - inner: RwLock::new(val), - } - } - - pub fn get(&self) -> Arc { - self.inner.read().deref().clone() - } - - pub fn set(&self, val: Arc) { - let mut lock = self.inner.write(); - *lock.deref_mut() = val - } - - pub fn update Option>>(&self, mut f: F) { - loop { - let initial = self.get(); - if let Some(new_val) = f(initial.deref()) { - let mut lock = self.inner.write(); - let current = lock.deref_mut(); - if Arc::ptr_eq(&initial, current) { - *current = new_val; - return; - } - } else { - return; - } - } - } -} - -impl Clone for Atom { - fn clone(&self) -> Self { - Atom::new(self.get()) - } -} - -impl Debug for Atom { - fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - self.get().fmt(f) - } -} diff --git a/crates/router-controller/src/controller.rs b/crates/router-controller/src/controller.rs deleted file mode 100644 index 06f5913..0000000 --- a/crates/router-controller/src/controller.rs +++ /dev/null @@ -1,586 +0,0 @@ -use std::cmp::{max, min}; -use std::collections::{HashMap, HashSet}; -use std::ops::{Deref, DerefMut}; -use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::Arc; -use std::time::{Duration, SystemTime}; - -use base64::engine::general_purpose::URL_SAFE_NO_PAD; -use base64::Engine; -use rand::prelude::SliceRandom; -use serde::{Deserialize, Serialize}; - -use super::messages::{DataChunk, Range, RangeSet, WorkerState}; - -use crate::atom::Atom; - -pub type WorkerId = String; -pub type Url = String; -pub type Dataset = Url; - -#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct Ping { - pub worker_id: String, - pub worker_url: String, - pub state: Option, - pub pause: bool, -} - -#[derive(Clone, Debug)] -struct Worker { - desired_state: Arc, - info: Arc>, - is_managed: Atom, -} - -#[derive(Clone, Debug)] -struct WorkerInfo { - id: WorkerId, - url: Url, - state: WorkerState, - suspended: bool, - last_ping: SystemTime, -} - -type Wi = usize; -type Ui = usize; -type Assignment = Vec>; - -#[derive(Clone, Debug)] -struct Schedule { - datasets: HashMap>, - assignment: HashMap, -} - -pub struct Controller { - schedule: parking_lot::Mutex, - workers: Atom>, - datasets_height: HashMap, - managed_datasets: HashMap, - managed_workers: parking_lot::RwLock>, - data_replication: usize, - data_management_unit: usize, -} - -unsafe impl Send for Controller {} -unsafe impl Sync for Controller {} - -impl Controller { - /// Get (worker_id, worker_url, encoded_dataset) - pub fn get_worker( - &self, - dataset_name: &str, - first_block: u64, - ) -> Option<(WorkerId, Url, String)> { - let dataset = self.managed_datasets.get(dataset_name)?; - - let now = SystemTime::now(); - - let select_candidate = |w: &Worker| { - if !w - .desired_state - .get(dataset) - .is_some_and(|ranges| ranges.has(first_block)) - { - return None; - } - let info = w.info.get(); - if info.suspended { - return None; - } - if now.duration_since(info.last_ping).unwrap() > Duration::from_secs(30) { - return None; - } - if info - .state - .get(dataset) - .is_some_and(|ranges| ranges.has(first_block)) - { - Some(info) - } else { - None - } - }; - - let workers = self.workers.get(); - - let candidates = { - let managed: Vec<_> = workers - .iter() - .filter(|w| *w.is_managed.get()) - .filter_map(select_candidate) - .collect(); - if !managed.is_empty() { - managed - } else { - workers - .iter() - .filter(|w| !*w.is_managed.get()) - .filter_map(select_candidate) - .collect() - } - }; - - candidates.choose(&mut rand::thread_rng()).map(|info| { - ( - info.id.clone(), - info.url.clone(), - URL_SAFE_NO_PAD.encode(dataset), - ) - }) - } - - pub fn get_height(&self, dataset_name: &str) -> Option { - let dataset = self.managed_datasets.get(dataset_name)?; - - self.datasets_height - .get(dataset) - .map(|height| height.load(Ordering::Relaxed)) - } - - pub fn update_managed_workers>(&self, workers: T) { - *self.managed_workers.write() = workers.into_iter().collect(); - } - - pub fn ping(&self, msg: Ping) -> Arc { - let info = Arc::new(WorkerInfo { - id: msg.worker_id.clone(), - url: msg.worker_url, - state: msg.state.unwrap_or_default(), - suspended: msg.pause, - last_ping: SystemTime::now(), - }); - - let mut desired_state: Option> = None; - let is_managed = Arc::new(self.managed_workers.read().contains(&msg.worker_id)); - - self.workers.update(|workers| { - if let Some(w) = workers.iter().find(|w| w.info.get().id == msg.worker_id) { - w.info.set(info.clone()); - w.is_managed.set(is_managed.clone()); // Set of managed workers can change - desired_state = Some(w.desired_state.clone()); - None - } else { - let new_worker = Worker { - desired_state: Arc::new(info.state.clone()), - info: Arc::new(Atom::new(info.clone())), - is_managed: Atom::new(is_managed.clone()), - }; - desired_state = Some(new_worker.desired_state.clone()); - Some(Arc::new( - workers - .iter() - .cloned() - .chain(std::iter::once(new_worker)) - .collect(), - )) - } - }); - - desired_state.unwrap() - } - - pub fn schedule(&self, mut f: F) - where - F: FnMut(&Dataset, u64) -> Result, ()>, - { - let mut schedule_lock = self.schedule.lock(); - let schedule = schedule_lock.deref_mut(); - let mut workers = self.workers.get().deref().clone(); - - Self::remove_dead_workers(&mut workers); - - let managed_workers: Vec<_> = workers - .iter() - .filter(|w| *w.is_managed.get()) - .cloned() - .collect(); - - if managed_workers.len() < self.managed_workers.read().len() { - self.workers.set(Arc::new(workers)); - return; - } - - let mut desired_state: Vec = std::iter::repeat_with(Default::default) - .take(managed_workers.len()) - .collect(); - - for (dataset, chunks) in schedule.datasets.iter_mut() { - if Self::import_new_chunks(chunks, |next_block| f(dataset, next_block)) { - if let Some(chunk) = chunks.last() { - let height = self.datasets_height.get(dataset).unwrap(); - let last_block = chunk.last_block(); - height.store(last_block, Ordering::Relaxed); - } - let plan = self.schedule_dataset( - &managed_workers, - &mut schedule.assignment, - dataset, - chunks, - ); - for (w, ranges) in plan.into_iter().enumerate() { - desired_state[w].insert(dataset.clone(), ranges); - } - } - } - - for worker in workers.iter_mut().filter(|w| *w.is_managed.get()) { - let i = managed_workers - .iter() - .position(|w| w.info.get().id == worker.info.get().id) - .unwrap(); - worker.desired_state = Arc::new(desired_state[i].clone()); - } - - self.workers.set(Arc::new(workers)); - } - - fn remove_dead_workers(workers: &mut Vec) { - let now = SystemTime::now(); - workers.retain(|w| { - *w.is_managed.get() || { - let since_last_ping = now - .duration_since(w.info.get().last_ping) - .unwrap_or(Duration::from_secs(0)); - since_last_ping < Duration::from_secs(5 * 60) - } - }) - } - - fn import_new_chunks(chunks: &mut Vec, f: F) -> bool - where - F: FnOnce(u64) -> Result, ()>, - { - let mut next_block = chunks.last().map_or(0, |c| c.last_block() + 1); - match f(next_block) { - Err(_) => false, - Ok(mut new_chunks) => { - new_chunks.sort(); - // check, that the new chunks are non-overlapping and span a continuous range - for (i, c) in new_chunks.iter().enumerate() { - if next_block != c.first_block() && next_block != 0 { - let p = if i > 0 { - &new_chunks[i - 1] - } else { - chunks.last().unwrap() - }; - if next_block > c.first_block() { - log::error!("Received overlapping chunks: {} and {}", p, c); - return false; - } else { - log::error!("There is a gap between {} and {}", p, c); - return false; - } - } - next_block = c.last_block() + 1 - } - chunks.append(&mut new_chunks); - true - } - } - } - - fn schedule_dataset( - &self, - workers: &[Worker], - assignment_map: &mut HashMap, - dataset: &Dataset, - chunks: &[DataChunk], - ) -> Vec { - let units: Vec = chunks - .chunks(self.data_management_unit) - .map(|unit| { - Range::new( - unit.first().unwrap().first_block(), - unit.last().unwrap().last_block(), - ) - }) - .collect(); - - let no_state = RangeSet::empty(); - let infos: Vec<_> = workers.iter().map(|w| w.info.get()).collect(); - - let actual: Assignment = infos - .iter() - .map(|info| { - let s = info.state.get(dataset).unwrap_or(&no_state); - units - .iter() - .enumerate() - .filter_map(|(i, &u)| if s.has(u.begin) { Some(i) } else { None }) - .collect() - }) - .collect(); - - if let Some(goal) = assignment_map.get_mut(dataset) { - for u in 0..units.len() { - let n_holders = Self::get_holders(goal, &u).count(); - if n_holders < self.data_replication { - Self::assign(goal, self.data_replication - n_holders, u) - } - } - } else { - let mut goal = actual.clone(); - - for u in 0..units.len() { - let mut holders: Vec<_> = Self::get_holders(&goal, &u).collect(); - if holders.len() > self.data_replication { - holders.sort_by_key(|&w| goal[w].len()); - for &w in holders.iter().skip(self.data_replication) { - goal[w].remove(&u); - } - } else { - Self::assign(&mut goal, self.data_replication - holders.len(), u); - } - } - - if !goal.is_empty() { - let mut order: Vec = (0..goal.len()).collect(); - let lst = goal.len() - 1; - let target_size = goal.iter().map(|a| a.len()).sum::() / goal.len(); - loop { - order.sort_by_key(|i| goal[*i].len()); - let s = order[0]; - let l = order[lst]; - let l_size = goal[l].len(); - let s_size = goal[s].len(); - if l_size - s_size < 2 { - break; - } - let to_move = max(1, min(target_size - s_size, l_size - target_size)); - for u in Self::select_randomly(to_move, goal[l].difference(&goal[s]).cloned()) { - goal[s].insert(u); - goal[l].remove(&u); - } - } - } - - assignment_map.insert(dataset.clone(), goal); - } - - let mut plan = assignment_map.get(dataset).unwrap().clone(); - - let actual_and_planned: Assignment = (0..workers.len()) - .map(|w| actual[w].intersection(&plan[w]).cloned().collect()) - .collect(); - - for u in 0..units.len() { - let n_holders = Self::get_holders(&actual_and_planned, &u).count(); - if n_holders < self.data_replication { - for w in Self::get_holders(&actual, &u) - .filter(|&w| !actual_and_planned[w].contains(&u)) - .take(self.data_replication - n_holders) - { - plan[w].insert(u); - } - } - } - - plan.iter() - .map(|a| { - let ranges: Vec = a.iter().map(|&u| units[u]).collect(); - RangeSet::from(ranges) - }) - .collect() - } - - fn get_holders<'a>(assignment: &'a Assignment, u: &'a Ui) -> impl Iterator + 'a { - (0..assignment.len()).filter(|&w| assignment[w].contains(u)) - } - - fn assign(goal: &mut Assignment, replicas: usize, u: Ui) { - for w in Self::select_randomly(replicas, (0..goal.len()).filter(|&w| !goal[w].contains(&u))) - { - goal[w].insert(u); - } - } - - fn select_randomly>(count: usize, candidates: I) -> Vec { - if count == 0 { - return Vec::new(); - } - - let mut vec: Vec = candidates.into_iter().collect(); - if count >= vec.len() { - return vec; - } - - let mut offset = 0; - let mut len = vec.len(); - while len > 0 && offset < count { - let i = rand::random::() % len; - vec.swap(offset, offset + i); - offset += 1; - len -= 1; - } - vec.truncate(count); - vec - } -} - -pub struct ControllerBuilder { - managed_datasets: HashMap, - managed_workers: HashSet, - replication: usize, - data_management_unit: usize, -} - -impl Default for ControllerBuilder { - fn default() -> Self { - Self::new() - } -} - -impl ControllerBuilder { - pub fn new() -> Self { - ControllerBuilder { - managed_datasets: HashMap::new(), - managed_workers: HashSet::new(), - replication: 1, - data_management_unit: 50, - } - } - - pub fn set_data_replication(&mut self, n: usize) -> &mut Self { - self.replication = n; - self - } - - pub fn set_data_management_unit(&mut self, n_chunks: usize) -> &mut Self { - self.data_management_unit = n_chunks; - self - } - - pub fn add_worker(&mut self, worker_id: WorkerId) -> &mut Self { - self.managed_workers.insert(worker_id); - self - } - - pub fn set_workers(&mut self, workers: I) -> &mut Self - where - I: IntoIterator, - { - self.managed_workers.clear(); - self.managed_workers.extend(workers); - self - } - - pub fn add_dataset(&mut self, name: String, dataset: Dataset) -> &mut Self { - self.managed_datasets.insert(name, dataset); - self - } - - pub fn set_datasets(&mut self, datasets: I) -> &mut Self - where - I: IntoIterator, - { - self.managed_datasets.clear(); - self.managed_datasets.extend(datasets); - self - } - - pub fn build(&self) -> Controller { - Controller { - schedule: parking_lot::Mutex::new(Schedule { - datasets: self - .managed_datasets - .values() - .map(|ds| (ds.clone(), Vec::new())) - .collect(), - assignment: HashMap::new(), - }), - datasets_height: self - .managed_datasets - .values() - .map(|name| (name.clone(), AtomicU64::new(0))) - .collect(), - workers: Atom::new(Arc::new(Vec::new())), - managed_datasets: self.managed_datasets.clone(), - managed_workers: parking_lot::RwLock::new(self.managed_workers.clone()), - data_replication: self.replication, - data_management_unit: self.data_management_unit, - } - } -} - -#[cfg(test)] -mod tests { - use std::ops::Deref; - - use super::Ping; - - use crate::controller::ControllerBuilder; - use sqd_messages::data_chunk::DataChunk; - - #[test] - fn basic() { - let controller = ControllerBuilder::new() - .set_data_management_unit(1) - .set_data_replication(2) - .set_workers((0..8).map(|i| i.to_string())) - .set_datasets((0..2).map(|i| (i.to_string(), i.to_string()))) - .build(); - - let chunk = |top, first_block, last_block| { - DataChunk::new(top, first_block, last_block, "abcde").expect("valid chunk") - }; - let chunks = [ - vec![chunk(0, 0, 10), chunk(0, 11, 200)], - vec![chunk(0, 0, 5), chunk(6, 6, 20)], - ]; - - for w in 0..8 { - controller.ping(Ping { - worker_id: w.to_string(), - worker_url: w.to_string(), - state: Some(Default::default()), - pause: false, - }); - } - - controller.schedule(|ds, _from_block| Ok(chunks[ds.parse::().unwrap()].clone())); - - let desired_state: Vec<_> = (0..8) - .map(|w| { - controller.ping(Ping { - worker_id: w.to_string(), - worker_url: w.to_string(), - state: Some(Default::default()), - pause: false, - }) - }) - .collect(); - - assert_eq!(controller.get_worker("0", 5), None); - - for (w, state) in desired_state.iter().enumerate() { - controller.ping(Ping { - worker_id: w.to_string(), - worker_url: w.to_string(), - state: Some(state.deref().clone()), - pause: false, - }); - } - - let holders: Vec<_> = desired_state - .iter() - .enumerate() - .filter_map(|(wi, s)| { - s.get("0") - .map(|range| { - if range.has(10) && range.has(0) { - Some((wi.to_string(), wi.to_string(), "MA".to_string())) - } else { - None - } - }) - .unwrap_or(None) - }) - .collect(); - - assert_eq!(holders.len(), 2); - assert!(holders.contains(&controller.get_worker("0", 0).unwrap())); - assert!(holders.contains(&controller.get_worker("0", 5).unwrap())); - assert!(holders.contains(&controller.get_worker("0", 10).unwrap())); - } -} diff --git a/crates/router-controller/src/lib.rs b/crates/router-controller/src/lib.rs deleted file mode 100644 index a69c08d..0000000 --- a/crates/router-controller/src/lib.rs +++ /dev/null @@ -1,3 +0,0 @@ -mod atom; -pub mod controller; -pub mod messages; diff --git a/crates/router-controller/src/messages.rs b/crates/router-controller/src/messages.rs deleted file mode 100644 index ed15def..0000000 --- a/crates/router-controller/src/messages.rs +++ /dev/null @@ -1,31 +0,0 @@ -use std::{ - collections::HashMap, - ops::{Deref, DerefMut}, -}; - -pub use sqd_messages::{data_chunk::DataChunk, Range, RangeSet}; - -#[derive(serde::Serialize, serde::Deserialize, Default, Debug, Clone, PartialEq, Eq)] -pub struct WorkerState { - pub datasets: HashMap, -} - -impl Deref for WorkerState { - type Target = HashMap; - - fn deref(&self) -> &Self::Target { - &self.datasets - } -} - -impl DerefMut for WorkerState { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.datasets - } -} - -impl From> for WorkerState { - fn from(datasets: HashMap) -> Self { - Self { datasets } - } -} diff --git a/crates/router/Cargo.toml b/crates/router/Cargo.toml deleted file mode 100644 index 97d536f..0000000 --- a/crates/router/Cargo.toml +++ /dev/null @@ -1,22 +0,0 @@ -[package] -name = "router" -version = "0.2.1" -edition = "2021" - -[dependencies] -anyhow = "1" -aws-sdk-s3 = "0.21.0" -aws-config = "0.51.0" -axum = "0.6" -axum-macros = "0.3" -clap = { version = "4.0.18", features = ["derive"] } -tokio = { version = "1.21.2", features = ["full"] } -url = "2.3.1" -tracing = "0.1" -tracing-subscriber = { version = "0.3.0", features = ["env-filter", "json"] } -libc = "0.2" -lazy_static = "1.4.0" -prometheus = { version = "0.13.3", features = ["process"] } - -router-controller = { version = "0.1", path = "../router-controller" } -sqd-messages = { workspace = true } # FIXME: Remove this dependency diff --git a/crates/router/src/cli.rs b/crates/router/src/cli.rs deleted file mode 100644 index ab6330e..0000000 --- a/crates/router/src/cli.rs +++ /dev/null @@ -1,42 +0,0 @@ -use clap::Parser; - -fn parse_dataset(s: &str) -> Result<(String, String), String> { - let pos = s - .find('=') - .ok_or_else(|| format!("invalid KEY=value: no `=` found in `{}`", s))?; - - let name = s[..pos].to_string(); - if name.is_empty() { - return Err(format!("invalid KEY=value: the KEY is empty in `{}`", s)); - } - - let url = s[pos + 1..].to_string(); - if !url.starts_with("s3://") { - return Err(format!("invalid S3 URL: `{}`", url)); - } - - Ok((name, url)) -} - -#[derive(Parser)] -pub struct Cli { - /// Add dataset `NAME` pointing to S3 `URL` - #[clap(short, long, value_parser = parse_dataset, value_name = "NAME=URL")] - pub dataset: Vec<(String, String)>, - - /// Add managed worker - #[clap(short, long, value_name = "ID")] - pub worker: Vec, - - /// Data replication factor - #[clap(short, long, value_name = "N")] - pub replication: usize, - - /// Size of a data scheduling unit (in chunks) - #[clap(short = 'u', long, value_name = "N")] - pub scheduling_unit: usize, - - /// Scheduling interval (in seconds) - #[clap(short = 'i', long, default_value = "300", value_name = "N")] - pub scheduling_interval: u64, -} diff --git a/crates/router/src/dataset.rs b/crates/router/src/dataset.rs deleted file mode 100644 index 92a0aca..0000000 --- a/crates/router/src/dataset.rs +++ /dev/null @@ -1,170 +0,0 @@ -use std::str::FromStr; - -use aws_sdk_s3::Client; -use tokio::runtime::Handle; - -use sqd_messages::data_chunk::DataChunk; - -pub trait Storage { - /// Get data chunks in the dataset. - fn get_chunks(&mut self, next_block: u64) -> Result, String>; -} - -fn invalid_object_key(key: &str) -> String { - format!("invalid object key - {key}") -} - -/// The chunk of a `/blocks.parquet` key; `None` for the chunk's other files. -fn chunk_of_blocks_key(key: &str) -> Result, String> { - let Some(chunk_id) = key.strip_suffix("/blocks.parquet") else { - return Ok(None); - }; - - DataChunk::from_str(chunk_id) - .map(Some) - .map_err(|()| invalid_object_key(key)) -} - -fn trim_trailing_slash(value: &str) -> String { - let mut value = value.to_string(); - value.pop(); - value -} - -pub struct S3Storage { - client: Client, - bucket: String, -} - -impl Storage for S3Storage { - fn get_chunks(&mut self, next_block: u64) -> Result, String> { - let mut objects = vec![]; - let handle = Handle::current(); - - let prefix = None; - let tops = self.ls(prefix)?; - - let top = tops.iter().rev().find(|top| { - let top: u64 = top.parse().unwrap(); - top <= next_block - }); - - if let Some(top) = top { - let prefix = format!("{}/", top); - let top_chunks = self.ls(Some(&prefix))?; - - let next_chunk = top_chunks.into_iter().find_map(|chunk| { - let chunk = DataChunk::from_str(&chunk).unwrap(); - if chunk.first_block() == next_block { - Some(chunk) - } else { - None - } - }); - - if let Some(chunk) = next_chunk { - let start_after = chunk.to_string(); - let builder = self - .client - .list_objects_v2() - .bucket(&self.bucket) - .start_after(start_after); - let output = handle - .block_on(builder.send()) - .map_err(|err| err.to_string())?; - let mut continuation_token = output.next_continuation_token.clone(); - if let Some(contents) = output.contents() { - objects.extend_from_slice(contents); - } - while let Some(token) = continuation_token { - let future = self - .client - .list_objects_v2() - .bucket(&self.bucket) - .continuation_token(token) - .send(); - let output = handle.block_on(future).map_err(|err| err.to_string())?; - continuation_token = output.next_continuation_token.clone(); - if let Some(contents) = output.contents() { - objects.extend_from_slice(contents); - } - } - } - } - - let mut chunks = vec![]; - for object in &objects { - if let Some(key) = object.key() { - if let Some(chunk) = chunk_of_blocks_key(key)? { - chunks.push(chunk); - } - } - } - - Ok(chunks) - } -} - -impl S3Storage { - pub fn new(client: Client, bucket: String) -> Self { - S3Storage { client, bucket } - } - - fn ls(&self, prefix: Option<&str>) -> Result, String> { - let handle = Handle::current(); - let mut builder = self - .client - .list_objects_v2() - .bucket(&self.bucket) - .delimiter('/'); - if let Some(prefix) = prefix { - builder = builder.prefix(prefix) - } - let output = handle - .block_on(builder.send()) - .map_err(|err| err.to_string())?; - let mut items = vec![]; - if let Some(prefixes) = output.common_prefixes() { - for prefix in prefixes { - if let Some(value) = prefix.prefix() { - let value = trim_trailing_slash(value); - items.push(value); - } - } - } - Ok(items) - } -} - -#[cfg(test)] -mod tests { - use super::chunk_of_blocks_key; - - #[test] - fn reads_the_chunk_from_a_full_object_key() { - let key = "0000000000/0000000000-0000000099-abcde/blocks.parquet"; - - let chunk = chunk_of_blocks_key(key) - .unwrap() - .expect("a blocks.parquet key"); - - assert_eq!(chunk.to_string(), "0000000000/0000000000-0000000099-abcde"); - } - - #[test] - fn skips_the_other_files_of_a_chunk() { - let key = "0000000000/0000000000-0000000099-abcde/transactions.parquet"; - - assert_eq!(chunk_of_blocks_key(key), Ok(None)); - } - - #[test] - fn rejects_a_malformed_chunk_id() { - let key = "0000000000/0000000099-abcde/blocks.parquet"; - - assert_eq!( - chunk_of_blocks_key(key), - Err(format!("invalid object key - {key}")) - ); - } -} diff --git a/crates/router/src/http_server.rs b/crates/router/src/http_server.rs deleted file mode 100644 index cdd0abc..0000000 --- a/crates/router/src/http_server.rs +++ /dev/null @@ -1,109 +0,0 @@ -use std::net::SocketAddr; -use std::ops::Deref; -use std::sync::Arc; - -use axum::body::{boxed, Body}; -use axum::extract::{Extension, Path}; -use axum::http::header::CONTENT_TYPE; -use axum::http::StatusCode; -use axum::middleware::from_fn; -use axum::response::{IntoResponse, Response}; -use axum::routing::{get, post}; -use axum::{Json, Router}; -use prometheus::{gather, Encoder, TextEncoder}; -use tracing::info; - -use router_controller::controller::{Controller, Ping}; -use router_controller::messages::WorkerState; - -mod middleware; - -#[axum_macros::debug_handler] -async fn ping( - Extension(controller): Extension>, - Json(msg): Json, -) -> Json { - let worker_id = msg.worker_id.clone(); - let worker_url = msg.worker_url.clone(); - let current_state = format!("{:?}", msg.state); - let desired_state = controller.ping(msg); - info!( - ping_from = worker_id, - worker_url, - current_state, - desired_state = format!("{:?}", desired_state) - ); - Json(desired_state.deref().clone()) -} - -#[axum_macros::debug_handler] -async fn get_worker( - Path((dataset, start_block)): Path<(String, u64)>, - Extension(controller): Extension>, -) -> Response { - match controller.get_worker(&dataset, start_block) { - Some((_, url, encoded_dataset)) => format!("{url}/{encoded_dataset}").into_response(), - None => ( - StatusCode::SERVICE_UNAVAILABLE, - format!( - "not ready to serve block {} of dataset {}", - start_block, dataset - ), - ) - .into_response(), - } -} - -#[axum_macros::debug_handler] -async fn get_height( - Path(dataset): Path, - Extension(controller): Extension>, -) -> Response { - match controller.get_height(&dataset) { - Some(height) => height.to_string().into_response(), - None => ( - StatusCode::SERVICE_UNAVAILABLE, - format!("height for dataset {} is unknown", dataset), - ) - .into_response(), - } -} - -#[axum_macros::debug_handler] -async fn get_metrics() -> Response { - let encoder = TextEncoder::new(); - let mut buffer = vec![]; - encoder - .encode(&gather(), &mut buffer) - .expect("Failed to encode metrics"); - Response::builder() - .status(200) - .header(CONTENT_TYPE, encoder.format_type()) - .body(boxed(Body::from(buffer))) - .unwrap() -} - -pub struct Server { - controller: Arc, -} - -impl Server { - pub fn new(controller: Arc) -> Self { - Server { controller } - } - - pub async fn run(&self) { - let app = Router::new() - .route("/ping", post(ping)) - .route("/network/:dataset/:start_block/worker", get(get_worker)) - .route("/network/:dataset/height", get(get_height)) - .route("/metrics", get(get_metrics)) - .layer(from_fn(middleware::logging)) - .layer(Extension(self.controller.clone())); - let addr = SocketAddr::from(([0, 0, 0, 0], 3000)); - axum::Server::bind(&addr) - .serve(app.into_make_service()) - .await - .unwrap() - } -} diff --git a/crates/router/src/http_server/middleware.rs b/crates/router/src/http_server/middleware.rs deleted file mode 100644 index 84dd46d..0000000 --- a/crates/router/src/http_server/middleware.rs +++ /dev/null @@ -1,16 +0,0 @@ -use axum::http::Request; -use axum::middleware::Next; -use axum::response::IntoResponse; -use tracing::info; - -pub async fn logging(req: Request, next: Next) -> impl IntoResponse { - let method = req.method().to_string(); - let uri = req.uri().to_string(); - let version = format!("{:?}", req.version()); - - let response = next.run(req).await; - let status = response.status().to_string(); - info!(method, uri, version, status); - - response -} diff --git a/crates/router/src/logger.rs b/crates/router/src/logger.rs deleted file mode 100644 index 66b361c..0000000 --- a/crates/router/src/logger.rs +++ /dev/null @@ -1,23 +0,0 @@ -use tracing_subscriber::EnvFilter; - -fn is_tty() -> bool { - unsafe { libc::isatty(libc::STDOUT_FILENO) != 0 } -} - -pub fn init() { - if is_tty() { - tracing_subscriber::fmt() - .with_env_filter(EnvFilter::from_default_env()) - .with_target(false) - .init(); - } else { - tracing_subscriber::fmt() - .with_env_filter(EnvFilter::from_default_env()) - .with_target(false) - .json() - .flatten_event(true) - .with_span_list(false) - .with_current_span(false) - .init(); - } -} diff --git a/crates/router/src/main.rs b/crates/router/src/main.rs deleted file mode 100644 index 71bf760..0000000 --- a/crates/router/src/main.rs +++ /dev/null @@ -1,71 +0,0 @@ -use std::collections::HashMap; -use std::env; -use std::sync::Arc; -use std::time::Duration; - -use clap::Parser; -use url::Url; - -use cli::Cli; -use dataset::{S3Storage, Storage}; -use router_controller::controller::ControllerBuilder; - -mod cli; -mod dataset; -mod http_server; -mod logger; -mod metrics; -mod scheduler; - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - let args = Cli::parse(); - logger::init(); - - let mut storages: HashMap> = HashMap::new(); - for (_name, dataset) in &args.dataset { - let storage = create_storage(dataset).await; - storages.insert(dataset.clone(), storage); - } - - let controller = ControllerBuilder::new() - .set_data_replication(args.replication) - .set_data_management_unit(args.scheduling_unit) - .set_workers(args.worker) - .set_datasets(args.dataset) - .build(); - - let controller = Arc::new(controller); - - let scheduling_interval = Duration::from_secs(args.scheduling_interval); - scheduler::start(controller.clone(), storages, scheduling_interval); - - http_server::Server::new(controller).run().await; - - Ok(()) -} - -async fn create_storage(dataset: &String) -> Box { - let url = Url::parse(dataset); - match url { - Ok(url) => match url.scheme() { - "s3" => { - let mut config_loader = aws_config::from_env(); - let s3_endpoint = env::var("AWS_S3_ENDPOINT").ok(); - if let Some(s3_endpoint) = &s3_endpoint { - let uri = s3_endpoint.parse().expect("invalid s3-endpoint"); - let endpoint = aws_sdk_s3::Endpoint::immutable(uri); - config_loader = config_loader.endpoint_resolver(endpoint); - } - let config = config_loader.load().await; - - let client = aws_sdk_s3::Client::new(&config); - let host = url.host_str().expect("invalid dataset host").to_string(); - let bucket = host + url.path(); - Box::new(S3Storage::new(client, bucket)) - } - _ => panic!("unsupported filesystem - {}", url.scheme()), - }, - Err(..) => panic!("unsupported dataset - {}", dataset), - } -} diff --git a/crates/router/src/metrics.rs b/crates/router/src/metrics.rs deleted file mode 100644 index 5fb71ff..0000000 --- a/crates/router/src/metrics.rs +++ /dev/null @@ -1,10 +0,0 @@ -use lazy_static::lazy_static; -use prometheus::{opts, register_int_counter_vec, IntCounterVec}; - -lazy_static! { - pub static ref DATASET_SYNC_ERRORS: IntCounterVec = register_int_counter_vec!( - opts!("sqd_dataset_sync_errors", "Dataset synchronization errors"), - &["dataset"] - ) - .expect("Can't create a metric"); -} diff --git a/crates/router/src/scheduler.rs b/crates/router/src/scheduler.rs deleted file mode 100644 index d07de8e..0000000 --- a/crates/router/src/scheduler.rs +++ /dev/null @@ -1,41 +0,0 @@ -use std::collections::HashMap; -use std::sync::Arc; -use std::thread; -use std::time::Duration; - -use tracing::{debug, error, info}; - -use router_controller::controller::Controller; - -use crate::dataset::Storage; -use crate::metrics::DATASET_SYNC_ERRORS; - -pub fn start( - controller: Arc, - mut storages: HashMap>, - interval: Duration, -) { - tokio::task::spawn_blocking(move || { - info!("started scheduling task with {:?} interval", interval); - loop { - thread::sleep(interval); - info!("started scheduling"); - controller.schedule(|dataset, next_block| { - info!("downloading new chunks for {}", dataset); - let storage = storages.get_mut(dataset).unwrap(); - match storage.get_chunks(next_block) { - Ok(chunks) => { - debug!("found new chunks in {}: {:?}", dataset, chunks); - Ok(chunks) - } - Err(err) => { - error!("failed to download new chunks for {}: {:?}", dataset, err); - DATASET_SYNC_ERRORS.with_label_values(&[dataset]).inc(); - Err(()) - } - } - }); - info!("finished scheduling"); - } - }); -}