mirror of
https://github.com/deuxfleurs-org/garage.git
synced 2026-08-14 16:13:11 +00:00
Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Alex Auvolat
|
def78c5e6f | ||
|
Alex
|
277a20ec44 | ||
|
Alex Auvolat
|
c9ef3e461b | ||
|
Alex Auvolat
|
c93008d333 | ||
|
Alex Auvolat
|
e5341ca47b | ||
|
withinboredom
|
a4f9f19ac3 | ||
|
Baptiste Jonglez
|
47e57518ec |
Generated
+6
-6
@@ -840,7 +840,7 @@ dependencies = [
|
||||
"http",
|
||||
"hyper",
|
||||
"kuska-sodiumoxide",
|
||||
"netapp 0.4.2",
|
||||
"netapp 0.4.4",
|
||||
"pretty_env_logger",
|
||||
"rand 0.8.5",
|
||||
"rmp-serde 0.15.5",
|
||||
@@ -975,7 +975,7 @@ dependencies = [
|
||||
"garage_table 0.7.0",
|
||||
"garage_util 0.7.0",
|
||||
"hex",
|
||||
"netapp 0.4.2",
|
||||
"netapp 0.4.4",
|
||||
"opentelemetry",
|
||||
"rand 0.8.5",
|
||||
"rmp-serde 0.15.5",
|
||||
@@ -1031,7 +1031,7 @@ dependencies = [
|
||||
"k8s-openapi",
|
||||
"kube",
|
||||
"kuska-sodiumoxide",
|
||||
"netapp 0.4.2",
|
||||
"netapp 0.4.4",
|
||||
"openssl",
|
||||
"opentelemetry",
|
||||
"pnet_datalink",
|
||||
@@ -1126,7 +1126,7 @@ dependencies = [
|
||||
"hex",
|
||||
"http",
|
||||
"hyper",
|
||||
"netapp 0.4.2",
|
||||
"netapp 0.4.4",
|
||||
"opentelemetry",
|
||||
"rand 0.8.5",
|
||||
"rmp-serde 0.15.5",
|
||||
@@ -1833,9 +1833,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "netapp"
|
||||
version = "0.4.2"
|
||||
version = "0.4.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d1a19af9ad24e6cdb166e2c5882a7ff9326cab8d45c9d82379fa638c0cbc84df"
|
||||
checksum = "c6419a4b836774192e13fedb05c0e5f414ee8df9ca0c467456a0bacde06c29ee"
|
||||
dependencies = [
|
||||
"arc-swap",
|
||||
"async-trait",
|
||||
|
||||
@@ -619,7 +619,7 @@ in
|
||||
registry = "registry+https://github.com/rust-lang/crates.io-index";
|
||||
src = fetchCratesIo { inherit name version; sha256 = "59a6001667ab124aebae2a495118e11d30984c3a653e99d86d58971708cf5e4b"; };
|
||||
dependencies = {
|
||||
${ if hostPlatform.config == "aarch64-apple-darwin" || hostPlatform.config == "aarch64-linux-android" || hostPlatform.parsed.cpu.name == "aarch64" && hostPlatform.parsed.kernel.name == "linux" then "libc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".libc."0.2.121" { inherit profileName; };
|
||||
${ if hostPlatform.config == "aarch64-linux-android" || hostPlatform.parsed.cpu.name == "aarch64" && hostPlatform.parsed.kernel.name == "linux" || hostPlatform.config == "aarch64-apple-darwin" then "libc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".libc."0.2.121" { inherit profileName; };
|
||||
};
|
||||
});
|
||||
|
||||
@@ -1193,7 +1193,7 @@ in
|
||||
git_version = rustPackages."registry+https://github.com/rust-lang/crates.io-index".git-version."0.3.5" { inherit profileName; };
|
||||
hex = rustPackages."registry+https://github.com/rust-lang/crates.io-index".hex."0.4.3" { inherit profileName; };
|
||||
sodiumoxide = rustPackages."registry+https://github.com/rust-lang/crates.io-index".kuska-sodiumoxide."0.2.5-0" { inherit profileName; };
|
||||
netapp = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.2" { inherit profileName; };
|
||||
netapp = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.4" { inherit profileName; };
|
||||
pretty_env_logger = rustPackages."registry+https://github.com/rust-lang/crates.io-index".pretty_env_logger."0.4.0" { inherit profileName; };
|
||||
rand = rustPackages."registry+https://github.com/rust-lang/crates.io-index".rand."0.8.5" { inherit profileName; };
|
||||
rmp_serde = rustPackages."registry+https://github.com/rust-lang/crates.io-index".rmp-serde."0.15.5" { inherit profileName; };
|
||||
@@ -1348,7 +1348,7 @@ in
|
||||
garage_table = rustPackages."unknown".garage_table."0.7.0" { inherit profileName; };
|
||||
garage_util = rustPackages."unknown".garage_util."0.7.0" { inherit profileName; };
|
||||
hex = rustPackages."registry+https://github.com/rust-lang/crates.io-index".hex."0.4.3" { inherit profileName; };
|
||||
netapp = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.2" { inherit profileName; };
|
||||
netapp = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.4" { inherit profileName; };
|
||||
opentelemetry = rustPackages."registry+https://github.com/rust-lang/crates.io-index".opentelemetry."0.17.0" { inherit profileName; };
|
||||
rand = rustPackages."registry+https://github.com/rust-lang/crates.io-index".rand."0.8.5" { inherit profileName; };
|
||||
rmp_serde = rustPackages."registry+https://github.com/rust-lang/crates.io-index".rmp-serde."0.15.5" { inherit profileName; };
|
||||
@@ -1415,7 +1415,7 @@ in
|
||||
${ if rootFeatures' ? "garage_rpc" then "k8s_openapi" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".k8s-openapi."0.13.1" { inherit profileName; };
|
||||
${ if rootFeatures' ? "garage_rpc" then "kube" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".kube."0.62.0" { inherit profileName; };
|
||||
${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "sodiumoxide" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".kuska-sodiumoxide."0.2.5-0" { inherit profileName; };
|
||||
${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "netapp" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.2" { inherit profileName; };
|
||||
${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "netapp" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.4" { inherit profileName; };
|
||||
${ if rootFeatures' ? "garage_rpc" then "openssl" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".openssl."0.10.38" { inherit profileName; };
|
||||
${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "opentelemetry" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".opentelemetry."0.17.0" { inherit profileName; };
|
||||
${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "pnet_datalink" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".pnet_datalink."0.28.0" { inherit profileName; };
|
||||
@@ -1518,7 +1518,7 @@ in
|
||||
hex = rustPackages."registry+https://github.com/rust-lang/crates.io-index".hex."0.4.3" { inherit profileName; };
|
||||
http = rustPackages."registry+https://github.com/rust-lang/crates.io-index".http."0.2.6" { inherit profileName; };
|
||||
hyper = rustPackages."registry+https://github.com/rust-lang/crates.io-index".hyper."0.14.18" { inherit profileName; };
|
||||
netapp = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.2" { inherit profileName; };
|
||||
netapp = rustPackages."registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.4" { inherit profileName; };
|
||||
opentelemetry = rustPackages."registry+https://github.com/rust-lang/crates.io-index".opentelemetry."0.17.0" { inherit profileName; };
|
||||
rand = rustPackages."registry+https://github.com/rust-lang/crates.io-index".rand."0.8.5" { inherit profileName; };
|
||||
rmp_serde = rustPackages."registry+https://github.com/rust-lang/crates.io-index".rmp-serde."0.15.5" { inherit profileName; };
|
||||
@@ -2456,11 +2456,11 @@ in
|
||||
};
|
||||
});
|
||||
|
||||
"registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.2" = overridableMkRustCrate (profileName: rec {
|
||||
"registry+https://github.com/rust-lang/crates.io-index".netapp."0.4.4" = overridableMkRustCrate (profileName: rec {
|
||||
name = "netapp";
|
||||
version = "0.4.2";
|
||||
version = "0.4.4";
|
||||
registry = "registry+https://github.com/rust-lang/crates.io-index";
|
||||
src = fetchCratesIo { inherit name version; sha256 = "d1a19af9ad24e6cdb166e2c5882a7ff9326cab8d45c9d82379fa638c0cbc84df"; };
|
||||
src = fetchCratesIo { inherit name version; sha256 = "c6419a4b836774192e13fedb05c0e5f414ee8df9ca0c467456a0bacde06c29ee"; };
|
||||
features = builtins.concatLists [
|
||||
[ "default" ]
|
||||
(lib.optional (rootFeatures' ? "garage" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web") "opentelemetry")
|
||||
@@ -3342,7 +3342,7 @@ in
|
||||
];
|
||||
dependencies = {
|
||||
${ if hostPlatform.parsed.kernel.name == "android" || hostPlatform.parsed.kernel.name == "linux" then "libc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".libc."0.2.121" { inherit profileName; };
|
||||
${ if hostPlatform.parsed.kernel.name == "dragonfly" || hostPlatform.parsed.kernel.name == "freebsd" || hostPlatform.parsed.kernel.name == "illumos" || hostPlatform.parsed.kernel.name == "netbsd" || hostPlatform.parsed.kernel.name == "openbsd" || hostPlatform.parsed.kernel.name == "solaris" || hostPlatform.parsed.kernel.name == "android" || hostPlatform.parsed.kernel.name == "linux" then "once_cell" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".once_cell."1.10.0" { inherit profileName; };
|
||||
${ if hostPlatform.parsed.kernel.name == "android" || hostPlatform.parsed.kernel.name == "linux" || hostPlatform.parsed.kernel.name == "dragonfly" || hostPlatform.parsed.kernel.name == "freebsd" || hostPlatform.parsed.kernel.name == "illumos" || hostPlatform.parsed.kernel.name == "netbsd" || hostPlatform.parsed.kernel.name == "openbsd" || hostPlatform.parsed.kernel.name == "solaris" then "once_cell" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".once_cell."1.10.0" { inherit profileName; };
|
||||
${ if hostPlatform.parsed.cpu.name == "i686" || hostPlatform.parsed.cpu.name == "x86_64" || (hostPlatform.parsed.cpu.name == "aarch64" || hostPlatform.parsed.cpu.name == "armv6l" || hostPlatform.parsed.cpu.name == "armv7l") && (hostPlatform.parsed.kernel.name == "android" || hostPlatform.parsed.kernel.name == "fuchsia" || hostPlatform.parsed.kernel.name == "linux") then "spin" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".spin."0.5.2" { inherit profileName; };
|
||||
untrusted = rustPackages."registry+https://github.com/rust-lang/crates.io-index".untrusted."0.7.1" { inherit profileName; };
|
||||
${ if hostPlatform.parsed.cpu.name == "wasm32" && hostPlatform.parsed.vendor.name == "unknown" && hostPlatform.parsed.kernel.name == "unknown" && hostPlatform.parsed.abi.name == "" then "web_sys" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".web-sys."0.3.56" { inherit profileName; };
|
||||
@@ -3819,7 +3819,7 @@ in
|
||||
];
|
||||
dependencies = {
|
||||
bitflags = rustPackages."registry+https://github.com/rust-lang/crates.io-index".bitflags."1.3.2" { inherit profileName; };
|
||||
${ if hostPlatform.parsed.kernel.name == "android" || hostPlatform.parsed.kernel.name == "linux" then "libc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".libc."0.2.121" { inherit profileName; };
|
||||
${ if hostPlatform.parsed.kernel.name == "linux" || hostPlatform.parsed.kernel.name == "android" then "libc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".libc."0.2.121" { inherit profileName; };
|
||||
${ if !(hostPlatform.parsed.kernel.name == "linux" || hostPlatform.parsed.kernel.name == "android") then "parking_lot" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".parking_lot."0.11.2" { inherit profileName; };
|
||||
${ if !(hostPlatform.parsed.kernel.name == "linux" || hostPlatform.parsed.kernel.name == "android") then "parking_lot_core" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".parking_lot_core."0.8.5" { inherit profileName; };
|
||||
static_init_macro = buildRustPackages."registry+https://github.com/rust-lang/crates.io-index".static_init_macro."1.0.2" { profileName = "__noProfile"; };
|
||||
@@ -4789,11 +4789,11 @@ in
|
||||
[ "default" ]
|
||||
];
|
||||
dependencies = {
|
||||
${ if hostPlatform.config == "aarch64-uwp-windows-msvc" || hostPlatform.config == "aarch64-pc-windows-msvc" then "windows_aarch64_msvc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".windows_aarch64_msvc."0.32.0" { inherit profileName; };
|
||||
${ if hostPlatform.config == "aarch64-pc-windows-msvc" || hostPlatform.config == "aarch64-uwp-windows-msvc" then "windows_aarch64_msvc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".windows_aarch64_msvc."0.32.0" { inherit profileName; };
|
||||
${ if hostPlatform.config == "i686-uwp-windows-gnu" || hostPlatform.config == "i686-pc-windows-gnu" then "windows_i686_gnu" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".windows_i686_gnu."0.32.0" { inherit profileName; };
|
||||
${ if hostPlatform.config == "i686-pc-windows-msvc" || hostPlatform.config == "i686-uwp-windows-msvc" then "windows_i686_msvc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".windows_i686_msvc."0.32.0" { inherit profileName; };
|
||||
${ if hostPlatform.config == "x86_64-uwp-windows-gnu" || hostPlatform.config == "x86_64-pc-windows-gnu" then "windows_x86_64_gnu" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".windows_x86_64_gnu."0.32.0" { inherit profileName; };
|
||||
${ if hostPlatform.config == "x86_64-pc-windows-msvc" || hostPlatform.config == "x86_64-uwp-windows-msvc" then "windows_x86_64_msvc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".windows_x86_64_msvc."0.32.0" { inherit profileName; };
|
||||
${ if hostPlatform.config == "x86_64-uwp-windows-msvc" || hostPlatform.config == "x86_64-pc-windows-msvc" then "windows_x86_64_msvc" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".windows_x86_64_msvc."0.32.0" { inherit profileName; };
|
||||
};
|
||||
});
|
||||
|
||||
|
||||
@@ -50,3 +50,24 @@ Click `Test connection` and then no when asked `The bucket name should start wit
|
||||
|
||||
*External links:* [Kopia Documentation > Repositories](https://kopia.io/docs/repositories/#amazon-s3)
|
||||
|
||||
To create the Kopia repository, you need to specify the region, the HTTP(S) endpoint, the bucket name and the access keys.
|
||||
For instance, if you have an instance of garage running on `https://garage.example.com`:
|
||||
|
||||
```
|
||||
kopia repository create s3 --region=garage --bucket=mybackups --access-key=KEY_ID --secret-access-key=SECRET_KEY --endpoint=garage.example.com
|
||||
```
|
||||
|
||||
Or if you have an instance running on localhost, without TLS:
|
||||
|
||||
```
|
||||
kopia repository create s3 --region=garage --bucket=mybackups --access-key=KEY_ID --secret-access-key=SECRET_KEY --endpoint=localhost:3900 --disable-tls
|
||||
```
|
||||
|
||||
After the repository has been created, check that everything works as expected:
|
||||
|
||||
```
|
||||
kopia repository validate-provider
|
||||
```
|
||||
|
||||
You can then run all the standard kopia commands: `kopia snapshot create`, `kopia mount`...
|
||||
Everything should work out-of-the-box.
|
||||
|
||||
+99
-54
@@ -1,7 +1,8 @@
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use futures::TryFutureExt;
|
||||
use futures::{stream, stream::Stream, StreamExt, TryFutureExt};
|
||||
use md5::{Digest as Md5Digest, Md5};
|
||||
|
||||
use hyper::{Body, Request, Response};
|
||||
@@ -268,7 +269,6 @@ pub async fn handle_upload_part_copy(
|
||||
|
||||
let mut blocks_to_copy = vec![];
|
||||
let mut current_offset = 0;
|
||||
let mut size_to_copy = 0;
|
||||
for (_bk, block) in source_version.blocks.items().iter() {
|
||||
let (block_begin, block_end) = (current_offset, current_offset + block.size);
|
||||
|
||||
@@ -289,10 +289,6 @@ pub async fn handle_upload_part_copy(
|
||||
(Some(b), None) => Some(b as usize..block.size as usize),
|
||||
(None, None) => None,
|
||||
};
|
||||
size_to_copy += range_to_copy
|
||||
.as_ref()
|
||||
.map(|x| x.len() as u64)
|
||||
.unwrap_or(block.size);
|
||||
|
||||
blocks_to_copy.push((block.hash, range_to_copy));
|
||||
}
|
||||
@@ -300,34 +296,49 @@ pub async fn handle_upload_part_copy(
|
||||
current_offset = block_end;
|
||||
}
|
||||
|
||||
if size_to_copy < 1024 * 1024 {
|
||||
return Err(Error::BadRequest(format!(
|
||||
"Not enough data to copy: {} bytes (minimum: 1MB)",
|
||||
size_to_copy
|
||||
)));
|
||||
}
|
||||
|
||||
// Now, actually copy the blocks
|
||||
let mut md5hasher = Md5::new();
|
||||
|
||||
let mut block = Some(
|
||||
garage
|
||||
.block_manager
|
||||
.rpc_get_block(&blocks_to_copy[0].0)
|
||||
.await?,
|
||||
);
|
||||
// First, create a stream that is able to read the source blocks
|
||||
// and extract the subrange if necessary.
|
||||
// The second returned value is an Option<Hash>, that is Some
|
||||
// if and only if the block returned is a block that already existed
|
||||
// in the Garage data store (thus we don't need to save it again).
|
||||
let garage2 = garage.clone();
|
||||
let source_blocks = stream::iter(blocks_to_copy)
|
||||
.flat_map(|(block_hash, range_to_copy)| {
|
||||
let garage3 = garage2.clone();
|
||||
stream::once(async move {
|
||||
let data = garage3.block_manager.rpc_get_block(&block_hash).await?;
|
||||
match range_to_copy {
|
||||
Some(r) => Ok((data[r].to_vec(), None)),
|
||||
None => Ok((data, Some(block_hash))),
|
||||
}
|
||||
})
|
||||
})
|
||||
.peekable();
|
||||
|
||||
// The defragmenter is a custom stream (defined below) that concatenates
|
||||
// consecutive block parts when they are too small.
|
||||
// It returns a series of (Vec<u8>, Option<Hash>).
|
||||
// When it is done, it returns an empty vec.
|
||||
// Same as the previous iterator, the Option is Some(_) if and only if
|
||||
// it's an existing block of the Garage data store.
|
||||
let mut defragmenter = Defragmenter::new(garage.config.block_size, Box::pin(source_blocks));
|
||||
|
||||
let mut current_offset = 0;
|
||||
for (i, (block_hash, range_to_copy)) in blocks_to_copy.iter().enumerate() {
|
||||
let (current_block, subrange_hash) = match range_to_copy.clone() {
|
||||
Some(r) => {
|
||||
let subrange = block.take().unwrap()[r].to_vec();
|
||||
let hash = blake2sum(&subrange);
|
||||
(subrange, hash)
|
||||
}
|
||||
None => (block.take().unwrap(), *block_hash),
|
||||
};
|
||||
md5hasher.update(¤t_block[..]);
|
||||
let mut next_block = defragmenter.next().await?;
|
||||
|
||||
loop {
|
||||
let (data, existing_block_hash) = next_block;
|
||||
if data.is_empty() {
|
||||
break;
|
||||
}
|
||||
|
||||
md5hasher.update(&data[..]);
|
||||
|
||||
let must_upload = existing_block_hash.is_none();
|
||||
let final_hash = existing_block_hash.unwrap_or_else(|| blake2sum(&data[..]));
|
||||
|
||||
let mut version = Version::new(dest_version_uuid, dest_bucket_id, dest_key.clone(), false);
|
||||
version.blocks.put(
|
||||
@@ -336,33 +347,25 @@ pub async fn handle_upload_part_copy(
|
||||
offset: current_offset,
|
||||
},
|
||||
VersionBlock {
|
||||
hash: subrange_hash,
|
||||
size: current_block.len() as u64,
|
||||
hash: final_hash,
|
||||
size: data.len() as u64,
|
||||
},
|
||||
);
|
||||
current_offset += current_block.len() as u64;
|
||||
current_offset += data.len() as u64;
|
||||
|
||||
let block_ref = BlockRef {
|
||||
block: subrange_hash,
|
||||
block: final_hash,
|
||||
version: dest_version_uuid,
|
||||
deleted: false.into(),
|
||||
};
|
||||
|
||||
let next_block_hash = blocks_to_copy.get(i + 1).map(|(h, _)| *h);
|
||||
|
||||
let garage2 = garage.clone();
|
||||
let garage3 = garage.clone();
|
||||
let is_subrange = range_to_copy.is_some();
|
||||
|
||||
let (_, _, _, next_block) = futures::try_join!(
|
||||
// Thing 1: if we are taking a subrange of the source block,
|
||||
// we need to insert that subrange as a new block.
|
||||
let res = futures::try_join!(
|
||||
// Thing 1: if the block is not exactly a block that existed before,
|
||||
// we need to insert that data as a new block.
|
||||
async move {
|
||||
if is_subrange {
|
||||
garage2
|
||||
.block_manager
|
||||
.rpc_put_block(subrange_hash, current_block)
|
||||
.await
|
||||
if must_upload {
|
||||
garage2.block_manager.rpc_put_block(final_hash, data).await
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
@@ -372,15 +375,9 @@ pub async fn handle_upload_part_copy(
|
||||
// Thing 3: we need to add a block reference
|
||||
garage.block_ref_table.insert(&block_ref),
|
||||
// Thing 4: we need to prefetch the next block
|
||||
async move {
|
||||
match next_block_hash {
|
||||
Some(h) => Ok(Some(garage3.block_manager.rpc_get_block(&h).await?)),
|
||||
None => Ok(None),
|
||||
}
|
||||
},
|
||||
defragmenter.next(),
|
||||
)?;
|
||||
|
||||
block = next_block;
|
||||
next_block = res.3;
|
||||
}
|
||||
|
||||
let data_md5sum = md5hasher.finalize();
|
||||
@@ -553,6 +550,54 @@ impl CopyPreconditionHeaders {
|
||||
}
|
||||
}
|
||||
|
||||
type BlockStreamItemOk = (Vec<u8>, Option<Hash>);
|
||||
type BlockStreamItem = Result<BlockStreamItemOk, garage_util::error::Error>;
|
||||
|
||||
struct Defragmenter<S: Stream<Item = BlockStreamItem>> {
|
||||
block_size: usize,
|
||||
block_stream: Pin<Box<stream::Peekable<S>>>,
|
||||
buffer: Vec<u8>,
|
||||
hash: Option<Hash>,
|
||||
}
|
||||
|
||||
impl<S: Stream<Item = BlockStreamItem>> Defragmenter<S> {
|
||||
fn new(block_size: usize, block_stream: Pin<Box<stream::Peekable<S>>>) -> Self {
|
||||
Self {
|
||||
block_size,
|
||||
block_stream,
|
||||
buffer: vec![],
|
||||
hash: None,
|
||||
}
|
||||
}
|
||||
|
||||
async fn next(&mut self) -> BlockStreamItem {
|
||||
// Fill buffer while we can
|
||||
while let Some(res) = self.block_stream.as_mut().peek().await {
|
||||
let (peeked_next_block, _) = match res {
|
||||
Ok(t) => t,
|
||||
Err(_) => {
|
||||
self.block_stream.next().await.unwrap()?;
|
||||
unreachable!()
|
||||
}
|
||||
};
|
||||
|
||||
if self.buffer.is_empty() {
|
||||
let (next_block, next_block_hash) = self.block_stream.next().await.unwrap()?;
|
||||
self.buffer = next_block;
|
||||
self.hash = next_block_hash;
|
||||
} else if self.buffer.len() + peeked_next_block.len() > self.block_size {
|
||||
break;
|
||||
} else {
|
||||
let (next_block, _) = self.block_stream.next().await.unwrap()?;
|
||||
self.buffer.extend(next_block);
|
||||
self.hash = None;
|
||||
}
|
||||
}
|
||||
|
||||
Ok((std::mem::take(&mut self.buffer), self.hash.take()))
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, PartialEq)]
|
||||
pub struct CopyObjectResult {
|
||||
#[serde(rename = "LastModified")]
|
||||
|
||||
@@ -43,14 +43,22 @@ pub async fn cmd_assign_role(
|
||||
resp => return Err(Error::Message(format!("Invalid RPC response: {:?}", resp))),
|
||||
};
|
||||
|
||||
let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
|
||||
|
||||
let added_nodes = args
|
||||
.node_ids
|
||||
.iter()
|
||||
.map(|node_id| find_matching_node(status.iter().map(|adv| adv.id), node_id))
|
||||
.map(|node_id| {
|
||||
find_matching_node(
|
||||
status
|
||||
.iter()
|
||||
.map(|adv| adv.id)
|
||||
.chain(layout.node_ids().iter().cloned()),
|
||||
node_id,
|
||||
)
|
||||
})
|
||||
.collect::<Result<Vec<_>, _>>()?;
|
||||
|
||||
let mut layout = fetch_layout(rpc_cli, rpc_host).await?;
|
||||
|
||||
let mut roles = layout.roles.clone();
|
||||
roles.merge(&layout.staging);
|
||||
|
||||
@@ -323,11 +331,20 @@ pub fn print_cluster_layout(layout: &ClusterLayout) -> bool {
|
||||
}
|
||||
|
||||
pub fn print_staging_role_changes(layout: &ClusterLayout) -> bool {
|
||||
if !layout.staging.items().is_empty() {
|
||||
let has_changes = layout
|
||||
.staging
|
||||
.items()
|
||||
.iter()
|
||||
.any(|(k, _, v)| layout.roles.get(k) != Some(v));
|
||||
|
||||
if has_changes {
|
||||
println!();
|
||||
println!("==== STAGED ROLE CHANGES ====");
|
||||
let mut table = vec!["ID\tTags\tZone\tCapacity".to_string()];
|
||||
for (id, _, role) in layout.staging.items().iter() {
|
||||
if layout.roles.get(id) == Some(role) {
|
||||
continue;
|
||||
}
|
||||
if let Some(role) = &role.0 {
|
||||
let tags = role.tags.join(",");
|
||||
table.push(format!(
|
||||
|
||||
@@ -208,7 +208,7 @@ pub fn find_matching_node(
|
||||
) -> Result<Uuid, Error> {
|
||||
let mut candidates = vec![];
|
||||
for c in cand {
|
||||
if hex::encode(&c).starts_with(&pattern) {
|
||||
if hex::encode(&c).starts_with(&pattern) && !candidates.contains(&c) {
|
||||
candidates.push(c);
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -48,7 +48,7 @@ opentelemetry = "0.17"
|
||||
|
||||
#netapp = { version = "0.3.0", git = "https://git.deuxfleurs.fr/lx/netapp" }
|
||||
#netapp = { version = "0.4", path = "../../../netapp", features = ["telemetry"] }
|
||||
netapp = { version = "0.4.2", features = ["telemetry"] }
|
||||
netapp = { version = "0.4.4", features = ["telemetry"] }
|
||||
|
||||
hyper = { version = "0.14", features = ["client", "http1", "runtime", "tcp"] }
|
||||
|
||||
|
||||
Reference in New Issue
Block a user