MNT Removing the Fluentd setup and updating to default to GCP logging (#4668)

Author: vitaliset

configs/test/gce/android-init.bash

@@ -80,22 +80,6 @@ echo "Adding workaround to prevent /dev/random hangs."
 rm /dev/random
 ln -s /dev/urandom /dev/random
 
-echo "Setting up google-fluentd."
-curl -sSO https://dl.google.com/cloudagents/install-logging-agent.sh
-sudo bash install-logging-agent.sh
-echo "
-<source>
-  type tcp
-  format json
-  port 5170
-  bind 127.0.0.1
-  tag bot
-</source>
-" > /etc/google-fluentd/config.d/clusterfuzz.conf
-sed -i 's/flush_interval 5s/flush_interval 60s/' \
-  /etc/google-fluentd/google-fluentd.conf
-sudo service google-fluentd restart
-
 echo "Installing ClusterFuzz package dependencies."
 pip install crcmod==1.7 psutil==5.9.4 cryptography==37.0.4 pyOpenSSL==22.0.0
 

configs/test/gce/windows-init.ps1

@@ -210,32 +210,6 @@ if (!(Test-Path ($fileName))) {
   unzip $fileName
 }
 
-# Download and install google-fluentd
-$fileName = "$tmp\StackdrvierLogging-v1-3.exe"
-if (!(Test-Path ($fileName))) {
-  $webClient.DownloadFile("https://dl.google.com/cloudagents/windows/StackdriverLogging-v1-3.exe", $fileName)
-  cmd /c $fileName /S
-
-  $configFile = "C:\GoogleStackdriverLoggingAgent\fluent.conf"
-  $loggingConfig = @"
-    `r
-    <source>`r
-      type tcp`r
-      format json`r
-      port 5170`r
-      bind 127.0.0.1`r
-      tag bot`r
-    </source>`r
-"@
-  Add-Content $configFile $loggingConfig
-  (Get-Content $configFile) -replace "flush_interval 5s","flush_interval 60s" | out-file -encoding ASCII $configFile
-
-  Start-Sleep -s 30
-
-  net stop fluentdwinsvc
-  net start fluentdwinsvc
-}
-
 # Install NVIDIA driver (Tesla P100).
 $nvidiaDriverVersion = "391.29"
 $fileName = "$tmp\$nvidiaDriverVersion-tesla-desktop-winserver2016-international.exe"

docker/base/Dockerfile

@@ -124,14 +124,6 @@ RUN echo "deb https://packages.cloud.google.com/apt cloud-sdk main" \
     apt-get update -y && \
     apt-get install -y google-cloud-sdk
 
-# Set up google-fluentd
-# We ignore errors in install-logging-agent.sh as it always fails to start
-# after installation without a metadata server.
-RUN curl -sSO https://dl.google.com/cloudagents/add-logging-agent-repo.sh && \
-    bash add-logging-agent-repo.sh --also-install --version=1.10.1 || true && \
-    sed -i 's/flush_interval 5s/flush_interval 60s/' /etc/google-fluentd/google-fluentd.conf
-COPY clusterfuzz-fluentd.conf /etc/google-fluentd/config.d/clusterfuzz.conf
-
 # Common environment variables.
 ENV USER=clusterfuzz
 ENV INSTALL_DIRECTORY /mnt/scratch0

docker/base/clusterfuzz-fluentd.conf

@@ -47,11 +47,6 @@ then
   ln -s /dev/urandom /dev/random
 fi
 
-# Running without credentials will cause this to fail.
-if [[ -z "$LOCAL_SRC" ]]; then
-  /etc/init.d/google-fluentd restart
-fi
-
 # Prevent anything from being written to downloads directory.
 mkdir -p /home/$USER/Downloads
 chmod 111 /home/$USER/Downloads

docker/base/setup_common.sh

@@ -392,9 +392,10 @@ def do_post(self):
     trusted_agreement_signed = request.get(
         'trustedAgreement') == TRUSTED_AGREEMENT_TEXT.strip()
 
-    # Chrome is the only ClusterFuzz deployment where there are trusted bots running utasks.
+    # Chrome is the only ClusterFuzz deployment where there are trusted bots
+    # running utasks.
     # This check also fails on oss-fuzz because of the way it abuses platform.
-    if (not trusted_agreement_signed and utils.is_chromium() and        
+    if (not trusted_agreement_signed and utils.is_chromium() and
         task_utils.is_remotely_executing_utasks() and
         ((platform_id and platform_id != 'Linux') or
          job.platform.lower() != 'linux')):

src/appengine/handlers/upload_testcase.py

@@ -84,23 +84,15 @@ def _file_logging_enabled():
       'True')) and not _is_running_on_app_engine() and not _is_running_on_k8s()
 
 
-def _fluentd_logging_enabled():
-  """Return bool True where fluentd logging is enabled.
-  This is enabled by default.
-  This is disabled for local development and if we are running in app engine or
-    kubernetes as these have their dedicated loggers, see configure_appengine()
-    and configure_k8s()."""
-  return bool(os.getenv('LOG_TO_FLUENTD', 'True')) and not _is_local(
-  ) and not _is_running_on_app_engine() and not _is_running_on_k8s()
-
-
 def _cloud_logging_enabled():
   """Return bool True where Google Cloud Logging is enabled.
+  This is enabled by default.
   This is disabled for local development and if we are running in a app engine
     or kubernetes as these have their dedicated loggers, see
     configure_appengine() and configure_k8s()."""
-  return bool(os.getenv('LOG_TO_GCP')) and not _is_local(
-  ) and not _is_running_on_app_engine() and not _is_running_on_k8s()
+  return (bool(os.getenv('LOG_TO_GCP', 'True')) and
+          not os.getenv("PY_UNITTESTS") and not _is_local() and
+          not _is_running_on_app_engine() and not _is_running_on_k8s())
 
 
 def suppress_unwanted_warnings():
@@ -184,53 +176,57 @@ def truncate(msg, limit):
   ])
 
 
-def format_record(record: logging.LogRecord) -> str:
-  """Format LogEntry into JSON string."""
-  entry = {
-      'message':
-          truncate(record.getMessage(), STACKDRIVER_LOG_MESSAGE_LIMIT),
-      'created': (
-          datetime.datetime.utcfromtimestamp(record.created).isoformat() + 'Z'),
-      'severity':
-          record.levelname,
-      'bot_name':
-          os.getenv('BOT_NAME'),
-      'task_payload':
-          os.getenv('TASK_PAYLOAD'),
-      'name':
-          record.name,
-      'pid':
-          os.getpid(),
-      'task_id':
-          os.getenv('CF_TASK_ID', 'null'),
-  }
+class JsonFormatter(logging.Formatter):
+  """Formats log records as JSON."""
 
-  initial_payload = os.getenv('INITIAL_TASK_PAYLOAD')
-  if initial_payload:
-    entry['actual_task_payload'] = entry['task_payload']
-    entry['task_payload'] = initial_payload
+  def format(self, record: logging.LogRecord) -> str:
+    """Format LogEntry into JSON string."""
+    entry = {
+        'message':
+            truncate(record.getMessage(), STACKDRIVER_LOG_MESSAGE_LIMIT),
+        'created': (
+            datetime.datetime.utcfromtimestamp(record.created).isoformat() + 'Z'
+        ),
+        'severity':
+            record.levelname,
+        'bot_name':
+            os.getenv('BOT_NAME'),
+        'task_payload':
+            os.getenv('TASK_PAYLOAD'),
+        'name':
+            record.name,
+        'pid':
+            os.getpid(),
+        'task_id':
+            os.getenv('CF_TASK_ID', 'null'),
+    }
 
-  entry['location'] = getattr(record, 'location', {'error': True})
-  entry['extras'] = getattr(record, 'extras', {})
-  update_entry_with_exc(entry, record.exc_info)
+    initial_payload = os.getenv('INITIAL_TASK_PAYLOAD')
+    if initial_payload:
+      entry['actual_task_payload'] = entry['task_payload']
+      entry['task_payload'] = initial_payload
 
-  if not entry['extras']:
-    del entry['extras']
+    entry['location'] = getattr(record, 'location', {'error': True})
+    entry['extras'] = getattr(record, 'extras', {})
+    update_entry_with_exc(entry, record.exc_info)
 
-  worker_bot_name = os.environ.get('WORKER_BOT_NAME')
-  if worker_bot_name:
-    entry['worker_bot_name'] = worker_bot_name
+    if not entry['extras']:
+      del entry['extras']
 
-  fuzz_target = os.getenv('FUZZ_TARGET')
-  if fuzz_target:
-    entry['fuzz_target'] = fuzz_target
+    worker_bot_name = os.environ.get('WORKER_BOT_NAME')
+    if worker_bot_name:
+      entry['worker_bot_name'] = worker_bot_name
 
-  # Log bot shutdown cases as WARNINGs since this is expected for preemptibles.
-  if (entry['severity'] in ['ERROR', 'CRITICAL'] and
-      'IOError: [Errno 4] Interrupted function call' in entry['message']):
-    entry['severity'] = 'WARNING'
+    fuzz_target = os.getenv('FUZZ_TARGET')
+    if fuzz_target:
+      entry['fuzz_target'] = fuzz_target
 
-  return json.dumps(entry, default=_handle_unserializable)
+    # Log bot shutdown cases as WARNINGs (this is expected for preemptibles).
+    if (entry['severity'] in ['ERROR', 'CRITICAL'] and
+        'IOError: [Errno 4] Interrupted function call' in entry['message']):
+      entry['severity'] = 'WARNING'
+
+    return json.dumps(entry, default=_handle_unserializable)
 
 
 def _handle_unserializable(unserializable: Any) -> str:
@@ -271,16 +267,6 @@ def update_entry_with_exc(entry, exc_info):
     }
 
 
-class JsonSocketHandler(logging.handlers.SocketHandler):
-  """Format log into JSON string before sending it to fluentd. We need this
-    because SocketHandler doesn't respect the formatter attribute."""
-
-  def makePickle(self, record: logging.LogRecord):
-    """Format LogEntry into JSON string."""
-    # \n is the recognized delimiter by fluentd's in_tcp. Don't remove.
-    return (format_record(record) + '\n').encode('utf-8')
-
-
 def uncaught_exception_handler(exception_type, exception_value,
                                exception_traceback):
   """Handles any exception that are uncaught by logging an error and calling
@@ -358,15 +344,6 @@ def record_factory(*args, **kwargs):
   logging.getLogger().setLevel(logging.INFO)
 
 
-def configure_fluentd_logging():
-  fluentd_handler = JsonSocketHandler(
-      host='127.0.0.1',
-      port=5170,
-  )
-  fluentd_handler.setLevel(logging.INFO)
-  logging.getLogger().addHandler(fluentd_handler)
-
-
 def configure_cloud_logging():
   """ Configure Google cloud logging, for bots not running on appengine nor k8s.
   """
@@ -412,6 +389,8 @@ def cloud_label_filter(record):
 
   handler.addFilter(cloud_label_filter)
   handler.setLevel(logging.INFO)
+  formatter = JsonFormatter()
+  handler.setFormatter(formatter)
 
   logging.getLogger().addHandler(handler)
 
@@ -434,8 +413,6 @@ def configure(name, extras=None):
     logging.basicConfig(level=logging.INFO)
   if _file_logging_enabled():
     config.dictConfig(get_logging_config_dict(name))
-  if _fluentd_logging_enabled():
-    configure_fluentd_logging()
   if _cloud_logging_enabled():
     configure_cloud_logging()
   logger = logging.getLogger(name)