ref: Give backends a name (#101)

This introduces a `name` to the generic backends, which we emit as a
separate metrics tag now.

Author: Swatinem

objectstore-service/src/backend/bigtable.rs

@@ -203,6 +203,10 @@ impl BigTableBackend {
 
 #[async_trait::async_trait]
 impl Backend for BigTableBackend {
+    fn name(&self) -> &'static str {
+        "bigtable"
+    }
+
     async fn put_object(
         &self,
         path: &ObjectPath,

objectstore-service/src/backend/gcs.rs

@@ -262,6 +262,10 @@ impl fmt::Debug for GcsBackend {
 
 #[async_trait::async_trait]
 impl Backend for GcsBackend {
+    fn name(&self) -> &'static str {
+        "gcs"
+    }
+
     async fn put_object(
         &self,
         path: &ObjectPath,

objectstore-service/src/backend/local_fs.rs

@@ -28,6 +28,10 @@ impl LocalFsBackend {
 
 #[async_trait::async_trait]
 impl Backend for LocalFsBackend {
+    fn name(&self) -> &'static str {
+        "local-fs"
+    }
+
     async fn put_object(
         &self,
         path: &ObjectPath,

objectstore-service/src/backend/mod.rs

@@ -22,6 +22,9 @@ pub type BackendStream = BoxStream<'static, io::Result<Bytes>>;
 
 #[async_trait::async_trait]
 pub trait Backend: Debug + Send + Sync + 'static {
+    /// The backend name, used for diagnostics.
+    fn name(&self) -> &'static str;
+
     async fn put_object(
         &self,
         path: &ObjectPath,

objectstore-service/src/backend/s3_compatible.rs

@@ -126,6 +126,10 @@ impl S3CompatibleBackend<NoToken> {
 
 #[async_trait::async_trait]
 impl<T: TokenProvider> Backend for S3CompatibleBackend<T> {
+    fn name(&self) -> &'static str {
+        "s3-compatible"
+    }
+
     async fn put_object(
         &self,
         path: &ObjectPath,

objectstore-service/src/lib.rs

@@ -114,38 +114,59 @@ impl StorageService {
             }
         }
 
-        let stored_size = Arc::new(AtomicU64::new(0));
-        let stream = futures_util::stream::once(async { Ok(first_chunk.into()) })
-            .chain(stream)
-            .inspect({
-                let stored_size = Arc::clone(&stored_size);
-                move |res| {
-                    if let Ok(chunk) = res {
-                        stored_size.fetch_add(chunk.len() as u64, Ordering::Relaxed);
-                    }
-                }
-            })
-            .boxed();
-
-        let backend_tag = match backend {
+        let (backend_choice, backend_ty, stored_size) = match backend {
             BackendChoice::HighVolume => {
+                let stored_size = first_chunk.len() as u64;
+                let stream = futures_util::stream::once(async { Ok(first_chunk.into()) }).boxed();
+
                 self.0
                     .high_volume_backend
                     .put_object(&path, metadata, stream)
                     .await?;
-                "high-volume"
+                (
+                    "high-volume",
+                    self.0.high_volume_backend.name(),
+                    stored_size,
+                )
             }
             BackendChoice::LongTerm => {
+                let stored_size = Arc::new(AtomicU64::new(0));
+                let stream = futures_util::stream::once(async { Ok(first_chunk.into()) })
+                    .chain(stream)
+                    .inspect({
+                        let stored_size = Arc::clone(&stored_size);
+                        move |res| {
+                            if let Ok(chunk) = res {
+                                stored_size.fetch_add(chunk.len() as u64, Ordering::Relaxed);
+                            }
+                        }
+                    })
+                    .boxed();
+
                 self.0
                     .long_term_backend
                     .put_object(&path, metadata, stream)
                     .await?;
-                "long-term"
+                (
+                    "long-term",
+                    self.0.long_term_backend.name(),
+                    stored_size.load(Ordering::Acquire),
+                )
             }
         };
 
-        merni::distribution!("put.latency"@s: start.elapsed(), "usecase" => path.usecase, "backend" => backend_tag);
-        merni::distribution!("put.size"@b: stored_size.load(Ordering::Acquire), "usecase" => path.usecase, "backend" => backend_tag);
+        merni::distribution!(
+            "put.latency"@s: start.elapsed(),
+            "usecase" => path.usecase,
+            "backend_choice" => backend_choice,
+            "backend_type" => backend_ty
+        );
+        merni::distribution!(
+            "put.size"@b: stored_size,
+            "usecase" => path.usecase,
+            "backend_choice" => backend_choice,
+            "backend_type" => backend_ty
+        );
 
         Ok(path)
     }

objectstore-service/src/path.rs

@@ -3,7 +3,7 @@ use std::fmt::{self, Display};
 /// The fully scoped path of an object.
 ///
 /// This consists of a usecase, the scope, and the user-defined object key.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct ObjectPath {
     /// The usecase, or "product" this object belongs to.
     ///