feat(ds): implement pegboard dynamic server (#1158)

<!-- Please make sure there is an issue that this PR is correlated to. -->

## Changes

<!-- If there are frontend changes, please include screenshots. -->

lib/pegboard/manager/src/container/setup.rs

@@ -499,34 +499,50 @@ impl Container {
 			.config
 			.ports
 			.iter()
-			.filter(|(_, port)| matches!(port.proxy_protocol, protocol::TransportProtocol::Tcp))
+			.filter(|(_, port)| matches!(port.proxy_protocol(), protocol::TransportProtocol::Tcp))
 			.zip(
 				rows.iter()
 					.filter(|(_, protocol)| *protocol == protocol::TransportProtocol::Tcp as i64),
 			)
 			.map(|((_, port), (host_port, _))| {
-				json!({
-					"HostPort": host_port,
-					"ContainerPort": port.internal_port,
-					"Protocol": port.proxy_protocol.to_string(),
-				})
+				match port {
+					protocol::Port::GameGuard { target, proxy_protocol } => {
+						Ok(json!({
+							"HostPort": host_port,
+							"ContainerPort": port.target,
+							"Protocol": port.proxy_protocol.to_string(),
+						}))
+					}
+					protocol::Port::Host { .. } => {
+						// TODO:
+						bail!("host ports not implemented");
+					}
+				}
 			})
 			.chain(
 				self.config
 					.ports
 					.iter()
 					.filter(|(_, port)| {
-						matches!(port.proxy_protocol, protocol::TransportProtocol::Udp)
+						matches!(port.proxy_protocol(), protocol::TransportProtocol::Udp)
 					})
 					.zip(rows.iter().filter(|(_, protocol)| {
 						*protocol == protocol::TransportProtocol::Udp as i64
 					}))
 					.map(|((_, port), (host_port, _))| {
-						json!({
-							"HostPort": host_port,
-							"ContainerPort": port.internal_port,
-							"Protocol": port.proxy_protocol.to_string(),
-						})
+						match port {
+							protocol::Port::GameGuard { target, proxy_protocol } => {
+								Ok(json!({
+									"HostPort": host_port,
+									"ContainerPort": port.target,
+									"Protocol": port.proxy_protocol.to_string(),
+								}))
+							}
+							protocol::Port::Host { .. } => {
+								// TODO:
+								bail!("host ports not implemented");
+							}
+						}
 					}),
 			)
 			.collect::<Vec<_>>();

svc/api/servers/src/route/servers.rs

@@ -47,12 +47,16 @@ pub async fn create(
 		.check_game(ctx.op_ctx(), game_id, env_id, true)
 		.await?;
 
-	let clusters_get = ctx
-		.op(cluster::ops::get_for_game::Input {
+	let (clusters_res, game_configs_res) = tokio::try_join!(
+		ctx.op(cluster::ops::get_for_game::Input {
 			game_ids: vec![game_id],
-		})
-		.await?;
-	let cluster_id = unwrap!(clusters_get.games.first()).cluster_id;
+		}),
+		ctx.op(ds::ops::game_config::get::Input {
+			game_ids: vec![game_id],
+		}),
+	)?;
+	let cluster_id = unwrap!(clusters_res.games.first()).cluster_id;
+	let game_config = unwrap!(game_configs_res.game_configs.first());
 
 	let datacenters = ctx
 		.op(cluster::ops::datacenter::list::Input {
@@ -86,8 +90,9 @@ pub async fn create(
 	ctx.workflow(ds::workflows::server::Input {
 		server_id,
 		env_id,
-		cluster_id,
 		datacenter_id: body.datacenter,
+		cluster_id,
+		client: game_config.client,
 		tags,
 		resources: (*body.resources).api_into(),
 		kill_timeout_ms: body
@@ -96,6 +101,7 @@ pub async fn create(
 			.and_then(|x| x.kill_timeout)
 			.unwrap_or_default(),
 		image_id: body.runtime.build,
+		root_user_enabled: game_config.root_user_enabled,
 		args: body.runtime.arguments.unwrap_or_default(),
 		network_mode: body.network.mode.unwrap_or_default().api_into(),
 		environment: body.runtime.environment.unwrap_or_default(),
@@ -191,7 +197,7 @@ pub async fn destroy(
 	);
 
 	let mut sub = ctx
-		.subscribe::<ds::workflows::server::destroy::DestroyStarted>(&json!({
+		.subscribe::<ds::workflows::server::DestroyStarted>(&json!({
 			"server_id": server_id,
 		}))
 		.await?;

svc/api/traefik-provider/src/route/game_guard/dynamic_servers.rs

@@ -15,8 +15,8 @@ struct DynamicServer {
 	server_id: Uuid,
 	datacenter_id: Uuid,
 	label: String,
-	nomad_ip: String,
-	nomad_source: i64,
+	ip: String,
+	source: i64,
 	port_number: i64,
 	gg_port: i64,
 	port_name: String,
@@ -52,9 +52,9 @@ pub async fn build_ds(
 				SELECT
 					s.server_id,
 					s.datacenter_id,
-					ip.nomad_label AS label,
-					ip.nomad_ip,
-					ip.nomad_source,
+					ip.label,
+					ip.ip,
+					ip.source,
 					gg.port_number,
 					gg.gg_port,
 					gg.port_name,
@@ -65,7 +65,7 @@ pub async fn build_ds(
 				JOIN db_ds.docker_ports_protocol_game_guard AS gg
 				ON
 					ip.server_id = gg.server_id AND
-					ip.nomad_label = CONCAT('ds_', REPLACE(gg.port_name, '-', '_'))
+					ip.label = CONCAT('ds_', REPLACE(gg.port_name, '-', '_'))
 				WHERE
 					s.datacenter_id = $1 AND
 					s.destroy_ts IS NULL
@@ -128,8 +128,8 @@ fn ds_register_proxied_port(
 	config: &mut types::TraefikConfigResponse,
 ) -> GlobalResult<()> {
 	let ingress_port = proxied_port.gg_port;
-	let target_nomad_port_label = proxied_port.label.clone();
-	let service_id = format!("ds-run:{}:{}", run_id, target_nomad_port_label);
+	let target_port_label = proxied_port.label.clone();
+	let service_id = format!("ds-run:{}:{}", run_id, target_port_label);
 	let proxy_protocol = unwrap!(ds::types::GameGuardProtocol::from_repr(
 		proxied_port.protocol.try_into()?
 	));
@@ -144,7 +144,7 @@ fn ds_register_proxied_port(
 						servers: vec![types::TraefikServer {
 							url: Some(format!(
 								"http://{}:{}",
-								proxied_port.nomad_ip, proxied_port.nomad_source
+								proxied_port.ip, proxied_port.source
 							)),
 							address: None,
 						}],
@@ -160,10 +160,7 @@ fn ds_register_proxied_port(
 					load_balancer: types::TraefikLoadBalancer {
 						servers: vec![types::TraefikServer {
 							url: None,
-							address: Some(format!(
-								"{}:{}",
-								proxied_port.nomad_ip, proxied_port.nomad_source
-							)),
+							address: Some(format!("{}:{}", proxied_port.ip, proxied_port.source)),
 						}],
 						sticky: None,
 					},
@@ -177,10 +174,7 @@ fn ds_register_proxied_port(
 					load_balancer: types::TraefikLoadBalancer {
 						servers: vec![types::TraefikServer {
 							url: None,
-							address: Some(format!(
-								"{}:{}",
-								proxied_port.nomad_ip, proxied_port.nomad_source
-							)),
+							address: Some(format!("{}:{}", proxied_port.ip, proxied_port.source)),
 						}],
 						sticky: None,
 					},
@@ -197,7 +191,7 @@ fn ds_register_proxied_port(
 			let rule = format_http_rule(proxied_port)?;
 
 			// Hash key
-			let unique_key = (&run_id, &target_nomad_port_label, &rule, &middlewares);
+			let unique_key = (&run_id, &target_port_label, &rule, &middlewares);
 			let mut hasher = DefaultHasher::new();
 			unique_key.hash(&mut hasher);
 			let hash = hasher.finish();
@@ -220,7 +214,7 @@ fn ds_register_proxied_port(
 			let rule = format_http_rule(proxied_port)?;
 
 			// Hash key
-			let unique_key = (&run_id, &target_nomad_port_label, &rule, &middlewares);
+			let unique_key = (&run_id, &target_port_label, &rule, &middlewares);
 			let mut hasher = DefaultHasher::new();
 			unique_key.hash(&mut hasher);
 			let hash = hasher.finish();
@@ -239,7 +233,7 @@ fn ds_register_proxied_port(
 		}
 		ds::types::GameGuardProtocol::Tcp => {
 			config.tcp.routers.insert(
-				format!("ds-run:{}:{}:tcp", run_id, target_nomad_port_label),
+				format!("ds-run:{}:{}:tcp", run_id, target_port_label),
 				types::TraefikRouter {
 					entry_points: vec![format!("lb-{ingress_port}-tcp")],
 					rule: Some("HostSNI(`*`)".into()),
@@ -252,7 +246,7 @@ fn ds_register_proxied_port(
 		}
 		ds::types::GameGuardProtocol::TcpTls => {
 			config.tcp.routers.insert(
-				format!("ds-run:{}:{}:tcp-tls", run_id, target_nomad_port_label),
+				format!("ds-run:{}:{}:tcp-tls", run_id, target_port_label),
 				types::TraefikRouter {
 					entry_points: vec![format!("lb-{ingress_port}-tcp")],
 					rule: Some("HostSNI(`*`)".into()),
@@ -265,7 +259,7 @@ fn ds_register_proxied_port(
 		}
 		ds::types::GameGuardProtocol::Udp => {
 			config.udp.routers.insert(
-				format!("ds-run:{}:{}:udp", run_id, target_nomad_port_label),
+				format!("ds-run:{}:{}:udp", run_id, target_port_label),
 				types::TraefikRouter {
 					entry_points: vec![format!("lb-{ingress_port}-udp")],
 					rule: None,

svc/pkg/ds/Cargo.toml

@@ -16,8 +16,9 @@ heck = "0.3"
 hex = "0.4"
 http = "0.2"
 lazy_static = "1.4.0"
-nomad-util = { path = "../../../lib/nomad-util" }
+nix = { version = "0.27", default-features = false, features = ["signal"] }
 nomad-client-old = { package = "nomad-client", version = "0.0.9" }
+nomad-util = { path = "../../../lib/nomad-util" }
 rand = "0.8"
 regex = "1.10"
 reqwest = { version = "0.12", features = ["json"] }
@@ -49,6 +50,7 @@ ip-info = { path = "../ip/ops/info" }
 mm-config-version-get = { path = "../mm-config/ops/version-get" }
 mm-lobby-get = { path = "../mm/ops/lobby-get" }
 mm-lobby-list-for-user-id = { path = "../mm/ops/lobby-list-for-user-id" }
+pegboard = { path = "../pegboard" }
 region-get = { path = "../region/ops/get" }
 tier-list = { path = "../tier/ops/list" }
 token-create = { path = "../token/ops/create" }

svc/pkg/ds/db/servers/migrations/20240917192549_add_pegboard.up.sql

@@ -1,7 +1,13 @@
 CREATE TABLE servers_pegboard (
 	server_id UUID PRIMARY KEY REFERENCES servers,
 	pegboard_container_id UUID NOT NULL,
-	pegboard_client_id STRING,
+	pegboard_client_id UUID,
 
 	INDEX (pegboard_container_id)
 );
+
+-- Agnostify
+ALTER TABLE internal_ports
+	RENAME COLUMN nomad_label TO label,
+	RENAME COLUMN nomad_ip TO ip,
+	RENAME COLUMN nomad_source TO source;

svc/pkg/ds/src/lib.rs

@@ -17,7 +17,8 @@ pub fn registry() -> WorkflowResult<Registry> {
 	registry.register_workflow::<server::nomad::alloc_update::Workflow>()?;
 	registry.register_workflow::<server::nomad::eval_update::Workflow>()?;
 	registry.register_workflow::<server::nomad::eval_update::Workflow>()?;
-	// registry.register_workflow::<server::pegboard::Workflow>()?;
+	registry.register_workflow::<server::pegboard::Workflow>()?;
+	registry.register_workflow::<server::pegboard::destroy::Workflow>()?;
 
 	Ok(registry)
 }

svc/pkg/ds/src/ops/game_config/get.rs

@@ -0,0 +1,55 @@
+use std::convert::{TryFrom, TryInto};
+
+use chirp_workflow::prelude::*;
+
+use crate::types::{GameClient, GameConfig};
+
+#[derive(Debug, Default)]
+pub struct Input {
+	pub game_ids: Vec<Uuid>,
+}
+
+#[derive(Debug)]
+pub struct Output {
+	pub game_configs: Vec<GameConfig>,
+}
+
+#[derive(sqlx::FromRow)]
+struct GameConfigRow {
+	game_id: Uuid,
+	host_networking_enabled: bool,
+	root_user_enabled: bool,
+	client: i64,
+}
+
+impl TryFrom<GameConfigRow> for GameConfig {
+	type Error = GlobalError;
+
+	fn try_from(value: GameConfigRow) -> GlobalResult<GameConfig> {
+		Ok(GameConfig {
+			game_id: value.game_id,
+			host_networking_enabled: value.host_networking_enabled,
+			root_user_enabled: value.root_user_enabled,
+			client: unwrap!(GameClient::from_repr(value.client.try_into()?)),
+		})
+	}
+}
+
+#[operation]
+pub async fn ds_game_config_get(ctx: &OperationCtx, input: &Input) -> GlobalResult<Output> {
+	let game_configs = sql_fetch_all!(
+		[ctx, GameConfigRow]
+		"
+		SELECT game_id, host_networking_enabled, root_user_enabled, client
+		FROM db_ds.game_config
+		WHERE game_id = ANY($1)
+		",
+		&input.game_ids,
+	)
+	.await?
+	.into_iter()
+	.map(TryInto::try_into)
+	.collect::<GlobalResult<_>>()?;
+
+	Ok(Output { game_configs })
+}

svc/pkg/ds/src/ops/game_config/mod.rs

@@ -0,0 +1,2 @@
+pub mod get;
+pub mod update;