From 29a93751183902b1bac36a9a9250a0910231e6ed Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Tue, 25 Aug 2026 09:51:39 +0800 Subject: [PATCH 1/4] feat(crates-sync): add K8s Job entrypoint and Harbor image Add run_job.py for wait/bootstrap against mono, freighter hostPath paths, and keep-crate-cache so shared caches are not deleted. Ship Dockerfile plus crates-sync deploy workflow for onprem Jobs. --- .github/workflows/crates-sync-deploy.yml | 53 +++ .github/workflows/release.yml | 4 + Cargo.lock | 174 ++++++--- Cargo.toml | 2 +- scripts/crates-sync/Dockerfile | 24 ++ scripts/crates-sync/README.md | 45 ++- scripts/crates-sync/crates-sync.py | 456 ++++++++++++++--------- scripts/crates-sync/run_job.py | 256 +++++++++++++ scripts/demo/README.md | 1 + scripts/demo/build-demo-images-local.sh | 6 +- vault/src/pki.rs | 2 +- 11 files changed, 799 insertions(+), 224 deletions(-) create mode 100644 .github/workflows/crates-sync-deploy.yml create mode 100644 scripts/crates-sync/Dockerfile create mode 100644 scripts/crates-sync/run_job.py diff --git a/.github/workflows/crates-sync-deploy.yml b/.github/workflows/crates-sync-deploy.yml new file mode 100644 index 000000000..dbbe328ae --- /dev/null +++ b/.github/workflows/crates-sync-deploy.yml @@ -0,0 +1,53 @@ +name: Crates Sync deploy + +on: + push: + branches: + - main + paths: + - ".github/workflows/crates-sync-deploy.yml" + - "scripts/crates-sync/**" + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +permissions: + contents: read + +env: + REPOSITORY: mega/crates-sync + IMAGE_TAG: latest + HARBOR_REGISTRY: registry.xuanwu.openatom.cn + +jobs: + build-and-push: + runs-on: ubuntu-latest + timeout-minutes: 15 + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Login to Harbor + uses: docker/login-action@v3 + with: + registry: ${{ env.HARBOR_REGISTRY }} + username: ${{ secrets.HARBOR_USERNAME }} + password: ${{ secrets.HARBOR_PASSWORD }} + + - name: Build, tag, and push docker image to Harbor + run: | + set -euo pipefail + + HARBOR_IMAGE_BASE="${{ env.HARBOR_REGISTRY }}/${{ env.REPOSITORY }}" + IMAGE_TAG="${{ env.IMAGE_TAG }}" + SHORT_SHA="${GITHUB_SHA:0:7}" + + docker build \ + -f scripts/crates-sync/Dockerfile \ + -t "$HARBOR_IMAGE_BASE:$IMAGE_TAG" \ + -t "$HARBOR_IMAGE_BASE:$SHORT_SHA" \ + scripts/crates-sync + + docker push "$HARBOR_IMAGE_BASE:$IMAGE_TAG" + docker push "$HARBOR_IMAGE_BASE:$SHORT_SHA" diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 767b75b14..8a55313e3 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -31,6 +31,10 @@ jobs: context: . dockerfile: scripts/init_mega/Dockerfile + - name: crates-sync + context: scripts/crates-sync + dockerfile: scripts/crates-sync/Dockerfile + steps: - name: Checkout uses: actions/checkout@v4 diff --git a/Cargo.lock b/Cargo.lock index 74fc2ec70..988d77aa5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -68,16 +68,16 @@ dependencies = [ [[package]] name = "aes-gcm" -version = "0.11.0" +version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fdf011db2e21ce0d575593d749db5554b47fed37aff429e4dc50bc91ac93a028" +checksum = "7f2b8006a0c83f52b62ba44a97b58bf76fe2f70a329e588f67f89691d93d498f" dependencies = [ "aead 0.6.1", "aes 0.9.2", "cipher 0.5.2", "ctr 0.10.1", + "ctutils", "ghash 0.6.0", - "subtle", "zeroize", ] @@ -608,7 +608,7 @@ checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -1272,7 +1272,7 @@ checksum = "46d07918caa9eeaaf06b7873925c53a61daac173539b4f7715090745e44e4e69" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -1345,9 +1345,9 @@ dependencies = [ [[package]] name = "cc" -version = "1.4.3" +version = "1.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" +checksum = "0ad534f4357a5264cce5019c989cf66a4f0dc4e0d1b1d15f8aacec0ff7360273" dependencies = [ "find-msvc-tools", "jobserver", @@ -1589,7 +1589,7 @@ dependencies = [ "heck 0.5.0", "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -1884,9 +1884,9 @@ checksum = "fd121741cf3eb82c08dd3023eb55bf2665e5f60ec20f89760cf836ae4562e6a0" [[package]] name = "crc32fast" -version = "1.5.0" +version = "1.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +checksum = "8498c871161e1742aaa9d52551b2d6ebdd4c3d45a3be423e3728f33b955be550" dependencies = [ "cfg-if", ] @@ -2522,7 +2522,7 @@ checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -2670,14 +2670,14 @@ dependencies = [ "enum-ordinalize", "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] name = "either" -version = "1.17.0" +version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e5e8f6c15a24b9a3ee5efec809ccd006d3b30e8b3bb63c39af737c7f87daa1d" +checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" dependencies = [ "serde", ] @@ -2818,7 +2818,7 @@ checksum = "a65863d15a4ce2888bd2f0f543cc963d3879c3a022c8ee43f6141d479a3ac815" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -3137,7 +3137,7 @@ checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -3256,6 +3256,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2eecf2d5dc9b66b732b97707a0210906b1d30523eb773193ab777c0c84b3e8d5" dependencies = [ "polyval 0.7.3", + "zeroize", ] [[package]] @@ -3364,9 +3365,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.17" +version = "0.4.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f877e75f39e9827ec50a572dd592684ac28c029578726c85f1b2aa6ab807449" +checksum = "839c0e8a181239723652be9062bb56ca5bf5f64011f73b623f6f4fc59086a228" dependencies = [ "atomic-waker", "bytes", @@ -3854,9 +3855,9 @@ checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa" [[package]] name = "icu_provider" -version = "2.3.0" +version = "2.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92a7ed671a6aad807a8651a2e1782a6598fda9ce5185dd8158549e95a91c6428" +checksum = "d27bbb9d3abbefac45d55f647c9de1d44aafcd1186eb91879afef17c396c3e73" dependencies = [ "displaydoc", "icu_locale_core", @@ -4316,9 +4317,9 @@ dependencies = [ [[package]] name = "keccak" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ffd9697dc4a9a62e2da93389f34400b77a28f0287711263cabb203b3ccb9c0e4" +checksum = "d8f198d1db720e4940b5a493201d199d9f24f568f8f746bd13706243a2f71598" dependencies = [ "cfg-if", "cpufeatures 0.3.0", @@ -4567,7 +4568,7 @@ dependencies = [ "bitflags 2.13.1", "libc", "plain", - "redox_syscall 0.9.2", + "redox_syscall 0.9.3", ] [[package]] @@ -4705,9 +4706,9 @@ checksum = "5be1cf190319c74ba3e45923624626ae2e43fe42ad7e60ff38ded81044c37630" [[package]] name = "log" -version = "0.4.33" +version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" [[package]] name = "logos" @@ -4994,7 +4995,7 @@ dependencies = [ "rand 0.10.2", "regex", "reqwest 0.13.4", - "russh", + "russh 0.63.1", "saturn", "serde", "serde_json", @@ -6213,7 +6214,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63d440a804ec8d6fafbb6b84471e013286658d373248927692ab3366686220ca" dependencies = [ "aes 0.9.2", - "aes-gcm 0.11.0", + "aes-gcm 0.11.1", "cbc", "der 0.8.1", "pbkdf2", @@ -6299,6 +6300,7 @@ dependencies = [ "cpubits", "cpufeatures 0.3.0", "universal-hash 0.6.1", + "zeroize", ] [[package]] @@ -6482,7 +6484,7 @@ checksum = "1c8d9ca532f185d5d4db7a7c9d51420b452168ea1c2b913953281bd6fe1fcbd0" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -6562,7 +6564,7 @@ dependencies = [ "qapi", "rand 0.10.2", "reqwest 0.13.4", - "russh", + "russh 0.62.7", "russh-sftp", "serde", "serde_json", @@ -6875,9 +6877,9 @@ dependencies = [ [[package]] name = "redox_syscall" -version = "0.9.2" +version = "0.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1c93da5bb2c5d4e6c0ef7abeead62c89169a0a4882bfb83ac892f2423aea2fe" +checksum = "d678d17679829e73d371e96880897e98fee2ded7acc0a50bdf8af2affa4b2fe5" dependencies = [ "bitflags 2.13.1", ] @@ -6910,7 +6912,7 @@ checksum = "92ecd8964f8453721699a1ed72037b0db49ce2f5a5138486ee89bed6f67cdf3a" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -7161,7 +7163,7 @@ checksum = "1c25ef604ac7dd839d44d64648952ea23c97866f124ff671b0ed2cf3ad9bb06e" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -7252,7 +7254,79 @@ dependencies = [ "hmac 0.13.0", "inout 0.2.2", "internal-russh-num-bigint", - "keccak 0.2.1", + "keccak 0.2.2", + "log", + "md5", + "ml-kem", + "module-lattice", + "num-bigint 0.4.8", + "p256 0.14.0", + "p384 0.14.0", + "p521 0.14.0", + "pageant", + "pbkdf2", + "pkcs1 0.8.0-rc.4", + "pkcs5", + "pkcs8 0.11.0", + "polyval 0.7.3", + "rand 0.10.2", + "rand_core 0.10.1", + "rsa 0.10.0-rc.18", + "russh-cryptovec", + "russh-util", + "salsa20", + "scrypt", + "sec1 0.8.1", + "sha1 0.11.0", + "sha2 0.11.0", + "sha3 0.12.0", + "signature 3.0.0", + "spki 0.8.0", + "ssh-encoding 0.3.0", + "ssh-key 0.7.0-rc.11", + "subtle", + "thiserror 2.0.20", + "tokio", + "typenum", + "universal-hash 0.6.1", + "zeroize", +] + +[[package]] +name = "russh" +version = "0.63.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "35bab1b87d915817d5d9cc352637cd40d5f0b298a48c6309af9156a4addc3031" +dependencies = [ + "aes 0.9.2", + "aws-lc-rs", + "bitflags 2.13.1", + "block-padding 0.4.2", + "byteorder", + "bytes", + "cbc", + "cipher 0.5.2", + "crypto-bigint 0.7.5", + "ctr 0.10.1", + "curve25519-dalek 5.0.0", + "data-encoding", + "delegate", + "der 0.8.1", + "digest 0.11.3", + "ecdsa 0.17.0", + "ed25519-dalek 3.0.0", + "elliptic-curve 0.14.1", + "enum_dispatch", + "flate2", + "futures", + "generic-array 1.4.5", + "getrandom 0.4.3", + "ghash 0.6.0", + "hex-literal", + "hmac 0.13.0", + "inout 0.2.2", + "internal-russh-num-bigint", + "keccak 0.2.2", "log", "md5", "ml-kem", @@ -7528,9 +7602,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" [[package]] name = "rustls-webpki" -version = "0.103.14" +version = "0.103.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0527518605e68109d875e248ea259b6758801cf165e4b2c2733ae3b51f12535a" +checksum = "f3c3cf1d8b1e7d4927e2d154c3fcb02979afb9939629c62cd9048d4f07b60ac2" dependencies = [ "aws-lc-rs", "ring", @@ -7981,7 +8055,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -8147,7 +8221,7 @@ checksum = "a22144e767da4ddd8416dbf383700542ffd8a5dc493dfecedfe1fe3ad03c98ae" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -8228,7 +8302,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be176f1a57ce4e3d31c1a166222d9768de5954f811601fb7ca06fc8203905ce1" dependencies = [ "digest 0.11.3", - "keccak 0.2.1", + "keccak 0.2.2", ] [[package]] @@ -8238,7 +8312,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bc9bad02c26382724b2d2692c6f179285e4b54eeecd7968f52a50059c3c11759" dependencies = [ "digest 0.11.3", - "keccak 0.2.1", + "keccak 0.2.2", "sponge-cursor", ] @@ -8675,7 +8749,7 @@ checksum = "d801accda99469cde6d73da741422610fdf6508a72d9a69d1b55cb241c720597" dependencies = [ "aead 0.6.1", "aes 0.9.2", - "aes-gcm 0.11.0", + "aes-gcm 0.11.1", "chacha20", "cipher 0.5.2", "ctutils", @@ -8941,9 +9015,9 @@ dependencies = [ [[package]] name = "syn" -version = "3.0.3" +version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +checksum = "e6275cddf4610d1775e6d1fe9469b2e77d0f39fd98fb7450901b821e0c53649f" dependencies = [ "proc-macro2", "quote", @@ -9138,7 +9212,7 @@ checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -9248,7 +9322,7 @@ checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] @@ -10042,9 +10116,9 @@ checksum = "e2eebbbfe4093922c2b6734d7c679ebfebd704a0d7e56dfcb0d05818ce28977d" [[package]] name = "uuid" -version = "1.24.1" +version = "1.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9" +checksum = "f053576934f05a761a402421fbbe3d425d9366f75f978806a037b3ca481abecc" dependencies = [ "getrandom 0.4.3", "js-sys", @@ -10909,13 +10983,13 @@ dependencies = [ [[package]] name = "zerovec-derive" -version = "0.11.5" +version = "0.11.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f212a141d820099d57ffafb9569be9617a6f27d3dc881fbee8fb56642f917a9" +checksum = "34df6fc39dbd26ddc9c10e6a2984476e13acce22e64e4487636ef494369225da" dependencies = [ "proc-macro2", "quote", - "syn 3.0.3", + "syn 3.0.4", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index c0136d3dd..b5d6f3c1a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -63,7 +63,7 @@ futures = "0.3.34" futures-util = "0.3.34" axum = { version = "0.8.9", features = ["macros", "json"] } axum-extra = "0.12.6" -russh = "0.62.7" +russh = "0.63.1" tower-http = "0.7.0" tower = "0.5.3" tower-sessions = { version = "0.15", features = ["memory-store"] } diff --git a/scripts/crates-sync/Dockerfile b/scripts/crates-sync/Dockerfile new file mode 100644 index 000000000..e1a79f377 --- /dev/null +++ b/scripts/crates-sync/Dockerfile @@ -0,0 +1,24 @@ +# crates-sync: python3 + git for the K8s crates import Job. +# Build from repo root: +# docker build -f scripts/crates-sync/Dockerfile -t mega/crates-sync scripts/crates-sync +# Or from this directory: +# docker build -t mega/crates-sync . +FROM python:3.12-slim-bookworm + +# Debian slim defaults to http://deb.debian.org. Through corporate proxies: +# - plain HTTP hangs +# - HTTPS works but MITM breaks apt's CA verify +# Switch to HTTPS and relax peer verify only for this apt fetch. +RUN sed -i 's|http://deb.debian.org|https://deb.debian.org|g' /etc/apt/sources.list.d/debian.sources \ + && apt-get -o Acquire::https::Verify-Peer=false -o Acquire::https::Verify-Host=false update -qq \ + && apt-get -o Acquire::https::Verify-Peer=false -o Acquire::https::Verify-Host=false install -y --no-install-recommends \ + ca-certificates \ + git \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +# Build context must be the scripts/crates-sync directory (this folder). +COPY . scripts/crates-sync/ + +ENTRYPOINT ["python3", "scripts/crates-sync/run_job.py"] diff --git a/scripts/crates-sync/README.md b/scripts/crates-sync/README.md index 0fab1182a..548357427 100644 --- a/scripts/crates-sync/README.md +++ b/scripts/crates-sync/README.md @@ -24,7 +24,7 @@ It reads a local `crates.io-index` checkout, downloads `.crate` tarballs, extrac - Python 3 - Git (installed and accessible from command line) -## Installation +## Installation 1. Clone this repository or download the script. 2. Ensure `git` is available on your PATH. @@ -200,3 +200,46 @@ Notes: - The manifest ensures already-imported `crate@version` entries are skipped on subsequent runs. - Adjust `--jobs` based on your CPU/IO and Mega server capacity. +- Use `--keep-crate-cache` when `--crates-dir` is a shared host cache (e.g. freighter) so successful pushes do not delete `.crate` files. + +## Kubernetes Job (onprem) + +CI builds `mega/crates-sync` from `scripts/crates-sync/Dockerfile` (workflow: `.github/workflows/crates-sync-deploy.yml`). + +The Job entrypoint is `run_job.py`: wait for mono → `bootstrap-init` bot token → optional `git pull` on index → `crates-sync.py` with `--keep-crate-cache` and `--max-versions-per-crate 0` (all versions). + +On **mega-rust**, data is expected on the node hostPath: + +| Host path | Container mount | +|-----------|-----------------| +| `/opt/data/freighter` | `/freighter` | +| `.../crates.io-index` | `--index` | +| `.../crates` | `--crates-dir` | +| `.../mega-crates-work` | `--workdir` + manifest | + +Terraform (`gitmono_stack`): + +```hcl +enable_crates_sync = true +crates_sync_freighter_host_path = "/opt/data/freighter" +crates_sync_node_hostname = "storage-server-01" # must match kubectl get nodes +# crates_sync_image = "registry.xuanwu.openatom.cn/mega/crates-sync:" +``` + +The Job is scheduled onto that node (`nodeAffinity`), mounts the freighter hostPath, and runs asynchronously (`wait_for_completion = false`). Re-run by bumping `crates_sync_image` (or delete the Job and re-apply). + +Manual run on the freighter host (no Job): + +```bash +export MEGA_TOKEN="..." +python3 scripts/crates-sync/crates-sync.py \ + --index /opt/data/freighter/crates.io-index \ + --crates-dir /opt/data/freighter/crates \ + --workdir /opt/data/freighter/mega-crates-work \ + --manifest /opt/data/freighter/mega-crates-work/crates-import-manifest.jsonl \ + --git-base-url https://git.rust.xuanwu.openatom.cn \ + --token "$MEGA_TOKEN" \ + --max-versions-per-crate 0 \ + --jobs 2 \ + --keep-crate-cache +``` diff --git a/scripts/crates-sync/crates-sync.py b/scripts/crates-sync/crates-sync.py index 373d616d1..02b289830 100644 --- a/scripts/crates-sync/crates-sync.py +++ b/scripts/crates-sync/crates-sync.py @@ -54,6 +54,17 @@ _push_fail_total = 0 _run_start_mono: float | None = None +# Overall import progress (shared with heartbeat). +_progress_lock = threading.Lock() +_progress_index_crates = 0 +_progress_versions_queued = 0 +_progress_versions_done = 0 +_progress_ok = 0 +_progress_skip = 0 +_progress_fail = 0 +_progress_scan_complete = False +_progress_total: int | None = None # known after scan finishes (or non-streaming start) + def _record_push_ok() -> None: now = time.monotonic() with _push_ok_lock: @@ -102,6 +113,92 @@ def _pushes_per_min_since_start() -> float: ok_total, fail_total = _push_totals() return (ok_total + fail_total) / max(1e-6, mins) +def _progress_reset() -> None: + global _progress_index_crates, _progress_versions_queued, _progress_versions_done + global _progress_ok, _progress_skip, _progress_fail + global _progress_scan_complete, _progress_total + with _progress_lock: + _progress_index_crates = 0 + _progress_versions_queued = 0 + _progress_versions_done = 0 + _progress_ok = 0 + _progress_skip = 0 + _progress_fail = 0 + _progress_scan_complete = False + _progress_total = None + +def _progress_note_index_crate() -> None: + global _progress_index_crates + with _progress_lock: + _progress_index_crates += 1 + +def _progress_note_queued(n: int = 1) -> None: + global _progress_versions_queued + with _progress_lock: + _progress_versions_queued += n + +def _progress_note_result(status: str) -> None: + global _progress_versions_done, _progress_ok, _progress_skip, _progress_fail + with _progress_lock: + _progress_versions_done += 1 + if status == "ok": + _progress_ok += 1 + elif status == "skip": + _progress_skip += 1 + else: + _progress_fail += 1 + +def _progress_mark_scan_complete() -> None: + global _progress_scan_complete, _progress_total + with _progress_lock: + _progress_scan_complete = True + _progress_total = _progress_versions_queued + +def _progress_set_total(total: int) -> None: + global _progress_scan_complete, _progress_total, _progress_versions_queued + with _progress_lock: + _progress_scan_complete = True + _progress_total = total + _progress_versions_queued = total + +def _format_eta(done: int, total: int | None) -> str: + if _run_start_mono is None or done <= 0 or not total or total <= done: + return "eta=?" + elapsed = max(1e-6, time.monotonic() - _run_start_mono) + rate = done / elapsed + remain = (total - done) / max(1e-9, rate) + if remain < 60: + return f"eta={remain:.0f}s" + if remain < 3600: + return f"eta={remain / 60:.1f}m" + return f"eta={remain / 3600:.1f}h" + +def _format_progress_line() -> str: + with _progress_lock: + crates = _progress_index_crates + queued = _progress_versions_queued + done = _progress_versions_done + ok_n = _progress_ok + skip_n = _progress_skip + fail_n = _progress_fail + scan_done = _progress_scan_complete + total = _progress_total + pending = max(0, queued - done) + if total and total > 0: + pct = (done / total) * 100.0 + return ( + f"progress: {done}/{total} ({pct:.1f}%) " + f"ok={ok_n} skip={skip_n} fail={fail_n} pending={pending} " + f"index_crates={crates} scan={'done' if scan_done else 'running'} " + f"{_format_eta(done, total)}" + ) + return ( + f"progress: done={done} queued={queued} pending={pending} " + f"ok={ok_n} skip={skip_n} fail={fail_n} " + f"index_crates={crates} scan={'done' if scan_done else 'running'} " + f"{_format_eta(done, None)}" + ) + def _clear_status_block_locked() -> None: global _status_block_active, _status_block_last_lens if not _status_block_active: @@ -145,6 +242,7 @@ def _format_status_block() -> list[str]: ok_total, fail_total = _push_totals() ppm = _pushes_per_min_since_start() return [ + _format_progress_line(), f"status: downloading={d} extracting={x} waiting_push={w} pushing={p}", ( f"push: ok_60s={ok60} fail_60s={fail60} " @@ -612,11 +710,10 @@ def process_crate_version( force: bool, force_with_lease: bool, push_sema: threading.Semaphore, + keep_crate_cache: bool = False, ) -> bool: - # Record start time for the entire crate - crate_start_time = datetime.now() if VERBOSE: - info(f"Started {crate_name} at {crate_start_time}") + info(f"Started {crate_name} at {datetime.now()}") # Process a specific version of a crate rel = mega_third_party_crates_rel_path(crate_name, version) @@ -693,16 +790,18 @@ def process_crate_version( warn(f"{_fmt_repo(crate_name, version)} push failed") _record_push_fail() return False - else: - ok(f"{_fmt_repo(crate_name, version)} pushed") - _record_push_ok() - # On success, remove local repo directory and cached crate to save disk space. - try: - shutil.rmtree(repo_path) - if VERBOSE: - info(f"Removed local repo: {repo_path}") - except Exception as e: - warn(f"Failed to remove local repo {repo_path}: {e}") + + ok(f"{_fmt_repo(crate_name, version)} pushed") + _record_push_ok() + # On success, remove local repo directory to save disk space. + try: + shutil.rmtree(repo_path) + if VERBOSE: + info(f"Removed local repo: {repo_path}") + except Exception as e: + warn(f"Failed to remove local repo {repo_path}: {e}") + # Optionally drop cached .crate (keep when sharing a host freighter cache). + if not keep_crate_cache: try: if os.path.exists(crate_path): os.remove(crate_path) @@ -710,17 +809,7 @@ def process_crate_version( info(f"Removed cached crate file: {crate_path}") except Exception as e: warn(f"Failed to remove cached crate file {crate_path}: {e}") - return True - - # Record end time and calculate duration for the entire crate - crate_end_time = datetime.now() - crate_duration = crate_end_time - crate_start_time - if VERBOSE: - info(f"Finished {crate_name} at {crate_end_time} (duration {crate_duration})") - - # Keep output minimal by default - if VERBOSE: - print("------------------") + return True def stream_index_crate_versions(index_path: str, max_versions_per_crate: int): """ @@ -769,6 +858,8 @@ def stream_index_crate_versions(index_path: str, max_versions_per_crate: int): if max_versions_per_crate > 0: vs = vs[-max_versions_per_crate:] + _progress_note_index_crate() + _progress_note_queued(len(vs)) yield crate_name, vs files_seen += 1 if files_seen % 1000 == 0: @@ -899,8 +990,10 @@ def scan_and_process_crates( manifest: Dict[Tuple[str, str], dict], manifest_path: str, reimport_ok: bool, -): + keep_crate_cache: bool = False, +) -> tuple[int, int, int]: info("Scanning crates.io index...") + _progress_reset() stop_evt = threading.Event() hb_thread = None @@ -908,158 +1001,174 @@ def scan_and_process_crates( hb_thread = threading.Thread(target=_heartbeat_thread, args=(stop_evt,), daemon=True) hb_thread.start() - # Read the config.json to get the dl base URL (needed before any processing) - config_path = os.path.join(index_path, "config.json") try: - with open(config_path, "r") as config_file: - config = json.load(config_file) - dl_base_url = config.get("dl") - if not dl_base_url: - warn("Error: 'dl' key not found in config.json") - sys.exit(1) - except Exception as e: - warn(f"Error reading config.json: {e}") - sys.exit(1) + # Read the config.json to get the dl base URL (needed before any processing) + config_path = os.path.join(index_path, "config.json") + try: + with open(config_path, "r") as config_file: + config = json.load(config_file) + dl_base_url = config.get("dl") + if not dl_base_url: + warn("Error: 'dl' key not found in config.json") + sys.exit(1) + except Exception as e: + warn(f"Error reading config.json: {e}") + sys.exit(1) - use_streaming_full_index = not only_crates and not (limit_crates and limit_crates > 0) - - if use_streaming_full_index: - info("Streaming index: will download/commit/push while walking crate files.") - elif only_crates: - crates = scan_selected_crates_index(index_path, only_crates) - info(f"Found {len(crates)} crates.") - crates_items = list(crates.items()) - allow = {c.strip() for c in only_crates if c.strip()} - crates_items = [(n, v) for (n, v) in crates_items if n in allow] - info(f"Filtered to {len(crates_items)} crates via --crate.") - else: - names = load_or_build_crate_name_cache(index_path, crate_name_cache) - random.shuffle(names) - picked = names[:limit_crates] - info(f"Sampling {len(picked)} crates via --limit-crates (no full content scan).") - crates = scan_selected_crates_index(index_path, picked) - info(f"Loaded {len(crates)} crates' versions.") - crates_items = list(crates.items()) - if VERBOSE: - info("Shuffling crates list...") - random.shuffle(crates_items) - - push_sema = threading.BoundedSemaphore(max(1, int(jobs))) - succeeded = 0 - failed = 0 - skipped = 0 - lock = threading.Lock() - - def process_one(crate_name: str, v: str) -> tuple[str, str, str]: - key = (crate_name, v) - rec = manifest.get(key) - # If already successfully imported, skip unless --reimport-ok (git --force does not disable this). - if rec and rec.get("status") == "ok" and not reimport_ok: - info(f"{_fmt_repo(crate_name, v)} already imported; skipping (manifest)") - return ("skip", crate_name, v) - - rel = mega_third_party_crates_rel_path(crate_name, v) - repo_path = os.path.join(git_repos_dir, rel) - - # Existing repo path - if os.path.exists(repo_path) and os.path.exists(os.path.join(repo_path, ".git")): - if repush_existing and not dry_run: - ok_push = ensure_remote_and_push_existing( - repo_path, - rel, + use_streaming_full_index = not only_crates and not (limit_crates and limit_crates > 0) + + if use_streaming_full_index: + info("Streaming index: will download/commit/push while walking crate files.") + elif only_crates: + crates = scan_selected_crates_index(index_path, only_crates) + info(f"Found {len(crates)} crates.") + crates_items = list(crates.items()) + allow = {c.strip() for c in only_crates if c.strip()} + crates_items = [(n, v) for (n, v) in crates_items if n in allow] + info(f"Filtered to {len(crates_items)} crates via --crate.") + else: + names = load_or_build_crate_name_cache(index_path, crate_name_cache) + random.shuffle(names) + picked = names[:limit_crates] + info(f"Sampling {len(picked)} crates via --limit-crates (no full content scan).") + crates = scan_selected_crates_index(index_path, picked) + info(f"Loaded {len(crates)} crates' versions.") + crates_items = list(crates.items()) + if VERBOSE: + info("Shuffling crates list...") + random.shuffle(crates_items) + + push_sema = threading.BoundedSemaphore(max(1, int(jobs))) + succeeded = 0 + failed = 0 + skipped = 0 + lock = threading.Lock() + + def process_one(crate_name: str, v: str) -> tuple[str, str, str]: + key = (crate_name, v) + rec = manifest.get(key) + # If already successfully imported, skip unless --reimport-ok (git --force does not disable this). + if rec and rec.get("status") == "ok" and not reimport_ok: + if VERBOSE: + info(f"{_fmt_repo(crate_name, v)} already imported; skipping (manifest)") + return ("skip", crate_name, v) + + rel = mega_third_party_crates_rel_path(crate_name, v) + repo_path = os.path.join(git_repos_dir, rel) + + # Existing repo path + if os.path.exists(repo_path) and os.path.exists(os.path.join(repo_path, ".git")): + if repush_existing and not dry_run: + ok_push = ensure_remote_and_push_existing( + repo_path, + rel, + git_base_url, + crate_name=crate_name, + version=v, + commit_signoff=commit_signoff, + auth_token=auth_token, + force=force, + force_with_lease=force_with_lease, + push_sema=push_sema, + ) + return ("ok" if ok_push else "fail", crate_name, v) + else: + if VERBOSE: + info(f"{_fmt_repo(crate_name, v)} exists; skipping") + return ("skip", crate_name, v) + + crate_path = check_and_download_crate(crates_dir, crate_name, v, dl_base_url) + if crate_path is None: + return ("fail", crate_name, v) + try: + ok_done = process_crate_version( + 0, + crate_name, + v, + crate_path, + git_repos_dir, git_base_url, - crate_name=crate_name, - version=v, commit_signoff=commit_signoff, + dry_run=dry_run, auth_token=auth_token, force=force, force_with_lease=force_with_lease, push_sema=push_sema, + keep_crate_cache=keep_crate_cache, ) - return ("ok" if ok_push else "fail", crate_name, v) - else: - info(f"{_fmt_repo(crate_name, v)} exists; skipping") - return ("skip", crate_name, v) - - crate_path = check_and_download_crate(crates_dir, crate_name, v, dl_base_url) - if crate_path is None: - return ("fail", crate_name, v) - try: - ok_done = process_crate_version( - 0, - crate_name, - v, - crate_path, - git_repos_dir, - git_base_url, - commit_signoff=commit_signoff, - dry_run=dry_run, - auth_token=auth_token, - force=force, - force_with_lease=force_with_lease, - push_sema=push_sema, - ) - return ("ok" if ok_done else "fail", crate_name, v) - except Exception as e: - warn(f"{_fmt_repo(crate_name, v)} failed: {e}") - return ("fail", crate_name, v) - - def record_result(fut) -> None: - nonlocal succeeded, failed, skipped - status, c_name, v = fut.result() - with lock: - if status == "ok": - succeeded += 1 - elif status == "skip": - skipped += 1 + return ("ok" if ok_done else "fail", crate_name, v) + except Exception as e: + warn(f"{_fmt_repo(crate_name, v)} failed: {e}") + return ("fail", crate_name, v) + + def record_result(fut) -> None: + nonlocal succeeded, failed, skipped + status, c_name, v = fut.result() + _progress_note_result(status) + with lock: + if status == "ok": + succeeded += 1 + elif status == "skip": + skipped += 1 + else: + failed += 1 + key = (c_name, v) + rel = mega_third_party_crates_rel_path(c_name, v) + rec = { + "crate": c_name, + "version": v, + "status": status, + "remote": f"{git_base_url.rstrip('/')}/{rel}", + "last_import_time": datetime.now(timezone.utc).isoformat(), + } + manifest[key] = rec + append_manifest_record(manifest_path, rec) + + # Concurrency: bounded pending futures in streaming mode to avoid RAM spikes. + with ThreadPoolExecutor(max_workers=max(1, int(jobs))) as ex: + if use_streaming_full_index: + max_pending = max(32, int(jobs) * 8) + pending: set = set() + for crate_name, versions in stream_index_crate_versions( + index_path, max_versions_per_crate + ): + for v in versions: + pending.add(ex.submit(process_one, crate_name, v)) + while len(pending) >= max_pending: + done, _ = wait(pending, return_when=FIRST_COMPLETED) + for df in done: + pending.discard(df) + record_result(df) + _progress_mark_scan_complete() + info( + f"Index scan complete: {_progress_index_crates} crates, " + f"{_progress_versions_queued} versions queued; draining workers..." + ) + for df in as_completed(pending): + record_result(df) else: - failed += 1 - key = (c_name, v) - rel = mega_third_party_crates_rel_path(c_name, v) - rec = { - "crate": c_name, - "version": v, - "status": status, - "remote": f"{git_base_url.rstrip('/')}/{rel}", - "last_import_time": datetime.now(timezone.utc).isoformat(), - } - manifest[key] = rec - append_manifest_record(manifest_path, rec) - - # Concurrency: bounded pending futures in streaming mode to avoid RAM spikes. - with ThreadPoolExecutor(max_workers=max(1, int(jobs))) as ex: - if use_streaming_full_index: - max_pending = max(32, int(jobs) * 8) - pending: set = set() - for crate_name, versions in stream_index_crate_versions( - index_path, max_versions_per_crate - ): - for v in versions: - pending.add(ex.submit(process_one, crate_name, v)) - while len(pending) >= max_pending: - done, _ = wait(pending, return_when=FIRST_COMPLETED) - for df in done: - pending.discard(df) - record_result(df) - for df in as_completed(pending): - record_result(df) - else: - tasks: list[tuple[str, str]] = [] - for crate_name, versions in crates_items: - vs = sorted(versions) - if max_versions_per_crate > 0: - vs = vs[-max_versions_per_crate:] - for v in vs: - tasks.append((crate_name, v)) - info(f"Starting to process {len(tasks)} crate versions...") - futures = [ex.submit(process_one, c, v) for c, v in tasks] - for f in as_completed(futures): - record_result(f) - - info(f"Summary: ok={succeeded}, skipped={skipped}, failed={failed}") - # Compact manifest (dedupe append-only history to one line per crate@version). - write_manifest(manifest_path, manifest) - return succeeded + skipped + failed + tasks: list[tuple[str, str]] = [] + for crate_name, versions in crates_items: + vs = sorted(versions) + if max_versions_per_crate > 0: + vs = vs[-max_versions_per_crate:] + for v in vs: + tasks.append((crate_name, v)) + _progress_set_total(len(tasks)) + info(f"Starting to process {len(tasks)} crate versions...") + futures = [ex.submit(process_one, c, v) for c, v in tasks] + for f in as_completed(futures): + record_result(f) + + info(f"Summary: ok={succeeded}, skipped={skipped}, failed={failed}") + info(_format_progress_line()) + # Compact manifest (dedupe append-only history to one line per crate@version). + write_manifest(manifest_path, manifest) + return succeeded, skipped, failed + finally: + stop_evt.set() + if hb_thread is not None: + hb_thread.join(timeout=max(1.0, float(STATUS_HEARTBEAT_INTERVAL_S) + 1.0)) def main(): @@ -1166,6 +1275,11 @@ def main(): action="store_true", help="Re-process crate versions even if manifest status is ok (default: skip ok).", ) + p.add_argument( + "--keep-crate-cache", + action="store_true", + help="Do not delete downloaded .crate files after a successful push (for shared host caches).", + ) args = p.parse_args() global VERBOSE @@ -1200,7 +1314,7 @@ def main(): manifest_path = str(Path(git_repos_dir) / "crates-import-manifest.jsonl") manifest = load_manifest(manifest_path) - total_crates = scan_and_process_crates( + succeeded, skipped, failed = scan_and_process_crates( index_path, crates_dir, git_repos_dir, @@ -1219,13 +1333,17 @@ def main(): manifest=manifest, manifest_path=manifest_path, reimport_ok=args.reimport_ok, + keep_crate_cache=bool(args.keep_crate_cache), ) # Record end time and calculate duration for the entire process total_end_time = datetime.now() total_duration = total_end_time - total_start_time - info(f"Total processed: {total_crates}") + total_crates = succeeded + skipped + failed + info(f"Total processed: {total_crates} (ok={succeeded}, skipped={skipped}, failed={failed})") info(f"Finished at {total_end_time} (duration {total_duration})") + if failed > 0: + sys.exit(1) if __name__ == "__main__": main() # Run the main function if this script is executed directly diff --git a/scripts/crates-sync/run_job.py b/scripts/crates-sync/run_job.py new file mode 100644 index 000000000..79a7ad508 --- /dev/null +++ b/scripts/crates-sync/run_job.py @@ -0,0 +1,256 @@ +#!/usr/bin/env python3 +"""K8s Job entrypoint for crates-sync. + +Waits for mono-engine, bootstraps a bot push token via MEGA_INIT_BOOTSTRAP_SECRET, +optionally refreshes a local crates.io-index checkout, then runs crates-sync.py. + +Typical freighter hostPath layout (mounted at --freighter-root): + + /crates.io-index -> --index + /crates -> --crates-dir + /mega-crates-work -> --workdir (+ manifest) +""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +import time +import urllib.error +import urllib.request +from pathlib import Path + +INIT_BOOTSTRAP_SECRET_ENV = "MEGA_INIT_BOOTSTRAP_SECRET" +INIT_BOOTSTRAP_SECRET_HEADER = "X-Mega-Init-Secret" +INIT_BOOTSTRAP_SECRET_MIN_LEN = 32 + +SCRIPT_DIR = Path(__file__).resolve().parent +CRATES_SYNC_PY = SCRIPT_DIR / "crates-sync.py" + + +def api_request(method, url, data=None, headers=None, timeout=10): + if headers is None: + headers = {} + if "accept" not in headers: + headers["accept"] = "application/json" + req_data = None + if data is not None: + req_data = json.dumps(data).encode("utf-8") + if "Content-Type" not in headers: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=req_data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=timeout) as response: + resp_body = response.read().decode("utf-8") + if 200 <= response.status < 300: + return json.loads(resp_body) if resp_body else {} + raise RuntimeError(f"API request failed with status {response.status}: {resp_body}") + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + raise RuntimeError(f"API request to {url} failed: HTTP Error {e.code}: {e.reason}; body={body}") from e + except Exception as e: + raise RuntimeError(f"API request to {url} failed: {e}") from e + + +def wait_for_server(base_url, timeout=300): + status_url = f"{base_url.rstrip('/')}/api/v1/status" + start = time.time() + print(f"Waiting for server at {status_url}...") + while time.time() - start < timeout: + try: + api_request("GET", status_url) + print("Server is ready.") + return + except Exception: + time.sleep(2) + raise RuntimeError(f"Server at {base_url} did not become ready within {timeout}s") + + +def resolve_init_bootstrap_secret(cli_secret=None): + secret = (cli_secret or "").strip() or os.environ.get(INIT_BOOTSTRAP_SECRET_ENV, "").strip() + if not secret: + raise RuntimeError( + f"bootstrap-init requires {INIT_BOOTSTRAP_SECRET_ENV} " + "(or --init-secret); must match mono-engine" + ) + if len(secret) < INIT_BOOTSTRAP_SECRET_MIN_LEN: + raise RuntimeError( + f"{INIT_BOOTSTRAP_SECRET_ENV} must be at least " + f"{INIT_BOOTSTRAP_SECRET_MIN_LEN} characters" + ) + return secret + + +def bootstrap_init_bot_token(base_url, init_secret): + url = f"{base_url.rstrip('/')}/api/v1/bots/bootstrap-init" + print(f"Bootstrapping init bot token via {url}...") + resp = api_request( + "POST", + url, + data={}, + headers={INIT_BOOTSTRAP_SECRET_HEADER: init_secret}, + timeout=120, + ) + if not resp.get("req_result"): + raise RuntimeError(f"bootstrap-init failed: {resp.get('err_message') or resp}") + data = resp.get("data") or {} + token = data.get("token") + if not token: + raise RuntimeError(f"bootstrap-init returned no token: {resp}") + print(f"Got bot token for bot_name={data.get('bot_name')} bot_id={data.get('bot_id')}") + return token + + +def maybe_pull_index(index_path: Path) -> None: + git_dir = index_path / ".git" + if not git_dir.exists(): + print(f"Index at {index_path} is not a git checkout; skipping pull.") + return + print(f"Refreshing crates.io-index at {index_path}...") + result = subprocess.run( + ["git", "-C", str(index_path), "pull", "--ff-only"], + capture_output=True, + text=True, + ) + if result.returncode != 0: + print( + f"Warning: git pull --ff-only failed (continuing with existing index): " + f"{(result.stderr or result.stdout or '').strip()}" + ) + else: + print((result.stdout or "").strip() or "Index up to date.") + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(prog="run_job.py") + p.add_argument( + "--base-url", + default=os.environ.get("MEGA_BASE_URL", "http://mono-engine:8000"), + help="Mega mono-engine base URL (ClusterIP in-cluster).", + ) + p.add_argument( + "--freighter-root", + default=os.environ.get("CRATES_SYNC_FREIGHTER_ROOT", "/freighter"), + help="Host freighter mount root (default: /freighter).", + ) + p.add_argument("--index", default="", help="Override index path (default: /crates.io-index).") + p.add_argument("--crates-dir", default="", help="Override crates cache (default: /crates).") + p.add_argument( + "--workdir", + default="", + help="Override workdir (default: /mega-crates-work).", + ) + p.add_argument("--manifest", default="", help="Override manifest path.") + p.add_argument("--init-secret", default="", help="Bootstrap secret (or env MEGA_INIT_BOOTSTRAP_SECRET).") + p.add_argument( + "--jobs", + type=int, + default=2, + help="Concurrent workers passed to crates-sync (default: 2).", + ) + p.add_argument( + "--max-versions-per-crate", + type=int, + default=0, + help="0 = all versions per crate (default for Job).", + ) + p.add_argument( + "--no-pull-index", + action="store_true", + help="Do not attempt git pull on the index checkout.", + ) + p.add_argument( + "--wait-timeout", + type=int, + default=300, + help="Seconds to wait for mono /api/v1/status (default: 300).", + ) + args, extra = p.parse_known_args(argv) + + freighter = Path(args.freighter_root) + index_path = Path(args.index) if args.index else freighter / "crates.io-index" + crates_dir = Path(args.crates_dir) if args.crates_dir else freighter / "crates" + workdir = Path(args.workdir) if args.workdir else freighter / "mega-crates-work" + manifest = ( + Path(args.manifest) + if args.manifest + else workdir / "crates-import-manifest.jsonl" + ) + + if not index_path.is_dir(): + raise SystemExit(f"Index directory not found: {index_path}") + if not (index_path / "config.json").is_file(): + raise SystemExit(f"Index config.json not found under {index_path}") + crates_dir.mkdir(parents=True, exist_ok=True) + workdir.mkdir(parents=True, exist_ok=True) + + base_url = args.base_url.rstrip("/") + wait_for_server(base_url, timeout=args.wait_timeout) + init_secret = resolve_init_bootstrap_secret(args.init_secret) + token = bootstrap_init_bot_token(base_url, init_secret) + + if not args.no_pull_index: + maybe_pull_index(index_path) + + if not CRATES_SYNC_PY.is_file(): + raise SystemExit(f"crates-sync.py not found next to run_job.py: {CRATES_SYNC_PY}") + + cmd = [ + sys.executable, + str(CRATES_SYNC_PY), + "--index", + str(index_path), + "--crates-dir", + str(crates_dir), + "--workdir", + str(workdir), + "--manifest", + str(manifest), + "--git-base-url", + base_url, + "--token", + token, + "--max-versions-per-crate", + str(args.max_versions_per_crate), + "--jobs", + str(args.jobs), + "--keep-crate-cache", + "--no-status-sticky", + ] + cmd.extend(extra) + + printable = [ + sys.executable, + str(CRATES_SYNC_PY), + "--index", + str(index_path), + "--crates-dir", + str(crates_dir), + "--workdir", + str(workdir), + "--manifest", + str(manifest), + "--git-base-url", + base_url, + "--token", + "***", + "--max-versions-per-crate", + str(args.max_versions_per_crate), + "--jobs", + str(args.jobs), + "--keep-crate-cache", + "--no-status-sticky", + *extra, + ] + print("Running:", " ".join(printable)) + return subprocess.call(cmd) + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except KeyboardInterrupt: + raise SystemExit(130) diff --git a/scripts/demo/README.md b/scripts/demo/README.md index 0888262e7..68eb8efcc 100644 --- a/scripts/demo/README.md +++ b/scripts/demo/README.md @@ -121,6 +121,7 @@ set TARGET_PLATFORMS=linux/amd64 | `mega/orion-server` | `orion-server/Dockerfile` | `.` (repo root) | `latest` | | `mega/mega-ui` | `moon/apps/web/Dockerfile` | `moon` | `latest` | | `mega/mega-init` | `scripts/init_mega/Dockerfile` | `.` (repo root) | `latest` | +| `mega/crates-sync` | `scripts/crates-sync/Dockerfile` | `scripts/crates-sync` | `latest` | ## Image Tags diff --git a/scripts/demo/build-demo-images-local.sh b/scripts/demo/build-demo-images-local.sh index 2ec279f01..107ed489b 100644 --- a/scripts/demo/build-demo-images-local.sh +++ b/scripts/demo/build-demo-images-local.sh @@ -82,13 +82,14 @@ else fi # Image configurations (ordered for consistent build order) -declare -a IMAGE_ORDER=("mono-engine" "orion-server" "mega-ui" "mega-init") +declare -a IMAGE_ORDER=("mono-engine" "orion-server" "mega-ui" "mega-init" "crates-sync") get_image_config() { case "$1" in "mono-engine") echo "mono/Dockerfile:." ;; "mega-ui") echo "moon/apps/web/Dockerfile:moon" ;; "orion-server") echo "orion-server/Dockerfile:." ;; "mega-init") echo "scripts/init_mega/Dockerfile:." ;; + "crates-sync") echo "scripts/crates-sync/Dockerfile:scripts/crates-sync" ;; esac } @@ -98,12 +99,13 @@ get_image_tag() { "mega-ui") echo "latest" ;; "orion-server") echo "latest" ;; "mega-init") echo "latest" ;; + "crates-sync") echo "latest" ;; esac } is_valid_image() { case "$1" in - "mono-engine"|"mega-ui"|"orion-server"|"mega-init") return 0 ;; + "mono-engine"|"mega-ui"|"orion-server"|"mega-init"|"crates-sync") return 0 ;; *) return 1 ;; esac } diff --git a/vault/src/pki.rs b/vault/src/pki.rs index fdd252edc..b8a59b1e8 100644 --- a/vault/src/pki.rs +++ b/vault/src/pki.rs @@ -48,7 +48,7 @@ impl VaultCore { .expect("Failed to mount pki backend"); } - /// generate root cert, so that you can read from `pki/ca/pem` + /// generate root cert, so that you can read from `pki/ca/tls/pem` /// - if `exported` is true, then the response will contain `private key` async fn generate_root(&self, exported: bool) { let key_type = "rsa"; From 7ffbce4ac73d732270e68d78fe5ba76386886afc Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Tue, 25 Aug 2026 10:42:18 +0800 Subject: [PATCH 2/4] update crates sync script --- scripts/crates-sync/crates-sync.py | 37 +++++++++++++++++++----------- scripts/crates-sync/run_job.py | 4 ++-- 2 files changed, 25 insertions(+), 16 deletions(-) diff --git a/scripts/crates-sync/crates-sync.py b/scripts/crates-sync/crates-sync.py index 02b289830..c4e0c5ec4 100644 --- a/scripts/crates-sync/crates-sync.py +++ b/scripts/crates-sync/crates-sync.py @@ -273,18 +273,26 @@ def _stage_counts() -> tuple[int, int, int, int]: ) def _heartbeat_thread(stop_evt: threading.Event) -> None: - # Periodically emit a compact summary of what the script is doing. - # This is meant to answer "where is it stuck?" at a glance. - while not stop_evt.wait(max(0.5, float(STATUS_HEARTBEAT_INTERVAL_S))): + # Periodically refresh the sticky progress footer (always at bottom of stderr). + # Also paints once immediately so the footer exists before the first interval elapses. + while True: lines = _format_status_block() - if STATUS_STICKY: - with _print_lock: + with _print_lock: + if STATUS_STICKY: _render_status_block_locked(lines) - else: - info(" | ".join(lines)) + else: + print("\n".join(lines), file=sys.stderr, flush=True) + if stop_evt.wait(max(0.5, float(STATUS_HEARTBEAT_INTERVAL_S))): + break + if STATUS_STICKY: + with _print_lock: + _clear_status_block_locked() + # Final snapshot as normal lines so the last state remains in scrollback. + print("\n".join(_format_status_block()), file=sys.stderr, flush=True) def _log(level: str, msg: str) -> None: # Standardized, low-noise logging. Use --verbose for command outputs. + # Always write to stderr so sticky status (also on stderr) and logs share one stream. if level == "INFO": c = BLUE elif level == "WARN": @@ -294,15 +302,14 @@ def _log(level: str, msg: str) -> None: else: c = "" prefix = f"[{level}]" + line = f"{c}{prefix}{RESET} {msg}" if c else f"{prefix} {msg}" with _print_lock: if STATUS_STICKY: _clear_status_block_locked() - if c: - print(f"{c}{prefix}{RESET} {msg}") - else: - print(f"{prefix} {msg}") - if STATUS_STICKY and _status_block_last_lines: - _render_status_block_locked(_status_block_last_lines) + print(line, file=sys.stderr, flush=True) + if STATUS_STICKY: + # Re-paint with fresh progress so OK/INFO lines don't leave a stale block. + _render_status_block_locked(_format_status_block()) def info(msg: str) -> None: _log("INFO", msg) @@ -791,7 +798,6 @@ def process_crate_version( _record_push_fail() return False - ok(f"{_fmt_repo(crate_name, version)} pushed") _record_push_ok() # On success, remove local repo directory to save disk space. try: @@ -1105,6 +1111,9 @@ def record_result(fut) -> None: nonlocal succeeded, failed, skipped status, c_name, v = fut.result() _progress_note_result(status) + # Print after progress counters update so sticky footer shows the new totals. + if status == "ok": + ok(f"{_fmt_repo(c_name, v)} pushed") with lock: if status == "ok": succeeded += 1 diff --git a/scripts/crates-sync/run_job.py b/scripts/crates-sync/run_job.py index 79a7ad508..af1a51584 100644 --- a/scripts/crates-sync/run_job.py +++ b/scripts/crates-sync/run_job.py @@ -218,7 +218,7 @@ def main(argv: list[str] | None = None) -> int: "--jobs", str(args.jobs), "--keep-crate-cache", - "--no-status-sticky", + "--status-sticky", ] cmd.extend(extra) @@ -242,7 +242,7 @@ def main(argv: list[str] | None = None) -> int: "--jobs", str(args.jobs), "--keep-crate-cache", - "--no-status-sticky", + "--status-sticky", *extra, ] print("Running:", " ".join(printable)) From 8a0d711aa2683cc98585b8412428ab6c6264f108 Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Tue, 25 Aug 2026 10:56:03 +0800 Subject: [PATCH 3/4] Reject nested import repo creates under or over existing paths. Prevents parent paths like /third-party/rust from shadowing leaf import repos in tree browsing via find_git_repo_like_path prefix matching. Co-authored-by: Cursor --- ceres/src/transport/protocol/mod.rs | 22 ++++++- common/src/utils.rs | 87 +++++++++++++++++++++++++++ jupiter/src/storage/git_db_storage.rs | 55 ++++++++++++++++- 3 files changed, 161 insertions(+), 3 deletions(-) diff --git a/ceres/src/transport/protocol/mod.rs b/ceres/src/transport/protocol/mod.rs index ec0e825c3..756bf7606 100644 --- a/ceres/src/transport/protocol/mod.rs +++ b/ceres/src/transport/protocol/mod.rs @@ -7,7 +7,10 @@ use std::{ }; use callisto::sea_orm_active_enums::RefTypeEnum; -use common::errors::{MegaError, ProtocolError}; +use common::{ + errors::{MegaError, ProtocolError}, + utils::nested_import_repo_conflict_message, +}; use import_refs::RefCommand; use jupiter::redis::lock::RedLock; use repo::Repo; @@ -176,8 +179,23 @@ impl SmartSession { return Err(ProtocolError::NotFound("Repository not found.".to_owned())); } ServiceType::ReceivePack => { + if let Some(conflict) = storage + .find_nested_import_repo_conflict(path_str) + .await + .map_err(|e| { + ProtocolError::InvalidInput(format!( + "failed to check nested import repo conflict: {e}" + )) + })? + { + return Err(ProtocolError::InvalidInput( + nested_import_repo_conflict_message(path_str, &conflict.repo_path), + )); + } let repo = Repo::new(self.repo_path.clone(), false); - storage.save_git_repo(repo.clone().into()).await.unwrap(); + storage.save_git_repo(repo.clone().into()).await.map_err(|e| { + ProtocolError::InvalidInput(format!("failed to create import repo: {e}")) + })?; repo } } diff --git a/common/src/utils.rs b/common/src/utils.rs index 028f9d3e5..9ab41787e 100644 --- a/common/src/utils.rs +++ b/common/src/utils.rs @@ -130,10 +130,97 @@ pub fn get_current_bin_name() -> String { .to_owned() } +/// True when `parent` is a strict path-segment prefix of `child`. +/// +/// Equal paths return false. `/third-party/rust` is not a prefix of +/// `/third-party/rust_v1`. +pub fn is_strict_path_prefix(parent: &str, child: &str) -> bool { + let parent = parent.trim_end_matches('/'); + let child = child.trim_end_matches('/'); + if parent.is_empty() || child.is_empty() || parent == child { + return false; + } + child.starts_with(parent) && child.as_bytes().get(parent.len()) == Some(&b'/') +} + +/// If `new_path` would nest under or above any path in `existing`, return that +/// conflicting path. Same-path entries are ignored (update / idempotent create). +pub fn nested_import_repo_conflict<'a>( + new_path: &str, + existing: impl IntoIterator, +) -> Option<&'a str> { + let new_path = new_path.trim_end_matches('/'); + for path in existing { + let trimmed = path.trim_end_matches('/'); + if trimmed.is_empty() || trimmed == new_path { + continue; + } + if is_strict_path_prefix(trimmed, new_path) || is_strict_path_prefix(new_path, trimmed) { + return Some(path); + } + } + None +} + +/// Human-readable error when creating an import repo would nest with an existing one. +pub fn nested_import_repo_conflict_message(new_path: &str, conflicting_path: &str) -> String { + format!( + "cannot create import repo at '{new_path}': nested with existing import repo at '{conflicting_path}'" + ) +} + #[cfg(test)] mod test { use super::*; + #[test] + fn test_is_strict_path_prefix() { + assert!(is_strict_path_prefix( + "/third-party/rust", + "/third-party/rust/crates/sw/ay/swayws/1.3.0" + )); + assert!(is_strict_path_prefix( + "/third-party/rust/", + "/third-party/rust/crates" + )); + assert!(!is_strict_path_prefix("/third-party/rust", "/third-party/rust")); + assert!(!is_strict_path_prefix("/third-party/rust", "/third-party/rust_v1")); + assert!(!is_strict_path_prefix("/third-party/rust_v1", "/third-party/rust")); + assert!(!is_strict_path_prefix("/third-party/foo", "/third-party/bar")); + } + + #[test] + fn test_nested_import_repo_conflict() { + let existing = [ + "/third-party/rust/crates/sw/ay/swayws/1.3.0", + "/third-party/rust_v1", + ]; + assert_eq!( + nested_import_repo_conflict("/third-party/rust", existing), + Some("/third-party/rust/crates/sw/ay/swayws/1.3.0") + ); + assert_eq!( + nested_import_repo_conflict( + "/third-party/rust/crates/to/ki/tokio/1.0.0", + ["/third-party/rust"] + ), + Some("/third-party/rust") + ); + assert!( + nested_import_repo_conflict("/third-party/foo", existing).is_none() + ); + assert!( + nested_import_repo_conflict("/third-party/rust_v1", existing).is_none() + ); + assert!( + nested_import_repo_conflict( + "/third-party/rust/crates/sw/ay/swayws/1.3.0", + existing + ) + .is_none() + ); + } + #[test] fn test_is_full_hex_object_id() { // Valid SHA-1 (40 hex) diff --git a/jupiter/src/storage/git_db_storage.rs b/jupiter/src/storage/git_db_storage.rs index 70ff80238..c81c817a9 100644 --- a/jupiter/src/storage/git_db_storage.rs +++ b/jupiter/src/storage/git_db_storage.rs @@ -5,7 +5,12 @@ use callisto::{ git_blob, git_commit, git_repo, git_tag, git_tree, import_refs, sea_orm_active_enums::RefTypeEnum, }; -use common::{errors::MegaError, utils::generate_id}; +use common::{ + errors::MegaError, + utils::{ + generate_id, nested_import_repo_conflict_message, + }, +}; use futures::Stream; use sea_orm::{ ActiveModelTrait, ColumnTrait, DatabaseTransaction, DbBackend, DbErr, EntityTrait, @@ -41,6 +46,12 @@ impl GitDbStorage { let repo_id = if let Some(existing) = self.find_git_repo_exact_match(repo_path).await? { existing.id } else { + if let Some(conflict) = self.find_nested_import_repo_conflict(repo_path).await? { + return Err(MegaError::Conflict(nested_import_repo_conflict_message( + repo_path, + &conflict.repo_path, + ))); + } let repo_id = generate_id(); let repo = git_repo::Model { id: repo_id, @@ -293,6 +304,48 @@ impl GitDbStorage { Ok(result) } + /// Returns an existing import repo that would nest with `repo_path` if a new + /// import repo were created there (ancestor or descendant by path segment). + /// + /// Same-path repos are not conflicts (caller should use exact match first). + pub async fn find_nested_import_repo_conflict( + &self, + repo_path: &str, + ) -> Result, MegaError> { + let path = repo_path.trim_end_matches('/'); + if path.is_empty() || path == "/" { + return Ok(None); + } + + // Descendant of the new path (new path would become a parent import repo). + if let Some(descendant) = git_repo::Entity::find() + .filter(git_repo::Column::RepoPath.like(format!("{path}/%"))) + .one(self.get_connection()) + .await? + { + return Ok(Some(descendant)); + } + + // Ancestor of the new path (new path would nest under an existing import repo). + let mut current = std::path::PathBuf::from(path); + while current.pop() { + let parent = current.to_string_lossy(); + let parent = if parent.is_empty() { + "/".to_string() + } else { + parent.to_string() + }; + if parent == "/" { + break; + } + if let Some(ancestor) = self.find_git_repo_exact_match(&parent).await? { + return Ok(Some(ancestor)); + } + } + + Ok(None) + } + /// Finds a Git repository with a path that matches the beginning of the provided repository path using a LIKE query. /// /// # Arguments From cb27a7710fce79326569ef9fd0464b374d2f7cc5 Mon Sep 17 00:00:00 2001 From: "benjamin.747" Date: Tue, 25 Aug 2026 11:26:40 +0800 Subject: [PATCH 4/4] update crates sync progess --- scripts/crates-sync/README.md | 2 + scripts/crates-sync/crates-sync.py | 204 +++++++--- ...12\346\211\213\346\214\207\345\215\227.md" | 368 ++++++++++++++++++ 3 files changed, 526 insertions(+), 48 deletions(-) create mode 100644 "scripts/crates-sync/\344\270\212\346\211\213\346\214\207\345\215\227.md" diff --git a/scripts/crates-sync/README.md b/scripts/crates-sync/README.md index 548357427..19a58fe9f 100644 --- a/scripts/crates-sync/README.md +++ b/scripts/crates-sync/README.md @@ -1,5 +1,7 @@ # Crates.io crates import (third-party/rust/crates) +> **中文上手（架构 / 本地脚本 / Terraform Job 调试）：** [上手指南.md](./上手指南.md) + ## Overview This script imports crates from `crates.io` into Mega as **path-based git repositories** under: diff --git a/scripts/crates-sync/crates-sync.py b/scripts/crates-sync/crates-sync.py index c4e0c5ec4..7b9fe09fd 100644 --- a/scripts/crates-sync/crates-sync.py +++ b/scripts/crates-sync/crates-sync.py @@ -5,6 +5,7 @@ import argparse import json import os +import queue import random import shutil import subprocess @@ -13,7 +14,7 @@ import time import urllib.request from collections import defaultdict -from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, as_completed, wait +from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone from pathlib import Path import threading @@ -64,6 +65,8 @@ _progress_fail = 0 _progress_scan_complete = False _progress_total: int | None = None # known after scan finishes (or non-streaming start) +_progress_jobs = 1 +_work_queue: queue.Queue | None = None def _record_push_ok() -> None: now = time.monotonic() @@ -173,6 +176,18 @@ def _format_eta(done: int, total: int | None) -> str: return f"eta={remain / 60:.1f}m" return f"eta={remain / 3600:.1f}h" +def _format_progress_bar(done: int, total: int | None, width: int = 30) -> str: + """ASCII progress bar, e.g. [############--------------] 40.0%""" + if not total or total <= 0: + return f"[{'-' * width}] ?.??%" + frac = min(1.0, max(0.0, done / total)) + filled = int(round(width * frac)) + if done > 0 and filled == 0: + filled = 1 + filled = min(width, filled) + bar = "#" * filled + "-" * (width - filled) + return f"[{bar}] {frac * 100.0:5.1f}%" + def _format_progress_line() -> str: with _progress_lock: crates = _progress_index_crates @@ -183,20 +198,24 @@ def _format_progress_line() -> str: fail_n = _progress_fail scan_done = _progress_scan_complete total = _progress_total + jobs = _progress_jobs + # Denominator = final total after scan, else versions discovered so far (grows toward ~2M). + denom = total if (total and total > 0) else (queued if queued > 0 else None) pending = max(0, queued - done) - if total and total > 0: - pct = (done / total) * 100.0 - return ( - f"progress: {done}/{total} ({pct:.1f}%) " - f"ok={ok_n} skip={skip_n} fail={fail_n} pending={pending} " - f"index_crates={crates} scan={'done' if scan_done else 'running'} " - f"{_format_eta(done, total)}" - ) + qdepth = 0 + if _work_queue is not None: + try: + qdepth = _work_queue.qsize() + except NotImplementedError: + qdepth = pending + bar = _format_progress_bar(done, denom) + scan_tag = "scan=done" if scan_done else "scan=running" + count_s = f"{done}/{denom}" if denom else f"done={done}" return ( - f"progress: done={done} queued={queued} pending={pending} " - f"ok={ok_n} skip={skip_n} fail={fail_n} " - f"index_crates={crates} scan={'done' if scan_done else 'running'} " - f"{_format_eta(done, None)}" + f"progress: {bar} {count_s} " + f"jobs={jobs} ok={ok_n} skip={skip_n} fail={fail_n} " + f"queue={qdepth} crates={crates} {scan_tag} " + f"{_format_eta(done, denom)}" ) def _clear_status_block_locked() -> None: @@ -241,8 +260,37 @@ def _format_status_block() -> list[str]: fail60 = _push_fail_last_60s() ok_total, fail_total = _push_totals() ppm = _pushes_per_min_since_start() + with _progress_lock: + done = _progress_versions_done + queued = _progress_versions_queued + total = _progress_total + ok_n = _progress_ok + skip_n = _progress_skip + fail_n = _progress_fail + crates = _progress_index_crates + scan_done = _progress_scan_complete + jobs = _progress_jobs + # Growing denominator while index scan runs; locks to final total when scan completes. + denom = total if (total and total > 0) else (queued if queued > 0 else None) + qdepth = 0 + if _work_queue is not None: + try: + qdepth = _work_queue.qsize() + except NotImplementedError: + qdepth = max(0, queued - done) + if denom: + head = f"progress: {_format_progress_bar(done, denom)} {done}/{denom} {_format_eta(done, denom)}" + else: + head = f"progress: {_format_progress_bar(0, None)} scanning index..." + scan_line = ( + f"scan: crates={crates} versions_found={queued} " + f"status={'done' if scan_done else 'running (denom grows until full index walk)'}" + ) return [ - _format_progress_line(), + head, + scan_line, + f"config: jobs={jobs} queue_depth={qdepth}", + f"counts: ok={ok_n} skip={skip_n} fail={fail_n} done={done}", f"status: downloading={d} extracting={x} waiting_push={w} pushing={p}", ( f"push: ok_60s={ok60} fail_60s={fail60} " @@ -998,6 +1046,7 @@ def scan_and_process_crates( reimport_ok: bool, keep_crate_cache: bool = False, ) -> tuple[int, int, int]: + global _progress_jobs, _work_queue info("Scanning crates.io index...") _progress_reset() @@ -1024,7 +1073,10 @@ def scan_and_process_crates( use_streaming_full_index = not only_crates and not (limit_crates and limit_crates > 0) if use_streaming_full_index: - info("Streaming index: will download/commit/push while walking crate files.") + info( + "Streaming index in parallel: producer walks full crates.io-index " + "while workers import/push (denominator grows with scan ~2M versions)." + ) elif only_crates: crates = scan_selected_crates_index(index_path, only_crates) info(f"Found {len(crates)} crates.") @@ -1049,6 +1101,13 @@ def scan_and_process_crates( failed = 0 skipped = 0 lock = threading.Lock() + _progress_jobs = max(1, int(jobs)) + info( + f"Config: jobs={_progress_jobs} " + f"max_versions_per_crate={max_versions_per_crate} " + f"keep_crate_cache={keep_crate_cache} " + f"reimport_ok={reimport_ok} dry_run={dry_run}" + ) def process_one(crate_name: str, v: str) -> tuple[str, str, str]: key = (crate_name, v) @@ -1107,9 +1166,8 @@ def process_one(crate_name: str, v: str) -> tuple[str, str, str]: warn(f"{_fmt_repo(crate_name, v)} failed: {e}") return ("fail", crate_name, v) - def record_result(fut) -> None: + def record_result_status(status: str, c_name: str, v: str) -> None: nonlocal succeeded, failed, skipped - status, c_name, v = fut.result() _progress_note_result(status) # Print after progress counters update so sticky footer shows the new totals. if status == "ok": @@ -1133,38 +1191,87 @@ def record_result(fut) -> None: manifest[key] = rec append_manifest_record(manifest_path, rec) - # Concurrency: bounded pending futures in streaming mode to avoid RAM spikes. - with ThreadPoolExecutor(max_workers=max(1, int(jobs))) as ex: - if use_streaming_full_index: - max_pending = max(32, int(jobs) * 8) - pending: set = set() - for crate_name, versions in stream_index_crate_versions( - index_path, max_versions_per_crate - ): - for v in versions: - pending.add(ex.submit(process_one, crate_name, v)) - while len(pending) >= max_pending: - done, _ = wait(pending, return_when=FIRST_COMPLETED) - for df in done: - pending.discard(df) - record_result(df) - _progress_mark_scan_complete() - info( - f"Index scan complete: {_progress_index_crates} crates, " - f"{_progress_versions_queued} versions queued; draining workers..." + def record_result(fut) -> None: + status, c_name, v = fut.result() + record_result_status(status, c_name, v) + + n_jobs = max(1, int(jobs)) + if use_streaming_full_index: + # Producer (index scan) runs at full speed into a large work queue while + # workers download/push in parallel. Denominator (versions_found) grows + # with the scan toward the full crates.io size (~2M versions), not with + # the tiny in-flight window that previously stalled the walk. + # + # Queue holds (name, version) only — ~2M entries is acceptable within Job memory. + work_q: queue.Queue = queue.Queue(maxsize=0) + _work_queue = work_q + scan_error: list[BaseException] = [] + + def index_producer() -> None: + try: + for crate_name, versions in stream_index_crate_versions( + index_path, max_versions_per_crate + ): + for v in versions: + work_q.put((crate_name, v)) + _progress_mark_scan_complete() + info( + f"Index scan complete: {_progress_index_crates} crates, " + f"{_progress_versions_queued} versions found; workers draining queue..." + ) + except BaseException as e: + scan_error.append(e) + warn(f"Index scan failed: {e}") + finally: + for _ in range(n_jobs): + work_q.put(None) + + def worker_loop() -> None: + while True: + item = work_q.get() + try: + if item is None: + return + crate_name, v = item + status, c_name, ver = process_one(crate_name, v) + record_result_status(status, c_name, ver) + finally: + work_q.task_done() + + info( + "Parallel mode: index scan producer + " + f"{n_jobs} import workers (queue unbounded for version descriptors)." + ) + producer = threading.Thread( + target=index_producer, name="crates-index-scan", daemon=True + ) + workers = [ + threading.Thread( + target=worker_loop, name=f"crates-worker-{i}", daemon=True ) - for df in as_completed(pending): - record_result(df) - else: - tasks: list[tuple[str, str]] = [] - for crate_name, versions in crates_items: - vs = sorted(versions) - if max_versions_per_crate > 0: - vs = vs[-max_versions_per_crate:] - for v in vs: - tasks.append((crate_name, v)) - _progress_set_total(len(tasks)) - info(f"Starting to process {len(tasks)} crate versions...") + for i in range(n_jobs) + ] + producer.start() + for t in workers: + t.start() + producer.join() + for t in workers: + t.join() + _work_queue = None + if scan_error: + raise scan_error[0] + else: + _work_queue = None + tasks: list[tuple[str, str]] = [] + for crate_name, versions in crates_items: + vs = sorted(versions) + if max_versions_per_crate > 0: + vs = vs[-max_versions_per_crate:] + for v in vs: + tasks.append((crate_name, v)) + _progress_set_total(len(tasks)) + info(f"Starting to process {len(tasks)} crate versions...") + with ThreadPoolExecutor(max_workers=n_jobs) as ex: futures = [ex.submit(process_one, c, v) for c, v in tasks] for f in as_completed(futures): record_result(f) @@ -1175,6 +1282,7 @@ def record_result(fut) -> None: write_manifest(manifest_path, manifest) return succeeded, skipped, failed finally: + _work_queue = None stop_evt.set() if hb_thread is not None: hb_thread.join(timeout=max(1.0, float(STATUS_HEARTBEAT_INTERVAL_S) + 1.0)) diff --git "a/scripts/crates-sync/\344\270\212\346\211\213\346\214\207\345\215\227.md" "b/scripts/crates-sync/\344\270\212\346\211\213\346\214\207\345\215\227.md" new file mode 100644 index 000000000..f9d1ee74d --- /dev/null +++ "b/scripts/crates-sync/\344\270\212\346\211\213\346\214\207\345\215\227.md" @@ -0,0 +1,368 @@ +# Crates Sync 上手指南（架构 + 脚本 + Terraform） + +面向**不熟悉当前系统**的同学：搞清「要导入什么、数据在哪、怎么本地试、怎么用 Terraform 在 k3s-rust 上跑 Job、怎么看进度/排错」。集群部署测试需 **fork mega + mega-terraform**，在 mega 配 Harbor Secret，等 CI 出镜像后改 tf 的 image tag 并 `terraform apply`（详见 §3）。 + +--- + +## 1. 一句话目标 + +把 [crates.io](https://crates.io) 上的 crate 源码，按版本导入到 Mega 的 **monorepo 路径仓库**里： + +```text +third-party/rust/crates/// +``` + +例如 `tokio@1.37.0` → `third-party/rust/crates/to/ki/tokio/1.37.0`。 + +**不要**指望在 Web UI 里对整个 mega 仓库点 Sync 来完成这件事；大规模导入的正确入口是本目录脚本（本地或 K8s Job）。 + +--- + +## 2. 系统架构（先建立心智模型） + +### 2.1 相关仓库 + +| 仓库 | 作用 | +|------|------| +| **mega** | 应用与本脚本：`scripts/crates-sync/`；CI 构建 `crates-sync` 镜像 | +| **mega-terraform** | 把栈部署到 K8s；`envs/onprem/k3s-rust` 是 rust 环境 | + +### 2.2 rust 环境里有什么（`mega-rust` 命名空间） + +```text +用户 / CI + │ + ▼ +┌─────────────────────────────────────────────────────────┐ +│ k3s 集群 · namespace: mega-rust │ +│ │ +│ mono-engine (git HTTP) ←── git push 目标 │ +│ git.rust.xuanwu.openatom.cn │ +│ │ +│ mega-ui / campsite-api / … │ +│ │ +│ RustFS (S3) ←── mono 存 pack/对象（PVC，当前约 50Gi） │ +│ │ +│ Job: crates-sync ──hostPath──► storage-server-01 │ +│ │ /opt/data/freighter/ │ +│ │ ├── crates.io-index │ +│ │ ├── crates/ (.crate) │ +│ │ └── mega-crates-work/ │ +│ └── run_job.py → crates-sync.py │ +└─────────────────────────────────────────────────────────┘ +``` + +要点： + +1. **导入进程**跑在节点 `storage-server-01` 上（有 freighter 磁盘），不是随便一个 worker。 +2. **index / .crate 缓存**在宿主机 `/opt/data/freighter`，Job 里挂载为 `/freighter`；**不要**在集群里重新 clone 整份 index。 +3. **git push** 打到集群内的 **mono-engine**；对象最终进 **RustFS**。磁盘不够会导致 push/保存失败。 +4. 该节点有污点 `observe-only=true:NoSchedule`；Terraform 已给 Job 配了 toleration，否则 Pod 会一直 Pending。 + +### 2.3 脚本在干什么（流水线） + +```text +读 crates.io-index（流式扫文件） + │ 并行 + ▼ +下载 .crate（若缓存没有）→ 解压 → git init/commit → git push → Mega + │ + ▼ +写 manifest（JSONL）：crate@version → ok | skip | fail +``` + +全量模式下： + +- **Producer**：全速扫 index，分母（`versions_found`）涨到真实规模（约 **200 万** version）。 +- **Workers（`--jobs`）**：同时从队列取任务做下载/推送。 +- 已成功写入 manifest 的 `status=ok` **默认跳过**（可断点续跑）。 + +--- + +## 3. 改代码到集群部署测试（完整流程） + +在集群上验证 / 跑 Job，需要同时使用 **两个仓库**，并走完「镜像构建 → 改 tag → terraform apply」。 + +### 3.0 线上测试流程图 + +```mermaid +flowchart LR + A[Fork 两仓 + Harbor Secret] --> B[改脚本 / 推送 mega] + B --> C[等 CI 出镜像 tag] + C --> D[tfvars 换 crates_sync_image] + D --> E[terraform apply] + E --> F[kubectl logs -f job/crates-sync] +``` + +### 3.1 Fork 两个仓库 + +1. Fork **mega**（脚本、Dockerfile、CI） +2. Fork **mega-terraform**（k3s-rust 环境、`crates_sync` Job 配置） + +本地分别 clone 自己的 fork，按团队习惯从 upstream 同步 `main` 后再开分支改动。 + +### 3.2 在 mega 仓库配置 Harbor Secret + +CI workflow：`.github/workflows/crates-sync-deploy.yml`。 +推送镜像到 `registry.xuanwu.openatom.cn/mega/crates-sync`，登录依赖 GitHub Actions Secrets： + +| Secret 名称 | 用途 | +|-------------|------| +| `HARBOR_USERNAME` | Harbor 用户名 | +| `HARBOR_PASSWORD` | Harbor 密码 / Robot token | + +在 **你的 mega fork** 上配置： + +1. GitHub → 仓库 → **Settings** → **Secrets and variables** → **Actions** +2. 新增上述两个 Secret（向有权限的同学索取 Harbor 账号；勿把密码写进代码或 tfvars） +3. 确认 Actions 已启用（fork 上可能要手动允许 workflow） + +未配置 Secret 时，CI 会在 Login to Harbor 步骤失败，镜像不会进仓库。 + +### 3.3 改脚本并等 CI 出镜像 + +1. 在 mega fork 修改 `scripts/crates-sync/**`（或 workflow 本身） +2. 合并 / 推送到会触发 workflow 的分支（当前 workflow 监听 **`main`**，且仅当上述路径变更时触发） +3. 打开 Actions → **Crates Sync deploy**，等到 **build-and-push 成功** +4. 成功后 Harbor 中会有类似 tag： + - `registry.xuanwu.openatom.cn/mega/crates-sync:<短 sha>`（commit 前 7 位，**部署请用这个**） + - `registry.xuanwu.openatom.cn/mega/crates-sync:latest` + +从 Actions 日志或 `GITHUB_SHA` 前 7 位确认实际 tag。 + +### 3.4 在 mega-terraform 中替换 image tag + +在 **mega-terraform** fork 的环境文件中更新镜像，例如 `envs/onprem/k3s-rust/terraform.tfvars`： + +```hcl +enable_crates_sync = true +crates_sync_freighter_host_path = "/opt/data/freighter" +crates_sync_node_hostname = "storage-server-01" +crates_sync_image = "registry.xuanwu.openatom.cn/mega/crates-sync:<短sha>" + +# 部署测试可先限流，确认链路后再去掉： +# crates_sync_args = ["--jobs", "2", "--limit-crates", "20", "--max-versions-per-crate", "1"] +``` + +`crates_sync_image` 每换一次 tag，Terraform 会 **替换重建** `crates-sync` Job（异步，apply 不等待导入跑完）。 + +### 3.5 执行 `terraform apply` 部署测试 + +```bash +cd mega-terraform/envs/onprem/k3s-rust +terraform init # 首次或 backend/provider 变更后 +terraform plan # 确认会替换 Job / 更新 image-trigger +terraform apply +``` + +然后跟日志验证： + +```bash +kubectl -n mega-rust get pods -l job-name=crates-sync -o wide +kubectl -n mega-rust logs -f job/crates-sync +``` + +仅本地跑脚本、不发 Job 时，可跳过 Harbor / CI / apply，见 §5。 + +--- + +## 4. 代码与关键文件位置 + +| 路径 | 说明 | +|------|------| +| `mega/scripts/crates-sync/crates-sync.py` | 核心导入逻辑 | +| `mega/scripts/crates-sync/run_job.py` | K8s Job 入口：等 mono、bootstrap token、调 sync | +| `mega/scripts/crates-sync/Dockerfile` | 镜像构建上下文（目录即本目录） | +| `mega/.github/workflows/crates-sync-deploy.yml` | 构建并推送镜像到 Harbor | +| `mega-terraform/modules/.../gitmono_stack/crates_sync.tf` | Job / hostPath / 亲和 / toleration | +| `mega-terraform/envs/onprem/k3s-rust/terraform.tfvars` | 该环境开关与镜像 tag | + +英文细节与参数列表见同目录 [README.md](./README.md)。 + +--- + +## 5. 本地怎么用脚本（调试推荐） + +### 5.1 依赖 + +- Python 3 +- Git +- 一份本地 [crates.io-index](https://github.com/rust-lang/crates.io-index) checkout +- 能访问目标 Mega 的 **Bearer token**（`MEGA_TOKEN` 或 `--token`） + +### 5.2 小规模试跑（强烈推荐先做） + +只导几个 crate，**不扫全量 index**： + +```bash +export MEGA_TOKEN="你的token" + +python3 scripts/crates-sync/crates-sync.py \ + --index ~/crates.io-index \ + --crates-dir /tmp/crates-cache \ + --workdir /tmp/mega-crates-work \ + --git-base-url https://git.rust.xuanwu.openatom.cn \ + --token "$MEGA_TOKEN" \ + --crate tokio --crate serde \ + --max-versions-per-crate 1 \ + --jobs 2 +``` + +干跑（只下载解压，不 commit/push）： + +```bash +python3 scripts/crates-sync/crates-sync.py \ + --index ~/crates.io-index \ + --crates-dir /tmp/crates-cache \ + --workdir /tmp/mega-crates-work \ + --git-base-url https://git.rust.xuanwu.openatom.cn \ + --crate tokio \ + --max-versions-per-crate 1 \ + --dry-run +``` + +采样 20 个 crate： + +```bash +python3 scripts/crates-sync/crates-sync.py \ + --index ~/crates.io-index \ + --crates-dir /tmp/crates-cache \ + --workdir /tmp/mega-crates-work \ + --git-base-url https://git.rust.xuanwu.openatom.cn \ + --token "$MEGA_TOKEN" \ + --limit-crates 20 \ + --max-versions-per-crate 1 \ + --jobs 4 +``` + +### 5.3 常用参数 + +| 参数 | 含义 | +|------|------| +| `--jobs N` | 并发 worker 数；同时限制并发 `git push` | +| `--max-versions-per-crate N` | 每 crate 只保留最近 N 个版本；`0` = 全部 | +| `--keep-crate-cache` | 成功后不删 `.crate`（共享 freighter 缓存时必须开） | +| `--manifest PATH` | 断点清单；默认 `/crates-import-manifest.jsonl` | +| `--reimport-ok` | 强制重导 manifest 里已是 `ok` 的版本 | +| `--force` / `--force-with-lease` | 只影响 git push，**不会**绕过 manifest 的 ok 跳过 | + +### 5.4 进度条（本地 / Job 日志底部） + +默认开启 sticky heartbeat，大致形如： + +```text +progress: [##----------------------------] 0.8% 16000/1850000 eta=48.2h +scan: crates=... versions_found=... status=running ... +config: jobs=2 queue_depth=... +counts: ok=... skip=... fail=... done=... +status: downloading=... pushing=... +push: ok_60s=... per_min=... +``` + +说明： + +- 扫 index 未完成时，分母会随 `versions_found` **一直涨**（目标约 200 万），不要用早期百分比当「快做完了」。 +- `[OK] xxx pushed` 会刷日志，进度条钉在底部（sticky）。 + +--- + +## 6. 用 Terraform 在 k3s-rust 上跑 Job（细节） + +> 端到端步骤见 **§3**。本节补充前置条件与运维细节。 + +环境目录：`mega-terraform/envs/onprem/k3s-rust` +命名空间：`mega-rust` +域名示例：`https://git.rust.xuanwu.openatom.cn`（mono）、`https://app.rust.xuanwu.openatom.cn`（UI）。 + +### 6.1 前置条件 Checklist + +1. 已按 §3 fork 两仓库，mega 已配 Harbor Secret，CI 已产出目标镜像 tag。 +2. 能访问集群：`kubeconfig`（如 `~/.kube/k3s.yaml`），`kubectl -n mega-rust get pods` 正常。 +3. 节点 `storage-server-01` 存在，且已有目录： + - `/opt/data/freighter/crates.io-index`（完整 index checkout） + - `/opt/data/freighter/crates`（.crate 缓存，可空） +4. mono / RustFS 已在该命名空间跑着（Job `depends_on` apps）。 + +### 6.2 配置要点 + +```hcl +enable_crates_sync = true +crates_sync_freighter_host_path = "/opt/data/freighter" +crates_sync_node_hostname = "storage-server-01" # 与 kubectl get nodes 主机名一致 +crates_sync_image = "registry.xuanwu.openatom.cn/mega/crates-sync:<短sha>" + +# 小流量试跑可加（全量导入时不要限制）: +# crates_sync_args = ["--jobs", "4", "--limit-crates", "20", "--max-versions-per-crate", "1"] + +# 全量常见写法（run_job 默认已是 max-versions=0、keep-cache）: +# crates_sync_args = ["--jobs", "4"] +``` + +说明： + +- `run_job.py` 默认：`--jobs 2`、`--max-versions-per-crate 0`、`--keep-crate-cache`、等 mono 就绪后用 `MEGA_INIT_BOOTSTRAP_SECRET` 换 bot token。 +- 额外参数通过 `crates_sync_args` 拼到 `run_job.py` 后面。 + +### 6.3 Apply（异步 Job） + +```bash +cd mega-terraform/envs/onprem/k3s-rust +terraform init # 首次 +terraform plan +terraform apply +``` + +- Job **`wait_for_completion = false`**：apply 成功 ≠ 导入完成，只表示 Job 对象已创建/替换。 +- **换镜像 tag** 或改触发 `terraform_data.crates_sync_image` 的内容 → Job **替换重建** → 新一轮导入（manifest 仍在 freighter 盘上，ok 会继续 skip）。 + +### 6.4 看日志与状态 + +```bash +kubectl -n mega-rust get job crates-sync +kubectl -n mega-rust get pods -l job-name=crates-sync -o wide +kubectl -n mega-rust logs -f job/crates-sync +``` + +期望：Pod 在 **`storage-server-01`**，状态 Running。 + +### 6.5 重新跑一轮 + +1. mega：等 **Crates Sync deploy** CI 成功，记下新 `<短sha>`。 +2. mega-terraform：改 `crates_sync_image`。 +3. `terraform apply`。 + +### 6.6 小流量建议 + +先用 `crates_sync_args` 加 `--limit-crates` / `--max-versions-per-crate` 通链路，再去掉限流做全量；盯 RustFS PVC 与 freighter 磁盘。 + +--- + +## 7. Manifest 与「杀 Job 再跑」行为 + +- 路径（Job）：`/freighter/mega-crates-work/crates-import-manifest.jsonl` +- `status=ok` → 下次默认 **skip**（省时间、可续跑）。 +- `--force` **不会**重导 ok；要重导用 `--reimport-ok`（经 `crates_sync_args` 传入）。 +- 杀 Pod/Job **不会**清 freighter 上的 index、crates 缓存、manifest。 + +--- + +## 8. 常见问题 + +全量导入体量大，优先关注资源是否够用： + +| 现象 | 可能原因 | 处理 | +|------|----------|------| +| `git push` / S3 报错、对象写失败 | RustFS **磁盘**不足（PVC 写满） | 扩容 PVC（如 `data-rustfs-0`），确认 Longhorn 已 `allowVolumeExpansion`；盯 `kubectl -n mega-rust get pvc` | +| RustFS / mono OOM、请求超时 | RustFS 或 mono **内存**不足 | 调高 `rustfs_resources` / mono 资源后 apply；查对应 Pod 是否被 OOMKilled | +| crates-sync Pod OOMKilled / Evicted | Job **内存/CPU** limit 过紧（全量扫队列也会占内存） | 调高 `crates_sync.tf` 里 container resources，或减小 `--jobs` | +| 节点磁盘打满、下载/解压失败 | freighter 宿主机盘（`/opt/data/freighter`）空间不足 | 清理或扩容 `storage-server-01` 上 freighter 目录所在磁盘 | + +--- + +## 9. 相关链接（按环境替换） + +- Git：`https://git.rust.xuanwu.openatom.cn` +- App：`https://app.rust.xuanwu.openatom.cn` +- Harbor：`registry.xuanwu.openatom.cn`（镜像 `mega/crates-sync`） +- Terraform 环境说明：`mega-terraform/envs/onprem/k3s-rust/README.md` +- 脚本英文 README：`scripts/crates-sync/README.md`