diff --git a/src/sentry/conf/server.py b/src/sentry/conf/server.py index 5137e1fc3899..bc8f241713c4 100644 --- a/src/sentry/conf/server.py +++ b/src/sentry/conf/server.py @@ -861,6 +861,11 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str: # Expected to be a JSON encoded dictionary of namespace:topic TASKWORKER_ROUTES = os.getenv("TASKWORKER_ROUTES") +# If true, taskbroker-client's TaskProducer will be used to produce messages to Kafka +# from within tasks. +# Set to True in the worker child entrypoint in taskworker/bootstrap.py. +TASKWORKER_USE_TASK_PRODUCER: bool = False + # The list of modules that workers will import after starting up # Taskworkers need to import task modules to make tasks # accessible to the worker. diff --git a/src/sentry/issues/producer.py b/src/sentry/issues/producer.py index 905667e28de1..9e9e76a496d4 100644 --- a/src/sentry/issues/producer.py +++ b/src/sentry/issues/producer.py @@ -2,6 +2,7 @@ import logging from collections.abc import MutableMapping +from functools import partial from typing import Any, cast from arroyo import Topic as ArroyoTopic @@ -9,12 +10,15 @@ from arroyo.types import Message, Value from confluent_kafka import KafkaException from django.conf import settings +from taskbroker_client.worker.producer import TaskProducer from sentry.conf.types.kafka_definition import Topic from sentry.hybridcloud.rpc import ValueEqualityEnum from sentry.issues.issue_occurrence import IssueOccurrence from sentry.issues.run import process_message from sentry.issues.status_change_message import StatusChangeMessage +from sentry.options.rollout import in_random_rollout +from sentry.taskworker.adapters import SentryMetricsBackend from sentry.utils import json from sentry.utils.arroyo_producer import SingletonProducer, get_arroyo_producer from sentry.utils.kafka_config import get_topic_definition @@ -34,9 +38,9 @@ class PayloadType(ValueEqualityEnum): STATUS_CHANGE = "status_change" -def _get_occurrence_producer() -> KafkaProducer: +def _get_occurrence_producer(name: str = "sentry.issues.producer") -> KafkaProducer: return get_arroyo_producer( - "sentry.issues.producer", + name, Topic.INGEST_OCCURRENCES, exclude_config_keys=["compression.type", "message.max.bytes"], ) @@ -46,6 +50,12 @@ def _get_occurrence_producer() -> KafkaProducer: _get_occurrence_producer, max_futures=settings.SENTRY_ISSUE_PLATFORM_FUTURES_MAX_LIMIT ) +_occurrence_task_producer = TaskProducer( + name="sentry.issues.tasks.producer", + producer_factory=partial(_get_occurrence_producer, name="sentry.issues.tasks.producer"), + metrics_backend=SentryMetricsBackend(), +) + def produce_occurrence_to_kafka( payload_type: PayloadType = PayloadType.OCCURRENCE, @@ -78,7 +88,12 @@ def produce_occurrence_to_kafka( try: topic = get_topic_definition(Topic.INGEST_OCCURRENCES)["real_topic_name"] - _occurrence_producer.produce(ArroyoTopic(topic), payload) + if settings.TASKWORKER_USE_TASK_PRODUCER and in_random_rollout( + "tasks.producer.occurrences.rollout" + ): + _occurrence_task_producer.produce(ArroyoTopic(topic), payload) + else: + _occurrence_producer.produce(ArroyoTopic(topic), payload) except KafkaException: logger.exception( "Failed to send occurrence to issue platform", diff --git a/src/sentry/options/defaults.py b/src/sentry/options/defaults.py index d1505a79bb33..85028d959f11 100644 --- a/src/sentry/options/defaults.py +++ b/src/sentry/options/defaults.py @@ -3732,3 +3732,11 @@ type=Bool, flags=FLAG_MODIFIABLE_BOOL | FLAG_AUTOMATOR_MODIFIABLE, ) + +# Rolls out the new TaskProducer to calls of produce_occurrence_to_kafka() from within taskworkers +register( + "tasks.producer.occurrences.rollout", + type=Float, + default=0.0, + flags=FLAG_AUTOMATOR_MODIFIABLE, +) diff --git a/src/sentry/taskworker/bootstrap.py b/src/sentry/taskworker/bootstrap.py index 99e38b53c89e..cb7f2197d003 100644 --- a/src/sentry/taskworker/bootstrap.py +++ b/src/sentry/taskworker/bootstrap.py @@ -8,6 +8,10 @@ configure() +from django.conf import settings + from sentry.taskworker.runtime import app +settings.TASKWORKER_USE_TASK_PRODUCER = True + __all__ = ("app",)