add gcp batch feature flag

Signed-off-by: Javan Lacerda <javanlacerda@google.com>

Author: javanlacerda

src/clusterfuzz/_internal/base/errors.py

@@ -115,5 +115,5 @@ def __str__(self):
 class QueueLimitReachedError(Error):
   """Error thrown when the queue limit is reached."""
 
-  def __init__(self, size):
-    super().__init__(f'Queue limit reached {size}.')
+  def __init__(self, size, queue):
+    super().__init__(f'Queue {queue} has reached the limit of {size}.')

src/clusterfuzz/_internal/base/feature_flags.py

@@ -31,6 +31,8 @@ class FeatureFlags(Enum):
 
   UTASK_MAIN_QUEUE_LIMIT = 'utask_main_queue_limit'
 
+  GCP_BATCH_JOBS_FREQUENCY = 'gcp_batch_jobs_frequency'
+
   @property
   def flag(self):
     """Get the feature flag."""

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

@@ -109,6 +109,12 @@ def get_task_duration(command):
 # scheduling on batch.
 MAX_UTASKS = 3000
 
+# Time window to get the metrics.
+# We should look for metrics in
+# start = now - _QUEUE_LIMIT_INTERVAL
+# end = now
+_QUEUE_LIMIT_INTERVAL = 5 * 60  # 5 minutes.
+
 
 class Error(Exception):
   """Base exception class."""
@@ -707,7 +713,7 @@ def get_utask_main_queue_size():
     now = time.time()
     interval = monitoring_v3.TimeInterval(
         end_time={'seconds': int(now)},
-        start_time={'seconds': int(now - 5 * 60)},
+        start_time={'seconds': int(now - _QUEUE_LIMIT_INTERVAL)},
     )
 
     results = client.list_time_series(

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

@@ -20,6 +20,7 @@
 from clusterfuzz._internal.base import feature_flags
 from clusterfuzz._internal.base import tasks
 from clusterfuzz._internal.base.tasks import task_utils
+from clusterfuzz._internal.base.tasks import UTASK_MAIN_QUEUE
 from clusterfuzz._internal.batch import service as batch_service
 from clusterfuzz._internal.bot.tasks import utasks
 from clusterfuzz._internal.metrics import events
@@ -165,7 +166,10 @@ def execute(self, task_argument, job_type, uworker_env):
     if utask_flag and utask_flag.enabled:
       utask_main_queue_limit = utask_flag.content
     if utask_main_queue_size > utask_main_queue_limit:
-      raise errors.QueueLimitReachedError(utask_main_queue_size)
+      base_os_version = environment.get_value('BASE_OS_VERSION')
+      queue_name = UTASK_MAIN_QUEUE if not base_os_version else \
+        f'{UTASK_MAIN_QUEUE}-{base_os_version}'
+      raise errors.QueueLimitReachedError(utask_main_queue_size, queue_name)
 
     logs.info('Preprocessing utask.')
     download_url = self.preprocess(task_argument, job_type, uworker_env)

src/clusterfuzz/_internal/remote_task/remote_task_adapters.py

@@ -35,7 +35,8 @@ class RemoteTaskAdapters(Enum):
   """
   KUBERNETES = ('kubernetes', k8s_service.KubernetesService,
                 feature_flags.FeatureFlags.K8S_JOBS_FREQUENCY, 0.0)
-  GCP_BATCH = ('gcp_batch', batch_service.GcpBatchService, None, 1.0)
+  GCP_BATCH = ('gcp_batch', batch_service.GcpBatchService,
+               feature_flags.FeatureFlags.GCP_BATCH_JOBS_FREQUENCY, 1.0)
 
   def __init__(self, adapter_id, service, feature_flag, default_weight):
     self.id = adapter_id