Add metrics for various file/blob uploads (#95416)

This adds `PUT` size and latency metrics to all the use-cases that we
care about:
- Attachments
- Debug Files and Artifact Bundles
- Replays
- Profiles
- Nodestore

The size metrics are being emitted for both compressed and uncompressed
sizes, tagged with the used compression algorithm. The `latency` metric
is only relevant for the compressed payload however.

Author: Swatinem

src/sentry/eventstore/processing/base.py

@@ -45,6 +45,10 @@ def store(self, event: Event, unprocessed: bool = False) -> str:
         if unprocessed:
             key = self.__get_unprocessed_key(key)
         self.inner.set(key, event, self.timeout)
+        # TODO(swatinem): we would like to gather size metrics for things stored
+        # in the processing store, though we need `bytes` for that, and it looks
+        # like the processing store is used with `dict`s directly, for which the
+        # encoding is non-obvious.
         return key
 
     def get(self, key: str, unprocessed: bool = False) -> MutableMapping[str, Any] | None:

src/sentry/models/debugfile.py

@@ -35,7 +35,7 @@
 from sentry.db.models.manager.base import BaseManager
 from sentry.models.files.file import File
 from sentry.models.files.utils import clear_cached_files
-from sentry.utils import json
+from sentry.utils import json, metrics
 from sentry.utils.zip import safe_extract_zip
 
 if TYPE_CHECKING:
@@ -315,6 +315,13 @@ def create_dif_from_id(
         file.headers["Content-Type"] = DIF_MIMETYPES[meta.file_format]
         file.save()
 
+    metrics.distribution(
+        "storage.put.size",
+        file.size,
+        tags={"usecase": "debug-files", "compression": "none"},
+        unit="byte",
+    )
+
     dif = ProjectDebugFile.objects.create(
         file=file,
         checksum=file.checksum,

src/sentry/models/eventattachment.py

@@ -17,6 +17,7 @@
 from sentry.db.models.fields.bounded import BoundedIntegerField
 from sentry.db.models.manager.base_query_set import BaseQuerySet
 from sentry.models.files.utils import get_size_and_checksum, get_storage
+from sentry.utils import metrics
 
 # Attachment file types that are considered a crash report (PII relevant)
 CRASH_REPORT_TYPES = ("event.minidump", "event.applecrashreport")
@@ -147,14 +148,29 @@ def putfile(cls, project_id: int, attachment: CachedAttachment) -> PutfileResult
         blob = BytesIO(data)
         size, checksum = get_size_and_checksum(blob)
 
+        metrics.distribution(
+            "storage.put.size",
+            size,
+            tags={"usecase": "attachments", "compression": "none"},
+            unit="byte",
+        )
+
         if can_store_inline(data):
             blob_path = ":" + data.decode()
         else:
             blob_path = "eventattachments/v1/" + FileBlob.generate_unique_path()
 
             storage = get_storage()
-            compressed_blob = BytesIO(zstandard.compress(data))
-            storage.save(blob_path, compressed_blob)
+            compressed_blob = zstandard.compress(data)
+
+            metrics.distribution(
+                "storage.put.size",
+                len(compressed_blob),
+                tags={"usecase": "attachments", "compression": "zstd"},
+                unit="byte",
+            )
+            with metrics.timer("storage.put.latency", tags={"usecase": "attachments"}):
+                storage.save(blob_path, BytesIO(compressed_blob))
 
         return PutfileResult(
             content_type=content_type, size=size, sha1=checksum, blob_path=blob_path

src/sentry/profiles/task.py

@@ -1406,13 +1406,27 @@ def _process_vroomrs_chunk_profile(profile: Profile) -> bool:
                     len(json_profile),
                     tags={"type": "chunk", "platform": profile["platform"]},
                 )
+                metrics.distribution(
+                    "storage.put.size",
+                    len(json_profile),
+                    tags={"usecase": "profiling", "compression": "none"},
+                    unit="byte",
+                )
             with sentry_sdk.start_span(op="json.unmarshal"):
                 chunk = vroomrs.profile_chunk_from_json_str(json_profile, profile["platform"])
             chunk.normalize()
             with sentry_sdk.start_span(op="gcs.write", name="compress and write"):
                 storage = get_profiles_storage()
                 compressed_chunk = chunk.compress()
-                storage.save(chunk.storage_path(), io.BytesIO(compressed_chunk))
+
+                metrics.distribution(
+                    "storage.put.size",
+                    len(compressed_chunk),
+                    tags={"usecase": "profiling", "compression": "lz4"},
+                    unit="byte",
+                )
+                with metrics.timer("storage.put.latency", tags={"usecase": "profiling"}):
+                    storage.save(chunk.storage_path(), io.BytesIO(compressed_chunk))
             with sentry_sdk.start_span(op="processing", name="send chunk to kafka"):
                 payload = build_chunk_kafka_message(chunk)
                 topic = ArroyoTopic(get_topic_definition(Topic.PROFILE_CHUNKS)["real_topic_name"])

src/sentry/replays/lib/storage.py

@@ -206,7 +206,14 @@ def get(self, key: str) -> bytes | None:
     def set(self, key: str, value: bytes) -> None:
         storage = get_storage(self._make_storage_options())
         try:
-            storage.save(key, BytesIO(value))
+            metrics.distribution(
+                "storage.put.size",
+                len(value),
+                tags={"usecase": "replays", "compression": "gzip"},
+                unit="byte",
+            )
+            with metrics.timer("storage.put.latency", tags={"usecase": "replays"}):
+                storage.save(key, BytesIO(value))
         except TooManyRequests:
             # if we 429 because of a dupe segment problem, ignore it
             metrics.incr("replays.lib.storage.TooManyRequests")

src/sentry/tasks/assemble.py

@@ -515,10 +515,19 @@ def _create_or_update_artifact_bundle(
 
         # In case there is not ArtifactBundle with a specific bundle_id, we just create it and return.
         if existing_artifact_bundle is None:
+            file = self.assemble_result.bundle
+
+            metrics.distribution(
+                "storage.put.size",
+                file.size,
+                tags={"usecase": "artifact-bundles", "compression": "none"},
+                unit="byte",
+            )
+
             artifact_bundle = ArtifactBundle.objects.create(
                 organization_id=self.organization.id,
                 bundle_id=bundle_id,
-                file=self.assemble_result.bundle,
+                file=file,
                 artifact_count=self.archive.artifact_count,
                 # By default, a bundle is not indexed.
                 indexing_state=ArtifactBundleIndexingState.NOT_INDEXED.value,

src/sentry/utils/kvstore/bigtable.py

@@ -15,6 +15,7 @@
 from google.cloud.bigtable.row_set import RowSet
 from google.cloud.bigtable.table import Table
 
+from sentry.utils import metrics
 from sentry.utils.codecs import Codec, ZlibCodec, ZstdCodec
 from sentry.utils.kvstore.abstract import KVStorage
 
@@ -235,11 +236,24 @@ def _set(self, key: str, value: bytes, ttl: timedelta | None = None) -> None:
         # tracking now is whether compression is on or not for the data column.
         flags = self.Flags(0)
 
+        metrics.distribution(
+            "storage.put.size",
+            len(value),
+            tags={"usecase": "nodestore", "compression": "none"},
+            unit="byte",
+        )
         if self.compression:
             compression_flag, strategy = self.compression_strategies[self.compression]
             flags |= compression_flag
             value = strategy.encode(value)
 
+            metrics.distribution(
+                "storage.put.size",
+                len(value),
+                tags={"usecase": "nodestore", "compression": self.compression},
+                unit="byte",
+            )
+
         # Only need to write the column at all if any flags are enabled. And if
         # so, pack it into a single byte.
         if flags: