get rid job limiter

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

Author: javanlacerda

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

@@ -509,7 +509,6 @@ def __init__(self, pubsub_message):
     }
 
     self.eta = datetime.datetime.utcfromtimestamp(float(self.attribute('eta')))
-    self.do_not_ack = False
 
   def attribute(self, key):
     """Return attribute value."""
@@ -555,10 +554,9 @@ def lease(self, _event=None):  # pylint: disable=arguments-differ
     finally:
       _event.set()
       leaser_thread.join()
-
+    
     # If we get here the task succeeded in running. Acknowledge the message.
-    if not self.do_not_ack:
-      self._pubsub_message.ack()
+    self._pubsub_message.ack()  
     track_task_end()
 
   def dont_retry(self):
@@ -595,8 +593,7 @@ def lease(self, _event=None):  # pylint: disable=arguments-differ
       leaser_thread.join()
 
     # If we get here the task succeeded in running. Acknowledge the message.
-    if not self.do_not_ack:
-      self._pubsub_message.ack()
+    self._pubsub_message.ack()
     track_task_end()
 
 
@@ -739,7 +736,6 @@ def __init__(self,
     grandparent_class.__init__(command, output_url_argument, job_type, eta,
                                is_command_override, high_end)
     self._pubsub_message = pubsub_message
-    self.do_not_ack = False
 
   def ack(self):
     self._pubsub_message.ack()

src/clusterfuzz/_internal/datastore/feature_flags.py

@@ -27,7 +27,6 @@ class FeatureFlags(Enum):
   TEST_FLOAT_FLAG = 'test_float_flag'
 
   K8S_JOBS_FREQUENCY = 'k8s_jobs_frequency'
-  K8S_PENDING_JOBS_LIMITER = 'k8s_pending_jobs_limiter'
 
   @property
   def flag(self):

src/clusterfuzz/_internal/k8s/service.py

@@ -31,7 +31,6 @@
 from clusterfuzz._internal.base.tasks import task_utils
 from clusterfuzz._internal.config import local_config
 from clusterfuzz._internal.datastore import data_types
-from clusterfuzz._internal.datastore import feature_flags
 from clusterfuzz._internal.datastore import ndb_utils
 from clusterfuzz._internal.metrics import logs
 from clusterfuzz._internal.remote_task import remote_task_types
@@ -295,19 +294,6 @@ def create_utask_main_jobs(
     separate batch job for each group. This allows tasks with similar
     requirements to be processed together, which can improve efficiency.
     """
-    limit = None
-    if feature_flags.FeatureFlags.K8S_PENDING_JOBS_LIMITER.enabled:
-      limit = feature_flags.FeatureFlags.K8S_PENDING_JOBS_LIMITER.content
-
-    if limit is not None and self._get_pending_jobs_count() >= int(limit):
-      logs.warning(
-          f'Kubernetes job limit reached. Not acking {len(remote_tasks)} tasks.'
-      )
-      for task in remote_tasks:
-        if task.pubsub_task:
-          task.pubsub_task.do_not_ack = True
-      return []
-
     job_specs = collections.defaultdict(list)
     configs = _get_k8s_job_configs(remote_tasks)
     for remote_task in remote_tasks:

src/clusterfuzz/_internal/remote_task/remote_task_gate.py

@@ -149,7 +149,10 @@ def create_utask_main_jobs(self,
     results = []
     for adapter_id, tasks in tasks_by_adapter.items():
       if tasks:
-        logs.info(f'Sending {len(tasks)} tasks to {adapter_id}.')
-        service = self._service_map[adapter_id]
-        results.extend(service.create_utask_main_jobs(tasks))
+        try:
+          logs.info(f'Sending {len(tasks)} tasks to {adapter_id}.')
+          service = self._service_map[adapter_id]
+          results.extend(service.create_utask_main_jobs(tasks))
+        except Exception:  # pylint: disable=broad-except
+          logs.error(f'Failed to send {len(tasks)} tasks to {adapter_id}.')
     return results

src/clusterfuzz/_internal/tests/core/k8s/k8s_service_limit_test.py

@@ -17,7 +17,6 @@
 from unittest import mock
 
 from clusterfuzz._internal.datastore import data_types
-from clusterfuzz._internal.datastore import feature_flags
 from clusterfuzz._internal.k8s import service
 from clusterfuzz._internal.remote_task import remote_task_types
 from clusterfuzz._internal.tests.test_libs import test_utils
@@ -42,11 +41,6 @@ def setUp(self):
         name='job2', platform='LINUX',
         environment_string='CUSTOM_VAR = value').put()
 
-    data_types.FeatureFlag(
-        id=feature_flags.FeatureFlags.K8S_PENDING_JOBS_LIMITER.value,
-        enabled=True,
-        value=100.0).put()
-
   @mock.patch.object(service.KubernetesService, '_get_pending_jobs_count')
   @mock.patch.object(service.KubernetesService, 'create_job')
   def test_create_uworker_main_batch_jobs(self, mock_create_job,
@@ -85,22 +79,6 @@ def test_get_pending_jobs_count(self, mock_core_api_cls, _):
         label_selector='app.kubernetes.io/name=clusterfuzz-kata-job',
         field_selector='status.phase=Pending')
 
-  @mock.patch.object(service.KubernetesService, '_get_pending_jobs_count')
-  def test_create_uworker_main_batch_jobs_limit_reached(
-      self, mock_get_pending_count, _):
-    """Tests that create_utask_main_jobs nacks when limit reached."""
-    mock_get_pending_count.return_value = 101
-    kube_service = service.KubernetesService()
-
-    mock_pubsub_task = mock.Mock()
-    mock_pubsub_task.do_not_ack = False
-    task = remote_task_types.RemoteTask(
-        'fuzz', 'job1', 'url1', pubsub_task=mock_pubsub_task)
-
-    result = kube_service.create_utask_main_jobs([task])
-    self.assertEqual(result, [])
-    self.assertTrue(mock_pubsub_task.do_not_ack)
-
   @mock.patch('kubernetes.client.BatchV1Api')
   def test_create_job_kata(self, mock_batch_api_cls, _):
     """Tests that create_job generates the correct spec for Kata."""

src/clusterfuzz/_internal/tests/core/k8s/k8s_service_test.py

@@ -98,11 +98,9 @@ def test_create_uworker_main_batch_jobs_success(self, _, mock_k8s_create,
     """Test successful creation."""
     mock_get_frequency.return_value = {'kubernetes': 1.0}
     mock_pubsub_task = mock.Mock()
-    mock_pubsub_task.do_not_ack = False
     task = remote_task_types.RemoteTask(
         'fuzz', 'job1', 'url1', pubsub_task=mock_pubsub_task)
 
     self.gate.create_utask_main_jobs([task])
 
     self.assertTrue(mock_k8s_create.called)
-    self.assertFalse(mock_pubsub_task.do_not_ack)