[Tracing] Emiting TestcaseRejectionEvent during grouper

src/clusterfuzz/_internal/cron/grouper.py

@@ -22,6 +22,7 @@
 from clusterfuzz._internal.datastore import data_handler
 from clusterfuzz._internal.datastore import data_types
 from clusterfuzz._internal.issue_management import issue_tracker_utils
+from clusterfuzz._internal.metrics import events
 from clusterfuzz._internal.metrics import logs
 
 from . import cleanup
@@ -416,6 +417,10 @@ def _key_func(testcase):
             ('Deleting testcase {testcase_id} due to overflowing group '
              '{group_id}.').format(
                  testcase_id=testcase.id, group_id=testcase.group_id))
+        events.emit(
+            events.TestcaseRejectionEvent(
+                testcase=testcase_entity,
+                rejection_reason=events.RejectionReason.GROUPER_OVERFLOW))
         testcase_entity.key.delete()
 
 
@@ -426,6 +431,10 @@ def _get_testcase_attributes(testcase, testcase_map, cached_issue_map):
   if (not testcase.bug_information and not testcase.uploader_email and
       _has_testcase_with_same_params(testcase, testcase_map)):
     logs.info('Deleting duplicate testcase %d.' % testcase_id)
+    events.emit(
+        events.TestcaseRejectionEvent(
+            testcase=testcase,
+            rejection_reason=events.RejectionReason.GROUPER_DUPLICATE))
     testcase.key.delete()
     return
 

src/clusterfuzz/_internal/metrics/events.py

@@ -54,6 +54,8 @@ class RejectionReason:
   """Explanation for the testcase rejection values."""
   ANALYZE_NO_REPRO = 'analyze_no_repro'
   ANALYZE_FLAKE_ON_FIRST_ATTEMPT = 'analyze_flake_on_first_attempt'
+  GROUPER_DUPLICATE = 'grouper_duplicate'
+  GROUPER_OVERFLOW = 'grouper_overflow'
 
 
 @dataclass(kw_only=True)

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

@@ -18,6 +18,7 @@
 
 from clusterfuzz._internal.cron import grouper
 from clusterfuzz._internal.datastore import data_handler
+from clusterfuzz._internal.metrics import events
 from clusterfuzz._internal.tests.test_libs import helpers
 from clusterfuzz._internal.tests.test_libs import test_utils
 
@@ -592,3 +593,59 @@ def test_group_exceed_max_testcases(self):
     expected_testcase_ids = [3, 4, 5] + list(range(
         9, 31)) + [unrelated_testcase.key.id()]
     self.assertEqual(expected_testcase_ids, testcase_ids)
+
+
+@test_utils.with_cloud_emulators('datastore')
+class GrouperRejectionEventsTest(unittest.TestCase):
+  """Tests for rejection event emissions in grouper."""
+
+  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 = {}
+
+    self.emitted_events = []
+    self.mock.emit.side_effect = self.emitted_events.append
+
+  def test_duplicate_rejection_event(self):
+    """Test that a duplicate testcase triggers a rejection event."""
+    testcase1 = test_utils.create_generic_testcase()
+    testcase2 = test_utils.create_generic_testcase()
+    testcase1.crash_type = 'Overflow'
+    testcase2.crash_type = 'Overflow'
+    testcase1.crash_state = 'state'
+    testcase2.crash_state = 'state'
+    testcase1.put()
+    testcase2.put()
+    original_testcase_ids = {testcase1.key.id(), testcase2.key.id()}
+
+    grouper.group_testcases()
+
+    self.assertEqual(1, len(self.emitted_events))
+    emitted_event = self.emitted_events[0]
+    self.assertEqual(events.RejectionReason.GROUPER_DUPLICATE,
+                     emitted_event.rejection_reason)
+    self.assertIn(emitted_event.testcase_id, original_testcase_ids)
+
+  def test_group_overflow_rejection_events(self):
+    """Test that removing testcases from large groups emits rejection events."""
+    for i in range(1, 31):
+      testcase = test_utils.create_generic_testcase()
+      testcase.crash_type = 'Heap-buffer-overflow'
+      testcase.crash_state = 'state' + str(i)
+      testcase.project_name = 'project'
+      testcase.one_time_crasher_flag = False
+      testcase.put()
+
+    grouper.group_testcases()
+
+    self.assertEqual(5, len(self.emitted_events))
+    for event in self.emitted_events:
+      self.assertEqual(event.rejection_reason,
+                       events.RejectionReason.GROUPER_OVERFLOW)
+    self.assertEqual(5, self.mock.emit.call_count)