Linehaul Field Guide

Start here

Run it on your machine

You need Python 3.11. The repo pins 3.11.2. Both functions use the python311 runtime. You do not need cloud credentials for this section.

git clone git@github.com:pypi/linehaul-cloud-function.git
cd linehaul-cloud-function
uv venv --python 3.11
uv pip install -r requirements.txt -r requirements-test.txt
python -m pytest test_functions.py tests
python -m mypy -p linehaul

CI runs the same pytest command. The argument list looks strange, but you need it. test_functions.py is at the root of the repo, because it does import main, and main.py is only at the root. The file is not in tests/. If you run pytest tests, pytest does not collect it.

The .txt requirements files have pins and hashes. They come from the .in files. Edit the .in file, then compile again. Do not edit a .txt file by hand.

The ingest path now runs on your machine. test_functions.py replaces Google Cloud Storage with pretend. It sends the two gzip log files in fixtures/ through the function body. You do not need an emulator, a network, or a GCP project.

Orientation

Two products, one name

This is what confuses most new readers: the repo holds two products in one tree, and they share almost nothing.

1 · The Cloud Functions: main.py at the root of the repo holds two Google Cloud Functions, linehaul-ingestor and linehaul-publisher, which turn Fastly CDN logs into BigQuery rows. Cloud Build deploys them from cloudbuild.yaml when you merge to main. No other code imports this file.
2 · The linehaul library: The linehaul/ package is a user-agent parser, published to PyPI as linehaul by a git tag. Warehouse pins it in requirements/main.in and imports it in warehouse/events/models.py and warehouse/accounts/services.py, where it supplies the installer name for a security event.

So a change to linehaul/ua/parser.py can break two systems, the public download data and the account-event log on pypi.org, while a change to main.py can only break one.

Linehaul is a word from the trucking industry for the long leg between two terminals.

The mechanism

End to end, once

Fig. 1. The pipeline has two legs. Events drive the ingest leg, which runs thousands of times each hour. Cron drives the publish leg, which runs 5 times each hour. The two legs meet only at gs://linehaul-bigquery-data.

01

Fastly writes the line
pypi/infra → terraform/warehouse/main.tf, terraform/file-hosting/{fastly-service.tf,vcl/files.vcl}
Each of the two Fastly services has a logging_gcs endpoint. The name of the endpoint is Linehaul GCS.

pypi.org uses a Fastly condition called Linehaul Log. The condition is true for a GET on /simple/ that returns 200 or 304. The condition is false for a shield request.

File-hosting cannot put its rule in a condition. Its endpoint uses a condition called Never, which is always false. The VCL writes the line by hand in vcl_log.

Both services read the VCL variable var.Ship-Logs-To-Line-Haul. A log snippet at priority 100 sets the variable. The Terraform input linehaul_enabled gives the value.

The two Terraform modules build 4 Fastly services. linehaul_enabled is true for pypi.org and files.pythonhosted.org. It is false for test.pypi.org and test-files.pythonhosted.org. The two test services point at a linehaul-logs-staging bucket. The flag is false, so nothing writes to that bucket. No function reads it.
02

The line goes to GCS
gs://linehaul-logs/{simple,downloads}/%Y/%m/%d/%H/%M/
Fastly collects lines for 120 seconds. It compresses them with gzip at level 9. Then it writes one object. The upload uses the service account linehaul-logs@the-psf.iam.gserviceaccount.com.
03

The new object starts linehaul-ingestor
main.py → process_fastly_log
The trigger is google.storage.object.finalize on the linehaul-logs bucket. The deploy command includes --retry.

The function downloads the object to a temporary file. It decompresses the file. It sends each line to linehaul.events.parser.parse. It puts the results in 3 temporary files: simple, downloads, and unprocessed. It uploads each file that has data. Then it deletes the source object.
04

The results go to a second bucket
gs://linehaul-bigquery-data/processed/<YYYYMMDD>/
The function writes NDJSON. The partition day is the lowest timestamp in the file. The name of the file does not set the partition.

Lines that do not parse go to unprocessed/<YYYYMMDD>/<name>.txt. A bare except: pass holds that upload. A failure to save these lines is acceptable.
05

Cron starts the publisher
Cloud Scheduler job linehaul-publisher (us-east1) → Pub/Sub linehaul-publisher-topic
Cloud Scheduler sends an empty Pub/Sub message every 12 minutes. The message has no attributes. The absence of attributes is the signal. It tells the function to calculate the days itself.
06

linehaul-publisher loads BigQuery
main.py → load_processed_files_into_bigquery
The function lists a maximum of MAX_BLOBS_PER_RUN objects. The value is 1000. It does this for the downloads- prefix and for the simple- prefix.

Then it starts a load_table_from_uri job for each table in each dataset. The location is US. It waits for the job. Then it deletes the source objects in one batch. The delete is necessary, because it makes the next run find new work.

Checkpoint 1 · the flow

Q1The publisher runs every 12 minutes, but the message has no data. How does it find the correct day?

AIt reads the newest object in the bucket and uses the day of that object. BIt looks at yesterday first. If that prefix is empty, it uses today. CIt loads every partition prefix that it can find, oldest first. DCloud Scheduler puts the partition in a Pub/Sub attribute.

B. _fetch_blobs takes a past_partition and a partition. On the cron path, past_partition is utcnow() minus 1 day. The function checks that prefix first, and returns it only if it finds objects. If it finds none, it uses today. This loads the late logs from yesterday before the pipeline moves on. On the manual path an attribute is present, past_partition is None, and only the given partition changes.

Q2The file-hosting endpoint uses a condition called Never. Its statement is req.http.Fastly-Client-IP == "127.0.0.1" && req.http.Fastly-Client-IP != "127.0.0.1". Why?

AIt is a kill switch from an old incident. BFastly needs the endpoint to exist before the VCL can use it, but the rule is too complex for a condition. CIt reduces the number of logs to almost zero, to control cost. DIt stops the logs in staging but not in production.

B. The Terraform comment gives this reason. The real rule is in files.vcl. It needs a non-shield request, a path that matches ^/packages/[a-f0-9]{2}/[a-f0-9]{2}/[a-f0-9]{60}/, a GET or OPTIONS method, and a status of 200 or 206. The rule also needs segmented_caching.is_inner_req to be false. Without that test, one range request on a large wheel counts as a second download. The input linehaul_enabled controls staging.

Wire format

The parts of a log line

Ten pipe-delimited fields feed all of the pipeline. This is a real line, from fixtures/:

download|Thu, 07 Jan 2021 20:54:54 GMT|US|/packages/f7/12/.../threadpoolctl-2.1.0-py3-none-any.whl|TLSv1.2|ECDHE-RSA-AES128-GCM-SHA256|threadpoolctl|2.1.0|bdist_wheel|pip/20.1.1 {"ci":null,"cpu":"x86_64",...}

simple|Thu, 07 Jan 2021 20:54:52 GMT|US|/simple/pyrsistent/|TLSv1.3|AES256-GCM||||pip/20.0.2 {...}

The simple line has the same shape. Its 3 project fields are empty. The Fastly format string shows the literal ||||. This is why MESSAGE_SIMPLE in the grammar has 4 PIPE tokens together.

	Field	Grammar token	Result
1	event header	`V3_HEADER` or `SIMPLE_HEADER`	selects `Download` or `Simple`
2	timestamp	`TIMESTAMP`	`.timestamp`, from the slice `d[5:-4]`
3	country code	`COUNTRY_CODE` (optional)	`.country_code`
4	URL path	`URL`	`.url`, and `file.filename` from the basename
5	TLS protocol	`TLS_PROTOCOL`	`.tls_protocol`
6	TLS cipher	`TLS_CIPHER`	`.tls_cipher`
7	project	`PROJECT_NAME`	`.project` and `file.project`
8	version	`VERSION`	`file.version`
9	package type	`PACKAGE_TYPE` (enum)	`file.type`
10	user agent	`USER_AGENT` = `rest_of_line`	`.details`, after the parse of the UA

Remember 3 details:

The header comes from the log format, not from the path. Fastly writes the literal word download or simple as the first field.
The code finds project in a different way for each event. A download uses field 7, which Fastly reads from the x-pypi-file-project response header. A simple request uses parsed.url.split("/")[2], which is a position in the path. This is why the VCL condition needs ^/simple/.+/.
printables is subtractive. parser.py makes its character class from all printables, plus the space and the tab, minus | and @. A field can hold a space, so rest_of_line is safe for the UA. One @ in any field breaks the line.

Checkpoint 2 · the grammar

Q3MESSAGE = MESSAGE_SIMPLE | MESSAGE_v3. What happens to a line whose first field is not simple and not download?

AThe code reads it as a Download with empty fields. Bpyparsing raises ParseException. That becomes UnparseableEvent. The caller catches it and writes the line to unprocessed/. CThe function fails, and Cloud Functions sends the file again forever. DThe code drops the line and keeps no record.

B. parse() raises UnparseableEvent. In process_fastly_log, a try/except Exception around each line catches every error and writes the raw bytes to the unprocessed file. This wide except is deliberate. One bad line must not cost you the other few thousand lines in the object.

Q4The timestamp hook is datetime.strptime(d[5:-4], "%d %b %Y %H:%M:%S"). What does the slice do?

AIt removes milliseconds from both ends. BIt removes the "Thu, " prefix and the " GMT" suffix from the Fastly date, and then sets UTC. CIt takes only the time. DIt removes a timezone offset such as +0000.

B. The date Thu, 07 Jan 2021 20:54:54 GMT loses 5 characters at the start and 4 at the end. The code reads the middle, then sets tzinfo=utc. The slice uses a fixed position, and it is fragile on purpose. The input comes from a Fastly format string, not from the open web. If you change timestamp_format in Terraform and not this hook, every line goes to unprocessed/.

The interesting part

The user-agent parser

The real work happens here, and this is the part you are most likely to change. linehaul/ua/ has three files:

impl.py: The machinery: the UserAgentParser abstract class, two concrete classes, and ParserSet. The @ua_parser decorator makes a CallbackUserAgentParser, and @regex_ua_parser(...) makes a RegexUserAgentParser.
parser.py: The rules: about 25 parsers, the _ignore_re denylist, and the public parse().
datastructures.py: The shape of the output, as frozen attrs classes. UserAgent holds Installer, Implementation, Distro, LibC, and System.

RegexUserAgentParser does something clever: it reads matched.re.groupindex and passes each named group to your handler as a keyword argument, each unnamed group as a positional one. That is why every handler looks like def PoetryUserAgent(*, version, impl_name, ...). The signature is the regex's group list, so when the two disagree Python raises TypeError straight away instead of quietly producing a wrong result.

Two things about ParserSet look odd:

It shuffles the list every time you register a parser. register() calls random.shuffle(self._parsers), which buys no speed at all. It exists so that a parser depending on order fails a test, rather than becoming a bug nobody finds. The set is unordered on purpose, so every parser has to work in any position.
It sorts itself under load. A counter runs down, first at 250k calls and then at every 1M, and the set re-sorts parsers by hit count, halving the counts each time as a decay. In a warm function instance, pip and uv move to the front.

The _ignore_re denylist is not a parser, and a TODO in the source explains why: ParserSet has no order, so a wide ignore pattern could hide a real parser. The code checks the denylist last, only after every parser has failed.

Fig. 2. Three exits, and 3 different results. Look at the middle row. The code does not drop an ignored UA. The line goes to unprocessed/.

Caution

THE CODE IN main.py DOES NOT DROP AN IGNORED USER AGENT. READ THIS BEFORE YOU CHANGE THE FUNCTION. THE INTENT AND THE BEHAVIOUR DO NOT AGREE.

res = parse(line.decode())
min_timestamp = min(min_timestamp, res.timestamp)   # this runs first
if res is not None:                                 # this is never False

If a UA matches _ignore_re, parse() returns None. Then res.timestamp raises AttributeError, before the code reaches the test for None. The wide except Exception catches the error and writes the line to unprocessed/. The else branch on line 109 is dead code. The denylist holds these lines. It does not drop them.

The fixtures in the repo show this. test_functions.py asserts that the (null) user-agent line is in the unprocessed output.

Destination

Where the data goes

Two tables sit in a public dataset that anyone on the internet can query.

Table	Partition	Cluster
`bigquery-public-data.pypi.file_downloads`	DAY on `timestamp`	`project`
`bigquery-public-data.pypi.simple_requests`	DAY on `timestamp`	`project`
`bigquery-public-data.pypi.distribution_metadata`	MONTH on `upload_time`	none

This repo does not produce distribution_metadata. Warehouse does. Do not look for it here.

A second cattr.Converter in main.py writes the JSON, using two custom unstructure hooks:

A datetime becomes "%Y-%m-%d %H:%M:%S +00:00", the literal string BigQuery wants for a TIMESTAMP.
Installer.subcommand passes through shlex.join(...), because pip and uv report the subcommand as a list while the BigQuery column is a single STRING. That is why the fixture shows "subcommand": "install 'something with a space'".

The load job sets ignore_unknown_values = True, and that matters in both directions. The table carries an http record this code never fills, and any field you add to datastructures.py gets dropped on load until someone adds the matching column.

Checkpoint 4 · the sink

Q8You add a uv_build_backend field to the UserAgent class and you ship it. What happens?

AThe load jobs fail with a schema error. BYou see nothing. The code writes it to the NDJSON, and the load job drops it until the column exists. CBigQuery adds the column with schema update options. DBigQuery rejects every row for that day into an error table.

B. The cause is job_config.ignore_unknown_values = True. Nothing breaks, and nothing tells you. You can believe that you shipped a feature while the data goes nowhere. A new field always needs 2 steps. Add the column to the public dataset first. Then change the code.

Q9The ingestor has BIGQUERY_DATASET = "pypi bigquery-public-data.pypi". The publisher has only "bigquery-public-data.pypi". Why?

AThe ingestor writes to two datasets. BThe ingestor never uses BigQuery. The variable does nothing on that function. CThe ingestor uses the-psf:pypi if the public dataset is not available. DOnly a 1st-gen function reads datasets that a space separates.

B. Only load_processed_files_into_bigquery reads DATASETS. The ingestor writes GCS objects and nothing else. The two functions share one main.py, so both read the module-level environment, but the value has no effect on the ingestor. It is old config from a time when the pipeline also wrote to a private the-psf:pypi dataset. That dataset no longer exists. Remove the value when you get the chance.

Operations

Faults, and what to do

Load an old day again

Use the manual path of the publisher. Send a message to the topic with attributes:

gcloud pubsub topics publish linehaul-publisher-topic \
  --project the-psf \
  --attribute partition=20260810,continue_publishing=true

If you set continue_publishing, the function sends a new message to its own topic at the end of the run. It does this only if it moved objects. The loop stops itself when no work remains. Each pass moves a maximum of 1000 objects.

CAUTION: THE CODE CALLS bool() ON THE ATTRIBUTE STRING. THE STRING "false" IS TRUE. SEND THE ATTRIBUTE ONLY WHEN YOU WANT THE LOOP.

Backlog

Two GCP limits set MAX_BLOBS_PER_RUN = 1000, and the code gives both. A load job takes 10,000 URIs. A batch delete call takes 1,000 objects. The cron runs 5 times each hour, so 5000 objects each hour is the maximum rate. If processed/ grows, raise the cap or the cron rate.

Bad objects

The deploy uses --retry. If the handler raises an error, Cloud Functions sends the object again, and it can do this forever. An endless retry on an object that never parses adds cost until someone sees it.

This is why a bad gzip file gets special code. The function catches BadGzipFile, EOFError, and zlib.error. It deletes the object and returns. A clean return is the only way to stop the retries.

Two smaller defences work in the same way. If get_blob returns None, the function returns at once, because the object is already processed. Both delete() calls accept NotFound, because a second delivery is normal.

Logs and errors

Sentry starts from SENTRY_DSN. The decorator @serverless_function wraps both entry points. All other output is print() into Cloud Logging. This is the summary line that you will search for:

Processed gs://linehaul-logs/downloads/2026/08/12/14/22/...: \
  4211 lines, 0 simple_requests, 4198 file_downloads, 13 unprocessed

Datadog and Grafana do not cover this pipeline. The Fastly side has Datadog. The GCP side has neither.

Caution

THE CODE USES os.path.basename(data["name"]).rstrip(".log.gz"). str.rstrip TAKES A SET OF CHARACTERS, NOT A SUFFIX.

It removes any run of ., l, o, g, or z at the end. A Fastly name ends with a random token, so it usually stops at once. A token that ends with zz or gol loses characters from the result name. The names stay unique enough, so this does no damage today. It is still wrong. Use removesuffix.

Checkpoint 5 · operations

Q10A bad object arrives in linehaul-logs and the ingestor raises an error. Why does the code delete the object, and not move it?

AA GCS lifecycle rule would delete it later. BThe deploy uses --retry. While the object exists and the handler raises, Cloud Functions sends it again. CStorage cost. DA move would start object.finalize again in the same bucket.

B. An endless retry on an object that never parses adds cost until someone sees it. The code catches the 3 gzip errors, prints a "Skipping malformed gzip" line, deletes the object, and returns. Only a clean return stops the next delivery. D is also a real risk, and that is why the results go to a different bucket.

Q11Every log line in a file has a timestamp from yesterday, but Fastly wrote the object today. Which partition prefix gets the results?

AYesterday. The partition is the lowest timestamp of all the lines. BToday. The partition comes from the object name. CToday. The partition is utcnow() at the time of the run. DBoth days, with one object for each day.

A. min_timestamp starts at utcnow(). Each line can lower it. The code then formats it as %Y%m%d. The value only goes down, so a file that crosses midnight goes fully into the earlier day. The publisher looks at yesterday first, so these late results still reach BigQuery.

The work

Your first change: add a new installer

A new UA parser makes a good first pull request: small, well covered by fixtures, and it moves numbers people read. You already have the environment from the top of this page.

Get a real sample. Read the unrecognised UAs in gs://linehaul-bigquery-data/unprocessed/, or query the rows where details IS NULL.
Add a parser to linehaul/ua/parser.py. Put @_parser.register above @regex_ua_parser(...). Use @ua_parser instead if you need logic, such as a version gate.
Anchor the regex. Almost every regex in the file starts with ^.
Return a plain dict with the shape of UserAgent. cattr.structure does the rest. Do not build the attrs objects yourself.
Add tests/unit/ua/fixtures/<tool>.yml. It holds a list of {ua, result} pairs. test_parser.py reads the whole directory, so the new file is the registration. A result of null asserts that the code ignores the UA.
Run python -m pytest test_functions.py tests and python -m mypy -p linehaul.
If a fixture fails sometimes, you have an overlap between two regexes. Correct the overlap. Do not run the test again until it passes.

@_parser.register
@regex_ua_parser(r"^pdm/(?P<version>\S+) (?P<impl_name>\S+)/(?P<impl_version>\S+)$")
def PDMUserAgent(*, version, impl_name, impl_version):
    return {
        "installer": {"name": "pdm", "version": version},
        "implementation": {"name": impl_name, "version": impl_version},
    }

How to ship it

There are 2 paths, and they are independent. New people usually choose the wrong one.

Merge to main. A Cloud Build trigger runs cloudbuild.yaml and deploys both functions from the source. This changes the pipeline. It does not release the library.
Push a git tag. release.yml builds an sdist and a wheel, and publishes them to PyPI with Trusted Publishing. It uses OIDC and the release environment. It holds no token. This releases the library. It does not deploy the functions.

A change to a UA parser needs both paths. It also needs a Warehouse pull request that raises the pin. Only then does pypi.org see it.

Checkpoint 6 · how to ship

Q12You merge a new UA parser to main. Which statement is true 5 minutes later?

AThe pipeline and pypi.org both use it. BThe Cloud Functions have it. pypi.org still uses the last linehaul release from PyPI. CNeither. Every deploy needs a tag. Dpypi.org has it. The functions wait for the next deploy.

B. Cloud Build deploys from the merged tree, so the functions get your parser at once. Warehouse pins linehaul from PyPI. It stays on the last tag until you cut a new tag and raise the pin. The two systems ship on different clocks. A parser can be live in the download data for weeks before it is live in the account-event log.

The map

Where the other parts are

Repo or system	What it holds	Start at
`pypi/infra`	Both Fastly services, the log format strings, the VCL gate, the bucket names, and the split between production and staging	`terraform/warehouse/main.tf`, `terraform/file-hosting/vcl/files.vcl`, `terraform/main.tf`
`pypi/warehouse`	The user of the `linehaul` library, the source of `distribution_metadata`, and the source of the `x-pypi-file-*` response headers that Fastly reads	`warehouse/events/models.py`, `warehouse/accounts/services.py`, `docs/user/api/bigquery.md`
Fastly	The live config for the above. Use `fastly service-version list` and `fastly logging gcs describe`	Terraform is the source of truth. Use the CLI to confirm what is active
GCP `the-psf`	Both functions, both buckets, the topic, the scheduler job, and the BigQuery tables	`gcloud functions describe linehaul-ingestor --region us-central1`
`psf/kubernetes-infra`	Nothing. This pipeline is serverless GCP from end to end. There is no cluster and no Cabotage	none
AWS	Nothing on this path. The bucket `psf-fastly-logs-eu-west-1` holds Fastly error logs, which linehaul does not use	none

One more item, while you are in pypi/infra. The file terraform/config.tf declares linehaul_token and linehaul_creds as sensitive variables. No other file uses them. They are from the old syslog linehaul daemon. Delete them, so that nobody believes a live credential is behind the name.

A first week

Read main.py from start to end. It is 294 lines, and it is the whole pipeline.
Read linehaul/events/parser.py with a fixture line open next to it.
Break a test on purpose. Change one byte in an expected NDJSON string, then read the diff. This shows you the shape of the output faster than the schema does.
Query the public dataset for a project that you maintain. Compare a number there against the field table above.
Open terraform/warehouse/main.tf and find the Linehaul Log condition. That one statement decides what data exists.
Get a sample of unrecognised UAs, then add a parser.

Sources: pypi/linehaul-cloud-function at 520c9bb, the pypi/infra terraform, pypi/warehouse, and live gcloud and bq output from the project the-psf, 2026-08-12.
The deploy values here (bucket names, memory, cron, environment variables) come from the running functions. The repo does not hold them.

Linehaul Field Guide

Run it on your machine

Two products, one name

End to end, once

Fastly writes the line

The line goes to GCS

The new object starts `linehaul-ingestor`

The results go to a second bucket

Cron starts the publisher

`linehaul-publisher` loads BigQuery

The parts of a log line

The user-agent parser

Where the data goes

Faults, and what to do

Load an old day again

Backlog

Bad objects

Logs and errors

Your first change: add a new installer

How to ship it

Where the other parts are

A first week

Run it on your machine

Two products, one name

End to end, once

Fastly writes the line

The line goes to GCS

The new object starts linehaul-ingestor

The results go to a second bucket

Cron starts the publisher

linehaul-publisher loads BigQuery

The parts of a log line

The user-agent parser

Where the data goes

Faults, and what to do

Load an old day again

Backlog

Bad objects

Logs and errors

Your first change: add a new installer

How to ship it

Where the other parts are

A first week

The new object starts `linehaul-ingestor`

`linehaul-publisher` loads BigQuery