Skip to content
Sign up

flagon-io/g1t

public
00

Where people and agents ship software together. The open-source git platform for the whole job: issues, agents, checks and deploys to the edge.

g1t.sh
gitagentscloudflarerustforgemcp
290 linesCodeBlameRaw

Pick any line to see why it is the way it is: the commit, the pull request and issue it came from, and what the agent was thinking.

A deploy makes the runner's image in the registry, without Docker: it reads the base's manifest and config from Cloudflare's registry, uploads one gzipped layer holding the runner binary at /usr/local/bin/g1t-runner with mode 0755, and writes a config (the entrypoint g1t-runner, the base's command cleared, as the Dockerfile's ENTRYPOINT does) and a manifest that name the base's layers and that one, under the runner's tag, for linux/amd64 (scripts/deploy/registry.mjs, the registry's own upload calls with the credentials Wrangler gives). Deploy #256 ran out of disk because docker build pulled the 4.5 GB base into the job's own Docker Engine, compressed and unpacked, on a 20 GB sandbox whose own filesystem is that same base; now nothing of the base is downloaded and the job's Docker Engine is never started. Docker is asked for only when it is needed: a dry run (build, or image --no-push) still builds services/runner/Dockerfile with it, and so does a deploy whose registry step fails, logging df -h and docker system df before and after. The image job removes npm's downloads, which it only restores, once its install has read them, and logs the disk before and after the deploy. Tests cover the layer's tar, the config and manifest, the calls against a registry kept in memory (the base's layers are never fetched), the token flow, and that the Dockerfile says the same path, mode and entrypoint; the image was also checked against a local registry, where it runs and has the same config as a docker build of the Dockerfile.1// The runner's image made in the registry, with no Docker: the base's
2// manifest and config are read from Cloudflare's registry, one layer
3// holding the runner binary is uploaded beside the base's layers, and a
4// new config and manifest that name the base's layers and that one are
5// written under the runner's tag. Nothing of the base is downloaded, so a
6// deploy job needs no room for the 4.5 GB base, which a `docker build`
7// pulls and unpacks on the same disk the job runs on.
8//
9// What it makes is what services/runner/Dockerfile builds on the same
10// base: the binary at /usr/local/bin/g1t-runner with mode 0755 (COPY
11// --chmod=0755), ENTRYPOINT ["g1t-runner"] with the base's CMD cleared
12// (as a Dockerfile's ENTRYPOINT clears an inherited one), for linux/amd64.
13// A test holds the Dockerfile to these.
14//
15// The calls are the OCI distribution API (docs.docker.com/registry/spec/api/),
16// as `docker push` makes them: HEAD a blob to skip one already there; POST
17// to start an upload, PATCH its bytes, PUT with its digest to finish; PUT
18// the manifest by tag.
19
20import { createHash } from "node:crypto";
21import { gzipSync } from "node:zlib";
22
23export const RUNNER_BINARY_PATH = "usr/local/bin/g1t-runner";
24export const RUNNER_BINARY_MODE = 0o755;
25export const RUNNER_ENTRYPOINT = ["g1t-runner"];
26
27export const OCI_MANIFEST = "application/vnd.oci.image.manifest.v1+json";
28export const DOCKER_MANIFEST = "application/vnd.docker.distribution.manifest.v2+json";
29const OCI_INDEX = "application/vnd.oci.image.index.v1+json";
30const DOCKER_LIST = "application/vnd.docker.distribution.manifest.list.v2+json";
31const ACCEPT = [OCI_MANIFEST, DOCKER_MANIFEST, OCI_INDEX, DOCKER_LIST].join(", ");
32
33/** Each manifest type's gzipped layer type. */
34const LAYER_TYPE = {
35 [OCI_MANIFEST]: "application/vnd.oci.image.layer.v1.tar+gzip",
36 [DOCKER_MANIFEST]: "application/vnd.docker.image.rootfs.diff.tar.gzip",
37};
38
39export const sha256 = (bytes) => `sha256:${createHash("sha256").update(bytes).digest("hex")}`;
40
41/** A tar header field: octal, zero-padded, ending in NUL. */
42function octal(value, width) {
43 return `${value.toString(8).padStart(width - 1, "0")}\0`;
44}
45
46/**
47 * A tar archive (ustar) of one regular file, owned by root. Only the file:
48 * its folders are the base's, so a layer that names them would replace
49 * their owner, mode and time with its own.
50 */
51export function tarOf(path, data, { mode = 0o644, mtime = 0 } = {}) {
52 if (Buffer.byteLength(path) > 100) throw new Error(`a path this layer cannot hold: ${path}`);
53 const header = Buffer.alloc(512);
54 header.write(path, 0, 100, "utf8");
55 header.write(octal(mode, 8), 100, 8, "ascii");
56 header.write(octal(0, 8), 108, 8, "ascii"); // uid
57 header.write(octal(0, 8), 116, 8, "ascii"); // gid
58 header.write(octal(data.length, 12), 124, 12, "ascii");
59 header.write(octal(Math.floor(mtime), 12), 136, 12, "ascii");
60 header.write(" ", 148, 8, "ascii"); // the checksum, as spaces while it is summed
61 header.write("0", 156, 1, "ascii"); // a regular file
62 header.write("ustar\0", 257, 6, "ascii");
63 header.write("00", 263, 2, "ascii");
64 header.write("root", 265, 32, "ascii");
65 header.write("root", 297, 32, "ascii");
66 let sum = 0;
67 for (const byte of header) sum += byte;
68 header.write(`${sum.toString(8).padStart(6, "0")}\0 `, 148, 8, "ascii");
69 const padding = Buffer.alloc((512 - (data.length % 512)) % 512);
70 return Buffer.concat([header, data, padding, Buffer.alloc(1024)]);
71}
72
73/**
74 * The runner's layer: the binary at its path and mode, gzipped. Returns
75 * the compressed bytes with their digest and size, and the uncompressed
76 * tar's digest (the config's diff_id).
77 */
78export function runnerLayer(binary, { mtime = 0 } = {}) {
79 const tar = tarOf(RUNNER_BINARY_PATH, binary, { mode: RUNNER_BINARY_MODE, mtime });
80 const bytes = gzipSync(tar, { level: 9 });
81 return { bytes, digest: sha256(bytes), size: bytes.length, diffId: sha256(tar) };
82}
83
84/**
85 * The runner image's config from the base's: one more layer, the
86 * entrypoint, and the history a Dockerfile build would add.
87 */
88export function runnerConfig(base, { diffId, created }) {
89 if (base.os !== "linux" || base.architecture !== "amd64") {
90 throw new Error(`the base is ${base.os}/${base.architecture}, not linux/amd64`);
91 }
92 const config = structuredClone(base);
93 config.created = created;
94 config.config = { ...(config.config ?? {}), Entrypoint: [...RUNNER_ENTRYPOINT] };
95 // An ENTRYPOINT clears the CMD a Dockerfile inherited.
96 delete config.config.Cmd;
97 config.rootfs = { type: "layers", diff_ids: [...(base.rootfs?.diff_ids ?? []), diffId] };
98 config.history = [
99 ...(base.history ?? []),
100 { created, created_by: `COPY --chmod=0755 g1t-runner /${RUNNER_BINARY_PATH} # g1t deploy`, comment: "scripts/deploy/registry.mjs" },
101 { created, created_by: `ENTRYPOINT ${JSON.stringify(RUNNER_ENTRYPOINT)}`, comment: "scripts/deploy/registry.mjs", empty_layer: true },
102 ];
103 return config;
104}
105
106/** The runner image's manifest: the base's layers, then the runner's. */
107export function runnerManifest(base, { config, layer }) {
108 const type = manifestType(base);
109 if (!LAYER_TYPE[type]) throw new Error(`a manifest of a type this cannot extend: ${type}`);
110 const manifest = {
111 schemaVersion: 2,
112 mediaType: type,
113 config: { mediaType: base.config.mediaType, digest: config.digest, size: config.size },
114 layers: [...base.layers, { mediaType: LAYER_TYPE[type], digest: layer.digest, size: layer.size }],
115 };
116 return manifest;
117}
118
119/** A manifest's type: its own, or (OCI lets it go unsaid) its config's. */
120export function manifestType(manifest) {
121 return manifest.mediaType ?? (manifest.config?.mediaType === "application/vnd.oci.image.config.v1+json" ? OCI_MANIFEST : DOCKER_MANIFEST);
122}
123
124/** A `WWW-Authenticate: Bearer realm=...,service=...,scope=...` challenge, or null. */
125export function bearerChallenge(header) {
126 if (!header || !/^bearer\s/i.test(header)) return null;
127 const params = Object.fromEntries([...header.slice(7).matchAll(/(\w+)="([^"]*)"/g)].map((m) => [m[1].toLowerCase(), m[2]]));
128 return params.realm ? params : null;
129}
130
131/** `host/name:tag`, `host/name@digest` or `host/name:tag@digest`, split. */
132export function parseRef(ref) {
133 const match = /^([^/]+)\/(.+?)(?::([^:/@]+))?(?:@(sha256:[0-9a-f]{64}))?$/.exec(ref);
134 if (!match || (!match[3] && !match[4])) throw new Error(`not an image reference with a tag or digest: ${ref}`);
135 const [, host, name, tag, digest] = match;
136 return { host, name, tag: tag ?? null, digest: digest ?? null };
137}
138
139/**
140 * A client for one repository: requests carry Basic auth from
141 * `credentials` (none when it returns null), or the bearer token a
142 * registry that asks for one gives for them; paths are under /v2/<name>/.
143 */
144export function repositoryClient(host, name, { credentials, fetchImpl = fetch, scheme = "https" }) {
145 const root = `${scheme}://${host}`;
146 let bearer = null;
147 const basic = async () => {
148 const auth = await credentials();
149 return auth ? `Basic ${Buffer.from(`${auth.username}:${auth.password}`).toString("base64")}` : null;
150 };
151 const once = async (method, url, headers, body) => {
152 const authorization = bearer ? `Bearer ${bearer}` : await basic();
153 return fetchImpl(new URL(url, root).toString(), { method, headers: authorization ? { ...headers, authorization } : headers, body });
154 };
155 const send = async (method, url, { headers = {}, body } = {}) => {
156 const response = await once(method, url, headers, body);
157 const challenge = response.status === 401 ? bearerChallenge(response.headers.get("www-authenticate")) : null;
158 if (!challenge) return response;
159 // The token flow: the realm trades the credentials for a token.
160 const realm = new URL(challenge.realm);
161 if (challenge.service) realm.searchParams.set("service", challenge.service);
162 realm.searchParams.set("scope", challenge.scope ?? `repository:${name}:pull,push`);
163 const authorization = await basic();
164 const got = await fetchImpl(realm.toString(), { method: "GET", headers: authorization ? { authorization } : {} });
165 if (!got.ok) return response;
166 const json = await got.json();
167 bearer = json.token ?? json.access_token ?? null;
168 return bearer ? once(method, url, headers, body) : response;
169 };
170 const at = (path) => `${root}/v2/${name}/${path}`;
171 const fail = async (what, response) => {
172 let detail = "";
173 try {
174 detail = (await response.text()).slice(0, 500);
175 } catch {}
176 return new Error(`${what}: the registry answered ${response.status}${detail ? ` ${detail}` : ""}`);
177 };
178 return {
179 name,
180 async manifest(reference) {
181 const response = await send("GET", at(`manifests/${reference}`), { headers: { accept: ACCEPT } });
182 if (!response.ok) throw await fail(`reading ${name} ${reference}`, response);
183 const bytes = Buffer.from(await response.arrayBuffer());
184 return { bytes, digest: sha256(bytes), json: JSON.parse(bytes.toString("utf8")) };
185 },
186 async blob(digest) {
187 const response = await send("GET", at(`blobs/${digest}`));
188 if (!response.ok) throw await fail(`reading blob ${digest}`, response);
189 const bytes = Buffer.from(await response.arrayBuffer());
190 if (sha256(bytes) !== digest) throw new Error(`blob ${digest} came back as ${sha256(bytes)}`);
191 return bytes;
192 },
193 async hasBlob(digest) {
194 const response = await send("HEAD", at(`blobs/${digest}`));
195 if (response.status === 404) return false;
196 if (!response.ok) throw await fail(`asking for blob ${digest}`, response);
197 return true;
198 },
199 /** Uploads `bytes` unless the repository has them. Returns whether it did. */
200 async putBlob(bytes, digest = sha256(bytes)) {
201 if (await this.hasBlob(digest)) return false;
202 const started = await send("POST", at("blobs/uploads/"));
203 if (started.status !== 202) throw await fail(`starting an upload of ${digest}`, started);
204 const location = started.headers.get("location");
205 if (!location) throw new Error(`the registry started an upload of ${digest} without saying where`);
206 const patched = await send("PATCH", location, {
207 headers: { "content-type": "application/octet-stream" },
208 body: bytes,
209 });
210 if (patched.status !== 202) throw await fail(`uploading ${digest}`, patched);
211 const done = new URL(patched.headers.get("location") ?? location, root);
212 done.searchParams.set("digest", digest);
213 const put = await send("PUT", done.toString());
214 if (put.status !== 201) throw await fail(`finishing the upload of ${digest}`, put);
215 return true;
216 },
217 async putManifest(tag, bytes, type) {
218 const response = await send("PUT", at(`manifests/${tag}`), { headers: { "content-type": type }, body: bytes });
219 if (response.status !== 201) throw await fail(`writing ${name}:${tag}`, response);
220 return response.headers.get("docker-content-digest");
221 },
222 };
223}
224
225/**
226 * Makes the runner's image `ref` in the registry from the base `base` (by
227 * digest, as base.json records it) and the binary's bytes. Returns the new
228 * manifest's digest. The base must be in the same registry; its layers
229 * are named, not copied, so it must be in the same repository too.
230 */
231export async function appendRunnerImage({
232 ref,
233 base,
234 binary,
235 credentials,
236 fetchImpl = fetch,
237 scheme = "https",
238 created = new Date().toISOString(),
239 onLine = () => {},
240 attempts = 3,
241 wait = (ms) => new Promise((resolve) => setTimeout(resolve, ms)),
242}) {
243 const target = parseRef(ref);
244 const from = parseRef(base);
245 if (!target.tag) throw new Error(`the runner's image needs a tag: ${ref}`);
246 if (from.host !== target.host || from.name !== target.name) {
247 throw new Error(`the base ${base} is not in the runner's repository ${target.host}/${target.name}`);
248 }
249 const client = repositoryClient(target.host, target.name, { credentials, fetchImpl, scheme });
250
251 let manifest = await client.manifest(from.digest ?? from.tag);
252 if (from.digest && manifest.digest !== from.digest) throw new Error(`the base's manifest came back as ${manifest.digest}, not ${from.digest}`);
253 const listed = manifest.json.mediaType;
254 if (listed === OCI_INDEX || listed === DOCKER_LIST) {
255 // An index: its linux/amd64 image, as `docker build --platform linux/amd64` would pick.
256 const entry = manifest.json.manifests.find((m) => m.platform?.os === "linux" && m.platform?.architecture === "amd64");
257 if (!entry) throw new Error(`the base ${base} has no linux/amd64 image`);
258 manifest = await client.manifest(entry.digest);
259 }
260 const baseConfig = JSON.parse((await client.blob(manifest.json.config.digest)).toString("utf8"));
261 // Every layer the new manifest names must be there.
262 for (const layer of manifest.json.layers) {
263 if (!(await client.hasBlob(layer.digest))) throw new Error(`the base's layer ${layer.digest} is not in ${target.name}`);
264 }
265
266 const layer = runnerLayer(binary, { mtime: Date.parse(created) / 1000 });
267 const configBytes = Buffer.from(JSON.stringify(runnerConfig(baseConfig, { diffId: layer.diffId, created })));
268 const config = { digest: sha256(configBytes), size: configBytes.length };
269 const type = manifestType(manifest.json);
270 const manifestBytes = Buffer.from(JSON.stringify(runnerManifest(manifest.json, { config, layer })));
271 const digest = sha256(manifestBytes);
272
273 onLine(`${target.tag}: the binary's layer ${layer.digest.slice(7, 19)}, ${(layer.size / 1e6).toFixed(1)} MB, on ${manifest.json.layers.length} layers of the base`);
274 for (let attempt = 1; ; attempt++) {
275 try {
276 if (await client.putBlob(layer.bytes, layer.digest)) onLine(`uploaded the binary's layer`);
277 await client.putBlob(configBytes, config.digest);
278 const answered = await client.putManifest(target.tag, manifestBytes, type);
279 if (answered && answered !== digest) throw new Error(`the registry kept ${target.tag} as ${answered}, not ${digest}`);
280 onLine(`${target.tag}: ${digest}`);
281 return digest;
282 } catch (error) {
283 // Cloudflare's registry can answer "blob unknown" for a blob it has
284 // just taken; asking again finds it there.
285 if (attempt >= attempts) throw error;
286 onLine(`making ${target.tag} in the registry did not finish (attempt ${attempt} of ${attempts}): ${error.message ?? error}; trying again`);
287 await wait(2000 * attempt);
288 }
289 }
290}