fix(ds): implement nomad monitors with signals

lib/chirp-workflow/macros/src/lib.rs

@@ -311,9 +311,9 @@ pub fn signal(attr: TokenStream, item: TokenStream) -> TokenStream {
 	if !name
 		.value()
 		.chars()
-		.all(|c| c.is_alphanumeric() || c == '_')
+		.all(|c| c.is_ascii_lowercase() || c == '_')
 	{
-		return error(name.span(), "invalid signal name, must be [A-Za-z_]");
+		return error(name.span(), "invalid signal name, must be [a-z_]");
 	}
 
 	let item_struct = parse_macro_input!(item as ItemStruct);
@@ -362,9 +362,9 @@ pub fn message(attr: TokenStream, item: TokenStream) -> TokenStream {
 	if !name
 		.value()
 		.chars()
-		.all(|c| c.is_alphanumeric() || c == '_')
+		.all(|c| c.is_ascii_lowercase() || c == '_')
 	{
-		return error(name.span(), "invalid message name, must be [A-Za-z_]");
+		return error(name.span(), "invalid message name, must be [a-z_]");
 	}
 
 	let item_struct = parse_macro_input!(item as ItemStruct);

svc/pkg/ds/src/lib.rs

@@ -12,6 +12,9 @@ pub fn registry() -> WorkflowResult<Registry> {
 	let mut registry = Registry::new();
 	registry.register_workflow::<server::Workflow>()?;
 	registry.register_workflow::<server::destroy::Workflow>()?;
+	registry.register_workflow::<server::nomad_alloc_plan::Workflow>()?;
+	registry.register_workflow::<server::nomad_alloc_update::Workflow>()?;
+	registry.register_workflow::<server::nomad_eval_update::Workflow>()?;
 
 	Ok(registry)
 }

svc/pkg/ds/src/util/mod.rs

@@ -6,8 +6,14 @@ mod oci_config;
 mod seccomp;
 pub mod test;
 
+pub const NOMAD_REGION: &str = "global";
+pub const RUNC_SETUP_CPU: i32 = 50;
+pub const RUNC_SETUP_MEMORY: i32 = 32;
+pub const RUNC_CLEANUP_CPU: i32 = 50;
+pub const RUNC_CLEANUP_MEMORY: i32 = 32;
+
 lazy_static::lazy_static! {
-	pub static ref NEW_NOMAD_CONFIG: nomad_client::apis::configuration::Configuration =
+	pub static ref NOMAD_CONFIG: nomad_client::apis::configuration::Configuration =
 		nomad_util::new_config_from_env().unwrap();
 }
 
@@ -37,7 +43,68 @@ pub fn format_nomad_port_label(port_label: &str) -> String {
 	format!("ds_{snake_port_label}")
 }
 
-pub const RUNC_SETUP_CPU: i32 = 50;
-pub const RUNC_SETUP_MEMORY: i32 = 32;
-pub const RUNC_CLEANUP_CPU: i32 = 50;
-pub const RUNC_CLEANUP_MEMORY: i32 = 32;
+// Have to patch `nomad_client::apis::allocations_api::signal_allocation` because it uses `/allocation`
+// instead of `/client/allocation`
+pub async fn signal_allocation(
+	configuration: &nomad_client::apis::configuration::Configuration,
+	alloc_id: &str,
+	namespace: Option<&str>,
+	region: Option<&str>,
+	index: Option<i64>,
+	wait: Option<&str>,
+	alloc_signal_request: Option<nomad_client_old::models::AllocSignalRequest>,
+) -> Result<
+	(),
+	nomad_client::apis::Error<nomad_client_old::apis::allocations_api::SignalAllocationError>,
+> {
+	let local_var_client = &configuration.client;
+
+	let local_var_uri_str = format!(
+		"{}/client/allocation/{alloc_id}/signal",
+		configuration.base_path,
+		alloc_id = nomad_client::apis::urlencode(alloc_id),
+	);
+	let mut local_var_req_builder = local_var_client.post(local_var_uri_str.as_str());
+
+	if let Some(ref local_var_str) = namespace {
+		local_var_req_builder =
+			local_var_req_builder.query(&[("namespace", &local_var_str.to_string())]);
+	}
+	if let Some(ref local_var_str) = region {
+		local_var_req_builder =
+			local_var_req_builder.query(&[("region", &local_var_str.to_string())]);
+	}
+	if let Some(ref local_var_str) = index {
+		local_var_req_builder =
+			local_var_req_builder.query(&[("index", &local_var_str.to_string())]);
+	}
+	if let Some(ref local_var_str) = wait {
+		local_var_req_builder =
+			local_var_req_builder.query(&[("wait", &local_var_str.to_string())]);
+	}
+	if let Some(ref local_var_user_agent) = configuration.user_agent {
+		local_var_req_builder =
+			local_var_req_builder.header(http::header::USER_AGENT, local_var_user_agent.clone());
+	}
+	local_var_req_builder = local_var_req_builder.json(&alloc_signal_request);
+
+	let local_var_req = local_var_req_builder.build()?;
+	let local_var_resp = local_var_client.execute(local_var_req).await?;
+
+	let local_var_status = local_var_resp.status();
+	let local_var_content = local_var_resp.text().await?;
+
+	if !local_var_status.is_client_error() && !local_var_status.is_server_error() {
+		Ok(())
+	} else {
+		let local_var_entity: Option<
+			nomad_client_old::apis::allocations_api::SignalAllocationError,
+		> = serde_json::from_str(&local_var_content).ok();
+		let local_var_error = nomad_client::apis::ResponseContent {
+			status: local_var_status,
+			content: local_var_content,
+			entity: local_var_entity,
+		};
+		Err(nomad_client::apis::Error::ResponseError(local_var_error))
+	}
+}

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

@@ -30,7 +30,8 @@ async fn worker(ctx: &OperationContext<ds::msg::drain_all::Message>) -> GlobalRe
 				"server_id": server_id,
 			}),
 			crate::workflows::server::Destroy {
-				override_kill_timeout_ms: (drain_timeout < kill_timeout_ms).then(|| drain_timeout),
+				override_kill_timeout_ms: (drain_timeout < kill_timeout_ms)
+					.then_some(drain_timeout),
 			},
 		)
 		.await?;

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

@@ -1,12 +1,4 @@
 mod drain_all;
-pub mod nomad_monitor_alloc_plan;
-pub mod nomad_monitor_alloc_update;
-pub mod nomad_monitor_eval_update;
 mod webhook;
 
-chirp_worker::workers![
-	drain_all,
-	nomad_monitor_alloc_plan,
-	nomad_monitor_alloc_update,
-	nomad_monitor_eval_update
-];
+chirp_worker::workers![drain_all,];