ref(data_export): add path param to gcs_data_sink (#101993)

This allows us to specify a prefix under the destination bucket when
transferring data from another bucket.

This will make it possible to use the same bucket across several event
types and specify the right prefix for each.

Author: viglia

src/sentry/profiles/data_export.py

@@ -14,6 +14,7 @@ def export_profiles_data(
     organization_id: int,
     gcp_project_id: str,
     destination_bucket: str,
+    destination_prefix: str,
     blob_export_job_duration: timedelta = EXPORT_JOB_DURATION_DEFAULT,
     source_bucket: str = EXPORT_JOB_SOURCE_BUCKET,
     pubsub_topic_name: str | None = None,
@@ -43,6 +44,7 @@ def export_profiles_data(
         source_bucket=source_bucket,
         source_prefix=f"{organization_id}",
         destination_bucket=destination_bucket,
+        destination_prefix=destination_prefix,
         notification_topic=pubsub_topic_name,
         job_description="Profiles EU Compliance Export",
         job_duration=blob_export_job_duration,

src/sentry/replays/data_export.py

@@ -139,6 +139,7 @@ def create_transfer_job[T](
     source_bucket: str,
     source_prefix: str,
     destination_bucket: str,
+    destination_prefix: str,
     job_description: str,
     job_duration: timedelta,
     do_create_transfer_job: Callable[[CreateTransferJobRequest], T],
@@ -164,6 +165,7 @@ def create_transfer_job[T](
     :param destination_bucket:
     :param job_duration: The amount of time the job should take to complete. Longer runs put less
         pressure on our buckets.
+    :param destination_prefix:
     :param notification_topic: Specifying a topic will enable automatic run retries on failure.
     :param do_create_transfer_job: Injected function which creates the transfer-job.
     :param get_current_datetime: Injected function which computes the current datetime.
@@ -177,7 +179,7 @@ def create_transfer_job[T](
         status=storage_transfer_v1.TransferJob.Status.ENABLED,
         transfer_spec=TransferSpec(
             gcs_data_source=GcsData(bucket_name=source_bucket, path=source_prefix),
-            gcs_data_sink=GcsData(bucket_name=destination_bucket),
+            gcs_data_sink=GcsData(bucket_name=destination_bucket, path=destination_prefix),
         ),
         schedule=Schedule(
             schedule_start_date=date_pb2.Date(
@@ -570,6 +572,7 @@ def export_replay_blob_data[T](
     project_id: int,
     gcp_project_id: str,
     destination_bucket: str,
+    destination_prefix: str,
     job_duration: timedelta,
     do_create_transfer_job: Callable[[CreateTransferJobRequest], T],
     pubsub_topic_name: str | None = None,
@@ -589,6 +592,7 @@ def export_replay_blob_data[T](
             source_bucket=source_bucket,
             source_prefix=f"{retention_days}/{project_id}",
             destination_bucket=destination_bucket,
+            destination_prefix=destination_prefix,
             notification_topic=pubsub_topic_name,
             job_description="Session Replay EU Compliance Export",
             job_duration=job_duration,
@@ -600,6 +604,7 @@ def export_replay_data(
     organization_id: int,
     gcp_project_id: str,
     destination_bucket: str,
+    destination_prefix: str,
     blob_export_job_duration: timedelta = EXPORT_JOB_DURATION_DEFAULT,
     database_rows_per_page: int = EXPORT_QUERY_ROWS_PER_PAGE,
     database_pages_per_task: int = EXPORT_QUERY_PAGES_PER_TASK,
@@ -647,6 +652,7 @@ def export_replay_data(
             project_id=project.id,
             gcp_project_id=gcp_project_id,
             destination_bucket=destination_bucket,
+            destination_prefix=destination_prefix,
             pubsub_topic_name=pubsub_topic_name,
             source_bucket=source_bucket,
             job_duration=blob_export_job_duration,

tests/sentry/replays/test_data_export.py

@@ -52,6 +52,7 @@ def test_replay_data_export(default_organization, default_project, replay_store)
             organization_id=default_organization.id,
             gcp_project_id="1",
             destination_bucket="destination",
+            destination_prefix="destination_prefix/",
             database_rows_per_page=1,
         )
         assert create_job.called
@@ -82,6 +83,7 @@ def test_replay_data_export_invalid_organization(default_project, replay_store)
             organization_id=1,
             gcp_project_id="1",
             destination_bucket="destination",
+            destination_prefix="destination_prefix/",
             database_rows_per_page=1,
         )
         assert not create_job.called
@@ -107,6 +109,7 @@ def test_replay_data_export_no_replay_projects(  # type: ignore[no-untyped-def]
             organization_id=default_organization.id,
             gcp_project_id="1",
             destination_bucket="destination",
+            destination_prefix="destination_prefix/",
             database_rows_per_page=1,
         )
         assert not create_job.called
@@ -133,6 +136,7 @@ def test_replay_data_export_no_replay_data(  # type: ignore[no-untyped-def]
             organization_id=default_organization.id,
             gcp_project_id="1",
             destination_bucket="destination",
+            destination_prefix="destination_prefix/",
             database_rows_per_page=1,
         )
 

tests/sentry/replays/unit/test_data_export.py

@@ -37,6 +37,7 @@ def test_export_blob_data() -> None:
         source_bucket=bucket_name,
         source_prefix=bucket_prefix,
         destination_bucket="b",
+        destination_prefix="destination_prefix/",
         notification_topic=pubsub_topic,
         job_description=job_description,
         job_duration=job_duration,
@@ -52,7 +53,7 @@ def test_export_blob_data() -> None:
             status=storage_transfer_v1.TransferJob.Status.ENABLED,
             transfer_spec=TransferSpec(
                 gcs_data_source=GcsData(bucket_name=bucket_name, path=bucket_prefix),
-                gcs_data_sink=GcsData(bucket_name="b"),
+                gcs_data_sink=GcsData(bucket_name="b", path="destination_prefix/"),
             ),
             schedule=Schedule(
                 schedule_start_date=date_pb2.Date(
@@ -97,7 +98,9 @@ def test_retry_export_blob_data() -> None:
 
 def test_export_replay_blob_data() -> None:
     jobs = []
-    export_replay_blob_data(1, "1", "test", timedelta(days=1), lambda job: jobs.append(job))
+    export_replay_blob_data(
+        1, "1", "test", "dest_prefix/", timedelta(days=1), lambda job: jobs.append(job)
+    )
 
     # Assert a job is created for each retention-period.
     assert len(jobs) == 3