feat: read accumulative result files (#1837)

* feat: read accumulative result files * fix: build results dir * fix: build results dir * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: selection algo * fix: debug * fix: debug * fix: debug * fix: debug * fix: debug * fix: debug * fix: debug select algo * fix: debug select algo * fix: debug select algo * fix: debug select algo * fix: filtering * fix: review comments * fix: rm test_filter
2026-09-26 14:42:04 +00:00 · 2023-05-12 17:44:54 +00:00
parent 5c2bd2a13a
commit 6ea28621ac
2 changed files with 43 additions and 33 deletions
@@ -137,7 +137,11 @@ changed_notebooks = execute_changed_notebooks_helper.get_changed_notebooks(


 results_bucket = f"{args.artifacts_bucket}"
-results_file = f"{args.build_id}.json"
+# artifacts_bucket may get set by trigger to a full gs:// folder path
+if results_bucket.startswith("gs://"):
+    results_bucket = results_bucket[5:]
+results_bucket = results_bucket.split('/')[0]
+results_file = f"build_results/{args.build_id}.json"

 if args.test_percent == 100:
    notebooks = changed_notebooks
@@ -158,7 +162,6 @@ else:
        staging_bucket=args.staging_bucket,
        artifacts_bucket=args.artifacts_bucket,
        results_file=results_file,
-        accumulative_results=accumulative_results,
        should_parallelize=args.should_parallelize,
        timeout=args.timeout,
        variable_project_id=args.variable_project_id,
@@ -95,8 +95,20 @@ def load_results(results_bucket: str,
    print("Loading existing accumulative results ...")
    accumulative_results = {}
    try:
-        content = util.download_blob_into_memory(results_bucket, results_file, download_as_text=True)
-        accumulative_results = json.loads(content)
+        client = storage.Client()
+        bucket = client.bucket(results_bucket)
+
+        build_results_dir = os.path.dirname(results_file)
+        blobs = client.list_blobs(results_bucket, prefix=build_results_dir)
+        for blob in blobs:
+            content = util.download_blob_into_memory(results_bucket, blob.name, download_as_text=True)
+            build_results = json.loads(content)
+            for notebook in build_results:
+                if notebook in accumulative_results:
+                    accumulative_results[notebook] += build_results[notebook]
+                else:
+                    accumulative_results[notebook] = build_results[notebook]
+
        print(accumulative_results)
    except Exception as e:
        print(e)
@@ -118,7 +130,15 @@ def select_notebook(changed_notebook: str,
        pass_count = 1
        fail_count = 0

-    return (random.randint(1, 100) * (1 + (fail_count / (pass_count + fail_count))) < test_percent)
+    inferred_failure_rate = fail_count / (pass_count + fail_count)
+
+    # If failure rate is high, the chance of testing should be higher
+    should_test_due_to_failure = random.uniform(0, 1) < inferred_failure_rate
+
+    # Additionally, only test a percentage of these
+    should_test_due_to_random_subset = random.uniform(0, 1) < test_percent
+
+    return should_test_due_to_failure and should_test_due_to_random_subset


 def _process_notebook(
@@ -379,39 +399,30 @@ def get_changed_notebooks(
    return notebooks

 def _save_results(results: List[NotebookExecutionResult],
-                  accumulative_results: Dict[str,Any],
                  artifacts_bucket: str,
                  results_file: str):

    artifacts_bucket = artifacts_bucket.replace("gs://", "").split('/')[0]

-    print("Updating accumulative results ...")
+    print("Updating build results ...")
+    build_results = {}
    for result in results:
-        if result.path in accumulative_results:
-            accumulative_results[result.path]['duration'] = result.duration.total_seconds()
-            accumulative_results[result.path]['start_time'] = str(result.start_time)
-            if result.is_pass:
-                accumulative_results[result.path]['passed'] += 1
-            else:
-                accumulative_results[result.path]['failed'] += 1
-            print(f"updating {result.path}")
+        if result.is_pass:
+            pass_count = 1
+            fail_count = 0
        else:
-            if result.is_pass:
-                pass_count = 1
-                fail_count = 0
-            else:
-                pass_count = 0
-                fail_count = 1
-            accumulative_results[result.path] = {
-                    'duration': result.duration.total_seconds(),
-                    'start_time': str(result.start_time),
-                    'passed': pass_count,
-                    'failed': fail_count
-                    }
-            print(f"adding {result.path}")
+            pass_count = 0
+            fail_count = 1
+        build_results[result.path] = {
+                'duration': result.duration.total_seconds(),
+                'start_time': str(result.start_time),
+                'passed': pass_count,
+                'failed': fail_count
+        }
+        print(f"adding {result.path}")

    print("Saving accumulative results ...")
-    content = json.dumps(accumulative_results)
+    content = json.dumps(build_results)

    client = storage.Client()
    bucket = client.get_bucket(artifacts_bucket)
@@ -425,7 +436,6 @@ def process_and_execute_notebooks(
    staging_bucket: str,
    artifacts_bucket: str,
    results_file: str,
-    accumulative_results: List[NotebookExecutionResult],
    should_parallelize: bool,
    timeout: int,
    variable_project_id: str,
@@ -455,8 +465,6 @@ def process_and_execute_notebooks(
            Required. The GCS staging bucket to write executed notebooks to.
        results_file (str):
            Required: The path to the artifacts bucket to save results 
-        accumulative_results (List):
-            Required: The in-memory previous accumulative notebook CI/CD test results.
        variable_project_id (str):
            Required. The value for PROJECT_ID to inject into notebooks.
        variable_region (str):
@@ -578,7 +586,6 @@ def process_and_execute_notebooks(
                print(log_contents)

        _save_results(results_sorted, 
-                      accumulative_results,
                      artifacts_bucket, 
                      results_file)