Repository: vectordotdev/vector (language: rust)

Issue:
Allow file sink to accept metric events
### A note for the community

<!-- Please keep this note for the community -->
* Please vote on this issue by adding a 👍 [reaction](https://blog.github.com/2016-03-10-add-reactions-to-pull-requests-issues-and-comments/) to the original issue to help the community and maintainers prioritize this request
* If you are interested in working on this issue or have submitted a pull request, please leave a comment
<!-- Thank you for keeping this note for the community -->


### Use Cases

I am attempting to save metrics to disk using the `native_json` codec using `http_server` source & the `file` sink, but it appears to only support logs - see below https://github.com/vectordotdev/vector/blob/a8e82dd71a38065d32db872e489e5dc4edb6dbc3/src/sinks/file/mod.rs#L197

### Attempted Solutions

Similar to the workaround mentioned in https://github.com/vectordotdev/vector/issues/20756, I was able to work around this by using the `metric_to_log` transform to save metrics to disk, and then applied the inverse `log_to_metric` transform downstream.

### Proposal

Allow the file sink to accept metric events

### References

* https://github.com/vectordotdev/vector/issues/20756 (similar request for the socket sink)

### Version

vector 0.42.0 (aarch64-unknown-linux-gnu 3d16e34 2024-10-21 14:10:14.375255220)


Discussion / hints:
cc @nionata 
Hi @rahulsmehta, thank you for the request. 

It's been a while since I looked at the file sink implementation but I don't think it will require too much modification to support metrics. If you wanted to test how it works with metrics, you can remove the `& DataType::Log` and run a config.  The native JSON codecs support all data types.

Fix patch:
diff --git a/changelog.d/21704_file_sink_supports_encoding_input.feature.md b/changelog.d/21704_file_sink_supports_encoding_input.feature.md
new file mode 100644
index 0000000000000..762bbde189128
--- /dev/null
+++ b/changelog.d/21704_file_sink_supports_encoding_input.feature.md
@@ -0,0 +1,3 @@
+The file sink now supports any input event type that the configured encoding supports. It previously only supported log events.
+
+authors: nionata
diff --git a/src/sinks/file/mod.rs b/src/sinks/file/mod.rs
index 64110daff576f..f07a4a907b86e 100644
--- a/src/sinks/file/mod.rs
+++ b/src/sinks/file/mod.rs
@@ -27,7 +27,7 @@ use vector_lib::{
 
 use crate::{
     codecs::{Encoder, EncodingConfigWithFraming, SinkType, Transformer},
-    config::{AcknowledgementsConfig, DataType, GenerateConfig, Input, SinkConfig, SinkContext},
+    config::{AcknowledgementsConfig, GenerateConfig, Input, SinkConfig, SinkContext},
     event::{Event, EventStatus, Finalizable},
     expiring_hash_map::ExpiringHashMap,
     internal_events::{
@@ -194,7 +194,7 @@ impl SinkConfig for FileSinkConfig {
     }
 
     fn input(&self) -> Input {
-        Input::new(self.encoding.config().1.input_type() & DataType::Log)
+        Input::new(self.encoding.config().1.input_type())
     }
 
     fn acknowledgements(&self) -> &AcknowledgementsConfig {
@@ -443,9 +443,14 @@ impl StreamSink<Event> for FileSink {
 mod tests {
     use std::convert::TryInto;
 
+    use chrono::{SubsecRound, Utc};
     use futures::{stream, SinkExt};
     use similar_asserts::assert_eq;
-    use vector_lib::{event::LogEvent, sink::VectorSink};
+    use vector_lib::{
+        codecs::JsonSerializerConfig,
+        event::{LogEvent, TraceEvent},
+        sink::VectorSink,
+    };
 
     use super::*;
     use crate::{
@@ -453,7 +458,8 @@ mod tests {
         test_util::{
             components::{assert_sink_compliance, FILE_SINK_TAGS},
             lines_from_file, lines_from_gzip_file, lines_from_zstd_file, random_events_with_stream,
-            random_lines_with_stream, temp_dir, temp_file, trace_init,
+            random_lines_with_stream, random_metrics_with_stream,
+            random_metrics_with_stream_timestamp, temp_dir, temp_file, trace_init,
         },
     };
 
@@ -463,7 +469,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn single_partition() {
+    async fn log_single_partition() {
         let template = temp_file();
 
         let config = FileSinkConfig {
@@ -480,7 +486,7 @@ mod tests {
 
         let (input, _events) = random_lines_with_stream(100, 64, None);
 
-        run_assert_log_sink(config, input.clone()).await;
+        run_assert_log_sink(&config, input.clone()).await;
 
         let output = lines_from_file(template);
         for (input, output) in input.into_iter().zip(output) {
@@ -489,7 +495,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn single_partition_gzip() {
+    async fn log_single_partition_gzip() {
         let template = temp_file();
 
         let config = FileSinkConfig {
@@ -506,7 +512,7 @@ mod tests {
 
         let (input, _) = random_lines_with_stream(100, 64, None);
 
-        run_assert_log_sink(config, input.clone()).await;
+        run_assert_log_sink(&config, input.clone()).await;
 
         let output = lines_from_gzip_file(template);
         for (input, output) in input.into_iter().zip(output) {
@@ -515,7 +521,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn single_partition_zstd() {
+    async fn log_single_partition_zstd() {
         let template = temp_file();
 
         let config = FileSinkConfig {
@@ -532,7 +538,7 @@ mod tests {
 
         let (input, _) = random_lines_with_stream(100, 64, None);
 
-        run_assert_log_sink(config, input.clone()).await;
+        run_assert_log_sink(&config, input.clone()).await;
 
         let output = lines_from_zstd_file(template);
         for (input, output) in input.into_iter().zip(output) {
@@ -541,7 +547,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn many_partitions() {
+    async fn log_many_partitions() {
         let directory = temp_dir();
 
         let mut template = directory.to_string_lossy().to_string();
@@ -579,7 +585,7 @@ mod tests {
         input[7].as_mut_log().insert("date", "2019-29-07");
         input[7].as_mut_log().insert("level", "error");
 
-        run_assert_sink(config, input.clone().into_iter()).await;
+        run_assert_sink(&config, input.clone().into_iter()).await;
 
         let output = [
             lines_from_file(directory.join("warnings-2019-26-07.log")),
@@ -626,7 +632,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn reopening() {
+    async fn log_reopening() {
         trace_init();
 
         let template = temp_file();
@@ -683,7 +689,115 @@ mod tests {
         sink_handle.await.unwrap();
     }
 
-    async fn run_assert_log_sink(config: FileSinkConfig, events: Vec<String>) {
+    #[tokio::test]
+    async fn metric_single_partition() {
+        let template = temp_file();
+
+        let config = FileSinkConfig {
+            path: template.clone().try_into().unwrap(),
+            idle_timeout: default_idle_timeout(),
+            encoding: (None::<FramingConfig>, TextSerializerConfig::default()).into(),
+            compression: Compression::None,
+            acknowledgements: Default::default(),
+            timezone: Default::default(),
+            internal_metrics: FileInternalMetricsConfig {
+                include_file_tag: true,
+            },
+        };
+
+        let (input, _events) = random_metrics_with_stream(100, None, None);
+
+        run_assert_sink(&config, input.clone().into_iter()).await;
+
+        let output = lines_from_file(template);
+        for (input, output) in input.into_iter().zip(output) {
+            let metric_name = input.as_metric().name();
+            assert!(output.contains(metric_name));
+        }
+    }
+
+    #[tokio::test]
+    async fn metric_many_partitions() {
+        let directory = temp_dir();
+
+        let format = "%Y-%m-%d-%H-%M-%S";
+        let mut template = directory.to_string_lossy().to_string();
+        template.push_str(&format!("/{}.log", format));
+
+        let config = FileSinkConfig {
+            path: template.try_into().unwrap(),
+            idle_timeout: default_idle_timeout(),
+            encoding: (None::<FramingConfig>, TextSerializerConfig::default()).into(),
+            compression: Compression::None,
+            acknowledgements: Default::default(),
+            timezone: Default::default(),
+            internal_metrics: FileInternalMetricsConfig {
+                include_file_tag: true,
+            },
+        };
+
+        let metric_count = 3;
+        let timestamp = Utc::now().trunc_subsecs(3);
+        let timestamp_offset = Duration::from_secs(1);
+
+        let (input, _events) = random_metrics_with_stream_timestamp(
+            metric_count,
+            None,
+            None,
+            timestamp,
+            timestamp_offset,
+        );
+
+        run_assert_sink(&config, input.clone().into_iter()).await;
+
+        let output = (0..metric_count).map(|index| {
+            let expected_timestamp = timestamp + (timestamp_offset * index as u32);
+            let expected_filename =
+                directory.join(format!("{}.log", expected_timestamp.format(format)));
+
+            lines_from_file(expected_filename)
+        });
+        for (input, output) in input.iter().zip(output) {
+            // The format will partition by second and metrics are a second apart.
+            assert_eq!(
+                output.len(),
+                1,
+                "Expected the output file to contain one metric"
+            );
+            let output = &output[0];
+
+            let metric_name = input.as_metric().name();
+            assert!(output.contains(metric_name));
+        }
+    }
+
+    #[tokio::test]
+    async fn trace_single_partition() {
+        let template = temp_file();
+
+        let config = FileSinkConfig {
+            path: template.clone().try_into().unwrap(),
+            idle_timeout: default_idle_timeout(),
+            encoding: (None::<FramingConfig>, JsonSerializerConfig::default()).into(),
+            compression: Compression::None,
+            acknowledgements: Default::default(),
+            timezone: Default::default(),
+            internal_metrics: FileInternalMetricsConfig {
+                include_file_tag: true,
+            },
+        };
+
+        let (input, _events) = random_lines_with_stream(100, 64, None);
+
+        run_assert_trace_sink(&config, input.clone()).await;
+
+        let output = lines_from_file(template);
+        for (input, output) in input.iter().zip(output) {
+            assert!(output.contains(input));
+        }
+    }
+
+    async fn run_assert_log_sink(config: &FileSinkConfig, events: Vec<String>) {
         run_assert_sink(
             config,
             events.into_iter().map(LogEvent::from).map(Event::Log),
@@ -691,9 +805,21 @@ mod tests {
         .await;
     }
 
-    async fn run_assert_sink(config: FileSinkConfig, events: impl Iterator<Item = Event> + Send) {
+    async fn run_assert_trace_sink(config: &FileSinkConfig, events: Vec<String>) {
+        run_assert_sink(
+            config,
+            events
+                .into_iter()
+                .map(LogEvent::from)
+                .map(TraceEvent::from)
+                .map(Event::Trace),
+        )
+        .await;
+    }
+
+    async fn run_assert_sink(config: &FileSinkConfig, events: impl Iterator<Item = Event> + Send) {
         assert_sink_compliance(&FILE_SINK_TAGS, async move {
-            let sink = FileSink::new(&config, SinkContext::default()).unwrap();
+            let sink = FileSink::new(config, SinkContext::default()).unwrap();
             VectorSink::from_event_streamsink(sink)
                 .run(Box::pin(stream::iter(events.map(Into::into))))
                 .await


Test patch:
diff --git a/src/test_util/mod.rs b/src/test_util/mod.rs
index 7197ddfbd3692..51e2a75ee894c 100644
--- a/src/test_util/mod.rs
+++ b/src/test_util/mod.rs
@@ -16,7 +16,7 @@ use std::{
     task::{ready, Context, Poll},
 };
 
-use chrono::{SubsecRound, Utc};
+use chrono::{DateTime, SubsecRound, Utc};
 use flate2::read::MultiGzDecoder;
 use futures::{stream, task::noop_waker_ref, FutureExt, SinkExt, Stream, StreamExt, TryStreamExt};
 use openssl::ssl::{SslConnector, SslFiletype, SslMethod, SslVerifyMode};
@@ -286,10 +286,36 @@ pub fn random_metrics_with_stream(
     batch: Option<BatchNotifier>,
     tags: Option<MetricTags>,
 ) -> (Vec<Event>, impl Stream<Item = EventArray>) {
-    let timestamp = Utc::now().trunc_subsecs(3);
+    random_metrics_with_stream_timestamp(
+        count,
+        batch,
+        tags,
+        Utc::now().trunc_subsecs(3),
+        std::time::Duration::from_secs(2),
+    )
+}
+
+/// Generates event metrics with the provided tags and timestamp.
+///
+/// # Parameters
+/// - `count`: the number of metrics to generate
+/// - `batch`: the batch notifier to use with the stream
+/// - `tags`: the tags to apply to each metric event
+/// - `timestamp`: the timestamp to use for each metric event
+/// - `timestamp_offset`: the offset from the `timestamp` to use for each additional metric
+///
+/// # Returns
+/// A tuple of the generated metric events and the stream of the generated events
+pub fn random_metrics_with_stream_timestamp(
+    count: usize,
+    batch: Option<BatchNotifier>,
+    tags: Option<MetricTags>,
+    timestamp: DateTime<Utc>,
+    timestamp_offset: std::time::Duration,
+) -> (Vec<Event>, impl Stream<Item = EventArray>) {
     let events: Vec<_> = (0..count)
         .map(|index| {
-            let ts = timestamp + (std::time::Duration::from_secs(2) * index as u32);
+            let ts = timestamp + (timestamp_offset * index as u32);
             Event::Metric(
                 Metric::new(
                     format!("counter_{}", thread_rng().gen::<u32>()),