Implementation guide for freshness scanners#

This guide walks step-by-step through implementing a custom freshness scanner using pw_fortifier.

Overview#

Building a freshness scanner consists of the following steps:

Create an IssueTracker to access Buganizer.
Set up the FreshnessScanner and register package analyzers.
(Optional) Implement custom PackageAnalyzers for unsupported package formats.
(Optional) Implement custom PackageUpdaters for custom roll tooling.
Define Build targets to run the scanner.

Pipeline architecture#

The diagram below shows how pipeline stages connect. Orange nodes represent implementer-provided components; light blue nodes represent library-provided components.

        flowchart TD
   subgraph Discovery ["Discovery & Generation"]
      Emitter --> AnalyzerMux["AnalyzerMux"]
      AnalyzerMux --> BazelCipd["BazelCipd"]
      AnalyzerMux --> BazelDep["BazelDep"]
      AnalyzerMux --> BazelMaven["BazelMaven"]
      AnalyzerMux --> CargoAnalyzer["Cargo"]
      AnalyzerMux --> CipdSetup["CipdSetup"]
      AnalyzerMux --> Copybara["Copybara"]
      AnalyzerMux --> GoModAnalyzer["GoMod"]
      AnalyzerMux --> NpmAnalyzer["Npm"]
      AnalyzerMux --> PipAnalyzer["Pip"]
      AnalyzerMux --> CustomAnalyzer["Custom"]
      BazelCipd --> AnalyzerDemux["AnalyzerDemux"]
      BazelDep --> AnalyzerDemux
      BazelMaven --> AnalyzerDemux
      CargoAnalyzer --> AnalyzerDemux
      CipdSetup --> AnalyzerDemux
      Copybara --> AnalyzerDemux
      GoModAnalyzer --> AnalyzerDemux
      NpmAnalyzer --> AnalyzerDemux
      PipAnalyzer --> AnalyzerDemux
      CustomAnalyzer --> AnalyzerDemux
      AnalyzerDemux --> Deduplicator
   end

   subgraph Tracking ["Triage & Tracking"]
      IssueTracker[("IssueTracker")]
      Deduplicator --> Triager
      Triager --> IssueWriter
      IssueWriter --> IssueDemux["Issue Demux"]
      IssueReader --> IssueDemux
      Deduplicator -.- IssueTracker
      IssueWriter -.- IssueTracker
      IssueReader -.- IssueTracker
   end

   subgraph Action ["Action & Output"]
      IssueDemux --> RollGenerator
      subgraph Updaters ["PackageUpdaters"]
         CargoUpdater["Cargo"]
         GoModUpdater["GoMod"]
         NpmUpdater["Npm"]
         PipUpdater["Pip"]
         CustomUpdater["Custom"]
      end
      RollGenerator -.- Updaters
      RollGenerator --> Collector
   end

   classDef implementer fill:#ffe0b2,stroke:#f57c00,stroke-width:2px;
   classDef library fill:#e1f5fe,stroke:#0288d1,stroke-width:1px;

   class IssueTracker,CustomAnalyzer,CustomUpdater implementer;
   class Emitter,AnalyzerMux,BazelCipd,BazelDep,BazelMaven library;
   class CargoAnalyzer,CipdSetup,Copybara,GoModAnalyzer library;
   class NpmAnalyzer,PipAnalyzer,AnalyzerDemux,Deduplicator library;
   class Triager,IssueWriter,IssueDemux,IssueReader library;
   class RollGenerator,Collector,CargoUpdater library;
   class GoModUpdater,NpmUpdater,PipUpdater library;

1. IssueTracker#

The first step is to implement an IssueTracker to manage communication between the scanner and Buganizer.

Create a new file and define a subclass of IssueTracker:

from pw_fortifier.issue_tracker import IssueTracker

class MyIssueTracker(IssueTracker):

    def __init__(self) -> None:
        super().__init__()

Set your project’s Buganizer component ID (required):
```
self.component_id = 1234567
```
Set the primary hotlist ID used for deduplication and tracking (required):
```
self.primary_hotlist_id = 7654321
```
Set additional hotlists to attach to filed issues (optional):
```
self.extra_hotlist_ids = [9876543]
```
Set the default assignee email address if no owner is found (required):
```
self.default_assignee = 'sheriff@example.com'
```
Set default CC email addresses (optional):
```
self.ccs = ['team-alerts@example.com']
```

Implement create() to file new issues. This is largely a passthrough to Buganizer APIs:

async def create(self, issue: Issue) -> Issue:
    bug_id = await my_buganizer_api.create(
        component_id=self.component_id,
        title=issue.title,
        description=issue.description,
        assignee=issue.assignee or self.default_assignee,
        ccs=self.ccs,
        hotlists=self.hotlist_ids,
    )
    return issue._replace(issue_id=bug_id)

Implement read() to fetch an existing issue by ID:

async def read(self, issue_id: int) -> Issue:
    record = await my_buganizer_api.query(issue=issue_id)
    return Issue(
        issue_id=record.id,
        title=record.title,
        description=record.description,
        assignee=record.assignee,
    )

Implement read_hotlist() to stream issues matching a hotlist ID:

async def read_hotlist(self, hotlist_id: int) -> AsyncIterator[Issue]:
    records = await my_buganizer_api.query(hotlist=hotlist_id)
    for record in records:
        yield Issue(
            issue_id=record.id,
            title=record.title,
            description=record.description,
            assignee=record.assignee,
        )

For a complete working example, see pw_fortifier/py/pw_fortifier/demo_issue_tracker.py.

Note

The Pigweed team maintains an internal IssueTracker implementation using Google Buganizer APIs. Googlers can contact Pigweed at go/pigweed-communication for help adapting it to their projects.

2. FreshnessScanner#

Next, set up the scanner executable by subclassing FreshnessScanner.

Create a script (e.g. my_freshness_scanner.py) and define the scanner skeleton:

import asyncio
import sys

from pw_fortifier.freshness_scanner import FreshnessScanner

class MyFreshnessScanner(FreshnessScanner):

    def __init__(self) -> None:
        super().__init__('my_freshness_scanner')

async def main() -> None:
    scanner = MyFreshnessScanner()
    await scanner.run(*sys.argv[1:])

if __name__ == '__main__':
    asyncio.run(main())

Set the repository URL to scan:

self.repo_url = 'sso://repo-host/my-project'

Instantiate the IssueTracker defined in Step 1:

from my_issue_tracker import MyIssueTracker

# In MyFreshnessScanner.__init__:
self.issue_tracker = MyIssueTracker()

Import and register a provided package analyzer via register():

from pw_fortifier.bazel_dep import BazelDepAnalyzer

# In MyFreshnessScanner.__init__:
self.register(BazelDepAnalyzer())

Repeat for each package type in your repository. For analyzers with companion updaters, pass the updater as well:

from pw_fortifier.cargo import CargoAnalyzer, CargoUpdater
from pw_fortifier.npm import NpmAnalyzer, NpmUpdater

# In MyFreshnessScanner.__init__:
self.register(CargoAnalyzer(), CargoUpdater())
self.register(NpmAnalyzer(), NpmUpdater())

For a complete working example, see pw_fortifier/py/demo_freshness_scanner.py.

Available package analyzers#

The following package analyzers are provided by pw_fortifier:

BazelCipdAnalyzer: CIPD packages in MODULE.bazel.
BazelDepAnalyzer: BCR dependencies in MODULE.bazel.
BazelMavenAnalyzer: Maven artifacts in MODULE.bazel.
CargoAnalyzer: Rust crates in Cargo.toml and Cargo.lock. (Pairs with CargoUpdater).
CipdSetupAnalyzer: CIPD JSON manifests in pw_env_setup.
CopybaraAnalyzer: Third-party repositories in copy.bara.sky.
GoModAnalyzer: Go modules in go.mod. (Pairs with GoModUpdater).
NpmAnalyzer: Node packages in package.json and package-lock.json. (Pairs with NpmUpdater).
PipAnalyzer: Python requirements in setup.cfg. (Pairs with PipUpdater).

3. Custom PackageAnalyzers#

To scan dependency types not covered by provided analyzers, implement a custom PackageAnalyzer.

Subclass PackageAnalyzer and set PKG_TYPE:

from pw_fortifier.package_analyzer import PackageAnalyzer


class MyCustomAnalyzer(PackageAnalyzer):
    PKG_TYPE = 'custom_pkg'

Configure target file discovery:
- For distributed files matching a specific name across the repository, set TARGET and implement _process_one():
```
TARGET = 'dependencies.json'

async def _process_one(self, path: AsyncPath) -> None:
    # Parse file at path...
```
- For a single fixed-location file (e.g. root workspace configuration), override _set_up() to read the file directly and enqueue results.
In the parsing logic, query available versions from your registry and call find_lowest() to find the earliest valid version:
```
earliest = self.find_lowest(
    TIER2_DEVHOST, current_version, available_versions
)
```

Construct and emit a FreshnessResult using _send_result():

if earliest and earliest.version != current_ver:
    result = FreshnessResult(
        package=pkg_name,
        location=f'{rel_path}:{line}',
        pkg_type=self.PKG_TYPE,
        current=current_version,
        earliest=earliest,
        tier=TIER2_DEVHOST,
    )
    await self._send_result(result)

Register your custom analyzer in your scanner via register():

from my_custom_analyzer import MyCustomAnalyzer

# In MyFreshnessScanner.__init__:
self.register(MyCustomAnalyzer())

4. Custom PackageUpdaters#

By default, RollGenerator attempts rolls by text replacement of version strings. For package types that require tool invocations (e.g. cargo update or lockfile generation), implement a custom PackageUpdater.

Subclass PackageUpdater and set PKG_TYPE or TARGET:

from pw_fortifier.package_updater import PackageUpdater


class MyCustomUpdater(PackageUpdater):
    PKG_TYPE = 'custom_pkg'
    TARGET = 'dependencies.json'

Implement update() to apply changes in the workspace and verify the build:

async def update(
    self,
    dst_repo: WritableGitWorkspace,
    result: FreshnessResult,
) -> bool:
    # Modify files in dst_repo using relevant tools
    # Run build and presubmit validation
    return True

Register the updater alongside your analyzer via register():

from my_custom_analyzer import MyCustomAnalyzer
from my_custom_updater import MyCustomUpdater

# In MyFreshnessScanner.__init__:
self.register(MyCustomAnalyzer(), MyCustomUpdater())

5. Build targets#

Define a pw_py_binary target in BUILD.bazel to run your scanner:

Load pw_py_binary:

load("//pw_build:python.bzl", "pw_py_binary")

Define the binary target and include pw_fortifier in deps:

pw_py_binary(
    name = "my_freshness_scanner",
    srcs = [
        "my_freshness_scanner.py",
    ],
    deps = [
        "@pigweed//pw_fortifier/py:pw_fortifier",
        ":my_custom_analyzer_lib",
        ":my_issue_tracker_lib",
    ],
)

Summary#

Implementation checklist:

IssueTracker
- [ ] Subclass IssueTracker.
- [ ] Set component_id, primary_hotlist_id, and default_assignee.
- [ ] (Optional) Set extra_hotlist_ids and ccs.
- [ ] Implement create() to file bugs in Buganizer.
- [ ] Implement read() to query bugs by ID.
- [ ] Implement read_hotlist() to stream bugs by hotlist.
FreshnessScanner
- [ ] Subclass FreshnessScanner and call super().__init__().
- [ ] Set repo_url.
- [ ] Instantiate and set self.issue_tracker.
- [ ] Register desired package analyzers (and optional updaters).
Custom PackageAnalyzers (optional)
- [ ] Subclass PackageAnalyzer and set PKG_TYPE.
- [ ] Set TARGET and implement _process_one() (or _set_up()).
- [ ] Query available versions and call find_lowest().
- [ ] Emit findings via _send_result().
- [ ] Register with self.register().
Custom PackageUpdaters (optional)
- [ ] Subclass PackageUpdater and set PKG_TYPE or TARGET.
- [ ] Implement update() and return success boolean.
- [ ] Register via self.register(analyzer, updater).
Build targets
- [ ] Define pw_py_binary in BUILD.bazel.
- [ ] Add //pw_fortifier/py:pw_fortifier to deps.

Testing and debugging#

Run the scanner with:

# Local dry run across the repository (no bugs filed, no CLs uploaded)
$ bazelisk run //path/to:my_freshness_scanner -- -s /path/to/my/project

# Scan specific files and output a CSV report
$ bazelisk run //path/to:my_freshness_scanner -- \
    -s /path/to/my/project \
    -f "**/MODULE.bazel" "**/Cargo.toml" -o freshness.csv

# Run a full scan, filing bugs and uploading roll CLs
$ bazelisk run //path/to:my_freshness_scanner -- \
    -s /path/to/my/project -b -u

The Scanner framework includes a few features to assist implementers in testing and debugging pipeline stages.

Execution flags and permissions#

Scanners operate with safe defaults that do not modify remote trackers or repositories unless explicitly permitted via command-line flags:

Findings are reported locally and not filed in Buganizer unless -b,–create-bugs is specified.
Code modifications and validation builds are skipped unless -e,–allow-edits is specified.
Local Git changes are kept locally and not uploaded to Gerrit unless -u,–allow-uploads is specified.

Run resumption#

If an invocation of the scanner is interrupted, the in-flight inputs will be saved as files under stage-specific subdirectories of the working directory. Implementers may examine these files to debug assertions and exceptions. They may also resume an interrupted run via the -r,–resume command line argument.

Retry configuration#

By default, consumers will try to process each input, and retry on request or subprocess failure up to the maximum number of times specified via the -m,–max-retries command line argument. When the maximum number of retries has been reached, the input will be moved to a stage-specific “error” subdirectory of the working directory, e.g. “triager_err”. By setting the maximum number of retries to zero, i.e. -m 0, implementers can cause the scanner pipeline to fail fast and then inspect the offending inputs to debug.

Input preservation#

Each pipeline consumer has a _preserve_inputs attribute.

When _preserve_inputs=False (the default for most intermediate stages), the consumer moves incoming files from the upstream stage’s output directory into its own <stage>_in/ directory and deletes (unlinks) them once processing succeeds.
When _preserve_inputs=True, the consumer processes input files directly in place in the upstream directory without moving or deleting them.

This attribute is enabled by default on initial analysis stages (such as CodeAnalyzer and PackageAnalyzer) so that source repository files are never moved or deleted.

Stage implementers can also enable _preserve_inputs=True on their stage and the subsequent stage to debug transformations: the inputs to the stage remain in <prev_stage>_out/ and the outputs remain in <stage>_out/, allowing direct comparison of the input and output state.