Revert "Swarming: Implement JSON env var encapsulation and v2 API protos" (#5188)

Reverts #5170

Author: jardondiego

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

@@ -115,8 +115,8 @@ def is_remote_utask(command, job):
     # Return True even if we can't query the db.
     return True
 
-  return batch_service.is_remote_task(command,
-                                      job) or swarming.is_swarming_task(job)
+  return batch_service.is_remote_task(
+      command, job) or swarming.is_swarming_task(command, job)
 
 
 def task_main_runs_on_uworker():
@@ -178,12 +178,11 @@ def execute(self, task_argument, job_type, uworker_env):
       return
 
     logs.info('Queueing utask for remote execution.', download_url=download_url)
-    if request := swarming.create_new_task_request(command, job_type,
-                                                   download_url):
-      swarming.push_swarming_task(request)
-    else:
-      assert batch_service.is_remote_task(command, job_type)
+    if batch_service.is_remote_task(command, job_type):
       tasks.add_utask_main(command, download_url, job_type)
+    else:
+      assert swarming.is_swarming_task(command, job_type)
+      swarming.push_swarming_task(command, download_url, job_type)
 
   @logs.task_stage_context(logs.Stage.PREPROCESS)
   def preprocess(self, task_argument, job_type, uworker_env):

src/clusterfuzz/_internal/bot/tasks/utasks/__init__.py

@@ -63,9 +63,10 @@ def _timestamp_now() -> Timestamp:
   return ts
 
 
-def _get_execution_mode(job_type):
+def _get_execution_mode(utask_module, job_type):
   """Determines whether this task in executed on swarming or batch."""
-  if swarming.is_swarming_task(job_type):
+  command = task_utils.get_command_from_module(utask_module.__name__)
+  if swarming.is_swarming_task(command, job_type):
     return Mode.SWARMING
   return Mode.BATCH
 
@@ -409,7 +410,7 @@ def tworker_preprocess(utask_module, task_argument, job_type, uworker_env):
   signed download URL for the uworker's input and the (unsigned) download URL
   for its output."""
   with _MetricRecorder(_Subtask.PREPROCESS) as recorder:
-    execution_mode = _get_execution_mode(job_type)
+    execution_mode = _get_execution_mode(utask_module, job_type)
     uworker_input = _preprocess(utask_module, task_argument, job_type,
                                 uworker_env, recorder, execution_mode)
     if not uworker_input:
@@ -500,7 +501,8 @@ def tworker_postprocess(output_download_url) -> None:
     task_utils.reset_task_stage_env()
 
     utask_module = get_utask_module(uworker_output.uworker_input.module_name)
-    execution_mode = _get_execution_mode(uworker_output.uworker_input.job_type)
+    execution_mode = _get_execution_mode(utask_module,
+                                         uworker_output.uworker_input.job_type)
     recorder.set_task_details(
         utask_module,
         uworker_output.uworker_input.job_type,