chore(dev): Add wrapper for file-source in vector-lib (vectordotd…

…ev#18984)

Cargo.toml

@@ -131,7 +131,6 @@ pin-project.workspace = true
 # Internal libs
 dnsmsg-parser = { path = "lib/dnsmsg-parser", optional = true }
 fakedata = { path = "lib/fakedata", optional = true }
-file-source = { path = "lib/file-source", optional = true }
 lookup = { package = "vector-lookup", path = "lib/vector-lookup" }
 portpicker = { path = "lib/portpicker" }
 prometheus-parser = { path = "lib/prometheus-parser", optional = true }
@@ -525,7 +524,7 @@ sources-dnstap = ["dep:base64", "dep:trust-dns-proto", "dep:dnsmsg-parser", "pro
 sources-docker_logs = ["docker"]
 sources-eventstoredb_metrics = []
 sources-exec = []
-sources-file = ["dep:file-source"]
+sources-file = ["vector-lib/file-source"]
 sources-file-descriptor = ["tokio-util/io"]
 sources-fluent = ["dep:base64", "sources-utils-net-tcp", "tokio-util/net", "dep:rmpv", "dep:rmp-serde", "dep:serde_bytes"]
 sources-gcp_pubsub = ["gcp", "dep:h2", "dep:prost-types", "protobuf-build", "dep:tonic"]
@@ -537,7 +536,7 @@ sources-internal_logs = []
 sources-internal_metrics = []
 sources-journald = []
 sources-kafka = ["dep:rdkafka"]
-sources-kubernetes_logs = ["dep:file-source", "kubernetes", "transforms-reduce"]
+sources-kubernetes_logs = ["vector-lib/file-source", "kubernetes", "transforms-reduce"]
 sources-logstash = ["sources-utils-net-tcp", "tokio-util/net"]
 sources-mongodb_metrics = ["dep:mongodb"]
 sources-nats = ["dep:async-nats", "dep:nkeys"]

lib/vector-lib/Cargo.toml

@@ -8,6 +8,7 @@ publish = false
 [dependencies]
 codecs = { path = "../codecs", default-features = false }
 enrichment = { path = "../enrichment" }
+file-source = { path = "../file-source", optional = true }
 vector-buffers = { path = "../vector-buffers", default-features = false }
 vector-common = { path = "../vector-common" }
 vector-config = { path = "../vector-config" }
@@ -17,6 +18,7 @@ vector-stream = { path = "../vector-stream" }
 [features]
 api = ["vector-core/api"]
 lua = ["vector-core/lua"]
+file-source = ["dep:file-source"]
 syslog = ["codecs/syslog"]
 test = ["vector-core/test"]
 vrl = ["vector-core/vrl"]

lib/vector-lib/src/lib.rs

@@ -1,5 +1,7 @@
 pub use codecs;
 pub use enrichment;
+#[cfg(feature = "file-source")]
+pub use file_source;
 pub use vector_buffers as buffers;
 pub use vector_common::{
     assert_event_data_eq, btreemap, byte_size_of, byte_size_of::ByteSizeOf, conversion,

src/internal_events/file.rs

@@ -82,8 +82,8 @@ impl<'a, P: std::fmt::Debug> InternalEvent for FileIoError<'a, P> {
 mod source {
     use std::{io::Error, path::Path, time::Duration};
 
-    use file_source::FileSourceInternalEvents;
     use metrics::counter;
+    use vector_lib::file_source::FileSourceInternalEvents;
 
     use super::{FileOpen, InternalEvent};
     use crate::emit;

src/sources/file.rs

@@ -2,12 +2,6 @@ use std::{convert::TryInto, future, path::PathBuf, time::Duration};
 
 use bytes::Bytes;
 use chrono::Utc;
-use file_source::{
-    calculate_ignore_before,
-    paths_provider::glob::{Glob, MatchOptions},
-    Checkpointer, FileFingerprint, FileServer, FingerprintStrategy, Fingerprinter, Line, ReadFrom,
-    ReadFromConfig,
-};
 use futures::{FutureExt, Stream, StreamExt, TryFutureExt};
 use lookup::{lookup_v2::OptionalValuePath, owned_value_path, path, OwnedValuePath};
 use regex::bytes::Regex;
@@ -17,6 +11,12 @@ use tokio::{sync::oneshot, task::spawn_blocking};
 use tracing::{Instrument, Span};
 use vector_lib::codecs::{BytesDeserializer, BytesDeserializerConfig};
 use vector_lib::configurable::configurable_component;
+use vector_lib::file_source::{
+    calculate_ignore_before,
+    paths_provider::glob::{Glob, MatchOptions},
+    Checkpointer, FileFingerprint, FileServer, FingerprintStrategy, Fingerprinter, Line, ReadFrom,
+    ReadFromConfig,
+};
 use vector_lib::finalizer::OrderedFinalizer;
 use vector_lib::{
     config::{LegacyKey, LogNamespace},

src/sources/kubernetes_logs/k8s_paths_provider.rs

@@ -4,9 +4,9 @@
 
 use std::path::PathBuf;
 
-use file_source::paths_provider::PathsProvider;
 use k8s_openapi::api::core::v1::{Namespace, Pod};
 use kube::runtime::reflector::{store::Store, ObjectRef};
+use vector_lib::file_source::paths_provider::PathsProvider;
 
 use super::path_helpers::build_pod_logs_directory;
 use crate::kubernetes::pod_manager_logic::extract_static_pod_config_hashsum;

src/sources/kubernetes_logs/mod.rs

@@ -8,10 +8,6 @@ use std::{path::PathBuf, time::Duration};
 
 use bytes::Bytes;
 use chrono::Utc;
-use file_source::{
-    calculate_ignore_before, Checkpointer, FileServer, FileServerShutdown, FingerprintStrategy,
-    Fingerprinter, Line, ReadFrom, ReadFromConfig,
-};
 use futures::{future::FutureExt, stream::StreamExt};
 use futures_util::Stream;
 use k8s_openapi::api::core::v1::{Namespace, Node, Pod};
@@ -27,6 +23,10 @@ use lookup::{lookup_v2::OptionalTargetPath, owned_value_path, path, OwnedTargetP
 use serde_with::serde_as;
 use vector_lib::codecs::{BytesDeserializer, BytesDeserializerConfig};
 use vector_lib::configurable::configurable_component;
+use vector_lib::file_source::{
+    calculate_ignore_before, Checkpointer, FileServer, FileServerShutdown, FingerprintStrategy,
+    Fingerprinter, Line, ReadFrom, ReadFromConfig,
+};
 use vector_lib::{config::LegacyKey, config::LogNamespace, EstimatedJsonEncodedSizeOf};
 use vector_lib::{
     internal_event::{ByteSize, BytesReceived, InternalEventHandle as _, Protocol},

src/sources/kubernetes_logs/util.rs

@@ -1,14 +1,14 @@
 use std::{error::Error, future::Future, time::Duration};
 
-use file_source::{
-    paths_provider::PathsProvider, Checkpointer, FileServer, FileServerShutdown,
-    FileSourceInternalEvents, Line,
-};
 use futures::{
     future::{select, Either},
     pin_mut, FutureExt, Sink,
 };
 use tokio::task::spawn_blocking;
+use vector_lib::file_source::{
+    paths_provider::PathsProvider, Checkpointer, FileServer, FileServerShutdown,
+    FileSourceInternalEvents, Line,
+};
 
 /// A tiny wrapper around a [`FileServer`] that runs it as a [`spawn_blocking`]
 /// task.