Log content type to event streams (#28999)

Gets ContentType and StorageId in the logging information.
Ideally it would include `Id<_storage>`, but it's a bit annoying to
get, so just doing StorageUuid for now which is 1-1 equivalent
for the time being.

This makes changes to both usher/backend - which I ensure compatibility of via the `optional` field.

GitOrigin-RevId: a9d40d4a790348c25850d22893b8cbe9b2361348

Author: nipunn1313

Cargo.lock

@@ -38,7 +38,6 @@ keybroker = { path = "../keybroker" }
 lru = { workspace = true }
 maplit = { workspace = true }
 metrics = { path = "../metrics" }
-mime = { workspace = true }
 mime2ext = { workspace = true }
 minitrace = { workspace = true }
 model = { path = "../model" }

crates/application/Cargo.toml

@@ -593,8 +593,18 @@ impl<RT: Runtime> ExportWorker<RT> {
                             file_storage_entry.storage_key,
                         )
                     })?;
+
+                let content_type = file_storage_entry
+                    .content_type
+                    .as_ref()
+                    .map(|ct| ct.parse())
+                    .transpose()?;
                 usage
-                    .track_storage_call("snapshot_export")
+                    .track_storage_call(
+                        "snapshot_export",
+                        Some(file_storage_entry.storage_id.clone()),
+                        content_type,
+                    )
                     .track_storage_egress_size(file_stream.content_length as u64);
                 zip_snapshot_upload
                     .stream_full_file(path, file_stream.stream)

crates/application/src/export_worker.rs

@@ -57,6 +57,7 @@ use common::{
         FieldName,
         MemberId,
         ObjectKey,
+        StorageUuid,
         TableName,
         UdfIdentifier,
     },
@@ -112,7 +113,6 @@ use model::{
         DeploymentAuditLogModel,
     },
     file_storage::{
-        types::StorageUuid,
         FILE_STORAGE_TABLE,
         FILE_STORAGE_VIRTUAL_TABLE,
     },
@@ -1902,7 +1902,7 @@ async fn import_storage_table<RT: Runtime>(
         let content_length = metadata.size.map(|size| ContentLength(size as u64));
         let content_type = metadata
             .content_type
-            .map(|content_type| anyhow::Ok(ContentType::from(mime::Mime::from_str(&content_type)?)))
+            .map(|content_type| anyhow::Ok(ContentType::from_str(&content_type)?))
             .transpose()
             .map_err(|e| ImportError::InvalidConvexValue(lineno, e))?;
         let sha256 = metadata
@@ -1912,7 +1912,9 @@ async fn import_storage_table<RT: Runtime>(
             .map_err(|e| ImportError::InvalidConvexValue(lineno, e))?;
         let storage_id = metadata
             .internal_id
-            .map(|storage_id| StorageUuid::from_str(&storage_id))
+            .map(|storage_id| {
+                StorageUuid::from_str(&storage_id).context("Couldn't parse storage_id")
+            })
             .transpose()
             .map_err(|e| ImportError::InvalidConvexValue(lineno, e))?;
         let creation_time = metadata
@@ -2003,8 +2005,13 @@ async fn import_storage_table<RT: Runtime>(
                 },
             )
             .await?;
+        let content_type = entry
+            .content_type
+            .as_ref()
+            .map(|ct| ct.parse())
+            .transpose()?;
         usage
-            .track_storage_call("snapshot_import")
+            .track_storage_call("snapshot_import", Some(entry.storage_id), content_type)
             .track_storage_ingress_size(file_size);
         num_files += 1;
         if let Some(import_id) = import_id {

crates/application/src/snapshot_import.rs

@@ -0,0 +1,62 @@
+use uuid::Uuid;
+use value::ConvexValue;
+
+#[cfg_attr(any(test, feature = "testing"), derive(proptest_derive::Arbitrary))]
+#[derive(
+    Clone, Debug, Eq, PartialEq, Ord, PartialOrd, derive_more::Display, derive_more::FromStr,
+)]
+pub struct StorageUuid(
+    #[cfg_attr(
+        any(test, feature = "testing"),
+        proptest(strategy = "Uuid::new_v4 as fn() -> Uuid")
+    )]
+    Uuid,
+);
+
+impl From<Uuid> for StorageUuid {
+    fn from(u: Uuid) -> Self {
+        Self(u)
+    }
+}
+
+impl TryFrom<StorageUuid> for ConvexValue {
+    type Error = anyhow::Error;
+
+    fn try_from(s: StorageUuid) -> anyhow::Result<Self> {
+        s.to_string().try_into()
+    }
+}
+
+impl TryFrom<ConvexValue> for StorageUuid {
+    type Error = anyhow::Error;
+
+    fn try_from(v: ConvexValue) -> anyhow::Result<Self> {
+        match v {
+            ConvexValue::String(s) => Ok(StorageUuid(Uuid::try_parse(&s)?)),
+            _ => anyhow::bail!("Can only convert Value::String to StorageUuid"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use proptest::prelude::*;
+    use value::ConvexValue;
+
+    use crate::{
+        testing::assert_roundtrips,
+        types::StorageUuid,
+    };
+
+    proptest! {
+        #![proptest_config(
+            ProptestConfig { failure_persistence: None, ..ProptestConfig::default() }
+        )]
+
+        #[test]
+        fn test_storage_roundtrip(v in any::<StorageUuid>()) {
+            assert_roundtrips::<StorageUuid, ConvexValue>(v);
+        }
+
+    }
+}

crates/common/src/types/file_storage.rs

@@ -18,6 +18,7 @@ mod actions;
 mod admin_key;
 mod backend_state;
 mod environment_variables;
+mod file_storage;
 mod functions;
 mod index;
 mod maybe_value;
@@ -52,6 +53,7 @@ pub use environment_variables::{
     EnvironmentVariable,
     ENV_VAR_LIMIT,
 };
+pub use file_storage::StorageUuid;
 pub use functions::{
     AllowedVisibility,
     FunctionCaller,

crates/common/src/types/mod.rs

@@ -64,7 +64,9 @@ pub enum UsageEvent {
     /// snapshot import/export)
     StorageCall {
         id: String,
+        storage_id: Option<String>,
         call: String,
+        content_type: Option<String>,
     },
     /// Bandwidth from a storage call outside of a user function (e.g. snapshot
     /// import/export).

crates/events/src/usage.rs

@@ -19,7 +19,6 @@ headers = { workspace = true }
 keybroker = { path = "../keybroker" }
 maplit = { workspace = true }
 metrics = { path = "../metrics" }
-mime = { workspace = true }
 model = { path = "../model" }
 storage = { path = "../storage" }
 tracing = { workspace = true }

crates/file_storage/Cargo.toml

@@ -13,7 +13,10 @@ use common::{
         UnixTimestamp,
     },
     sha256::Sha256Digest,
-    types::ConvexOrigin,
+    types::{
+        ConvexOrigin,
+        StorageUuid,
+    },
 };
 use database::Transaction;
 use errors::ErrorMetadata;
@@ -35,12 +38,8 @@ use keybroker::{
     KeyBroker,
 };
 use maplit::btreemap;
-use mime::Mime;
 use model::file_storage::{
-    types::{
-        FileStorageEntry,
-        StorageUuid,
-    },
+    types::FileStorageEntry,
     BatchKey,
     FileStorageId,
     FileStorageModel,
@@ -223,17 +222,14 @@ impl<RT: Runtime> TransactionalFileStorage<RT> {
         get_file_type: GetFileType,
     ) -> anyhow::Result<FileRangeStream> {
         let FileStorageEntry {
-            storage_id: _,
+            storage_id,
             storage_key,
             sha256: _,
             size,
             content_type,
         } = file;
 
-        let content_type = match content_type {
-            None => None,
-            Some(ct) => Some(ct.parse::<Mime>()?.into()),
-        };
+        let content_type = content_type.as_ref().map(|ct| ct.parse()).transpose()?;
 
         let storage_get_stream = self
             .storage
@@ -244,7 +240,8 @@ impl<RT: Runtime> TransactionalFileStorage<RT> {
         let stream = storage_get_stream.stream;
         let content_length = ContentLength(storage_get_stream.content_length as u64);
 
-        let call_tracker = usage_tracker.track_storage_call("get range");
+        let call_tracker =
+            usage_tracker.track_storage_call("get range", Some(storage_id), content_type.clone());
 
         Ok(FileRangeStream {
             content_length,
@@ -402,9 +399,16 @@ impl<RT: Runtime> FileStorage<RT> {
         entry: FileStorageEntry,
         usage_tracker: &dyn StorageUsageTracker,
     ) -> anyhow::Result<DeveloperDocumentId> {
+        let storage_id = entry.storage_id.clone();
+        let size = entry.size;
+        let content_type = entry
+            .content_type
+            .as_ref()
+            .map(|ct| ct.parse())
+            .transpose()?;
+
         // Start/Complete transaction after the slow upload process
         // to avoid OCC risk.
-        let size = entry.size;
         let mut tx = self.database.begin(Identity::system()).await?;
         let virtual_id = self
             .transactional_file_storage
@@ -415,7 +419,7 @@ impl<RT: Runtime> FileStorage<RT> {
             .await?;
 
         usage_tracker
-            .track_storage_call("store")
+            .track_storage_call("store", Some(storage_id), content_type)
             .track_storage_ingress_size(size as u64);
         Ok(virtual_id)
     }

crates/file_storage/src/core.rs

@@ -69,19 +69,20 @@ impl<RT: Runtime> TaskExecutor<RT> {
 
         let entry = self
             .file_storage
-            .upload_file(content_length, content_type, body_stream, digest)
+            .upload_file(content_length, content_type.clone(), body_stream, digest)
             .await?;
+        let storage_id = entry.storage_id.clone();
         let size = entry.size;
-        let storage_id = self
+        let storage_doc_id = self
             .action_callbacks
             .storage_store_file_entry(self.identity.clone(), self.component_id(), entry)
             .await?;
 
         self.usage_tracker
-            .track_storage_call("store")
+            .track_storage_call("store", Some(storage_id), content_type)
             .track_storage_ingress_size(size as u64);
 
-        Ok(storage_id)
+        Ok(storage_doc_id)
     }
 
     #[convex_macro::instrument_future]