vector/internal_events/

file.rs

#![allow(dead_code)] // TODO requires optional feature compilation

use std::borrow::Cow;

use metrics::{counter, gauge};
use vector_lib::{
    NamedInternalEvent,
    configurable::configurable_component,
    internal_event::{
        ComponentEventsDropped, InternalEvent, UNINTENTIONAL, error_stage, error_type,
    },
};

#[cfg(any(feature = "sources-file", feature = "sources-kubernetes_logs"))]
pub use self::source::*;

/// Configuration of internal metrics for file-based components.
#[configurable_component]
#[derive(Clone, Debug, PartialEq, Eq, Default)]
#[serde(deny_unknown_fields)]
pub struct FileInternalMetricsConfig {
    /// Whether or not to include the "file" tag on the component's corresponding internal metrics.
    ///
    /// This is useful for distinguishing between different files while monitoring. However, the tag's
    /// cardinality is unbounded.
    #[serde(default = "crate::serde::default_false")]
    pub include_file_tag: bool,
}

#[derive(Debug, NamedInternalEvent)]
pub struct FileOpen {
    pub count: usize,
}

impl InternalEvent for FileOpen {
    fn emit(self) {
        gauge!("open_files").set(self.count as f64);
    }
}

#[derive(Debug, NamedInternalEvent)]
pub struct FileBytesSent<'a> {
    pub byte_size: usize,
    pub file: Cow<'a, str>,
    pub include_file_metric_tag: bool,
}

impl InternalEvent for FileBytesSent<'_> {
    fn emit(self) {
        trace!(
            message = "Bytes sent.",
            byte_size = %self.byte_size,
            protocol = "file",
            file = %self.file,
        );
        if self.include_file_metric_tag {
            counter!(
                "component_sent_bytes_total",
                "protocol" => "file",
                "file" => self.file.clone().into_owned(),
            )
        } else {
            counter!(
                "component_sent_bytes_total",
                "protocol" => "file",
            )
        }
        .increment(self.byte_size as u64);
    }
}

#[derive(Debug, NamedInternalEvent)]
pub struct FileIoError<'a, P> {
    pub error: std::io::Error,
    pub code: &'static str,
    pub message: &'static str,
    pub path: &'a P,
    pub dropped_events: usize,
}

impl<P: std::fmt::Debug> InternalEvent for FileIoError<'_, P> {
    fn emit(self) {
        error!(
            message = %self.message,
            path = ?self.path,
            error = %self.error,
            error_code = %self.code,
            error_type = error_type::IO_FAILED,
            stage = error_stage::SENDING,
        );
        counter!(
            "component_errors_total",
            "error_code" => self.code,
            "error_type" => error_type::IO_FAILED,
            "stage" => error_stage::SENDING,
        )
        .increment(1);

        if self.dropped_events > 0 {
            emit!(ComponentEventsDropped::<UNINTENTIONAL> {
                count: self.dropped_events,
                reason: self.message,
            });
        }
    }
}

#[cfg(any(feature = "sources-file", feature = "sources-kubernetes_logs"))]
mod source {
    use std::{io::Error, path::Path, time::Duration};

    use bytes::BytesMut;
    use metrics::counter;
    use vector_lib::{
        NamedInternalEvent, emit,
        file_source_common::internal_events::FileSourceInternalEvents,
        internal_event::{ComponentEventsDropped, INTENTIONAL, error_stage, error_type},
        json_size::JsonSize,
    };

    use super::{FileOpen, InternalEvent};

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileBytesReceived<'a> {
        pub byte_size: usize,
        pub file: &'a str,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileBytesReceived<'_> {
        fn emit(self) {
            trace!(
                message = "Bytes received.",
                byte_size = %self.byte_size,
                protocol = "file",
                file = %self.file,
            );
            if self.include_file_metric_tag {
                counter!(
                    "component_received_bytes_total",
                    "protocol" => "file",
                    "file" => self.file.to_owned()
                )
            } else {
                counter!(
                    "component_received_bytes_total",
                    "protocol" => "file",
                )
            }
            .increment(self.byte_size as u64);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileEventsReceived<'a> {
        pub count: usize,
        pub file: &'a str,
        pub byte_size: JsonSize,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileEventsReceived<'_> {
        fn emit(self) {
            trace!(
                message = "Events received.",
                count = %self.count,
                byte_size = %self.byte_size,
                file = %self.file
            );
            if self.include_file_metric_tag {
                counter!(
                    "component_received_events_total",
                    "file" => self.file.to_owned(),
                )
                .increment(self.count as u64);
                counter!(
                    "component_received_event_bytes_total",
                    "file" => self.file.to_owned(),
                )
                .increment(self.byte_size.get() as u64);
            } else {
                counter!("component_received_events_total").increment(self.count as u64);
                counter!("component_received_event_bytes_total")
                    .increment(self.byte_size.get() as u64);
            }
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileChecksumFailed<'a> {
        pub file: &'a Path,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileChecksumFailed<'_> {
        fn emit(self) {
            warn!(
                message = "Currently ignoring file too small to fingerprint.",
                file = %self.file.display(),
            );
            if self.include_file_metric_tag {
                counter!(
                    "checksum_errors_total",
                    "file" => self.file.to_string_lossy().into_owned(),
                )
            } else {
                counter!("checksum_errors_total")
            }
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileFingerprintReadError<'a> {
        pub file: &'a Path,
        pub error: Error,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileFingerprintReadError<'_> {
        fn emit(self) {
            error!(
                message = "Failed reading file for fingerprinting.",
                file = %self.file.display(),
                error = %self.error,
                error_code = "reading_fingerprint",
                error_type = error_type::READER_FAILED,
                stage = error_stage::RECEIVING,
            );
            if self.include_file_metric_tag {
                counter!(
                    "component_errors_total",
                    "error_code" => "reading_fingerprint",
                    "error_type" => error_type::READER_FAILED,
                    "stage" => error_stage::RECEIVING,
                    "file" => self.file.to_string_lossy().into_owned(),
                )
            } else {
                counter!(
                    "component_errors_total",
                    "error_code" => "reading_fingerprint",
                    "error_type" => error_type::READER_FAILED,
                    "stage" => error_stage::RECEIVING,
                )
            }
            .increment(1);
        }
    }

    const DELETION_FAILED: &str = "deletion_failed";

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileDeleteError<'a> {
        pub file: &'a Path,
        pub error: Error,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileDeleteError<'_> {
        fn emit(self) {
            error!(
                message = "Failed in deleting file.",
                file = %self.file.display(),
                error = %self.error,
                error_code = DELETION_FAILED,
                error_type = error_type::COMMAND_FAILED,
                stage = error_stage::RECEIVING,
            );
            if self.include_file_metric_tag {
                counter!(
                    "component_errors_total",
                    "file" => self.file.to_string_lossy().into_owned(),
                    "error_code" => DELETION_FAILED,
                    "error_type" => error_type::COMMAND_FAILED,
                    "stage" => error_stage::RECEIVING,
                )
            } else {
                counter!(
                    "component_errors_total",
                    "error_code" => DELETION_FAILED,
                    "error_type" => error_type::COMMAND_FAILED,
                    "stage" => error_stage::RECEIVING,
                )
            }
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileDeleted<'a> {
        pub file: &'a Path,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileDeleted<'_> {
        fn emit(self) {
            info!(
                message = "File deleted.",
                file = %self.file.display(),
            );
            if self.include_file_metric_tag {
                counter!(
                    "files_deleted_total",
                    "file" => self.file.to_string_lossy().into_owned(),
                )
            } else {
                counter!("files_deleted_total")
            }
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileUnwatched<'a> {
        pub file: &'a Path,
        pub include_file_metric_tag: bool,
        pub reached_eof: bool,
    }

    impl InternalEvent for FileUnwatched<'_> {
        fn emit(self) {
            let reached_eof = if self.reached_eof { "true" } else { "false" };
            info!(
                message = "Stopped watching file.",
                file = %self.file.display(),
                reached_eof
            );
            if self.include_file_metric_tag {
                counter!(
                    "files_unwatched_total",
                    "file" => self.file.to_string_lossy().into_owned(),
                    "reached_eof" => reached_eof,
                )
            } else {
                counter!(
                    "files_unwatched_total",
                    "reached_eof" => reached_eof,
                )
            }
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    struct FileWatchError<'a> {
        pub file: &'a Path,
        pub error: Error,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileWatchError<'_> {
        fn emit(self) {
            error!(
                message = "Failed to watch file.",
                error = %self.error,
                error_code = "watching",
                error_type = error_type::COMMAND_FAILED,
                stage = error_stage::RECEIVING,
                file = %self.file.display(),
            );
            if self.include_file_metric_tag {
                counter!(
                    "component_errors_total",
                    "error_code" => "watching",
                    "error_type" => error_type::COMMAND_FAILED,
                    "stage" => error_stage::RECEIVING,
                    "file" => self.file.to_string_lossy().into_owned(),
                )
            } else {
                counter!(
                    "component_errors_total",
                    "error_code" => "watching",
                    "error_type" => error_type::COMMAND_FAILED,
                    "stage" => error_stage::RECEIVING,
                )
            }
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileResumed<'a> {
        pub file: &'a Path,
        pub file_position: u64,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileResumed<'_> {
        fn emit(self) {
            info!(
                message = "Resuming to watch file.",
                file = %self.file.display(),
                file_position = %self.file_position
            );
            if self.include_file_metric_tag {
                counter!(
                    "files_resumed_total",
                    "file" => self.file.to_string_lossy().into_owned(),
                )
            } else {
                counter!("files_resumed_total")
            }
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileAdded<'a> {
        pub file: &'a Path,
        pub include_file_metric_tag: bool,
    }

    impl InternalEvent for FileAdded<'_> {
        fn emit(self) {
            info!(
                message = "Found new file to watch.",
                file = %self.file.display(),
            );
            if self.include_file_metric_tag {
                counter!(
                    "files_added_total",
                    "file" => self.file.to_string_lossy().into_owned(),
                )
            } else {
                counter!("files_added_total")
            }
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileCheckpointed {
        pub count: usize,
        pub duration: Duration,
    }

    impl InternalEvent for FileCheckpointed {
        fn emit(self) {
            debug!(
                message = "Files checkpointed.",
                count = %self.count,
                duration_ms = self.duration.as_millis() as u64,
            );
            counter!("checkpoints_total").increment(self.count as u64);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileCheckpointWriteError {
        pub error: Error,
    }

    impl InternalEvent for FileCheckpointWriteError {
        fn emit(self) {
            error!(
                message = "Failed writing checkpoints.",
                error = %self.error,
                error_code = "writing_checkpoints",
                error_type = error_type::WRITER_FAILED,
                stage = error_stage::RECEIVING,
            );
            counter!(
                "component_errors_total",
                "error_code" => "writing_checkpoints",
                "error_type" => error_type::WRITER_FAILED,
                "stage" => error_stage::RECEIVING,
            )
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct PathGlobbingError<'a> {
        pub path: &'a Path,
        pub error: &'a Error,
    }

    impl InternalEvent for PathGlobbingError<'_> {
        fn emit(self) {
            error!(
                message = "Failed to glob path.",
                error = %self.error,
                error_code = "globbing",
                error_type = error_type::READER_FAILED,
                stage = error_stage::RECEIVING,
                path = %self.path.display(),
            );
            counter!(
                "component_errors_total",
                "error_code" => "globbing",
                "error_type" => error_type::READER_FAILED,
                "stage" => error_stage::RECEIVING,
            )
            .increment(1);
        }
    }

    #[derive(Debug, NamedInternalEvent)]
    pub struct FileLineTooBigError<'a> {
        pub truncated_bytes: &'a BytesMut,
        pub configured_limit: usize,
        pub encountered_size_so_far: usize,
    }

    impl InternalEvent for FileLineTooBigError<'_> {
        fn emit(self) {
            error!(
                message = "Found line that exceeds max_line_bytes; discarding.",
                truncated_bytes = ?self.truncated_bytes,
                configured_limit = self.configured_limit,
                encountered_size_so_far = self.encountered_size_so_far,
                error_type = error_type::CONDITION_FAILED,
                stage = error_stage::RECEIVING,
            );
            counter!(
                "component_errors_total",
                "error_code" => "reading_line_from_file",
                "error_type" => error_type::CONDITION_FAILED,
                "stage" => error_stage::RECEIVING,
            )
            .increment(1);
            emit!(ComponentEventsDropped::<INTENTIONAL> {
                count: 1,
                reason: "Found line that exceeds max_line_bytes; discarding.",
            });
        }
    }

    #[derive(Clone)]
    pub struct FileSourceInternalEventsEmitter {
        pub include_file_metric_tag: bool,
    }

    impl FileSourceInternalEvents for FileSourceInternalEventsEmitter {
        fn emit_file_added(&self, file: &Path) {
            emit!(FileAdded {
                file,
                include_file_metric_tag: self.include_file_metric_tag
            });
        }

        fn emit_file_resumed(&self, file: &Path, file_position: u64) {
            emit!(FileResumed {
                file,
                file_position,
                include_file_metric_tag: self.include_file_metric_tag
            });
        }

        fn emit_file_watch_error(&self, file: &Path, error: Error) {
            emit!(FileWatchError {
                file,
                error,
                include_file_metric_tag: self.include_file_metric_tag
            });
        }

        fn emit_file_unwatched(&self, file: &Path, reached_eof: bool) {
            emit!(FileUnwatched {
                file,
                include_file_metric_tag: self.include_file_metric_tag,
                reached_eof
            });
        }

        fn emit_file_deleted(&self, file: &Path) {
            emit!(FileDeleted {
                file,
                include_file_metric_tag: self.include_file_metric_tag
            });
        }

        fn emit_file_delete_error(&self, file: &Path, error: Error) {
            emit!(FileDeleteError {
                file,
                error,
                include_file_metric_tag: self.include_file_metric_tag
            });
        }

        fn emit_file_fingerprint_read_error(&self, file: &Path, error: Error) {
            emit!(FileFingerprintReadError {
                file,
                error,
                include_file_metric_tag: self.include_file_metric_tag
            });
        }

        fn emit_file_checksum_failed(&self, file: &Path) {
            emit!(FileChecksumFailed {
                file,
                include_file_metric_tag: self.include_file_metric_tag
            });
        }

        fn emit_file_checkpointed(&self, count: usize, duration: Duration) {
            emit!(FileCheckpointed { count, duration });
        }

        fn emit_file_checkpoint_write_error(&self, error: Error) {
            emit!(FileCheckpointWriteError { error });
        }

        fn emit_files_open(&self, count: usize) {
            emit!(FileOpen { count });
        }

        fn emit_path_globbing_failed(&self, path: &Path, error: &Error) {
            emit!(PathGlobbingError { path, error });
        }

        fn emit_file_line_too_long(
            &self,
            truncated_bytes: &bytes::BytesMut,
            configured_limit: usize,
            encountered_size_so_far: usize,
        ) {
            emit!(FileLineTooBigError {
                truncated_bytes,
                configured_limit,
                encountered_size_so_far
            });
        }
    }
}