Restack, add various fixes branch

lib/bolt/core/src/dep/terraform/gen.rs

@@ -261,7 +261,7 @@ async fn vars(ctx: &ProjectContext) {
 				"name": "Nomad",
 				"service": "http://nomad-server.nomad.svc.cluster.local:4646",
 				"access_groups": access.as_ref().map(|x| vec![x.groups.engineering.clone()]).unwrap_or_default(),
-				"service_tokens": []
+				"service_tokens": [],
 			}),
 		);
 

svc/pkg/faker/ops/job-template/src/lib.rs

@@ -161,7 +161,7 @@ fn gen_task(ctx: &OperationContext<faker::job_template::Request>) -> GlobalResul
 									});
 								})
 								.listen(process.env.NOMAD_PORT_tcp);
-						"#
+							"#
 						)
 						.into(),
 					),
@@ -212,7 +212,7 @@ fn gen_task(ctx: &OperationContext<faker::job_template::Request>) -> GlobalResul
 									});
 								})
 								.bind(process.env.NOMAD_PORT_udp);
-						"#
+							"#
 						)
 						.into(),
 					),
@@ -243,10 +243,10 @@ fn gen_task(ctx: &OperationContext<faker::job_template::Request>) -> GlobalResul
 						embedded_tmpl: Some(
 							indoc!(
 								r#"
-							#!/bin/sh
-							cat ${NOMAD_TASK_DIR}/stdout.txt
-							cat ${NOMAD_TASK_DIR}/stderr.txt > /dev/stderr
-							"#
+								#!/bin/sh
+								cat ${NOMAD_TASK_DIR}/stdout.txt
+								cat ${NOMAD_TASK_DIR}/stderr.txt > /dev/stderr
+								"#
 							)
 							.into(),
 						),
@@ -318,14 +318,14 @@ fn gen_task(ctx: &OperationContext<faker::job_template::Request>) -> GlobalResul
 					dest_path: Some("local/run.sh".into()),
 					embedded_tmpl: Some(formatdoc!(
 						r#"
-					#!/bin/sh
-					counter=0
-					while [ true ]; do
-						echo "Counter: $counter"
-						let 'counter++'
-						sleep {sleep}
-					done
-					"#,
+						#!/bin/sh
+						counter=0
+						while [ true ]; do
+							echo "Counter: $counter"
+							let 'counter++'
+							sleep {sleep}
+						done
+						"#,
 						sleep = counter.interval_ms as f64 / 1000.,
 					)),
 					..Template::new()

svc/pkg/faker/ops/job-template/templates/allocate.rs

@@ -0,0 +1,35 @@
+// cargo-deps: ctrlc="3.4.2"
+
+use std::{env, thread, time::Duration, sync::{Arc, atomic::{AtomicBool, Ordering}}};
+
+fn main() {
+	// Parse the first command-line argument as the array size
+	let args: Vec<String> = env::args().collect();
+	if args.len() < 2 {
+		eprintln!("missing size arg");
+		std::process::exit(1);
+	}
+	let size = args[1].parse::<usize>().expect("Error parsing size");
+
+	// Allocate the array
+	println!("allocating {size} bytes");
+	let _array = vec![0u8; size];
+
+	handle_ctrl_c();
+
+	println!("exiting");
+}
+
+fn handle_ctrl_c() {
+	let running = Arc::new(AtomicBool::new(true));
+	let r = running.clone();
+
+	ctrlc::set_handler(move || {
+		r.store(false, Ordering::SeqCst);
+	}).expect("Error setting Ctrl-C handler");
+
+	// Wait for ctrl + c
+	while running.load(Ordering::SeqCst) {
+		thread::sleep(Duration::from_secs(1));
+	}
+}

svc/pkg/game/ops/version-validate/src/lib.rs

@@ -17,6 +17,7 @@ use std::collections::HashSet;
 
 // See version-info-game-mode.ts
 const MAX_LOBBY_SIZE: u32 = 256;
+const MAX_MAX_PLAYERS: u32 = 256;
 const MAX_MIN_IDLE_LOBBY_COUNT: u32 = 16;
 const MAX_MAX_IDLE_LOBBY_COUNT: u32 = 32;
 const MAX_CUSTOM_DISPLAY_NAME_LEN: usize = 11;
@@ -463,7 +464,7 @@ async fn handle(
 					"max-players-normal",
 					"too-low",
 				]);
-			} else if lobby_group.max_players_normal > MAX_LOBBY_SIZE {
+			} else if lobby_group.max_players_normal > MAX_MAX_PLAYERS {
 				errors.push(util::err_path![
 					"config",
 					"matchmaker",
@@ -485,7 +486,7 @@ async fn handle(
 					"max-players-direct",
 					"too-low",
 				]);
-			} else if lobby_group.max_players_direct > MAX_LOBBY_SIZE {
+			} else if lobby_group.max_players_direct > MAX_MAX_PLAYERS {
 				errors.push(util::err_path![
 					"config",
 					"matchmaker",
@@ -507,7 +508,7 @@ async fn handle(
 					"max-players-party",
 					"too-low",
 				]);
-			} else if lobby_group.max_players_party > MAX_LOBBY_SIZE {
+			} else if lobby_group.max_players_party > MAX_MAX_PLAYERS {
 				errors.push(util::err_path![
 					"config",
 					"matchmaker",

svc/pkg/job-run/ops/metrics-log/Cargo.toml

@@ -6,13 +6,13 @@ authors = ["Rivet Gaming, LLC <developer@rivet.gg>"]
 license = "Apache-2.0"
 
 [dependencies]
-rivet-operation = { path = "../../../../../lib/operation/core" }
 chirp-client = { path = "../../../../../lib/chirp/client" }
+indoc = "1.0"
 prost = "0.10"
-
+reqwest = "0.11"
+rivet-operation = { path = "../../../../../lib/operation/core" }
 serde = { version = "1.0", features = ["derive"] }
 serde_urlencoded = "0.7.0"
-reqwest = "0.11"
 
 [dev-dependencies]
 chirp-worker = { path = "../../../../../lib/chirp/worker" }

svc/pkg/job-run/ops/metrics-log/src/lib.rs

@@ -1,3 +1,4 @@
+use indoc::formatdoc;
 use proto::backend::pkg::*;
 use reqwest::StatusCode;
 use rivet_operation::prelude::*;
@@ -59,26 +60,45 @@ async fn handle(
 			handle_request(
 				&prometheus_url,
 				None,
-				format!("last_over_time(nomad_client_allocs_memory_allocated{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}} [15m:15s]) or vector(0)",
+				formatdoc!(
+					"
+					last_over_time(
+						nomad_client_allocs_memory_allocated{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}}
+						[15m:15s]
+					) or vector(0)
+					",
 					nomad_job_id = metric.job,
 					task = metric.task
-			)),
+				)
+			),
 			handle_request(
 				&prometheus_url,
 				query_timing.as_ref(),
-				format!("max(nomad_client_allocs_cpu_total_percent{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}}) or vector(0)",
+				formatdoc!(
+					"
+					max(
+						nomad_client_allocs_cpu_total_ticks{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}} /
+						nomad_client_allocs_cpu_allocated{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}}
+					) or vector(0)
+					",
 					nomad_job_id = metric.job,
 					task = metric.task
-			)),
+				)
+			),
 			handle_request(
 				&prometheus_url,
 				query_timing.as_ref(),
-				// Fall back to `nomad_client_allocs_memory_rss` since `nomadusage_memory_usage` is
+				// Fall back to `nomad_client_allocs_memory_rss` since `nomad_client_allocs_memory_usage` is
 				// not available in `raw_exec`.
-				format!("max(nomad_client_allocs_memory_usage{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}}) or max(nomad_client_allocs_memory_rss{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}}) or vector(0)",
+				formatdoc!(
+					"
+					max(nomad_client_allocs_memory_usage{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}}) or
+					max(nomad_client_allocs_memory_rss{{exported_job=\"{nomad_job_id}\",task=\"{task}\"}}) or
+					vector(0)",
 					nomad_job_id = metric.job,
 					task = metric.task
-			)),
+				)
+			),
 		)?;
 
 		let (_, mem_allocated) = unwrap!(mem_allocated.value);

svc/pkg/job-run/ops/metrics-log/tests/integration.rs

@@ -4,13 +4,19 @@ use proto::backend::{self, pkg::*};
 use std::time::Duration;
 
 #[worker_test]
-async fn empty(ctx: TestCtx) {
+async fn cpu_stress(ctx: TestCtx) {
 	if !util::feature::job_run() {
 		return;
 	}
 
 	let template_res = op!([ctx] faker_job_template {
-		kind: Some(faker::job_template::request::Kind::Stress(faker::job_template::request::Stress { flags: "-c 1 -l 50".into() })),
+		kind: Some(
+			faker::job_template::request::Kind::Stress(
+				faker::job_template::request::Stress {
+					flags: "-c 1 -l 50".into(),
+				}
+			)
+		),
 		..Default::default()
 	})
 	.await
@@ -65,10 +71,80 @@ async fn empty(ctx: TestCtx) {
 				tracing::info!(?cpu, "received valid cpu metrics");
 				break;
 			} else {
-				tracing::info!("cpu metrics not high enough",)
+				tracing::info!("cpu metrics not high enough");
 			}
 		} else {
 			tracing::info!("received zeroed metrics, either Prometheus has not polled this client yet or requests are failing");
 		}
 	}
 }
+
+#[worker_test]
+async fn memory_stress(ctx: TestCtx) {
+	if !util::feature::job_run() {
+		return;
+	}
+
+	let template_res = op!([ctx] faker_job_template {
+		kind: Some(
+			faker::job_template::request::Kind::Stress(
+				faker::job_template::request::Stress {
+					flags: "--vm 1 --vm-bytes 4K --vm-hang 0".into(),
+				},
+			)
+		),
+		..Default::default()
+	})
+	.await
+	.unwrap();
+
+	let run_res = op!([ctx] faker_job_run {
+		job_spec_json: Some(template_res.job_spec_json.clone()),
+		..Default::default()
+	})
+	.await
+	.unwrap();
+	let run_id = run_res.run_id.as_ref().unwrap();
+
+	let job_res = op!([ctx] job_run_get {
+		run_ids: vec![*run_id],
+	})
+	.await
+	.unwrap();
+	let job_run = job_res.runs.first().unwrap();
+	let nomad_job_id = if let Some(backend::job::RunMeta {
+		kind: Some(backend::job::run_meta::Kind::Nomad(nomad)),
+	}) = job_run.run_meta.as_ref()
+	{
+		nomad.dispatched_job_id.as_ref().unwrap()
+	} else {
+		unreachable!()
+	};
+
+	// Poll metrics until they return non-0 results
+	loop {
+		tokio::time::sleep(Duration::from_secs(2)).await;
+
+		let now = util::timestamp::now();
+		let metrics_res = op!([ctx] job_run_metrics_log {
+			start: (now - util::duration::minutes(15)),
+			end: now,
+			step: 15000,
+			metrics: vec![job_run::metrics_log::request::Metric {
+				job: nomad_job_id.clone(),
+				task: util_job::RUN_MAIN_TASK_NAME.to_owned(),
+			}],
+		})
+		.await
+		.unwrap();
+
+		let metrics = metrics_res.metrics.first().unwrap();
+		let memory = *metrics.memory.last().unwrap();
+		if memory > 10000000000 {
+			tracing::info!(?memory, "received valid memory metrics");
+			break;
+		} else {
+			tracing::info!("memory metrics not high enough");
+		}
+	}
+}