feat(launchpad): create kafka producer (#94726)

<!-- Describe your PR here. -->

<!--

  Sentry employees and contractors can delete or ignore the following.

-->

### Legal Boilerplate

Look, I get it. The entity doing business as "Sentry" was incorporated
in the State of Delaware in 2015 as Functional Software, Inc. and is
gonna need some rights from me in order to utilize my contributions in
this here PR. So here's the deal: I retain all rights, title and
interest in and to my contributions, and by keeping this boilerplate
intact I confirm that Sentry can use, modify, copy, and redistribute my
contributions, under Sentry's choice of terms.

Author: NicoHinderling

src/sentry/conf/server.py

@@ -3391,6 +3391,7 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
     "buffered-segments": "default",
     "buffered-segments-dlq": "default",
     "snuba-ourlogs": "default",
+    "preprod-artifact-events": "default",
     # Taskworker topics
     "taskworker": "default",
     "taskworker-dlq": "default",
@@ -3715,6 +3716,8 @@ def custom_parameter_sort(parameter: dict) -> tuple[str, int]:
 SENTRY_PROFILE_FUNCTIONS_FUTURES_MAX_LIMIT = 10000
 SENTRY_PROFILE_CHUNKS_FUTURES_MAX_LIMIT = 10000
 
+SENTRY_PREPROD_ARTIFACT_EVENTS_FUTURES_MAX_LIMIT = 10000
+
 # How long we should wait for a gateway proxy request to return before giving up
 GATEWAY_PROXY_TIMEOUT: int | None = None
 

src/sentry/conf/types/kafka_definition.py

@@ -55,6 +55,7 @@ class Topic(Enum):
     INGEST_REPLAYS_RECORDINGS = "ingest-replay-recordings"
     INGEST_OCCURRENCES = "ingest-occurrences"
     INGEST_MONITORS = "ingest-monitors"
+    PREPROD_ARTIFACT_EVENTS = "preprod-artifact-events"
     MONITORS_CLOCK_TICK = "monitors-clock-tick"
     MONITORS_CLOCK_TASKS = "monitors-clock-tasks"
     MONITORS_INCIDENT_OCCURRENCES = "monitors-incident-occurrences"

src/sentry/preprod/producer.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from arroyo import Topic as ArroyoTopic
+from arroyo.backends.kafka import KafkaPayload, KafkaProducer, build_kafka_configuration
+from confluent_kafka import KafkaException
+from django.conf import settings
+
+from sentry.conf.types.kafka_definition import Topic
+from sentry.utils import json
+from sentry.utils.arroyo_producer import SingletonProducer
+from sentry.utils.kafka_config import get_kafka_producer_cluster_options, get_topic_definition
+
+logger = logging.getLogger(__name__)
+
+
+def _get_preprod_producer() -> KafkaProducer:
+    cluster_name = get_topic_definition(Topic.PREPROD_ARTIFACT_EVENTS)["cluster"]
+    producer_config = get_kafka_producer_cluster_options(cluster_name)
+    producer_config.pop("compression.type", None)
+    producer_config.pop("message.max.bytes", None)
+    return KafkaProducer(build_kafka_configuration(default_config=producer_config))
+
+
+_preprod_producer = SingletonProducer(
+    _get_preprod_producer, max_futures=settings.SENTRY_PREPROD_ARTIFACT_EVENTS_FUTURES_MAX_LIMIT
+)
+
+
+def produce_preprod_artifact_to_kafka(
+    project_id: int,
+    organization_id: int,
+    artifact_id: int,
+    **kwargs: Any,
+) -> None:
+    payload_data = {
+        "artifact_id": str(artifact_id),
+        "project_id": str(project_id),
+        "organization_id": str(organization_id),
+        **kwargs,
+    }
+
+    partition_key = f"{project_id}-{artifact_id}".encode()
+    payload = KafkaPayload(partition_key, json.dumps(payload_data).encode("utf-8"), [])
+
+    try:
+        topic = get_topic_definition(Topic.PREPROD_ARTIFACT_EVENTS)["real_topic_name"]
+        _preprod_producer.produce(ArroyoTopic(topic), payload)
+    except KafkaException:
+        logger.exception(
+            "Failed to send preprod artifact message to Kafka",
+            extra={"artifact_id": artifact_id, "project_id": project_id},
+        )

src/sentry/preprod/tasks.py

@@ -6,10 +6,12 @@
 from collections.abc import Callable
 from typing import Any
 
+import sentry_sdk
 from django.db import router, transaction
 
 from sentry.models.organization import Organization
 from sentry.models.project import Project
+from sentry.preprod.producer import produce_preprod_artifact_to_kafka
 from sentry.silo.base import SiloMode
 from sentry.tasks.assemble import (
     AssembleResult,
@@ -133,16 +135,6 @@ def assemble_preprod_artifact(
             },
         )
 
-        logger.info(
-            "Finished preprod artifact assembly",
-            extra={
-                "timestamp": datetime.datetime.now().isoformat(),
-                "project_id": project_id,
-                "organization_id": org_id,
-                "checksum": checksum,
-            },
-        )
-
         # where next set of changes will happen
         # TODO: Trigger artifact processing (size analysis, etc.)
         # This is where you'd add logic to:
@@ -152,11 +144,13 @@ def assemble_preprod_artifact(
         # 4. Update state to PROCESSED when done (also update the date_built value to reflect when the artifact was built, among other fields)
 
     except Exception as e:
+        sentry_sdk.capture_exception(e)
         logger.exception(
             "Failed to assemble preprod artifact",
             extra={
                 "project_id": project_id,
                 "organization_id": org_id,
+                "checksum": checksum,
             },
         )
         set_assemble_status(
@@ -166,8 +160,29 @@ def assemble_preprod_artifact(
             ChunkFileState.ERROR,
             detail=str(e),
         )
-    else:
-        set_assemble_status(AssembleTask.PREPROD_ARTIFACT, project_id, checksum, ChunkFileState.OK)
+        return
+
+    # Mark assembly as successful since the artifact was created successfully
+    set_assemble_status(AssembleTask.PREPROD_ARTIFACT, project_id, checksum, ChunkFileState.OK)
+
+    produce_preprod_artifact_to_kafka(
+        project_id=project_id,
+        organization_id=org_id,
+        artifact_id=preprod_artifact.id,
+        checksum=checksum,
+        git_sha=git_sha,
+        build_configuration=build_configuration,
+    )
+
+    logger.info(
+        "Finished preprod artifact assembly and Kafka dispatch",
+        extra={
+            "preprod_artifact_id": preprod_artifact.id,
+            "project_id": project_id,
+            "organization_id": org_id,
+            "checksum": checksum,
+        },
+    )
 
 
 def _assemble_preprod_artifact(