Skip to content

Commit d01c822

Browse files
barrygreengusbarrygreengus
committed
refactor(pylon): use tracing macros for registration logs
Emit registration and discovery warnings at the failure sites. Remove logging dispatch wrappers and the custom connection-log macro. Keep safe endpoint and error formatting separate from event emission. Preserve certificate ERROR diagnostics and their existing suppression. Log each failed retry at WARN and report closed streams directly. Relates to #1817
1 parent 664865e commit d01c822

4 files changed

Lines changed: 178 additions & 215 deletions

File tree

‎src/libraries/rust/stargate/crates/pylon-lib/src/registration/discovery.rs‎

Lines changed: 38 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ use stargate_runtime::{OwnedTask, TASK_SHUTDOWN_TIMEOUT};
2828
use tracing::warn;
2929

3030
use super::grpc_endpoint::{
31-
StargateGrpcEndpoint, log_stargate_grpc_connect_attempt, log_stargate_grpc_failure,
31+
StargateGrpcEndpoint, grpc_error_chain, log_stargate_grpc_certificate_failure,
3232
};
3333
use super::topology::{RegistrationRouterTopology, publish_registration_router_topology};
3434

@@ -165,7 +165,13 @@ async fn watch_stargate_endpoint(
165165
return;
166166
}
167167

168-
log_stargate_grpc_connect_attempt(&target, "watch_stargates", "lazy");
168+
tracing::debug!(
169+
transport = "grpc",
170+
operation = "watch_stargates",
171+
endpoint = %target,
172+
connect_mode = "lazy",
173+
"attempting Stargate gRPC connection"
174+
);
169175
let stream = match target.channel_endpoint(grpc_tls_ca_cert_pem.as_deref()) {
170176
Ok(endpoint) => {
171177
let mut client = StargateControlPlaneClient::new(endpoint.connect_lazy());
@@ -176,7 +182,16 @@ async fn watch_stargate_endpoint(
176182
Some(response.into_inner())
177183
}
178184
Err(error) => {
179-
log_stargate_grpc_failure(&target, "watch_stargates", &error, &mut last_certificate_failure);
185+
last_certificate_failure = log_stargate_grpc_certificate_failure(
186+
&target, "watch_stargates", &error, last_certificate_failure,
187+
);
188+
warn!(
189+
transport = "grpc",
190+
operation = "watch_stargates",
191+
endpoint = %target,
192+
error = %grpc_error_chain(&error),
193+
"Stargate gRPC operation failed"
194+
);
180195
None
181196
}
182197
}
@@ -185,11 +200,18 @@ async fn watch_stargate_endpoint(
185200
}
186201
}
187202
Err(error) => {
188-
log_stargate_grpc_failure(
203+
last_certificate_failure = log_stargate_grpc_certificate_failure(
189204
&target,
190205
"watch_stargates",
191206
error.as_ref(),
192-
&mut last_certificate_failure,
207+
last_certificate_failure,
208+
);
209+
warn!(
210+
transport = "grpc",
211+
operation = "watch_stargates",
212+
endpoint = %target,
213+
error = %grpc_error_chain(error.as_ref()),
214+
"Stargate gRPC operation failed"
193215
);
194216
None
195217
}
@@ -205,23 +227,21 @@ async fn watch_stargate_endpoint(
205227
Some(watch_endpoint_snapshot_from_response(&watch_url, response))
206228
}
207229
Ok(None) => {
208-
log_stargate_grpc_failure(
209-
&target,
210-
"watch_stargates_stream",
211-
&std::io::Error::new(
212-
std::io::ErrorKind::UnexpectedEof,
213-
"discovery response stream ended",
214-
),
215-
&mut last_certificate_failure,
230+
warn!(
231+
transport = "grpc",
232+
operation = "watch_stargates_stream",
233+
endpoint = %target,
234+
"Stargate discovery response stream ended"
216235
);
217236
None
218237
}
219238
Err(error) => {
220-
log_stargate_grpc_failure(
221-
&target,
222-
"watch_stargates_stream",
223-
&error,
224-
&mut last_certificate_failure,
239+
warn!(
240+
transport = "grpc",
241+
operation = "watch_stargates_stream",
242+
endpoint = %target,
243+
error = %grpc_error_chain(&error),
244+
"Stargate gRPC operation failed"
225245
);
226246
None
227247
}

‎src/libraries/rust/stargate/crates/pylon-lib/src/registration/grpc_endpoint.rs‎

Lines changed: 26 additions & 94 deletions
Original file line numberDiff line numberDiff line change
@@ -124,10 +124,19 @@ pub(super) fn grpc_origin_uri(
124124

125125
impl fmt::Display for StargateGrpcEndpoint {
126126
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
127+
let safe_endpoint = |endpoint: &str| match stargate_grpc_debug_target(endpoint) {
128+
Ok(target) => format!("{}://{}:{}", target.scheme, target.host, target.port),
129+
Err(_) => "<invalid endpoint>".to_string(),
130+
};
131+
let authority = safe_endpoint(&self.authority_endpoint());
127132
if self.uses_authority_override() {
128-
write!(f, "{} via {}", self.authority_addr, self.dial_addr)
133+
write!(
134+
f,
135+
"{authority} via {}",
136+
safe_endpoint(&self.dial_endpoint())
137+
)
129138
} else {
130-
write!(f, "{}", self.authority_addr)
139+
f.write_str(&authority)
131140
}
132141
}
133142
}
@@ -164,12 +173,23 @@ pub(super) async fn connect_stargate_grpc_channel(
164173
grpc_tls_ca_cert_pem: Option<&[u8]>,
165174
operation: &'static str,
166175
) -> anyhow::Result<Channel> {
167-
log_stargate_grpc_connect_attempt(router_endpoint, operation, "eager");
176+
tracing::debug!(
177+
transport = "grpc",
178+
operation,
179+
endpoint = %router_endpoint,
180+
connect_mode = "eager",
181+
"attempting Stargate gRPC connection"
182+
);
168183
let channel = router_endpoint
169184
.channel_endpoint(grpc_tls_ca_cert_pem)?
170185
.connect()
171186
.await?;
172-
log_stargate_grpc_channel_connected(router_endpoint, operation);
187+
tracing::debug!(
188+
transport = "grpc",
189+
operation,
190+
endpoint = %router_endpoint,
191+
"Stargate gRPC channel connected"
192+
);
173193
Ok(channel)
174194
}
175195

@@ -287,69 +307,9 @@ pub(super) fn log_stargate_grpc_certificate_failure(
287307
Some(failure)
288308
}
289309

290-
macro_rules! log_stargate_grpc_target {
291-
($level:ident, $target:expr, $operation:expr, [$($extra:tt)*], $message:literal, $error_message:literal) => {{
292-
if !tracing::enabled!(tracing::Level::$level) {
293-
return;
294-
}
295-
let dial_endpoint = $target.dial_endpoint();
296-
let authority_endpoint = $target.authority_endpoint();
297-
let override_authority = dial_endpoint != authority_endpoint;
298-
match (
299-
stargate_grpc_debug_target(&dial_endpoint),
300-
stargate_grpc_debug_target(&authority_endpoint),
301-
) {
302-
(Ok(dial), Ok(authority)) => tracing::event!(tracing::Level::$level,
303-
transport = "grpc",
304-
operation = $operation,
305-
http_version = "h2",
306-
dial_scheme = %dial.scheme,
307-
tls = dial.scheme == "https",
308-
dial_host = %dial.host,
309-
dial_port = dial.port,
310-
authority_host = %authority.host,
311-
authority_port = authority.port,
312-
override_authority,
313-
$($extra)*
314-
$message
315-
),
316-
(Err(_), _) | (_, Err(_)) => tracing::event!(tracing::Level::$level,
317-
transport = "grpc",
318-
operation = $operation,
319-
override_authority,
320-
$($extra)*
321-
$error_message
322-
),
323-
}
324-
}};
325-
}
326-
327-
pub(super) fn log_stargate_grpc_failure(
328-
target: &StargateGrpcEndpoint,
329-
operation: &'static str,
330-
error: &(dyn Error + 'static),
331-
last_certificate_failure: &mut Option<StargateGrpcCertificateFailure>,
332-
) {
333-
if classify_stargate_grpc_certificate_failure(error).is_some() {
334-
*last_certificate_failure = log_stargate_grpc_certificate_failure(
335-
target,
336-
operation,
337-
error,
338-
*last_certificate_failure,
339-
);
340-
return;
341-
}
342-
log_stargate_grpc_target!(
343-
WARN, target, operation,
344-
[error = %grpc_error_chain(error),],
345-
"Stargate gRPC operation failed",
346-
"Stargate gRPC operation failed"
347-
);
348-
}
349-
350-
fn grpc_error_chain(mut error: &(dyn Error + 'static)) -> String {
310+
pub(super) fn grpc_error_chain(error: &(dyn Error + 'static)) -> String {
351311
let mut causes = Vec::new();
352-
loop {
312+
for error in anyhow::Chain::new(error) {
353313
// Parser diagnostics can include input excerpts from token files.
354314
let detail = if let Some(error) = error.downcast_ref::<sonic_rs::Error>() {
355315
format!(
@@ -376,38 +336,10 @@ fn grpc_error_chain(mut error: &(dyn Error + 'static)) -> String {
376336
if causes.last() != Some(&detail) {
377337
causes.push(detail);
378338
}
379-
let Some(source) = error.source() else { break };
380-
error = source;
381339
}
382340
causes.join(": ")
383341
}
384342

385-
pub(super) fn log_stargate_grpc_connect_attempt(
386-
target: &StargateGrpcEndpoint,
387-
operation: &'static str,
388-
connect_mode: &'static str,
389-
) {
390-
log_stargate_grpc_target!(
391-
DEBUG,
392-
target,
393-
operation,
394-
[connect_mode,],
395-
"attempting Stargate gRPC connection",
396-
"could not parse Stargate gRPC endpoint for connection debug logging"
397-
);
398-
}
399-
400-
fn log_stargate_grpc_channel_connected(target: &StargateGrpcEndpoint, operation: &'static str) {
401-
log_stargate_grpc_target!(
402-
DEBUG,
403-
target,
404-
operation,
405-
[],
406-
"Stargate gRPC channel connected",
407-
"Stargate gRPC channel connected but endpoint metadata could not be parsed"
408-
);
409-
}
410-
411343
fn normalize_addr_with_default_scheme(addr: &str, default_scheme: &str) -> String {
412344
if addr.starts_with("http://") || addr.starts_with("https://") {
413345
addr.to_string()

‎src/libraries/rust/stargate/crates/pylon-lib/src/registration/router_stream.rs‎

Lines changed: 27 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,8 @@ use stargate_proto::pb::{InferenceServerAck, InferenceServerRegistration, Infere
2929
use stargate_runtime::{OwnedTask, TASK_SHUTDOWN_TIMEOUT};
3030

3131
use super::grpc_endpoint::{
32-
StargateGrpcEndpoint, connect_stargate_grpc_channel, log_stargate_grpc_failure,
32+
StargateGrpcEndpoint, connect_stargate_grpc_channel, grpc_error_chain,
33+
log_stargate_grpc_certificate_failure,
3334
};
3435
use super::reverse_tunnel::{
3536
ReverseTunnelState, reverse_tunnel_endpoint_from_ack, run_reverse_tunnel_loop,
@@ -60,11 +61,19 @@ pub(super) async fn run_router_registration_stream(
6061
let (mut ack_stream, update_tx) = match connection {
6162
Ok(connection) => connection,
6263
Err(error) => {
63-
log_stargate_grpc_failure(
64+
last_certificate_failure = log_stargate_grpc_certificate_failure(
6465
&router_endpoint,
6566
"register_inference_server",
6667
error.as_ref(),
67-
&mut last_certificate_failure,
68+
last_certificate_failure,
69+
);
70+
tracing::warn!(
71+
transport = "grpc",
72+
operation = "register_inference_server",
73+
endpoint = %router_endpoint,
74+
cluster_id = %config.cluster_id,
75+
error = %grpc_error_chain(error.as_ref()),
76+
"Stargate gRPC operation failed"
6877
);
6978
if stop
7079
.run_until_cancelled(tokio::time::sleep(Duration::from_secs(1)))
@@ -163,13 +172,24 @@ pub(super) async fn run_router_registration_stream(
163172
ack
164173
}
165174
Ok(None) => {
166-
log_stargate_grpc_failure(&router_endpoint, "register_inference_server_stream",
167-
&std::io::Error::new(std::io::ErrorKind::UnexpectedEof, "registration response stream ended"),
168-
&mut last_certificate_failure);
175+
tracing::warn!(
176+
transport = "grpc",
177+
operation = "register_inference_server_stream",
178+
endpoint = %router_endpoint,
179+
cluster_id = %config.cluster_id,
180+
"Stargate registration response stream ended"
181+
);
169182
break false;
170183
}
171184
Err(error) => {
172-
log_stargate_grpc_failure(&router_endpoint, "register_inference_server_stream", &error, &mut last_certificate_failure);
185+
tracing::warn!(
186+
transport = "grpc",
187+
operation = "register_inference_server_stream",
188+
endpoint = %router_endpoint,
189+
cluster_id = %config.cluster_id,
190+
error = %grpc_error_chain(&error),
191+
"Stargate gRPC operation failed"
192+
);
173193
break false;
174194
}
175195
};

0 commit comments

Comments
 (0)