fix: read epoxy config from local replica id

Author: NathanFlurry

packages/common/metrics/src/providers.rs

@@ -1,9 +1,8 @@
 // Based off of https://github.com/tokio-rs/tracing-opentelemetry/blob/v0.1.x/examples/opentelemetry-otlp.rs
 // Based off of https://github.com/tokio-rs/tracing-opentelemetry/blob/v0.1.x/examples/opentelemetry-otlp.rs
 
-use std::sync::{Arc, RwLock, OnceLock};
-use opentelemetry::{KeyValue, global};
 use opentelemetry::trace::{SamplingResult, SpanKind};
+use opentelemetry::{KeyValue, global};
 use opentelemetry_otlp::WithExportConfig;
 use opentelemetry_sdk::{
 	Resource,
@@ -12,6 +11,7 @@ use opentelemetry_sdk::{
 	trace::{RandomIdGenerator, Sampler, SdkTracerProvider},
 };
 use opentelemetry_semantic_conventions::{SCHEMA_URL, attribute::SERVICE_VERSION};
+use std::sync::{Arc, OnceLock, RwLock};
 
 /// Dynamic sampler that can be updated at runtime
 #[derive(Clone, Debug)]

packages/common/runtime/src/traces.rs

@@ -26,8 +26,8 @@ pub fn init_tracing_subscriber(otel_providers: &Option<OtelProviderGuard>) {
 		Some(providers) => {
 			let tracer = providers.tracer_provider.tracer("tracing-otel-subscriber");
 
-			let otel_trace_layer =
-				OpenTelemetryLayer::new(tracer).with_filter(build_filter_from_env_var("RUST_TRACE"));
+			let otel_trace_layer = OpenTelemetryLayer::new(tracer)
+				.with_filter(build_filter_from_env_var("RUST_TRACE"));
 
 			let otel_metric_layer = MetricsLayer::new(providers.meter_provider.clone())
 				.with_filter(build_filter_from_env_var("RUST_TRACE"));

packages/core/api-peer/src/internal.rs

@@ -47,6 +47,7 @@ pub async fn bump_serverless_autoscaler(
 }
 
 #[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct SetTracingConfigRequest {
 	#[serde(default, skip_serializing_if = "Option::is_none")]
 	pub filter: Option<Option<String>>,
@@ -55,6 +56,7 @@ pub struct SetTracingConfigRequest {
 }
 
 #[derive(Serialize)]
+#[serde(deny_unknown_fields)]
 pub struct SetTracingConfigResponse {}
 
 #[tracing::instrument(skip_all)]

packages/core/api-public/src/actors/get_or_create.rs

@@ -4,8 +4,8 @@ use axum::{
 	response::{IntoResponse, Response},
 };
 use rivet_api_builder::{
-	extract::{Extension, Json, Query},
 	ApiError,
+	extract::{Extension, Json, Query},
 };
 use rivet_types::actors::CrashPolicy;
 use rivet_util::Id;

packages/core/api-public/src/actors/utils.rs

@@ -10,11 +10,7 @@ use std::collections::HashMap;
 /// Helper function to fetch an actor by ID, automatically routing to the correct datacenter
 /// based on the actor ID's label.
 #[tracing::instrument(skip_all)]
-pub async fn fetch_actor_by_id(
-	ctx: &ApiCtx,
-	actor_id: Id,
-	namespace: String,
-) -> Result<Actor> {
+pub async fn fetch_actor_by_id(ctx: &ApiCtx, actor_id: Id, namespace: String) -> Result<Actor> {
 	let list_query = rivet_api_types::actors::list::ListQuery {
 		namespace,
 		actor_ids: Some(actor_id.to_string()),

packages/core/api-public/src/namespaces.rs

@@ -21,10 +21,7 @@ use crate::ctx::ApiCtx;
 	security(("bearer_auth" = [])),
 )]
 #[tracing::instrument(skip_all)]
-pub async fn list(
-	Extension(ctx): Extension<ApiCtx>,
-	Query(query): Query<ListQuery>,
-) -> Response {
+pub async fn list(Extension(ctx): Extension<ApiCtx>, Query(query): Query<ListQuery>) -> Response {
 	match list_inner(ctx, query).await {
 		Ok(response) => Json(response).into_response(),
 		Err(err) => ApiError::from(err).into_response(),

packages/core/api-public/src/runners.rs

@@ -22,10 +22,7 @@ use crate::ctx::ApiCtx;
 	security(("bearer_auth" = [])),
 )]
 #[tracing::instrument(skip_all)]
-pub async fn list(
-	Extension(ctx): Extension<ApiCtx>,
-	Query(query): Query<ListQuery>,
-) -> Response {
+pub async fn list(Extension(ctx): Extension<ApiCtx>, Query(query): Query<ListQuery>) -> Response {
 	match list_inner(ctx, query).await {
 		Ok(response) => Json(response).into_response(),
 		Err(err) => ApiError::from(err).into_response(),

packages/infra/engine/src/run_config.rs

@@ -22,9 +22,11 @@ pub fn config(_rivet_config: rivet_config::Config) -> Result<RunConfigData> {
 			ServiceKind::Standalone,
 			|config, pools| Box::pin(pegboard_serverless::start(config, pools)),
 		),
-		Service::new("tracing_reconfigure", ServiceKind::Standalone, |config, pools| {
-			Box::pin(rivet_tracing_reconfigure::start(config, pools))
-		}),
+		Service::new(
+			"tracing_reconfigure",
+			ServiceKind::Standalone,
+			|config, pools| Box::pin(rivet_tracing_reconfigure::start(config, pools)),
+		),
 	];
 
 	Ok(RunConfigData { services })

packages/services/epoxy/src/http_client.rs

@@ -5,6 +5,7 @@ use epoxy_protocol::{
 	versioned,
 };
 use futures_util::{StreamExt, stream::FuturesUnordered};
+use gas::prelude::*;
 use rivet_api_builder::ApiCtx;
 use std::future::Future;
 use vbare::OwnedVersionedData;
@@ -24,6 +25,7 @@ fn find_replica_address(
 		.map(|r| r.api_peer_url.clone())
 }
 
+#[tracing::instrument(skip_all, fields(%from_replica_id, ?replica_ids, ?quorum_type))]
 pub async fn fanout_to_replicas<F, Fut, T>(
 	from_replica_id: ReplicaId,
 	replica_ids: &[ReplicaId],
@@ -93,6 +95,7 @@ where
 	Ok(successful_responses)
 }
 
+#[tracing::instrument(skip_all)]
 pub async fn send_message(
 	ctx: &ApiCtx,
 	config: &protocol::ClusterConfig,
@@ -102,6 +105,7 @@ pub async fn send_message(
 	send_message_to_address(ctx, replica_url, request).await
 }
 
+#[tracing::instrument(skip_all, fields(%replica_url))]
 pub async fn send_message_to_address(
 	ctx: &ApiCtx,
 	replica_url: String,
@@ -116,8 +120,7 @@ pub async fn send_message_to_address(
 			"sending message to replica directly"
 		);
 
-		return crate::replica::message_request::message_request(&ctx, from_replica_id, request)
-			.await;
+		return crate::replica::message_request::message_request(&ctx, request).await;
 	}
 
 	let mut replica_url = url::Url::parse(&replica_url)?;
@@ -139,6 +142,7 @@ pub async fn send_message_to_address(
 		.post(replica_url.to_string())
 		.body(request.serialize()?)
 		.send()
+		.custom_instrument(tracing::info_span!("http_request"))
 		.await;
 
 	let response = match response_result {

packages/services/epoxy/src/http_routes.rs

@@ -28,8 +28,7 @@ pub async fn message(ctx: ApiCtx, path: VersionedPath, _query: (), body: Bytes)
 	);
 
 	// Process message directly using ops
-	let response =
-		crate::replica::message_request::message_request(&ctx, current_replica_id, request).await?;
+	let response = crate::replica::message_request::message_request(&ctx, request).await?;
 
 	versioned::Response::latest(response).serialize(path.version)
 }