Skip to content
Sign up

flagon-io/g1t

public
0
608 linesCodeBlameRaw
1//! `g1t-runner` on your own machine: a self-hosted runner for g1t.
2//!
3//! ```text
4//! g1t-runner register --url https://g1t.sh --token g1trt_… [--name build-01]
5//! [--labels gpu,cuda] [--group Default] [--ephemeral] [--work-dir DIR]
6//! [--no-docker] [--image IMAGE] [--agent-image IMAGE] [--replace]
7//! [--memory 8g] [--cpus 4] [--pids-limit 4096] [--no-job-docker]
8//! g1t-runner run [--once]
9//! g1t-runner service install|start|stop|status|uninstall
10//! g1t-runner remove
11//! g1t-runner update
12//! g1t-runner version
13//! ```
14//!
15//! It only ever calls out, to `api.g1t.sh` over HTTPS: it polls for work
16//! (a poll waits up to 20 seconds for some, and is the runner's heartbeat),
17//! runs what it is given with the same harness g1t's sandboxes run (this
18//! program in another mode, see `exec`), and says how it ended. Nothing
19//! listens on the machine. Every command takes `--dir` (default
20//! `~/.g1t-runner`, or `G1T_RUNNER_DIR`), where its configuration is kept.
21//!
22//! How it keeps jobs from the machine is `sandbox`: in its image under
23//! Sysbox (`G1T_RUNNER_SANDBOX=nested`) it runs its own Docker Engine
24//! (`engine`) and refuses to start if it can reach the machine's.
25
26mod api;
27mod config;
28#[cfg(unix)]
29mod engine;
30mod exec;
31mod sandbox;
32mod service;
33mod update;
34
35use std::path::{Path, PathBuf};
36use std::sync::atomic::{AtomicBool, Ordering};
37use std::time::{Duration, Instant};
38
39use anyhow::{Context, Result, bail};
40use serde_json::json;
41
42use api::{Api, Failure};
43use config::Config;
44
45pub const VERSION: &str = env!("CARGO_PKG_VERSION");
46
47/// Set to stop the loop between polls: a Windows service's stop request.
48pub static STOP: AtomicBool = AtomicBool::new(false);
49
50/// How long a busy runner waits between heartbeats.
51const HEARTBEAT: Duration = Duration::from_secs(10);
52/// How long an idle poll waits for work.
53const IDLE_WAIT_MS: u64 = 20_000;
54/// How often an idle runner checks for a new version.
55const UPDATE_EVERY: Duration = Duration::from_secs(6 * 60 * 60);
56
57const COMMANDS: &[&str] = &["register", "register-and-run", "run", "service", "remove", "update", "version", "--version", "help", "--help", "-h"];
58
59/// Whether `args` (without the program) are a command of this one, rather
60/// than the harness's `MODE`.
61pub fn is_command(args: &[String]) -> bool {
62 args.first().is_some_and(|first| COMMANDS.contains(&first.as_str()))
63}
64
65pub fn log(message: &str) {
66 let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0);
67 println!("{} {message}", g1t_now(now));
68}
69
70/// RFC 3339 for the log, without pulling in a date library.
71fn g1t_now(ms: u64) -> String {
72 let seconds = ms / 1000;
73 let (days, rest) = (seconds / 86_400, seconds % 86_400);
74 let z = days as i64 + 719_468;
75 let era = z.div_euclid(146_097);
76 let doe = z.rem_euclid(146_097);
77 let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
78 let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
79 let mp = (5 * doy + 2) / 153;
80 let day = doy - (153 * mp + 2) / 5 + 1;
81 let month = if mp < 10 { mp + 3 } else { mp - 9 };
82 let year = yoe + era * 400 + i64::from(month <= 2);
83 format!("{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}Z", rest / 3600, rest % 3600 / 60, rest % 60)
84}
85
86/// `--name value` and `--flag` options.
87struct Options {
88 values: Vec<(String, String)>,
89 flags: Vec<String>,
90 rest: Vec<String>,
91}
92
93impl Options {
94 fn parse(args: &[String], takes_value: &[&str]) -> Result<Options> {
95 let mut options = Options { values: Vec::new(), flags: Vec::new(), rest: Vec::new() };
96 let mut i = 0;
97 while i < args.len() {
98 let arg = &args[i];
99 if let Some(name) = arg.strip_prefix("--") {
100 if let Some((name, value)) = name.split_once('=') {
101 options.values.push((name.to_owned(), value.to_owned()));
102 } else if takes_value.contains(&name) {
103 let Some(value) = args.get(i + 1) else { bail!("--{name} needs a value") };
104 options.values.push((name.to_owned(), value.clone()));
105 i += 1;
106 } else {
107 options.flags.push(name.to_owned());
108 }
109 } else {
110 options.rest.push(arg.clone());
111 }
112 i += 1;
113 }
114 Ok(options)
115 }
116
117 fn value(&self, name: &str) -> Option<&str> {
118 self.values.iter().rev().find(|(n, _)| n == name).map(|(_, v)| v.as_str())
119 }
120
121 fn flag(&self, name: &str) -> bool {
122 self.flags.iter().any(|f| f == name)
123 }
124
125 /// For `register-and-run` in a container: each option may come from
126 /// `G1T_RUNNER_<NAME>` instead (`G1T_RUNNER_TOKEN`, `G1T_RUNNER_LABELS`,
127 /// `G1T_RUNNER_EPHEMERAL=1`), so a Kubernetes secret can hold the token.
128 fn with_environment(mut self) -> Options {
129 let env_name = |name: &str| format!("G1T_RUNNER_{}", name.to_ascii_uppercase().replace('-', "_"));
130 for name in VALUE_OPTIONS {
131 if self.value(name).is_none()
132 && let Ok(value) = std::env::var(env_name(name))
133 && !value.is_empty()
134 {
135 self.values.push((name.to_owned(), value));
136 }
137 }
138 for name in ["ephemeral", "no-docker", "replace", "no-auto-update", "no-job-docker"] {
139 if !self.flag(name) && std::env::var(env_name(name)).is_ok_and(|v| matches!(v.as_str(), "1" | "true" | "yes")) {
140 self.flags.push(name.to_owned());
141 }
142 }
143 self
144 }
145}
146
147/// The options that take a value.
148const VALUE_OPTIONS: [&str; 14] =
149 ["url", "api", "token", "name", "labels", "group", "work-dir", "image", "agent-image", "harness", "memory", "cpus", "pids-limit", "job-docker-image"];
150
151const HELP: &str = "g1t-runner: a self-hosted runner for g1t.
152
153 g1t-runner register --url https://g1t.sh --token <registration token>
154 [--name NAME] [--labels a,b] [--group GROUP] [--ephemeral]
155 [--work-dir DIR] [--no-docker] [--image IMAGE] [--agent-image IMAGE]
156 [--api URL] [--harness PATH] [--replace] [--no-auto-update]
157 [--memory 8g] [--cpus 4] [--pids-limit 4096] [--no-job-docker]
158 [--job-docker-image IMAGE]
159 g1t-runner run [--once] poll for work and run it
160 g1t-runner register-and-run ... both, for containers; options may be
161 G1T_RUNNER_URL, _TOKEN, _LABELS, _EPHEMERAL…
162 g1t-runner service install|start|stop|status|uninstall
163 g1t-runner remove unregister this runner
164 g1t-runner update update to the newest release
165 g1t-runner version
166
167Every command takes --dir DIR (default ~/.g1t-runner, or G1T_RUNNER_DIR).
168Run it as a container under Sysbox so neither it nor its jobs reach the
169machine: docker run --runtime=sysbox-runc … g1t.sh/flagon-io/g1t-runner.
170--no-docker (host mode) is deprecated.
171Make a registration token under Settings, Runners on g1t.sh.
172Docs: https://docs.g1t.sh/guides/self-hosted-runners/";
173
174/// Runs a command. Returns the process's exit code.
175pub fn main(args: Vec<String>) -> i32 {
176 let mut takes = VALUE_OPTIONS.to_vec();
177 takes.push("dir");
178 let options = match Options::parse(&args[1..], &takes) {
179 Ok(options) => options,
180 Err(error) => {
181 eprintln!("g1t-runner: {error:#}");
182 return 2;
183 }
184 };
185 let folder = config::folder(options.value("dir"));
186 let outcome = match args[0].as_str() {
187 "register" => register(&options, &folder),
188 // A container's one command: register unless it already is, then run.
189 "register-and-run" => {
190 let options = options.with_environment();
191 let ready = if config::load(&folder).is_ok() { Ok(()) } else { register(&options, &folder) };
192 ready.and_then(|()| run(&folder, options.flag("once")))
193 }
194 "run" => run(&folder, options.flag("once")),
195 "service" => match options.rest.first().map(String::as_str) {
196 Some(action) => config::load(&folder).and_then(|config| service::main(action, &config, &folder)),
197 None => Err(anyhow::anyhow!("service takes install, uninstall, start, stop or status")),
198 },
199 "remove" => remove(&folder),
200 "update" => config::load(&folder).and_then(|config| match update::update(&config)? {
201 Some(version) => {
202 log(&format!("Updated to {version}. Restart the runner (or its service) to use it."));
203 Ok(())
204 }
205 None => {
206 log(&format!("{VERSION} is the newest release."));
207 Ok(())
208 }
209 }),
210 "version" | "--version" => {
211 println!("g1t-runner {VERSION} ({})", update::platform());
212 Ok(())
213 }
214 _ => {
215 println!("{HELP}");
216 Ok(())
217 }
218 };
219 match outcome {
220 Ok(()) => 0,
221 Err(error) => {
222 eprintln!("g1t-runner: {error:#}");
223 1
224 }
225 }
226}
227
228fn hostname() -> String {
229 for name in ["COMPUTERNAME", "HOSTNAME"] {
230 if let Ok(value) = std::env::var(name).map(|v| v.trim().to_owned())
231 && !value.is_empty()
232 {
233 return value;
234 }
235 }
236 std::process::Command::new("hostname")
237 .output()
238 .ok()
239 .map(|out| String::from_utf8_lossy(&out.stdout).trim().to_owned())
240 .filter(|name| !name.is_empty())
241 .unwrap_or_else(|| "runner".to_owned())
242}
243
244/// A runner's name from a machine's: what g1t accepts.
245fn runner_name(given: &str) -> String {
246 let name: String = given
247 .chars()
248 .map(|c| if c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.') { c } else { '-' })
249 .take(64)
250 .collect();
251 if name.is_empty() { "runner".into() } else { name }
252}
253
254fn register(options: &Options, folder: &Path) -> Result<()> {
255 let Some(url) = options.value("url") else { bail!("--url is needed: the g1t you register with, such as https://g1t.sh") };
256 let Some(token) = options.value("token") else { bail!("--token is needed: make a registration token under Settings, Runners") };
257 let url = url.trim_end_matches('/').to_owned();
258 let api = options.value("api").map(|a| a.trim_end_matches('/').to_owned()).unwrap_or_else(|| config::api_for(&url));
259 let name = runner_name(options.value("name").map(str::to_owned).unwrap_or_else(hostname).as_str());
260 let labels: Vec<String> = options
261 .value("labels")
262 .unwrap_or_default()
263 .split(',')
264 .map(|l| l.trim().to_owned())
265 .filter(|l| !l.is_empty())
266 .collect();
267 let docker = !options.flag("no-docker");
268 let isolation = isolation(docker)?;
269 if isolation == sandbox::Isolation::Host {
270 log(sandbox::HOST_MODE_AT_START);
271 } else if !sandbox::nested_requested() && !exec::docker_ready() {
272 log("Docker does not answer here. Jobs will fail until it does.");
273 }
274 let pids_limit = match options.value("pids-limit") {
275 Some(text) => Some(text.parse::<u32>().ok().filter(|n| *n > 0).with_context(|| format!("--pids-limit takes a number of processes, not {text}"))?),
276 None => None,
277 };
278 let existing = config::load(folder).ok();
279 if existing.is_some() && !options.flag("replace") {
280 bail!("{} already holds a registered runner: run `g1t-runner remove` first, or register with --replace or another --dir", folder.display());
281 }
282 // The OS jobs run on: in Docker that is Linux, whatever the machine
283 // (Docker Desktop runs Linux containers on macOS and Windows).
284 let os = match std::env::consts::OS {
285 _ if docker => "linux",
286 "macos" => "macos",
287 "windows" => "windows",
288 _ => "linux",
289 };
290 let arch = if std::env::consts::ARCH == "aarch64" { "arm64" } else { "x64" };
291 let registered = Api::register(
292 &api,
293 &json!({
294 "token": token,
295 "name": name,
296 "labels": labels,
297 "os": os,
298 "arch": arch,
299 "version": VERSION,
300 "ephemeral": options.flag("ephemeral"),
301 "group": options.value("group"),
302 "replace": options.flag("replace"),
303 "isolation": isolation.as_str(),
304 }),
305 )?;
306 let work_dir = options.value("work-dir").map(PathBuf::from).unwrap_or_else(|| folder.join("work"));
307 let config = Config {
308 url,
309 api,
310 runner: registered.runner.id.clone(),
311 name: registered.runner.name.clone(),
312 credential: registered.credential,
313 workspace: registered.runner.workspace.clone(),
314 repo: registered.runner.repo.clone(),
315 group: registered.runner.group.clone(),
316 labels: registered.runner.labels.clone(),
317 ephemeral: options.flag("ephemeral"),
318 work_dir,
319 docker,
320 image: options.value("image").map(str::to_owned),
321 agent_image: options.value("agent-image").map(str::to_owned),
322 harness: options.value("harness").map(PathBuf::from),
323 auto_update: !options.flag("no-auto-update"),
324 memory: options.value("memory").map(str::to_owned),
325 cpus: options.value("cpus").map(str::to_owned),
326 pids_limit,
327 job_docker: !options.flag("no-job-docker"),
328 job_docker_image: options.value("job-docker-image").map(str::to_owned),
329 };
330 config::save(folder, &config)?;
331 let place = match &config.repo {
332 Some(repo) => format!("for {repo}"),
333 None => format!("in {}{}", config.workspace, config.group.as_ref().map(|g| format!(", group {g}")).unwrap_or_default()),
334 };
335 log(&format!("Registered {} {place}, with labels {}.", config.name, config.labels.join(", ")));
336 log("Start it with `g1t-runner run`, or `g1t-runner service install` to keep it running.");
337 Ok(())
338}
339
340/// How this runner keeps work from its machine; refuses a nested runner
341/// that can reach the machine (sandbox.rs).
342fn isolation(docker: bool) -> Result<sandbox::Isolation> {
343 use sandbox::{Isolation, SandboxDriver};
344 if !docker {
345 // What is coming for jobs native to this OS.
346 if let Some(driver) = sandbox::native_driver()
347 && let Err(why) = driver.preflight()
348 {
349 log(&format!("Sandboxed {} jobs, in {}: {why}.", std::env::consts::OS, driver.name()));
350 }
351 return Ok(Isolation::Host);
352 }
353 if !sandbox::nested_requested() {
354 return sandbox::SharedDocker.preflight();
355 }
356 if !cfg!(target_os = "linux") {
357 bail!("{}=nested is for the runner's Linux image", sandbox::SANDBOX_ENV);
358 }
359 let isolation = sandbox::NestedDocker { allow_unsafe: sandbox::unsafe_allowed() }.preflight()?;
360 if isolation != Isolation::NestedContainer {
361 for problem in sandbox::Findings::read().problems() {
362 log(&format!("UNSAFE: {problem}"));
363 }
364 log(&format!("UNSAFE: {}=1 is set, so the runner starts anyway, and its jobs can reach this machine.", sandbox::UNSAFE_ENV));
365 }
366 Ok(isolation)
367}
368
369/// A nested runner's own Docker Engine, which it uses whatever else it
370/// can see (with the override, it starts despite that).
371fn prepare(docker: bool) -> Result<()> {
372 if !docker || !sandbox::nested_requested() {
373 return Ok(());
374 }
375 #[cfg(unix)]
376 {
377 let version = engine::start()?;
378 log(&format!("Started this runner's own Docker Engine {version}, inside its container."));
379 }
380 Ok(())
381}
382
383fn remove(folder: &Path) -> Result<()> {
384 let config = config::load(folder)?;
385 let api = Api { base: config.api.clone(), runner: config.runner.clone(), credential: config.credential.clone() };
386 match api.remove() {
387 Ok(()) | Err(Failure::Refused(401, _)) | Err(Failure::Refused(404, _)) => {}
388 Err(failure) => bail!("{failure}; the runner is still registered"),
389 }
390 config::forget(folder);
391 log(&format!("Removed {}.", config.name));
392 Ok(())
393}
394
395/// The loop, for a Windows service: its exit code.
396#[cfg_attr(not(windows), allow(dead_code))]
397pub fn serve_from_service() -> i32 {
398 let folder = config::folder(None);
399 let folder = std::env::args()
400 .collect::<Vec<_>>()
401 .windows(2)
402 .find(|pair| pair[0] == "--dir")
403 .map(|pair| PathBuf::from(&pair[1]))
404 .unwrap_or(folder);
405 match run(&folder, false) {
406 Ok(()) => 0,
407 Err(error) => {
408 eprintln!("g1t-runner: {error:#}");
409 1
410 }
411 }
412}
413
414/// Sleeps up to `duration`, waking early when `stop` says so.
415fn nap(duration: Duration, mut stop: impl FnMut() -> bool) {
416 let until = Instant::now() + duration;
417 while Instant::now() < until {
418 if STOP.load(Ordering::SeqCst) || stop() {
419 return;
420 }
421 std::thread::sleep(Duration::from_millis(250));
422 }
423}
424
425fn run(folder: &Path, once: bool) -> Result<()> {
426 let mut config = config::load(folder)?;
427 let mut api = Api { base: config.api.clone(), runner: config.runner.clone(), credential: config.credential.clone() };
428 std::fs::create_dir_all(&config.work_dir).with_context(|| format!("could not make {}", config.work_dir.display()))?;
429 let isolation = isolation(config.docker)?;
430 prepare(config.docker)?;
431 if isolation != sandbox::Isolation::Host {
432 exec::sweep(&config);
433 }
434 log(&format!(
435 "g1t-runner {VERSION}: {} ({}) is listening for work from {}. Isolation: {}.",
436 config.name,
437 config.labels.join(", "),
438 config.url,
439 isolation.describe()
440 ));
441 if isolation == sandbox::Isolation::Host {
442 log(sandbox::HOST_MODE_AT_START);
443 }
444 let mut current: Option<exec::Running> = None;
445 let mut ran_one = false;
446 let mut backoff = Duration::from_secs(2);
447 let mut last_update_check = Instant::now().checked_sub(UPDATE_EVERY).unwrap_or_else(Instant::now);
448 let mut said_no_key = false;
449 loop {
450 if STOP.load(Ordering::SeqCst) {
451 if let Some(mut running) = current.take() {
452 log(&format!("Stopping: {} is cut short.", running.name));
453 running.stop();
454 let _ = api.finished(&running.id, 1, Some("The self-hosted runner was stopped."));
455 running.clean();
456 }
457 return Ok(());
458 }
459 // What was running has ended: say how.
460 if let Some(running) = current.as_mut()
461 && let Some(code) = running.ended()
462 {
463 let mut running = current.take().expect("checked above");
464 log(&format!("{} {} ended ({code}).", running.kind, running.name));
465 if let Err(failure) = api.finished(&running.id, code, None) {
466 log(&format!("Could not say how it ended: {failure}"));
467 }
468 running.clean();
469 ran_one = true;
470 if config.ephemeral || once {
471 if config.ephemeral {
472 let _ = api.remove();
473 config::forget(folder);
474 log("The ephemeral runner ran its job and removed itself.");
475 }
476 return Ok(());
477 }
478 }
479 // Idle: a newer release, now and then.
480 if current.is_none() && config.auto_update && last_update_check.elapsed() >= UPDATE_EVERY {
481 last_update_check = Instant::now();
482 if update::can_update() {
483 match update::update(&config) {
484 Ok(Some(version)) => {
485 log(&format!("Updated to {version}; starting it."));
486 return restart();
487 }
488 Ok(None) => {}
489 Err(error) => log(&format!("Could not check for a new version: {error:#}")),
490 }
491 } else if !said_no_key {
492 said_no_key = true;
493 log("This build cannot check releases, so it will not update itself.");
494 }
495 }
496 let running: Vec<String> = current.iter().map(|r| r.id.clone()).collect();
497 let wait = if current.is_some() { 0 } else { IDLE_WAIT_MS };
498 let poll = match api.poll(&running, wait) {
499 Ok(poll) => {
500 backoff = Duration::from_secs(2);
501 poll
502 }
503 Err(Failure::Refused(401, message)) => {
504 if let Some(mut running) = current.take() {
505 running.stop();
506 running.clean();
507 }
508 bail!("g1t no longer knows this runner ({message}). Register it again.");
509 }
510 Err(failure) => {
511 log(&format!("{failure}; trying again in {}s", backoff.as_secs()));
512 nap(backoff, || false);
513 backoff = (backoff * 2).min(Duration::from_secs(60));
514 continue;
515 }
516 };
517 if let Some(fresh) = poll.credential {
518 config.credential = fresh.clone();
519 api.credential = fresh;
520 config::save(folder, &config)?;
521 }
522 if poll.removed {
523 config::forget(folder);
524 log("This runner was removed from g1t; stopping.");
525 return Ok(());
526 }
527 if let Some(running) = current.as_mut()
528 && poll.cancel.contains(&running.id)
529 {
530 log(&format!("{} was cancelled; stopping it.", running.name));
531 running.stop();
532 }
533 if let Some(running) = current.as_mut()
534 && running.overdue()
535 {
536 log(&format!("{} ran past its time limit; stopping it.", running.name));
537 running.stop();
538 }
539 if let Some(work) = poll.assignment {
540 log(&format!("Took {} {} ({}).", work.kind, work.name, work.repo));
541 match exec::start(&config, isolation, folder, &work) {
542 Ok(running) => current = Some(running),
543 Err(error) => {
544 log(&format!("Could not start it: {error:#}"));
545 let _ = api.finished(&work.id, 1, Some(&format!("The self-hosted runner {} could not start it: {error:#}", config.name)));
546 if config.ephemeral || once {
547 return Ok(());
548 }
549 }
550 }
551 continue;
552 }
553 if once && ran_one {
554 return Ok(());
555 }
556 // Busy: a heartbeat every few seconds until it ends.
557 if current.is_some() {
558 let mut check = || current.as_mut().is_some_and(|r| r.ended().is_some());
559 nap(HEARTBEAT, &mut check);
560 }
561 }
562}
563
564/// Runs the updated program in this one's place, with the same arguments.
565fn restart() -> Result<()> {
566 let me = std::env::current_exe()?;
567 let status = std::process::Command::new(me).args(std::env::args().skip(1)).status()?;
568 std::process::exit(status.code().unwrap_or(1));
569}
570
571#[cfg(test)]
572mod tests {
573 use super::*;
574
575 fn args(items: &[&str]) -> Vec<String> {
576 items.iter().map(|s| s.to_string()).collect()
577 }
578
579 #[test]
580 fn commands_are_told_apart_from_the_harness() {
581 assert!(is_command(&args(&["register", "--url", "x"])));
582 assert!(is_command(&args(&["run"])));
583 assert!(!is_command(&args(&[])));
584 assert!(!is_command(&args(&["something"])));
585 }
586
587 #[test]
588 fn options_read_both_ways() {
589 let options = Options::parse(&args(&["--url", "https://g1t.sh", "--labels=gpu,cuda", "--ephemeral", "install"]), &["url", "labels"]).unwrap();
590 assert_eq!(options.value("url"), Some("https://g1t.sh"));
591 assert_eq!(options.value("labels"), Some("gpu,cuda"));
592 assert!(options.flag("ephemeral"));
593 assert_eq!(options.rest, vec!["install".to_owned()]);
594 assert!(Options::parse(&args(&["--url"]), &["url"]).is_err());
595 }
596
597 #[test]
598 fn a_machines_name_becomes_a_runners() {
599 assert_eq!(runner_name("Build Box #1"), "Build-Box--1");
600 assert_eq!(runner_name(""), "runner");
601 assert_eq!(runner_name(&"x".repeat(80)).len(), 64);
602 }
603
604 #[test]
605 fn the_log_says_when() {
606 assert_eq!(g1t_now(1_790_918_179_123), "2026-10-02T05:16:19Z");
607 }
608}