fix clippy

Cargo.nix

@@ -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.parsed.cpu.name == "aarch64" && hostPlatform.parsed.kernel.name == "linux" || hostPlatform.config == "aarch64-linux-android" || 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; };
     };
   });
   
@@ -2659,23 +2659,23 @@ in
     features = builtins.concatLists [
       [ "async-trait" ]
       [ "crossbeam-channel" ]
-      [ "dashmap" ]
+      (lib.optional (rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web") "dashmap")
       [ "default" ]
-      [ "fnv" ]
+      (lib.optional (rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web") "fnv")
-      [ "metrics" ]
+      (lib.optional (rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web") "metrics")
       [ "percent-encoding" ]
       [ "pin-project" ]
       [ "rand" ]
-      [ "rt-tokio" ]
+      (lib.optional (rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web") "rt-tokio")
-      [ "tokio" ]
+      (lib.optional (rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web") "tokio")
-      [ "tokio-stream" ]
+      (lib.optional (rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web") "tokio-stream")
       [ "trace" ]
     ];
     dependencies = {
       async_trait = buildRustPackages."registry+https://github.com/rust-lang/crates.io-index".async-trait."0.1.52" { profileName = "__noProfile"; };
       crossbeam_channel = rustPackages."registry+https://github.com/rust-lang/crates.io-index".crossbeam-channel."0.5.4" { inherit profileName; };
-      dashmap = rustPackages."registry+https://github.com/rust-lang/crates.io-index".dashmap."4.0.2" { inherit profileName; };
+      ${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "dashmap" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".dashmap."4.0.2" { inherit profileName; };
-      fnv = rustPackages."registry+https://github.com/rust-lang/crates.io-index".fnv."1.0.7" { inherit profileName; };
+      ${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "fnv" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".fnv."1.0.7" { inherit profileName; };
       futures_channel = rustPackages."registry+https://github.com/rust-lang/crates.io-index".futures-channel."0.3.21" { inherit profileName; };
       futures_executor = rustPackages."registry+https://github.com/rust-lang/crates.io-index".futures-executor."0.3.21" { inherit profileName; };
       futures_util = rustPackages."registry+https://github.com/rust-lang/crates.io-index".futures-util."0.3.21" { inherit profileName; };
@@ -2685,8 +2685,8 @@ in
       pin_project = rustPackages."registry+https://github.com/rust-lang/crates.io-index".pin-project."1.0.10" { inherit profileName; };
       rand = rustPackages."registry+https://github.com/rust-lang/crates.io-index".rand."0.8.5" { inherit profileName; };
       thiserror = rustPackages."registry+https://github.com/rust-lang/crates.io-index".thiserror."1.0.30" { inherit profileName; };
-      tokio = rustPackages."registry+https://github.com/rust-lang/crates.io-index".tokio."1.17.0" { inherit profileName; };
+      ${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "tokio" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".tokio."1.17.0" { inherit profileName; };
-      tokio_stream = rustPackages."registry+https://github.com/rust-lang/crates.io-index".tokio-stream."0.1.8" { inherit profileName; };
+      ${ if rootFeatures' ? "garage" || rootFeatures' ? "garage_admin" || rootFeatures' ? "garage_api" || rootFeatures' ? "garage_block" || rootFeatures' ? "garage_model" || rootFeatures' ? "garage_rpc" || rootFeatures' ? "garage_table" || rootFeatures' ? "garage_web" then "tokio_stream" else null } = rustPackages."registry+https://github.com/rust-lang/crates.io-index".tokio-stream."0.1.8" { inherit profileName; };
     };
   });
   
@@ -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; };
@@ -4791,8 +4791,8 @@ in
     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 == "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 == "i686-uwp-windows-msvc" || hostPlatform.config == "i686-pc-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-gnu" || hostPlatform.config == "x86_64-uwp-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; };
     };
   });

script/k8s/README.md

@@ -1,13 +0,0 @@
-Spawn a cluster with minikube
-
-```bash
-minikube start
-minikube kubectl -- apply -f config.yaml
-minikube kubectl -- apply -f daemon.yaml
-minikube dashboard
-
-minikube kubectl -- exec -it garage-0 --container garage -- /garage status
-# etc.
-```
-
-

script/k8s/admin.yaml

@@ -1,12 +0,0 @@
-apiVersion: rbac.authorization.k8s.io/v1
-kind: ClusterRoleBinding
-metadata:
-  name: garage-admin
-roleRef:
-  apiGroup: rbac.authorization.k8s.io
-  kind: ClusterRole
-  name: cluster-admin
-subjects:
-- apiGroup: rbac.authorization.k8s.io
-  kind: User
-  name: system:serviceaccount:default:default

script/k8s/config.yaml

@@ -1,30 +0,0 @@
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: garage-config
-  namespace: default
-data:
-  garage.toml: |-
-    metadata_dir = "/tmp/meta"
-    data_dir = "/tmp/data"
-
-    replication_mode = "3"
-
-    rpc_bind_addr = "[::]:3901"
-    rpc_secret = "1799bccfd7411eddcf9ebd316bc1f5287ad12a68094e1c6ac6abde7e6feae1ec"
-
-    bootstrap_peers = []
-
-    kubernetes_namespace = "default"
-    kubernetes_service_name = "garage-daemon"
-    kubernetes_skip_crd = false
-
-    [s3_api]
-    s3_region = "garage"
-    api_bind_addr = "[::]:3900"
-    root_domain = ".s3.garage.tld"
-
-    [s3_web]
-    bind_addr = "[::]:3902"
-    root_domain = ".web.garage.tld"
-    index = "index.html"

script/k8s/daemon.yaml

@@ -1,52 +0,0 @@
-apiVersion: apps/v1
-kind: StatefulSet
-metadata:
-  name: garage
-spec:
-  selector:
-    matchLabels:
-      app: garage
-  serviceName: "garage"
-  replicas: 3 
-  template:
-    metadata:
-      labels:
-        app: garage
-    spec:
-      terminationGracePeriodSeconds: 10
-      containers:
-      - name: garage
-        image: dxflrs/amd64_garage:v0.7.0-rc1
-        ports:
-        - containerPort: 3900
-          name: s3-api
-        - containerPort: 3902
-          name: web-api
-        volumeMounts:
-        - name: fast
-          mountPath: /mnt/fast
-        - name: slow
-          mountPath: /mnt/slow
-        - name: etc
-          mountPath: /etc/garage.toml
-          subPath: garage.toml
-      volumes:
-      - name: etc
-        configMap:
-          name: garage-config
-  volumeClaimTemplates:
-  - metadata:
-      name: fast
-    spec:
-      accessModes: [ "ReadWriteOnce" ]
-      resources:
-        requests:
-          storage: 100Mi
-  - metadata:
-      name: slow
-    spec:
-      accessModes: [ "ReadWriteOnce" ]
-      resources:
-        requests:
-          storage: 100Mi
-  

script/telemetry/README.md

@@ -1,21 +0,0 @@
-Configure your `[admin-api]` endpoint:
-
-```
-[admin]
-api_bind_addr = "0.0.0.0:3903"
-trace_sink = "http://localhost:4317"
-```
-
-Start the test stack:
-
-```
-cd telemetry
-docker-compose up
-```
-
-Access the web interfaces:
-  - [Kibana](http://localhost:5601) - Click on the hamburger menu, in the Observability section, click APM
-  - [Grafana](http://localhost:3000) - Set a password, then on the left menu, click Dashboard -> Browse. On the new page click Import -> Choose the test dashboard we ship `grafana-garage-dashboard-elasticsearch.json` 
-
-
-

script/telemetry/elastic/grafana/provisioning/datasources/elastic.yaml

@@ -4,16 +4,17 @@ datasources:
   - name: DS_ELASTICSEARCH
     type: elasticsearch
     access: proxy
-    url: http://localhost:9200
+    url: http://elastic:9700
     password: ''
     user: ''
-    database: apm-*
+    database: metricbeat-*
     basicAuth: false
     isDefault: true
     jsonData:
-      esVersion: 7.10.0
+      esVersion: 70
       logLevelField: ''
       logMessageField: ''
       maxConcurrentShardRequests: 5
       timeField: "@timestamp"
+      timeInterval: 10s
     readOnly: false

script/telemetry/elastic/otel-config.yaml

@@ -13,7 +13,7 @@ receivers:
         - job_name: "garage"
           scrape_interval: 5s
           static_configs:
-            - targets: ["localhost:3903"]
+            - targets: ["localhost:3909"]
 
 exporters:
   logging:

src/api/s3_copy.rs

@@ -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};
@@ -310,24 +311,46 @@ pub async fn handle_upload_part_copy(
 	// Now, actually copy the blocks
 	let mut md5hasher = Md5::new();
 
-	let mut block = Some(
+	// First, create a stream that is able to read the source blocks
-		garage
+	// and extract the subrange if necessary.
-			.block_manager
+	// The second returned value is an Option<Hash>, that is Some
-			.rpc_get_block(&blocks_to_copy[0].0)
+	// if and only if the block returned is a block that already existed
-			.await?,
+	// 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 mut next_block = defragmenter.next().await?;
-		let (current_block, subrange_hash) = match range_to_copy.clone() {
+
-			Some(r) => {
+	loop {
-				let subrange = block.take().unwrap()[r].to_vec();
+		let (data, existing_block_hash) = next_block;
-				let hash = blake2sum(&subrange);
+		if data.is_empty() {
-				(subrange, hash)
+			break;
-			}
+		}
-			None => (block.take().unwrap(), *block_hash),
+
-		};
+		md5hasher.update(&data[..]);
-		md5hasher.update(&current_block[..]);
+
+		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 +359,25 @@ pub async fn handle_upload_part_copy(
 				offset: current_offset,
 			},
 			VersionBlock {
-				hash: subrange_hash,
+				hash: final_hash,
-				size: current_block.len() as u64,
+				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 res = futures::try_join!(
-		let is_subrange = range_to_copy.is_some();
+			// Thing 1: if the block is not exactly a block that existed before,
-
+			// we need to insert that data as a new block.
-		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.
 			async move {
-				if is_subrange {
+				if must_upload {
-					garage2
+					garage2.block_manager.rpc_put_block(final_hash, data).await
-						.block_manager
-						.rpc_put_block(subrange_hash, current_block)
-						.await
 				} else {
 					Ok(())
 				}
@@ -372,15 +387,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 {
+			defragmenter.next(),
-				match next_block_hash {
-					Some(h) => Ok(Some(garage3.block_manager.rpc_get_block(&h).await?)),
-					None => Ok(None),
-				}
-			},
 		)?;
-
+		next_block = res.3;
-		block = next_block;
 	}
 
 	let data_md5sum = md5hasher.finalize();
@@ -553,6 +562,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")]

src/util/Cargo.toml

@@ -40,4 +40,4 @@ netapp = "0.4"
 http = "0.2"
 hyper = "0.14"
 
-opentelemetry = { version = "0.17", features = [ "rt-tokio", "metrics", "trace" ] }
+opentelemetry = "0.17"