Only set trace subscriber when useful (#602)

Author: Sushisource

Cargo.toml

@@ -1,5 +1,6 @@
 [workspace]
 members = ["core", "client", "core-api", "fsm", "test-utils", "sdk-core-protos", "sdk"]
+resolver = "2"
 
 [workspace.dependencies]
 tonic = "0.9"

core-api/src/telemetry/metrics.rs

@@ -136,7 +136,7 @@ impl MetricAttributes {
                 Arc::make_mut(kvs).extend(new_kvs.into_iter().map(Into::into));
             }
             MetricAttributes::Lang(ref mut attrs, ..) => {
-                attrs.new_attributes.extend(new_kvs.into_iter());
+                attrs.new_attributes.extend(new_kvs);
             }
         }
     }
@@ -149,7 +149,7 @@ pub struct MetricsAttributesOptions {
 }
 impl MetricsAttributesOptions {
     pub fn extend(&mut self, new_kvs: impl IntoIterator<Item = MetricKeyValue>) {
-        self.attributes.extend(new_kvs.into_iter())
+        self.attributes.extend(new_kvs)
     }
 }
 

core/Cargo.toml

@@ -43,7 +43,7 @@ itertools = "0.11"
 lazy_static = "1.4"
 lru = "0.11"
 mockall = "0.11"
-nix = { version = "0.26", optional = true }
+nix = { version = "0.27", optional = true, features = ["process", "signal"] }
 once_cell = "1.5"
 opentelemetry = { workspace = true, features = ["rt-tokio", "metrics"] }
 opentelemetry_sdk = { version = "0.20", features = ["metrics"] }
@@ -60,7 +60,7 @@ ringbuf = "0.3"
 rmp-serde = { version = "1.1", optional = true }
 serde = "1.0"
 serde_json = "1.0"
-siphasher = "0.3"
+siphasher = "1.0"
 slotmap = "1.0"
 tar = { version = "0.4", optional = true }
 thiserror = "1.0"

core/src/internal_flags.rs

@@ -75,7 +75,7 @@ impl InternalFlags {
             ..
         } = self
         {
-            lang_since_last_complete.extend(flags.into_iter());
+            lang_since_last_complete.extend(flags);
         }
     }
 

core/src/lib.rs

@@ -223,7 +223,9 @@ impl CoreRuntime {
         let runtime = tokio_builder
             .enable_all()
             .on_thread_start(move || {
-                set_trace_subscriber_for_current_thread(subscriber.clone());
+                if let Some(sub) = subscriber.as_ref() {
+                    set_trace_subscriber_for_current_thread(sub.clone());
+                }
             })
             .build()?;
         let _rg = runtime.enter();
@@ -249,7 +251,9 @@ impl CoreRuntime {
     /// If there is no currently active Tokio runtime
     pub fn new_assume_tokio_initialized_telem(telemetry: TelemetryInstance) -> Self {
         let runtime_handle = tokio::runtime::Handle::current();
-        set_trace_subscriber_for_current_thread(telemetry.trace_subscriber());
+        if let Some(sub) = telemetry.trace_subscriber() {
+            set_trace_subscriber_for_current_thread(sub);
+        }
         Self {
             telemetry,
             runtime: None,

core/src/telemetry/log_export.rs

@@ -166,7 +166,7 @@ mod tests {
             .build()
             .unwrap();
         let instance = telemetry_init(opts).unwrap();
-        let _g = tracing::subscriber::set_default(instance.trace_subscriber.clone());
+        let _g = tracing::subscriber::set_default(instance.trace_subscriber().unwrap().clone());
 
         let top_span = span!(Level::INFO, "yayspan", huh = "wat");
         let _guard = top_span.enter();

core/src/telemetry/metrics.rs

@@ -813,7 +813,7 @@ mod tests {
         let no_op_subscriber = Arc::new(NoSubscriber::new());
         let call_buffer = Arc::new(MetricsCallBuffer::new(100));
         let telem_instance = TelemetryInstance::new(
-            no_op_subscriber,
+            Some(no_op_subscriber),
             None,
             METRIC_PREFIX.to_string(),
             Some(call_buffer.clone()),

core/src/telemetry/mod.rs

@@ -50,13 +50,15 @@ pub struct TelemetryInstance {
     metric_prefix: String,
     logs_out: Option<Mutex<CoreLogsOut>>,
     metrics: Option<Arc<dyn CoreMeter + 'static>>,
-    trace_subscriber: Arc<dyn Subscriber + Send + Sync>,
+    /// The tracing subscriber which is associated with this telemetry instance. May be `None` if
+    /// the user has not opted into any tracing configuration.
+    trace_subscriber: Option<Arc<dyn Subscriber + Send + Sync>>,
     attach_service_name: bool,
 }
 
 impl TelemetryInstance {
     fn new(
-        trace_subscriber: Arc<dyn Subscriber + Send + Sync>,
+        trace_subscriber: Option<Arc<dyn Subscriber + Send + Sync>>,
         logs_out: Option<Mutex<CoreLogsOut>>,
         metric_prefix: String,
         metrics: Option<Arc<dyn CoreMeter + 'static>>,
@@ -73,8 +75,8 @@ impl TelemetryInstance {
 
     /// Return the trace subscriber associated with the telemetry options/instance. Can be used
     /// to manually set the default for a thread or globally using the `tracing` crate, or with
-    /// [set_trace_subscriber_for_current_thread]
-    pub fn trace_subscriber(&self) -> Arc<dyn Subscriber + Send + Sync> {
+    /// [set_trace_subscriber_for_current_thread].
+    pub fn trace_subscriber(&self) -> Option<Arc<dyn Subscriber + Send + Sync>> {
         self.trace_subscriber.clone()
     }
 
@@ -172,7 +174,7 @@ pub fn telemetry_init(opts: TelemetryOptions) -> Result<TelemetryInstance, anyho
     let mut forward_layer = None;
     // ===================================
 
-    if let Some(ref logger) = opts.logging {
+    let tracing_sub = opts.logging.map(|logger| {
         match logger {
             Logger::Console { filter } => {
                 // This is silly dupe but can't be avoided without boxing.
@@ -206,18 +208,18 @@ pub fn telemetry_init(opts: TelemetryOptions) -> Result<TelemetryInstance, anyho
                 forward_layer = Some(export_layer.with_filter(EnvFilter::new(filter)));
             }
         };
-    };
+        let reg = tracing_subscriber::registry()
+            .with(console_pretty_layer)
+            .with(console_compact_layer)
+            .with(forward_layer);
 
-    let reg = tracing_subscriber::registry()
-        .with(console_pretty_layer)
-        .with(console_compact_layer)
-        .with(forward_layer);
-
-    #[cfg(feature = "tokio-console")]
-    let reg = reg.with(console_subscriber::spawn());
+        #[cfg(feature = "tokio-console")]
+        let reg = reg.with(console_subscriber::spawn());
+        Arc::new(reg) as Arc<dyn Subscriber + Send + Sync>
+    });
 
     Ok(TelemetryInstance::new(
-        Arc::new(reg),
+        tracing_sub,
         logs_out,
         opts.metric_prefix,
         opts.metrics,
@@ -234,7 +236,9 @@ pub fn telemetry_init_global(opts: TelemetryOptions) -> Result<(), anyhow::Error
         .is_ok()
     {
         let ti = telemetry_init(opts)?;
-        tracing::subscriber::set_global_default(ti.trace_subscriber())?;
+        if let Some(ts) = ti.trace_subscriber() {
+            tracing::subscriber::set_global_default(ts)?;
+        }
     }
     Ok(())
 }

core/src/worker/workflow/mod.rs

@@ -185,7 +185,7 @@ impl Workflows {
         let (start_polling_tx, start_polling_rx) = oneshot::channel();
         // We must spawn a task to constantly poll the activation stream, because otherwise
         // activation completions would not cause anything to happen until the next poll.
-        let tracing_sub = telem_instance.map(|ti| ti.trace_subscriber());
+        let tracing_sub = telem_instance.and_then(|ti| ti.trace_subscriber());
         let processing_task = thread::spawn(move || {
             if let Some(ts) = tracing_sub {
                 set_trace_subscriber_for_current_thread(ts);
@@ -620,7 +620,7 @@ impl Workflows {
             }
             let with_permits = reserved_act_permits
                 .into_iter()
-                .zip(eager_acts.into_iter())
+                .zip(eager_acts)
                 .map(|(permit, resp)| TrackedPermittedTqResp { permit, resp });
             if with_permits.len() > 0 {
                 debug!(

core/src/worker/workflow/workflow_stream.rs

@@ -187,7 +187,7 @@ impl WFStream {
                     }
                 };
 
-                activations.extend(maybe_act.into_iter());
+                activations.extend(maybe_act);
                 activations.extend(state.reconcile_buffered());
 
                 // Always flush *after* actually handling the input, as this allows LA sink