chore(streaming): enable actor metrics in release mode

src/stream/src/executor/debug.rs

@@ -58,13 +58,12 @@ impl DebugExecutor {
         }
     }
 
-    #[allow(clippy::redundant_clone)]
     #[allow(clippy::let_and_return)]
-    fn wrap(
+    fn wrap_debug(
         info: Arc<ExecutorInfo>,
         extra: DebugExtraInfo,
-        stream: impl MessageStream,
-    ) -> impl MessageStream {
+        stream: impl MessageStream + 'static,
+    ) -> impl MessageStream + 'static {
         // Trace
         let stream = trace::trace(
             info.clone(),
@@ -73,15 +72,43 @@ impl DebugExecutor {
             extra.metrics,
             stream,
         );
+
         // Schema check
         let stream = schema_check::schema_check(info.clone(), stream);
         // Epoch check
         let stream = epoch_check::epoch_check(info.clone(), stream);
         // Update check
-        let stream = update_check::update_check(info.clone(), stream);
+        let stream = update_check::update_check(info, stream);
 
         stream
     }
+
+    #[allow(clippy::let_and_return)]
+    fn wrap_release(
+        info: Arc<ExecutorInfo>,
+        extra: DebugExtraInfo,
+        stream: impl MessageStream + 'static,
+    ) -> impl MessageStream + 'static {
+        // Metrics
+        let stream = trace::metrics(extra.actor_id, extra.metrics, stream);
+
+        // Epoch check
+        let stream = epoch_check::epoch_check(info, stream);
+
+        stream
+    }
+
+    fn wrap(
+        info: Arc<ExecutorInfo>,
+        extra: DebugExtraInfo,
+        stream: impl MessageStream + 'static,
+    ) -> BoxedMessageStream {
+        if cfg!(debug_assertions) {
+            Self::wrap_debug(info, extra, stream).boxed()
+        } else {
+            Self::wrap_release(info, extra, stream).boxed()
+        }
+    }
 }
 
 impl Executor for DebugExecutor {

src/stream/src/executor/debug/trace.rs

@@ -61,3 +61,24 @@ pub async fn trace(
         yield message;
     }
 }
+
+/// Streams wrapped by `metrics` will update actor metrics.
+#[try_stream(ok = Message, error = StreamExecutorError)]
+pub async fn metrics(actor_id: ActorId, metrics: Arc<StreamingMetrics>, input: impl MessageStream) {
+    let actor_id_string = actor_id.to_string();
+
+    pin_mut!(input);
+
+    while let Some(message) = input.next().await.transpose()? {
+        if let Message::Chunk(chunk) = &message {
+            if chunk.cardinality() > 0 {
+                metrics
+                    .actor_row_count
+                    .with_label_values(&[&actor_id_string])
+                    .inc_by(chunk.cardinality() as u64);
+            }
+        }
+
+        yield message;
+    }
+}