Merge pull request ClickHouse#78089 from ClickHouse/ci_support_for_s3usage_meters_in_praktika

CI: Add S3 Usage meters to praktika

Author: maxknv

ci/praktika/cidb.py

@@ -5,10 +5,11 @@
 from typing import Optional
 
 import requests
-from praktika.info import Info
 
 from ._environment import _Environment
+from .info import Info
 from .result import Result
+from .s3 import StorageUsage
 from .settings import Settings
 from .utils import Utils
 
@@ -103,7 +104,7 @@ def json_data_generator(cls, result: Result):
             record.test_context_raw = result_.info
             yield json.dumps(dataclasses.asdict(record))
 
-    def insert(self, result: Result):
+    def insert_rows(self, jsons, retries=3):
         params = {
             "database": Settings.CI_DB_DB_NAME,
             "query": f"INSERT INTO {Settings.CI_DB_TABLE_NAME} FORMAT JSONEachRow",
@@ -112,12 +113,7 @@ def insert(self, result: Result):
         }
         assert Settings.CI_DB_TABLE_NAME
 
-        MAX_RETRIES = 3
-        jsons = []
-        for json_str in self.json_data_generator(result):
-            jsons.append(json_str)
-
-        for retry in range(MAX_RETRIES):
+        for retry in range(retries):
             try:
                 response = requests.post(
                     url=self.url,
@@ -131,14 +127,58 @@ def insert(self, result: Result):
                     print(f"INFO: {len(jsons)} rows inserted into CIDB")
                     break
                 else:
-                    if retry == MAX_RETRIES - 1:
+                    if retry == retries - 1:
                         raise RuntimeError(
                             f"Failed to write to CI DB, response code [{response.status_code}]"
                         )
             except Exception as ex:
-                if retry == MAX_RETRIES - 1:
+                if retry == retries - 1:
                     raise ex
 
+    def insert(self, result: Result):
+        jsons = []
+        for json_str in self.json_data_generator(result):
+            jsons.append(json_str)
+        self.insert_rows(jsons)
+        return self
+
+    def insert_storage_usage(self, storage_usage: StorageUsage):
+        info = Info()
+        json_rows = []
+        record = self.TableRecord(
+            pull_request_number=info.pr_number,
+            commit_sha=info.sha,
+            commit_url=info.commit_url,
+            check_name="Usage",
+            check_status=Result.Status.SUCCESS,
+            check_duration_ms=0,
+            check_start_time=Utils.timestamp_to_str(Utils.timestamp()),
+            report_url=info.get_report_url(),
+            pull_request_url=info.change_url,
+            base_ref=info.base_branch,
+            base_repo=info.repo_name,
+            head_ref=info.git_branch,
+            head_repo=info.fork_name,
+            task_url="",
+            instance_type=",".join(
+                filter(None, [info.instance_type, info.instance_lifecycle])
+            ),
+            instance_id=info.instance_id,
+            test_name="storage_usage_uploaded_bytes",
+            test_status="OK",
+            test_duration_ms=storage_usage.uploaded,
+            test_context_raw="test_duration_ms shows total size uploaded in bytes",
+        )
+        json_rows.append(json.dumps(dataclasses.asdict(record)))
+        record.test_name = "storage_usage_uploaded_items"
+        record.test_duration_ms = len(storage_usage.uploaded_details)
+        record.test_context_raw = (
+            "test_duration_ms shows total number of uniq object names uploaded"
+        )
+        json_rows.append(json.dumps(dataclasses.asdict(record)))
+        self.insert_rows(json_rows)
+        return self
+
     def check(self):
         # Create a session object
         params = {

ci/praktika/hook_html.py

@@ -12,7 +12,7 @@
 from .parser import WorkflowConfigParser
 from .result import Result, ResultInfo, _ResultS3
 from .runtime import RunConfig
-from .s3 import S3
+from .s3 import S3, StorageUsage
 from .settings import Settings
 from .utils import Utils
 
@@ -225,6 +225,14 @@ def run(cls, _workflow, _job):
     def post_run(cls, _workflow, _job, info_errors):
         result = Result.from_fs(_job.name)
         _ResultS3.upload_result_files_to_s3(result).dump()
+        storage_usage = None
+        if StorageUsage.exist():
+            StorageUsage.add_uploaded(
+                result.file_name()
+            )  # add Result file beforehand to upload actual storage usage data
+            print("Storage usage data found - add to Result")
+            storage_usage = StorageUsage.from_fs()
+            result.ext["storage_usage"] = storage_usage
         _ResultS3.copy_result_to_s3(result)
 
         env = _Environment.get()
@@ -282,6 +290,7 @@ def add_dependees(job_name):
             new_info=new_result_info,
             new_sub_results=new_sub_results,
             workflow_name=_workflow.name,
+            storage_usage=storage_usage,
         )
 
         if updated_status:

ci/praktika/html_prepare.py

@@ -7,9 +7,23 @@
 class Html:
     @classmethod
     def prepare(cls, is_test):
+        import htmlmin
+
         if is_test:
             page_file = Settings.HTML_PAGE_FILE.removesuffix(".html") + "_test.html"
             Shell.check(f"cp {Settings.HTML_PAGE_FILE} {page_file}")
         else:
             page_file = Settings.HTML_PAGE_FILE
+
+        with open(page_file, "r", encoding="utf-8") as f:
+            html = f.read()
+
+        minified = htmlmin.minify(html, remove_comments=True, remove_empty_space=True)
+
+        with open(page_file, "w", encoding="utf-8") as f:
+            f.write(minified)
+
         S3.copy_file_to_s3(s3_path=Settings.HTML_S3_PATH, local_path=page_file)
+
+        with open(page_file, "w", encoding="utf-8") as f:
+            f.write(html)

ci/praktika/info.py

@@ -3,8 +3,8 @@
 from pathlib import Path
 from typing import Optional
 
-from praktika.runtime import RunConfig
-from praktika.settings import Settings
+from .runtime import RunConfig
+from .settings import Settings
 
 
 class Info:
@@ -51,10 +51,18 @@ def pr_url(self):
     def commit_url(self):
         return self.env.COMMIT_URL
 
+    @property
+    def change_url(self):
+        return self.pr_url if self.pr_number else self.commit_url
+
     @property
     def git_branch(self):
         return self.env.BRANCH
 
+    @property
+    def base_branch(self):
+        return self.env.BASE_BRANCH
+
     @property
     def git_sha(self):
         return self.env.SHA
@@ -83,6 +91,10 @@ def pr_labels(self):
     def instance_type(self):
         return self.env.INSTANCE_TYPE
 
+    @property
+    def instance_lifecycle(self):
+        return self.env.INSTANCE_LIFE_CYCLE
+
     @property
     def instance_id(self):
         return self.env.INSTANCE_ID
@@ -93,7 +105,7 @@ def is_local_run(self):
 
     # TODO: Consider defining secrets outside of workflow as it project data in most of the cases
     def get_secret(self, name):
-        from praktika.mangle import _get_workflows
+        from .mangle import _get_workflows
 
         if not self.workflow:
             self.workflow = _get_workflows(self.env.WORKFLOW_NAME)[0]
@@ -115,7 +127,7 @@ def dump(self):
         self.env.dump()
 
     def get_specific_report_url(self, pr_number, branch, sha, job_name=""):
-        from praktika.settings import Settings
+        from .settings import Settings
 
         if pr_number:
             ref_param = f"PR={pr_number}"
@@ -134,7 +146,7 @@ def get_specific_report_url(self, pr_number, branch, sha, job_name=""):
 
     @staticmethod
     def get_workflow_input_value(input_name) -> Optional[str]:
-        from praktika.settings import _Settings
+        from .settings import _Settings
 
         try:
             with open(_Settings.WORKFLOW_INPUTS_FILE, "r", encoding="utf8") as f:
@@ -174,7 +186,7 @@ def is_workflow_ok(self):
         Experimental function
         :return:
         """
-        from praktika.result import Result
+        from .result import Result
 
         result = Result.from_fs(self.env.WORKFLOW_NAME)
         for subresult in result.results: