kbuild: Support persistent http for uploads, speedup

kernelci/api/__init__.py

@@ -382,7 +382,11 @@ def find(
 
         @abc.abstractmethod
         def stats(self, attributes: Dict[str, str]) -> list:
-            """Get aggregated telemetry statistics"""
+            """Return aggregated telemetry statistics for matched events.
+
+            The result contains grouped counters only; it does not run any
+            anomaly detection.
+            """
 
     @property
     def telemetry(self) -> Telemetry:

kernelci/api/latest.py

@@ -168,7 +168,11 @@ def find(
             return self._get_paginated(params, 'telemetry', offset, limit)
 
         def stats(self, attributes: Dict[str, str]) -> list:
-            """Get aggregated telemetry statistics"""
+            """Return aggregated telemetry statistics for matched events.
+
+            This endpoint only performs grouped aggregation of telemetry counters
+            and does not perform anomaly detection.
+            """
             params = attributes.copy() if attributes else {}
             return self._get('telemetry/stats', params=params).json()
 

kernelci/kbuild.py

@@ -1,4 +1,4 @@
 # SPDX-License-Identifier: LGPL-2.1-or-later
 #
 # Copyright (C) 2023 Collabora Limited
 # Author: Denys Fedoryshchenko <denys.f@collabora.com>
@@ -44,12 +44,12 @@
 from typing import Dict, Tuple, List
 
 CIP_CONFIG_URL = \
     "https://gitlab.com/cip-project/cip-kernel/cip-kernel-config/-/raw/master/{branch}/{config}"  # noqa
 CROS_CONFIG_URL = \
     "https://chromium.googlesource.com/chromiumos/third_party/kernel/+archive/refs/heads/{branch}/chromeos/config.tar.gz"  # noqa
 FW_GIT = "https://git.kernel.org/pub/scm/linux/kernel/git/firmware/linux-firmware.git"  # noqa
 
 # TODO: find a way to automatically fetch this information
 LATEST_LTS_MAJOR = 6
 LATEST_LTS_MINOR = 12
 
@@ -106,19 +106,19 @@
 REDIR = ' > >(tee {}) 2> >(tee {} >&1)'
 
 
 def _download_file_inner(url, file_path):
     try:
         r = requests.get(url, stream=True, timeout=60)
     except requests.exceptions.RequestException as e:
         print(f"[_download_file_inner] Error: {e}")
         return False
     except requests.exceptions.Timeout as e:
         print(f"[_download_file_inner] Timeout: {e}")
         return False
     except requests.exceptions.ConnectionError as e:
         print(f"[_download_file_inner] Connection error: {e}")
         return False
     except Exception as e:
         print(f"[_download_file_inner] Exception: {e}")
         return False
     if r.status_code == 200:
@@ -143,7 +143,7 @@
     return False
 
 
 class KBuild():
     '''
     Build class that represents kernel build
     if node, jobname and params are provided, create new build object
@@ -626,8 +626,8 @@
         for i in range(0, fragnum):
             self.addcmd("./scripts/kconfig/merge_config.sh" +
                         f" -m .config {self._fragments_dir}/{i}.config")
         # TODO: olddefconfig should be optional/configurable
         # TODO: log all warnings/errors of olddefconfig to separate file
         self.addcmd("make olddefconfig")
         self.addcmd(f"cp .config {self._af_dir}/")
         self.addcmd("cd ..")
@@ -640,7 +640,7 @@
         fragnum = self._parse_fragments(firmware=True)
         self._merge_frags(fragnum)
         if not self._dtbs_check:
             # TODO: verify if CONFIG_EXTRA_FIRMWARE have any files
             # We can check that if fragments have CONFIG_EXTRA_FIRMWARE
             self._fetch_firmware()
             self._build_kernel()
@@ -903,6 +903,9 @@
         '''
         Get storage object
         '''
+        if hasattr(self, '_storage') and self._storage is not None:
+            return self._storage
+
         storage_config = kernelci.config.storage.StorageFactory.from_yaml(
             name='storage', config=yaml.safe_load(self._storage_config)
         )
@@ -911,7 +914,8 @@
         storage_cred = os.getenv('KCI_STORAGE_CREDENTIALS')
         if not storage_cred:
             raise ValueError("KCI_STORAGE_CREDENTIALS not set")
-        return kernelci.storage.get_storage(storage_config, storage_cred)
+        self._storage = kernelci.storage.get_storage(storage_config, storage_cred)
+        return self._storage
 
     def map_artifact_name(self, artifact):
         '''
@@ -980,12 +984,12 @@
                     os.unlink(upload_path)
 
                 print(f"[_upload_artifacts] Uploaded {artifact} to {stored_url}")
                 return artifact, stored_url, None
             except Exception as e:
                 print(f"[_upload_artifacts] Error uploading {artifact}: {e}")
                 if compressed_file and os.path.exists(upload_path):
                     os.unlink(upload_path)
                 return artifact, None, str(e)
 
         # Process uploads in parallel
         max_workers = min(10, len(upload_tasks))  # Limit concurrent uploads

kernelci/storage/backend.py

@@ -11,6 +11,7 @@
 import time
 from urllib.parse import urljoin
 import requests
+from requests.adapters import HTTPAdapter
 from . import Storage
 
 
@@ -20,13 +21,19 @@ class StorageBackend(Storage):
     This class implements the Storage interface for the kernelci-backend API.
     It requires an API token as credentials.
     """
+    _HTTP_TIMEOUT = 300
+    _HTTP_POOL_SIZE = 20
 
     def _close_files(self, files):
         """Helper to close all file handles in the files dictionary."""
         for file_tuple in files.values():
             if hasattr(file_tuple[1], 'close'):
                 file_tuple[1].close()
 
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._session = None
+
     def _handle_http_error(self, exc, attempt, max_retries, retry_delay):
         """Handle HTTP errors during upload with retry logic."""
         # Only retry on server errors (5xx status codes)
@@ -57,7 +64,17 @@ def _handle_network_error(self, exc, attempt, max_retries, retry_delay):
             print(f"Upload failed after {max_retries} attempts")
         return exc
 
+    def _connect(self):
+        if self._session is not None:
+            return
+        self._session = requests.Session()
+        adapter = HTTPAdapter(pool_connections=self._HTTP_POOL_SIZE,
+                             pool_maxsize=self._HTTP_POOL_SIZE)
+        self._session.mount('https://', adapter)
+        self._session.mount('http://', adapter)
+
     def _upload(self, file_paths, dest_path):
+        self._connect()
         headers = {
             'Authorization': self.credentials,
         }
@@ -78,8 +95,9 @@ def _upload(self, file_paths, dest_path):
                 }
 
                 url = urljoin(self.config.api_url, 'upload')
-                resp = requests.post(
-                    url, headers=headers, data=data, files=files, timeout=300
+                resp = self._session.post(
+                    url, headers=headers, data=data, files=files,
+                    timeout=self._HTTP_TIMEOUT
                 )
                 resp.raise_for_status()