[STREAM-611] Move Launchpad to Taskbroker

devenv/sync.py

@@ -288,7 +288,7 @@ def main(context: dict[str, str]) -> int:
         (
             "docker",
             "exec",
-            "sentry-postgres-1",
+            "postgres-postgres-1",
             "psql",
             "sentry",
             "postgres",

pyproject.toml

@@ -171,7 +171,7 @@ dev = [
     "pyupgrade>=3.19.1",
     "responses>=0.23.1",
     "selenium>=4.16.0",
-    "sentry-cli>=2.16.0",
+    "sentry-cli>=3.1.0",
     "sentry-covdefaults-disable-branch-coverage>=1.0.2",
     "sentry-devenv>=1.22.0",
     "sentry-forked-django-stubs>=5.2.2.post2",

src/sentry/preprod/tasks.py

@@ -26,7 +26,7 @@
     PreprodArtifactSizeMetrics,
     PreprodBuildConfiguration,
 )
-from sentry.preprod.producer import PreprodFeature, produce_preprod_artifact_to_kafka
+from sentry.preprod.producer import PreprodFeature
 from sentry.preprod.quotas import (
     has_installable_quota,
     has_size_quota,
@@ -54,6 +54,18 @@
 
 logger = logging.getLogger(__name__)
 
+from sentry.taskworker.registry import TaskRegistry
+
+# Separate registry for launchpad tasks
+launchpad_registry = TaskRegistry(application="launchpad")
+launchpad_namespace = launchpad_registry.create_namespace(name="default")
+
+
+@launchpad_namespace.register(name="process_artifact")
+def process_artifact(*args: list[Any], **kwargs: dict[str, Any]) -> None:
+    """Stub - actual implementation in Launchpad service"""
+    pass
+
 
 @instrumented_task(
     name="sentry.preprod.tasks.assemble_preprod_artifact",
@@ -195,11 +207,15 @@ def assemble_preprod_artifact(
         if run_distribution:
             requested_features.append(PreprodFeature.BUILD_DISTRIBUTION)
 
-        produce_preprod_artifact_to_kafka(
-            project_id=project_id,
-            organization_id=org_id,
-            artifact_id=artifact_id,
-            requested_features=requested_features,
+        print("> APPLYING ASYNC process_...")
+
+        process_artifact.apply_async(
+            kwargs={
+                "artifact_id": str(artifact_id),
+                "project_id": str(project_id),
+                "organization_id": str(org_id),
+                "requested_features": [feature.value for feature in requested_features],
+            }
         )
     except Exception as e:
         user_friendly_error_message = "Failed to dispatch preprod artifact event for analysis"

src/sentry/taskworker/registry.py

@@ -156,6 +156,8 @@ def _handle_produce_future(self, future: ProducerFuture, tags: dict[str, str]) -
     def send_task(self, activation: TaskActivation, wait_for_delivery: bool = False) -> None:
         topic = self.router.route_namespace(self.name)
 
+        print(f"> SENDING {activation.taskname} TO TOPIC {topic}...")
+
         with sentry_sdk.start_span(
             op=OP.QUEUE_PUBLISH,
             name=activation.taskname,

src/sentry/taskworker/task.py

@@ -124,6 +124,9 @@ def apply_async(
         activation = self.create_activation(
             args=args, kwargs=kwargs, headers=headers, expires=expires, countdown=countdown
         )
+
+        print(f"> CREATED ACTIVATION {activation.taskname} (application {activation.application})")
+
         if settings.TASKWORKER_ALWAYS_EAGER:
             self._func(*args, **kwargs)
         else:

src/sentry/taskworker/worker.py

@@ -192,6 +192,7 @@ def _add_task(self) -> bool:
 
         inflight = self.fetch_task()
         if inflight:
+            print(f"> GOT TASK {inflight.activation.taskname}")
             try:
                 start_time = time.monotonic()
                 self._child_tasks.put(inflight)

src/sentry/taskworker/workerchild.py

@@ -382,6 +382,7 @@ def _execute_activation(task_func: Task[Any, Any], activation: TaskActivation) -
                 if "__start_time" in kwargs:
                     kwargs.pop("__start_time")
 
+                print(f"> EXECUTING TASK {activation.namespace}:{activation.taskname}")
                 try:
                     task_func(*args, **kwargs)
                     transaction.set_status(SPANSTATUS.OK)