Fully migrate gsutil module to gcloud storage

src/clusterfuzz/_internal/bot/tasks/setup.py

@@ -508,7 +508,7 @@ def update_data_bundle(
 
       file_host.create_directory(
           worker_data_bundle_directory, create_intermediates=True)
-      result = untrusted_corpus_manager.RemoteGSUtilRunner().rsync(
+      result = untrusted_corpus_manager.RemoteGCloudStorageRunner().rsync(
           data_bundle_corpus.gcs_url,
           worker_data_bundle_directory,
           delete=False)

src/clusterfuzz/_internal/bot/untrusted_runner/corpus_manager.py

@@ -19,8 +19,8 @@
 from . import remote_process_host
 
 
-class RemoteGSUtilRunner(gsutil.GSUtilRunner):
-  """Remote GSUtil runner."""
+class RemoteGCloudStorageRunner(gsutil.GCloudStorageRunner):
+  """Remote GCloud Storage runner."""
 
   def __init__(self):
     super().__init__(process_runner=remote_process_host.RemoteProcessRunner)
@@ -36,4 +36,4 @@ def __init__(self, fuzzer_name, fuzzer_executable_name, quarantine=False):
         quarantine,
         # Never log results for remote corpora since the state is on the worker.
         log_results=False,
-        gsutil_runner_func=RemoteGSUtilRunner)
+        gcloud_storage_runner_func=RemoteGCloudStorageRunner)

src/clusterfuzz/_internal/bot/untrusted_runner/environment.py

@@ -39,6 +39,7 @@
         r'^FUZZ_TARGET$',
         r'^FUZZ_TEST_TIMEOUT$',
         r'^GSUTIL_PATH$',
+        r'^GCLOUD_PATH$',
         r'^JOB_NAME$',
         r'^LOCAL_DEVELOPMENT$',
         r'^MSAN_OPTIONS$',

src/clusterfuzz/_internal/cron/chrome_tests_syncer.py

@@ -240,7 +240,7 @@ def filter_members(member, path):
   remote_archive = f'gs://autozilli/autozilli-{archive_suffix}.tgz'
   logs.info(f'Processing {remote_archive}')
   local_archive = os.path.join(fuzzilli_tests_directory, 'tmp.tgz')
-  gsutil.GSUtilRunner().download_file(remote_archive, local_archive)
+  gsutil.GCloudStorageRunner().download_file(remote_archive, local_archive)
 
   # Extract relevant files.
   with tarfile.open(local_archive) as tar:
@@ -333,7 +333,8 @@ def sync_tests(tests_archive_bucket: str, tests_archive_name: str,
           '*.svn*',
       ],
       cwd=tests_directory)
-  gsutil.GSUtilRunner().upload_file(tests_archive_local, tests_archive_remote)
+  gsutil.GCloudStorageRunner().upload_file(tests_archive_local,
+                                           tests_archive_remote)
 
   logs.info('Sync complete.')
   monitoring_metrics.CHROME_TEST_SYNCER_SUCCESS.increment()

src/clusterfuzz/_internal/cron/job_exporter.py

@@ -52,12 +52,12 @@ class GCloudCLIRSync(RSyncClient):
     Unsuitable for unit testing."""
 
   def __init__(self):
-    self._runner = gsutil.GSUtilRunner()
+    self._runner = gsutil.GCloudStorageRunner()
 
   def rsync(self, source: str, target: str):
     """Rsyncs a source to a target destination. Returns True if
     successful, False if there was any failure. Considers successful
-     any gsutil execution with a 0 return code."""
+     any gcloud execution with a 0 return code."""
     rsync_process_output = self._runner.rsync(source, target)
     return_code = rsync_process_output.return_code
     return return_code == 0

src/clusterfuzz/_internal/fuzzing/corpus_manager.py

@@ -36,11 +36,11 @@
 
   # Disable "invalid-name" because fixing the issue will cause pylint to
   # complain the None assignment is incorrectly named.
-  DEFAULT_GSUTIL_RUNNER = gsutil.GSUtilRunner  # pylint: disable=invalid-name
+  DEFAULT_GCLOUD_STORAGE_RUNNER = gsutil.GCloudStorageRunner  # pylint: disable=invalid-name
 except:
   # This is expected to fail on App Engine.
   gsutil = None
-  DEFAULT_GSUTIL_RUNNER = None
+  DEFAULT_GCLOUD_STORAGE_RUNNER = None
 
 BACKUP_ARCHIVE_FORMAT = 'zip'
 CORPUS_FILES_SYNC_TIMEOUT = 60 * 60
@@ -61,33 +61,33 @@
 COPY_BUFFER_SIZE = 16 * 1024
 
 
-def _rsync_errors_below_threshold(gsutil_result, max_errors):
+def _rsync_errors_below_threshold(gcloud_result, max_errors):
   """Check if the number of errors during rsync is lower than our threshold."""
-  match = re.search(RSYNC_ERROR_REGEX, gsutil_result.output, re.MULTILINE)
+  match = re.search(RSYNC_ERROR_REGEX, gcloud_result.output, re.MULTILINE)
   if not match:
     return False
 
   error_count = int(match.group(1))
 
   # Ignore NotFoundException(s) since they can happen when files can get deleted
   # e.g. when pruning task is updating corpus.
-  error_count -= gsutil_result.output.count(b'NotFoundException')
-  error_count -= gsutil_result.output.count(b'No such file or directory')
+  error_count -= gcloud_result.output.count(b'NotFoundException')
+  error_count -= gcloud_result.output.count(b'No such file or directory')
 
   return error_count <= max_errors
 
 
-def _handle_rsync_result(gsutil_result, max_errors):
+def _handle_rsync_result(gcloud_result, max_errors):
   """Handle rsync result."""
-  if gsutil_result.return_code == 0:
+  if gcloud_result.return_code == 0:
     sync_succeeded = True
   else:
-    logs.warning('gsutil rsync got non-zero:\n'
+    logs.warning('gcloud rsync got non-zero:\n'
                  'Command: %s\n'
-                 'Output: %s\n' % (gsutil_result.command, gsutil_result.output))
-    sync_succeeded = _rsync_errors_below_threshold(gsutil_result, max_errors)
+                 'Output: %s\n' % (gcloud_result.command, gcloud_result.output))
+    sync_succeeded = _rsync_errors_below_threshold(gcloud_result, max_errors)
 
-  return sync_succeeded and not gsutil_result.timed_out
+  return sync_succeeded and not gcloud_result.timed_out
 
 
 def _count_corpus_files(directory):
@@ -153,7 +153,7 @@ def __init__(self,
                bucket_name,
                bucket_path='/',
                log_results=True,
-               gsutil_runner_func=DEFAULT_GSUTIL_RUNNER):
+               gcloud_storage_runner_func=DEFAULT_GCLOUD_STORAGE_RUNNER):
     """Inits the GcsCorpus.
 
     Args:
@@ -163,7 +163,7 @@ def __init__(self,
     self._bucket_name = bucket_name
     self._bucket_path = bucket_path
     self._log_results = log_results
-    self._gsutil_runner = gsutil_runner_func()
+    self._gcloud_storage_runner = gcloud_storage_runner_func()
 
   @property
   def bucket_name(self):
@@ -174,7 +174,7 @@ def bucket_path(self):
     return self._bucket_path
 
   def get_gcs_url(self):
-    """Build corpus GCS URL for gsutil.
+    """Build corpus GCS URL for gcloud storage.
     Returns:
       A string giving the GCS URL.
     """
@@ -195,16 +195,16 @@ def rsync_from_disk(self,
 
     Args:
       directory: Path to directory to sync from.
-      timeout: Timeout for gsutil.
+      timeout: Timeout for gcloud storage.
       delete: Whether or not to delete files on GCS that don't exist locally.
 
     Returns:
       A bool indicating whether or not the command succeeded.
     """
     corpus_gcs_url = self.get_gcs_url()
     legalize_corpus_files(directory)
-    result = self._gsutil_runner.rsync(
-        directory, corpus_gcs_url, timeout, delete=delete)
+    result = self._gcloud_storage_runner.rsync(
+        directory, corpus_gcs_url, timeout=timeout, delete=delete)
 
     # Allow a small number of files to fail to be synced.
     return _handle_rsync_result(result, max_errors=MAX_SYNC_ERRORS)
@@ -213,11 +213,11 @@ def rsync_to_disk(self,
                     directory,
                     timeout=CORPUS_FILES_SYNC_TIMEOUT,
                     delete=True):
-    """Run gsutil to download corpus files from GCS.
+    """Run gcloud storage to download corpus files from GCS.
 
     Args:
       directory: Path to directory to sync to.
-      timeout: Timeout for gsutil.
+      timeout: Timeout for gcloud storage.
       delete: Whether or not to delete files on disk that don't exist locally.
 
     Returns:
@@ -226,8 +226,8 @@ def rsync_to_disk(self,
     shell.create_directory(directory, create_intermediates=True)
 
     corpus_gcs_url = self.get_gcs_url()
-    result = self._gsutil_runner.rsync(corpus_gcs_url, directory, timeout,
-                                       delete)
+    result = self._gcloud_storage_runner.rsync(
+        corpus_gcs_url, directory, timeout=timeout, delete=delete)
 
     # Allow a small number of files to fail to be synced.
     return _handle_rsync_result(result, max_errors=MAX_SYNC_ERRORS)
@@ -248,7 +248,7 @@ def upload_files(self, file_paths, timeout=CORPUS_FILES_SYNC_TIMEOUT):
     # legal on Windows.
     file_paths = legalize_filenames(file_paths)
     gcs_url = self.get_gcs_url()
-    return self._gsutil_runner.upload_files_to_url(
+    return self._gcloud_storage_runner.upload_files_to_url(
         file_paths, gcs_url, timeout=timeout)
 
 
@@ -261,7 +261,7 @@ def __init__(self,
                quarantine=False,
                log_results=True,
                include_regressions=False,
-               gsutil_runner_func=DEFAULT_GSUTIL_RUNNER):
+               gcloud_storage_runner_func=DEFAULT_GCLOUD_STORAGE_RUNNER):
     """Inits the FuzzTargetCorpus.
 
     Args:
@@ -292,15 +292,16 @@ def __init__(self,
         sync_corpus_bucket_name,
         f'/{self._engine}/{self._project_qualified_target_name}',
         log_results=log_results,
-        gsutil_runner_func=gsutil_runner_func,
+        gcloud_storage_runner_func=gcloud_storage_runner_func,
     )
 
     self._regressions_corpus = GcsCorpus(
         sync_corpus_bucket_name,
         f'/{self._engine}/{self._project_qualified_target_name}'
         f'{REGRESSIONS_GCS_PATH_SUFFIX}',
         log_results=log_results,
-        gsutil_runner_func=gsutil_runner_func) if include_regressions else None
+        gcloud_storage_runner_func=gcloud_storage_runner_func
+    ) if include_regressions else None
 
   @property
   def engine(self):
@@ -320,7 +321,7 @@ def rsync_from_disk(self,
 
     Args:
       directory: Path to directory to sync to.
-      timeout: Timeout for gsutil.
+      timeout: Timeout for gcloud storage.
       delete: Whether or not to delete files on GCS that don't exist locally.
 
     Returns:
@@ -339,13 +340,13 @@ def rsync_to_disk(self,
                     directory,
                     timeout=CORPUS_FILES_SYNC_TIMEOUT,
                     delete=True):
-    """Run gsutil to download corpus files from GCS.
+    """Run gcloud storage to download corpus files from GCS.
 
     Overridden to have additional logging.
 
     Args:
       directory: Path to directory to sync to.
-      timeout: Timeout for gsutil.
+      timeout: Timeout for gcloud storage.
       delete: Whether or not to delete files on disk that don't exist locally.
 
     Returns:
@@ -413,7 +414,7 @@ def rsync_from_disk(self,
 
     Args:
       directory: Path to directory to sync to.
-      timeout: Timeout for gsutil.
+      timeout: Timeout for gcloud storage.
       delete: Whether or not to delete files on GCS that don't exist locally.
 
     Returns:
@@ -613,7 +614,7 @@ def _get_regressions_corpus_gcs_url(bucket_name, bucket_path):
 
 
 def _get_gcs_url(bucket_name, bucket_path, suffix=''):
-  """Build corpus GCS URL for gsutil.
+  """Build corpus GCS URL for gcloud storage.
   Returns:
     A string giving the GCS URL.
   """
@@ -661,7 +662,7 @@ def sync_data_bundle_corpus_to_disk(data_bundle_corpus, directory):
   if (not task_types.task_main_runs_on_uworker() and
       not environment.is_uworker()):
     # Fast path for when we don't need an untrusted worker to run a task.
-    return gsutil.GSUtilRunner().rsync(
+    return gsutil.GCloudStorageRunner().rsync(
         data_bundle_corpus.gcs_url, directory, delete=False).return_code == 0
   results = storage.download_signed_urls(data_bundle_corpus.corpus_urls,
                                          directory)