feat(subscriptions): Use RPC in subscriptions pipeline #6499

shruthilayaj · 2024-11-04T12:00:12Z

Updates the subscription pipeline to support RPCSubscriptionData.

codecov · 2024-11-04T12:12:31Z

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
2535	1	2534	5

View the top 1 failed tests by shortest run time

tests.subscriptions.test_subscription.TestEAPSpansRPCSubscriptionCreator::test[EAP spans RPC subscription]

Stack Traces | 0.184s run time

Traceback (most recent call last):
  File ".../tests/subscriptions/test_subscription.py", line 397, in test
    identifier = creator.create(subscription, self.timer)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File ".../snuba/subscriptions/subscription.py", line 36, in create
    data.validate()
  File ".../snuba/subscriptions/data.py", line 189, in validate
    raise InvalidSubscriptionError("Multiple project IDs not supported.")
snuba.datasets.entity_subscriptions.validators.InvalidSubscriptionError: Multiple project IDs not supported.

To view more test analytics, go to the Test Analytics Dashboard
Got feedback? Let us know on Github

…oint

shruthilayaj · 2024-11-08T19:58:03Z

snuba/subscriptions/data.py

+        rounded_ts = (
+            int(timestamp.replace(tzinfo=UTC).timestamp() / self.time_window_sec)
+            * self.time_window_sec
+        )


hack for now, we want to round this to the lowest available granularity (15s) more context here

shruthilayaj · 2024-11-08T20:14:57Z

snuba/subscriptions/data.py

+        if (self.request_name, self.request_version) not in REQUEST_TYPE_ALLOWLIST:
+            raise InvalidSubscriptionError(
+                f"{self.request_name} {self.request_version} not supported."
+            )
+
+        # TODO: Validate no group by, having, order by etc


I'll add more validation in a follow up so it's easier to review

volokluev

Why can the create subscription RPC call not be its own PR?

shruthilayaj · 2024-11-08T21:37:55Z

Why can the create subscription RPC call not be its own PR?

@volokluev Creating a subscription requires validation, so I needed to build and run the request before storing it. That introduces the RPCSubscriptionData class. And basically once I introduced it in one place, I kinda had to follow all the typing issues and handle it everywhere.

…oint

shruthilayaj · 2024-11-21T12:45:55Z

snuba/subscriptions/data.py

@@ -95,7 +104,7 @@ def from_string(cls, value: str) -> SubscriptionIdentifier:


 @dataclass(frozen=True, kw_only=True)
-class SubscriptionData(ABC):
+class _SubscriptionData(ABC, Generic[TRequest]):


build_request returns generic type TRequest and run_query accepts TRequest.
So for SnQLSubscriptionData, build_request returns Request and run_query accepts Request
and for RPCSubscriptionData, build_request returns TimeSeriesRequest and run_query accepts TimeSeriesRequest

…oint

volokluev

I have some cleanup comments but overall this is good to go

volokluev · 2024-11-22T19:15:47Z

snuba/subscriptions/data.py

+        # TODO: update it to round to the lowest granularity
+        # rounded_ts = int(timestamp.replace(tzinfo=UTC).timestamp() / 15) * 15


is this TODO still relevant?

Yup, waiting on https://github.com/getsentry/projects/issues/364 to be finished

volokluev · 2024-11-25T19:25:18Z

tests/subscriptions/test_scheduler_consumer.py

+        entity_key,
+        PartitionId(partition_index),
+    )
+    entity = get_entity(EntityKey.EVENTS)


why is this line here

oops, bad copy paste

volokluev · 2024-11-25T19:26:01Z

tests/subscriptions/test_scheduler_consumer.py

+        PartitionId(partition_index),
+    )
+    entity = get_entity(EntityKey.EVENTS)
+    store.create(


in your tests, you should be creating subscriptions through the api call, not the implementation detail as you are doing here

addressed here a642b1b#diff-599503c09e9756bcc58fc90a4f840f7241091b3c7cf3a270d698808e122a2107R183

volokluev · 2024-11-25T19:38:54Z

tests/subscriptions/test_subscription.py

+    def setup_method(self) -> None:
+        self.dataset = get_dataset("metrics")


volokluev · 2024-11-25T20:07:03Z

tests/subscriptions/test_filter_subscriptions.py

+
+
+def build_rpc_subscription(resolution: timedelta, org_id: int) -> Subscription:
+    return Subscription(
+        SubscriptionIdentifier(PartitionId(1), uuid.uuid4()),
+        RPCSubscriptionData.from_proto(
+            CreateSubscriptionRequestProto(
+                time_series_request=TimeSeriesRequest(
+                    meta=RequestMeta(
+                        project_ids=[1],
+                        organization_id=org_id,
+                        cogs_category="something",
+                        referrer="something",
+                    ),
+                    aggregations=[
+                        AttributeAggregation(
+                            aggregate=Function.FUNCTION_SUM,
+                            key=AttributeKey(
+                                type=AttributeKey.TYPE_FLOAT, name="test_metric"
+                            ),
+                            label="sum",
+                            extrapolation_mode=ExtrapolationMode.EXTRAPOLATION_MODE_SAMPLE_WEIGHTED,
+                        ),
+                    ],
+                    filter=TraceItemFilter(
+                        comparison_filter=ComparisonFilter(
+                            key=AttributeKey(type=AttributeKey.TYPE_STRING, name="foo"),
+                            op=ComparisonFilter.OP_NOT_EQUALS,
+                            value=AttributeValue(val_str="bar"),
+                        )
+                    ),
+                ),
+                time_window_secs=300,
+                resolution_secs=int(resolution.total_seconds()),
+            ),
+            EntityKey.EAP_SPANS,
+        ),
+    )
+
+
+@pytest.fixture
+def expected_rpc_subs() -> MutableSequence[Subscription]:
+    return [
+        build_rpc_subscription(timedelta(minutes=1), 2)
+        for count in range(randint(1, 50))
+    ]
+
+
+@pytest.fixture
+def extra_rpc_subs() -> MutableSequence[Subscription]:
+    return [
+        build_rpc_subscription(timedelta(minutes=3), 1)
+        for count in range(randint(1, 50))
+    ]
+
+
+@patch("snuba.settings.SLICED_STORAGE_SETS", {"events_analytics_platform": 3})
+@patch(
+    "snuba.settings.LOGICAL_PARTITION_MAPPING",
+    {"events_analytics_platform": {0: 0, 1: 1, 2: 2}},
+)
+def test_filter_rpc_subscriptions(expected_rpc_subs, extra_rpc_subs) -> None:  # type: ignore
+    importlib.reload(scheduler)
+
+    filtered_subs = filter_subscriptions(
+        subscriptions=expected_rpc_subs + extra_rpc_subs,
+        entity_key=EntityKey.EAP_SPANS,
+        metrics=DummyMetricsBackend(strict=True),
+        slice_id=2,
+    )
+    assert filtered_subs == expected_rpc_subs


this test is not necessary.

eap_spans is not a sliced storage

this is far in the weeds of the implementation details of the subscription scheduler

volokluev · 2024-11-25T20:16:43Z

snuba/datasets/slicing.py

@@ -3,6 +3,7 @@
 should be stored. These do not require individual physical partitions but allow
 for repartitioning with less code changes per physical change.
 """
+


volokluev · 2024-11-25T20:18:14Z

snuba/subscriptions/codecs.py

@@ -98,15 +116,20 @@ def decode(self, value: KafkaPayload) -> ScheduledSubscriptionTask:

        entity_key = EntityKey(scheduled_subscription_dict["entity"])

+        data = scheduled_subscription_dict["task"]["data"]


FYI: the docstring of this class should probably change

a642b1b#diff-f48e31c677eccff7bf14b091fe348640c394684c53c8d3a5a16d27e265d8b1abR87

volokluev · 2024-11-25T22:09:47Z

When you go to merge this, please notify the oncall in #discuss-eng-sns this change is touching a lot of crucial infra. It's good you have done an end to end test locally. We should make sure that it doesn't break things in S4S when it's rolled out

shruthilayaj · 2024-11-25T22:19:08Z

When you go to merge this, please notify the oncall in #discuss-eng-sns this change is touching a lot of crucial infra. It's good you have done an end to end test locally. We should make sure that it doesn't break things in S4S when it's rolled out

Will do, I'll address your comments and merge tomorrow 👍

…oint

shruthilayaj added 2 commits November 4, 2024 06:58

feat(subscriptions): Add create subscriptions RPC

fb123b4

create subscriptions file

0f862e9

shruthilayaj added 11 commits November 4, 2024 08:56

fix typing

0e90ffc

tests

efa8aaa

subscription data test

69cfc87

merge conflicts

01f952c

fix typing

f532342

codec tests

2cd4713

Merge branch 'master' into shruthi/feat/add-create-subscriptions-endp…

6c05e65

…oint

task result encoder

958926d

fix typing

9a8f16e

not deterministic

67372da

fix test

c16e31b

shruthilayaj commented Nov 8, 2024

View reviewed changes

remove breakpoint

d152941

shruthilayaj commented Nov 8, 2024

View reviewed changes

shruthilayaj marked this pull request as ready for review November 8, 2024 21:29

shruthilayaj requested review from a team as code owners November 8, 2024 21:29

volokluev reviewed Nov 8, 2024

View reviewed changes

edwardgou-sentry mentioned this pull request Nov 13, 2024

feat(alerts): Create eap snuba subscription using rpc timeseries request getsentry/sentry#80665

Merged

shruthilayaj added 5 commits November 18, 2024 11:26

feat(subscriptions): Add a create subscription rpc endpoint

76a7f0d

move validate to base class

5726ade

inspect what's stored in redis in the test

66270bf

Merge branch 'master' into shruthi/feat/add-create-subscriptions-endp…

baaa03c

…oint

merge conflicts

7191833

shruthilayaj changed the base branch from master to shruthi/feat/create-rpc-subscriptions November 18, 2024 16:56

shruthilayaj changed the title ~~feat(subscriptions): Add create subscriptions RPC~~ feat(subscriptions): Use RPC in subscriptions pipeline Nov 18, 2024

shruthilayaj added 9 commits November 19, 2024 13:44

serialize what's stored in redis

c99d49c

add a comment

17a49fe

add more validation

83f1693

merge

49e6e9b

fix tests

e7fa5fb

minimize diff

15d29f6

remove rpc helper file

df861f5

update test subscriptions to be valid

356d368

fix typing

baf5566

Base automatically changed from shruthi/feat/create-rpc-subscriptions to master November 20, 2024 18:43

shruthilayaj added 3 commits November 20, 2024 14:11

fix merge conflcits

6e9424a

remove rpc helper file

6321e0c

return none, not 0 when no data is found

f7f8448

shruthilayaj requested a review from volokluev November 21, 2024 12:42

shruthilayaj commented Nov 21, 2024

View reviewed changes

Merge branch 'master' into shruthi/feat/add-create-subscriptions-endp…

f4b81d3

…oint

volokluev approved these changes Nov 25, 2024

View reviewed changes

shruthilayaj added 2 commits November 26, 2024 10:45

Merge branch 'master' into shruthi/feat/add-create-subscriptions-endp…

dc11a7e

…oint

address comments

a642b1b

shruthilayaj merged commit ffa50b5 into master Nov 26, 2024
30 checks passed

shruthilayaj deleted the shruthi/feat/add-create-subscriptions-endpoint branch November 26, 2024 16:56

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(subscriptions): Use RPC in subscriptions pipeline #6499

feat(subscriptions): Use RPC in subscriptions pipeline #6499

shruthilayaj commented Nov 4, 2024 •

edited

Loading

codecov bot commented Nov 4, 2024 •

edited

Loading

shruthilayaj Nov 8, 2024 •

edited

Loading

shruthilayaj Nov 8, 2024

volokluev left a comment

shruthilayaj commented Nov 8, 2024 •

edited

Loading

shruthilayaj Nov 21, 2024

volokluev left a comment

volokluev Nov 22, 2024

shruthilayaj Nov 26, 2024

volokluev Nov 25, 2024

shruthilayaj Nov 26, 2024

volokluev Nov 25, 2024

shruthilayaj Nov 26, 2024

volokluev Nov 25, 2024

volokluev Nov 25, 2024

volokluev Nov 25, 2024

volokluev Nov 25, 2024

shruthilayaj Nov 26, 2024

volokluev commented Nov 25, 2024

shruthilayaj commented Nov 25, 2024

		# TODO: update it to round to the lowest granularity
		# rounded_ts = int(timestamp.replace(tzinfo=UTC).timestamp() / 15) * 15

		def setup_method(self) -> None:
		self.dataset = get_dataset("metrics")

		@@ -98,15 +116,20 @@ def decode(self, value: KafkaPayload) -> ScheduledSubscriptionTask:

		entity_key = EntityKey(scheduled_subscription_dict["entity"])

		data = scheduled_subscription_dict["task"]["data"]

feat(subscriptions): Use RPC in subscriptions pipeline #6499

feat(subscriptions): Use RPC in subscriptions pipeline #6499

Conversation

shruthilayaj commented Nov 4, 2024 • edited Loading

codecov bot commented Nov 4, 2024 • edited Loading

❌ 1 Tests Failed:

shruthilayaj Nov 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

volokluev left a comment

Choose a reason for hiding this comment

shruthilayaj commented Nov 8, 2024 • edited Loading

Choose a reason for hiding this comment

volokluev left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

volokluev commented Nov 25, 2024

shruthilayaj commented Nov 25, 2024

shruthilayaj commented Nov 4, 2024 •

edited

Loading

codecov bot commented Nov 4, 2024 •

edited

Loading

shruthilayaj Nov 8, 2024 •

edited

Loading

shruthilayaj commented Nov 8, 2024 •

edited

Loading