Fix stdout configuration format handling and defaults

crates/logfwd-config/src/lib.rs

@@ -189,8 +189,9 @@ pub enum Format {
     Raw,
     Auto,
     /// Human-readable colored console output for debugging/testing.
-    #[serde(alias = "text")]
     Console,
+    /// Plain text output.
+    Text,
 }
 
 impl fmt::Display for Format {
@@ -203,6 +204,7 @@ impl fmt::Display for Format {
             Format::Raw => f.write_str("raw"),
             Format::Auto => f.write_str("auto"),
             Format::Console => f.write_str("console"),
+            Format::Text => f.write_str("text"),
         }
     }
 }
@@ -2437,9 +2439,8 @@ input:
 output:
   type: stdout
 ";
-        let cfg =
-            Config::load_str(yaml).expect("format: text should be accepted as alias for console");
+        let cfg = Config::load_str(yaml).expect("format: text should be accepted");
         let pipe = &cfg.pipelines["default"];
-        assert_eq!(pipe.inputs[0].format, Some(Format::Console));
+        assert_eq!(pipe.inputs[0].format, Some(Format::Text));
     }
 }

crates/logfwd-output/src/arrow_ipc_sink.rs

@@ -340,7 +340,7 @@ mod tests {
         ]));
         let message = StringArray::from(vec![Some("hello world"), Some("goodbye"), None]);
         let status = Int64Array::from(vec![Some(200), Some(500), Some(404)]);
-        let latency = Float64Array::from(vec![Some(1.5), None, Some(3.14)]);
+        let latency = Float64Array::from(vec![Some(1.5), None, Some(3.5)]);
         RecordBatch::try_new(
             schema,
             vec![Arc::new(message), Arc::new(status), Arc::new(latency)],

crates/logfwd-output/src/lib.rs

@@ -757,9 +757,20 @@ pub fn build_sink_factory(
         }
         OutputType::Stdout => {
             let fmt = match cfg.format.as_ref() {
-                Some(Format::Json) => StdoutFormat::Json,
+                Some(Format::Cri) | Some(Format::Raw) | Some(Format::Auto) => {
+                    return Err(OutputError::Construction(format!(
+                        "output '{name}': format '{}' is only supported for inputs",
+                        cfg.format.as_ref().unwrap()
+                    )));
+                }
                 Some(Format::Console) => StdoutFormat::Console,
-                _ => StdoutFormat::Text,
+                Some(Format::Text) => StdoutFormat::Text,
+                Some(Format::Json) | None => StdoutFormat::Json,
+                Some(f) => {
+                    return Err(OutputError::Construction(format!(
+                        "output '{name}': format '{f}' is not supported for stdout"
+                    )));
+                }
             };
             Ok(Arc::new(StdoutSinkFactory::new(
                 name.to_string(),
@@ -1163,6 +1174,49 @@ mod tests {
         assert_eq!(factory.name(), "test");
     }
 
+    #[test]
+    fn test_build_sink_factory_stdout_input_only_formats() {
+        for format in [Format::Cri, Format::Raw, Format::Auto] {
+            let cfg = OutputConfig {
+                name: Some("test".to_string()),
+                output_type: OutputType::Stdout,
+                endpoint: None,
+                protocol: None,
+                compression: None,
+                request_mode: None,
+                format: Some(format.clone()),
+                path: None,
+                index: None,
+                auth: None,
+            };
+            let result = build_sink_factory("test", &cfg, Arc::new(ComponentStats::new()));
+            assert!(result.is_err(), "Expected error for format {:?}", format);
+            let err = result.err().unwrap();
+            assert!(
+                err.to_string().contains("only supported for inputs"),
+                "error should mention input-only format: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_build_sink_factory_stdout_default_format() {
+        let cfg = OutputConfig {
+            name: Some("test".to_string()),
+            output_type: OutputType::Stdout,
+            endpoint: None,
+            protocol: None,
+            compression: None,
+            request_mode: None,
+            format: None,
+            path: None,
+            index: None,
+            auth: None,
+        };
+        let factory = build_sink_factory("test", &cfg, Arc::new(ComponentStats::new())).unwrap();
+        assert_eq!(factory.name(), "test");
+    }
+
     #[test]
     fn test_build_output_sink_otlp() {
         let cfg = OutputConfig {
@@ -1807,17 +1861,17 @@ mod write_row_json_tests {
     #[test]
     fn float32_serializes_as_number() {
         use arrow::array::Float32Array;
-        let batch = make_batch(vec![("val", Arc::new(Float32Array::from(vec![3.14_f32])))]);
+        let batch = make_batch(vec![("val", Arc::new(Float32Array::from(vec![3.5_f32])))]);
         let json = render(&batch, 0);
         let v: serde_json::Value = serde_json::from_str(&json).expect("must be valid JSON");
         assert!(
             v["val"].is_number(),
             "Float32 should serialize as JSON number, got {json}"
         );
-        let diff = (v["val"].as_f64().unwrap() - 3.14_f64).abs();
+        let diff = (v["val"].as_f64().unwrap() - 3.5_f64).abs();
         assert!(
             diff < 0.001,
-            "Float32 value should be ~3.14, got {}",
+            "Float32 value should be ~3.5, got {}",
             v["val"]
         );
     }
@@ -1987,7 +2041,7 @@ mod write_row_json_tests {
         let batch = make_batch(vec![
             ("dur_int", Arc::new(Int64Array::from(vec![42]))),
             ("label_str", Arc::new(StringArray::from(vec!["hello"]))),
-            ("score_float", Arc::new(Float64Array::from(vec![3.14]))),
+            ("score_float", Arc::new(Float64Array::from(vec![3.5]))),
         ]);
         let json = render(&batch, 0);
         let v: serde_json::Value = serde_json::from_str(&json).expect("must be valid JSON");
@@ -2001,7 +2055,7 @@ mod write_row_json_tests {
             "alias 'label_str' must be preserved, got: {json}"
         );
         assert!(
-            (v["score_float"].as_f64().unwrap() - 3.14).abs() < 0.01,
+            (v["score_float"].as_f64().unwrap() - 3.5).abs() < 0.01,
             "alias 'score_float' must be preserved, got: {json}"
         );
     }

crates/logfwd-output/src/sink.rs

@@ -351,7 +351,7 @@ mod tests {
             Box::pin(async { Ok(()) })
         }
 
-        fn name(&self) -> &str {
+        fn name(&self) -> &'static str {
             "stub"
         }
 

crates/logfwd-output/tests/it/elasticsearch_arrow_ipc.rs

@@ -114,7 +114,7 @@ async fn setup_test_data(sink: &mut Box<dyn logfwd_output::Sink>) -> RecordBatch
 }
 
 #[test]
-#[ignore] // Run with: cargo test --test elasticsearch_arrow_ipc -- --ignored
+#[ignore = "needs es instance"] // Run with: cargo test --test elasticsearch_arrow_ipc -- --ignored
 fn test_query_arrow_all_documents() {
     let rt = tokio::runtime::Runtime::new().expect("failed to create runtime");
     rt.block_on(async {
@@ -188,7 +188,7 @@ fn test_query_arrow_all_documents() {
 }
 
 #[test]
-#[ignore]
+#[ignore = "needs es instance"]
 fn test_query_arrow_with_filter() {
     let rt = tokio::runtime::Runtime::new().expect("failed to create runtime");
     rt.block_on(async {
@@ -249,7 +249,7 @@ fn test_query_arrow_with_filter() {
 }
 
 #[test]
-#[ignore]
+#[ignore = "needs es instance"]
 fn test_query_arrow_with_projection() {
     let rt = tokio::runtime::Runtime::new().expect("failed to create runtime");
     rt.block_on(async {
@@ -318,7 +318,7 @@ fn test_query_arrow_with_projection() {
 }
 
 #[test]
-#[ignore]
+#[ignore = "needs es instance"]
 fn test_query_arrow_empty_result() {
     let rt = tokio::runtime::Runtime::new().expect("failed to create runtime");
     rt.block_on(async {