Repository: numaproj/numaflow

Issue:
Update the resources for sidecar container( monitor ) used for vertex runtime information.


Discussion:

Solution patch:
diff --git a/pkg/apis/numaflow/v1alpha1/monitor_container.go b/pkg/apis/numaflow/v1alpha1/monitor_container.go
index 0b9e36a6..8c522fb8 100644
--- a/pkg/apis/numaflow/v1alpha1/monitor_container.go
+++ b/pkg/apis/numaflow/v1alpha1/monitor_container.go
@@ -15,7 +15,10 @@ limitations under the License.
 */
 package v1alpha1
 
-import corev1 "k8s.io/api/core/v1"
+import (
+	corev1 "k8s.io/api/core/v1"
+	resource "k8s.io/apimachinery/pkg/api/resource"
+)
 
 // volume mount to the runtime path
 func getRuntimeVolumeMount() []corev1.VolumeMount {
@@ -28,6 +31,13 @@ func getRuntimeVolumeMount() []corev1.VolumeMount {
 }
 func buildMonitorContainer(req getContainerReq) corev1.Container {
 	return containerBuilder{}.init(req).
+		resources(corev1.ResourceRequirements{
+			Limits: corev1.ResourceList{},
+			Requests: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("10m"),
+				corev1.ResourceMemory: resource.MustParse("20Mi"),
+			},
+		}).
 		volumeMounts(getRuntimeVolumeMount()...).
 		name(CtrMonitor).
 		asSidecar().
diff --git a/rust/numaflow-core/src/lib.rs b/rust/numaflow-core/src/lib.rs
index 45441ff3..93602b5f 100644
--- a/rust/numaflow-core/src/lib.rs
+++ b/rust/numaflow-core/src/lib.rs
@@ -64,15 +64,12 @@ mod serving_store;
 /// [Watermark]: https://numaflow.numaproj.io/core-concepts/watermarks/
 mod watermark;
 
-use numaflow_monitor::runtime::Runtime;
+use numaflow_monitor::runtime;
 
 pub async fn run() -> Result<()> {
     let cln_token = CancellationToken::new();
     let shutdown_cln_token = cln_token.clone();
 
-    // Initialize runtime for persisting errors
-    let runtime = Runtime::new(None);
-
     // wait for SIG{INT,TERM} and invoke cancellation token.
     let shutdown_handle: JoinHandle<Result<()>> = tokio::spawn(async move {
         shutdown_signal().await;
@@ -88,7 +85,7 @@ pub async fn run() -> Result<()> {
             if let Err(e) = monovertex::start_forwarder(cln_token, &config).await {
                 if let Error::Grpc(e) = e {
                     error!(error=?e, "Monovertex failed because of UDF failure");
-                    runtime.persist_application_error(e)
+                    runtime::persist_application_error(e);
                 } else {
                     error!(?e, "Error running monovertex");
                 }
@@ -103,7 +100,7 @@ pub async fn run() -> Result<()> {
             if let Err(e) = pipeline::start_forwarder(cln_token, config).await {
                 if let Error::Grpc(e) = e {
                     error!(error=?e, "Pipeline failed because of UDF failure");
-                    runtime.persist_application_error(e)
+                    runtime::persist_application_error(e);
                 } else {
                     error!(?e, "Error running pipeline");
                 }
diff --git a/rust/numaflow-core/src/sink/user_defined.rs b/rust/numaflow-core/src/sink/user_defined.rs
index ab3ed001..d960bd55 100644
--- a/rust/numaflow-core/src/sink/user_defined.rs
+++ b/rust/numaflow-core/src/sink/user_defined.rs
@@ -1,9 +1,11 @@
+use bytes::Bytes;
+use numaflow_monitor::runtime;
 use numaflow_pb::clients::sink::sink_client::SinkClient;
 use numaflow_pb::clients::sink::{Handshake, SinkRequest, SinkResponse, TransmissionStatus};
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::transport::Channel;
-use tonic::{Request, Streaming};
+use tonic::{Code, Request, Status, Streaming};
 use tracing::error;
 
 use crate::message::Message;
@@ -125,6 +127,16 @@ impl Sink for UserDefinedSink {
             if response.status.is_some_and(|s| s.eot) {
                 if responses.len() != num_requests {
                     error!("received EOT message before all responses are received, we will wait indefinitely for the remaining responses");
+                    // persist the error for debugging
+                    runtime::persist_application_error(Status::with_details(
+                        Code::Internal,
+                        "UDF_PARTIAL_RESPONSE(udsink)",
+                        Bytes::from_static(
+                            b"received End-Of-Transmission (EOT) before all responses are received from the ud-sink,\
+                            we will wait indefinitely for the remaining responses. This indicates that there is a bug\
+                            in the user-code. Please check whether you are accidentally skipping the messages.",
+                        ),
+                    ));
                 } else {
                     break;
                 }
diff --git a/rust/numaflow-core/src/transformer.rs b/rust/numaflow-core/src/transformer.rs
index 6a890fca..c5a30056 100644
--- a/rust/numaflow-core/src/transformer.rs
+++ b/rust/numaflow-core/src/transformer.rs
@@ -1,4 +1,12 @@
+use bytes::Bytes;
+use numaflow_monitor::runtime;
+use numaflow_pb::clients::sourcetransformer::source_transform_client::SourceTransformClient;
 use std::sync::Arc;
+use tokio::sync::{mpsc, oneshot, Semaphore};
+use tokio_util::sync::CancellationToken;
+use tonic::transport::Channel;
+use tonic::{Code, Status};
+use tracing::error;
 
 use crate::error::Error;
 use crate::message::Message;
@@ -6,11 +14,6 @@ use crate::metrics::{monovertex_metrics, mvtx_forward_metric_labels};
 use crate::tracker::TrackerHandle;
 use crate::transformer::user_defined::UserDefinedTransformer;
 use crate::Result;
-use numaflow_pb::clients::sourcetransformer::source_transform_client::SourceTransformClient;
-use tokio::sync::{mpsc, oneshot, Semaphore};
-use tokio_util::sync::CancellationToken;
-use tonic::transport::Channel;
-use tracing::error;
 
 /// User-Defined Transformer is a custom transformer that can be built by the user.
 ///
@@ -121,6 +124,21 @@ impl Transformer {
             }
         };
 
+        if response.is_empty() {
+            error!("received empty response from server (transformer), we will wait indefinitely");
+            // persist the error for debugging
+            runtime::persist_application_error(Status::with_details(
+                Code::Internal,
+                "UDF_PARTIAL_RESPONSE(transformer)",
+                Bytes::from_static(
+                    b"received End-Of-Transmission (EOT) before all responses are received from the transformer,\
+                            we will wait indefinitely for the remaining responses. This indicates that there is a bug\
+                            in the user-code. Please check whether you are accidentally skipping the messages.",
+                ),
+            ));
+            futures::future::pending::<()>().await;
+        }
+
         monovertex_metrics()
             .transformer
             .time
diff --git a/rust/numaflow-monitor/src/app.rs b/rust/numaflow-monitor/src/app.rs
index 8afd63c7..36344694 100644
--- a/rust/numaflow-monitor/src/app.rs
+++ b/rust/numaflow-monitor/src/app.rs
@@ -116,6 +116,7 @@ async fn graceful_shutdown(handle: Handle, server_config: MonitorServerConfig) {
 mod tests {
     use crate::config::generate_certs;
     use crate::error::Result;
+    use crate::runtime::persist_application_error_to_file;
 
     use super::*;
     use axum::body::Body;
@@ -166,7 +167,7 @@ mod tests {
 
         // Create a Runtime instance with the temporary directory path
         let runtime = Runtime::new(Some(RuntimeInfoConfig {
-            app_error_path: application_error_path,
+            app_error_path: application_error_path.clone(),
             max_error_files_per_container: 2,
         }));
 
@@ -174,7 +175,7 @@ mod tests {
         let grpc_status = Status::internal("UDF_EXECUTION_ERROR(udsource): Test error message");
 
         // Call the function to persist error in temp app directory
-        runtime.persist_application_error(grpc_status.clone());
+        persist_application_error_to_file(application_error_path, 5, grpc_status.clone());
 
         let state = Arc::new(AppState {
             runtime: Arc::new(runtime),
diff --git a/rust/numaflow-monitor/src/runtime.rs b/rust/numaflow-monitor/src/runtime.rs
index 98fbee50..5d9a08c1 100644
--- a/rust/numaflow-monitor/src/runtime.rs
+++ b/rust/numaflow-monitor/src/runtime.rs
@@ -7,11 +7,14 @@ use std::convert::TryFrom;
 use std::fs;
 use std::fs::File;
 use std::io::Write;
+use std::os::unix::fs::DirBuilderExt as _;
 use std::path::Path;
 use std::str;
+use std::sync::OnceLock;
 use tonic::Status;
 use tracing::error;
 
+static PERSIST_APPLICATION_ERROR_ONCE: OnceLock<()> = OnceLock::new();
 const CURRENT_FILE: &str = "current-numa.json";
 
 /// Represents a single runtime error entry persisted by the application.
@@ -66,6 +69,121 @@ impl From<RuntimeErrorEntry> for String {
     }
 }
 
+/// Persists a gRPC error as a JSON file in the appropriate container directory.
+/// It organizes error files in a directory structure based on container names and ensures that the
+/// number of error files per container (files may be written from udf) does not exceed a specified limit.
+/// If the limit is exceeded, the oldest file is removed to make room for new entry. This function runs only once.
+///
+/// # Parameters:
+/// - `grpc_status`: The gRPC error (`tonic::Status`) to be persisted.
+///
+/// # Example:
+/// ```no_run
+///  use numaflow_monitor::runtime;
+///  let grpc_status = tonic::Status::internal("UDF_EXECUTION_ERROR(container-name): Test error");
+///  runtime::persist_application_error(grpc_status);
+/// ```
+//
+pub fn persist_application_error(grpc_status: Status) {
+    if PERSIST_APPLICATION_ERROR_ONCE.set(()).is_err() {
+        return;
+    }
+    persist_application_error_to_file(
+        RuntimeInfoConfig::default().app_error_path,
+        RuntimeInfoConfig::default().max_error_files_per_container,
+        grpc_status,
+    );
+}
+
+pub(crate) fn persist_application_error_to_file(
+    application_error_path: String,
+    max_error_files_per_container: usize,
+    grpc_status: Status,
+) {
+    // extract the type of udf container based on the error message
+    let container_name = extract_container_name(grpc_status.message());
+    // skip processing if the container name is empty. This happens only if
+    // the gRPC status is not created by us (e.g., unknown bugs like https://github.com/grpc/grpc-go/issues/7641)
+    // TODO: we should try to expose this in the UI if we encounter a few of this in prod.
+    if container_name.is_empty() {
+        error!(?grpc_status, "unknown-container");
+        return;
+    }
+    // create a directory for the container if it doesn't exist with permissions to read, write, and execute for all
+    let dir_path = Path::new(&application_error_path.clone()).join(&container_name);
+    if !dir_path.exists() {
+        let mut builder = fs::DirBuilder::new();
+        builder.recursive(true);
+        builder.mode(0o777);
+        builder
+            .create(&dir_path)
+            .expect("Failed to create application errors directory");
+    }
+
+    // this is to check the number of files in the directory
+    // additional check in place to process only files and ignore directories
+    // ignore files starting with prefix `current`
+    let mut files: Vec<_> = fs::read_dir(&dir_path)
+        .expect("Failed to read application errors directory")
+        .filter_map(|entry| entry.ok())
+        .filter(|entry| entry.path().is_file())
+        .filter(|entry| {
+            entry
+                .file_name()
+                .to_str()
+                .map(|name| !name.starts_with("current"))
+                .unwrap_or(false)
+        })
+        .collect();
+
+    // sort the files based on timestamp
+    files.sort_by_key(|e| {
+        e.file_name()
+            .to_str()
+            .and_then(|name| name.split('-').next())
+            .and_then(|timestamp| timestamp.parse::<i64>().ok())
+    });
+
+    // remove the oldest files until the number of files is within the max limit
+    // this is to ensure that we don't exceed the max limit of files in the container directory
+    while files.len() >= max_error_files_per_container {
+        if let Some(oldest_file) = files.first() {
+            if let Err(e) = fs::remove_file(oldest_file.path()) {
+                error!(
+                    "Failed to remove the oldest application error file: {:?}, error: {:?}",
+                    oldest_file.path(),
+                    e
+                );
+                break;
+            }
+            files.remove(0);
+        }
+    }
+
+    let timestamp = Utc::now().timestamp();
+    let runtime_error_entry =
+        RuntimeErrorEntry::from((&grpc_status, container_name.as_str(), timestamp));
+    let json_str: String = runtime_error_entry.into();
+
+    // Write the error details to a temporary file and rename it to a
+    // timestamped file once the write operation is complete.
+    // this is to ensure that while reading we skip this file to avoid race condition
+    let current_file_path = dir_path.join(CURRENT_FILE);
+    // append numa to the file name to denote files created by numa container
+    let file_name = format!("{}-numa.json", timestamp);
+    let final_file_path = dir_path.join(&file_name);
+
+    let mut current_file =
+        File::create(&current_file_path).expect("Failed to create current application errors file");
+    current_file
+        .write_all(json_str.as_bytes())
+        .expect("Failed to write to current application error file");
+
+    // rename the current file to the final file name once write operation completes
+    fs::rename(&current_file_path, &final_file_path)
+        .expect("Failed to rename current file to final file name");
+}
+
 /// A structure used to represent API responses containing runtime error entries.
 #[derive(serde::Serialize, serde::Deserialize)]
 pub(crate) struct ApiResponse {
@@ -76,6 +194,7 @@ pub(crate) struct ApiResponse {
     pub(crate) data: Vec<RuntimeErrorEntry>,
 }
 
+#[allow(dead_code)]
 /// Response for managing runtime error persistence and retrieval.
 pub struct Runtime {
     /// The root directory where error files are stored.
@@ -94,84 +213,6 @@ impl Runtime {
         }
     }
 
-    /// Persists a gRPC error as a JSON file in the appropriate container directory.
-    /// It organizes error files in a directory structure based on container names and ensures that the
-    /// number of error files per container does not exceed a specified limit. If the limit is exceeded,
-    /// the oldest file is removed to make room for new entries.
-    ///
-    /// # Parameters:
-    /// - `grpc_status`: The gRPC error (`tonic::Status`) to be persisted.
-    ///
-    /// # Example:
-    /// ```no_run
-    ///  use numaflow_monitor::runtime::Runtime;
-    ///  let grpc_status = tonic::Status::internal("UDF_EXECUTION_ERROR(container-name): Test error");
-    ///  Runtime::new(None).persist_application_error(grpc_status);
-    /// ```
-    pub fn persist_application_error(&self, grpc_status: Status) {
-        // extract the type of udf container based on the error message
-        let container_name = extract_container_name(grpc_status.message());
-        // skip processing if the container name is empty. This happens only if
-        // the gRPC status is not created by us (e.g., unknown bugs like https://github.com/grpc/grpc-go/issues/7641)
-        // TODO: we should try to expose this in the UI if we encounter a few of this in prod.
-        if container_name.is_empty() {
-            error!(?grpc_status, "unknown-container");
-            return;
-        }
-        // create a directory for the container if it doesn't exist
-        let dir_path = Path::new(&self.application_error_path).join(&container_name);
-        if !dir_path.exists() {
-            fs::create_dir_all(&dir_path).expect("Failed to create application errors directory");
-        }
-
-        // this is to check the number of files in the directory
-        // additional check in place to process only files and ignore directories
-        let mut files: Vec<_> = fs::read_dir(&dir_path)
-            .expect("Failed to read application errors directory")
-            .filter_map(|entry| entry.ok())
-            .filter(|entry| entry.path().is_file())
-            .collect();
-
-        // sort the files based on timestamp
-        files.sort_by_key(|e| {
-            e.file_name()
-                .to_str()
-                .and_then(|name| name.split('-').next())
-                .and_then(|timestamp| timestamp.parse::<i64>().ok())
-        });
-
-        // remove the oldest file if the number of files exceeds the limit
-        if files.len() >= self.max_error_files_per_container {
-            if let Some(oldest_file) = files.first() {
-                fs::remove_file(oldest_file.path())
-                    .expect("Failed to remove the oldest application error file");
-            }
-        }
-
-        let timestamp = Utc::now().timestamp();
-        let runtime_error_entry =
-            RuntimeErrorEntry::from((&grpc_status, container_name.as_str(), timestamp));
-        let json_str: String = runtime_error_entry.into();
-
-        // Write the error details to a temporary file (`current-numa.json`) and rename it to a
-        // timestamped file once the write operation is complete.
-        // this is to ensure that while reading we skip this file to avoid race condition
-        let current_file_path = dir_path.join(CURRENT_FILE);
-        // append numa to the file name to avoid collision with files written from udf with same timestamp
-        let file_name = format!("{}-numa.json", timestamp);
-        let final_file_path = dir_path.join(&file_name);
-
-        let mut current_file = File::create(&current_file_path)
-            .expect("Failed to create current application errors file");
-        current_file
-            .write_all(json_str.as_bytes())
-            .expect("Failed to write to current application error file");
-
-        // rename the current file to the final file name once write operation completes
-        fs::rename(&current_file_path, &final_file_path)
-            .expect("Failed to rename current file to final file name");
-    }
-
     // ## File Structure
     // The runtime module organizes application error files in a directory structure as follows:
     // Root: /var/numaflow/runtime/
@@ -217,7 +258,7 @@ impl Runtime {
                     .file_name()
                     .to_str()
                     .expect("file name should be valid")
-                    == CURRENT_FILE
+                    .starts_with("current")
                 {
                     continue;
                 }
@@ -308,26 +349,20 @@ mod tests {
     }
 
     #[test]
-    fn test_persist_application_error() {
+    fn test_persist_application_error_to_file() {
         // Create a temporary directory for testing
         let temp_dir = tempdir().unwrap();
         let application_error_path = temp_dir.path().to_str().unwrap().to_string();
 
-        // Create a Runtime instance with the temporary directory path
-        let runtime = Runtime {
-            application_error_path,
-            max_error_files_per_container: 5,
-        };
-
         // Create a mock gRPC status
         let grpc_status = Status::internal("UDF_EXECUTION_ERROR(udsource): Test error message");
 
         // Call the function to test
-        runtime.persist_application_error(grpc_status.clone());
+        persist_application_error_to_file(application_error_path.clone(), 5, grpc_status.clone());
 
         // Verify that the directory for the container was created
         let container_name = extract_container_name(grpc_status.message());
-        let dir_path = Path::new(&runtime.application_error_path).join(&container_name);
+        let dir_path = Path::new(&application_error_path).join(&container_name);
         assert!(dir_path.exists());
 
         // Verify that a new error file was created
@@ -347,13 +382,6 @@ mod tests {
         // Create a temporary directory for testing
         let temp_dir = tempdir().unwrap();
         let application_error_path = temp_dir.path().to_str().unwrap().to_string();
-
-        // Create a Runtime instance with the temporary directory path
-        let runtime = Runtime {
-            application_error_path,
-            max_error_files_per_container: 5,
-        };
-
         // Create a mock gRPC status with an empty container name
         let grpc_status = Status::internal("UDF_EXECUTION_ERROR: Test error message");
 
@@ -362,10 +390,10 @@ mod tests {
         assert!(container_name.is_empty());
 
         // Call the function to test
-        runtime.persist_application_error(grpc_status.clone());
+        persist_application_error_to_file(application_error_path.clone(), 5, grpc_status.clone());
 
         // Verify that no files were created in the directory
-        let dir_path = Path::new(&runtime.application_error_path).join(&container_name);
+        let dir_path = Path::new(&application_error_path).join(&container_name);
         if dir_path.exists() {
             let files: Vec<_> = fs::read_dir(&dir_path)
                 .unwrap()



Test patch:
diff --git a/pkg/apis/numaflow/v1alpha1/monitor_container_test.go b/pkg/apis/numaflow/v1alpha1/monitor_container_test.go
index 7a4cf5af..8aeb71e5 100644
--- a/pkg/apis/numaflow/v1alpha1/monitor_container_test.go
+++ b/pkg/apis/numaflow/v1alpha1/monitor_container_test.go
@@ -60,8 +60,11 @@ func TestBuildMonitorContainer(t *testing.T) {
 	assert.Equal(t, "test-value", container.Env[0].Value, "Environment variable value mismatch")
 
 	// Verify resource requirements
-	assert.Equal(t, req.resources.Limits, container.Resources.Limits, "Resource limits mismatch")
-	assert.Equal(t, req.resources.Requests, container.Resources.Requests, "Resource requests mismatch")
+	assert.Equal(t, container.Resources.Limits, corev1.ResourceList{}, "Resource limits mismatch")
+	assert.Equal(t, container.Resources.Requests, corev1.ResourceList{
+		corev1.ResourceCPU:    resource.MustParse("10m"),
+		corev1.ResourceMemory: resource.MustParse("20Mi"),
+	}, "Resource requests mismatch")
 
 	// Verify volume mounts --> should overwrite req's volume mounts
 	assert.Equal(t, 1, len(container.VolumeMounts), "Volume mount count mismatch")