Kubernetes Job service

.github/workflows/kubernetes-e2e-tests.yaml

@@ -0,0 +1,42 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: Run Kubernetes e2e tests
+on: [pull_request]
+
+permissions: read-all
+
+jobs:
+  build:
+    runs-on: ubuntu-24.04
+
+    steps:
+      - uses: actions/checkout@v4
+      - run: |  # Needed for git diff to work.
+          git fetch origin master --depth 1
+          git symbolic-ref refs/remotes/origin/HEAD refs/remotes/origin/master
+
+      - name: Setup python environment
+        uses: actions/setup-python@b55428b1882923874294fa556849718a1d7f2ca5
+        with:
+          python-version: 3.11
+
+      - name: Set up JDK 21
+        uses: actions/setup-java@v4
+        with:
+          java-version: '21'
+          distribution: 'temurin'
+
+      - name: Run Kubernetes e2e tests
+        run: ./local/tests/kubernetes_e2e_test.bash

src/clusterfuzz/_internal/batch/kubernetes.py → local/tests/kubernetes_e2e_test.bash

@@ -1,3 +1,5 @@
+#!/bin/bash -ex
+#
 # Copyright 2025 Google LLC
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -11,17 +13,17 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-"""Kubernetes batch client."""
-from clusterfuzz._internal.remote_task import RemoteTaskInterface
 
+# This script is for running the Kubernetes end-to-end test in CI.
+
+pip install pipenv
+
+# Install dependencies.
+pipenv --python 3.11
+pipenv install
 
-class KubernetesJobClient(RemoteTaskInterface):
-  """A remote task execution client for Kubernetes.
-  
-  This class is a placeholder for a future implementation of a remote task
-  execution client that uses Kubernetes. It is not yet implemented.
-  """
+./local/install_deps.bash
 
-  def create_job(self, spec, input_urls):
-    """Creates a Kubernetes job."""
-    raise NotImplementedError('Kubernetes batch client is not implemented yet.')
+# Run the test.
+export K8S_E2E=1
+pipenv run python butler.py py_unittest -t core -p k8s_service_e2e_test.py

src/Pipfile

@@ -27,14 +27,16 @@ google-crc32c = "==1.5.0"
 grpcio = "==1.62.2"
 httplib2 = "==0.19.0"
 jira = "==2.0.0"
+Jinja2 = "==3.1.4"
+kubernetes = "==34.1.0"
 mozprocess = "==1.3.1"
 oauth2client = "==4.1.3"
 psutil = "==5.9.4"
 protobuf = "==4.23.4"
 pygithub = "==1.55"
 pyOpenSSL = "==22.0.0"
 python-dateutil = "==2.8.1"
-PyYAML = "==6.0"
+PyYAML = "==6.0.1"
 pytz = "==2023.3"
 redis = "==4.6.0"
 requests = "==2.21.0"
@@ -49,7 +51,6 @@ charset-normalizer = "==3.3.2"
 firebase-admin = "==6.2.0"
 Flask = "==2.2.2"
 itsdangerous = "==2.0.1"
-Jinja2 = "==3.1.4"
 PyJWT = "==2.7.0"
 requests-toolbelt = "==0.9.1"
 werkzeug = "==2.2.2"

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

@@ -64,6 +64,12 @@
     'regression': 24 * 60 * 60,
 }
 
+
+def get_task_duration(command):
+  """Gets the duration of a task."""
+  return TASK_LEASE_SECONDS_BY_COMMAND.get(command, TASK_LEASE_SECONDS)
+
+
 TASK_QUEUE_DISPLAY_NAMES = {
     'LINUX': 'Linux',
     'LINUX_WITH_GPU': 'Linux (with GPU)',
@@ -503,6 +509,7 @@ 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."""
@@ -550,7 +557,8 @@ def lease(self, _event=None):  # pylint: disable=arguments-differ
       leaser_thread.join()
 
     # If we get here the task succeeded in running. Acknowledge the message.
-    self._pubsub_message.ack()
+    if not self.do_not_ack:
+      self._pubsub_message.ack()
     track_task_end()
 
   def dont_retry(self):
@@ -587,7 +595,8 @@ def lease(self, _event=None):  # pylint: disable=arguments-differ
       leaser_thread.join()
 
     # If we get here the task succeeded in running. Acknowledge the message.
-    self._pubsub_message.ack()
+    if not self.do_not_ack:
+      self._pubsub_message.ack()
     track_task_end()
 
 
@@ -730,6 +739,7 @@ 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()