mirror of
https://github.com/Portabase/agent.git
synced 2026-09-11 02:27:10 +00:00
Compare commits
49 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0b7b7ef59b | |||
| 038c5523b7 | |||
| ea010ce713 | |||
| 8897568281 | |||
| 66cad4e12e | |||
| 221ed4e7e1 | |||
| 84735ac399 | |||
| f7639de096 | |||
| e504d09cb3 | |||
| 0363b300a4 | |||
| bf6e7d41ab | |||
| 2c7065ae95 | |||
| 1fece577cc | |||
| 273475fa3b | |||
| 628c017584 | |||
| f02218708a | |||
| 16152328b0 | |||
| 7328827435 | |||
| 348eaac81b | |||
| 0f6c93ecd0 | |||
| befec0deae | |||
| 2052ab0ff5 | |||
| 461e92d67e | |||
| 23678bf2d6 | |||
| 02105a8171 | |||
| f94656a39b | |||
| b94ff4e987 | |||
| dc32c442f3 | |||
| 1453851555 | |||
| 185a7ee612 | |||
| d8328f60ca | |||
| 531a25f292 | |||
| 740ee43038 | |||
| 1fccc0bc19 | |||
| 2c15ea64ae | |||
| ac8f7fd8d8 | |||
| 6edf2890f1 | |||
| 5f579690ff | |||
| 5c35375df7 | |||
| 61c7224104 | |||
| d7958e03b9 | |||
| b269d98d3c | |||
| 2d44b62844 | |||
| 092f760431 | |||
| 2c5c805308 | |||
| 38a3274c43 | |||
| bb45c1961f | |||
| ee7016d1fa | |||
| c46c301371 |
@@ -0,0 +1,96 @@
|
||||
name: Build Windows release
|
||||
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
push:
|
||||
tags:
|
||||
- '[0-9]+.[0-9]+.[0-9]+'
|
||||
branches:
|
||||
- main
|
||||
- master
|
||||
|
||||
jobs:
|
||||
build-windows:
|
||||
runs-on: windows-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Set up Rust toolchain (MSVC)
|
||||
uses: actions-rs/toolchain@v1
|
||||
with:
|
||||
toolchain: stable-x86_64-pc-windows-msvc
|
||||
profile: minimal
|
||||
override: true
|
||||
|
||||
- name: Install vcpkg and OpenSSL (x64)
|
||||
shell: pwsh
|
||||
run: |
|
||||
# Install vcpkg and the prebuilt OpenSSL package
|
||||
git clone https://github.com/microsoft/vcpkg C:\vcpkg
|
||||
C:\vcpkg\bootstrap-vcpkg.bat
|
||||
C:\vcpkg\vcpkg install openssl:x64-windows
|
||||
# Export variables for subsequent steps
|
||||
'VCPKG_ROOT=C:\vcpkg' | Out-File -FilePath $env:GITHUB_ENV -Encoding utf8 -Append
|
||||
'OPENSSL_DIR=C:\vcpkg\installed\x64-windows' | Out-File -FilePath $env:GITHUB_ENV -Encoding utf8 -Append
|
||||
|
||||
- name: Build (cargo release)
|
||||
shell: pwsh
|
||||
env:
|
||||
# Cargo / openssl-sys will pick up OPENSSL_DIR from the environment
|
||||
OPENSSL_DIR: ${{ env.OPENSSL_DIR }}
|
||||
run: |
|
||||
# Ensure the environment variable is present for this step
|
||||
if (-Not $env:OPENSSL_DIR) { Write-Host "OPENSSL_DIR not set, printing env for debugging"; Get-ChildItem Env: | ForEach-Object { Write-Host $_ } }
|
||||
# Build the declared bin target explicitly (Cargo.toml [[bin]] name = "app")
|
||||
cargo build --release --bin app
|
||||
|
||||
- name: Prepare artifact zip
|
||||
id: prepare_artifact
|
||||
shell: pwsh
|
||||
env:
|
||||
RELEASE_TAG: ${{ github.ref_name }}
|
||||
run: |
|
||||
$tag = $env:RELEASE_TAG
|
||||
if (-not $tag) { $tag = $env:GITHUB_SHA }
|
||||
|
||||
# Package the declared bin target deterministically (Cargo.toml [[bin]] name = "app")
|
||||
$exe = "target\release\app.exe"
|
||||
if (-not (Test-Path $exe)) { Write-Error "Built binary $exe not found in target/release"; exit 1 }
|
||||
|
||||
$outDir = "artifact"
|
||||
New-Item -ItemType Directory -Path $outDir -Force | Out-Null
|
||||
# Ship under the package name, not the internal bin name "app"
|
||||
Copy-Item -Path $exe -Destination "$outDir\portabase-agent.exe"
|
||||
|
||||
$zipName = "windows-release-$tag.zip"
|
||||
if (Test-Path $zipName) { Remove-Item $zipName }
|
||||
Compress-Archive -Path "$outDir\*" -DestinationPath $zipName -Force
|
||||
Write-Host "ZIP=$zipName"
|
||||
Write-Output "zip=$zipName" | Out-File -FilePath $env:GITHUB_OUTPUT -Encoding utf8 -Append
|
||||
|
||||
- name: Upload build artifact
|
||||
uses: actions/upload-artifact@v4
|
||||
with:
|
||||
name: windows-release
|
||||
path: windows-release-*.zip
|
||||
|
||||
- name: Create GitHub Release
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
id: create_release
|
||||
uses: softprops/action-gh-release@v1
|
||||
with:
|
||||
tag_name: ${{ github.ref_name }}
|
||||
env:
|
||||
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
- name: Upload release asset
|
||||
if: startsWith(github.ref, 'refs/tags/')
|
||||
uses: actions/upload-release-asset@v1
|
||||
with:
|
||||
upload_url: ${{ steps.create_release.outputs.upload_url }}
|
||||
asset_path: windows-release-${{ github.ref_name }}.zip
|
||||
asset_name: windows-release-${{ github.ref_name }}.zip
|
||||
asset_content_type: application/zip
|
||||
+2
-1
@@ -6,4 +6,5 @@
|
||||
.env
|
||||
|
||||
.claude
|
||||
/docs
|
||||
|
||||
/docs
|
||||
|
||||
+1
-1
@@ -27,5 +27,5 @@ keywords:
|
||||
- self-hosted
|
||||
- portabase
|
||||
license: Apache-2.0
|
||||
version: 1.11.1
|
||||
version: 1.13.1
|
||||
date-released: '2026-02-24'
|
||||
|
||||
Generated
+191
-1
@@ -136,6 +136,17 @@ dependencies = [
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-lock"
|
||||
version = "3.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311"
|
||||
dependencies = [
|
||||
"event-listener",
|
||||
"event-listener-strategy",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-stream"
|
||||
version = "0.3.6"
|
||||
@@ -679,6 +690,58 @@ dependencies = [
|
||||
"tower-service",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "azure_core"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "fe6a26a7d374b440015cbbcbf2d9d8be5a133aa940599f5e5dc569504baa262e"
|
||||
dependencies = [
|
||||
"async-lock",
|
||||
"async-trait",
|
||||
"azure_core_macros",
|
||||
"bytes",
|
||||
"futures",
|
||||
"pin-project",
|
||||
"rustc_version",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"typespec",
|
||||
"typespec_client_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "azure_core_macros"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b9b52dba6a345f3ad2d42ff8d0d63df9d0994cfa29657bf18ffdbf149f78a4f5"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "azure_storage_blob"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1756febbcca86c862ef718b983b505d08bd65a9bc984a915b0a16af4a4c3fe5b"
|
||||
dependencies = [
|
||||
"async-stream",
|
||||
"async-trait",
|
||||
"azure_core",
|
||||
"bytes",
|
||||
"futures",
|
||||
"percent-encoding",
|
||||
"pin-project",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"time",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "base16ct"
|
||||
version = "0.1.1"
|
||||
@@ -910,6 +973,17 @@ version = "0.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724"
|
||||
|
||||
[[package]]
|
||||
name = "chacha20"
|
||||
version = "0.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6f8d983286843e49675a4b7a2d174efe136dc93a18d69130dd18198a6c167601"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
"rand_core 0.10.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "chrono"
|
||||
version = "0.4.44"
|
||||
@@ -981,6 +1055,15 @@ version = "0.4.31"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "75984efb6ed102a0d42db99afb6c1948f0380d1d91808d5529916e6c08b49d8d"
|
||||
|
||||
[[package]]
|
||||
name = "concurrent-queue"
|
||||
version = "2.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "connection-string"
|
||||
version = "0.2.0"
|
||||
@@ -1501,6 +1584,27 @@ dependencies = [
|
||||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "event-listener"
|
||||
version = "5.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e13b66accf52311f30a0db42147dadea9850cb48cd070028831ae5f5d4b856ab"
|
||||
dependencies = [
|
||||
"concurrent-queue",
|
||||
"parking",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "event-listener-strategy"
|
||||
version = "0.5.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93"
|
||||
dependencies = [
|
||||
"event-listener",
|
||||
"pin-project-lite",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fallible-iterator"
|
||||
version = "0.2.0"
|
||||
@@ -2934,6 +3038,12 @@ dependencies = [
|
||||
"sha2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parking"
|
||||
version = "2.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
|
||||
|
||||
[[package]]
|
||||
name = "parking_lot"
|
||||
version = "0.12.5"
|
||||
@@ -3083,7 +3193,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "portabase-agent"
|
||||
version = "1.11.1"
|
||||
version = "1.13.1"
|
||||
dependencies = [
|
||||
"aes",
|
||||
"aes-gcm",
|
||||
@@ -3093,6 +3203,8 @@ dependencies = [
|
||||
"async-trait",
|
||||
"aws-config",
|
||||
"aws-sdk-s3",
|
||||
"azure_core",
|
||||
"azure_storage_blob",
|
||||
"base64 0.22.1",
|
||||
"bytes",
|
||||
"chrono",
|
||||
@@ -3294,6 +3406,16 @@ dependencies = [
|
||||
"prost",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quick-xml"
|
||||
version = "0.39.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "cdcc8dd4e2f670d309a5f0e83fe36dfdc05af317008fea29144da1a2ac858e5e"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
"serde",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "quinn"
|
||||
version = "0.11.9"
|
||||
@@ -3398,6 +3520,17 @@ dependencies = [
|
||||
"rand_core 0.9.5",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand"
|
||||
version = "0.10.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d2e8e8bcc7961af1fdac401278c6a831614941f6164ee3bf4ce61b7edb162207"
|
||||
dependencies = [
|
||||
"chacha20",
|
||||
"getrandom 0.4.2",
|
||||
"rand_core 0.10.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rand_chacha"
|
||||
version = "0.3.1"
|
||||
@@ -4752,13 +4885,18 @@ version = "0.6.8"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d4e6559d53cc268e5031cd8429d05415bc4cb4aefc4aa5d6cc35fbf5b924a1f8"
|
||||
dependencies = [
|
||||
"async-compression",
|
||||
"bitflags 2.11.0",
|
||||
"bytes",
|
||||
"futures-core",
|
||||
"futures-util",
|
||||
"http 1.4.0",
|
||||
"http-body 1.0.1",
|
||||
"http-body-util",
|
||||
"iri-string",
|
||||
"pin-project-lite",
|
||||
"tokio",
|
||||
"tokio-util",
|
||||
"tower",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
@@ -4884,6 +5022,58 @@ version = "1.19.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb"
|
||||
|
||||
[[package]]
|
||||
name = "typespec"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "21666a31293beab8f41d38c2849ddbc342cd9c7cb4d71a9818868287a8934e53"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"bytes",
|
||||
"futures",
|
||||
"quick-xml",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "typespec_client_core"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "924f0c734e0ac3b881ab99d032bd28fcc969d2bb73ef1b8dd4772fd8e518a382"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"base64 0.22.1",
|
||||
"bytes",
|
||||
"dyn-clone",
|
||||
"futures",
|
||||
"pin-project",
|
||||
"rand 0.10.1",
|
||||
"reqwest 0.13.2",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"time",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"typespec",
|
||||
"typespec_macros",
|
||||
"url",
|
||||
"uuid",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "typespec_macros"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "2c608f4427943f8adb211abc95c87672b1b98847152783507d54e3246e502f60"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"rustc_version",
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "unicase"
|
||||
version = "2.9.0"
|
||||
|
||||
+3
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "portabase-agent"
|
||||
version = "1.11.1"
|
||||
version = "1.13.1"
|
||||
edition = "2024"
|
||||
|
||||
[dependencies]
|
||||
@@ -39,6 +39,8 @@ tokio-util = { version = "0.7.18", features = ["compat"] }
|
||||
tiberius = { version = "0.12", default-features = false, features = ["rustls", "chrono"] }
|
||||
aws-config = "1.8.13"
|
||||
aws-sdk-s3 = { version = "1.122.0", features = ["behavior-version-latest"] }
|
||||
azure_core = "1.0.0"
|
||||
azure_storage_blob = "1.0.0"
|
||||
async-compression = { version = "0.4.37", features = ["tokio", "gzip"] }
|
||||
tokio-tar = "0.3.1"
|
||||
oauth2 = "5.0.0"
|
||||
|
||||
+9
-1
@@ -19,7 +19,7 @@ services:
|
||||
APP_ENV: development
|
||||
LOG: debug
|
||||
TZ: "Europe/Paris"
|
||||
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiMGZiNDYyMmUtMTMxNS00MzMxLTlkMTMtZWMzMjAyZjZiNTIwIiwibWFzdGVyS2V5QjY0IjoiMUh0djdtWCtYVkJxL0IzUEV2WDlZZjlQeUdVZW5oRHlXemo5THRqNW90WT0ifQ=="
|
||||
EDGE_KEY: "eyJzZXJ2ZXJVcmwiOiJodHRwOi8vbG9jYWxob3N0Ojg4ODciLCJhZ2VudElkIjoiNWRkZTE1NTctZWQ1ZC00MjUxLThiZDMtMDE0MjkxOTg2OGZjIiwibWFzdGVyS2V5QjY0IjoiQlhWM1hvbEM2NTZTVjdkTmdjV1BHUWxrKytycExJNmxHRGk3Q1BCNWllbz0ifQ=="
|
||||
#CHUNK_SIZE_MB: "1"
|
||||
#POOLING: 1
|
||||
#DATABASES_CONFIG_FILE: "config.toml"
|
||||
@@ -28,6 +28,14 @@ services:
|
||||
networks:
|
||||
- portabase
|
||||
|
||||
cpus: "1.50"
|
||||
|
||||
mem_limit: 4g
|
||||
memswap_limit: 4g
|
||||
|
||||
pids_limit: 512
|
||||
|
||||
|
||||
volumes:
|
||||
cargo-registry:
|
||||
cargo-git:
|
||||
|
||||
@@ -1,3 +1,11 @@
|
||||
# =========================
|
||||
# MySQL client tools
|
||||
# =========================
|
||||
FROM mysql:8.4 AS mysql-client-tools
|
||||
RUN mkdir -p /mysql-exports/bin /mysql-exports/lib \
|
||||
&& cp /usr/bin/mysqldump /mysql-exports/bin/ \
|
||||
&& find /usr/lib -name "libmysqlclient.so.21*" -exec cp {} /mysql-exports/lib/ \;
|
||||
|
||||
# =========================
|
||||
# Base image (shared)
|
||||
# =========================
|
||||
@@ -68,6 +76,13 @@ RUN if [ "$TARGETARCH" = "amd64" ]; then \
|
||||
COPY assets/tools/${TARGETARCH}/mongodb/ /usr/local/mongodb/
|
||||
RUN chmod +x /usr/local/mongodb/bin/*
|
||||
|
||||
# =========================
|
||||
# MySQL real mysqldump binary
|
||||
# =========================
|
||||
COPY --from=mysql-client-tools /mysql-exports/bin/mysqldump /usr/local/bin/mysqldump
|
||||
COPY --from=mysql-client-tools /mysql-exports/lib/ /usr/local/lib/
|
||||
RUN chmod +x /usr/local/bin/mysqldump && ldconfig
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
# =========================
|
||||
@@ -135,6 +150,9 @@ RUN chmod +x /entrypoint.sh
|
||||
COPY --from=base /usr/lib/postgresql/ /usr/lib/postgresql/
|
||||
COPY --from=base /usr/local/mongodb/bin/ /usr/local/mongodb/bin/
|
||||
COPY --from=base /root/.dotnet/tools/ /root/.dotnet/tools/
|
||||
COPY --from=mysql-client-tools /mysql-exports/bin/mysqldump /usr/local/bin/mysqldump
|
||||
COPY --from=mysql-client-tools /mysql-exports/lib/ /usr/local/lib/
|
||||
RUN chmod +x /usr/local/bin/mysqldump && ldconfig
|
||||
|
||||
ENV PATH="$PATH:/usr/local/dotnet:/root/.dotnet/tools"
|
||||
ENV APP_ENV=production
|
||||
|
||||
@@ -33,6 +33,11 @@ pub async fn run(
|
||||
let _mariadb_dump = select_mariadb_path(&version).join("mariadb-dump");
|
||||
|
||||
logger.log("debug", format!("Using mariadb-dump at {}", _mariadb_dump.display()));
|
||||
|
||||
if let Ok(out) = Command::new("mariadb-dump").arg("--version").output() {
|
||||
logger.log("debug", format!("mariadb-dump client: {}", String::from_utf8_lossy(&out.stdout).trim()));
|
||||
}
|
||||
|
||||
logger.log("info", format!("Running mariadb-dump for {}", cfg.name));
|
||||
|
||||
let start = Instant::now();
|
||||
@@ -50,7 +55,7 @@ pub async fn run(
|
||||
.arg("--skip-add-drop-table")
|
||||
.arg("--compress")
|
||||
.arg("--verbose")
|
||||
.arg("--max-allowed-packet=512M")
|
||||
.arg(format!("--max-allowed-packet={}", cfg.max_packet_size))
|
||||
.arg("--net-buffer-length=16K")
|
||||
.arg("--default-character-set=utf8mb4")
|
||||
.arg(&cfg.database)
|
||||
|
||||
@@ -2,7 +2,6 @@ use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::config::DatabaseConfig;
|
||||
use anyhow::{Context, Result};
|
||||
use std::fs::File;
|
||||
use std::io::{Read, Write};
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::sync::Arc;
|
||||
@@ -12,11 +11,8 @@ pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogg
|
||||
let handle = tokio::task::spawn_blocking(move || -> Result<()> {
|
||||
logger.log("info", format!("Starting restore for database {}", cfg.name));
|
||||
|
||||
let mut sql_content = String::new();
|
||||
let mut file = File::open(&restore_file)
|
||||
.with_context(|| format!("Failed to open restore file {}", restore_file.display()))?;
|
||||
file.read_to_string(&mut sql_content)
|
||||
.with_context(|| format!("Failed to read restore file {}", restore_file.display()))?;
|
||||
|
||||
let drop_create_cmd = format!(
|
||||
"DROP DATABASE IF EXISTS `{0}`; CREATE DATABASE `{0}`;",
|
||||
@@ -66,10 +62,8 @@ pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogg
|
||||
.with_context(|| format!("Failed to start MariaDB restore for {}", cfg.name))?;
|
||||
|
||||
let mut stdin = child.stdin.take().context("Failed to open child stdin")?;
|
||||
stdin
|
||||
.write_all(sql_content.as_bytes())
|
||||
.context("Failed to write SQL content to MariaDB stdin")?;
|
||||
stdin.flush()?;
|
||||
std::io::copy(&mut file, &mut stdin)
|
||||
.context("Failed to stream SQL content to MariaDB stdin")?;
|
||||
drop(stdin);
|
||||
|
||||
let output = child
|
||||
|
||||
@@ -31,6 +31,10 @@ pub async fn run(
|
||||
|
||||
let file_path = backup_dir.join(format!("{}{}", cfg.generated_id, file_extension));
|
||||
|
||||
if let Ok(out) = Command::new("mysqldump").arg("--version").output() {
|
||||
logger.log("debug", format!("mysqldump client: {}", String::from_utf8_lossy(&out.stdout).trim()));
|
||||
}
|
||||
|
||||
logger.log("info", format!("Running mysqldump for {}", cfg.name));
|
||||
|
||||
let start = Instant::now();
|
||||
@@ -43,11 +47,15 @@ pub async fn run(
|
||||
.arg("--triggers")
|
||||
.arg("--verbose")
|
||||
.arg("--single-transaction")
|
||||
.arg("--set-gtid-purged=OFF")
|
||||
.arg("--no-tablespaces")
|
||||
.arg("--quick")
|
||||
.arg("--skip-lock-tables")
|
||||
.arg("--skip-add-drop-table")
|
||||
.arg("--no-create-db")
|
||||
.arg("--default-character-set=utf8mb4")
|
||||
.arg("--network-timeout")
|
||||
.arg(format!("--max-allowed-packet={}", cfg.max_packet_size))
|
||||
.arg(&cfg.database)
|
||||
.arg("-r").arg(&file_path)
|
||||
.envs(env)
|
||||
|
||||
@@ -2,7 +2,6 @@ use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::config::DatabaseConfig;
|
||||
use anyhow::{Context, Result};
|
||||
use std::fs::File;
|
||||
use std::io::{Read, Write};
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::sync::Arc;
|
||||
@@ -12,11 +11,8 @@ pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogg
|
||||
let handle = tokio::task::spawn_blocking(move || -> Result<()> {
|
||||
logger.log("info", format!("Starting restore for database {}", cfg.name));
|
||||
|
||||
let mut sql_content = String::new();
|
||||
let mut file = File::open(&restore_file)
|
||||
.with_context(|| format!("Failed to open restore file {}", restore_file.display()))?;
|
||||
file.read_to_string(&mut sql_content)
|
||||
.with_context(|| format!("Failed to read restore file {}", restore_file.display()))?;
|
||||
|
||||
let drop_create_cmd = format!(
|
||||
"DROP DATABASE IF EXISTS `{0}`; CREATE DATABASE `{0}`;",
|
||||
@@ -66,10 +62,8 @@ pub async fn run(cfg: DatabaseConfig, restore_file: PathBuf, logger: Arc<JobLogg
|
||||
.with_context(|| format!("Failed to start mysql restore for {}", cfg.name))?;
|
||||
|
||||
let mut stdin = child.stdin.take().context("Failed to open child stdin")?;
|
||||
stdin
|
||||
.write_all(sql_content.as_bytes())
|
||||
.context("Failed to write SQL content to mysql stdin")?;
|
||||
stdin.flush()?;
|
||||
std::io::copy(&mut file, &mut stdin)
|
||||
.context("Failed to stream SQL content to mysql stdin")?;
|
||||
drop(stdin);
|
||||
|
||||
let output = child
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use anyhow::Result;
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::sync::Arc;
|
||||
@@ -13,6 +14,7 @@ pub async fn run(
|
||||
cfg: DatabaseConfig,
|
||||
format: PostgresDumpFormat,
|
||||
backup_dir: PathBuf,
|
||||
env: HashMap<String, String>,
|
||||
logger: Arc<JobLogger>,
|
||||
) -> Result<PathBuf> {
|
||||
tokio::task::spawn_blocking(move || -> Result<PathBuf> {
|
||||
@@ -37,18 +39,18 @@ pub async fn run(
|
||||
logger.log("info", format!("Running FC backup for {}", cfg.name));
|
||||
|
||||
let file_path = backup_dir.join(format!("{}.dump", cfg.generated_id));
|
||||
let url = format!(
|
||||
"postgresql://{}:{}@{}:{}/{}",
|
||||
cfg.username, cfg.password, cfg.host, cfg.port, cfg.database
|
||||
);
|
||||
|
||||
let start = Instant::now();
|
||||
let output = Command::new(&pg_dump)
|
||||
.arg("--dbname").arg(&url)
|
||||
.arg("--host").arg(&cfg.host)
|
||||
.arg("--port").arg(cfg.port.to_string())
|
||||
.arg("--username").arg(&cfg.username)
|
||||
.arg("--dbname").arg(&cfg.database)
|
||||
.arg("-Fc")
|
||||
.arg("-f").arg(&file_path)
|
||||
.arg("-v")
|
||||
.arg("--compress=3")
|
||||
.envs(env)
|
||||
.output();
|
||||
let duration_ms = start.elapsed().as_millis() as f64;
|
||||
|
||||
@@ -87,19 +89,19 @@ pub async fn run(
|
||||
return Err(e.into());
|
||||
}
|
||||
|
||||
let url = format!(
|
||||
"postgresql://{}:{}@{}:{}/{}",
|
||||
cfg.username, cfg.password, cfg.host, cfg.port, cfg.database
|
||||
);
|
||||
let cmd_label = format!("pg_dump -Fd {}", url);
|
||||
let cmd_label = format!("pg_dump -Fd {}@{}:{}/{}", cfg.username, cfg.host, cfg.port, cfg.database);
|
||||
|
||||
let start = Instant::now();
|
||||
let output = Command::new(&pg_dump)
|
||||
.arg("--dbname").arg(&url)
|
||||
.arg("--host").arg(&cfg.host)
|
||||
.arg("--port").arg(cfg.port.to_string())
|
||||
.arg("--username").arg(&cfg.username)
|
||||
.arg("--dbname").arg(&cfg.database)
|
||||
.arg("-Fd")
|
||||
.arg("-j").arg("4")
|
||||
.arg("-f").arg(&dump_dir)
|
||||
.arg("-v")
|
||||
.envs(env)
|
||||
.output();
|
||||
let duration_ms = start.elapsed().as_millis() as f64;
|
||||
|
||||
|
||||
@@ -1,18 +1,23 @@
|
||||
use crate::domain::postgres::format::PostgresDumpFormat;
|
||||
use crate::services::config::DatabaseConfig;
|
||||
use crate::settings::CONFIG;
|
||||
use anyhow::Result;
|
||||
use std::path::Path;
|
||||
use tokio_postgres::{Client, NoTls};
|
||||
use tokio_postgres::{Client, Config, NoTls};
|
||||
use tracing::{error, info};
|
||||
|
||||
pub async fn connect(cfg: &DatabaseConfig) -> Result<Client> {
|
||||
info!("Connecting to postgres database {}:{}", cfg.host, cfg.port);
|
||||
let dsn = format!(
|
||||
"host={} port={} user={} password={} dbname={}",
|
||||
cfg.host, cfg.port, cfg.username, cfg.password, cfg.database
|
||||
);
|
||||
|
||||
let (client, connection) = tokio_postgres::connect(&dsn, NoTls).await?;
|
||||
let mut config = Config::new();
|
||||
config
|
||||
.host(&cfg.host)
|
||||
.port(cfg.port)
|
||||
.user(&cfg.username)
|
||||
.password(&cfg.password)
|
||||
.dbname(&cfg.database);
|
||||
|
||||
let (client, connection) = config.connect(NoTls).await?;
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = connection.await {
|
||||
error!("Postgres connection error: {}", e);
|
||||
@@ -28,11 +33,91 @@ pub async fn server_version(cfg: &DatabaseConfig) -> Result<String> {
|
||||
Ok(version)
|
||||
}
|
||||
|
||||
/// Resolves the `bin` directory of a PostgreSQL installation for the given
|
||||
/// major version, in a cross-platform way.
|
||||
///
|
||||
/// Resolution order:
|
||||
/// 1. The `PG_BIN_DIR` environment variable, if set, is used as-is. This
|
||||
/// allows users/CI to override detection for non-standard installs
|
||||
/// (e.g. portable PostgreSQL distributions, custom install locations).
|
||||
/// 2. Platform-specific default install locations (Debian/Ubuntu packages,
|
||||
/// the official Windows installer, Homebrew/Postgres.app on macOS, and
|
||||
/// common RPM-based layouts on other Linux distros).
|
||||
/// 3. A `PATH` lookup for `pg_dump` (`pg_dump.exe` on Windows), returning
|
||||
/// its parent directory.
|
||||
/// 4. The historical Debian/Ubuntu path as a last-resort fallback, so the
|
||||
/// function keeps returning a `PathBuf` (never panics) even when nothing
|
||||
/// was found, preserving the previous behavior for callers.
|
||||
///
|
||||
/// The override is sourced from `CONFIG.pg_bin_dir` (the `PG_BIN_DIR`
|
||||
/// environment variable). An empty value means "unset" and falls through to
|
||||
/// detection.
|
||||
pub fn select_pg_path(version: &str) -> std::path::PathBuf {
|
||||
select_pg_path_with(version, &CONFIG.pg_bin_dir)
|
||||
}
|
||||
|
||||
/// Inner resolver behind [`select_pg_path`], parameterized over the
|
||||
/// `PG_BIN_DIR` override. Kept pure (no env / no `CONFIG` access) so it is
|
||||
/// unit-testable without mutating process-global state.
|
||||
pub(crate) fn select_pg_path_with(version: &str, pg_bin_dir: &str) -> std::path::PathBuf {
|
||||
let major = version.split('.').next().unwrap_or("17");
|
||||
|
||||
if !pg_bin_dir.is_empty() {
|
||||
return pg_bin_dir.into();
|
||||
}
|
||||
|
||||
let candidates: Vec<std::path::PathBuf> = if cfg!(target_os = "windows") {
|
||||
vec![
|
||||
// Default install path used by the official EDB Windows installer
|
||||
format!(r"C:\Program Files\PostgreSQL\{major}\bin").into(),
|
||||
format!(r"C:\Program Files (x86)\PostgreSQL\{major}\bin").into(),
|
||||
]
|
||||
} else if cfg!(target_os = "macos") {
|
||||
vec![
|
||||
// Homebrew on Apple Silicon
|
||||
format!("/opt/homebrew/opt/postgresql@{major}/bin").into(),
|
||||
// Homebrew on Intel
|
||||
format!("/usr/local/opt/postgresql@{major}/bin").into(),
|
||||
// Postgres.app
|
||||
format!("/Applications/Postgres.app/Contents/Versions/{major}/bin").into(),
|
||||
]
|
||||
} else {
|
||||
vec![
|
||||
// Debian/Ubuntu packages
|
||||
format!("/usr/lib/postgresql/{major}/bin").into(),
|
||||
// Common RPM-based distro layout
|
||||
format!("/usr/pgsql-{major}/bin").into(),
|
||||
]
|
||||
};
|
||||
|
||||
if let Some(found) = candidates.into_iter().find(|p| pg_dump_exists_in(p)) {
|
||||
return found;
|
||||
}
|
||||
|
||||
if let Some(dir) = find_pg_dump_in_path() {
|
||||
return dir;
|
||||
}
|
||||
|
||||
format!("/usr/lib/postgresql/{}/bin", major).into()
|
||||
}
|
||||
|
||||
pub(crate) fn pg_dump_binary_name() -> &'static str {
|
||||
if cfg!(target_os = "windows") {
|
||||
"pg_dump.exe"
|
||||
} else {
|
||||
"pg_dump"
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn pg_dump_exists_in(dir: &std::path::Path) -> bool {
|
||||
dir.join(pg_dump_binary_name()).is_file()
|
||||
}
|
||||
|
||||
fn find_pg_dump_in_path() -> Option<std::path::PathBuf> {
|
||||
let path_var = std::env::var_os("PATH")?;
|
||||
std::env::split_paths(&path_var).find(|dir| pg_dump_exists_in(dir))
|
||||
}
|
||||
|
||||
pub async fn terminate_connections(cfg: &DatabaseConfig) -> Result<()> {
|
||||
let mut admin = cfg.clone();
|
||||
admin.database = "postgres".to_string().into();
|
||||
@@ -68,6 +153,7 @@ pub async fn detect_format_from_size(cfg: &DatabaseConfig) -> PostgresDumpFormat
|
||||
"Detecting database format {:?} - {:?}",
|
||||
cfg.name, cfg.generated_id
|
||||
);
|
||||
|
||||
let client = match connect(cfg).await {
|
||||
Ok(c) => c,
|
||||
Err(_) => return PostgresDumpFormat::Fc,
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
use anyhow::Result;
|
||||
use async_trait::async_trait;
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use super::{backup, format::PostgresDumpFormat, ping, restore};
|
||||
use crate::domain::factory::Database;
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
@@ -18,6 +18,12 @@ impl PostgresDatabase {
|
||||
pub fn new(cfg: DatabaseConfig, format: PostgresDumpFormat) -> Self {
|
||||
Self { cfg, format }
|
||||
}
|
||||
|
||||
fn build_env(&self) -> HashMap<String, String> {
|
||||
let mut envs = std::env::vars().collect::<HashMap<_, _>>();
|
||||
envs.insert("PGPASSWORD".to_string(), self.cfg.password.to_string());
|
||||
envs
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -35,14 +41,28 @@ impl Database for PostgresDatabase {
|
||||
|
||||
async fn backup(&self, dir: &Path, logger: Arc<JobLogger>) -> Result<PathBuf> {
|
||||
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Backup.as_str()).await?;
|
||||
let res = backup::run(self.cfg.clone(), self.format, dir.to_path_buf(), logger).await;
|
||||
let res = backup::run(
|
||||
self.cfg.clone(),
|
||||
self.format,
|
||||
dir.to_path_buf(),
|
||||
self.build_env(),
|
||||
logger,
|
||||
)
|
||||
.await;
|
||||
FileLock::release(&self.cfg.generated_id).await?;
|
||||
res
|
||||
}
|
||||
|
||||
async fn restore(&self, file: &Path, logger: Arc<JobLogger>) -> Result<()> {
|
||||
FileLock::acquire(&self.cfg.generated_id, DbOpLock::Restore.as_str()).await?;
|
||||
let res = restore::run(self.cfg.clone(), self.format, file.to_path_buf(), logger).await;
|
||||
let res = restore::run(
|
||||
self.cfg.clone(),
|
||||
self.format,
|
||||
file.to_path_buf(),
|
||||
self.build_env(),
|
||||
logger,
|
||||
)
|
||||
.await;
|
||||
FileLock::release(&self.cfg.generated_id).await?;
|
||||
res
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
pub mod backup;
|
||||
mod connection;
|
||||
pub(crate) mod connection;
|
||||
pub mod database;
|
||||
mod format;
|
||||
mod ping;
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use anyhow::Result;
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::process::Command;
|
||||
use std::sync::Arc;
|
||||
@@ -13,6 +14,7 @@ pub async fn run(
|
||||
cfg: DatabaseConfig,
|
||||
format: PostgresDumpFormat,
|
||||
restore_file: PathBuf,
|
||||
env: HashMap<String, String>,
|
||||
logger: Arc<JobLogger>,
|
||||
) -> Result<()> {
|
||||
tokio::task::spawn_blocking(move || -> Result<()> {
|
||||
@@ -39,11 +41,6 @@ pub async fn run(
|
||||
}
|
||||
logger.log("info", format!("Connections terminated for database {}", cfg.name));
|
||||
|
||||
let url = format!(
|
||||
"postgresql://{}:{}@{}:{}/{}",
|
||||
cfg.username, cfg.password, cfg.host, cfg.port, cfg.database
|
||||
);
|
||||
|
||||
match format {
|
||||
PostgresDumpFormat::Fc => {
|
||||
logger.log("info", format!("Running FC restore for {}", cfg.name));
|
||||
@@ -54,11 +51,13 @@ pub async fn run(
|
||||
.arg("--clean")
|
||||
.arg("--if-exists")
|
||||
// .arg("--create")
|
||||
.arg("--dbname")
|
||||
.arg(&url)
|
||||
.arg("--host").arg(&cfg.host)
|
||||
.arg("--port").arg(cfg.port.to_string())
|
||||
.arg("--username").arg(&cfg.username)
|
||||
.arg("--dbname").arg(&cfg.database)
|
||||
.arg("-v")
|
||||
.arg(&restore_file)
|
||||
.env("PGPASSWORD", &cfg.password)
|
||||
.envs(env)
|
||||
.output();
|
||||
|
||||
let duration_ms = start.elapsed().as_millis() as f64;
|
||||
@@ -154,13 +153,15 @@ pub async fn run(
|
||||
.arg("--clean")
|
||||
.arg("--if-exists")
|
||||
// .arg("--create")
|
||||
.arg("--dbname")
|
||||
.arg(&url)
|
||||
.arg("--host").arg(&cfg.host)
|
||||
.arg("--port").arg(cfg.port.to_string())
|
||||
.arg("--username").arg(&cfg.username)
|
||||
.arg("--dbname").arg(&cfg.database)
|
||||
.arg("-v")
|
||||
.arg("-j")
|
||||
.arg("4")
|
||||
.arg(dump_dir)
|
||||
.env("PGPASSWORD", &cfg.password)
|
||||
.envs(env)
|
||||
.output();
|
||||
|
||||
let duration_ms = start.elapsed().as_millis() as f64;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
#![allow(dead_code)]
|
||||
|
||||
use crate::utils::deserializer::deserialize_snake_case;
|
||||
use crate::utils::deserializer::{deserialize_snake_case, string_or_number_to_string};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use toml::Value;
|
||||
|
||||
@@ -54,4 +54,6 @@ pub struct RestoreInfo {
|
||||
pub file: Option<String>,
|
||||
#[serde(rename = "metaFile")]
|
||||
pub meta_file: Option<String>,
|
||||
#[serde(default, deserialize_with = "string_or_number_to_string")]
|
||||
pub size: Option<String>,
|
||||
}
|
||||
|
||||
@@ -115,6 +115,28 @@ impl BackupService {
|
||||
storage_id,
|
||||
upload_result.error.as_deref().unwrap_or("unknown error")
|
||||
));
|
||||
|
||||
// `backup_upload_init` opened a per-storage record; close it as "failed"
|
||||
// so the server is notified of the failure (no path/size on this path).
|
||||
if let Err(err) = ctx_clone
|
||||
.api
|
||||
.backup_upload_status(
|
||||
ctx_clone.edge_key.agent_id.clone(),
|
||||
generated_id.clone(),
|
||||
backup_storage_id.clone(),
|
||||
status,
|
||||
String::new(),
|
||||
0u64,
|
||||
backup_id,
|
||||
)
|
||||
.await
|
||||
{
|
||||
logger_clone.log("error", format!(
|
||||
"Failed-status update failed for {}: {}",
|
||||
storage_id, err
|
||||
));
|
||||
}
|
||||
|
||||
return upload_result;
|
||||
}
|
||||
|
||||
|
||||
@@ -54,6 +54,7 @@ pub struct DatabaseConfig {
|
||||
pub host: String,
|
||||
pub generated_id: String,
|
||||
pub path: String,
|
||||
pub max_packet_size: String,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
@@ -75,6 +76,7 @@ pub struct InputDatabaseConfig {
|
||||
pub host: Option<String>,
|
||||
pub generated_id: String,
|
||||
pub path: Option<String>,
|
||||
pub max_packet_size: Option<String>,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
@@ -214,6 +216,13 @@ impl ConfigService {
|
||||
_ => optional(&db.path),
|
||||
};
|
||||
|
||||
let max_packet_size = match db.db_type {
|
||||
DbType::Mysql | DbType::Mariadb => {
|
||||
db.max_packet_size.unwrap_or_else(|| "512M".to_string())
|
||||
}
|
||||
_ => String::new(),
|
||||
};
|
||||
|
||||
databases.push(DatabaseConfig {
|
||||
name: db.name,
|
||||
database: database_name,
|
||||
@@ -224,6 +233,7 @@ impl ConfigService {
|
||||
port,
|
||||
generated_id: db.generated_id,
|
||||
path: path_val,
|
||||
max_packet_size,
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
+1
-1
@@ -4,4 +4,4 @@ pub mod config;
|
||||
pub mod cron;
|
||||
pub mod restore;
|
||||
pub mod status;
|
||||
mod storage;
|
||||
pub mod storage;
|
||||
|
||||
@@ -20,6 +20,8 @@ impl RestoreService {
|
||||
return;
|
||||
};
|
||||
|
||||
let expected_size = db.data.restore.size.clone();
|
||||
|
||||
let service = Self {
|
||||
ctx: self.ctx.clone(),
|
||||
};
|
||||
@@ -27,7 +29,10 @@ impl RestoreService {
|
||||
let db_cfg = cfg.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = service.execute_restore(db_cfg, file_to_restore).await {
|
||||
if let Err(e) = service
|
||||
.execute_restore(db_cfg, file_to_restore, expected_size)
|
||||
.await
|
||||
{
|
||||
error!("Restore failed: {}", e);
|
||||
}
|
||||
});
|
||||
|
||||
@@ -1,20 +1,40 @@
|
||||
use super::service::RestoreService;
|
||||
|
||||
use anyhow::Result;
|
||||
use futures::StreamExt;
|
||||
use reqwest::{Client, Url};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
|
||||
fn human_size(bytes: u64) -> String {
|
||||
if bytes >= 1024 * 1024 {
|
||||
format!("{} MB", bytes / 1024 / 1024)
|
||||
} else if bytes >= 1024 {
|
||||
format!("{} KB", bytes / 1024)
|
||||
} else {
|
||||
format!("{bytes} B")
|
||||
}
|
||||
}
|
||||
|
||||
impl RestoreService {
|
||||
pub async fn download_backup(&self, file_url: &str, tmp_path: &Path, logger: Arc<JobLogger>) -> Result<PathBuf> {
|
||||
pub async fn download_backup(
|
||||
&self,
|
||||
file_url: &str,
|
||||
tmp_path: &Path,
|
||||
logger: Arc<JobLogger>,
|
||||
expected_size: Option<String>,
|
||||
) -> Result<PathBuf> {
|
||||
logger.log("info", "Start downloading backup archive".to_string());
|
||||
|
||||
let client = Client::new();
|
||||
|
||||
let response = client.get(file_url).send().await?;
|
||||
let status = response.status();
|
||||
|
||||
if !response.status().is_success() {
|
||||
if !status.is_success() {
|
||||
logger.log("error", "Failed to download".to_string());
|
||||
anyhow::bail!("download failed");
|
||||
}
|
||||
@@ -39,11 +59,66 @@ impl RestoreService {
|
||||
|
||||
let path = tmp_path.join(&filename);
|
||||
|
||||
let bytes = response.bytes().await?;
|
||||
let total = expected_size
|
||||
.as_deref()
|
||||
.and_then(|s| s.trim().parse::<u64>().ok())
|
||||
.filter(|&n| n > 0);
|
||||
|
||||
tokio::fs::write(&path, &bytes).await?;
|
||||
logger.log(
|
||||
"info",
|
||||
format!(
|
||||
"Downloading backup '{}' ({})",
|
||||
filename,
|
||||
total.map(human_size).unwrap_or_else(|| "unknown size".to_string())
|
||||
),
|
||||
);
|
||||
|
||||
let start = Instant::now();
|
||||
let mut file = tokio::fs::File::create(&path).await?;
|
||||
let mut stream = response.bytes_stream();
|
||||
let mut downloaded: u64 = 0;
|
||||
let mut next_pct: u64 = 10;
|
||||
|
||||
while let Some(chunk) = stream.next().await {
|
||||
let chunk = chunk?;
|
||||
file.write_all(&chunk).await?;
|
||||
downloaded += chunk.len() as u64;
|
||||
|
||||
if let Some(total) = total {
|
||||
let pct = (downloaded.saturating_mul(100) / total).min(100);
|
||||
let milestone = pct / 10 * 10;
|
||||
if milestone >= next_pct {
|
||||
logger.log(
|
||||
"info",
|
||||
format!(
|
||||
"Download progress: {}% ({} / {} bytes)",
|
||||
milestone, downloaded, total
|
||||
),
|
||||
);
|
||||
next_pct = milestone + 10;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
file.flush().await?;
|
||||
|
||||
if downloaded == 0 {
|
||||
logger.log(
|
||||
"warn",
|
||||
format!("Downloaded 0 bytes (status {status}); backup body was empty"),
|
||||
);
|
||||
}
|
||||
|
||||
logger.log(
|
||||
"info",
|
||||
format!(
|
||||
"Backup downloaded to {} ( {} bytes in {:.1}s)",
|
||||
path.display(),
|
||||
downloaded,
|
||||
start.elapsed().as_secs_f64()
|
||||
),
|
||||
);
|
||||
|
||||
logger.log("info", format!("Backup downloaded to {}", path.display()));
|
||||
Ok(path)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,7 +7,12 @@ use std::time::Instant;
|
||||
use tempfile::TempDir;
|
||||
|
||||
impl RestoreService {
|
||||
pub async fn execute_restore(&self, cfg: DatabaseConfig, file_url: String) -> Result<()> {
|
||||
pub async fn execute_restore(
|
||||
&self,
|
||||
cfg: DatabaseConfig,
|
||||
file_url: String,
|
||||
expected_size: Option<String>,
|
||||
) -> Result<()> {
|
||||
let logger = Arc::new(JobLogger::new());
|
||||
let start = Instant::now();
|
||||
|
||||
@@ -18,7 +23,9 @@ impl RestoreService {
|
||||
|
||||
logger.log("info", format!("Created temp directory {}", tmp_path.display()));
|
||||
|
||||
let downloaded = self.download_backup(&file_url, tmp_path, Arc::clone(&logger)).await?;
|
||||
let downloaded = self
|
||||
.download_backup(&file_url, tmp_path, Arc::clone(&logger), expected_size)
|
||||
.await?;
|
||||
|
||||
let backup_file = self.prepare_archive(downloaded, tmp_path, Arc::clone(&logger)).await?;
|
||||
|
||||
|
||||
@@ -5,6 +5,7 @@ use crate::services::api::models::agent::status::DatabaseStorage;
|
||||
use crate::services::backup::models::{BackupResult, UploadResult};
|
||||
use crate::utils::common::BackupMethod;
|
||||
use async_trait::async_trait;
|
||||
use providers::azure_blob;
|
||||
use providers::google_drive;
|
||||
use providers::local;
|
||||
use providers::s3;
|
||||
@@ -31,6 +32,7 @@ pub fn get_provider(storage: &DatabaseStorage) -> Option<Box<dyn StorageProvider
|
||||
match storage.provider.as_str() {
|
||||
"local" => Some(Box::new(local::LocalProvider {})),
|
||||
"s3" => Some(Box::new(s3::S3Provider {})),
|
||||
"blob" => Some(Box::new(azure_blob::AzureBlobProvider {})),
|
||||
"google-drive" => Some(Box::new(google_drive::GoogleDriveProvider {})),
|
||||
_ => {
|
||||
error!("Unknown storage provider: {}", storage.provider);
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
use anyhow::{Context as _, Result, anyhow};
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use chrono::{Duration, Utc};
|
||||
use openssl::hash::MessageDigest;
|
||||
use openssl::pkey::PKey;
|
||||
use openssl::sign::Signer;
|
||||
use url::Url;
|
||||
use azure_core::http::RequestContent;
|
||||
use azure_storage_blob::clients::{BlobClient, BlockBlobClient};
|
||||
use azure_storage_blob::models::BlockLookupList;
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use futures::{Stream, StreamExt};
|
||||
use std::pin::Pin;
|
||||
use tracing::info;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct ResolvedAzure {
|
||||
pub account_name: String,
|
||||
pub account_key: String,
|
||||
pub blob_endpoint: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
pub enum SasResource {
|
||||
Blob,
|
||||
#[allow(dead_code)]
|
||||
Container,
|
||||
}
|
||||
|
||||
impl SasResource {
|
||||
fn code(self) -> &'static str {
|
||||
match self { SasResource::Blob => "b", SasResource::Container => "c" }
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) const SAS_VERSION: &str = "2022-11-02";
|
||||
|
||||
pub(crate) fn hmac_sha256_b64(key: &[u8], data: &str) -> Result<String> {
|
||||
let pkey = PKey::hmac(key).context("hmac key")?;
|
||||
let mut signer = Signer::new(MessageDigest::sha256(), &pkey).context("signer")?;
|
||||
signer.update(data.as_bytes()).context("signer update")?;
|
||||
let sig = signer.sign_to_vec().context("sign")?;
|
||||
Ok(STANDARD.encode(sig))
|
||||
}
|
||||
|
||||
/// Build Service SAS query pairs (raw, un-encoded) for `canonical_resource`
|
||||
/// e.g. `/blob/{account}/{container}/{blob}`.
|
||||
pub fn build_service_sas(
|
||||
resolved: &ResolvedAzure,
|
||||
canonical_resource: &str,
|
||||
resource: SasResource,
|
||||
permissions: &str,
|
||||
) -> Result<Vec<(String, String)>> {
|
||||
let key = STANDARD
|
||||
.decode(&resolved.account_key)
|
||||
.map_err(|_| anyhow!("account key is not valid base64"))?;
|
||||
|
||||
let signed_start = String::new();
|
||||
let signed_expiry = (Utc::now() + Duration::hours(1))
|
||||
.format("%Y-%m-%dT%H:%M:%SZ").to_string();
|
||||
let signed_protocol = "https,http"; // Azurite is http
|
||||
let signed_resource = resource.code();
|
||||
|
||||
let string_to_sign = format!(
|
||||
"{sp}\n{st}\n{se}\n{canon}\n{si}\n{sip}\n{spr}\n{sv}\n{sr}\n{snap}\n{enc}\n{rscc}\n{rscd}\n{rsce}\n{rscl}\n{rsct}",
|
||||
sp = permissions, st = signed_start, se = signed_expiry, canon = canonical_resource,
|
||||
si = "", sip = "", spr = signed_protocol, sv = SAS_VERSION, sr = signed_resource,
|
||||
snap = "", enc = "", rscc = "", rscd = "", rsce = "", rscl = "", rsct = "",
|
||||
);
|
||||
|
||||
let sig = hmac_sha256_b64(&key, &string_to_sign)?;
|
||||
|
||||
Ok(vec![
|
||||
("sv".into(), SAS_VERSION.into()),
|
||||
("sr".into(), signed_resource.into()),
|
||||
("sp".into(), permissions.into()),
|
||||
("se".into(), signed_expiry),
|
||||
("spr".into(), signed_protocol.into()),
|
||||
("sig".into(), sig),
|
||||
])
|
||||
}
|
||||
|
||||
/// Build a SAS-scoped URL for a blob (or container when `blob` is empty).
|
||||
pub fn build_sas_url(
|
||||
resolved: &ResolvedAzure,
|
||||
container: &str,
|
||||
blob: &str,
|
||||
resource: SasResource,
|
||||
permissions: &str,
|
||||
) -> Result<Url> {
|
||||
let canonical = if blob.is_empty() {
|
||||
format!("/blob/{}/{}", resolved.account_name, container)
|
||||
} else {
|
||||
format!("/blob/{}/{}/{}", resolved.account_name, container, blob)
|
||||
};
|
||||
let pairs = build_service_sas(resolved, &canonical, resource, permissions)?;
|
||||
|
||||
let base = if blob.is_empty() {
|
||||
format!("{}/{}", resolved.blob_endpoint.trim_end_matches('/'), container)
|
||||
} else {
|
||||
format!("{}/{}/{}", resolved.blob_endpoint.trim_end_matches('/'), container, blob)
|
||||
};
|
||||
|
||||
let mut url = Url::parse(&base).context("invalid blob endpoint/url")?;
|
||||
{
|
||||
let mut qp = url.query_pairs_mut();
|
||||
for (k, v) in pairs { qp.append_pair(&k, &v); }
|
||||
}
|
||||
Ok(url)
|
||||
}
|
||||
|
||||
/// Default block size for the provider path (mirrors the S3 provider's PART_SIZE).
|
||||
pub const BLOCK_SIZE: usize = 100 * 1024 * 1024;
|
||||
|
||||
type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, std::io::Error>> + Send>>;
|
||||
|
||||
/// Stage one block under a zero-padded sequential id; records the RAW id bytes.
|
||||
async fn stage_block(
|
||||
bbc: &BlockBlobClient,
|
||||
index: u32,
|
||||
block: Bytes,
|
||||
block_ids: &mut Vec<Vec<u8>>,
|
||||
) -> Result<()> {
|
||||
let raw_id = format!("{index:032}").into_bytes();
|
||||
let len = block.len() as u64;
|
||||
bbc.stage_block(&raw_id, len, RequestContent::from(block.to_vec()), None)
|
||||
.await
|
||||
.map_err(|e| anyhow!("stage_block {index} failed: {e}"))?;
|
||||
block_ids.push(raw_id);
|
||||
info!("staged azure block {index} ({len} bytes)");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Stream `body` to `{container}/{blob}` using Azure block upload. Never buffers the
|
||||
/// full payload: at most one `block_size` block plus one inbound chunk is resident
|
||||
/// (mirrors the S3 provider's per-part guarantee).
|
||||
///
|
||||
/// Assumes the container already exists — S3-faithful, no container creation. Uncommitted
|
||||
/// blocks are garbage-collected by Azure if `commit_block_list` is never reached, so no
|
||||
/// explicit abort is needed on the error path (unlike S3 multipart).
|
||||
pub async fn upload_stream_to_azure(
|
||||
resolved: &ResolvedAzure,
|
||||
container: &str,
|
||||
blob: &str,
|
||||
mut body: ByteStream,
|
||||
block_size: usize,
|
||||
) -> Result<()> {
|
||||
let url = build_sas_url(resolved, container, blob, SasResource::Blob, "cw")?;
|
||||
let blob_client = BlobClient::new(url, None, None).context("blob client")?;
|
||||
let bbc = blob_client.block_blob_client();
|
||||
|
||||
let mut buffer = BytesMut::with_capacity(block_size);
|
||||
let mut block_ids: Vec<Vec<u8>> = Vec::new();
|
||||
let mut index: u32 = 0;
|
||||
|
||||
while let Some(item) = body.next().await {
|
||||
let bytes = item.context("stream error during upload")?;
|
||||
buffer.extend_from_slice(&bytes);
|
||||
|
||||
while buffer.len() >= block_size {
|
||||
let block = buffer.split_to(block_size).freeze();
|
||||
stage_block(&bbc, index, block, &mut block_ids).await?;
|
||||
index += 1;
|
||||
}
|
||||
}
|
||||
|
||||
if !buffer.is_empty() {
|
||||
let block = buffer.split().freeze();
|
||||
stage_block(&bbc, index, block, &mut block_ids).await?;
|
||||
}
|
||||
|
||||
if block_ids.is_empty() {
|
||||
stage_block(&bbc, 0, Bytes::new(), &mut block_ids).await?;
|
||||
}
|
||||
|
||||
let block_list = BlockLookupList {
|
||||
latest: Some(block_ids),
|
||||
..Default::default()
|
||||
};
|
||||
bbc.commit_block_list(block_list.try_into()?, None)
|
||||
.await
|
||||
.map_err(|e| anyhow!("commit_block_list failed: {e}"))?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -0,0 +1,134 @@
|
||||
pub mod helpers;
|
||||
mod models;
|
||||
|
||||
use crate::core::context::Context;
|
||||
use crate::services::api::models::agent::status::DatabaseStorage;
|
||||
use crate::services::backup::models::{BackupResult, UploadResult};
|
||||
use crate::services::storage::StorageProvider;
|
||||
use crate::services::storage::providers::azure_blob::helpers::{BLOCK_SIZE, upload_stream_to_azure};
|
||||
use crate::services::storage::providers::azure_blob::models::AzureBlobProviderConfig;
|
||||
use crate::utils::common::BackupMethod;
|
||||
use crate::utils::file::{full_file_name, full_file_path};
|
||||
use crate::utils::stream::build_stream;
|
||||
use async_trait::async_trait;
|
||||
use std::sync::Arc;
|
||||
use tokio::fs;
|
||||
use tracing::{error, info};
|
||||
|
||||
pub struct AzureBlobProvider {}
|
||||
|
||||
#[async_trait]
|
||||
impl StorageProvider for AzureBlobProvider {
|
||||
async fn upload(
|
||||
&self,
|
||||
ctx: Arc<Context>,
|
||||
result: BackupResult,
|
||||
_method: BackupMethod,
|
||||
storage: &DatabaseStorage,
|
||||
encrypt: Option<bool>,
|
||||
) -> UploadResult {
|
||||
let Some(file_path) = result.backup_file else {
|
||||
return UploadResult {
|
||||
storage_id: storage.id.clone(),
|
||||
success: false,
|
||||
error: Some("Missing backup file path".to_string()),
|
||||
remote_file_path: None,
|
||||
total_size: None,
|
||||
};
|
||||
};
|
||||
|
||||
let total_size = match fs::metadata(&file_path).await {
|
||||
Ok(meta) => meta.len(),
|
||||
Err(e) => {
|
||||
error!("Failed to get file size: {}", e);
|
||||
return UploadResult {
|
||||
storage_id: storage.id.clone(),
|
||||
success: false,
|
||||
error: Some(e.to_string()),
|
||||
remote_file_path: None,
|
||||
total_size: None,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
let encrypt = encrypt.unwrap_or(false);
|
||||
|
||||
let upload = match build_stream(&file_path, encrypt, &ctx.edge_key.master_key_b64).await {
|
||||
Ok(u) => u,
|
||||
Err(e) => {
|
||||
error!("Stream build failed: {}", e);
|
||||
return UploadResult {
|
||||
storage_id: storage.id.clone(),
|
||||
success: false,
|
||||
error: Some(e.to_string()),
|
||||
remote_file_path: None,
|
||||
total_size: None,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
let config: AzureBlobProviderConfig = match storage.clone().config.try_into() {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
return UploadResult {
|
||||
storage_id: storage.id.clone(),
|
||||
success: false,
|
||||
error: Some(e.to_string()),
|
||||
remote_file_path: None,
|
||||
total_size: None,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
let resolved = match config.resolve() {
|
||||
Ok(r) => r,
|
||||
Err(e) => {
|
||||
return UploadResult {
|
||||
storage_id: storage.id.clone(),
|
||||
success: false,
|
||||
error: Some(e.to_string()),
|
||||
remote_file_path: None,
|
||||
total_size: None,
|
||||
};
|
||||
}
|
||||
};
|
||||
|
||||
let file_name = full_file_name(encrypt);
|
||||
let remote_file_path = full_file_path(&file_name);
|
||||
info!(
|
||||
"Starting block upload to azure blob {}/{}",
|
||||
config.container_name, remote_file_path
|
||||
);
|
||||
|
||||
match upload_stream_to_azure(
|
||||
&resolved,
|
||||
&config.container_name,
|
||||
&remote_file_path,
|
||||
upload.stream,
|
||||
BLOCK_SIZE,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(_) => {
|
||||
info!("Azure blob upload successful: {}", remote_file_path);
|
||||
UploadResult {
|
||||
storage_id: storage.id.clone(),
|
||||
success: true,
|
||||
error: None,
|
||||
remote_file_path: Some(remote_file_path),
|
||||
total_size: Some(total_size),
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Azure blob upload failed: {:?}", e);
|
||||
UploadResult {
|
||||
storage_id: storage.id.clone(),
|
||||
success: false,
|
||||
error: Some(e.to_string()),
|
||||
remote_file_path: None,
|
||||
total_size: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,64 @@
|
||||
use crate::services::storage::providers::azure_blob::helpers::ResolvedAzure;
|
||||
use anyhow::{Result, anyhow};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Debug, Deserialize, Serialize)]
|
||||
pub struct AzureBlobProviderConfig {
|
||||
pub account_name: String,
|
||||
pub account_key: String,
|
||||
pub container_name: String,
|
||||
#[serde(default)]
|
||||
pub connection_string: String,
|
||||
#[serde(default)]
|
||||
pub endpoint_url: Option<String>,
|
||||
}
|
||||
|
||||
fn parse_connection_string(cs: &str) -> std::collections::HashMap<String, String> {
|
||||
cs.split(';')
|
||||
.filter(|s| !s.trim().is_empty())
|
||||
.filter_map(|pair| {
|
||||
let mut it = pair.splitn(2, '=');
|
||||
let k = it.next()?.trim().to_string();
|
||||
let v = it.next()?.trim().to_string();
|
||||
Some((k, v))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
impl AzureBlobProviderConfig {
|
||||
/// Resolve effective connection params, preferring the connection string when non-empty.
|
||||
pub fn resolve(&self) -> Result<ResolvedAzure> {
|
||||
if !self.connection_string.trim().is_empty() {
|
||||
let map = parse_connection_string(&self.connection_string);
|
||||
let account_name = map
|
||||
.get("AccountName")
|
||||
.cloned()
|
||||
.unwrap_or_else(|| self.account_name.clone());
|
||||
let account_key = map
|
||||
.get("AccountKey")
|
||||
.cloned()
|
||||
.unwrap_or_else(|| self.account_key.clone());
|
||||
let blob_endpoint = map
|
||||
.get("BlobEndpoint")
|
||||
.cloned()
|
||||
.ok_or_else(|| anyhow!("connection string missing BlobEndpoint"))?;
|
||||
return Ok(ResolvedAzure {
|
||||
account_name,
|
||||
account_key,
|
||||
blob_endpoint,
|
||||
});
|
||||
}
|
||||
|
||||
let blob_endpoint = self
|
||||
.endpoint_url
|
||||
.clone()
|
||||
.filter(|s| !s.trim().is_empty())
|
||||
.ok_or_else(|| anyhow!("endpointUrl required when connectionString is empty"))?;
|
||||
|
||||
Ok(ResolvedAzure {
|
||||
account_name: self.account_name.clone(),
|
||||
account_key: self.account_key.clone(),
|
||||
blob_endpoint,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
pub mod azure_blob;
|
||||
pub mod google_drive;
|
||||
pub mod local;
|
||||
pub mod s3;
|
||||
|
||||
@@ -11,6 +11,7 @@ pub struct Settings {
|
||||
pub edge_key: String,
|
||||
pub databases_config_file: String,
|
||||
pub data_path: String,
|
||||
pub pg_bin_dir: String,
|
||||
pub pooling: usize,
|
||||
pub timezone: String,
|
||||
pub log: String,
|
||||
@@ -59,6 +60,7 @@ impl Settings {
|
||||
databases_config_file: env::var("DATABASES_CONFIG_FILE")
|
||||
.unwrap_or_else(|_| "config.json".into()),
|
||||
data_path: env::var("DATA_PATH").unwrap_or_else(|_| "/config".into()),
|
||||
pg_bin_dir: env::var("PG_BIN_DIR").unwrap_or_default(),
|
||||
pooling: pooling_seconds,
|
||||
timezone: tz,
|
||||
log: env::var("LOG").unwrap_or_else(|_| "info".into()),
|
||||
|
||||
@@ -39,6 +39,7 @@ async fn create_config() -> (ContainerAsync<GenericImage>, DatabaseConfig) {
|
||||
host,
|
||||
generated_id: "3c445eb4-c2c6-4bde-a423-ee1385dcf6d2".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "".to_string(),
|
||||
};
|
||||
|
||||
(container, config)
|
||||
|
||||
@@ -31,6 +31,7 @@ async fn create_config() -> (ContainerAsync<Mariadb>, DatabaseConfig) {
|
||||
host: host.to_string(),
|
||||
generated_id: "3c4b4eb4-c2c6-4bde-a423-ee1385dcf6d2".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "512M".to_string(),
|
||||
};
|
||||
|
||||
(container, config)
|
||||
|
||||
@@ -29,6 +29,7 @@ async fn create_config() -> (ContainerAsync<Mongo>, DatabaseConfig) {
|
||||
host: host.to_string(),
|
||||
generated_id: "96d30a9f-ff4b-47c9-aaab-f3147bb34f16".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "".to_string(),
|
||||
};
|
||||
|
||||
(container, config)
|
||||
|
||||
@@ -54,6 +54,7 @@ fn make_config(host: String, port: u16, database: &str, generated_id: &str) -> D
|
||||
host,
|
||||
generated_id: generated_id.to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "".to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ async fn create_config() -> (ContainerAsync<Mysql>, DatabaseConfig) {
|
||||
host: host.to_string(),
|
||||
generated_id: "0f1bb8f2-35a0-4c91-8098-e36873d3ce31".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "512M".to_string(),
|
||||
};
|
||||
|
||||
(container, config)
|
||||
|
||||
@@ -38,6 +38,7 @@ async fn create_config() -> (ContainerAsync<Postgres>, DatabaseConfig) {
|
||||
host: host.to_string(),
|
||||
generated_id: "40875631-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "".to_string(),
|
||||
};
|
||||
|
||||
(container, config)
|
||||
@@ -107,3 +108,105 @@ async fn postgres_backup_restore_test() {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn postgres_password_with_slash_test() {
|
||||
init_tracing_for_test();
|
||||
|
||||
let special_password = "ch/ange:me@1";
|
||||
|
||||
let container = Postgres::default()
|
||||
.with_env_var("POSTGRES_DB", "testdb")
|
||||
.with_env_var("POSTGRES_USER", "testuser")
|
||||
.with_env_var("POSTGRES_PASSWORD", special_password)
|
||||
.with_tag("17")
|
||||
.start()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let host = container
|
||||
.get_host()
|
||||
.await
|
||||
.unwrap_or(Host::parse("127.0.0.1").unwrap());
|
||||
|
||||
let port = container.get_host_port_ipv4(5432).await.unwrap_or(5432);
|
||||
|
||||
let config = DatabaseConfig {
|
||||
name: "My test Postgres Database with slash password".to_string(),
|
||||
database: "testdb".to_string(),
|
||||
db_type: DbType::Postgresql,
|
||||
username: "testuser".to_string(),
|
||||
password: special_password.to_string(),
|
||||
port,
|
||||
host: host.to_string(),
|
||||
generated_id: "5a1f0e3c-9b8a-4a8e-9b1b-0a1c2d3e4f5a".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "".to_string(),
|
||||
};
|
||||
|
||||
let db = DatabaseFactory::create_for_backup(config.clone()).await;
|
||||
let reachable = db.ping().await.unwrap_or(false);
|
||||
|
||||
assert_eq!(reachable, true);
|
||||
}
|
||||
|
||||
mod select_pg_path_tests {
|
||||
use crate::domain::postgres::connection::{
|
||||
pg_dump_binary_name, pg_dump_exists_in, select_pg_path_with,
|
||||
};
|
||||
|
||||
// `select_pg_path_with` takes the `PG_BIN_DIR` override as a plain
|
||||
// argument, so these tests never touch process-global env state or the
|
||||
// cached `CONFIG`. They stay deterministic regardless of whether — or at
|
||||
// which version — a real PostgreSQL install exists on the host.
|
||||
|
||||
#[test]
|
||||
fn respects_pg_bin_dir_override() {
|
||||
let custom = if cfg!(target_os = "windows") {
|
||||
r"C:\custom\pg\bin"
|
||||
} else {
|
||||
"/custom/pg/bin"
|
||||
};
|
||||
let path = select_pg_path_with("16.4", custom);
|
||||
assert_eq!(path, std::path::PathBuf::from(custom));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pg_bin_dir_override_ignores_requested_version() {
|
||||
// The override is taken as-is, regardless of which version was
|
||||
// requested — this documents/locks in that behavior.
|
||||
let custom = if cfg!(target_os = "windows") {
|
||||
r"C:\custom\pg\bin"
|
||||
} else {
|
||||
"/custom/pg/bin"
|
||||
};
|
||||
let path = select_pg_path_with("not-a-version", custom);
|
||||
assert_eq!(path, std::path::PathBuf::from(custom));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_pg_bin_dir_falls_through_to_detection() {
|
||||
// An empty override means "unset" (matches `CONFIG.pg_bin_dir` when
|
||||
// `PG_BIN_DIR` is absent). It must not be returned as a literal empty
|
||||
// path — resolution falls through to platform defaults / PATH lookup
|
||||
// and yields a non-empty path.
|
||||
let path = select_pg_path_with("17", "");
|
||||
assert_ne!(path, std::path::PathBuf::from(""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pg_dump_binary_name_is_platform_specific() {
|
||||
let name = pg_dump_binary_name();
|
||||
if cfg!(target_os = "windows") {
|
||||
assert_eq!(name, "pg_dump.exe");
|
||||
} else {
|
||||
assert_eq!(name, "pg_dump");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pg_dump_exists_in_is_false_for_nonexistent_dir() {
|
||||
let dir = std::path::Path::new("this/path/almost-certainly/does-not-exist-12345");
|
||||
assert!(!pg_dump_exists_in(dir));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -28,6 +28,7 @@ async fn create_config() -> (ContainerAsync<Redis>, DatabaseConfig) {
|
||||
host: host.to_string(),
|
||||
generated_id: "40875631-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "".to_string(),
|
||||
};
|
||||
|
||||
(container, config)
|
||||
|
||||
@@ -27,6 +27,7 @@ async fn create_config() -> (ContainerAsync<Valkey>, DatabaseConfig) {
|
||||
host: host.to_string(),
|
||||
generated_id: "40875485-e3d2-4dfe-a26b-2a347ecc64fd".to_string(),
|
||||
path: "".to_string(),
|
||||
max_packet_size: "".to_string(),
|
||||
};
|
||||
|
||||
(container, config)
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
mod domain;
|
||||
mod services;
|
||||
mod storage;
|
||||
mod utils;
|
||||
|
||||
use once_cell::sync::Lazy;
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
//! Regression test: a per-storage upload failure must be reported to the server via
|
||||
//! `backup_upload_status("failed", ...)`. Previously the uploader early-returned on failure
|
||||
//! and skipped the status call, so `backup_upload_init` opened a record that was never closed.
|
||||
|
||||
use crate::core::context::Context;
|
||||
use crate::services::api::ApiClient;
|
||||
use crate::services::api::models::agent::status::DatabaseStorage;
|
||||
use crate::services::backup::BackupService;
|
||||
use crate::services::backup::logger::JobLogger;
|
||||
use crate::services::backup::models::BackupResult;
|
||||
use crate::services::config::DbType;
|
||||
use crate::tests::init_tracing_for_test;
|
||||
use crate::utils::common::BackupMethod;
|
||||
use crate::utils::edge_key::EdgeKey;
|
||||
|
||||
use serde_json::json;
|
||||
use std::sync::Arc;
|
||||
use wiremock::matchers::{body_partial_json, method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
fn ctx_pointing_at(base_url: String) -> Context {
|
||||
Context {
|
||||
edge_key: EdgeKey {
|
||||
server_url: String::new(),
|
||||
agent_id: "agent-1".to_string(),
|
||||
master_key_b64: String::new(),
|
||||
},
|
||||
api: ApiClient::new(base_url),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn failed_upload_reports_failed_status_to_server() {
|
||||
init_tracing_for_test();
|
||||
let server = MockServer::start().await;
|
||||
|
||||
// init opens the per-storage record and returns its id.
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/agent/agent-1/backup/upload/init"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"message": "ok",
|
||||
"backupStorage": { "id": "bs-1" }
|
||||
})))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
// The fix: on failure the uploader must PATCH the status as "failed".
|
||||
Mock::given(method("PATCH"))
|
||||
.and(path("/agent/agent-1/backup/upload/status"))
|
||||
.and(body_partial_json(json!({ "status": "failed" })))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({})))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let service = BackupService::new(Arc::new(ctx_pointing_at(server.uri())));
|
||||
|
||||
// backup_file = None makes the provider fail immediately ("Missing backup file path"),
|
||||
// exercising the failure path without any network/Azure dependency.
|
||||
let result = BackupResult {
|
||||
generated_id: "gen-1".to_string(),
|
||||
db_type: DbType::Postgresql,
|
||||
status: "success".to_string(),
|
||||
backup_file: None,
|
||||
code: None,
|
||||
};
|
||||
|
||||
let storage: DatabaseStorage = serde_json::from_value(json!({
|
||||
"id": "storage-1",
|
||||
"provider": "blob",
|
||||
"config": {}
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
let backup_id = "backup-1".to_string();
|
||||
let logger = Arc::new(JobLogger::new());
|
||||
|
||||
let results = service
|
||||
.upload(
|
||||
result,
|
||||
BackupMethod::Manual,
|
||||
vec![storage],
|
||||
false,
|
||||
&backup_id,
|
||||
logger,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(results.len(), 1);
|
||||
assert!(!results[0].success);
|
||||
|
||||
// MockServer drop verifies both `.expect(1)` mounts were hit — including the "failed" PATCH.
|
||||
}
|
||||
@@ -1 +1,2 @@
|
||||
mod api_models_tests;
|
||||
mod backup_uploader_tests;
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
use crate::services::storage::providers::azure_blob::helpers::{
|
||||
ResolvedAzure, SAS_VERSION, SasResource, build_sas_url, hmac_sha256_b64,
|
||||
};
|
||||
use crate::tests::init_tracing_for_test;
|
||||
|
||||
use anyhow::{Context as _, anyhow};
|
||||
use azure_core::http::RequestContent;
|
||||
use azure_storage_blob::clients::BlobClient;
|
||||
use azure_storage_blob::models::BlockLookupList;
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use bytes::Bytes;
|
||||
use chrono::{Duration, Utc};
|
||||
use testcontainers::core::{IntoContainerPort, WaitFor};
|
||||
use testcontainers::runners::AsyncRunner;
|
||||
use testcontainers::{GenericImage, ImageExt};
|
||||
use url::Url;
|
||||
|
||||
/// Build an Account SAS query set (test-only). Azurite cannot authorize container-create with
|
||||
/// a container-scoped Service SAS, so tests create the target container with an Account SAS.
|
||||
/// Reuses the production HMAC primitive (`hmac_sha256_b64`) to avoid duplicating signing logic.
|
||||
fn build_account_sas(
|
||||
resolved: &ResolvedAzure,
|
||||
services: &str,
|
||||
resource_types: &str,
|
||||
permissions: &str,
|
||||
) -> anyhow::Result<Vec<(String, String)>> {
|
||||
let key = STANDARD
|
||||
.decode(&resolved.account_key)
|
||||
.map_err(|_| anyhow!("account key is not valid base64"))?;
|
||||
|
||||
let signed_start = String::new();
|
||||
let signed_expiry = (Utc::now() + Duration::hours(1))
|
||||
.format("%Y-%m-%dT%H:%M:%SZ")
|
||||
.to_string();
|
||||
let signed_protocol = "https,http"; // Azurite is http
|
||||
let signed_ip = String::new();
|
||||
let encryption_scope = String::new();
|
||||
|
||||
// Account SAS string-to-sign for sv >= 2020-12-06:
|
||||
// account \n sp \n ss \n srt \n st \n se \n sip \n spr \n sv \n ses \n (trailing newline)
|
||||
let string_to_sign = format!(
|
||||
"{acc}\n{sp}\n{ss}\n{srt}\n{st}\n{se}\n{sip}\n{spr}\n{sv}\n{ses}\n",
|
||||
acc = resolved.account_name, sp = permissions, ss = services, srt = resource_types,
|
||||
st = signed_start, se = signed_expiry, sip = signed_ip, spr = signed_protocol,
|
||||
sv = SAS_VERSION, ses = encryption_scope,
|
||||
);
|
||||
|
||||
let sig = hmac_sha256_b64(&key, &string_to_sign)?;
|
||||
|
||||
Ok(vec![
|
||||
("sv".into(), SAS_VERSION.into()),
|
||||
("ss".into(), services.into()),
|
||||
("srt".into(), resource_types.into()),
|
||||
("sp".into(), permissions.into()),
|
||||
("se".into(), signed_expiry),
|
||||
("spr".into(), signed_protocol.into()),
|
||||
("sig".into(), sig),
|
||||
])
|
||||
}
|
||||
|
||||
/// Build an Account-SAS-scoped URL for a container (test-only container creation).
|
||||
fn build_account_sas_container_url(
|
||||
resolved: &ResolvedAzure,
|
||||
container: &str,
|
||||
services: &str,
|
||||
resource_types: &str,
|
||||
permissions: &str,
|
||||
) -> anyhow::Result<Url> {
|
||||
let pairs = build_account_sas(resolved, services, resource_types, permissions)?;
|
||||
let base = format!("{}/{}", resolved.blob_endpoint.trim_end_matches('/'), container);
|
||||
let mut url = Url::parse(&base).context("invalid blob endpoint/url")?;
|
||||
{
|
||||
let mut qp = url.query_pairs_mut();
|
||||
for (k, v) in pairs {
|
||||
qp.append_pair(&k, &v);
|
||||
}
|
||||
}
|
||||
Ok(url)
|
||||
}
|
||||
|
||||
const AZURITE_ACCOUNT: &str = "devstoreaccount1";
|
||||
const AZURITE_KEY: &str =
|
||||
"Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==";
|
||||
|
||||
async fn start_azurite() -> (testcontainers::ContainerAsync<GenericImage>, ResolvedAzure) {
|
||||
let container = GenericImage::new("mcr.microsoft.com/azure-storage/azurite", "latest")
|
||||
.with_exposed_port(10000.tcp())
|
||||
// The current `latest` image logs (on stdout):
|
||||
// "Azurite Blob service successfully listens on http://0.0.0.0:10000"
|
||||
// Older builds phrased it "...is successfully listening"; this substring matches
|
||||
// the wording the pulled image actually emits.
|
||||
.with_wait_for(WaitFor::message_on_stdout(
|
||||
"Azurite Blob service successfully listens on",
|
||||
))
|
||||
// The GA SDK sends a very recent `x-ms-version`; Azurite 3.35 rejects unknown
|
||||
// versions unless we tell it to skip that check.
|
||||
.with_cmd(["azurite-blob", "--blobHost", "0.0.0.0", "--skipApiVersionCheck"])
|
||||
.start().await.unwrap();
|
||||
// Use the testcontainers-resolved host (not a hardcoded 127.0.0.1): under
|
||||
// docker-out-of-docker / remote daemons the published port is not on the test
|
||||
// process's loopback. All other container tests (mssql, valkey, postgres, ...)
|
||||
// already do this; azure_blob was the only one hardcoding the host.
|
||||
let host = container.get_host().await.unwrap().to_string();
|
||||
let port = container.get_host_port_ipv4(10000).await.unwrap();
|
||||
let resolved = ResolvedAzure {
|
||||
account_name: AZURITE_ACCOUNT.to_string(),
|
||||
account_key: AZURITE_KEY.to_string(),
|
||||
blob_endpoint: format!("http://{host}:{port}/{AZURITE_ACCOUNT}"),
|
||||
};
|
||||
(container, resolved)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn spike_sas_block_roundtrip_against_azurite() {
|
||||
init_tracing_for_test();
|
||||
let (_container, resolved) = start_azurite().await;
|
||||
let container = "portabase";
|
||||
let blob = "spike/hello.txt";
|
||||
|
||||
// Container creation must use an Account SAS (service=blob, resource-type=container,
|
||||
// perms=create+write). Azurite cannot authorize container-create with a Service SAS.
|
||||
let container_url =
|
||||
build_account_sas_container_url(&resolved, container, "b", "c", "cw").unwrap();
|
||||
let container_client =
|
||||
azure_storage_blob::clients::BlobContainerClient::new(container_url, None, None).unwrap();
|
||||
container_client.create(None).await.unwrap();
|
||||
|
||||
let blob_url = build_sas_url(&resolved, container, blob, SasResource::Blob, "cw").unwrap();
|
||||
let blob_client = BlobClient::new(blob_url.clone(), None, None).unwrap();
|
||||
let bbc = blob_client.block_blob_client();
|
||||
|
||||
let payload = Bytes::from_static(b"hello azurite");
|
||||
let raw_id = format!("{:032}", 0u32).into_bytes();
|
||||
bbc.stage_block(&raw_id, payload.len() as u64, RequestContent::from(payload.to_vec()), None)
|
||||
.await.unwrap();
|
||||
|
||||
// `BlockLookupList.latest` is `Option<Vec<Vec<u8>>>` and base64-encodes each entry
|
||||
// internally during XML serialization, exactly as `stage_block` base64-encodes the
|
||||
// `blockid` query. So `latest` must hold the SAME RAW id bytes passed to `stage_block`.
|
||||
let block_list = BlockLookupList { latest: Some(vec![raw_id.clone()]), ..Default::default() };
|
||||
bbc.commit_block_list(block_list.try_into().unwrap(), None).await.unwrap();
|
||||
|
||||
let read_url = build_sas_url(&resolved, container, blob, SasResource::Blob, "r").unwrap();
|
||||
let read_client = BlobClient::new(read_url, None, None).unwrap();
|
||||
assert!(read_client.exists().await.unwrap());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn upload_stream_multi_block_roundtrip() {
|
||||
init_tracing_for_test();
|
||||
use crate::services::storage::providers::azure_blob::helpers::upload_stream_to_azure;
|
||||
use futures::stream;
|
||||
|
||||
let (_container, resolved) = start_azurite().await;
|
||||
let container = "portabase";
|
||||
let blob = "backups/multi.bin";
|
||||
|
||||
// Container setup (provider itself never creates it): Account SAS create.
|
||||
let container_url =
|
||||
build_account_sas_container_url(&resolved, container, "b", "c", "cw").unwrap();
|
||||
azure_storage_blob::clients::BlobContainerClient::new(container_url, None, None)
|
||||
.unwrap()
|
||||
.create(None)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// 10 KiB fed as 1 KiB chunks, forced into 4 KiB blocks => 3 blocks (multi-block path).
|
||||
let data = vec![7u8; 10 * 1024];
|
||||
let chunks: Vec<Result<Bytes, std::io::Error>> = data
|
||||
.chunks(1024)
|
||||
.map(|c| Ok(Bytes::copy_from_slice(c)))
|
||||
.collect();
|
||||
let body = Box::pin(stream::iter(chunks));
|
||||
|
||||
upload_stream_to_azure(&resolved, container, blob, body, 4 * 1024)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Verify the committed blob reassembles to the exact source bytes via a read-SAS GET.
|
||||
let read_url = build_sas_url(&resolved, container, blob, SasResource::Blob, "r").unwrap();
|
||||
let got = reqwest::get(read_url).await.unwrap().bytes().await.unwrap();
|
||||
assert_eq!(got.as_ref(), data.as_slice());
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
mod azure_blob;
|
||||
Reference in New Issue
Block a user