fix(ds): fix ds tests, traefik, nomad monitors, job server drain

lib/cache/build/src/req_config.rs

@@ -157,8 +157,7 @@ impl RequestConfig {
 		//
 		// Drop `keys` bc this is not the same as the keys list in `ctx`, so it should not be used
 		// again.
-		let mut ctx = GetterCtx::new(base_key.clone().into(), keys.to_vec());
-		drop(keys);
+		let mut ctx = GetterCtx::new(base_key.clone().into(), keys);
 
 		// Build keys to look up values in Redis
 		let redis_keys = ctx

proto/backend/ds.proto

@@ -6,6 +6,8 @@ import "proto/common.proto";
 import "proto/backend/captcha.proto";
 import "proto/backend/region.proto";
 
+// TODO: Delete file after converting ds-log pkg to native ops
+
 message Server {
 	reserved 8;
 

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

@@ -44,7 +44,7 @@ pub async fn build_ds(
 ) -> GlobalResult<()> {
 	let dynamic_servers = ctx
 		.cache()
-		.ttl(60)
+		.ttl(60_000)
 		.fetch_one_json("servers_ports", dc_id, |mut cache, dc_id| async move {
 			let rows = sql_fetch_all!(
 				[ctx, DynamicServer]
@@ -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_', gg.port_name)
+					ip.nomad_label = CONCAT('ds_', REPLACE(gg.port_name, '-', '_'))
 				WHERE
 					s.datacenter_id = $1 AND
 					s.destroy_ts IS NULL
@@ -116,8 +116,6 @@ pub async fn build_ds(
 		},
 	);
 
-	tracing::info!(?config, "config timeeee");
-
 	// TODO: add middleware & services & ports
 	// TODO: same as jobs, watch out for namespaces
 	Ok(())

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

@@ -27,14 +27,10 @@ pub async fn config(
 	let mut config = types::TraefikConfigResponse::default();
 
 	// Fetch configs and catch any errors
-	tracing::info!(?config, "traefik config ds");
-	tracing::info!("asdgaerwvsdfvasdf");
 
 	build_ds(&ctx, datacenter, &mut config).await?;
 	build_job(&ctx, datacenter, &mut config).await?;
 
-	tracing::info!(?config, "traefik config ds");
-
 	// tracing::info!(
 	// 	http_services = ?config.http.services.len(),
 	// 	http_routers = config.http.routers.len(),

svc/pkg/cluster/src/workflows/server/drain.rs

@@ -79,7 +79,7 @@ async fn drain_node(ctx: &ActivityCtx, input: &DrainNodeInput) -> GlobalResult<b
 
 	if let Some(nomad_node_id) = nomad_node_id {
 		// Drain complete message is caught by `cluster-nomad-node-drain-complete`
-		nodes_api::update_node_drain(
+		let res = nodes_api::update_node_drain(
 			&NOMAD_CONFIG,
 			&nomad_node_id,
 			models::NodeUpdateDrainRequest {
@@ -102,7 +102,19 @@ async fn drain_node(ctx: &ActivityCtx, input: &DrainNodeInput) -> GlobalResult<b
 			None,
 			None,
 		)
-		.await?;
+		.await;
+
+		// Catch "node not found" error
+		if let Err(nomad_client::apis::Error::ResponseError(
+			nomad_client::apis::ResponseContent { content, .. },
+		)) = res
+		{
+			if content == "node not found" {
+				tracing::warn!("node does not exist, not draining");
+
+				return Ok(false);
+			}
+		}
 
 		// Prevent new matchmaker requests to the node running on this server
 		msg!([ctx] mm::msg::nomad_node_closed_set(&nomad_node_id) {

svc/pkg/cluster/src/workflows/server/undrain.rs

@@ -67,7 +67,7 @@ async fn undrain_node(ctx: &ActivityCtx, input: &UndrainNodeInput) -> GlobalResu
 	.await?;
 
 	if let Some(nomad_node_id) = nomad_node_id {
-		nodes_api::update_node_drain(
+		let res = nodes_api::update_node_drain(
 			&NOMAD_CONFIG,
 			&nomad_node_id,
 			models::NodeUpdateDrainRequest {
@@ -86,7 +86,19 @@ async fn undrain_node(ctx: &ActivityCtx, input: &UndrainNodeInput) -> GlobalResu
 			None,
 			None,
 		)
-		.await?;
+		.await;
+
+		// Catch "node not found" error
+		if let Err(nomad_client::apis::Error::ResponseError(
+			nomad_client::apis::ResponseContent { content, .. },
+		)) = res
+		{
+			if content == "node not found" {
+				tracing::warn!("node does not exist, not undraining");
+
+				return Ok(());
+			}
+		}
 
 		// Allow new matchmaker requests to the node running on this server
 		msg!([ctx] mm::msg::nomad_node_closed_set(&nomad_node_id) {

svc/pkg/ds/src/workers/nomad_monitor_alloc_plan.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use chirp_worker::prelude::*;
-use rivet_operation::prelude::proto::backend::{self, pkg::*};
+use rivet_operation::prelude::proto::backend::pkg::*;
 use serde::Deserialize;
 
 use crate::util::NEW_NOMAD_CONFIG;
@@ -31,7 +31,15 @@ struct RunData {
 	nomad_node_name: String,
 	nomad_node_public_ipv4: String,
 	nomad_node_vlan_ipv4: String,
-	ports: Vec<backend::job::Port>,
+	ports: Vec<Port>,
+}
+
+#[derive(Clone, Debug)]
+struct Port {
+	label: String,
+	source: u32,
+	target: u32,
+	ip: String,
 }
 
 #[worker(name = "ds-nomad-monitor-alloc-plan")]
@@ -74,7 +82,7 @@ async fn worker(
 				for port in dynamic_ports {
 					// Don't share connect proxy ports
 					let label = unwrap_ref!(port.label);
-					ports.push(backend::job::Port {
+					ports.push(Port {
 						label: label.clone(),
 						source: *unwrap_ref!(port.value) as u32,
 						target: *unwrap_ref!(port.to) as u32,

svc/pkg/ds/src/workers/nomad_monitor_eval_update.rs

@@ -27,12 +27,12 @@ async fn worker(
 ) -> GlobalResult<()> {
 	let payload_value = serde_json::from_str::<serde_json::Value>(&ctx.payload_json)?;
 
-	let job_id = unwrap!(unwrap!(payload_value.get("JobID"), "eval has no job id").as_str());
-	let eval_status_raw = unwrap!(unwrap!(payload_value.get("Status")).as_str());
-
 	// We can't decode this with serde, so manually deserialize the response
 	let eval_value = unwrap!(payload_value.get("Evaluation"));
 
+	let job_id = unwrap!(unwrap!(eval_value.get("JobID"), "eval has no job id").as_str());
+	let eval_status_raw = unwrap!(unwrap!(eval_value.get("Status")).as_str());
+
 	if !util_job::is_nomad_job_run(job_id) {
 		tracing::info!(%job_id, "disregarding event");
 		return Ok(());