Refactor RemoteTask and fix Kubernetes E2E test

- Update RemoteTask interface to include create_uworker_main_batch_jobs.
- Refactor KubernetesService and GcpBatchService to match new interface.
- Fix TypeError in k8s_service_e2e_test.py by adding @classmethod to tearDownClass.
- Move and update tests for batch and k8s services.

Author: javanlacerda

src/clusterfuzz/_internal/batch/service.py

@@ -259,7 +259,7 @@ def _get_subconfig(batch_config, instance_spec):
   return all_subconfigs[weighted_subconfig.name]
 
 
-def _get_specs_from_config(batch_tasks) -> Dict:
+def _get_specs_from_config(batch_tasks: List[RemoteTask]) -> Dict:
   """Gets the configured specifications for a batch workload."""
   if not batch_tasks:
     return {}
@@ -383,19 +383,19 @@ def create_uworker_main_batch_job(self, module: str, job_type: str,
       return result
     return result[0]
 
-  def create_uworker_main_batch_jobs(self, batch_tasks: List[RemoteTask]):
+  def create_uworker_main_batch_jobs(self, remote_tasks: List[RemoteTask]):
     """Creates a batch job for a list of uworker main tasks.
     
     This method groups the tasks by their workload specification and creates a
     separate batch job for each group. This allows tasks with similar
     requirements to be processed together, which can improve efficiency.
     """
     job_specs = collections.defaultdict(list)
-    specs = _get_specs_from_config(batch_tasks)
-    for batch_task in batch_tasks:
-      logs.info(f'Scheduling {batch_task.command}, {batch_task.job_type}.')
-      spec = specs[(batch_task.command, batch_task.job_type)]
-      job_specs[spec].append(batch_task.input_download_url)
+    specs = _get_specs_from_config(remote_tasks)
+    for remote_task in remote_tasks:
+      logs.info(f'Scheduling {remote_task.command}, {remote_task.job_type}.')
+      spec = specs[(remote_task.command, remote_task.job_type)]
+      job_specs[spec].append(remote_task.input_download_url)
 
     logs.info('Creating batch jobs.')
     jobs = []

src/clusterfuzz/_internal/k8s/service.py

@@ -59,13 +59,14 @@ def _create_job_client_wrapper(self, container_image: str, job_spec: dict,
     return job_name
 
   def create_job(self, remote_task: RemoteTaskInterface,
-                 input_urls: List[str]) -> str:
+                 input_urls: List[str], docker_image: str) -> str:
     """Creates a Kubernetes job.
 
     Args:
       remote_task: The remote task specification.
       input_urls: A list of URLs to be passed as environment variables to the
         job's container.
+      docker_image: The Docker image to use for the job.
     Returns:
       The name of the created Kubernetes job.
     """
@@ -75,7 +76,8 @@ def create_job(self, remote_task: RemoteTaskInterface,
         'kind': 'Job',
         'metadata': {
             'name':
-                remote_task.job_type  # Use job_type as base name
+                getattr(remote_task, 'job_type',
+                        'clusterfuzz-job')  # Use job_type as base name
         },
         'spec': {
             'template': {
@@ -92,7 +94,7 @@ def create_job(self, remote_task: RemoteTaskInterface,
             'backoffLimit': 0
         }
     }
-    return self._create_job_client_wrapper(remote_task.docker_image, job_spec,
+    return self._create_job_client_wrapper(docker_image, job_spec,
                                            input_urls)
 
   def create_uworker_main_batch_job(self, module: str, job_type: str,
@@ -105,19 +107,19 @@ def create_uworker_main_batch_job(self, module: str, job_type: str,
       return result
     return result[0]
 
-  def create_uworker_main_batch_jobs(self, batch_tasks: List[RemoteTask]):
+  def create_uworker_main_batch_jobs(self, remote_tasks: List[RemoteTask]):
     """Creates a batch job for a list of uworker main tasks.
     
     This method groups the tasks by their workload specification and creates a
     separate batch job for each group. This allows tasks with similar
     requirements to be processed together, which can improve efficiency.
     """
     job_specs = collections.defaultdict(list)
-    specs = _get_specs_from_config(batch_tasks)
-    for batch_task in batch_tasks:
-      logs.info(f'Scheduling {batch_task.command}, {batch_task.job_type}.')
-      spec = specs[(batch_task.command, batch_task.job_type)]
-      job_specs[spec].append(batch_task.input_download_url)
+    specs = _get_specs_from_config(remote_tasks)
+    for remote_task in remote_tasks:
+      logs.info(f'Scheduling {remote_task.command}, {remote_task.job_type}.')
+      spec = specs[(remote_task.command, remote_task.job_type)]
+      job_specs[spec].append(remote_task.input_download_url)
 
     logs.info('Creating batch jobs.')
     jobs = []
@@ -126,7 +128,7 @@ def create_uworker_main_batch_jobs(self, batch_tasks: List[RemoteTask]):
     for spec, input_urls in job_specs.items():
       for input_urls_portion in utils.batched(input_urls,
                                               MAX_CONCURRENT_VMS_PER_JOB - 1):
-        jobs.append(self.create_job(spec, input_urls_portion))
+        jobs.append(self.create_job(spec, input_urls_portion, spec.docker_image))
 
     return jobs
 

src/clusterfuzz/_internal/remote_task/__init__.py

@@ -19,8 +19,11 @@
 the task creation logic to a specific implementation.
 """
 import abc
+import random
 from typing import List
 
+from clusterfuzz._internal.remote_task import job_frequency
+
 
 class RemoteTask:
   """Represents a single ClusterFuzz task to be executed on a remote worker.
@@ -45,22 +48,18 @@ class RemoteTaskInterface(abc.ABC):
   """
 
   @abc.abstractmethod
-  def create_job(self, remote_task: RemoteTask, input_urls: List[str]):
-    """Creates a remote job.
-    
-    This method is responsible for creating a new job in the remote execution
-    environment. It takes a workload specification and a list of input URLs,
-    and returns a representation of the created job.
-    """
+  def create_uworker_main_batch_job(self, module: str, job_type: str,
+                                    input_download_url: str):
+    """Creates a single remote task for a uworker main task."""
     raise NotImplementedError
 
-
-import random
-
-from clusterfuzz._internal.remote_task import job_frequency
+  @abc.abstractmethod
+  def create_uworker_main_batch_jobs(self, remote_tasks: List[RemoteTask]):
+    """Creates a many remote tasks for uworker main tasks."""
+    raise NotImplementedError
 
 
-class RemoteTaskGate:
+class RemoteTaskGate(RemoteTaskInterface):
   """A gatekeeper for remote task execution.
   
   This class is responsible for choosing the remote execution backend (GCP Batch
@@ -95,15 +94,15 @@ def create_uworker_main_batch_job(self, module: str, job_type: str,
     return self._gcp_batch_service.create_uworker_main_batch_job(
         module, job_type, input_download_url)
 
-  def create_uworker_main_batch_jobs(self, batch_tasks: List[RemoteTask]):
+  def create_uworker_main_batch_jobs(self, remote_tasks: List[RemoteTask]):
     """Creates batch jobs on either GCP Batch or Kubernetes.
     
     The tasks are grouped by their target backend (GCP Batch or Kubernetes) and
     then created in separate batches.
     """
     gcp_batch_tasks = []
     kubernetes_tasks = []
-    for task in batch_tasks:
+    for task in remote_tasks:
       if self._should_use_kubernetes(task.job_type):
         kubernetes_tasks.append(task)
       else: