Get batch ready for OSS-Fuzz (#4387)

This should not be landed before the corresponding internal config
change.
1. Fix utask-main scheduling for oss-fuzz.
2. Fix configuration of subnet, network, and gce region
3. Only test using skia for now.
4. Add tests for this functionality.

Author: jonathanmetzman

configs/test/batch/batch.yaml

@@ -21,7 +21,10 @@ mapping:
     disk_type: pd-standard
     service_account_email: test-clusterfuzz-service-account-email
     subnetwork: null
+    gce_region: 'gce-region'
     gce_zone: 'gce-zone'
+    network: 'projects/google.com:clusterfuzz/global/networks/networkname'
+    subnetwork: 'projects/google.com:clusterfuzz/regions/gce-region/subnetworks/subnetworkname'
     preemptible: false
     machine_type: n1-standard-1
   LINUX-NONPREEMPTIBLE-UNPRIVILEGED:
@@ -31,8 +34,10 @@ mapping:
     disk_size_gb: 110
     disk_type: pd-standard
     service_account_email: test-unpriv-clusterfuzz-service-account-email
-    subnetwork: null
+    gce_region: 'gce-region'
     gce_zone: 'gce-zone'
+    network: 'projects/google.com:clusterfuzz/global/networks/networkname'
+    subnetwork: 'projects/google.com:clusterfuzz/regions/gce-region/subnetworks/subnetworkname'
     preemptible: false
     machine_type: n1-standard-1
   LINUX-PREEMPTIBLE:
@@ -42,8 +47,10 @@ mapping:
     disk_size_gb: 75
     disk_type: pd-standard
     service_account_email: test-clusterfuzz-service-account-email
-    subnetwork: null
+    gce_region: 'gce-region'
     gce_zone: 'gce-zone'
+    network: 'projects/google.com:clusterfuzz/global/networks/networkname'
+    subnetwork: 'projects/google.com:clusterfuzz/regions/gce-region/subnetworks/subnetworkname'
     preemptible: true
     machine_type: n1-standard-1
   LINUX-PREEMPTIBLE-UNPRIVILEGED:
@@ -53,8 +60,10 @@ mapping:
     disk_size_gb: 75
     disk_type: pd-standard
     service_account_email: test-unpriv-clusterfuzz-service-account-email
-    subnetwork: null
+    gce_region: 'gce-region'
     gce_zone: 'gce-zone'
+    network: 'projects/google.com:clusterfuzz/global/networks/networkname'
+    subnetwork: 'projects/google.com:clusterfuzz/regions/gce-region/subnetworks/subnetworkname'
     preemptible: true
     machine_type: n1-standard-1
 project: 'test-clusterfuzz'

src/clusterfuzz/_internal/base/tasks/__init__.py

@@ -497,8 +497,6 @@ def get_task_from_message(message) -> Optional[PubSubTask]:
 def get_utask_mains() -> List[PubSubTask]:
   """Returns a list of tasks for preprocessing many utasks on this bot and then
   running the uworker_mains in the same batch job."""
-  if not task_utils.is_remotely_executing_utasks():
-    return None
   pubsub_puller = PubSubPuller(UTASK_MAINS_QUEUE)
   messages = pubsub_puller.get_messages_time_limited(MAX_UTASKS,
                                                      UTASK_QUEUE_PULL_SECONDS)

src/clusterfuzz/_internal/base/tasks/task_utils.py

@@ -40,6 +40,9 @@ def is_remotely_executing_utasks(task=None) -> bool:
 def is_task_opted_into_uworker_execution(task):
   # TODO(metzman): Remove this after OSS-Fuzz and Chrome are at parity.
   uworker_tasks = local_config.ProjectConfig().get('uworker_tasks', [])
+  if 'skia' not in environment.get_value('JOB_NAME', ''):
+    # This is just for testing OSS-Fuzz.
+    return False
   return task in uworker_tasks
 
 

src/clusterfuzz/_internal/google_cloud_utils/batch.py

@@ -46,17 +46,9 @@
 MAX_CONCURRENT_VMS_PER_JOB = 1000
 
 BatchWorkloadSpec = collections.namedtuple('BatchWorkloadSpec', [
-    'clusterfuzz_release',
-    'disk_size_gb',
-    'disk_type',
-    'docker_image',
-    'user_data',
-    'service_account_email',
-    'subnetwork',
-    'preemptible',
-    'project',
-    'gce_zone',
-    'machine_type',
+    'clusterfuzz_release', 'disk_size_gb', 'disk_type', 'docker_image',
+    'user_data', 'service_account_email', 'subnetwork', 'preemptible',
+    'project', 'gce_zone', 'machine_type', 'network', 'gce_region'
 ])
 
 
@@ -166,7 +158,6 @@ def _get_task_spec(batch_workload_spec):
   task_spec = batch.TaskSpec()
   task_spec.runnables = [runnable]
   task_spec.max_retry_count = RETRY_COUNT
-  # TODO(metzman): Change this for production.
   task_spec.max_run_duration = MAX_DURATION
   return task_spec
 
@@ -187,12 +178,8 @@ def _get_allocation_policy(spec):
   # unnecessary.
   network_interface = batch.AllocationPolicy.NetworkInterface()
   network_interface.no_external_ip_address = True
-  # TODO(metzman): Make configurable.
-  network_interface.network = (
-      'projects/google.com:clusterfuzz/global/networks/batch')
-  network_interface.subnetwork = (
-      'projects/google.com:clusterfuzz/regions/us-west1/subnetworks/us-west1a')
-
+  network_interface.network = spec.network
+  network_interface.subnetwork = spec.subnetwork
   network_interfaces = [network_interface]
   network_policy = batch.AllocationPolicy.NetworkPolicy()
   network_policy.network_interfaces = network_interfaces
@@ -231,8 +218,9 @@ def _create_job(spec, input_urls):
   job_name = get_job_name()
   create_request.job_id = job_name
   # The job's parent is the region in which the job will run
-  project_id = 'google.com:clusterfuzz'
-  create_request.parent = f'projects/{project_id}/locations/us-west1'
+  project_id = spec.project
+  create_request.parent = (
+      f'projects/{project_id}/locations/{spec.gce_region}')
   job_result = _send_create_job_request(create_request)
   logs.info(f'Created batch job id={job_name}.', spec=spec)
   return job_result
@@ -269,14 +257,26 @@ def is_remote_task(command, job_name):
     return False
 
 
-def _get_spec_from_config(command, job_name):
-  """Gets the configured specifications for a batch workload."""
+def _get_config_name(command, job_name):
+  """Returns the name of the config for |command| and |job_name|."""
   job = _get_job(job_name)
-  config_name = job.platform
+  # As of this writing, batch only supports LINUX.
+  if utils.is_oss_fuzz():
+    # TODO(b/377885331): In OSS-Fuzz, the platform can't be used because, as of
+    # it includes the project name.
+    config_name = 'LINUX'
+  else:
+    config_name = job.platform
   if command == 'fuzz':
     config_name += '-PREEMPTIBLE-UNPRIVILEGED'
   else:
     config_name += '-NONPREEMPTIBLE-UNPRIVILEGED'
+  return config_name
+
+
+def _get_spec_from_config(command, job_name):
+  """Gets the configured specifications for a batch workload."""
+  config_name = _get_config_name(command, job_name)
   batch_config = _get_batch_config()
   instance_spec = batch_config.get('mapping').get(config_name, None)
   if instance_spec is None:
@@ -285,18 +285,18 @@ def _get_spec_from_config(command, job_name):
   docker_image = instance_spec['docker_image']
   user_data = instance_spec['user_data']
   clusterfuzz_release = instance_spec.get('clusterfuzz_release', 'prod')
-  # TODO(https://github.com/google/clusterfuzz/issues/3008): Make this use a
-  # low-privilege account.
   spec = BatchWorkloadSpec(
       clusterfuzz_release=clusterfuzz_release,
       docker_image=docker_image,
       user_data=user_data,
       disk_size_gb=instance_spec['disk_size_gb'],
       disk_type=instance_spec['disk_type'],
       service_account_email=instance_spec['service_account_email'],
-      subnetwork=instance_spec['subnetwork'],
       gce_zone=instance_spec['gce_zone'],
+      gce_region=instance_spec['gce_region'],
       project=project_name,
+      network=instance_spec['network'],
+      subnetwork=instance_spec['subnetwork'],
       preemptible=instance_spec['preemptible'],
       machine_type=instance_spec['machine_type'])
   return spec

src/clusterfuzz/_internal/system/shell.py

@@ -176,6 +176,8 @@ def clear_temp_directory(clear_user_profile_directories=True):
 @environment.local_noop
 def clear_system_temp_directory():
   """Clear system specific temp directory."""
+  if environment.get_value('DEBUG_TASK'):
+    return
 
   def _delete_object(path, delete_func):
     """Delete a object with its delete function, ignoring any error."""

src/clusterfuzz/_internal/tests/core/base/tasks/task_utils_test.py

@@ -12,10 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 """Tests for task_utils."""
+import os
 import unittest
 
 from clusterfuzz._internal.base.tasks import task_utils
 from clusterfuzz._internal.bot.tasks import commands
+from clusterfuzz._internal.tests.test_libs import helpers
 
 
 class GetCommandFromModuleTest(unittest.TestCase):
@@ -36,9 +38,20 @@ def test_get_command_from_module(self):
 
 
 class IsTaskOptedIntoUworkerExecution(unittest.TestCase):
+  """Tests that is_task_opted_into_uworker_execution only returns True for the
+  tasks we are testing in oss-fuzz."""
+
+  def setUp(self):
+    helpers.patch_environ(self)
 
   def test_opt_in(self):
+    os.environ['JOB_NAME'] = 'libfuzzer_asan_skia'
     self.assertTrue(task_utils.is_task_opted_into_uworker_execution('analyze'))
 
+  def test_wrong_job(self):
+    os.environ['JOB_NAME'] = 'libfuzzer_asan_systemd'
+    self.assertFalse(task_utils.is_task_opted_into_uworker_execution('analyze'))
+
   def test_no_opt_in(self):
+    os.environ['JOB_NAME'] = 'libfuzzer_asan_skia'
     self.assertFalse(task_utils.is_task_opted_into_uworker_execution('fuzz'))

src/clusterfuzz/_internal/tests/core/google_cloud_utils/batch_test.py

@@ -40,7 +40,10 @@ def test_nonpreemptible_get_spec_from_config(self):
         disk_size_gb=110,
         disk_type='pd-standard',
         service_account_email='test-unpriv-clusterfuzz-service-account-email',
-        subnetwork=None,
+        subnetwork=
+        'projects/google.com:clusterfuzz/regions/gce-region/subnetworks/subnetworkname',
+        network='projects/google.com:clusterfuzz/global/networks/networkname',
+        gce_region='gce-region',
         gce_zone='gce-zone',
         project='test-clusterfuzz',
         preemptible=False,
@@ -60,8 +63,11 @@ def test_preemptible_get_spec_from_config(self):
         disk_size_gb=75,
         disk_type='pd-standard',
         service_account_email='test-unpriv-clusterfuzz-service-account-email',
-        subnetwork=None,
+        subnetwork=
+        'projects/google.com:clusterfuzz/regions/gce-region/subnetworks/subnetworkname',
+        network='projects/google.com:clusterfuzz/global/networks/networkname',
         gce_zone='gce-zone',
+        gce_region='gce-region',
         project='test-clusterfuzz',
         preemptible=True,
         machine_type='n1-standard-1')

src/python/bot/startup/run_bot.py

@@ -96,7 +96,6 @@ def schedule_utask_mains():
 
   logs.info(f'Combining {len(utask_mains)} batch tasks.')
 
-  batch_tasks = []
   with lease_all_tasks(utask_mains):
     batch_tasks = [
         batch.BatchTask(task.command, task.job, task.argument)