Merge pull request ClickHouse#78662 from ClickHouse/ci_compress_artifacts

maxknv · web-flow · commit 534587921328 · 2025-04-04T18:02:25.000Z
CI: Support artifacts compressing
diff --git a/ci/defs/defs.py b/ci/defs/defs.py
@@ -473,6 +473,7 @@ class ArtifactConfigs:
         name="...",
         type=Artifact.Type.S3,
         path=f"{TEMP_DIR}/build/src/unit_tests_dbms",
+        compress_zst=True,
     ).parametrize(
         names=[
             ArtifactNames.UNITTEST_AMD_ASAN,
diff --git a/ci/jobs/performance_tests.py b/ci/jobs/performance_tests.py
@@ -6,7 +6,6 @@
 import traceback
 from pathlib import Path
 
-from ci.jobs.scripts.clickhouse_version import CHVersion
 from ci.praktika.info import Info
 from ci.praktika.result import Result
 from ci.praktika.utils import MetaClasses, Shell, Utils
@@ -247,7 +246,7 @@ def main():
         compare_against_master or compare_against_release
     ), "test option: head_master or prev_release must be selected"
 
-    release_version = CHVersion.get_release_version_as_dict()
+    # release_version = CHVersion.get_release_version_as_dict()
     info = Info()
 
     if Utils.is_arm():
diff --git a/ci/jobs/scripts/workflow_hooks/filter_job.py b/ci/jobs/scripts/workflow_hooks/filter_job.py
@@ -1,7 +1,6 @@
-from ci.praktika.info import Info
-
 from ci.defs.defs import JobNames
 from ci.jobs.scripts.workflow_hooks.pr_description import Labels
+from ci.praktika.info import Info
 
 
 def only_docs(changed_files):
diff --git a/ci/praktika/artifact.py b/ci/praktika/artifact.py
@@ -20,6 +20,7 @@ class Config:
         name: str
         type: str
         path: Union[str, List[str]]
+        compress_zst: bool = False
         _provided_by: str = ""
         _s3_path: str = ""
 
diff --git a/ci/praktika/cache.py b/ci/praktika/cache.py
@@ -54,7 +54,10 @@ def push_success_record(cls, job_name, job_digest, sha, if_not_exist):
         record_file = Path(Settings.TEMP_DIR) / type_
         record.dump(record_file)
         S3.put(
-            s3_path=record_path, local_path=record_file, if_none_matched=if_not_exist
+            s3_path=record_path,
+            local_path=record_file,
+            if_none_matched=if_not_exist,
+            no_strict=True,
         )
         record_file.unlink()
 
@@ -74,6 +77,7 @@ def fetch_success(self, job_name, job_digest):
             s3_path=record_path,
             local_path=record_file_local_dir,
             _skip_download_counter=True,
+            no_strict=True,
         )
 
         if res:
diff --git a/ci/praktika/digest.py b/ci/praktika/digest.py
@@ -6,12 +6,10 @@
 from pathlib import Path
 from typing import List
 
-from praktika.utils import Shell
-
-from . import Job
 from .docker import Docker
+from .job import Job
 from .settings import Settings
-from .utils import Utils
+from .utils import Shell, Utils
 
 
 class Digest:
@@ -30,7 +28,7 @@ def _hash_digest_config(digest_config: Job.CacheDigestConfig) -> str:
     def get_null_digest(cls):
         return "f" * Settings.CACHE_DIGEST_LEN
 
-    def calc_job_digest(self, job_config: Job.Config, docker_digests):
+    def calc_job_digest(self, job_config: Job.Config, docker_digests, artifact_configs):
         config = job_config.digest_config
         if not config:
             return self.get_null_digest()
@@ -72,6 +70,7 @@ def calc_job_digest(self, job_config: Job.Config, docker_digests):
             digest = "-".join([docker_digest, digest])
 
         job_config_dict = dataclasses.asdict(job_config)
+
         drop_fields = [
             "requires",
             "enable_commit_status",
@@ -80,6 +79,15 @@ def calc_job_digest(self, job_config: Job.Config, docker_digests):
         filtered_job_dict = {
             k: v for k, v in job_config_dict.items() if k not in drop_fields
         }
+        # add Articat.Configs list to the job config dict so that changed Articat.Config object affects job digest
+        job_provides_artifact_configs = []
+        for a in job_config.provides:
+            if a in artifact_configs:
+                job_provides_artifact_configs.append(
+                    dataclasses.asdict(artifact_configs[a])
+                )
+        filtered_job_dict["provides"] = job_provides_artifact_configs
+
         config_digest = hashlib.md5(
             json.dumps(filtered_job_dict, sort_keys=True).encode()
         ).hexdigest()[: min(Settings.CACHE_DIGEST_LEN // 4, 4)]
diff --git a/ci/praktika/gh.py b/ci/praktika/gh.py
@@ -121,7 +121,7 @@ def get_pr_title_body_labels(cls, pr=None, repo=None):
         except Exception:
             print("ERROR: Failed to get PR data")
             traceback.print_exc()
-            Info().store_exception_traceback()
+            Info().store_traceback()
             return "", "", []
         return title, body, labels
 
diff --git a/ci/praktika/hook_cache.py b/ci/praktika/hook_cache.py
@@ -20,10 +20,15 @@ def configure(cls, workflow):
         ), f"Outdated yaml pipelines or BUG. Configuration must be run only for workflow with enabled cache, workflow [{workflow.name}]"
         artifact_digest_map = {}
         job_digest_map = {}
+        artifact_name_config_map = {}
+        for a in workflow.artifacts:
+            artifact_name_config_map[a.name] = a
 
         for job in workflow.jobs:
             digest = cache.digest.calc_job_digest(
-                job_config=job, docker_digests=docker_digests
+                job_config=job,
+                docker_digests=docker_digests,
+                artifact_configs=artifact_name_config_map,
             )
             job_digest_map[job.name] = digest
             if job.provides:
diff --git a/ci/praktika/hook_html.py b/ci/praktika/hook_html.py
@@ -99,7 +99,9 @@ def pull_from_s3(cls):
         local_path = Path(cls.file_name())
         file_name = local_path.name
         s3_path = f"{cls.get_s3_path()}/{file_name}"
-        if not S3.copy_file_from_s3(s3_path=s3_path, local_path=local_path):
+        if not S3.copy_file_from_s3(
+            s3_path=s3_path, local_path=local_path, no_strict=True
+        ):
             print(f"WARNING: failed to cp file [{s3_path}] from s3")
             return []
         return cls.from_json(local_path)
@@ -111,7 +113,9 @@ def push_to_s3(cls, commits):
         local_path = Path(cls.file_name())
         file_name = local_path.name
         s3_path = f"{cls.get_s3_path()}/{file_name}"
-        if not S3.copy_file_to_s3(s3_path=s3_path, local_path=local_path, text=True):
+        if not S3.copy_file_to_s3(
+            s3_path=s3_path, local_path=local_path, text=True, no_strict=True
+        ):
             print(f"WARNING: failed to cp file [{local_path}] to s3")
 
     @classmethod
diff --git a/ci/praktika/info.py b/ci/praktika/info.py
@@ -192,7 +192,7 @@ def get_custom_data(self, key=None):
             return custom_data.get(key, None)
         return custom_data
 
-    def store_exception_traceback(self):
+    def store_traceback(self):
         self.env.TRACEBACKS.append(traceback.format_exc())
         self.env.dump()
 
diff --git a/ci/praktika/job.py b/ci/praktika/job.py
@@ -3,8 +3,8 @@
 from dataclasses import dataclass, field
 from typing import Any, List, Optional
 
-from praktika import Artifact
-from praktika.utils import Utils
+from . import Artifact
+from .utils import Utils
 
 
 class Job:
diff --git a/ci/praktika/json.html b/ci/praktika/json.html
@@ -774,6 +774,7 @@
         function drawTimeline() {
             ctx.clearRect(0, 0, canvas.width, canvas.height);
             let taskRects = [];
+            const failedMarkers = [];
 
             const minStartTime = runtimeCache.json0.start_time;
             const currentUtcTimestamp = Date.now() / 1000;
@@ -804,8 +805,20 @@
                 ctx.fillRect(x, y, width, height);
 
                 taskRects.push({ x, y, width, height, task });
+
+                // Save failure marker position for second pass
+                if (task.status === "failure" || task.status === "error") {
+                    failedMarkers.push({ x: x + width, y: y + height / 2 });
+                }
+            });
+
+            // Draw failure markers on top
+            failedMarkers.forEach(({ x, y }) => {
+                ctx.font = "12px sans-serif";
+                ctx.fillText("❌", Math.max(x - 14, 0), y + 4);
             });
 
+            // Event listeners
             canvas.addEventListener("mousemove", (event) => {
                 const rect = canvas.getBoundingClientRect();
                 const mouseY = event.clientY - rect.top;
diff --git a/ci/praktika/result.py b/ci/praktika/result.py
@@ -521,15 +521,11 @@ def copy_result_to_s3(cls, result, clean=False):
         return url
 
     @classmethod
-    def copy_result_from_s3(cls, local_path, lock=False):
+    def copy_result_from_s3(cls, local_path):
         env = _Environment.get()
         file_name = Path(local_path).name
         s3_path = f"{Settings.HTML_S3_PATH}/{env.get_s3_prefix()}/{file_name}"
-        # if lock:
-        #     cls.lock(s3_path)
-        if not S3.copy_file_from_s3(s3_path=s3_path, local_path=local_path):
-            print(f"ERROR: failed to cp file [{s3_path}] from s3")
-            raise
+        S3.copy_file_from_s3(s3_path=s3_path, local_path=local_path)
 
     @classmethod
     def copy_result_from_s3_with_version(cls, local_path):
@@ -540,11 +536,9 @@ def copy_result_from_s3_with_version(cls, local_path):
         for file_path in local_dir.glob(file_name_pattern):
             file_path.unlink()
         s3_path = f"{Settings.HTML_S3_PATH}/{env.get_s3_prefix()}/"
-        if not S3.copy_file_from_s3_matching_pattern(
+        S3.copy_file_from_s3_matching_pattern(
             s3_path=s3_path, local_path=local_dir, include=file_name_pattern
-        ):
-            print(f"ERROR: failed to cp file [{s3_path}] from s3")
-            raise
+        )
         result_files = []
         for file_path in local_dir.glob(file_name_pattern):
             result_files.append(file_path)
@@ -555,7 +549,7 @@ def copy_result_from_s3_with_version(cls, local_path):
         return version
 
     @classmethod
-    def copy_result_to_s3_with_version(cls, result, version):
+    def copy_result_to_s3_with_version(cls, result, version, no_strict=False):
         result.dump()
         filename = Path(result.file_name()).name
         file_name_versioned = f"{filename}_{str(version).zfill(3)}"
@@ -568,10 +562,13 @@ def copy_result_to_s3_with_version(cls, result, version):
             s3_path=s3_path_versioned,
             local_path=result.file_name(),
             if_none_matched=True,
+            no_strict=no_strict,
         ):
             print("Failed to put versioned Result")
             return False
-        if not S3.put(s3_path=s3_path, local_path=result.file_name()):
+        if not S3.put(
+            s3_path=s3_path, local_path=result.file_name(), no_strict=no_strict
+        ):
             print("Failed to put non-versioned Result")
         return True
 
@@ -629,8 +626,9 @@ def update_workflow_results(
         attempt = 1
         prev_status = ""
         new_status = ""
-        done = False
-        while attempt < 50:
+        MAX_ATTEMPTS = 50
+
+        while attempt < MAX_ATTEMPTS:
             version = cls.copy_result_from_s3_with_version(
                 Result.file_name_static(workflow_name)
             )
@@ -652,15 +650,17 @@ def update_workflow_results(
                 workflow_result.ext["storage_usage"] = workflow_storage_usage
 
             new_status = workflow_result.status
-            if cls.copy_result_to_s3_with_version(workflow_result, version=version + 1):
-                done = True
+            if cls.copy_result_to_s3_with_version(
+                workflow_result,
+                version=version + 1,
+                no_strict=attempt < MAX_ATTEMPTS - 1,
+            ):
                 break
             print(f"Attempt [{attempt}] to upload workflow result failed")
             attempt += 1
             # random delay (0-2s) to reduce contention and minimize race conditions
             # when multiple concurrent jobs attempt to update the workflow report
             time.sleep(random.uniform(0, 2))
-        assert done
 
         if prev_status != new_status:
             return new_status
diff --git a/ci/praktika/runner.py b/ci/praktika/runner.py
diff --git a/ci/praktika/s3.py b/ci/praktika/s3.py
diff --git a/ci/praktika/utils.py b/ci/praktika/utils.py
