feat(metric_alerts): Make metric alerts work in local dev. (#21801)

This makes the subscription consumer run if `SENTRY_DEV_PROCESS_SUBSCRIPTIONS` is true. Also starts
Kafka and fixes some bugs around auto topic creation that we don't see in prod.

Author: wedamija

src/sentry/conf/server.py

@@ -1442,6 +1442,10 @@ def create_partitioned_queues(name):
 SENTRY_USE_RELAY = True
 SENTRY_RELAY_PORT = 7899
 
+# Controls whether we'll run the snuba subscription processor. If enabled, we'll run
+# it as a worker, and devservices will run Kafka.
+SENTRY_DEV_PROCESS_SUBSCRIPTIONS = False
+
 # The chunk size for attachments in blob store. Should be a power of two.
 SENTRY_ATTACHMENT_BLOB_SIZE = 8 * 1024 * 1024  # 8MB
 
@@ -1525,7 +1529,9 @@ def create_partitioned_queues(name):
         },
         "volumes": {"kafka": {"bind": "/var/lib/kafka"}},
         "only_if": lambda settings, options: (
-            "kafka" in settings.SENTRY_EVENTSTREAM or settings.SENTRY_USE_RELAY
+            "kafka" in settings.SENTRY_EVENTSTREAM
+            or settings.SENTRY_USE_RELAY
+            or settings.SENTRY_DEV_PROCESS_SUBSCRIPTIONS
         ),
     },
     "clickhouse": {
@@ -1883,6 +1889,10 @@ def get_sentry_sdk_config():
 KAFKA_OUTCOMES = "outcomes"
 KAFKA_EVENTS_SUBSCRIPTIONS_RESULTS = "events-subscription-results"
 KAFKA_TRANSACTIONS_SUBSCRIPTIONS_RESULTS = "transactions-subscription-results"
+KAFKA_SUBSCRIPTION_RESULT_TOPICS = {
+    "events": KAFKA_EVENTS_SUBSCRIPTIONS_RESULTS,
+    "transactions": KAFKA_TRANSACTIONS_SUBSCRIPTIONS_RESULTS,
+}
 KAFKA_INGEST_EVENTS = "ingest-events"
 KAFKA_INGEST_ATTACHMENTS = "ingest-attachments"
 KAFKA_INGEST_TRANSACTIONS = "ingest-transactions"

src/sentry/runner/commands/devserver.py

@@ -21,11 +21,22 @@
     "ingest": ["sentry", "run", "ingest-consumer", "--all-consumer-types"],
     "server": ["sentry", "run", "web"],
     "storybook": ["yarn", "storybook"],
+    "subscription-consumer": [
+        "sentry",
+        "run",
+        "query-subscription-consumer",
+        "--commit-batch-size",
+        "1",
+    ],
 }
 
 
-def _get_daemon(name):
-    return (name, _DEFAULT_DAEMONS[name])
+def _get_daemon(name, *args, **kwargs):
+    display_name = name
+    if "suffix" in kwargs:
+        display_name = u"{}-{}".format(name, kwargs["suffix"])
+
+    return (display_name, _DEFAULT_DAEMONS[name] + list(args))
 
 
 @click.command()
@@ -204,6 +215,15 @@ def devserver(
         if eventstream.requires_post_process_forwarder():
             daemons += [_get_daemon("post-process-forwarder")]
 
+        if settings.SENTRY_DEV_PROCESS_SUBSCRIPTIONS:
+            if not settings.SENTRY_EVENTSTREAM == "sentry.eventstream.kafka.KafkaEventStream":
+                raise click.ClickException(
+                    "`SENTRY_DEV_PROCESS_SUBSCRIPTIONS` can only be used when "
+                    "`SENTRY_EVENTSTREAM=sentry.eventstream.kafka.KafkaEventStream`."
+                )
+            for name, topic in settings.KAFKA_SUBSCRIPTION_RESULT_TOPICS.items():
+                daemons += [_get_daemon("subscription-consumer", "--topic", topic, suffix=name)]
+
     if settings.SENTRY_USE_RELAY:
         daemons += [_get_daemon("ingest")]
 

src/sentry/snuba/query_subscription_consumer.py

@@ -6,13 +6,15 @@
 import sentry_sdk
 import six
 from confluent_kafka import Consumer, KafkaException, OFFSET_INVALID, TopicPartition
+from confluent_kafka.admin import AdminClient
 from dateutil.parser import parse as parse_date
 from django.conf import settings
 
 from sentry.snuba.json_schemas import SUBSCRIPTION_PAYLOAD_VERSIONS, SUBSCRIPTION_WRAPPER_SCHEMA
 from sentry.snuba.models import QueryDatasets, QuerySubscription
 from sentry.snuba.tasks import _delete_from_snuba
 from sentry.utils import metrics, json
+from sentry.utils.batching_kafka_consumer import wait_for_topics
 
 logger = logging.getLogger(__name__)
 
@@ -107,6 +109,17 @@ def on_revoke(consumer, partitions):
                 },
             )
 
+        if settings.KAFKA_CONSUMER_AUTO_CREATE_TOPICS:
+            # This is required for confluent-kafka>=1.5.0, otherwise the topics will
+            # not be automatically created.
+            admin_client = AdminClient(
+                {
+                    "bootstrap.servers": conf["bootstrap.servers"],
+                    "allow.auto.create.topics": "true",
+                }
+            )
+            wait_for_topics(admin_client, [self.topic])
+
         self.consumer = Consumer(conf)
         self.consumer.subscribe([self.topic], on_assign=on_assign, on_revoke=on_revoke)
 

src/sentry/utils/batching_kafka_consumer.py

@@ -25,6 +25,42 @@
 DEFAULT_QUEUED_MIN_MESSAGES = 10000
 
 
+def wait_for_topics(admin_client, topics, timeout=10):
+    """
+    Make sure that the provided topics exist and have non-zero partitions in them.
+    """
+    for topic in topics:
+        start = time.time()
+        last_error = None
+
+        while True:
+            if time.time() > start + timeout:
+                raise RuntimeError(
+                    "Timeout when waiting for Kafka topic '%s' to become available, last error: %s".format(
+                        topic, last_error
+                    )
+                )
+
+            result = admin_client.list_topics(topic=topic)
+            topic_metadata = result.topics.get(topic)
+            if topic_metadata and topic_metadata.partitions and not topic_metadata.error:
+                logger.debug("Topic '%s' is ready", topic)
+                break
+            elif topic_metadata.error in {
+                KafkaError.UNKNOWN_TOPIC_OR_PART,
+                KafkaError.LEADER_NOT_AVAILABLE,
+            }:
+                last_error = topic_metadata.error
+                logger.warn("Topic '%s' or its partitions are not ready, retrying...", topic)
+                time.sleep(0.1)
+                continue
+            else:
+                raise RuntimeError(
+                    "Unknown error when waiting for Kafka topic '%s': %s"
+                    % (topic, topic_metadata.error)
+                )
+
+
 @six.add_metaclass(abc.ABCMeta)
 class AbstractBatchWorker(object):
     """The `BatchingKafkaConsumer` requires an instance of this class to
@@ -172,41 +208,6 @@ def __record_timing(self, metric, value, tags=None):
         sample_rate = self.__metrics_sample_rates.get(metric, settings.SENTRY_METRICS_SAMPLE_RATE)
         return self.__metrics.timing(metric, value, tags=tags, sample_rate=sample_rate)
 
-    def _wait_for_topics(self, admin_client, topics, timeout=10):
-        """
-        Make sure that the provided topics exist and have non-zero partitions in them.
-        """
-        for topic in topics:
-            start = time.time()
-            last_error = None
-
-            while True:
-                if time.time() > start + timeout:
-                    raise RuntimeError(
-                        "Timeout when waiting for Kafka topic '%s' to become available, last error: %s".format(
-                            topic, last_error
-                        )
-                    )
-
-                result = admin_client.list_topics(topic=topic)
-                topic_metadata = result.topics.get(topic)
-                if topic_metadata and topic_metadata.partitions and not topic_metadata.error:
-                    logger.debug("Topic '%s' is ready", topic)
-                    break
-                elif topic_metadata.error in {
-                    KafkaError.UNKNOWN_TOPIC_OR_PART,
-                    KafkaError.LEADER_NOT_AVAILABLE,
-                }:
-                    last_error = topic_metadata.error
-                    logger.warn("Topic '%s' or its partitions are not ready, retrying...", topic)
-                    time.sleep(0.1)
-                    continue
-                else:
-                    raise RuntimeError(
-                        "Unknown error when waiting for Kafka topic '%s': %s"
-                        % (topic, topic_metadata.error)
-                    )
-
     def create_consumer(
         self,
         topics,
@@ -236,7 +237,7 @@ def create_consumer(
                     "allow.auto.create.topics": "true",
                 }
             )
-            self._wait_for_topics(admin_client, topics)
+            wait_for_topics(admin_client, topics)
 
         consumer = Consumer(consumer_config)