Feature/kubernetes job client

.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@v3
+      - 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@v3
+        with:
+          java-version: '21'
+          distribution: 'temurin'
+
+      - name: Run Kubernetes e2e tests
+        run: ./local/tests/kubernetes_e2e_test.bash

Pipfile

@@ -10,6 +10,19 @@ future = "==0.17.1"
 protobuf = "==4.23.4"
 psutil = "==5.9.4"
 google-cloud-ndb = "==2.3.4"
+kubernetes = "==34.1.0"
+google-api-python-client = "==2.93.0"
+aiohttp = "==3.10.5"
+google-cloud-storage = "==2.10.0"
+google-cloud-secret-manager = "==2.17.0"
+google-cloud-logging = "==3.6.0"
+google-cloud-monitoring = "==2.15.1"
+google-cloud-datastore = "==2.16.1"
+oauth2client = "==4.1.3"
+requests = "==2.21.0"
+PyYAML = "==6.0"
+httplib2 = "==0.19.0"
+google-auth-oauthlib = "==0.4.1"
 
 [dev-packages]
 Fabric = "==1.14.1"

butler.py

@@ -436,6 +436,11 @@ def main():
   parser_clean_indexes.add_argument(
       '-c', '--config-dir', required=True, help='Path to application config.')
 
+  parser_clean_jobs = subparsers.add_parser(
+      'clean_jobs', help='Clean up pending kubernetes jobs.')
+  parser_clean_jobs.add_argument(
+      '-c', '--config-dir', required=True, help='Path to application config.')
+
   parser_create_config = subparsers.add_parser(
       'create_config', help='Create a new deployment config.')
   parser_create_config.add_argument(

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

out.log

@@ -0,0 +1,7 @@
+Running in dry-run mode, no datastore writes are committed. For permanent modifications, re-run with --non-dry-run.
+Attempting to combine batch tasks.
+Combining 2901 batch tasks.
+K8s result: ['libfuzzer-chrome-asan-debug-7ca9c46f']
+
+Please remember to run the migration individually on all projects.
+

src/Pipfile

@@ -27,14 +27,15 @@ google-crc32c = "==1.5.0"
 grpcio = "==1.62.2"
 httplib2 = "==0.19.0"
 jira = "==2.0.0"
+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"

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()