Merge branch 'master' into dynamic-max-pages

Author: JoshuaKGoldberg

.github/CODEOWNERS

@@ -276,6 +276,8 @@ pnpm-lock.yaml                                           @getsentry/owners-js-de
 /src/sentry/api/endpoints/organization_measurements_meta.py                 @getsentry/data-browsing
 src/sentry/api/endpoints/organization_attribute_mappings.py                 @getsentry/data-browsing
 
+/src/sentry/integrations/slack/unfurl/explore.py                           @getsentry/data-browsing
+
 /tests/snuba/api/endpoints/*                                                @getsentry/data-browsing
 /tests/snuba/api/endpoints/test_organization_tags.py                        @getsentry/data-browsing
 /tests/snuba/api/endpoints/test_organization_events_histogram.py            @getsentry/data-browsing

pyproject.toml

@@ -101,7 +101,7 @@ dependencies = [
   "statsd>=3.3.0",
   "structlog>=22.1.0",
   "symbolic>=12.14.1",
-  "taskbroker-client>=0.1.7",
+  "taskbroker-client>=0.1.8",
   "tiktoken>=0.8.0",
   "tokenizers>=0.22.0",
   "tldextract>=5.1.2",

src/sentry/features/temporary.py

@@ -577,7 +577,7 @@ def register_temporary_features(manager: FeatureManager) -> None:
     # Enable lightweight RCA clustering write path (generate embeddings on new issues)
     manager.add("organizations:supergroups-lightweight-rca-clustering-write", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Enable lightweight RCA clustering read path (query lightweight embeddings in supergroup APIs)
-    manager.add("organizations:supergroups-lightweight-rca-clustering-read", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
+    manager.add("organizations:supergroups-lightweight-rca-clustering-read", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
 
     manager.add("projects:workflow-engine-performance-detectors", ProjectFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
 

src/sentry/integrations/msteams/webhook.py

@@ -45,6 +45,7 @@
     IntegrationProviderSlug,
     IntegrationResponse,
 )
+from sentry.integrations.utils.webhook_viewer_context import webhook_viewer_context
 from sentry.models.activity import ActivityIntegration
 from sentry.models.apikey import ApiKey
 from sentry.models.group import Group
@@ -434,18 +435,19 @@ def _handle_team_member_removed(self, request: Request) -> Response:
         )
         if len(org_integrations) > 0:
             for org_integration in org_integrations:
-                create_audit_entry(
-                    request=request,
-                    organization_id=org_integration.organization_id,
-                    target_object=integration.id,
-                    event=audit_log.get_event_id("INTEGRATION_REMOVE"),
-                    actor_label="Teams User",
-                    data={
-                        "provider": integration.provider,
-                        "name": integration.name,
-                        "team_id": team_id,
-                    },
-                )
+                with webhook_viewer_context(org_integration.organization_id):
+                    create_audit_entry(
+                        request=request,
+                        organization_id=org_integration.organization_id,
+                        target_object=integration.id,
+                        event=audit_log.get_event_id("INTEGRATION_REMOVE"),
+                        actor_label="Teams User",
+                        data={
+                            "provider": integration.provider,
+                            "name": integration.name,
+                            "team_id": team_id,
+                        },
+                    )
 
         integration_service.delete_integration(integration_id=integration.id)
         return self.respond(status=204)
@@ -584,60 +586,61 @@ def _handle_action_submitted(self, request: Request) -> Response:
             )
             return self.respond(status=404)
 
-        idp = identity_service.get_provider(
-            provider_type=IntegrationProviderSlug.MSTEAMS.value, provider_ext_id=team_id
-        )
-        if idp is None:
-            logger.info(
-                "msteams.action.invalid-team-id",
-                extra={
-                    "team_id": team_id,
-                    "integration_id": integration.id,
-                    "organization_id": group.organization.id,
-                },
+        with webhook_viewer_context(group.organization.id):
+            idp = identity_service.get_provider(
+                provider_type=IntegrationProviderSlug.MSTEAMS.value, provider_ext_id=team_id
             )
-            return self.respond(status=404)
+            if idp is None:
+                logger.info(
+                    "msteams.action.invalid-team-id",
+                    extra={
+                        "team_id": team_id,
+                        "integration_id": integration.id,
+                        "organization_id": group.organization.id,
+                    },
+                )
+                return self.respond(status=404)
 
-        identity = identity_service.get_identity(
-            filter={"provider_id": idp.id, "identity_ext_id": user_id}
-        )
-        if identity is None:
-            associate_url = build_linking_url(
-                integration, group.organization, user_id, team_id, tenant_id
+            identity = identity_service.get_identity(
+                filter={"provider_id": idp.id, "identity_ext_id": user_id}
             )
+            if identity is None:
+                associate_url = build_linking_url(
+                    integration, group.organization, user_id, team_id, tenant_id
+                )
 
-            card = build_linking_card(associate_url)
-            user_conversation_id = client.get_user_conversation_id(user_id, tenant_id)
-            client.send_card(user_conversation_id, card)
-            return self.respond(status=201)
+                card = build_linking_card(associate_url)
+                user_conversation_id = client.get_user_conversation_id(user_id, tenant_id)
+                client.send_card(user_conversation_id, card)
+                return self.respond(status=201)
 
-        # update the state of the issue
-        issue_change_response = self._issue_state_change(group, identity, data["value"])
+            # update the state of the issue
+            issue_change_response = self._issue_state_change(group, identity, data["value"])
 
-        # get the rules from the payload
-        rules = tuple(Rule.objects.filter(id__in=payload["rules"]))
+            # get the rules from the payload
+            rules = tuple(Rule.objects.filter(id__in=payload["rules"]))
 
-        # pull the event based off our payload
-        event = eventstore.backend.get_event_by_id(group.project_id, payload["eventId"])
-        if event is None:
-            logger.info(
-                "msteams.action.event-missing",
-                extra={
-                    "team_id": team_id,
-                    "integration_id": integration.id,
-                    "organization_id": group.organization.id,
-                    "event_id": payload["eventId"],
-                    "project_id": group.project_id,
-                },
-            )
-            return self.respond(status=404)
+            # pull the event based off our payload
+            event = eventstore.backend.get_event_by_id(group.project_id, payload["eventId"])
+            if event is None:
+                logger.info(
+                    "msteams.action.event-missing",
+                    extra={
+                        "team_id": team_id,
+                        "integration_id": integration.id,
+                        "organization_id": group.organization.id,
+                        "event_id": payload["eventId"],
+                        "project_id": group.project_id,
+                    },
+                )
+                return self.respond(status=404)
 
-        # refresh issue and update card
-        group.refresh_from_db()
-        card = MSTeamsIssueMessageBuilder(group, event, rules, integration).build_group_card()
-        client.update_card(conversation_id, activity_id, card)
+            # refresh issue and update card
+            group.refresh_from_db()
+            card = MSTeamsIssueMessageBuilder(group, event, rules, integration).build_group_card()
+            client.update_card(conversation_id, activity_id, card)
 
-        return issue_change_response
+            return issue_change_response
 
     def _handle_channel_message(self, request: Request) -> Response:
         data = request.data

src/sentry/integrations/slack/unfurl/explore.py

@@ -4,7 +4,7 @@
 import logging
 import re
 from collections.abc import Mapping
-from typing import Any
+from typing import Any, TypedDict
 from urllib.parse import urlparse
 
 from django.http.request import QueryDict
@@ -25,6 +25,7 @@
 from sentry.integrations.slack.unfurl.types import Handler, UnfurlableUrl, UnfurledUrl
 from sentry.models.apikey import ApiKey
 from sentry.models.organization import Organization
+from sentry.search.eap.types import SupportedTraceItemType
 from sentry.snuba.referrer import Referrer
 from sentry.users.models.user import User
 from sentry.users.services.user import RpcUser
@@ -33,10 +34,40 @@
 _logger = logging.getLogger(__name__)
 
 DEFAULT_PERIOD = "14d"
-DEFAULT_Y_AXIS = "count(span.duration)"
 TOP_N = 5
 
 
+class ExploreDatasetDefaults(TypedDict):
+    title: str
+    y_axis: str
+
+
+EXPLORE_DATASET_DEFAULTS: dict[SupportedTraceItemType, ExploreDatasetDefaults] = {
+    SupportedTraceItemType.SPANS: {
+        "title": "Explore Traces",
+        "y_axis": "count(span.duration)",
+    },
+    SupportedTraceItemType.LOGS: {
+        "title": "Explore Logs",
+        "y_axis": "count(message)",
+    },
+}
+
+
+def _get_explore_dataset_defaults(dataset: SupportedTraceItemType) -> ExploreDatasetDefaults:
+    """Returns the default title and y_axis for the given explore dataset."""
+    return EXPLORE_DATASET_DEFAULTS.get(
+        dataset, EXPLORE_DATASET_DEFAULTS[SupportedTraceItemType.SPANS]
+    )
+
+
+def _get_explore_dataset(url: str) -> SupportedTraceItemType:
+    """Returns the dataset based on the explore URL."""
+    if explore_logs_link_regex.match(url) or customer_domain_explore_logs_link_regex.match(url):
+        return SupportedTraceItemType.LOGS
+    return SupportedTraceItemType.SPANS
+
+
 def unfurl_explore(
     integration: Integration | RpcIntegration,
     links: list[UnfurlableUrl],
@@ -83,9 +114,12 @@ def _unfurl_explore(
         params = link.args["query"]
         chart_type = link.args.get("chart_type")
 
+        explore_dataset = link.args.get("dataset", SupportedTraceItemType.SPANS)
+        defaults = _get_explore_dataset_defaults(explore_dataset)
+
         y_axes = params.getlist("yAxis")
         if not y_axes:
-            y_axes = [DEFAULT_Y_AXIS]
+            y_axes = [defaults["y_axis"]]
             params.setlist("yAxis", y_axes)
 
         group_bys = params.getlist("groupBy")
@@ -97,7 +131,7 @@ def _unfurl_explore(
         if not params.get("statsPeriod") and not params.get("start"):
             params["statsPeriod"] = DEFAULT_PERIOD
 
-        params["dataset"] = "spans"
+        params["dataset"] = explore_dataset.value
         params["referrer"] = Referrer.EXPLORE_SLACK_UNFURL.value
 
         try:
@@ -124,7 +158,7 @@ def _unfurl_explore(
             continue
 
         unfurls[link.url] = SlackDiscoverMessageBuilder(
-            title="Explore Traces",
+            title=defaults["title"],
             chart_url=url,
         ).build()
 
@@ -158,6 +192,8 @@ def map_explore_query_args(url: str, args: Mapping[str, str | None]) -> Mapping[
     parsed_url = urlparse(url)
     raw_query = QueryDict(parsed_url.query)
 
+    explore_dataset = _get_explore_dataset(url)
+
     # Parse visualize (spans explore) or aggregateField (logs explore) JSON params
     visualize_fields = raw_query.getlist("visualize") or raw_query.getlist("aggregateField")
     y_axes: list[str] = []
@@ -176,7 +212,7 @@ def map_explore_query_args(url: str, args: Mapping[str, str | None]) -> Mapping[
             continue
 
     if not y_axes:
-        y_axes = [DEFAULT_Y_AXIS]
+        y_axes = [_get_explore_dataset_defaults(explore_dataset)["y_axis"]]
 
     # Build query params
     query = QueryDict(mutable=True)
@@ -186,12 +222,12 @@ def map_explore_query_args(url: str, args: Mapping[str, str | None]) -> Mapping[
         query.setlist("groupBy", group_bys)
 
     # Copy standard params
-    for param in ("project", "statsPeriod", "start", "end", "query", "environment"):
+    for param in ("project", "statsPeriod", "start", "end", "query", "environment", "interval"):
         values = raw_query.getlist(param)
         if values:
             query.setlist(param, values)
 
-    return dict(**args, query=query, chart_type=chart_type)
+    return dict(**args, query=query, chart_type=chart_type, dataset=explore_dataset)
 
 
 explore_traces_link_regex = re.compile(
@@ -202,11 +238,21 @@ def map_explore_query_args(url: str, args: Mapping[str, str | None]) -> Mapping[
     r"^https?\://(?P<org_slug>[^.]+?)\.(?#url_prefix)[^/]+/explore/traces/"
 )
 
+explore_logs_link_regex = re.compile(
+    r"^https?\://(?#url_prefix)[^/]+/organizations/(?P<org_slug>[^/]+)/explore/logs/"
+)
+
+customer_domain_explore_logs_link_regex = re.compile(
+    r"^https?\://(?P<org_slug>[^.]+?)\.(?#url_prefix)[^/]+/explore/logs/"
+)
+
 explore_handler = Handler(
     fn=unfurl_explore,
     matcher=[
         explore_traces_link_regex,
         customer_domain_explore_traces_link_regex,
+        explore_logs_link_regex,
+        customer_domain_explore_logs_link_regex,
     ],
     arg_mapper=map_explore_query_args,
 )

src/sentry/runner/commands/run.py

@@ -135,11 +135,20 @@ def taskworker_scheduler(redis_cluster: str, **options: Any) -> None:
 
 
 @run.command()
+@click.option(
+    "--push-mode", help="Whether to run in PUSH or PULL mode.", default=False, is_flag=True
+)
 @click.option(
     "--rpc-host",
-    help="The hostname and port for the taskworker-rpc. When using num-brokers the hostname will be appended with `-{i}` to connect to individual brokers.",
+    help="The hostname and port for the taskbroker gRPC server. When using num-brokers the hostname will be appended with `-{i}` to connect to individual brokers.",
     default="127.0.0.1:50051",
 )
+@click.option(
+    "--worker-rpc-port",
+    help="Port for the taskworker gRPC server to listen on when it is running in push mode.",
+    default=50052,
+    type=int,
+)
 @click.option(
     "--num-brokers", help="Number of brokers available to connect to", default=None, type=int
 )
@@ -198,6 +207,8 @@ def taskworker(**options: Any) -> None:
 
 
 def run_taskworker(
+    push_mode: bool,
+    worker_rpc_port: int,
     rpc_host: str,
     num_brokers: int | None,
     rpc_host_list: str | None,
@@ -233,6 +244,8 @@ def run_taskworker(
             processing_pool_name=processing_pool_name,
             health_check_file_path=health_check_file_path,
             health_check_sec_per_touch=health_check_sec_per_touch,
+            grpc_port=worker_rpc_port,
+            push_mode=push_mode,
             **options,
         )
         exitcode = worker.start()

src/sentry/taskworker/adapters.py

@@ -59,6 +59,29 @@ def incr(
             sample_rate = settings.SENTRY_METRICS_SAMPLE_RATE
         sentry_metrics.incr(name, amount=int(value), tags=tags, sample_rate=sample_rate)
 
+    def gauge(
+        self,
+        key: str,
+        value: float,
+        instance: str | None = None,
+        tags: Tags | None = None,
+        sample_rate: float | None = None,
+        unit: str | None = None,
+        stacklevel: int = 0,
+    ) -> None:
+        if sample_rate is None:
+            sample_rate = settings.SENTRY_METRICS_SAMPLE_RATE
+
+        return sentry_metrics.gauge(
+            key,
+            value,
+            instance=instance,
+            tags=tags,
+            sample_rate=sample_rate,
+            unit=unit,
+            stacklevel=stacklevel,
+        )
+
     def distribution(
         self,
         name: str,