You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by ak...@apache.org on 2026/06/30 05:49:29 UTC
(tooling-agents) branch main updated: Bulk CouchDB operations and concurrent API calls across 7 agents
This is an automated email from the ASF dual-hosted git repository.
andrewmusselman pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-agents.git
The following commit(s) were added to refs/heads/main by this push:
new 30a473c Bulk CouchDB operations and concurrent API calls across 7 agents
30a473c is described below
commit 30a473c42ba63409fb29583d408ee4571be8d3db
Author: Andrew Musselman <ak...@apache.org>
AuthorDate: Mon Jun 29 22:49:21 2026 -0700
Bulk CouchDB operations and concurrent API calls across 7 agents
Replace serial per-key get/set loops with get_many/set_many and
sequential HTTP calls with asyncio.gather + Semaphore(10). Saves
roughly 9-10 minutes per full pipeline run.
gha_security: bulk-read all cached YAML at init via get_many (was
~5,000 serial gets); accumulate findings in dict, flush with set_many
(was ~1,200 serial sets); bulk-read cached findings in redacted mode.
gha_publishing_detail: 3x get_many for prefetch/classification/YAML
keys (was ~7,400 serial gets); concurrent GitHub API run-history
fetches with Semaphore(10) (was ~200 serial calls); set_many for
report writes.
gha_artifact_verify: concurrent registry API queries with
Semaphore(10) (was ~200 serial calls); set_many for results.
gha_brief: get_many for all finding keys (was ~1,200 serial gets).
gha_review: same get_many fix as brief.
gha_json_export: get_many for classification keys and finding keys
(was ~3,700 serial gets).
gha_prefetch: fetch_single_yaml returns content instead of writing
per-file; set_many batches all YAML + metadata per repo; composites
and extras batched into single set_many per repo.
---
gha-review/agents/gha_artifact_verify.py | 132 ++++++++++++++++-------------
gha-review/agents/gha_brief.py | 5 +-
gha-review/agents/gha_json_export.py | 12 ++-
gha-review/agents/gha_prefetch.py | 47 ++++++----
gha-review/agents/gha_publishing_detail.py | 102 ++++++++++++++--------
gha-review/agents/gha_review.py | 5 +-
gha-review/agents/gha_security.py | 52 ++++++++----
7 files changed, 224 insertions(+), 131 deletions(-)
diff --git a/gha-review/agents/gha_artifact_verify.py b/gha-review/agents/gha_artifact_verify.py
index 9cb98fc..b0d4956 100644
--- a/gha-review/agents/gha_artifact_verify.py
+++ b/gha-review/agents/gha_artifact_verify.py
@@ -147,11 +147,14 @@ async def run(input_dict, tools):
phantom = [] # {repo, channel, workflow} (workflow but no artifact)
errors = []
+ # ── Build verification tasks ──
+ import asyncio
+ verify_tasks = [] # (repo_data, channel, api_def, candidates_or_name)
+
for repo_data in enriched_repos:
repo_full = repo_data['repo']
repo_name = repo_full.split('/')[-1]
- # What channels does this repo claim to publish to?
repo_channels = set()
for wf in repo_data.get('workflows', []):
repo_channels.update(wf.get('production_channels', []))
@@ -163,7 +166,6 @@ async def run(input_dict, tools):
if not repo_channels:
continue
- # ── Try to discover package names from manifests ──
for channel in sorted(repo_channels):
if channel not in REGISTRY_APIS:
continue
@@ -171,11 +173,10 @@ async def run(input_dict, tools):
api = REGISTRY_APIS[channel]
package_name = None
- # Try to read manifest from cached YAML store
manifest_specs = MANIFEST_FILES.get(channel, [])
for manifest_file, pattern in manifest_specs:
if '*' in manifest_file:
- continue # Skip glob patterns for now
+ continue
content = wf_ns.get(f"__manifest__:{repo_name}:{manifest_file}")
if content and isinstance(content, str):
m = re.search(pattern, content, re.MULTILINE | re.DOTALL)
@@ -183,59 +184,67 @@ async def run(input_dict, tools):
package_name = m.group(1) or (m.group(2) if m.lastindex >= 2 else None)
break
- # Fallback: common naming conventions
- if not package_name:
- if channel == 'pypi':
- # Try apache-{repo} and {repo} patterns
- candidates = [f"apache-{repo_name}", repo_name,
- repo_name.replace('-', '_'), f"apache_{repo_name}"]
- elif channel == 'npm':
- candidates = [f"@apache-{repo_name}/{repo_name}", repo_name,
- f"apache-{repo_name}"]
- elif channel == 'crates_io':
- candidates = [repo_name, repo_name.replace('-', '_')]
- elif channel == 'docker_hub':
- candidates = [f"apache/{repo_name}"]
- else:
- candidates = [repo_name]
-
- # Try each candidate against the registry
- for candidate in candidates:
- result = await query_registry(http_client, channel, candidate, api)
- if result:
- package_name = candidate
- break
-
- if not package_name:
- # No package found — could be phantom workflow
- pub_wfs = [wf for wf in repo_data['workflows']
- if channel in wf.get('production_channels', []) + wf.get('staging_channels', [])]
- for wf in pub_wfs:
- phantom.append({
+ if package_name:
+ candidates = [package_name]
+ elif channel == 'pypi':
+ candidates = [f"apache-{repo_name}", repo_name,
+ repo_name.replace('-', '_'), f"apache_{repo_name}"]
+ elif channel == 'npm':
+ candidates = [f"@apache-{repo_name}/{repo_name}", repo_name,
+ f"apache-{repo_name}"]
+ elif channel == 'crates_io':
+ candidates = [repo_name, repo_name.replace('-', '_')]
+ elif channel == 'docker_hub':
+ candidates = [f"apache/{repo_name}"]
+ else:
+ candidates = [repo_name]
+
+ verify_tasks.append((repo_data, channel, api, candidates))
+
+ # ── Run verification concurrently (10-way parallel) ──
+ sem = asyncio.Semaphore(10)
+
+ async def verify_one(repo_data, channel, api_def, candidates):
+ async with sem:
+ repo_full = repo_data['repo']
+ for candidate in candidates:
+ result = await query_registry(http_client, channel, candidate, api_def)
+ if result:
+ pub_wfs = [wf for wf in repo_data['workflows']
+ if channel in wf.get('production_channels', []) + wf.get('staging_channels', [])]
+ return ('verified', {
'repo': repo_full,
'channel': channel,
- 'workflow': wf['file'],
- 'note': 'No package found in registry (may use non-standard name)',
+ 'package': candidate,
+ 'registry': result,
+ 'workflows': [wf['file'] for wf in pub_wfs],
+ 'last_run': pub_wfs[0].get('last_run') if pub_wfs else None,
})
- continue
-
- # ── Query registry for verification ──
- registry_data = await query_registry(http_client, channel, package_name, api)
-
- if registry_data:
- # Find matching workflow
- pub_wfs = [wf for wf in repo_data['workflows']
- if channel in wf.get('production_channels', []) + wf.get('staging_channels', [])]
- verified.append({
+ # No candidate matched — phantom
+ pub_wfs = [wf for wf in repo_data['workflows']
+ if channel in wf.get('production_channels', []) + wf.get('staging_channels', [])]
+ phantoms = []
+ for wf in pub_wfs:
+ phantoms.append({
'repo': repo_full,
'channel': channel,
- 'package': package_name,
- 'registry': registry_data,
- 'workflows': [wf['file'] for wf in pub_wfs],
- 'last_run': pub_wfs[0].get('last_run') if pub_wfs else None,
+ 'workflow': wf['file'],
+ 'note': 'No package found in registry (may use non-standard name)',
})
- print(f" ✅ {repo_full} → {channel}:{package_name} "
- f"(v{registry_data.get('latest_version', '?')})", flush=True)
+ return ('phantom', phantoms)
+
+ results = await asyncio.gather(*[
+ verify_one(rd, ch, api, cands)
+ for rd, ch, api, cands in verify_tasks
+ ])
+
+ for result_type, data in results:
+ if result_type == 'verified':
+ verified.append(data)
+ print(f" \u2705 {data['repo']} \u2192 {data['channel']}:{data['package']} "
+ f"(v{data['registry'].get('latest_version', '?')})", flush=True)
+ elif result_type == 'phantom':
+ phantom.extend(data)
print(f"\nResults: {len(verified)} verified, {len(phantom)} phantom, "
f"{len(orphaned)} orphaned, {len(errors)} errors", flush=True)
@@ -247,16 +256,19 @@ async def run(input_dict, tools):
files['atr-catalog.json'] = json.dumps(
gen_atr_catalog(verified, enriched_repos, github_owner), indent=2)
- # Store
+ # Store (one bulk write)
verify_ns = data_store.use_namespace(f"ci-artifact-verify:{github_owner}")
- verify_ns.set("latest_results", {
- 'generated_at': datetime.now(timezone.utc).isoformat(),
- 'verified': verified,
- 'phantom': phantom,
- 'orphaned': orphaned,
- })
+ couch_writes = {
+ "latest_results": {
+ 'generated_at': datetime.now(timezone.utc).isoformat(),
+ 'verified': verified,
+ 'phantom': phantom,
+ 'orphaned': orphaned,
+ }
+ }
for fname, content in files.items():
- verify_ns.set(f"report:{fname}", content)
+ couch_writes[f"report:{fname}"] = content
+ verify_ns.set_many(couch_writes)
return {"outputText": json.dumps({"files": files})}
@@ -391,4 +403,4 @@ def gen_atr_catalog(verified, enriched_repos, owner):
'generated_at': datetime.now(timezone.utc).isoformat(),
'description': 'Apache publishing catalog — verified packages with workflow provenance',
'repos': list(repos_dict.values()),
- }
+ }
\ No newline at end of file
diff --git a/gha-review/agents/gha_brief.py b/gha-review/agents/gha_brief.py
index 07c5c6e..f5e9c0f 100644
--- a/gha-review/agents/gha_brief.py
+++ b/gha-review/agents/gha_brief.py
@@ -55,6 +55,9 @@ async def run(input_dict, tools):
all_sec_keys = security_ns.list_keys()
finding_keys = [k for k in all_sec_keys if k.startswith("findings:")]
+ # Bulk-read all findings in one CouchDB call (was ~1,200 sequential GETs)
+ all_findings_data = security_ns.get_many(finding_keys) if finding_keys else {}
+
# Collect CRITICAL, HIGH, and composite injection findings with repo context
critical_findings = [] # (repo, file, check, detail)
high_publishing = [] # (repo, file, check, detail)
@@ -70,7 +73,7 @@ async def run(input_dict, tools):
for k in finding_keys:
repo = k.replace("findings:", "")
- findings = security_ns.get(k)
+ findings = all_findings_data.get(k)
if not findings or not isinstance(findings, list):
continue
diff --git a/gha-review/agents/gha_json_export.py b/gha-review/agents/gha_json_export.py
index 11ed8ff..11dfd8c 100644
--- a/gha-review/agents/gha_json_export.py
+++ b/gha-review/agents/gha_json_export.py
@@ -59,10 +59,13 @@ async def run(input_dict, tools):
all_cls_keys = classification_ns.list_keys()
meta_keys = [k for k in all_cls_keys if k.startswith("__meta__:")]
+ # Bulk-read all classification keys in one CouchDB call
+ all_cls_data = classification_ns.get_many(all_cls_keys) if all_cls_keys else {}
+
repo_workflows = {} # repo -> [workflow classifications]
for mk in meta_keys:
repo = mk.replace("__meta__:", "")
- meta = classification_ns.get(mk)
+ meta = all_cls_data.get(mk)
if not meta or not meta.get("complete"):
continue
wf_names = meta.get("workflows", [])
@@ -71,7 +74,7 @@ async def run(input_dict, tools):
continue
wfs = []
for wf_name in wf_names:
- cls = classification_ns.get(f"{repo}:{wf_name}")
+ cls = all_cls_data.get(f"{repo}:{wf_name}")
if cls:
wfs.append(cls)
repo_workflows[repo] = wfs
@@ -82,10 +85,13 @@ async def run(input_dict, tools):
all_sec_keys = security_ns.list_keys()
finding_keys = [k for k in all_sec_keys if k.startswith("findings:")]
+ # Bulk-read all findings in one CouchDB call
+ all_findings_data = security_ns.get_many(finding_keys) if finding_keys else {}
+
repo_findings = {}
for k in finding_keys:
repo = k.replace("findings:", "")
- findings = security_ns.get(k)
+ findings = all_findings_data.get(k)
if findings and isinstance(findings, list):
if redacted_severity:
findings = [f for f in findings if not is_severity_redacted(f.get("severity", "INFO"))]
diff --git a/gha-review/agents/gha_prefetch.py b/gha-review/agents/gha_prefetch.py
index d57e87d..bd4698c 100644
--- a/gha-review/agents/gha_prefetch.py
+++ b/gha-review/agents/gha_prefetch.py
@@ -144,21 +144,19 @@ async def run(input_dict, tools):
"errors": 0}
async def fetch_single_yaml(repo_name, wf_name, download_url):
- """Fetch one YAML file if not already cached."""
+ """Fetch one YAML file if not already cached. Returns (key, content) or None."""
cache_key = f"{repo_name}/{wf_name}"
if cache_key in all_cached_keys:
- return True, True # success, was_cached
+ return True, True, None, None # success, was_cached, key, content
async with api_sem:
try:
resp = await http_client.get(download_url, follow_redirects=True, timeout=30.0)
if resp.status_code == 200:
- workflow_cache.set(cache_key, resp.text)
- all_cached_keys.add(cache_key)
- return True, False # success, newly fetched
+ return True, False, cache_key, resp.text # success, newly fetched
except Exception:
pass
- return False, False
+ return False, False, None, None
def extract_extras_from_tree(tree):
"""Extract CODEOWNERS and dependabot booleans from tree listing."""
@@ -229,19 +227,25 @@ async def run(input_dict, tools):
tasks.append(fetch_single_yaml(repo_name, wf_name, dl_url))
results = await asyncio.gather(*tasks, return_exceptions=True)
+ repo_writes = {}
for r in results:
if isinstance(r, Exception):
stats["errors"] += 1
else:
- success, was_cached = r
+ success, was_cached, cache_key, content = r
if success:
if was_cached:
stats["wf_yaml_existed"] += 1
else:
stats["wf_yaml_cached"] += 1
+ if cache_key and content:
+ repo_writes[cache_key] = content
+ all_cached_keys.add(cache_key)
- workflow_cache.set(f"__prefetch__:{repo_name}",
- {"complete": True, "workflows": wf_names})
+ # Batch-write all YAML for this repo + metadata
+ repo_writes[f"__prefetch__:{repo_name}"] = {
+ "complete": True, "workflows": wf_names}
+ workflow_cache.set_many(repo_writes)
stats["wf_fetched"] += 1
# ---- Composites + Extras (share the tree fetch) ----
@@ -266,6 +270,7 @@ async def run(input_dict, tools):
stats["ca_skipped"] += 1
else:
composite_names = []
+ composite_writes = {}
try:
action_files = [
item["path"] for item in tree
@@ -293,7 +298,7 @@ async def run(input_dict, tools):
dl_resp = await http_client.get(
dl_url, follow_redirects=True, timeout=10.0)
if dl_resp.status_code == 200:
- workflow_cache.set(cache_key, dl_resp.text)
+ composite_writes[cache_key] = dl_resp.text
all_cached_keys.add(cache_key)
composite_names.append(short_path)
stats["ca_total"] += 1
@@ -305,16 +310,26 @@ async def run(input_dict, tools):
if composite_names:
stats["ca_repos_with"] += 1
- workflow_cache.set(f"__composites__:{repo_name}", {
+ composite_writes[f"__composites__:{repo_name}"] = {
"complete": True,
"actions": composite_names,
- })
+ }
stats["ca_fetched"] += 1
- # -- Extras from tree (zero extra API calls) --
- if ex_cached:
- stats["extras_skipped"] += 1
- else:
+ # -- Extras from tree (zero extra API calls) --
+ if ex_cached:
+ stats["extras_skipped"] += 1
+ else:
+ extras = extract_extras_from_tree(tree)
+ composite_writes[f"__extras__:{repo_name}"] = extras
+ stats["extras_fetched"] += 1
+
+ # Batch-write composites + extras for this repo
+ if composite_writes:
+ workflow_cache.set_many(composite_writes)
+
+ if ca_cached and not ex_cached:
+ # Extras only (no composites to write)
extras = extract_extras_from_tree(tree)
workflow_cache.set(f"__extras__:{repo_name}", extras)
stats["extras_fetched"] += 1
diff --git a/gha-review/agents/gha_publishing_detail.py b/gha-review/agents/gha_publishing_detail.py
index 1b01222..6596f19 100644
--- a/gha-review/agents/gha_publishing_detail.py
+++ b/gha-review/agents/gha_publishing_detail.py
@@ -180,18 +180,42 @@ async def run(input_dict, tools):
if read_pat:
github_headers["Authorization"] = f"Bearer {read_pat}"
- # ── Process each repo ──
+ # ── Bulk-read all CouchDB data (3 round trips instead of ~7,400) ──
+ import asyncio
+
+ # 1. All prefetch metadata
+ all_prefetch = wf_ns.get_many(prefetch_keys) if prefetch_keys else {}
+ print(f" Bulk-read {len(all_prefetch)} prefetch keys", flush=True)
+
+ # 2. All classification data
+ cls_keys = [f"classification:{r}" for r in repo_names]
+ all_cls = cls_ns.get_many(cls_keys) if cls_keys else {}
+ print(f" Bulk-read {len(all_cls)} classification keys", flush=True)
+
+ # 3. All workflow YAML — build key list from prefetch metadata
+ yaml_keys = []
+ for repo_name in repo_names:
+ prefetch = all_prefetch.get(f"__prefetch__:{repo_name}")
+ if not prefetch or not isinstance(prefetch, dict):
+ continue
+ for wf_file in prefetch.get("workflow_files", []):
+ yaml_keys.append(f"{repo_name}/{wf_file}")
+ all_yaml = wf_ns.get_many(yaml_keys) if yaml_keys else {}
+ print(f" Bulk-read {len(all_yaml)} YAML keys", flush=True)
+
+ # ── Process each repo (pure in-memory, no CouchDB calls) ──
enriched_repos = []
+ repos_needing_runs = [] # (repo_name, repo_wfs) for GitHub API
for repo_name in repo_names:
- prefetch = wf_ns.get(f"__prefetch__:{repo_name}")
+ prefetch = all_prefetch.get(f"__prefetch__:{repo_name}")
if not prefetch or not isinstance(prefetch, dict):
continue
wf_files = prefetch.get("workflow_files", [])
if not wf_files:
continue
- cls_data = cls_ns.get(f"classification:{repo_name}")
+ cls_data = all_cls.get(f"classification:{repo_name}")
wf_cls = {}
if cls_data and isinstance(cls_data, list):
for c in cls_data:
@@ -199,7 +223,7 @@ async def run(input_dict, tools):
repo_wfs = []
for wf_file in wf_files:
- yaml_content = wf_ns.get(f"{repo_name}/{wf_file}")
+ yaml_content = all_yaml.get(f"{repo_name}/{wf_file}")
if not yaml_content or not isinstance(yaml_content, str):
continue
@@ -207,7 +231,6 @@ async def run(input_dict, tools):
wf_data = parse_workflow(yaml_content, wf_file, repo_name, github_owner)
wf_data['category'] = cat
- # Skip if no channels detected and it's not classified as publishing
if not wf_data['channels'] and cat in ('unknown', 'ci_check', 'documentation'):
continue
@@ -223,37 +246,45 @@ async def run(input_dict, tools):
if channels_filter and not any(c in all_ch for c in channels_filter):
continue
- # ── Fetch run history ──
- if read_pat:
- try:
- url = f"https://api.github.com/repos/{github_owner}/{repo_name}/actions/runs?per_page=30"
- resp = await http_client.get(url, headers=github_headers, timeout=15)
- if resp.status_code == 200:
- runs_by_file = defaultdict(list)
- for r in resp.json().get("workflow_runs", []):
- fname = r.get("path", "").split("/")[-1]
- runs_by_file[fname].append({
- 'status': r.get('conclusion', r.get('status', '?')),
- 'trigger': r.get('event', '?'),
- 'created': r.get('created_at', ''),
- 'url': r.get('html_url', ''),
- })
- for wf in repo_wfs:
- wf['last_run'] = (runs_by_file.get(wf['file'], [None]) or [None])[0]
- wf['recent_runs'] = runs_by_file.get(wf['file'], [])[:5]
- except Exception as e:
- print(f" API error {repo_name}: {e}", flush=True)
- for wf in repo_wfs:
- wf['last_run'] = None; wf['recent_runs'] = []
- else:
- for wf in repo_wfs:
- wf['last_run'] = None; wf['recent_runs'] = []
+ # Initialize run history as empty — filled by concurrent API calls below
+ for wf in repo_wfs:
+ wf['last_run'] = None
+ wf['recent_runs'] = []
enriched_repos.append({
'repo': f"{github_owner}/{repo_name}",
'channels': sorted(all_ch),
'workflows': repo_wfs,
})
+ repos_needing_runs.append((repo_name, repo_wfs))
+
+ # ── Fetch run history concurrently (10-way parallel) ──
+ if read_pat and repos_needing_runs:
+ sem = asyncio.Semaphore(10)
+
+ async def fetch_runs(repo_name, repo_wfs):
+ async with sem:
+ try:
+ url = f"https://api.github.com/repos/{github_owner}/{repo_name}/actions/runs?per_page=30"
+ resp = await http_client.get(url, headers=github_headers, timeout=15)
+ if resp.status_code == 200:
+ runs_by_file = defaultdict(list)
+ for r in resp.json().get("workflow_runs", []):
+ fname = r.get("path", "").split("/")[-1]
+ runs_by_file[fname].append({
+ 'status': r.get('conclusion', r.get('status', '?')),
+ 'trigger': r.get('event', '?'),
+ 'created': r.get('created_at', ''),
+ 'url': r.get('html_url', ''),
+ })
+ for wf in repo_wfs:
+ wf['last_run'] = (runs_by_file.get(wf['file'], [None]) or [None])[0]
+ wf['recent_runs'] = runs_by_file.get(wf['file'], [])[:5]
+ except Exception as e:
+ print(f" API error {repo_name}: {e}", flush=True)
+
+ await asyncio.gather(*[fetch_runs(rn, wfs) for rn, wfs in repos_needing_runs])
+ print(f" Fetched run history for {len(repos_needing_runs)} repos (10-way parallel)", flush=True)
print(f"Enriched {len(enriched_repos)} repos", flush=True)
@@ -276,14 +307,15 @@ async def run(input_dict, tools):
fname = f"channel-{ch_key.replace('_', '-')}.md"
files[fname] = gen_channel_report(ch_key, display, items, enriched_repos, github_owner)
- # Store in CouchDB
+ # Store in CouchDB (one bulk write)
detail_ns = data_store.use_namespace(f"ci-publishing-detail:{github_owner}")
- detail_ns.set("latest_data", {
+ couch_writes = {"latest_data": {
'generated_at': datetime.now(timezone.utc).isoformat(),
'repos': enriched_repos,
- })
+ }}
for fname, content in files.items():
- detail_ns.set(f"report:{fname}", content)
+ couch_writes[f"report:{fname}"] = content
+ detail_ns.set_many(couch_writes)
print(f"Generated {len(files)} report files", flush=True)
@@ -496,4 +528,4 @@ def gen_channel_report(ch_key, display_name, items, all_repos, owner):
lines.append(f"- Last run: {_run_link(wf)}")
lines.append("")
- return "\n".join(lines)
+ return "\n".join(lines)
\ No newline at end of file
diff --git a/gha-review/agents/gha_review.py b/gha-review/agents/gha_review.py
index a00f9bb..35e3cce 100644
--- a/gha-review/agents/gha_review.py
+++ b/gha-review/agents/gha_review.py
@@ -73,6 +73,9 @@ async def run(input_dict, tools):
all_sec_keys = security_ns.list_keys()
finding_keys = [k for k in all_sec_keys if k.startswith("findings:")]
+ # Bulk-read all findings in one CouchDB call
+ all_findings_data = security_ns.get_many(finding_keys) if finding_keys else {}
+
repo_security = {}
# Accumulate org-level stats from filtered findings
@@ -82,7 +85,7 @@ async def run(input_dict, tools):
for k in finding_keys:
repo = k.replace("findings:", "")
- findings = security_ns.get(k)
+ findings = all_findings_data.get(k)
if not findings or not isinstance(findings, list):
continue
diff --git a/gha-review/agents/gha_security.py b/gha-review/agents/gha_security.py
index 3462b92..9d96df0 100644
--- a/gha-review/agents/gha_security.py
+++ b/gha-review/agents/gha_security.py
@@ -39,6 +39,12 @@ async def run(input_dict, tools):
return {"outputText": "Error: no cached workflows found in `ci-workflows:" + github_owner + "`. "
"Run the Publishing Analyzer agent first."}
+ # Bulk-read ALL cached data in one CouchDB call (~5,000 keys).
+ # Individual workflow_ns.get() calls in the scan loop now read
+ # from this in-memory dict instead of hitting CouchDB per file.
+ all_wf_data = workflow_ns.get_many(all_wf_keys) if all_wf_keys else {}
+ print(f"Bulk-read {len(all_wf_data)} cached keys", flush=True)
+
repos = {}
for key in all_wf_keys:
if "/" in key:
@@ -612,6 +618,15 @@ async def run(input_dict, tools):
# ===== Main scan loop =====
all_findings = {}
repos_scanned = 0
+ pending_writes = {} # accumulated for set_many after loop
+
+ # For redacted mode: bulk-read all cached findings upfront
+ if redacted_severity:
+ sec_keys = security_ns.list_keys()
+ finding_keys = [k for k in sec_keys if k.startswith("findings:")]
+ cached_findings = security_ns.get_many(finding_keys) if finding_keys else {}
+ else:
+ cached_findings = {}
for repo_name, wf_names in sorted(repos.items()):
repos_scanned += 1
@@ -619,9 +634,9 @@ async def run(input_dict, tools):
if repos_scanned % 10 == 1:
print(f"[{repos_scanned}/{len(repos)}] Scanning {repo_name}...", flush=True)
- # When redacting: read cached findings, filter, don't rescan
+ # When redacting: use bulk-read findings, filter, don't rescan
if redacted_severity:
- cached = security_ns.get(f"findings:{repo_name}")
+ cached = cached_findings.get(f"findings:{repo_name}")
if cached:
filtered = [f for f in cached if not is_severity_redacted(f.get("severity", "INFO"))]
if filtered:
@@ -638,7 +653,7 @@ async def run(input_dict, tools):
if ".github/actions/" in wf_name:
continue
- content = workflow_ns.get(f"{repo_name}/{wf_name}")
+ content = all_wf_data.get(f"{repo_name}/{wf_name}")
if not content or not isinstance(content, str):
continue
@@ -728,7 +743,7 @@ async def run(input_dict, tools):
# --- Read extra files from prefetch cache ---
# Check 7: CODEOWNERS (from prefetch __extras__)
- extras = workflow_ns.get(f"__extras__:{repo_name}")
+ extras = all_wf_data.get(f"__extras__:{repo_name}")
if extras:
if not extras.get("has_codeowners"):
repo_findings.append({
@@ -753,11 +768,11 @@ async def run(input_dict, tools):
# Collect (action_name, short_path, action_content) tuples
composite_items = []
- composites_meta = workflow_ns.get(f"__composites__:{repo_name}")
+ composites_meta = all_wf_data.get(f"__composites__:{repo_name}")
if composites_meta and composites_meta.get("complete"):
cached_actions = composites_meta.get("actions", [])
for short_path in cached_actions:
- action_content = workflow_ns.get(f"{repo_name}/{short_path}")
+ action_content = all_wf_data.get(f"{repo_name}/{short_path}")
if action_content:
action_name = short_path.replace(".github/actions/", "").rsplit("/", 1)[0]
composite_items.append((action_name, short_path, action_content))
@@ -850,13 +865,18 @@ async def run(input_dict, tools):
# Deduplicate all findings for this repo
repo_findings = deduplicate_findings(repo_findings)
- # Store
- security_ns.set(f"findings:{repo_name}", repo_findings)
+ # Accumulate for bulk write after loop
+ pending_writes[f"findings:{repo_name}"] = repo_findings
if repo_findings:
all_findings[repo_name] = repo_findings
await asyncio.sleep(0.1)
+ # Bulk-write all findings in one CouchDB call
+ if pending_writes:
+ security_ns.set_many(pending_writes)
+ print(f"Bulk-wrote findings for {len(pending_writes)} repos", flush=True)
+
print(f"\n{'=' * 60}", flush=True)
print(f"Security scan complete! {repos_scanned} repos", flush=True)
total_findings = sum(len(f) for f in all_findings.values())
@@ -1005,13 +1025,15 @@ async def run(input_dict, tools):
full_report = toc + "\n\n---\n\n" + report_body
if not redacted_severity:
- security_ns.set("latest_report", full_report)
- security_ns.set("latest_stats", {
- "repos_scanned": repos_scanned,
- "repos_with_findings": len(all_findings),
- "total_findings": total_findings,
- "severity_counts": severity_counts,
- "check_counts": check_counts,
+ security_ns.set_many({
+ "latest_report": full_report,
+ "latest_stats": {
+ "repos_scanned": repos_scanned,
+ "repos_with_findings": len(all_findings),
+ "total_findings": total_findings,
+ "severity_counts": severity_counts,
+ "check_counts": check_counts,
+ },
})
return {"outputText": full_report}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org