fix: add metrics to ops and guard (#2429)

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

## Changes

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

Author: MasterPtato

packages/common/chirp-workflow/core/src/ctx/common.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use futures_util::StreamExt;
 use global_error::{GlobalError, GlobalResult};
@@ -81,6 +81,16 @@ where
 {
 	tracing::debug!(?input, "operation call");
 
+	// Record metrics
+	crate::metrics::OPERATION_PENDING
+		.with_label_values(&[I::Operation::NAME])
+		.inc();
+	crate::metrics::OPERATION_TOTAL
+		.with_label_values(&[I::Operation::NAME])
+		.inc();
+
+	let start_instant = Instant::now();
+
 	let ctx = OperationCtx::new(
 		db.clone(),
 		config,
@@ -95,9 +105,45 @@ where
 
 	let res = tokio::time::timeout(I::Operation::TIMEOUT, I::Operation::run(&ctx, &input))
 		.await
-		.map_err(|_| WorkflowError::OperationTimeout(0))?
-		.map_err(WorkflowError::OperationFailure)
-		.map_err(GlobalError::raw);
+		.map_err(|_| WorkflowError::OperationTimeout(0))
+		.map(|res| {
+			res.map_err(WorkflowError::OperationFailure)
+				.map_err(GlobalError::raw)
+		});
+
+	// Record metrics
+	{
+		let error_code_str = match &res {
+			Ok(Err(GlobalError::Internal { ty, .. })) => {
+				let err_code_str = "__UNKNOWN__".to_string();
+				crate::metrics::OPERATION_ERRORS
+					.with_label_values(&[I::Operation::NAME, &err_code_str, ty])
+					.inc();
+
+				err_code_str
+			}
+			Ok(Err(GlobalError::BadRequest { code, .. })) => {
+				crate::metrics::OPERATION_ERRORS
+					.with_label_values(&[I::Operation::NAME, code, "bad_request"])
+					.inc();
+
+				code.clone()
+			}
+			Ok(_) => String::new(),
+			Err(_) => "timeout".to_string(),
+		};
+
+		// Other request metrics
+		let dt = start_instant.elapsed().as_secs_f64();
+		crate::metrics::OPERATION_PENDING
+			.with_label_values(&[I::Operation::NAME])
+			.dec();
+		crate::metrics::OPERATION_DURATION
+			.with_label_values(&[I::Operation::NAME, error_code_str.as_str()])
+			.observe(dt);
+	}
+
+	let res = res?;
 
 	tracing::debug!(?res, "operation response");
 

packages/common/chirp-workflow/core/src/ctx/listen.rs

@@ -83,6 +83,16 @@ impl<'a> ListenCtx<'a> {
 			.with_label_values(&[self.ctx.name(), &signal.signal_name])
 			.observe(recv_lag);
 
+		if recv_lag > 15.0 {
+			// We print an error here so the trace of this workflow does not get dropped
+			tracing::error!(
+				?recv_lag,
+				signal_id=%signal.signal_id,
+				signal_name=%signal.signal_name,
+				"long signal recv time",
+			);
+		}
+
 		tracing::debug!(
 			signal_id=%signal.signal_id,
 			signal_name=%signal.signal_name,

packages/common/chirp-workflow/core/src/db/fdb_sqlite_nats/mod.rs

@@ -364,7 +364,7 @@ impl Database for DatabaseFdbSqliteNats {
 	}
 
 	#[tracing::instrument(skip_all)]
-	async fn publish_metrics(&self, worker_instance_id: Uuid) -> WorkflowResult<()> {
+	async fn publish_metrics(&self, _worker_instance_id: Uuid) -> WorkflowResult<()> {
 		// Attempt to be the only worker publishing metrics by writing to the lock key
 		let acquired_lock = self
 			.pools

packages/common/chirp-workflow/core/src/metrics.rs

@@ -184,4 +184,31 @@ lazy_static::lazy_static! {
 		BUCKETS.to_vec(),
 		*REGISTRY,
 	).unwrap();
+
+	// MARK: Ops
+	pub static ref OPERATION_PENDING: IntGaugeVec = register_int_gauge_vec_with_registry!(
+		"chirp_workflow_operation_pending",
+		"Total number of operation calls in progress.",
+		&["operation_name"],
+		*REGISTRY,
+	).unwrap();
+	pub static ref OPERATION_TOTAL: IntCounterVec = register_int_counter_vec_with_registry!(
+		"chirp_workflow_operation_total",
+		"Total number of operation calls.",
+		&["operation_name"],
+		*REGISTRY,
+	).unwrap();
+	pub static ref OPERATION_DURATION: HistogramVec = register_histogram_vec_with_registry!(
+		"chirp_workflow_operation_duration",
+		"Total duration of an op call.",
+		&["operation_name", "error_code"],
+		BUCKETS.to_vec(),
+		*REGISTRY,
+	).unwrap();
+	pub static ref OPERATION_ERRORS: IntCounterVec = register_int_counter_vec_with_registry!(
+		"chirp_workflow_operation_errors",
+		"All errors made by this operation.",
+		&["operation_name", "error_code", "error_type"],
+		*REGISTRY,
+	).unwrap();
 }

packages/core/api/status/Cargo.toml

@@ -41,7 +41,7 @@ game-resolve-name-id.workspace = true
 game-namespace-resolve-name-id.workspace = true
 
 token-create.workspace = true
-trust-dns-resolver = "0.23.2"
+hickory-resolver = "0.25.1"
 rivet-config.workspace = true
 rivet-env.workspace = true
 

packages/core/api/status/src/route/actor.rs

@@ -240,6 +240,7 @@ pub async fn status(
 		Some(&system_test_env),
 		None,
 	)
+	.instrument(tracing::info_span!("actor destroy request"))
 	.await?;
 
 	// Unwrap res
@@ -259,6 +260,7 @@ pub async fn status(
 	Ok(serde_json::json!({}))
 }
 
+#[tracing::instrument]
 async fn test_actor_connection(hostname: &str, port: u16, actor_origin: &str) -> GlobalResult<()> {
 	// Look up IP for the actor's host
 	let gg_ips = lookup_dns(hostname).await?;
@@ -312,8 +314,9 @@ async fn test_actor_connection(hostname: &str, port: u16, actor_origin: &str) ->
 }
 
 /// Returns the IP addresses for a given hostname.
+#[tracing::instrument]
 async fn lookup_dns(hostname: &str) -> GlobalResult<Vec<IpAddr>> {
-	let resolver = trust_dns_resolver::TokioAsyncResolver::tokio_from_system_conf()?;
+	let resolver = hickory_resolver::Resolver::builder_tokio()?.build();
 	let addrs = resolver
 		.lookup_ip(hostname)
 		.await?
@@ -326,6 +329,7 @@ async fn lookup_dns(hostname: &str) -> GlobalResult<Vec<IpAddr>> {
 /// Tests HTTP connectivity to a hostname for a given address.
 ///
 /// This lets us isolate of a specific GG IP address is not behaving correctly.
+#[tracing::instrument]
 async fn test_http(
 	actor_origin: String,
 	hostname: String,
@@ -340,13 +344,15 @@ async fn test_http(
 	client
 		.get(format!("{actor_origin}/health"))
 		.send()
+		.instrument(tracing::info_span!("health request"))
 		.await?
 		.error_for_status()?;
 
 	Ok(())
 }
 
 /// Tests WebSocket connectivity to a hostname.
+#[tracing::instrument]
 async fn test_ws(actor_origin: &str) -> GlobalResult<()> {
 	let actor_origin = actor_origin
 		.replace("http://", "ws://")

packages/core/api/status/src/route/matchmaker.rs

@@ -246,7 +246,7 @@ async fn test_lobby_connection(
 
 /// Returns the IP addresses for a given hostname.
 async fn lookup_dns(hostname: &str) -> GlobalResult<Vec<IpAddr>> {
-	let resolver = trust_dns_resolver::TokioAsyncResolver::tokio_from_system_conf()?;
+	let resolver = hickory_resolver::Resolver::builder_tokio()?.build();
 	let addrs = resolver
 		.lookup_ip(hostname)
 		.await?

packages/core/services/cluster/src/workflows/server/install/install_scripts/files/otel_collector.sh

@@ -39,7 +39,20 @@ processors:
         status_code:
           status_codes:
             - ERROR
-
+      # Guard-only policy
+      - name: policy-2
+        type: and
+        and:
+          and_sub_policy:
+            - name: latency-policy-1
+              type: latency
+              latency:
+                threshold_ms: 15000
+            - name: span-name-policy-1
+              type: ottl_condition
+              ottl_condition:
+                span:
+                  - 'name == "routing_fn"'
 exporters:
   otlp:
     endpoint: 127.0.0.1:__TUNNEL_OTEL_PORT__

packages/core/services/cluster/src/workflows/server/install/install_scripts/files/rivet_guard_configure.sh

@@ -17,7 +17,7 @@ ConditionPathExists=/etc/rivet-server/
 [Service]
 Environment="RIVET_OTEL_ENABLED=1"
 Environment="RIVET_OTEL_SAMPLER_RATIO=1"
-Environment="RIVET_SERVICE_NAME=rivet-guard"
+Environment="RIVET_SERVICE_NAME=guard"
 Environment="RIVET_CLUSTER_ID=___CLUSTER_ID___"
 Environment="RIVET_DATACENTER_ID=___DATACENTER_ID___"
 Environment="RIVET_SERVER_ID=___SERVER_ID___"

packages/core/services/cluster/src/workflows/server/install/install_scripts/files/rivet_worker_configure.sh

@@ -18,7 +18,7 @@ ConditionPathExists=/etc/rivet-server/
 # OTeL env vars
 Environment="RIVET_OTEL_ENABLED=1"
 Environment="RIVET_OTEL_SAMPLER_RATIO=1"
-Environment="RIVET_SERVICE_NAME=rivet-edge"
+Environment="RIVET_SERVICE_NAME=edge"
 Environment="RIVET_CLUSTER_ID=___CLUSTER_ID___"
 Environment="RIVET_DATACENTER_ID=___DATACENTER_ID___"
 Environment="RIVET_SERVER_ID=___SERVER_ID___"