[Logging] Create and add structured logs for Cleanup task (#4781)

### Description
Adding structured logging to the cleanup cronjob. In order to enable
this, the following changes on logs module were done:
* Added a new log context for cronjobs.
* Added a filter to the logs handler for both GAE/GKE to propagate the
`extras` argument from python logging to the `json_fields` metadata in
google cloud logging ([GCP
reference](https://cloud.google.com/python/docs/reference/logging/latest/std-lib-integration#automatic-metadata-detection)).
(Note that this is a workaround to enable the observability project for
now. As a long-term solution, we should try to centralize the logs
handlers/filters for all environments).

For the cleanup module:
* Set env vars for task name and ID in the cron entrypoints (for k8s and
appengine).
* Instrumented the cleanup code with the cron-based log context and with
the testcase-based context. This latest is done for each testcase within
the clean up unneeded open testcases method.

### Tests
Running the cronjob locally with the debugger and sending logs to GCP.
* Setting `IS_K8S_ENV` to true ([link to
logs](https://cloudlogging.app.goo.gl/Us8aHBMUyNoaPFbL8)):

![image](https://github.com/user-attachments/assets/4d268895-804d-483d-a426-174a3c8b5f4b)

Author: ViniciustCosta

src/appengine/libs/handler.py

@@ -97,6 +97,9 @@ def wrapper(self):
       if not self.is_cron():
         raise helpers.AccessDeniedError('You are not a cron.')
 
+      # Add env vars used by logs context for cleanup/triage.
+      environment.set_task_id_vars(self.__module__)
+
       with monitor.wrap_with_monitoring():
         result = func(self)
         if result is None:

src/clusterfuzz/_internal/cron/cleanup.py

@@ -100,6 +100,60 @@ def cleanup_reports_metadata():
   ndb_utils.delete_multi(uploaded_reports)
 
 
+def _cleanup_testcases_and_issues(testcase, jobs,
+                                  top_crashes_by_project_and_platform_map,
+                                  empty_issue_tracker_policy):
+  """Clean up unneeded open testcase and its associated issues."""
+  testcase_id = testcase.key.id()
+  logs.info(f'Processing testcase {testcase_id}.')
+
+  try:
+    issue = issue_tracker_utils.get_issue_for_testcase(testcase)
+    policy = issue_tracker_utils.get_issue_tracker_policy_for_testcase(testcase)
+    if not policy:
+      logs.info('No policy')
+      policy = empty_issue_tracker_policy
+
+    # Issue updates.
+    update_os_labels(policy, testcase, issue)
+    logs.info('maybe updated os')
+    update_fuzz_blocker_label(policy, testcase, issue,
+                              top_crashes_by_project_and_platform_map)
+    logs.info('maybe updated fuzz blocker')
+    update_component_labels_and_id(policy, testcase, issue)
+    logs.info('maybe updated component labels and component id')
+    update_issue_ccs_from_owners_file(policy, testcase, issue)
+    logs.info('maybe updated issueccs')
+    update_issue_owner_and_ccs_from_predator_results(policy, testcase, issue)
+    logs.info('maybe updated update_issue_owner_and_ccs_from_predator_results')
+    update_issue_labels_for_flaky_testcase(policy, testcase, issue)
+
+    # Testcase marking rules.
+    mark_duplicate_testcase_as_closed_with_no_issue(testcase)
+    mark_issue_as_closed_if_testcase_is_fixed(policy, testcase, issue)
+    mark_testcase_as_closed_if_issue_is_closed(policy, testcase, issue)
+    mark_testcase_as_closed_if_job_is_invalid(testcase, jobs)
+    mark_unreproducible_testcase_as_fixed_if_issue_is_closed(testcase, issue)
+    mark_unreproducible_testcase_and_issue_as_closed_after_deadline(
+        policy, testcase, issue)
+    mark_na_testcase_issues_as_wontfix(policy, testcase, issue)
+
+    # Notification, to be done at end after testcase state is updated from
+    # previous rules.
+    notify_closed_issue_if_testcase_is_open(policy, testcase, issue)
+    notify_issue_if_testcase_is_invalid(policy, testcase, issue)
+    notify_uploader_when_testcase_is_processed(policy, testcase, issue)
+
+    # Mark testcase as triage complete if both testcase and associated issue
+    # are closed. This also need to be done before the deletion rules.
+    mark_testcase_as_triaged_if_needed(testcase, issue)
+
+    # Testcase deletion rules.
+    delete_unreproducible_testcase_with_no_issue(testcase)
+  except Exception:
+    logs.error(f'Failed to process testcase {testcase_id}.')
+
+
 def cleanup_testcases_and_issues():
   """Clean up unneeded open testcases and their associated issues."""
   logs.info('Getting all job type names.')
@@ -125,55 +179,10 @@ def cleanup_testcases_and_issues():
       # Already deleted.
       continue
 
-    logs.info(f'Processing testcase {testcase_id}.')
-
-    try:
-      issue = issue_tracker_utils.get_issue_for_testcase(testcase)
-      policy = issue_tracker_utils.get_issue_tracker_policy_for_testcase(
-          testcase)
-      if not policy:
-        logs.info('No policy')
-        policy = empty_issue_tracker_policy
-
-      # Issue updates.
-      update_os_labels(policy, testcase, issue)
-      logs.info('maybe updated os')
-      update_fuzz_blocker_label(policy, testcase, issue,
-                                top_crashes_by_project_and_platform_map)
-      logs.info('maybe updated fuzz blocker')
-      update_component_labels_and_id(policy, testcase, issue)
-      logs.info('maybe updated component labels and component id')
-      update_issue_ccs_from_owners_file(policy, testcase, issue)
-      logs.info('maybe updated issueccs')
-      update_issue_owner_and_ccs_from_predator_results(policy, testcase, issue)
-      logs.info(
-          'maybe updated update_issue_owner_and_ccs_from_predator_results')
-      update_issue_labels_for_flaky_testcase(policy, testcase, issue)
-
-      # Testcase marking rules.
-      mark_duplicate_testcase_as_closed_with_no_issue(testcase)
-      mark_issue_as_closed_if_testcase_is_fixed(policy, testcase, issue)
-      mark_testcase_as_closed_if_issue_is_closed(policy, testcase, issue)
-      mark_testcase_as_closed_if_job_is_invalid(testcase, jobs)
-      mark_unreproducible_testcase_as_fixed_if_issue_is_closed(testcase, issue)
-      mark_unreproducible_testcase_and_issue_as_closed_after_deadline(
-          policy, testcase, issue)
-      mark_na_testcase_issues_as_wontfix(policy, testcase, issue)
-
-      # Notification, to be done at end after testcase state is updated from
-      # previous rules.
-      notify_closed_issue_if_testcase_is_open(policy, testcase, issue)
-      notify_issue_if_testcase_is_invalid(policy, testcase, issue)
-      notify_uploader_when_testcase_is_processed(policy, testcase, issue)
-
-      # Mark testcase as triage complete if both testcase and associated issue
-      # are closed. This also need to be done before the deletion rules.
-      mark_testcase_as_triaged_if_needed(testcase, issue)
-
-      # Testcase deletion rules.
-      delete_unreproducible_testcase_with_no_issue(testcase)
-    except Exception:
-      logs.error(f'Failed to process testcase {testcase_id}.')
+    with logs.testcase_log_context(testcase, testcase.get_fuzz_target()):
+      _cleanup_testcases_and_issues(testcase, jobs,
+                                    top_crashes_by_project_and_platform_map,
+                                    empty_issue_tracker_policy)
 
     testcases_processed += 1
     if testcases_processed % 100 == 0:
@@ -1406,6 +1415,7 @@ def cleanup_unused_heartbeats():
   ndb_utils.delete_multi(unused_heartbeats)
 
 
+@logs.cron_log_context()
 def main():
   """Cleaning up unneeded testcases"""
   cleanup_testcases_and_issues()

src/clusterfuzz/_internal/metrics/logs.py

@@ -325,6 +325,23 @@ def uncaught_exception_handler(exception_type, exception_value,
   sys.__excepthook__(exception_type, exception_value, exception_traceback)
 
 
+def json_fields_filter(record):
+  """Add logs `extras` argument to `json_fields` metadata for cloud logging."""
+  # TODO(vtcosta): This is a workaround to allow structured logs for
+  # cleanup/triage cronjobs in GKE/GAE. We should try to refactor and
+  # centralize the logs configurations for all environments.
+  if not hasattr(record, 'json_fields'):
+    record.json_fields = {}
+
+  record.json_fields.update({
+      'extras': {
+          k: truncate(v, STACKDRIVER_LOG_MESSAGE_LIMIT)
+          for k, v in getattr(record, 'extras', {}).items()
+      }
+  })
+  return True
+
+
 def configure_appengine():
   """Configure logging for App Engine."""
   logging.getLogger().setLevel(logging.INFO)
@@ -335,6 +352,7 @@ def configure_appengine():
   import google.cloud.logging
   client = google.cloud.logging.Client()
   handler = client.get_default_handler()
+  handler.addFilter(json_fields_filter)
   logging.getLogger().addHandler(handler)
 
 
@@ -374,6 +392,7 @@ def record_factory(*args, **kwargs):
     return record
 
   logging.setLogRecordFactory(record_factory)
+  logging.getLogger().addFilter(json_fields_filter)
   logging.getLogger().setLevel(logging.INFO)
 
 
@@ -695,6 +714,11 @@ class TaskLogStruct(NamedTuple):
   stage: str
 
 
+class CronLogStruct(NamedTuple):
+  task_id: str
+  task_name: str
+
+
 class FuzzerLogStruct(NamedTuple):
   fuzz_target: str
   job: str
@@ -715,6 +739,7 @@ class LogContextType(enum.Enum):
   TASK = 'task'
   FUZZER = 'fuzzer'
   TESTCASE = 'testcase'
+  CRON = 'cron'
 
   def get_extras(self) -> NamedTuple:
     """Get the structured log for a given context"""
@@ -783,6 +808,17 @@ def get_extras(self) -> NamedTuple:
             ignore_context=True)
         return GenericLogStruct()
 
+    if self == LogContextType.CRON:
+      try:
+        return CronLogStruct(
+            task_name=os.getenv('CF_TASK_NAME', 'null'),
+            task_id=os.getenv('CF_TASK_ID', 'null'))
+      except:
+        error(
+            'Error retrieving context for cron-based logs.',
+            ignore_context=True)
+        return GenericLogStruct()
+
     return GenericLogStruct()
 
 
@@ -922,3 +958,13 @@ def testcase_log_context(testcase: 'Testcase',
       log_contexts.delete_metadata('fuzzer_name')
       log_contexts.delete_metadata('job_type')
       log_contexts.delete_metadata('fuzz_target')
+
+
+@contextlib.contextmanager
+def cron_log_context():
+  with wrap_log_context(contexts=[LogContextType.CRON]):
+    try:
+      yield
+    except Exception as e:
+      warning(message='Error during cronjob context.')
+      raise e

src/clusterfuzz/_internal/system/environment.py

@@ -20,6 +20,7 @@
 import socket
 import subprocess
 import sys
+import uuid
 
 import yaml
 
@@ -1014,6 +1015,14 @@ def set_tsan_max_history_size():
   set_value('TSAN_OPTIONS', tsan_options)
 
 
+def set_task_id_vars(task_name, task_id=None):
+  """Sets environment vars for task name and ID."""
+  if not task_id:
+    task_id = uuid.uuid4()
+  set_value('CF_TASK_ID', task_id)
+  set_value('CF_TASK_NAME', task_name)
+
+
 def set_value(environment_variable, value, env=None):
   """Set an environment variable."""
   if env is None:

src/python/bot/startup/run_cron.py

@@ -59,6 +59,9 @@ def main():
   task = sys.argv[1]
 
   task_module_name = f'clusterfuzz._internal.cron.{task}'
+
+  environment.set_task_id_vars(task)
+
   with monitor.wrap_with_monitoring(), ndb_init.context():
     task_module = importlib.import_module(task_module_name)
     return 0 if task_module.main() else 1