[Tracing] Emit grouping events (#4925)

Instrumenting grouper code in order to emit grouping events. These
events are only emitted for the testcases being moved to another group
(or to a newly formed one).

I.e., when a pair of testcases is considered similar:
* If both testcases are not grouped yet, an event is emitted for each
one.
* If only one testcase is already assigned to a group, the event is
emitted for the testcase being added to the group.
* If both testcases are in groups, an event is emitted for the one being
moved to the other group. Then for all the remaining testcases that are
also moved due to the group merge.

There is also the case of a testcase being ungrouped because its group
only contains one testcase. This also triggers an event since it is
moving the testcase to an ungrouped state.


Rational at: b/394051778

Author: ViniciustCosta

src/clusterfuzz/_internal/cron/grouper.py

@@ -73,6 +73,30 @@ def get_metadata(self, key=None, default=None):
     return getattr(self, key, default)
 
 
+def _emit_grouping_event(moved_testcase: int,
+                         new_group_id: int,
+                         prev_group_id: int,
+                         similar_testcase: int | None,
+                         reason: str,
+                         group_merge: bool = False):
+  """Helper for emitting a testcase grouping event."""
+  # If this is due to a group merge, we have to use the grouping reason as the
+  # reason for the merge itself.
+  group_merge_reason = None
+  if group_merge:
+    group_merge_reason = reason
+    reason = events.GroupingReason.GROUP_MERGE
+
+  events.emit(
+      events.TestcaseGroupingEvent(
+          testcase_id=moved_testcase,
+          group_id=new_group_id,
+          previous_group_id=prev_group_id,
+          similar_testcase_id=similar_testcase,
+          grouping_reason=reason,
+          group_merge_reason=group_merge_reason))
+
+
 def combine_testcases_into_group(
     testcase_1: TestcaseAttributes, testcase_2: TestcaseAttributes,
     testcase_map: dict[int, TestcaseAttributes], reason: str) -> None:
@@ -93,26 +117,49 @@ def combine_testcases_into_group(
     new_group_id = _get_new_group_id()
     testcase_1.group_id = new_group_id
     testcase_2.group_id = new_group_id
+    # Both testcases are moved, so emit an event for each.
+    _emit_grouping_event(testcase_1.id, new_group_id, 0, testcase_2.id, reason)
+    _emit_grouping_event(testcase_2.id, new_group_id, 0, testcase_1.id, reason)
     return
 
   # If one of the testcase has a group id, then assign the other to reuse that
   # group id.
   if testcase_1.group_id and not testcase_2.group_id:
     testcase_2.group_id = testcase_1.group_id
+    # Only emit event for moved testcase_2.
+    _emit_grouping_event(testcase_2.id, testcase_1.group_id, 0, testcase_1.id,
+                         reason)
     return
+
   if testcase_2.group_id and not testcase_1.group_id:
     testcase_1.group_id = testcase_2.group_id
+    # Only emit event for moved testcase_1.
+    _emit_grouping_event(testcase_1.id, testcase_2.group_id, 0, testcase_2.id,
+                         reason)
     return
 
   # If both the testcase have their own groups, then just merge the two groups
   # together and reuse one of their group ids.
   group_id_to_reuse = testcase_1.group_id
   group_id_to_move = testcase_2.group_id
+  # Emit event for testcase from group to be moved.
+  _emit_grouping_event(testcase_2.id, testcase_1.group_id, testcase_2.group_id,
+                       testcase_1.id, reason)
+
   moved_testcase_ids = []
   for testcase in testcase_map.values():
     if testcase.group_id == group_id_to_move:
       testcase.group_id = group_id_to_reuse
       moved_testcase_ids.append(str(testcase.id))
+      if testcase.id != testcase_2.id:
+        # Emit event for each testcase moved due to the group merge.
+        _emit_grouping_event(
+            testcase.id,
+            group_id_to_reuse,
+            group_id_to_move,
+            testcase_2.id,
+            reason,
+            group_merge=True)
 
   logs.info(f'Merged group {group_id_to_move} into {group_id_to_reuse}: ' +
             'moved testcases: ' + ', '.join(moved_testcase_ids))
@@ -270,7 +317,7 @@ def _group_testcases_based_on_variants(testcase_map):
           continue
 
         combine_testcases_into_group(testcase_1, testcase_2, testcase_map,
-                                     'identical variant')
+                                     events.GroupingReason.IDENTICAL_VARIANT)
 
 
 def _group_testcases_with_same_issues(testcase_map):
@@ -301,7 +348,7 @@ def _group_testcases_with_same_issues(testcase_map):
           continue
 
         combine_testcases_into_group(testcase_1, testcase_2, testcase_map,
-                                     'same issue')
+                                     events.GroupingReason.SAME_ISSUE)
 
 
 def _compare_testcases_crash_states(testcase_1, testcase_2) -> bool:
@@ -362,7 +409,7 @@ def _group_testcases_with_similar_states(testcase_map):
           continue
 
         combine_testcases_into_group(testcase_1, testcase_2, testcase_map,
-                                     'similar crashes')
+                                     events.GroupingReason.SIMILAR_CRASH)
 
 
 def _has_testcase_with_same_params(testcase, testcase_map):
@@ -603,6 +650,8 @@ def group_testcases():
         data_handler.delete_group(updated_group_id, update_testcases=False)
         logs.info(
             f'Deleted group {updated_group_id} used by only one testcase.')
+        _emit_grouping_event(testcase_id, 0, updated_group_id, None,
+                             events.GroupingReason.UNGROUPED)
         updated_group_id = 0
         updated_group_bug_information = 0
         updated_is_leader = True
@@ -635,8 +684,11 @@ def group_testcases():
 
 @logs.cron_log_context()
 def main():
-  """Group testcases (this will be used to run grouper as a standalone cron in
-  dev/staging environments)."""
+  """Group testcases.
+  
+  This will be useful to run grouper as a standalone cron in dev/staging
+  environments.
+  """
   try:
     logs.info('Grouping testcases.')
     group_testcases()

src/clusterfuzz/_internal/metrics/events.py

@@ -106,6 +106,7 @@ class GroupingReason:
   SAME_ISSUE = 'same_issue'
   IDENTICAL_VARIANT = 'identical_variant'
   GROUP_MERGE = 'group_merge'
+  UNGROUPED = 'ungrouped'
 
 
 @dataclass(kw_only=True)

src/clusterfuzz/_internal/tests/appengine/handlers/cron/grouper_test.py

@@ -39,8 +39,10 @@ def setUp(self):
 
     helpers.patch(self, [
         'clusterfuzz._internal.cron.cleanup.get_top_crashes_for_all_projects_and_platforms',
+        'clusterfuzz._internal.metrics.events.emit',
+        'clusterfuzz._internal.metrics.events._get_datetime_now',
     ])
-
+    self.mock._get_datetime_now.return_value = datetime.datetime(2025, 1, 1)  # pylint: disable=protected-access
     self.mock.get_top_crashes_for_all_projects_and_platforms.return_value = {
         'blah': {},
         'project1': {
@@ -77,6 +79,7 @@ def test_same_crash_different_security(self):
       self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
       self.assertEqual(self.testcases[index].group_id, 0)
       self.assertTrue(self.testcases[index].is_leader)
+    self.mock.emit.assert_not_called()
 
   def test_same_crash_same_security(self):
     """Test that crashes with same crash states and same security flags get
@@ -139,6 +142,7 @@ def test_different_crash_same_security(self):
       self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
       self.assertEqual(self.testcases[index].group_id, 0)
       self.assertTrue(self.testcases[index].is_leader)
+    self.mock.emit.assert_not_called()
 
   def test_group_of_one(self):
     """Test that a group id with just one testcase gets removed."""
@@ -152,6 +156,15 @@ def test_group_of_one(self):
     self.assertEqual(testcase.group_id, 0)
     self.assertTrue(testcase.is_leader)
 
+    self.mock.emit.assert_called_once_with(
+        events.TestcaseGroupingEvent(
+            testcase_id=testcase.key.id(),
+            group_id=0,
+            previous_group_id=1,
+            similar_testcase_id=None,
+            grouping_reason=events.GroupingReason.UNGROUPED,
+            group_merge_reason=None))
+
   def test_same_unique_crash_type_with_same_state(self):
     """Test that the crashes with same unique crash type and same state get
     de-duplicated with one of them removed.."""
@@ -196,6 +209,7 @@ def test_same_unique_crash_type_with_different_state(self):
       self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
       self.assertEqual(self.testcases[index].group_id, 0)
       self.assertTrue(self.testcases[index].is_leader)
+    self.mock.emit.assert_not_called()
 
   def test_different_unique_crash_type_with_same_state(self):
     """Test that the crashes with different unique crash type but same state
@@ -316,6 +330,7 @@ def test_top_crasher_for_variant_analysis(self):
     for testcase in self.testcases:
       self.assertEqual(testcase.group_id, 0)
       self.assertTrue(testcase.is_leader)
+    self.mock.emit.assert_not_called()
 
   def test_same_job_type_for_variant_analysis(self):
     """Tests that testcases with the same job_type don't get grouped together"""
@@ -404,6 +419,23 @@ def test_similar_variants_for_variant_analysis(self):
       self.assertEqual(self.testcases[i].group_id, 0)
       self.assertTrue(self.testcases[i].is_leader)
 
+    self.mock.emit.assert_any_call(
+        events.TestcaseGroupingEvent(
+            testcase_id=self.testcases[0].key.id(),
+            group_id=self.testcases[0].group_id,
+            previous_group_id=0,
+            similar_testcase_id=self.testcases[1].key.id(),
+            grouping_reason=events.GroupingReason.IDENTICAL_VARIANT,
+            group_merge_reason=None))
+    self.mock.emit.assert_any_call(
+        events.TestcaseGroupingEvent(
+            testcase_id=self.testcases[1].key.id(),
+            group_id=self.testcases[1].group_id,
+            previous_group_id=0,
+            similar_testcase_id=self.testcases[0].key.id(),
+            grouping_reason=events.GroupingReason.IDENTICAL_VARIANT,
+            group_merge_reason=None))
+
   def test_similar_but_anomalous_variants_for_variant_analysis(self):
     """Tests that testcases with similar variants but anomalous do not
     get deduplicated. Anomalous variant matches with more than threshold
@@ -476,6 +508,7 @@ def test_similar_but_anomalous_variants_for_variant_analysis(self):
       self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
       self.assertEqual(self.testcases[index].group_id, 0)
       self.assertTrue(self.testcases[index].is_leader)
+    self.mock.emit.assert_not_called()
 
   def test_no_reproducible_for_variant_analysis(self):
     """Tests that no-reproducible testcases with similar variants do not
@@ -515,6 +548,7 @@ def test_no_reproducible_for_variant_analysis(self):
       self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
       self.assertEqual(self.testcases[index].group_id, 0)
       self.assertTrue(self.testcases[index].is_leader)
+    self.mock.emit.assert_not_called()
 
   def test_ignored_crash_type_for_variant_analysis(self):
     """Tests that testcases of ignored crash type with similar variants
@@ -555,6 +589,143 @@ def test_ignored_crash_type_for_variant_analysis(self):
       self.assertEqual(self.testcases[index].group_id, 0)
       self.assertTrue(self.testcases[index].is_leader)
 
+  def test_grouping_event_new_group(self):
+    """Test correct grouping event for a newly formed group."""
+    self.testcases[0].security_flag = True
+    self.testcases[0].crash_state = 'abcdef'
+    self.testcases[1].security_flag = True
+    self.testcases[1].crash_state = 'abcde'
+
+    for t in self.testcases:
+      t.put()
+
+    grouper.group_testcases()
+
+    for index, t in enumerate(self.testcases):
+      self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
+
+    # Check testcases 0 and 1 are grouped together.
+    self.assertNotEqual(self.testcases[0].group_id, 0)
+    self.assertNotEqual(self.testcases[1].group_id, 0)
+    self.assertEqual(self.testcases[0].group_id, self.testcases[1].group_id)
+
+    self.mock.emit.assert_any_call(
+        events.TestcaseGroupingEvent(
+            testcase_id=self.testcases[0].key.id(),
+            group_id=self.testcases[0].group_id,
+            previous_group_id=0,
+            similar_testcase_id=self.testcases[1].key.id(),
+            grouping_reason=events.GroupingReason.SIMILAR_CRASH,
+            group_merge_reason=None))
+    self.mock.emit.assert_any_call(
+        events.TestcaseGroupingEvent(
+            testcase_id=self.testcases[1].key.id(),
+            group_id=self.testcases[1].group_id,
+            previous_group_id=0,
+            similar_testcase_id=self.testcases[0].key.id(),
+            grouping_reason=events.GroupingReason.SIMILAR_CRASH,
+            group_merge_reason=None))
+
+  def test_grouping_event_assign_one_testcase(self):
+    """Test correct grouping event for new testcase assigned to a group."""
+    self.testcases[0].security_flag = True
+    self.testcases[0].crash_state = 'abcdef'
+    self.testcases[0].group_id = 1
+    self.testcases[1].security_flag = True
+    self.testcases[1].crash_state = 'abcde'
+    self.testcases[1].group_id = 0
+
+    for t in self.testcases:
+      t.put()
+
+    grouper.group_testcases()
+
+    for index, t in enumerate(self.testcases):
+      self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
+
+    # Check testcases 0 and 1 are grouped together.
+    self.assertEqual(self.testcases[0].group_id, 1)
+    self.assertEqual(self.testcases[1].group_id, 1)
+
+    self.mock.emit.assert_called_once_with(
+        events.TestcaseGroupingEvent(
+            testcase_id=self.testcases[1].key.id(),
+            group_id=1,
+            previous_group_id=0,
+            similar_testcase_id=self.testcases[0].key.id(),
+            grouping_reason=events.GroupingReason.SIMILAR_CRASH,
+            group_merge_reason=None))
+
+  def test_grouping_event_merge_groups(self):
+    """Test correct grouping event for merging groups."""
+    self.testcases[0].security_flag = True
+    self.testcases[0].crash_state = 'abcdef'
+    self.testcases[0].group_id = 1
+    self.testcases[1].security_flag = True
+    self.testcases[1].crash_state = 'ghijk'
+    self.testcases[1].group_id = 1
+
+    self.testcases.append(test_utils.create_generic_testcase())
+    self.testcases[2].security_flag = True
+    self.testcases[2].crash_state = 'abcde'
+    self.testcases[2].group_id = 2
+    self.testcases.append(test_utils.create_generic_testcase())
+    self.testcases[3].security_flag = True
+    self.testcases[3].crash_state = 'lmnopq'
+    self.testcases[3].group_id = 2
+
+    for t in self.testcases:
+      t.put()
+
+    grouper.group_testcases()
+
+    for index, t in enumerate(self.testcases):
+      self.testcases[index] = data_handler.get_testcase_by_id(t.key.id())
+
+    common_group_id = self.testcases[0].group_id
+    self.assertIn(common_group_id, [1, 2])
+    # Check testcases are all grouped together.
+    self.assertEqual(self.testcases[0].group_id, common_group_id)
+    self.assertEqual(self.testcases[1].group_id, common_group_id)
+    self.assertEqual(self.testcases[2].group_id, common_group_id)
+    self.assertEqual(self.testcases[3].group_id, common_group_id)
+
+    # Avoid possibly flaky test by ensuring which group was kept during merge.
+    if common_group_id == 1:
+      self.mock.emit.assert_any_call(
+          events.TestcaseGroupingEvent(
+              testcase_id=self.testcases[2].key.id(),
+              group_id=1,
+              previous_group_id=2,
+              similar_testcase_id=self.testcases[0].key.id(),
+              grouping_reason=events.GroupingReason.SIMILAR_CRASH,
+              group_merge_reason=None))
+      self.mock.emit.assert_any_call(
+          events.TestcaseGroupingEvent(
+              testcase_id=self.testcases[3].key.id(),
+              group_id=1,
+              previous_group_id=2,
+              similar_testcase_id=self.testcases[2].key.id(),
+              grouping_reason=events.GroupingReason.GROUP_MERGE,
+              group_merge_reason=events.GroupingReason.SIMILAR_CRASH))
+    else:
+      self.mock.emit.assert_any_call(
+          events.TestcaseGroupingEvent(
+              testcase_id=self.testcases[0].key.id(),
+              group_id=2,
+              previous_group_id=1,
+              similar_testcase_id=self.testcases[2].key.id(),
+              grouping_reason=events.GroupingReason.SIMILAR_CRASH,
+              group_merge_reason=None))
+      self.mock.emit.assert_any_call(
+          events.TestcaseGroupingEvent(
+              testcase_id=self.testcases[1].key.id(),
+              group_id=2,
+              previous_group_id=1,
+              similar_testcase_id=self.testcases[0].key.id(),
+              grouping_reason=events.GroupingReason.GROUP_MERGE,
+              group_merge_reason=events.GroupingReason.SIMILAR_CRASH))
+
 
 @test_utils.with_cloud_emulators('datastore')
 class GroupExceedMaxTestcasesTest(unittest.TestCase):
@@ -644,8 +815,12 @@ def test_group_overflow_rejection_events(self):
 
     grouper.group_testcases()
 
-    self.assertEqual(5, len(self.emitted_events))
+    # Expected: 30 grouping events + 5 rejection events
+    self.assertEqual(35, len(self.emitted_events))
+    count_rejection_events = 0
     for event in self.emitted_events:
-      self.assertEqual(event.rejection_reason,
-                       events.RejectionReason.GROUPER_OVERFLOW)
-    self.assertEqual(5, self.mock.emit.call_count)
+      if event.event_type == events.EventTypes.TESTCASE_REJECTION:
+        self.assertEqual(event.rejection_reason,
+                         events.RejectionReason.GROUPER_OVERFLOW)
+        count_rejection_events += 1
+    self.assertEqual(5, count_rejection_events)