fix(observability): do not hold span guards across await points (#25482)

* add clippy rule

* remove span.enter from topology builders

* fix internal_logs source unit test holding a span guard across an await. The test still works because tokio:: test runs single-threaded by default, but better fixing it than silencing the clippy error

* add changelog

* apply suggestion

Author: gwenaskell

changelog.d/fix_topology_build_spans.fix.md

@@ -0,0 +1,7 @@
+Fixed a bug in the topology builder causing component metrics registered at build
+time to miss the component tags if the component build function awaits non-trivially.
+
+This notably affected sinks using a disk buffer, and source or sinks performing
+IO work in the build function.
+
+authors: gwenaskell

clippy.toml

@@ -20,3 +20,8 @@ disallowed-types = [
   { path = "once_cell::sync::Lazy", reason = "Use `std::sync::LazyLock` instead." },
   { path = "once_cell::unsync::Lazy", reason = "Use `std::sync::LazyCell` instead." },
 ]
+
+
+await-holding-invalid-types = [
+    { path = "tracing::span::Entered", reason = "Holding a tracing span guard across an `.await` is thread-local and silently breaks when the task resumes on a different thread. Use `#[tracing::instrument]`, `future.instrument(span)`, or `Span::in_scope(|| ...)` instead." },
+]

src/sources/internal_logs.rs

@@ -256,12 +256,16 @@ mod tests {
             component_id = "foo",
             component_type = "internal_logs",
         );
-        let _enter = span.enter();
+        let enter = span.enter();
 
         error!(message = "Before source started.", %test_id);
 
+        drop(enter); // don't hold the span guard across an await point
+
         let rx = start_source().await;
 
+        let enter = span.enter();
+
         error!(message = "After source started.", %test_id);
 
         {
@@ -276,6 +280,8 @@ mod tests {
             error!(message = "In a nested span.", %test_id);
         }
 
+        drop(enter);
+
         sleep(Duration::from_millis(1)).await;
         let mut events = collect_ready(rx).await;
         let test_id = Value::from(test_id.to_string());