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

This reverts commit 8e43b8b.

Author: vitaliset

configs/test/gce/android-init.bash

@@ -80,6 +80,22 @@ 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,6 +210,32 @@ 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,6 +124,14 @@ 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

@@ -0,0 +1,7 @@
+<source>
+  type tcp
+  format json
+  port 5170
+  bind 127.0.0.1
+  tag bot
+</source>

docker/base/setup_common.sh

@@ -47,6 +47,11 @@ 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

src/appengine/handlers/upload_testcase.py

@@ -392,10 +392,9 @@ 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/clusterfuzz/_internal/metrics/logs.py

@@ -84,15 +84,23 @@ 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', 'True')) and
-          not os.getenv("PY_UNITTESTS") and not _is_local() and
-          not _is_running_on_app_engine() and not _is_running_on_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()
 
 
 def suppress_unwanted_warnings():
@@ -176,57 +184,53 @@ def truncate(msg, limit):
   ])
 
 
-class JsonFormatter(logging.Formatter):
-  """Formats log records as JSON."""
-
-  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'),
-    }
+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'),
+  }
 
-    initial_payload = os.getenv('INITIAL_TASK_PAYLOAD')
-    if initial_payload:
-      entry['actual_task_payload'] = entry['task_payload']
-      entry['task_payload'] = initial_payload
+  initial_payload = os.getenv('INITIAL_TASK_PAYLOAD')
+  if initial_payload:
+    entry['actual_task_payload'] = entry['task_payload']
+    entry['task_payload'] = initial_payload
 
-    entry['location'] = getattr(record, 'location', {'error': True})
-    entry['extras'] = getattr(record, 'extras', {})
-    update_entry_with_exc(entry, record.exc_info)
+  entry['location'] = getattr(record, 'location', {'error': True})
+  entry['extras'] = getattr(record, 'extras', {})
+  update_entry_with_exc(entry, record.exc_info)
 
-    if not entry['extras']:
-      del entry['extras']
+  if not entry['extras']:
+    del entry['extras']
 
-    worker_bot_name = os.environ.get('WORKER_BOT_NAME')
-    if worker_bot_name:
-      entry['worker_bot_name'] = worker_bot_name
+  worker_bot_name = os.environ.get('WORKER_BOT_NAME')
+  if worker_bot_name:
+    entry['worker_bot_name'] = worker_bot_name
 
-    fuzz_target = os.getenv('FUZZ_TARGET')
-    if fuzz_target:
-      entry['fuzz_target'] = fuzz_target
+  fuzz_target = os.getenv('FUZZ_TARGET')
+  if fuzz_target:
+    entry['fuzz_target'] = fuzz_target
 
-    # 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'
+  # 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'
 
-    return json.dumps(entry, default=_handle_unserializable)
+  return json.dumps(entry, default=_handle_unserializable)
 
 
 def _handle_unserializable(unserializable: Any) -> str:
@@ -267,6 +271,16 @@ 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
@@ -344,6 +358,15 @@ 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.
   """
@@ -389,8 +412,6 @@ def cloud_label_filter(record):
 
   handler.addFilter(cloud_label_filter)
   handler.setLevel(logging.INFO)
-  formatter = JsonFormatter()
-  handler.setFormatter(formatter)
 
   logging.getLogger().addHandler(handler)
 
@@ -413,6 +434,8 @@ 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)