Repository: cloudflare/foundations (language: rust)

Issue:
Don't halt gRPC reporting on errors



Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index ed9e046..6e83abb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -50,7 +50,7 @@ prometools = "0.2.1"
 rand = "0.8"
 quote = "1"
 regex = "1.8"
-reqwest = { version = "0.11", default-features = false }
+reqwest = { version = "0.12", default-features = false }
 routerify = "3"
 socket2 = { version = "0.5.3", features = [ "all" ] }
 syn = "1"
@@ -59,15 +59,15 @@ serde_path_to_error = "0.1.15"
 serde_yaml = "0.8.26"
 serde_with = "3.3.0"
 slab = "0.4.9"
-slog = "2.4"
+slog = "2.7"
 slog-async = "2.3"
 slog-json = "2.3"
 slog-term = "2.4"
 tempfile = "3.7"
 tokio = "1.37.0"
 thread_local = "1.1"
-tikv-jemallocator = "0.5"
-tikv-jemalloc-ctl = "0.5"
+tikv-jemallocator = "0.6.0"
+tikv-jemalloc-ctl = "0.6.0"
 yaml-merge-keys = "0.5"
 
 # needed for minver
diff --git a/foundations/src/telemetry/tracing/internal.rs b/foundations/src/telemetry/tracing/internal.rs
index d2c8d64..113fde0 100644
--- a/foundations/src/telemetry/tracing/internal.rs
+++ b/foundations/src/telemetry/tracing/internal.rs
@@ -6,6 +6,7 @@ use crate::telemetry::tracing::rate_limit::RateLimitingProbabilisticSampler;
 use cf_rustracing::tag::Tag;
 use cf_rustracing_jaeger::span::{Span, SpanContext, SpanContextState};
 use std::borrow::Cow;
+use std::error::Error;
 use std::sync::Arc;
 
 pub(crate) type Tracer = cf_rustracing::Tracer<RateLimitingProbabilisticSampler, SpanContextState>;
@@ -90,6 +91,14 @@ pub(crate) fn start_trace(
     new_trace_root_span
 }
 
+pub(super) fn reporter_error(err: impl Error) {
+    #[cfg(feature = "logging")]
+    crate::telemetry::log::error!("failed to report traces to the traces sink"; "error" => %err);
+
+    #[cfg(not(feature = "logging"))]
+    drop(err);
+}
+
 // Link a newly created trace in the current span's ref span and vice-versa
 fn link_new_trace_with_current(
     current_span: &mut SharedSpan,
diff --git a/foundations/src/telemetry/tracing/output_jaeger_thrift_udp.rs b/foundations/src/telemetry/tracing/output_jaeger_thrift_udp.rs
index 8655dcf..7dfd8c0 100644
--- a/foundations/src/telemetry/tracing/output_jaeger_thrift_udp.rs
+++ b/foundations/src/telemetry/tracing/output_jaeger_thrift_udp.rs
@@ -1,3 +1,4 @@
+use super::internal::reporter_error;
 use crate::telemetry::settings::JaegerThriftUdpOutputSettings;
 use crate::{BootstrapResult, ServiceInfo};
 use anyhow::bail;
@@ -9,9 +10,6 @@ use std::net::SocketAddr;
 use std::net::{Ipv4Addr, Ipv6Addr};
 use std::sync::Arc;
 
-#[cfg(feature = "logging")]
-use crate::telemetry::log;
-
 pub(super) fn start(
     service_info: ServiceInfo,
     settings: &JaegerThriftUdpOutputSettings,
@@ -69,12 +67,8 @@ async fn do_export(reporter: JaegerCompactReporter, mut span_rx: SpanReceiver) {
             let reporter = Arc::clone(&reporter);
 
             async move {
-                if let Err(e) = reporter.report(&[span][..]).await {
-                    #[cfg(feature = "logging")]
-                    log::error!("failed to send a tracing span to the agent"; "error" => %e);
-
-                    #[cfg(not(feature = "logging"))]
-                    drop(e);
+                if let Err(err) = reporter.report(&[span][..]).await {
+                    reporter_error(err);
                 }
             }
         });
diff --git a/foundations/src/telemetry/tracing/output_otlp_grpc.rs b/foundations/src/telemetry/tracing/output_otlp_grpc.rs
index 70fc152..db5ce65 100644
--- a/foundations/src/telemetry/tracing/output_otlp_grpc.rs
+++ b/foundations/src/telemetry/tracing/output_otlp_grpc.rs
@@ -1,3 +1,4 @@
+use super::internal::reporter_error;
 use crate::telemetry::otlp_conversion::tracing::convert_span;
 use crate::telemetry::settings::OpenTelemetryGrpcOutputSettings;
 use crate::{BootstrapResult, ServiceInfo};
@@ -37,7 +38,9 @@ pub(super) fn start(
 
         let client = Grpc::new(grpc_channel);
 
-        do_export(client, service_info, span_rx, max_batch_size).await
+        do_export(client, service_info, span_rx, max_batch_size).await;
+
+        Ok(())
     }
     .boxed())
 }
@@ -47,7 +50,7 @@ async fn do_export(
     service_info: ServiceInfo,
     mut span_rx: SpanReceiver,
     max_batch_size: usize,
-) -> BootstrapResult<()> {
+) {
     let mut batch = Vec::with_capacity(max_batch_size);
 
     loop {
@@ -62,22 +65,23 @@ async fn do_export(
             .map(|span| convert_span(span, &service_info))
             .collect();
 
-        client
-            .ready()
-            .await
-            .context("lost gRPC connection to the trace service")?;
+        if let Err(err) = client.ready().await {
+            reporter_error(err);
+            continue;
+        }
 
-        client
+        let send_res = client
             .unary::<_, ExportTraceServiceResponse, _>(
                 create_request(resource_spans),
                 PathAndQuery::from_static(COLLECTOR_PATH),
                 ProstCodec::default(),
             )
-            .await
-            .context("failed to send gRPC request to the trace service")?;
-    }
+            .await;
 
-    Ok(())
+        if let Err(err) = send_res {
+            reporter_error(err);
+        }
+    }
 }
 
 fn create_request(resource_spans: Vec<ResourceSpans>) -> Request<ExportTraceServiceRequest> {


Test patch: