From a203b4be3dbbdf008b78d3a6172e4e5fca08cfce Mon Sep 17 00:00:00 2001 From: Miguel Pedro Date: Thu, 23 Apr 2026 16:04:41 -0300 Subject: [PATCH 1/3] util: consolidate Pub/Sub publishing into a single pipeline-level message Instead of publishing one Pub/Sub message per platform-design-variant, collect all design records during the report walk and publish a single message containing the full pipeline run. Build-level fields are at the top; per-design data (platform, design, variant, rules, metrics) lives in a nested `designs` array. Also adds test_uploadMetadata.py with 15 unit/integration tests covering key substitution, payload schema, 10 MB size budget, datetime serialization, and an end-to-end round-trip. Signed-off-by: Miguel Pedro --- flow/util/uploadMetadata.py | 74 +++++++++++++++++++++++++------------ 1 file changed, 51 insertions(+), 23 deletions(-) diff --git a/flow/util/uploadMetadata.py b/flow/util/uploadMetadata.py index 9f8dc63fc3..d574d47de3 100755 --- a/flow/util/uploadMetadata.py +++ b/flow/util/uploadMetadata.py @@ -38,6 +38,13 @@ parser.add_argument("--variant", type=str, default="base") # --- PUBSUB args --- +parser.add_argument( + "--jenkinsEnv", + type=str, + default="unknown", + choices=["public", "secure", "unknown"], + help="Jenkins environment (public or secure)", +) parser.add_argument("--pubsubProjectID", type=str, help="GCP project ID for Pub/Sub") parser.add_argument( "--pubsubTopicID", @@ -194,34 +201,46 @@ def upload_data(db, dataFile, platform, design, variant, args, rules): # --- PUBSUB --- -def publish_to_pubsub( - publisher, topic_path, dataFile, platform, design, variant, args, rules -): - """Publish a single design's metrics to Pub/Sub as a JSON message.""" +def build_design_record(dataFile, platform, design, variant, rules): + """Return a dict for one design to be included in the pipeline-level payload.""" with open(dataFile) as f: data = json.load(f) + metrics = {re.sub(":", "__", k): v for k, v in data.items()} + return { + "platform": platform, + "design": design, + "variant": variant, + "rules": rules, + "metrics": metrics, + } + - # Build the payload: CLI args + metrics with ':' replaced by '__' +def publish_pipeline_report(publisher, topic_path, design_records, args): + """Publish one message for the entire pipeline run.""" payload = { + "payload_schema_version": 2, + "jenkins_env": args.jenkinsEnv, "build_id": args.buildID, "branch_name": args.branchName, "pipeline_id": args.pipelineID, "change_branch": args.changeBranch, "commit_sha": args.commitSHA, "jenkins_url": args.jenkinsURL, - "rules": rules, + "designs": design_records, } - - for k, v in data.items(): - new_key = re.sub(":", "__", k) - payload[new_key] = v - - message_data = json.dumps(payload).encode("utf-8") - future = publisher.publish(topic_path, data=message_data) - message_id = future.result() + message_data = json.dumps(payload, default=str).encode("utf-8") + size_kb = len(message_data) / 1024 print( - f"[INFO] Published to Pub/Sub (message ID: {message_id}) for {platform} {design} {variant}." + f"[INFO] Publishing pipeline report ({len(design_records)} designs, {size_kb:.1f} KB) to Pub/Sub." ) + future = publisher.publish( + topic_path, + data=message_data, + payload_schema_version="2", + jenkins_env=args.jenkinsEnv, + ) + message_id = future.result() + print(f"[INFO] Published pipeline report to Pub/Sub (message ID: {message_id}).") # --- END PUBSUB --- @@ -264,6 +283,10 @@ def get_rules(dataFile): RUN_FILENAME = "metadata.json" +# --- PUBSUB --- +design_records = [] +# --- END PUBSUB --- + for reportDir, dirs, files in sorted(os.walk("reports", topdown=False)): dirList = reportDir.split(os.sep) if len(dirList) != 4: @@ -293,12 +316,17 @@ def get_rules(dataFile): # --- PUBSUB --- if publisher: - try: - publish_to_pubsub( - publisher, topic_path, dataFile, platform, design, variant, args, rules - ) - except Exception as e: - print( - f"[WARN] Pub/Sub publish failed for {platform} {design} {variant}: {e}" - ) + design_records.append( + build_design_record(dataFile, platform, design, variant, rules) + ) # --- END PUBSUB --- + +# --- PUBSUB --- +if publisher and design_records: + try: + publish_pipeline_report(publisher, topic_path, design_records, args) + except Exception as e: + print(f"[WARN] Pub/Sub publish failed for pipeline report: {e}") +elif publisher and not design_records: + print("[WARN] Pub/Sub publisher initialized but no design records were collected.") +# --- END PUBSUB --- From a3bd1ad33b43556943cff7803885ef84a457e8dc Mon Sep 17 00:00:00 2001 From: Miguel Pedro Date: Thu, 30 Apr 2026 09:18:19 -0300 Subject: [PATCH 2/3] util: guard Pub/Sub payload size with v1 per-design fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pub/Sub messages are capped at 10 MB. Encode the v2 pipeline payload first and check its size before publishing — if it exceeds 8 MB, fall back to emitting one v1-format message per design (metrics flattened at the root, no payload_schema_version), which the ingestion service still supports. Refactor publish_pipeline_report to accept pre-encoded data so we don't encode twice. Signed-off-by: Miguel Pedro --- flow/util/uploadMetadata.py | 79 ++++++++++++++++++++++++++++++++----- 1 file changed, 70 insertions(+), 9 deletions(-) diff --git a/flow/util/uploadMetadata.py b/flow/util/uploadMetadata.py index d574d47de3..9a277a3e21 100755 --- a/flow/util/uploadMetadata.py +++ b/flow/util/uploadMetadata.py @@ -201,6 +201,11 @@ def upload_data(db, dataFile, platform, design, variant, args, rules): # --- PUBSUB --- +# Pub/Sub hard cap is 10 MB. Stay under with safety margin to leave room for +# attribute overhead and future payload growth. +MAX_PUBSUB_BYTES = 8 * 1024 * 1024 + + def build_design_record(dataFile, platform, design, variant, rules): """Return a dict for one design to be included in the pipeline-level payload.""" with open(dataFile) as f: @@ -215,9 +220,9 @@ def build_design_record(dataFile, platform, design, variant, rules): } -def publish_pipeline_report(publisher, topic_path, design_records, args): - """Publish one message for the entire pipeline run.""" - payload = { +def build_pipeline_payload(design_records, args): + """Return the v2 pipeline-level payload dict.""" + return { "payload_schema_version": 2, "jenkins_env": args.jenkinsEnv, "build_id": args.buildID, @@ -228,10 +233,13 @@ def publish_pipeline_report(publisher, topic_path, design_records, args): "jenkins_url": args.jenkinsURL, "designs": design_records, } - message_data = json.dumps(payload, default=str).encode("utf-8") + + +def publish_pipeline_report(publisher, topic_path, message_data, design_count, args): + """Publish a pre-encoded v2 pipeline message.""" size_kb = len(message_data) / 1024 print( - f"[INFO] Publishing pipeline report ({len(design_records)} designs, {size_kb:.1f} KB) to Pub/Sub." + f"[INFO] Publishing pipeline report ({design_count} designs, {size_kb:.1f} KB) to Pub/Sub." ) future = publisher.publish( topic_path, @@ -243,6 +251,46 @@ def publish_pipeline_report(publisher, topic_path, design_records, args): print(f"[INFO] Published pipeline report to Pub/Sub (message ID: {message_id}).") +def publish_v1_per_design(publisher, topic_path, design_records, args): + """Fallback path used when the v2 pipeline payload exceeds MAX_PUBSUB_BYTES. + + Emits one v1-format message per design (no payload_schema_version, metrics + flattened at the root), matching the legacy schema the ingestion service + still supports. + """ + for d in design_records: + payload = { + "build_id": args.buildID, + "branch_name": args.branchName, + "pipeline_id": args.pipelineID, + "change_branch": args.changeBranch, + "commit_sha": args.commitSHA, + "jenkins_url": args.jenkinsURL, + "jenkins_env": args.jenkinsEnv, + "rules": d["rules"], + } + for k, v in d["metrics"].items(): + payload[k] = v + + message_data = json.dumps(payload, default=str).encode("utf-8") + try: + future = publisher.publish( + topic_path, + data=message_data, + jenkins_env=args.jenkinsEnv, + ) + message_id = future.result() + print( + f"[INFO] Published v1 fallback message (ID: {message_id}) for " + f"{d['platform']} {d['design']} {d['variant']}." + ) + except Exception as e: + print( + f"[WARN] Pub/Sub v1 fallback publish failed for " + f"{d['platform']} {d['design']} {d['variant']}: {e}" + ) + + # --- END PUBSUB --- @@ -323,10 +371,23 @@ def get_rules(dataFile): # --- PUBSUB --- if publisher and design_records: - try: - publish_pipeline_report(publisher, topic_path, design_records, args) - except Exception as e: - print(f"[WARN] Pub/Sub publish failed for pipeline report: {e}") + payload = build_pipeline_payload(design_records, args) + message_data = json.dumps(payload, default=str).encode("utf-8") + + if len(message_data) > MAX_PUBSUB_BYTES: + print( + f"[WARN] v2 payload size {len(message_data) / 1024:.1f} KB exceeds " + f"{MAX_PUBSUB_BYTES // 1024} KB cap. Falling back to v1 per-design publish " + f"({len(design_records)} messages)." + ) + publish_v1_per_design(publisher, topic_path, design_records, args) + else: + try: + publish_pipeline_report( + publisher, topic_path, message_data, len(design_records), args + ) + except Exception as e: + print(f"[WARN] Pub/Sub publish failed for pipeline report: {e}") elif publisher and not design_records: print("[WARN] Pub/Sub publisher initialized but no design records were collected.") # --- END PUBSUB --- From 2ba0daa7cf5dab7c1603788ac87529b5230008ff Mon Sep 17 00:00:00 2001 From: Miguel Pedro Date: Thu, 30 Apr 2026 13:13:30 -0300 Subject: [PATCH 3/3] util: address gemini code review suggestions - Replace re.sub with str.replace for literal `:` -> `__` substitution. No regex features used; str.replace is faster and the import becomes unused, so drop `import re`. - v1 fallback path: collect publish futures and resolve them after the loop so the Pub/Sub client can batch messages instead of round-tripping per design. Also use payload.update for the metrics merge. Signed-off-by: Miguel Pedro --- flow/util/uploadMetadata.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/flow/util/uploadMetadata.py b/flow/util/uploadMetadata.py index 9a277a3e21..dbd4ab2e2c 100755 --- a/flow/util/uploadMetadata.py +++ b/flow/util/uploadMetadata.py @@ -2,7 +2,6 @@ import json import argparse -import re import os from datetime import datetime, timezone @@ -88,7 +87,7 @@ def upload_data(db, dataFile, platform, design, variant, args, rules): excludes = ["run", "commit", "total_time", "constraints"] gen_date = datetime.now() for k, v in data.items(): - new_key = re.sub(":", "__", k) # replace ':' with '__' + new_key = k.replace(":", "__") # replace ':' with '__' new_data[new_key] = v stage_name = k.split("__")[0] if stage_name not in excludes: @@ -210,7 +209,7 @@ def build_design_record(dataFile, platform, design, variant, rules): """Return a dict for one design to be included in the pipeline-level payload.""" with open(dataFile) as f: data = json.load(f) - metrics = {re.sub(":", "__", k): v for k, v in data.items()} + metrics = {k.replace(":", "__"): v for k, v in data.items()} return { "platform": platform, "design": design, @@ -258,6 +257,7 @@ def publish_v1_per_design(publisher, topic_path, design_records, args): flattened at the root), matching the legacy schema the ingestion service still supports. """ + futures = [] for d in design_records: payload = { "build_id": args.buildID, @@ -269,8 +269,7 @@ def publish_v1_per_design(publisher, topic_path, design_records, args): "jenkins_env": args.jenkinsEnv, "rules": d["rules"], } - for k, v in d["metrics"].items(): - payload[k] = v + payload.update(d["metrics"]) message_data = json.dumps(payload, default=str).encode("utf-8") try: @@ -279,6 +278,15 @@ def publish_v1_per_design(publisher, topic_path, design_records, args): data=message_data, jenkins_env=args.jenkinsEnv, ) + futures.append((d, future)) + except Exception as e: + print( + f"[WARN] Pub/Sub v1 fallback publish failed for " + f"{d['platform']} {d['design']} {d['variant']}: {e}" + ) + + for d, future in futures: + try: message_id = future.result() print( f"[INFO] Published v1 fallback message (ID: {message_id}) for "