1from __future__ import annotations 2 3import argparse 4import json 5import os 6from pathlib import Path 7from tempfile import TemporaryDirectory 8from typing import Any 9 10from tools.stats.upload_stats_lib import ( 11 download_s3_artifacts, 12 upload_workflow_stats_to_s3, 13) 14 15 16def get_sccache_stats( 17 workflow_run_id: int, workflow_run_attempt: int 18) -> list[dict[str, Any]]: 19 with TemporaryDirectory() as temp_dir: 20 print("Using temporary directory:", temp_dir) 21 os.chdir(temp_dir) 22 23 # Download and extract all the reports (both GHA and S3) 24 download_s3_artifacts("sccache-stats", workflow_run_id, workflow_run_attempt) 25 26 stats_jsons = [] 27 for json_file in Path(".").glob("**/*.json"): 28 with open(json_file) as f: 29 stats_jsons.append(json.load(f)) 30 return stats_jsons 31 32 33if __name__ == "__main__": 34 parser = argparse.ArgumentParser(description="Upload test stats to Rockset") 35 parser.add_argument( 36 "--workflow-run-id", 37 type=int, 38 required=True, 39 help="id of the workflow to get artifacts from", 40 ) 41 parser.add_argument( 42 "--workflow-run-attempt", 43 type=int, 44 required=True, 45 help="which retry of the workflow this is", 46 ) 47 args = parser.parse_args() 48 stats = get_sccache_stats(args.workflow_run_id, args.workflow_run_attempt) 49 upload_workflow_stats_to_s3( 50 args.workflow_run_id, args.workflow_run_attempt, "sccache_stats", stats 51 ) 52