WIP: Clean up after full attachments rollout

src/sentry/api/endpoints/event_attachment_details.py

@@ -1,8 +1,5 @@
-import contextlib
 import posixpath
-from typing import IO, ContextManager
 
-import sentry_sdk
 from django.http import StreamingHttpResponse
 from rest_framework.request import Request
 from rest_framework.response import Response
@@ -17,11 +14,10 @@
 from sentry.auth.system import is_system_auth
 from sentry.constants import ATTACHMENTS_ROLE_DEFAULT
 from sentry.models.activity import Activity
-from sentry.models.eventattachment import V1_PREFIX, V2_PREFIX, EventAttachment
+from sentry.models.eventattachment import EventAttachment
 from sentry.models.organizationmember import OrganizationMember
 from sentry.services import eventstore
 from sentry.types.activity import ActivityType
-from sentry.utils import metrics
 
 
 class EventAttachmentDetailsPermission(ProjectPermission):
@@ -65,35 +61,12 @@ def download(self, attachment: EventAttachment):
         name = posixpath.basename(" ".join(attachment.name.split()))
 
         def stream_attachment():
-            attachment_file = attachment.getfile()
-            doublewrite_file: IO[bytes] | ContextManager[None] = contextlib.nullcontext()
-            blob_path = attachment.blob_path or ""
-            blob_path = blob_path.startswith(V1_PREFIX) and blob_path.removeprefix(V1_PREFIX) or ""
-            if blob_path.startswith(V2_PREFIX):
-                try:
-                    # We force the attachment model to use the objectstore backend
-                    # by changing its prefix. Its a big hack, but hey why not.
-                    attachment.blob_path = blob_path
-                    doublewrite_file = attachment.getfile()
-                    metrics.incr("storage.attachments.double_write.read")
-                except Exception:
-                    sentry_sdk.capture_exception()
-
             # TODO: We should pass along the `Accept-Encoding`, so we can avoid
             # decompressing on the API side, and just transfer the already
             # compressed bytes to the client as it indicated it can handle it.
-            with attachment_file as af, doublewrite_file as df:
-                while filestore_chunk := af.read(4096):
-                    if df:
-                        try:
-                            objectstore_chunk = df.read(4096)
-                            assert filestore_chunk == objectstore_chunk
-                        except Exception:
-                            # If we have encountered one error, clear the reference
-                            # to avoid spamming more errors for all the remaining chunks.
-                            df = None
-                            sentry_sdk.capture_exception()
-                    yield filestore_chunk
+            with attachment.getfile() as attachment_file:
+                while chunk := attachment_file.read(4096):
+                    yield chunk
 
         response = StreamingHttpResponse(
             stream_attachment(),

src/sentry/attachments.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+__all__ = [
+    "store_attachments_for_event",
+    "get_attachments_for_event",
+    "delete_cached_and_ratelimited_attachments",
+    "CachedAttachment",
+]
+
+from typing import TYPE_CHECKING, Any
+
+import sentry_sdk
+
+from sentry.objectstore import get_attachments_session
+from sentry.utils.json import prune_empty_keys
+
+if TYPE_CHECKING:
+    from sentry.models.project import Project
+
+
+UNINITIALIZED_DATA = object()
+
+
+class CachedAttachment:
+    def __init__(
+        self,
+        name=None,
+        content_type=None,
+        type=None,
+        data=UNINITIALIZED_DATA,
+        stored_id: str | None = None,
+        rate_limited=None,
+        size=None,
+        **kwargs,
+    ):
+        self.name = name
+        self.content_type = content_type
+        self.type = type or "event.attachment"
+        assert isinstance(self.type, str), self.type
+        self.rate_limited = rate_limited
+
+        if size is not None:
+            self.size = size
+        elif data not in (None, UNINITIALIZED_DATA):
+            self.size = len(data)
+        else:
+            self.size = 0
+
+        self.stored_id = stored_id
+        self._data = data
+
+    def load_data(self, project: Project | None = None) -> bytes:
+        if self.stored_id:
+            assert project
+            session = get_attachments_session(project.organization_id, project.id)
+            return session.get(self.stored_id).payload.read()
+
+        assert self._data is not UNINITIALIZED_DATA
+        return self._data
+
+    def meta(self) -> dict:
+        return prune_empty_keys(
+            {
+                "name": self.name,
+                "rate_limited": self.rate_limited,
+                "content_type": self.content_type,
+                "type": self.type,
+                "size": self.size,
+                "stored_id": self.stored_id,
+            }
+        )
+
+
+@sentry_sdk.trace
+def store_attachments_for_event(project: Project, event: Any, attachments: list[CachedAttachment]):
+    """
+    Stores the given list of `attachments` belonging to `event` for processing.
+
+    The attachment metadata is stored within the `event`, and attachment payloads
+    are stored in `objectstore`.
+    """
+
+    attachments_metadata: list[dict] = []
+    for attachment in attachments:
+        # if the attachment has non-empty data set, we want to store it, overwriting any existing data in case a `stored_id` is set.
+        if attachment._data and attachment._data is not UNINITIALIZED_DATA:
+            session = get_attachments_session(project.organization_id, project.id)
+            attachment.stored_id = session.put(attachment._data, key=attachment.stored_id)
+
+        attachments_metadata.append(attachment.meta())
+
+    event["_attachments"] = attachments_metadata
+
+
+def get_attachments_for_event(event: Any) -> list[CachedAttachment]:
+    """
+    Retrieves the attachments belonging to the given `event`.
+    """
+
+    return [CachedAttachment(**attachment) for attachment in event.get("_attachments", [])]
+
+
+@sentry_sdk.trace
+def delete_cached_and_ratelimited_attachments(
+    project: Project, attachments: list[CachedAttachment]
+):
+    """
+    This deletes all the `rate_limited` attachments from the `objectstore`.
+    Non-ratelimited attachments which are already stored in `objectstore` will
+    be retained there for long-term storage.
+    """
+    for attachment in attachments:
+        # deletes from objectstore if no long-term storage is desired
+        if attachment.rate_limited and attachment.stored_id:
+            get_attachments_session(project.organization_id, project.id).delete(
+                attachment.stored_id
+            )