kubernetes_logs.rs

use metrics::counter;
use vector_lib::{
    NamedInternalEvent,
    internal_event::{
        ComponentEventsDropped, INTENTIONAL, InternalEvent, UNINTENTIONAL, error_stage, error_type,
    },
    json_size::JsonSize,
};
use vrl::core::Value;

use crate::event::Event;

#[derive(Debug, NamedInternalEvent)]
pub struct KubernetesLogsEventsReceived<'a> {
    pub file: &'a str,
    pub byte_size: JsonSize,
    pub pod_info: Option<KubernetesLogsPodInfo>,
}

#[derive(Debug)]
pub struct KubernetesLogsPodInfo {
    pub name: String,
    pub namespace: String,
}

impl InternalEvent for KubernetesLogsEventsReceived<'_> {
    fn emit(self) {
        trace!(
            message = "Events received.",
            count = 1,
            byte_size = %self.byte_size,
            file = %self.file,
        );
        match self.pod_info {
            Some(pod_info) => {
                let pod_name = pod_info.name;
                let pod_namespace = pod_info.namespace;

                counter!(
                    "component_received_events_total",
                    "pod_name" => pod_name.clone(),
                    "pod_namespace" => pod_namespace.clone(),
                )
                .increment(1);
                counter!(
                    "component_received_event_bytes_total",
                    "pod_name" => pod_name,
                    "pod_namespace" => pod_namespace,
                )
                .increment(self.byte_size.get() as u64);
            }
            None => {
                counter!("component_received_events_total").increment(1);
                counter!("component_received_event_bytes_total")
                    .increment(self.byte_size.get() as u64);
            }
        }
    }
}

const ANNOTATION_FAILED: &str = "annotation_failed";

#[derive(Debug, NamedInternalEvent)]
pub struct KubernetesLogsEventAnnotationError<'a> {
    pub event: &'a Event,
}

impl InternalEvent for KubernetesLogsEventAnnotationError<'_> {
    fn emit(self) {
        error!(
            message = "Failed to annotate event with pod metadata.",
            event = ?self.event,
            error_code = ANNOTATION_FAILED,
            error_type = error_type::READER_FAILED,
            stage = error_stage::PROCESSING,
        );
        counter!(
            "component_errors_total",
            "error_code" => ANNOTATION_FAILED,
            "error_type" => error_type::READER_FAILED,
            "stage" => error_stage::PROCESSING,
        )
        .increment(1);
    }
}

#[derive(Debug, NamedInternalEvent)]
pub(crate) struct KubernetesLogsEventNamespaceAnnotationError<'a> {
    pub event: &'a Event,
}

impl InternalEvent for KubernetesLogsEventNamespaceAnnotationError<'_> {
    fn emit(self) {
        error!(
            message = "Failed to annotate event with namespace metadata.",
            event = ?self.event,
            error_code = ANNOTATION_FAILED,
            error_type = error_type::READER_FAILED,
            stage = error_stage::PROCESSING,
        );
        counter!(
            "component_errors_total",
            "error_code" => ANNOTATION_FAILED,
            "error_type" => error_type::READER_FAILED,
            "stage" => error_stage::PROCESSING,
        )
        .increment(1);
        counter!("k8s_event_namespace_annotation_failures_total").increment(1);
    }
}

#[derive(Debug, NamedInternalEvent)]
pub(crate) struct KubernetesLogsEventNodeAnnotationError<'a> {
    pub event: &'a Event,
}

impl InternalEvent for KubernetesLogsEventNodeAnnotationError<'_> {
    fn emit(self) {
        error!(
            message = "Failed to annotate event with node metadata.",
            event = ?self.event,
            error_code = ANNOTATION_FAILED,
            error_type = error_type::READER_FAILED,
            stage = error_stage::PROCESSING,
        );
        counter!(
            "component_errors_total",
            "error_code" => ANNOTATION_FAILED,
            "error_type" => error_type::READER_FAILED,
            "stage" => error_stage::PROCESSING,
        )
        .increment(1);
        counter!("k8s_event_node_annotation_failures_total").increment(1);
    }
}

#[derive(Debug, NamedInternalEvent)]
pub struct KubernetesLogsFormatPickerEdgeCase {
    pub what: &'static str,
}

impl InternalEvent for KubernetesLogsFormatPickerEdgeCase {
    fn emit(self) {
        warn!(
            message = "Encountered format picker edge case.",
            what = %self.what,
        );
        counter!("k8s_format_picker_edge_cases_total").increment(1);
    }
}

#[derive(Debug, NamedInternalEvent)]
pub struct KubernetesLogsDockerFormatParseError<'a> {
    pub error: &'a dyn std::error::Error,
}

impl InternalEvent for KubernetesLogsDockerFormatParseError<'_> {
    fn emit(self) {
        error!(
            message = "Failed to parse log line in docker format.",
            error = %self.error,
            error_type = error_type::PARSER_FAILED,
            stage = error_stage::PROCESSING,
        );
        counter!(
            "component_errors_total",
            "error_type" => error_type::PARSER_FAILED,
            "stage" => error_stage::PROCESSING,
        )
        .increment(1);
        counter!("k8s_docker_format_parse_failures_total").increment(1);
    }
}

const KUBERNETES_LIFECYCLE: &str = "kubernetes_lifecycle";

#[derive(Debug, NamedInternalEvent)]
pub struct KubernetesLifecycleError<E> {
    pub message: &'static str,
    pub error: E,
    pub count: usize,
}

impl<E: std::fmt::Display> InternalEvent for KubernetesLifecycleError<E> {
    fn emit(self) {
        error!(
            message = self.message,
            error = %self.error,
            error_code = KUBERNETES_LIFECYCLE,
            error_type = error_type::READER_FAILED,
            stage = error_stage::PROCESSING,
        );
        counter!(
            "component_errors_total",
            "error_code" => KUBERNETES_LIFECYCLE,
            "error_type" => error_type::READER_FAILED,
            "stage" => error_stage::PROCESSING,
        )
        .increment(1);
        emit!(ComponentEventsDropped::<UNINTENTIONAL> {
            count: self.count,
            reason: self.message,
        });
    }
}

#[derive(Debug, NamedInternalEvent)]
pub struct KubernetesMergedLineTooBigError<'a> {
    pub event: &'a Value,
    pub configured_limit: usize,
    pub encountered_size_so_far: usize,
}

impl InternalEvent for KubernetesMergedLineTooBigError<'_> {
    fn emit(self) {
        error!(
            message = "Found line that exceeds max_merged_line_bytes; discarding.",
            event = ?self.event,
            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_kubernetes_log",
            "error_type" => error_type::CONDITION_FAILED,
            "stage" => error_stage::RECEIVING,
        )
        .increment(1);
        emit!(ComponentEventsDropped::<INTENTIONAL> {
            count: 1,
            reason: "Found line that exceeds max_merged_line_bytes; discarding.",
        });
    }
}