tools/ci: Cancel the redundant pipelines in the CI.

tools/ci/cancel_pipelines.py

@@ -0,0 +1,65 @@
+# SPDX-FileCopyrightText: 2025 Espressif Systems (Shanghai) CO LTD
+
+# SPDX-License-Identifier: CC0-1.0
+
+import re
+import os
+import requests
+from gitlab_api import GitLabAPI
+
+def cancel_branch_pipelines(gitlab_api):
+    url = f"{gitlab_api.gitlab_api_url}/projects/{gitlab_api.ci_project_id}/pipelines?ref={gitlab_api.ci_commit_ref_name}&status=running"
+    headers = {"PRIVATE-TOKEN": gitlab_api.gitlab_token}
+    response = requests.get(url, headers=headers)
+
+    if response.status_code == 200:
+        running_pipelines = response.json()
+
+        for pipeline in running_pipelines:
+            pipeline_id = pipeline.get('id')
+            source = pipeline.get('source')
+            ref = pipeline.get('ref')
+            if pipeline_id != int(gitlab_api.ci_pipeline_id) and source != 'merge_request_event':
+                print(f"Cancelling branch pipeline with pipeline ID: {pipeline_id}, Source:{source}, Ref: {ref}")
+                gitlab_api.cancel_pipeline(pipeline_id)
+            else:
+                print(f"Skipping current branch pipeline with pipeline ID: {gitlab_api.ci_pipeline_id}")
+    else:
+        print(f"Failed to fetch branch pipelines, Status Code: {response.status_code}")
+
+def cancel_merge_request_pipelines(gitlab_api):
+    url = f"{gitlab_api.gitlab_api_url}/projects/{gitlab_api.ci_project_id}/pipelines?status=running"
+    headers = {"PRIVATE-TOKEN": gitlab_api.gitlab_token}
+    response = requests.get(url, headers=headers)
+
+    if response.status_code == 200:
+        running_pipelines = response.json()
+
+        for pipeline in running_pipelines:
+            pipeline_id = pipeline.get('id')
+            source = pipeline.get('source')
+            ref = pipeline.get('ref')
+            merge_request_id_pattern = re.match(r"refs/merge-requests/(\d+)/head", ref)
+            if merge_request_id_pattern:
+               print(f"MR pipeline id: {merge_request_id_pattern.group(1)}")
+               if source == "merge_request_event" and gitlab_api.ci_merge_request_iid == merge_request_id_pattern.group(1):
+                  if pipeline_id == int(gitlab_api.ci_pipeline_id):
+                      print(f"Skipping current MR pipeline with pipeline id: {pipeline_id} ref: {ref} source: {source}")
+                  else:
+                      print(f"Cancelling MR pipeline with pipeline id: {pipeline_id} ref:{ref} source: {source}")
+                      gitlab_api.cancel_pipeline(pipeline_id)
+            else:
+                print(f"Skipping pipeline with pipeline ID: {pipeline_id} (ref: {ref}, source: {source})")
+
+    else:
+        print(f"Failed to fetch pipelines, Status: {response.status_code}")
+
+
+def main():
+    gitlab_api = GitLabAPI()
+    cancel_branch_pipelines(gitlab_api)
+    cancel_merge_request_pipelines(gitlab_api)
+
+if __name__ == "__main__":
+    main()
+

tools/ci/gitlab_api.py

@@ -12,8 +12,10 @@ class GitLabAPI:
         self.gitlab_token = os.getenv("GITLAB_MR_COMMENT_TOKEN")
         self.ci_project_id = os.getenv("CI_PROJECT_ID")
         self.ci_merge_request_iid = os.getenv("CI_MERGE_REQUEST_IID")
+        self.ci_pipeline_id = os.getenv("CI_PIPELINE_ID")
+        self.ci_commit_ref_name = os.getenv("CI_COMMIT_REF_NAME")
         
-        if not all([self.gitlab_api_url, self.gitlab_token, self.ci_project_id, self.ci_merge_request_iid]):
+        if not all([self.gitlab_api_url, self.gitlab_token, self.ci_project_id, self.ci_pipeline_id]):
             raise ValueError("Required GitLab environment variables are not set")
 
     def fetch_merge_request_description(self):
@@ -64,3 +66,13 @@ class GitLabAPI:
             with open(output_file, 'wb') as f:
                 for chunk in response.iter_content(chunk_size=8192):
                     f.write(chunk) 
+
+    def cancel_pipeline(self, pipeline_id):
+        url = f"{self.gitlab_api_url}/projects/{self.ci_project_id}/pipelines/{pipeline_id}/cancel"
+        headers = {"PRIVATE-TOKEN": self.gitlab_token}
+        response = requests.post(url, headers=headers)
+
+        if response.status_code == 200:
+            print(f"Successfully cancelled Pipeline ID: {pipeline_id}")
+        else:
+            print(f"Failed to cancel Pipeline ID: {pipeline_id}, Status Code: {response.status_code}")