chore(hybridcloud) Cleanup logging and option usage in webhooks (#69192)

With multi-org webhook handling fixed, I don't need all this logging.
The parallel delivery can be used more frequently to help deliver small
backlogs faster.

I've also simplified the parser interface as we had two very similar
methods and no way to fully remove usage of the non-integration focused
one. I think having fewer methods makes the design simpler.

Author: markstory

src/sentry/hybridcloud/tasks/deliver_webhooks.py

@@ -94,7 +94,6 @@ def schedule_webhook_delivery(**kwargs) -> None:
     metrics.distribution(
         "hybridcloud.schedule_webhook_delivery.mailbox_count", scheduled_mailboxes.count()
     )
-    use_parallel = options.get("hybridcloud.webhookpayload.use_parallel")
     for record in scheduled_mailboxes[:BATCH_SIZE]:
         # Reschedule the records that we will attempt to deliver next.
         # We update schedule_for in an attempt to minimize races for potentially in-flight batches.
@@ -106,8 +105,8 @@ def schedule_webhook_delivery(**kwargs) -> None:
         updated_count = WebhookPayload.objects.filter(id__in=Subquery(mailbox_batch)).update(
             schedule_for=timezone.now() + BATCH_SCHEDULE_OFFSET
         )
-        # If we have a half-full batch we should process in parallel as we're likely behind.
-        if use_parallel and updated_count >= int(MAX_MAILBOX_DRAIN / 2):
+        # If we have 1/3 or more in a mailbox we should process in parallel as we're likely behind.
+        if updated_count >= int(MAX_MAILBOX_DRAIN / 3):
             drain_mailbox_parallel.delay(record["id"])
         else:
             drain_mailbox.delay(record["id"])
@@ -216,20 +215,11 @@ def drain_mailbox_parallel(payload_id: int) -> None:
             },
         )
         return
-    logger.info(
-        "drain_mailbox_parallel.start",
-        extra={"mailbox_name": payload.mailbox_name, "id": payload_id},
-    )
 
     # Remove batches payloads that have been backlogged for MAX_DELIVERY_AGE.
     # Once payloads are this old they are low value, and we're better off prioritizing new work.
     max_age = timezone.now() - MAX_DELIVERY_AGE
     if payload.date_added < max_age:
-        logger.info(
-            "drain_mailbox_parallel.max_age_start",
-            extra={"mailbox_name": payload.mailbox_name, "id": payload_id},
-        )
-
         # We delete chunks of stale messages using a subquery
         # because postgres cannot do delete with limit
         stale_query = WebhookPayload.objects.filter(
@@ -283,14 +273,6 @@ def drain_mailbox_parallel(payload_id: int) -> None:
                 threadpool.submit(deliver_message_parallel, record)
                 for record in query[:worker_threads]
             }
-            logger.info(
-                "drain_mailbox_parallel.send_batch",
-                extra={
-                    "mailbox_name": payload.mailbox_name,
-                    "count": len(futures),
-                    "threads": worker_threads,
-                },
-            )
             for future in as_completed(futures):
                 payload_record, err = future.result()
 

src/sentry/middleware/integrations/parsers/base.py

@@ -187,22 +187,14 @@ def get_mailbox_identifier(self, integration: RpcIntegration, data: Mapping[str,
         that can be delivered in parallel. Requires the integration to implement
         `mailbox_bucket_id`
         """
-        # One integration is misbehaving in saas, and only need logs from that instance
-        extra_logging = integration.id == 122177
-
         # If we get fewer than 3000 in 1 hour we don't need to split into buckets
         ratelimit_key = f"webhookpayload:{self.provider}:{integration.id}"
         use_buckets_key = f"{ratelimit_key}:use_buckets"
 
-        is_limited = False
-        ratelimit_val = None
-        reset = None
         use_buckets = cache.get(use_buckets_key)
-        if not use_buckets:
-            is_limited, ratelimit_val, reset = ratelimiter.is_limited_with_value(
-                key=ratelimit_key, window=60 * 60, limit=3000
-            )
-        if not use_buckets and is_limited:
+        if not use_buckets and ratelimiter.is_limited(
+            key=ratelimit_key, window=60 * 60, limit=3000
+        ):
             # Once we have gone over the rate limit in a day, we use smaller
             # buckets for the next day.
             cache.set(use_buckets_key, 1, timeout=ONE_DAY)
@@ -211,19 +203,6 @@ def get_mailbox_identifier(self, integration: RpcIntegration, data: Mapping[str,
                 "integrations.parser.activate_buckets",
                 extra={"provider": self.provider, "integration_id": integration.id},
             )
-
-        if extra_logging:
-            logger.info(
-                "integrations.parser.use_buckets",
-                extra={
-                    "provider": self.provider,
-                    "result": use_buckets or False,
-                    "ratelimit_key": ratelimit_key,
-                    "is_limited": is_limited,
-                    "ratelimit_val": ratelimit_val,
-                    "reset": reset,
-                },
-            )
         if not use_buckets:
             return str(integration.id)
 

src/sentry/middleware/integrations/parsers/github.py

@@ -65,6 +65,6 @@ def get_response(self):
         except (Integration.DoesNotExist, OrganizationIntegration.DoesNotExist):
             return self.get_default_missing_integration_response()
 
-        return self.get_response_from_webhookpayload_for_integration(
-            regions=regions, integration=integration
+        return self.get_response_from_webhookpayload(
+            regions=regions, identifier=integration.id, integration_id=integration.id
         )

src/sentry/middleware/integrations/parsers/gitlab.py

@@ -84,17 +84,9 @@ def get_response_from_gitlab_webhook(self):
         except ValueError:
             data = {}
 
-        identifier = self.get_mailbox_identifier(integration, data)
-        logger.info(
-            "gitlab.webhookpayload.save",
-            extra={
-                "identifier": identifier,
-                "integration_id": integration.id,
-            },
-        )
         return self.get_response_from_webhookpayload(
             regions=regions,
-            identifier=identifier,
+            identifier=self.get_mailbox_identifier(integration, data),
             integration_id=integration.id,
         )
 

src/sentry/middleware/integrations/parsers/jira.py

@@ -81,8 +81,8 @@ def get_response(self):
                 return self.get_response_from_control_silo()
 
         if self.view_class in self.outbox_response_region_classes:
-            return self.get_response_from_webhookpayload_for_integration(
-                regions=regions, integration=integration
+            return self.get_response_from_webhookpayload(
+                regions=regions, identifier=integration.id, integration_id=integration.id
             )
 
         return self.get_response_from_control_silo()

src/sentry/middleware/integrations/parsers/msteams.py

@@ -132,6 +132,6 @@ def get_response(self) -> HttpResponseBase:
             "Scheduling event for request",
             extra={"request_data": self.request_data},
         )
-        return self.get_response_from_webhookpayload_for_integration(
-            regions=regions, integration=integration
+        return self.get_response_from_webhookpayload(
+            regions=regions, identifier=integration.id, integration_id=integration.id
         )

src/sentry/middleware/integrations/parsers/vsts.py

@@ -45,6 +45,6 @@ def get_response(self) -> HttpResponseBase:
         except (Integration.DoesNotExist, OrganizationIntegration.DoesNotExist):
             return self.get_default_missing_integration_response()
 
-        return self.get_response_from_webhookpayload_for_integration(
-            regions=regions, integration=integration
+        return self.get_response_from_webhookpayload(
+            regions=regions, identifier=integration.id, integration_id=integration.id
         )

tests/sentry/hybridcloud/tasks/test_deliver_webhooks.py

@@ -17,7 +17,6 @@
 )
 from sentry.testutils.cases import TestCase
 from sentry.testutils.factories import Factories
-from sentry.testutils.helpers.options import override_options
 from sentry.testutils.region import override_regions
 from sentry.testutils.silo import control_silo_test
 from sentry.types.region import Region, RegionCategory, RegionResolutionError
@@ -135,10 +134,9 @@ def test_schedule_mailbox_with_more_than_batch_size_records(self):
         # No messages delivered
         assert WebhookPayload.objects.count() == num_records
 
-    @override_options({"hybridcloud.webhookpayload.use_parallel": True})
     @patch("sentry.hybridcloud.tasks.deliver_webhooks.drain_mailbox_parallel")
     def test_schedule_mailbox_parallel_task(self, mock_deliver):
-        for _ in range(0, int(MAX_MAILBOX_DRAIN / 2 + 1)):
+        for _ in range(0, int(MAX_MAILBOX_DRAIN / 3 + 1)):
             self.create_webhook_payload(
                 mailbox_name="github:123",
                 region_name="us",

tests/sentry/middleware/integrations/parsers/test_gitlab.py

@@ -175,9 +175,9 @@ def test_routing_webhook_with_mailbox_buckets(self):
             HTTP_X_GITLAB_EVENT="Push Hook",
         )
         with mock.patch(
-            "sentry.middleware.integrations.parsers.base.ratelimiter.is_limited_with_value"
+            "sentry.middleware.integrations.parsers.base.ratelimiter.is_limited"
         ) as mock_is_limited:
-            mock_is_limited.return_value = (True, 3500, 360)
+            mock_is_limited.return_value = True
             parser = GitlabRequestParser(request=request, response_handler=self.get_response)
             response = parser.get_response()
 

tests/sentry/middleware/integrations/parsers/test_jira_server.py

@@ -120,11 +120,11 @@ def test_routing_webhook_with_mailbox_buckets_high_volume(self):
         parser = JiraServerRequestParser(request=request, response_handler=self.get_response)
 
         with mock.patch(
-            "sentry.middleware.integrations.parsers.base.ratelimiter.is_limited_with_value"
+            "sentry.middleware.integrations.parsers.base.ratelimiter.is_limited"
         ) as mock_is_limited, mock.patch(
             "sentry.middleware.integrations.parsers.jira_server.get_integration_from_token"
         ) as mock_get_token:
-            mock_is_limited.return_value = (True, 3500, 360)
+            mock_is_limited.return_value = True
             mock_get_token.return_value = self.integration
             response = parser.get_response()
         assert isinstance(response, HttpResponse)