processor.rs

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

//! Set of traits and structures used to implement processors.
//!
//! A processor is a node in the pipeline that transforms, filters, or otherwise processes messages
//! as they flow through the pipeline. Processors can perform operations such as:
//!
//! 1. Filtering messages based on certain criteria
//! 2. Transforming message content or format
//! 3. Aggregating multiple messages into a single message
//! 4. Splitting a single message into multiple messages
//! 5. Adding or removing attributes from messages
//!
//! # Lifecycle
//!
//! 1. The processor is instantiated and configured
//! 2. The processor receives and processes both data messages and control messages
//! 3. For each message, the processor can transform it, filter it, or split it into multiple messages
//! 4. The processor can maintain state between processing calls if needed
//! 5. The processor responds to control messages such as Config, TimerTick, or Shutdown
//! 6. The processor shuts down when it receives a `Shutdown` control message or encounters a fatal error
//!
//! # Thread Safety
//!
//! This implementation is designed to be used in a single-threaded environment.
//! The `Processor` trait does not require the `Send` bound, allowing for the use of non-thread-safe
//! types.
//!
//! # Scalability
//!
//! To ensure scalability, the pipeline engine will start multiple instances of the same pipeline
//! in parallel on different cores, each with its own processor instance.

use crate::control::{AckMsg, NackMsg};
use crate::effect_handler::{EffectHandlerCore, TelemetryTimerCancelHandle, TimerCancelHandle};
use crate::error::{Error, ProcessorErrorKind, TypedError};
use crate::local::message::LocalSender;
use crate::message::Message;
use crate::node::NodeId;
use async_trait::async_trait;
use otap_df_config::PortName;
use otap_df_telemetry::error::Error as TelemetryError;
use otap_df_telemetry::metrics::{MetricSet, MetricSetHandler};
use otap_df_telemetry::reporter::MetricsReporter;
use std::collections::HashMap;
use std::time::{Duration, Instant};

/// A trait for processors in the pipeline (!Send definition).
#[async_trait(?Send)]
pub trait Processor<PData> {
    /// Processes a message and optionally produces effects, such as generating new pdata messages.
    ///
    /// This method is called by the pipeline engine for each message that arrives at the processor.
    /// Unlike receivers, processors have known inputs (messages from previous stages), so the pipeline
    /// engine can control when to call this method and when the processor executes.
    ///
    /// This approach allows for greater flexibility and optimization, giving the pipeline engine
    /// the ability to decide whether to spawn one task per processor or one task for a group of processors.
    /// The method signature uses `&mut self` rather than `Box<Self>` because the engine only wants to
    /// temporarily allow mutation of the processor instance, not transfer ownership.
    ///
    /// The processor can:
    /// - Transform the message and return a new message
    /// - Filter the message by returning None
    /// - Split the message into multiple messages by returning a vector
    /// - Handle control messages (e.g., Config, TimerTick, Shutdown)
    ///
    /// # Parameters
    ///
    /// - `msg`: The message to process, which can be either a data message or a control message
    /// - `effect_handler`: A handler to perform side effects such as sending messages to the next node.
    ///    This can be either Send or !Send depending on the processor's Mode type.
    ///
    /// # Returns
    ///
    /// - `Ok(())`: The processor successfully processed the message
    /// - `Err(Error)`: The processor encountered an error and could not process the message
    ///
    /// # Errors
    ///
    /// Returns an [`Error`] if the processor encounters an unrecoverable error.
    async fn process(
        &mut self,
        msg: Message<PData>,
        effect_handler: &mut EffectHandler<PData>,
    ) -> Result<(), Error>;
}

/// A `!Send` implementation of the EffectHandler.
#[derive(Clone)]
pub struct EffectHandler<PData> {
    pub(crate) core: EffectHandlerCore<PData>,

    /// A sender used to forward messages from the processor.
    /// Supports multiple named output ports.
    msg_senders: HashMap<PortName, LocalSender<PData>>,
    /// Cached default sender for fast access in the hot path
    default_sender: Option<LocalSender<PData>>,
    
    /// Local telemetry state for component-level logging.
    /// Stored directly in EffectHandler (not in EffectHandlerCore) because
    /// it contains Rc<RefCell<>> which would break Send bounds for shared handlers.
    telemetry_state: Option<std::rc::Rc<std::cell::RefCell<crate::effect_handler::LocalTelemetryState>>>,
}

/// Implementation for the `!Send` effect handler.
impl<PData> EffectHandler<PData> {
    /// Creates a new local (!Send) `EffectHandler` with the given processor name.
    #[must_use]
    pub fn new(
        node_id: NodeId,
        msg_senders: HashMap<PortName, LocalSender<PData>>,
        default_port: Option<PortName>,
        metrics_reporter: MetricsReporter,
    ) -> Self {
        let core = EffectHandlerCore::new(node_id, metrics_reporter);

        // Determine and cache the default sender
        let default_sender = if let Some(ref port) = default_port {
            msg_senders.get(port).cloned()
        } else if msg_senders.len() == 1 {
            msg_senders.values().next().cloned()
        } else {
            None
        };

        EffectHandler {
            core,
            msg_senders,
            default_sender,
            telemetry_state: None,
        }
    }

    /// Returns the id of the processor associated with this handler.
    #[must_use]
    pub fn processor_id(&self) -> NodeId {
        self.core.node_id()
    }

    /// Returns the list of connected out ports for this processor.
    #[must_use]
    pub fn connected_ports(&self) -> Vec<PortName> {
        self.msg_senders.keys().cloned().collect()
    }

    /// Sends a message to the next node(s) in the pipeline using the default port.
    ///
    /// If a default port is configured (either explicitly or deduced when a single port is
    /// connected), it will be used. Otherwise, an error is returned.
    ///
    /// # Errors
    ///
    /// Returns an [`Error::ChannelSendError`] if the message could not be sent or [`Error::ProcessorError`]
    /// if the default port is not configured.
    #[inline]
    pub async fn send_message(&self, data: PData) -> Result<(), TypedError<PData>> {
        match &self.default_sender {
            Some(sender) => sender
                .send(data)
                .await
                .map_err(TypedError::ChannelSendError),
            None => Err(TypedError::Error(Error::ProcessorError {
                processor: self.processor_id(),
                kind: ProcessorErrorKind::Configuration,
                error:
                    "Ambiguous default out port: multiple ports connected and no default configured"
                        .to_string(),
                source_detail: String::new(),
            })),
        }
    }

    /// Sends a message to a specific named out port.
    ///
    /// # Errors
    ///
    /// Returns an [`Error::ChannelSendError`] if the message could not be sent, or
    /// [`Error::ProcessorError`] if the port does not exist.
    #[inline]
    pub async fn send_message_to<P>(&self, port: P, data: PData) -> Result<(), TypedError<PData>>
    where
        P: Into<PortName>,
    {
        let port_name: PortName = port.into();
        match self.msg_senders.get(&port_name) {
            Some(sender) => sender
                .send(data)
                .await
                .map_err(TypedError::ChannelSendError),
            None => Err(TypedError::Error(Error::ProcessorError {
                processor: self.processor_id(),
                kind: ProcessorErrorKind::Configuration,
                error: format!(
                    "Unknown out port '{port_name}' for node {}",
                    self.processor_id()
                ),
                source_detail: String::new(),
            })),
        }
    }

    /// Print an info message to stdout.
    ///
    /// This method provides a standardized way for processors to output
    /// informational messages without blocking the async runtime.
    pub async fn info(&self, message: &str) {
        self.core.info(message).await;
    }

    /// Starts a cancellable periodic timer that emits TimerTick on the control channel.
    /// Returns a handle that can be used to cancel the timer.
    ///
    /// Current limitation: Only one timer can be started by an exporter at a time.
    pub async fn start_periodic_timer(
        &self,
        duration: Duration,
    ) -> Result<TimerCancelHandle<PData>, Error> {
        self.core.start_periodic_timer(duration).await
    }

    /// Starts a cancellable periodic telemetry timer that emits CollectTelemetry.
    pub async fn start_periodic_telemetry(
        &self,
        duration: Duration,
    ) -> Result<TelemetryTimerCancelHandle<PData>, Error> {
        self.core.start_periodic_telemetry(duration).await
    }

    /// Send an Ack to a node of known-interest.
    pub async fn route_ack<F>(&self, ack: AckMsg<PData>, cxf: F) -> Result<(), Error>
    where
        F: FnOnce(AckMsg<PData>) -> Option<(usize, AckMsg<PData>)>,
    {
        self.core.route_ack(ack, cxf).await
    }

    /// Send a Nack to a node of known-interest.
    pub async fn route_nack<F>(&self, nack: NackMsg<PData>, cxf: F) -> Result<(), Error>
    where
        F: FnOnce(NackMsg<PData>) -> Option<(usize, NackMsg<PData>)>,
    {
        self.core.route_nack(nack, cxf).await
    }

    /// Delay data.
    pub async fn delay_data(&self, when: Instant, data: Box<PData>) -> Result<(), PData> {
        self.core.delay_data(when, data).await
    }

    /// Reports metrics collected by the processor.
    #[allow(dead_code)] // Will be used in the future. ToDo report metrics from channel and messages.
    pub(crate) fn report_metrics<M: MetricSetHandler + 'static>(
        &mut self,
        metrics: &mut MetricSet<M>,
    ) -> Result<(), TelemetryError> {
        self.core.report_metrics(metrics)
    }

    /// Log a telemetry event from this component.
    ///
    /// This method encodes the log record into the component's telemetry buffer.
    /// On overflow (buffer exceeds threshold), it automatically flushes to the
    /// configured destination (ITR or SDK fallback) via non-blocking channel send.
    ///
    /// This method never fails - errors are silently dropped to prevent recursion.
    /// If the telemetry buffer is not configured, this is a no-op.
    pub fn log_event(&self, log_record: &impl otap_df_pdata::views::logs::LogRecordView) {
        if let Some(ref buffer) = self.telemetry_state {
            let mut state = buffer.borrow_mut();
            
            // Clone necessary data to avoid holding immutable borrows
            let resource_bytes = state.resource_bytes.clone();
            let scope_name = state.scope_name.clone();
            let threshold = state.flush_threshold_bytes;
            
            // Encode the log record (silent drop on error to prevent recursion)
            if state.encoder.encode_log_record(
                log_record,
                resource_bytes.as_bytes(),
                &scope_name,
            ).is_err() {
                // TODO: Increment dropped events counter
                return;
            }
            
            // Check overflow - non-blocking send to break recursion
            if state.encoder.len() >= threshold {
                let bytes = state.encoder.flush();
                // Non-blocking send - errors are silently dropped
                // Destination is either ITR or SDK fallback (OtlpBytesChannel)
                let _ = state.overflow_sender.send(bytes);
            }
        }
    }

    // More methods will be added in the future as needed.
}

#[cfg(test)]
mod tests {
    #![allow(missing_docs)]
    use super::*;
    use crate::local::message::LocalSender;
    use crate::testing::test_node;
    use otap_df_channel::mpsc;
    use std::borrow::Cow;
    use std::collections::{HashMap, HashSet};
    use tokio::time::{Duration, timeout};

    fn channel<T>(capacity: usize) -> (mpsc::Sender<T>, mpsc::Receiver<T>) {
        mpsc::Channel::new(capacity)
    }

    #[tokio::test]
    async fn effect_handler_send_message_to_named_port() {
        let (a_tx, a_rx) = channel::<u64>(10);
        let (b_tx, b_rx) = channel::<u64>(10);

        let mut senders = HashMap::new();
        let _ = senders.insert("a".into(), LocalSender::MpscSender(a_tx));
        let _ = senders.insert("b".into(), LocalSender::MpscSender(b_tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(test_node("proc"), senders, None, metrics_reporter);
        eh.send_message_to("b", 42).await.unwrap();

        // Ensure only 'b' received
        assert!(
            timeout(Duration::from_millis(50), a_rx.recv())
                .await
                .is_err()
        );
        assert_eq!(b_rx.recv().await.unwrap(), 42);
    }

    #[tokio::test]
    async fn effect_handler_send_message_single_port_fallback() {
        let (tx, rx) = channel::<u64>(10);
        let mut senders = HashMap::new();
        let _ = senders.insert("only".into(), LocalSender::MpscSender(tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(test_node("proc"), senders, None, metrics_reporter);

        eh.send_message(7).await.unwrap();
        assert_eq!(rx.recv().await.unwrap(), 7);
    }

    #[tokio::test]
    async fn effect_handler_send_message_uses_default_port() {
        let (a_tx, a_rx) = channel::<u64>(10);
        let (b_tx, b_rx) = channel::<u64>(10);

        let mut senders = HashMap::new();
        let _ = senders.insert("a".into(), LocalSender::MpscSender(a_tx));
        let _ = senders.insert("b".into(), LocalSender::MpscSender(b_tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(
            test_node("proc"),
            senders,
            Some("a".into()),
            metrics_reporter,
        );

        eh.send_message(11).await.unwrap();

        assert_eq!(a_rx.recv().await.unwrap(), 11);
        assert!(
            timeout(Duration::from_millis(50), b_rx.recv())
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn effect_handler_send_message_ambiguous_without_default() {
        let (a_tx, a_rx) = channel::<u64>(10);
        let (b_tx, b_rx) = channel::<u64>(10);

        let mut senders = HashMap::new();
        let _ = senders.insert("a".into(), LocalSender::MpscSender(a_tx));
        let _ = senders.insert("b".into(), LocalSender::MpscSender(b_tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(test_node("proc"), senders, None, metrics_reporter);

        let res = eh.send_message(5).await;
        assert!(res.is_err());

        // Nothing should be received on either port
        assert!(
            timeout(Duration::from_millis(50), a_rx.recv())
                .await
                .is_err()
        );
        assert!(
            timeout(Duration::from_millis(50), b_rx.recv())
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn effect_handler_connected_ports_lists_all() {
        let (a_tx, _a_rx) = channel::<u64>(1);
        let (b_tx, _b_rx) = channel::<u64>(1);

        let mut senders = HashMap::new();
        let _ = senders.insert("a".into(), LocalSender::MpscSender(a_tx));
        let _ = senders.insert("b".into(), LocalSender::MpscSender(b_tx));

        let (_metrics_rx, metrics_reporter) = MetricsReporter::create_new_and_receiver(1);
        let eh = EffectHandler::new(test_node("proc"), senders, None, metrics_reporter);

        let ports: HashSet<_> = eh.connected_ports().into_iter().collect();
        let expected: HashSet<_> = [Cow::from("a"), Cow::from("b")].into_iter().collect();
        assert_eq!(ports, expected);
    }
}