feat(aci milestone 3): call process_data_sources in subscription processor

[mifu67]

Wrap the subscription update in a DataPacket and call process_data_sources, then feed the results into process_detectors and log the result (process_detectors calls a dummy evaluate function for now). Step one of the pipeline.

[codecov]

Codecov Report

Attention: Patch coverage is 78.78788% with 7 lines in your changes missing coverage. Please review.

✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
src/sentry/incidents/grouptype.py	65.00%	7 Missing ⚠️

Additional details and impacted files

@@           Coverage Diff            @@
##           master   #81788    +/-   ##
========================================
  Coverage   80.35%   80.36%            
========================================
  Files        7252     7257     +5     
  Lines      320551   320812   +261     
  Branches    20859    20859            
========================================
+ Hits       257577   257812   +235     
- Misses      62579    62605    +26     
  Partials      395      395            

[saponifi3d]

🤔 - i was intending the query_type to be more of the DataSource, so this would be like snuba_subscription or something along those lines. maybe we should have those exported from data_sources as an enum and applied to the type field.

[mifu67]

I think Colleen is writing an enum for this purpose!

[mifu67]

TODO: change query_type to be DataSourceType.SNUBA_QUERY_SUBSCRIPTION once Colleen's PR lands

[mifu67]

This is pretty messy. Any suggestions for how we can better log the data flowing through the pipeline?

[saponifi3d]

I'm adding centralized logging to the process_* methods, so we shouldn't need to add much in the way of metrics / logging here. It will all be based off of the types set in each of those flows. We should only have to worry about anything that's specific to metric alerts implementation here.

If we make the change to have process_data_packets as i was talking about before, you would just have to verify that process_data_packets is being invoked correctly in different states, then you can assume the platform would be handling the rest. (yay #platforms)

If you wanted to test this stuff w/o that method, it may want to be like different snuba query results with a configured detector, and how does that invoke process_data_sources / process_detectors differently.

[saponifi3d]

Might be nice for this to be a little more complete for tests by adding the occurrence and event data,

occurrence, event_data = self.build_occurrence_and_event_data("dummy_group", 0, PriorityLevel.HIGH)

and then update the result / event_data,

Suggested change

	result=None,
	event_data=None,
	result=occurrence,
	event_data=event_data,

[saponifi3d]

should we be overriding the evaluate method from a stateful detector? just looking a little closer at the base class, it looks like this would call through to check the detectors conditions and that should all work from the base if the models are constructed. (lemme double check with dan on some of his intentions on code in this area though)

[saponifi3d]

i was thinking of adding another process method that wraps process_data_sources and process_detectors and then returns a list of the results.

That would turn this code into something along these lines:

Suggested change

	detectors = process_data_sources([data_packet], query_type=self.subscription.type)
	results = []
	for data_packet, detectors in detectors:
	results.append(process_detectors(data_packet, detectors))
	# NOTE: this is temporary, to verify in the tests that the right information is flowing through the pipeline.
	results = process_data_packets([data_packet], query_type=self.subscription.type)

Inside of process_data_packets i was thinking we could add centralize metrics / logging there. Whatcha think?

[mifu67]

I think this is an excellent idea.

[saponifi3d]

I'm adding centralized logging to the process_* methods, so we shouldn't need to add much in the way of metrics / logging here. It will all be based off of the types set in each of those flows. We should only have to worry about anything that's specific to metric alerts implementation here.

If we make the change to have process_data_packets as i was talking about before, you would just have to verify that process_data_packets is being invoked correctly in different states, then you can assume the platform would be handling the rest. (yay #platforms)

If you wanted to test this stuff w/o that method, it may want to be like different snuba query results with a configured detector, and how does that invoke process_data_sources / process_detectors differently.

[mifu67]

Closing because the rebase is too complicated and there aren't that many lines anyway—I'll start a new PR instead.