llm_optimizer_processor/
config.rs

1//! Configuration types for the stream processor
2//!
3//! This module provides configuration structures for all processor components
4//! including windows, watermarks, state backends, and aggregations.
5
6use crate::error::{ProcessorError, Result};
7use serde::{Deserialize, Serialize};
8use std::path::PathBuf;
9use std::time::Duration;
10
11/// Main processor configuration
12#[derive(Debug, Clone, Serialize, Deserialize)]
13pub struct ProcessorConfig {
14    /// Window configuration
15    pub window: WindowConfig,
16
17    /// Watermark configuration
18    pub watermark: WatermarkConfig,
19
20    /// State backend configuration
21    pub state: StateConfig,
22
23    /// Aggregation configuration
24    pub aggregation: AggregationConfig,
25
26    /// Deduplication configuration
27    #[serde(default)]
28    pub deduplication: DeduplicationConfig,
29
30    /// Time-series normalization configuration
31    #[serde(default)]
32    pub normalization: NormalizationConfig,
33
34    /// Number of parallel processing workers
35    #[serde(default = "default_parallelism")]
36    pub parallelism: usize,
37
38    /// Maximum events to buffer before backpressure
39    #[serde(default = "default_buffer_size")]
40    pub buffer_size: usize,
41
42    /// Enable metrics collection
43    #[serde(default = "default_true")]
44    pub metrics_enabled: bool,
45}
46
47impl Default for ProcessorConfig {
48    fn default() -> Self {
49        Self {
50            window: WindowConfig::default(),
51            watermark: WatermarkConfig::default(),
52            state: StateConfig::default(),
53            aggregation: AggregationConfig::default(),
54            deduplication: DeduplicationConfig::default(),
55            normalization: NormalizationConfig::default(),
56            parallelism: default_parallelism(),
57            buffer_size: default_buffer_size(),
58            metrics_enabled: true,
59        }
60    }
61}
62
63impl ProcessorConfig {
64    /// Validate the configuration
65    pub fn validate(&self) -> Result<()> {
66        self.window.validate()?;
67        self.watermark.validate()?;
68        self.state.validate()?;
69        self.aggregation.validate()?;
70        self.deduplication.validate()?;
71        self.normalization.validate()?;
72
73        if self.parallelism == 0 {
74            return Err(ProcessorError::Configuration {
75                source: "parallelism must be greater than 0".into(),
76            });
77        }
78
79        if self.buffer_size == 0 {
80            return Err(ProcessorError::Configuration {
81                source: "buffer_size must be greater than 0".into(),
82            });
83        }
84
85        Ok(())
86    }
87}
88
89/// Window configuration
90#[derive(Debug, Clone, Serialize, Deserialize)]
91pub struct WindowConfig {
92    /// Type of window
93    pub window_type: WindowType,
94
95    /// Window size in milliseconds (for tumbling/sliding/hopping windows)
96    pub size_ms: Option<u64>,
97
98    /// Slide interval in milliseconds (for sliding/hopping windows)
99    pub slide_ms: Option<u64>,
100
101    /// Session gap in milliseconds (for session windows)
102    pub gap_ms: Option<u64>,
103
104    /// Allowed lateness for late events (milliseconds)
105    #[serde(default = "default_allowed_lateness")]
106    pub allowed_lateness_ms: u64,
107
108    /// Whether to drop late events or process them
109    #[serde(default)]
110    pub drop_late_events: bool,
111}
112
113/// Type of windowing strategy
114#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
115#[serde(rename_all = "snake_case")]
116pub enum WindowType {
117    /// Fixed-size non-overlapping windows
118    Tumbling,
119    /// Fixed-size overlapping windows
120    Sliding,
121    /// Event-based windows with gaps
122    Session,
123    /// Count-based windows
124    Count,
125    /// Global window (all events in one window)
126    Global,
127}
128
129impl Default for WindowConfig {
130    fn default() -> Self {
131        Self {
132            window_type: WindowType::Tumbling,
133            size_ms: Some(60_000), // 1 minute
134            slide_ms: None,
135            gap_ms: None,
136            allowed_lateness_ms: default_allowed_lateness(),
137            drop_late_events: false,
138        }
139    }
140}
141
142impl WindowConfig {
143    /// Create a tumbling window configuration
144    pub fn tumbling(size_ms: u64) -> Self {
145        Self {
146            window_type: WindowType::Tumbling,
147            size_ms: Some(size_ms),
148            slide_ms: None,
149            gap_ms: None,
150            ..Default::default()
151        }
152    }
153
154    /// Create a sliding window configuration
155    pub fn sliding(size_ms: u64, slide_ms: u64) -> Self {
156        Self {
157            window_type: WindowType::Sliding,
158            size_ms: Some(size_ms),
159            slide_ms: Some(slide_ms),
160            gap_ms: None,
161            ..Default::default()
162        }
163    }
164
165    /// Create a session window configuration
166    pub fn session(gap_ms: u64) -> Self {
167        Self {
168            window_type: WindowType::Session,
169            size_ms: None,
170            slide_ms: None,
171            gap_ms: Some(gap_ms),
172            ..Default::default()
173        }
174    }
175
176    /// Validate window configuration
177    pub fn validate(&self) -> Result<()> {
178        match self.window_type {
179            WindowType::Tumbling => {
180                if let Some(size) = self.size_ms {
181                    if size == 0 {
182                        return Err(ProcessorError::Configuration {
183                            source: "tumbling window size must be greater than 0".into(),
184                        });
185                    }
186                } else {
187                    return Err(ProcessorError::Configuration {
188                        source: "tumbling window requires size_ms".into(),
189                    });
190                }
191            }
192            WindowType::Sliding => {
193                let size = self.size_ms.ok_or_else(|| ProcessorError::Configuration {
194                    source: "sliding window requires size_ms".into(),
195                })?;
196                let slide = self.slide_ms.ok_or_else(|| ProcessorError::Configuration {
197                    source: "sliding window requires slide_ms".into(),
198                })?;
199
200                if size == 0 {
201                    return Err(ProcessorError::Configuration {
202                        source: "sliding window size must be greater than 0".into(),
203                    });
204                }
205                if slide == 0 {
206                    return Err(ProcessorError::Configuration {
207                        source: "sliding window slide must be greater than 0".into(),
208                    });
209                }
210                if slide > size {
211                    return Err(ProcessorError::Configuration {
212                        source: "sliding window slide must be less than or equal to size".into(),
213                    });
214                }
215            }
216            WindowType::Session => {
217                if let Some(gap) = self.gap_ms {
218                    if gap == 0 {
219                        return Err(ProcessorError::Configuration {
220                            source: "session window gap must be greater than 0".into(),
221                        });
222                    }
223                } else {
224                    return Err(ProcessorError::Configuration {
225                        source: "session window requires gap_ms".into(),
226                    });
227                }
228            }
229            WindowType::Count | WindowType::Global => {
230                // No specific validation needed
231            }
232        }
233
234        Ok(())
235    }
236}
237
238/// Watermark configuration
239#[derive(Debug, Clone, Serialize, Deserialize)]
240pub struct WatermarkConfig {
241    /// Watermark generation strategy
242    pub strategy: WatermarkStrategy,
243
244    /// Maximum out-of-orderness delay (milliseconds)
245    #[serde(default = "default_max_delay")]
246    pub max_delay_ms: u64,
247
248    /// Idle source timeout (milliseconds)
249    #[serde(default = "default_idle_timeout")]
250    pub idle_timeout_ms: u64,
251
252    /// Watermark update interval (milliseconds)
253    #[serde(default = "default_update_interval")]
254    pub update_interval_ms: u64,
255
256    /// Whether to allow clock skew
257    #[serde(default = "default_true")]
258    pub allow_clock_skew: bool,
259
260    /// Maximum allowed clock skew (milliseconds)
261    #[serde(default = "default_max_clock_skew")]
262    pub max_clock_skew_ms: u64,
263}
264
265/// Watermark generation strategy
266#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
267#[serde(rename_all = "snake_case")]
268pub enum WatermarkStrategy {
269    /// Periodic watermark based on processing time
270    Periodic,
271    /// Bounded out-of-orderness based on event time
272    BoundedOutOfOrderness,
273    /// Ascending timestamps (no out-of-order events)
274    Ascending,
275    /// Manual watermark control
276    Manual,
277}
278
279impl Default for WatermarkConfig {
280    fn default() -> Self {
281        Self {
282            strategy: WatermarkStrategy::BoundedOutOfOrderness,
283            max_delay_ms: default_max_delay(),
284            idle_timeout_ms: default_idle_timeout(),
285            update_interval_ms: default_update_interval(),
286            allow_clock_skew: true,
287            max_clock_skew_ms: default_max_clock_skew(),
288        }
289    }
290}
291
292impl WatermarkConfig {
293    /// Validate watermark configuration
294    pub fn validate(&self) -> Result<()> {
295        if self.update_interval_ms == 0 {
296            return Err(ProcessorError::Configuration {
297                source: "watermark update_interval_ms must be greater than 0".into(),
298            });
299        }
300
301        if self.idle_timeout_ms == 0 {
302            return Err(ProcessorError::Configuration {
303                source: "watermark idle_timeout_ms must be greater than 0".into(),
304            });
305        }
306
307        Ok(())
308    }
309
310    /// Get max delay as Duration
311    pub fn max_delay(&self) -> Duration {
312        Duration::from_millis(self.max_delay_ms)
313    }
314
315    /// Get idle timeout as Duration
316    pub fn idle_timeout(&self) -> Duration {
317        Duration::from_millis(self.idle_timeout_ms)
318    }
319
320    /// Get update interval as Duration
321    pub fn update_interval(&self) -> Duration {
322        Duration::from_millis(self.update_interval_ms)
323    }
324}
325
326/// State backend configuration
327#[derive(Debug, Clone, Serialize, Deserialize)]
328pub struct StateConfig {
329    /// State backend type
330    pub backend: StateBackend,
331
332    /// State storage path (for file-based backends)
333    pub storage_path: Option<PathBuf>,
334
335    /// Checkpoint interval (milliseconds)
336    #[serde(default = "default_checkpoint_interval")]
337    pub checkpoint_interval_ms: u64,
338
339    /// Enable incremental checkpoints
340    #[serde(default = "default_true")]
341    pub incremental_checkpoints: bool,
342
343    /// Number of checkpoints to retain
344    #[serde(default = "default_checkpoint_retention")]
345    pub checkpoint_retention: usize,
346
347    /// State TTL (time-to-live) in milliseconds
348    pub ttl_ms: Option<u64>,
349
350    /// Maximum state size in bytes (for memory backends)
351    pub max_size_bytes: Option<u64>,
352
353    /// Enable state compression
354    #[serde(default)]
355    pub compression_enabled: bool,
356}
357
358/// State backend type
359#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
360#[serde(rename_all = "snake_case")]
361pub enum StateBackend {
362    /// In-memory state (not fault-tolerant)
363    Memory,
364    /// File-based state using RocksDB/Sled
365    FileSystem,
366    /// Redis-backed state
367    Redis,
368    /// PostgreSQL-backed state
369    PostgreSQL,
370}
371
372impl Default for StateConfig {
373    fn default() -> Self {
374        Self {
375            backend: StateBackend::Memory,
376            storage_path: None,
377            checkpoint_interval_ms: default_checkpoint_interval(),
378            incremental_checkpoints: true,
379            checkpoint_retention: default_checkpoint_retention(),
380            ttl_ms: None,
381            max_size_bytes: None,
382            compression_enabled: false,
383        }
384    }
385}
386
387impl StateConfig {
388    /// Validate state configuration
389    pub fn validate(&self) -> Result<()> {
390        if self.checkpoint_interval_ms == 0 {
391            return Err(ProcessorError::Configuration {
392                source: "checkpoint_interval_ms must be greater than 0".into(),
393            });
394        }
395
396        if self.checkpoint_retention == 0 {
397            return Err(ProcessorError::Configuration {
398                source: "checkpoint_retention must be greater than 0".into(),
399            });
400        }
401
402        if matches!(self.backend, StateBackend::FileSystem) && self.storage_path.is_none() {
403            return Err(ProcessorError::Configuration {
404                source: "file-based state backend requires storage_path".into(),
405            });
406        }
407
408        Ok(())
409    }
410
411    /// Get checkpoint interval as Duration
412    pub fn checkpoint_interval(&self) -> Duration {
413        Duration::from_millis(self.checkpoint_interval_ms)
414    }
415
416    /// Get TTL as Duration
417    pub fn ttl(&self) -> Option<Duration> {
418        self.ttl_ms.map(Duration::from_millis)
419    }
420}
421
422/// Aggregation configuration
423#[derive(Debug, Clone, Serialize, Deserialize)]
424pub struct AggregationConfig {
425    /// Enabled aggregation types
426    pub enabled_aggregations: Vec<AggregationType>,
427
428    /// Percentiles to compute (e.g., [50, 95, 99])
429    #[serde(default = "default_percentiles")]
430    pub percentiles: Vec<u8>,
431
432    /// Histogram bucket configuration
433    pub histogram_buckets: Option<Vec<f64>>,
434
435    /// Enable incremental aggregation
436    #[serde(default = "default_true")]
437    pub incremental: bool,
438
439    /// Minimum samples required for aggregation
440    #[serde(default = "default_min_samples")]
441    pub min_samples: usize,
442
443    /// Maximum samples to keep in memory
444    #[serde(default = "default_max_samples")]
445    pub max_samples: usize,
446
447    /// Enable outlier detection and removal
448    #[serde(default)]
449    pub remove_outliers: bool,
450
451    /// Outlier threshold (standard deviations)
452    #[serde(default = "default_outlier_threshold")]
453    pub outlier_threshold: f64,
454}
455
456/// Types of aggregations to compute
457#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
458#[serde(rename_all = "snake_case")]
459pub enum AggregationType {
460    /// Sum of all values
461    Sum,
462    /// Count of values
463    Count,
464    /// Arithmetic mean
465    Average,
466    /// Minimum value
467    Min,
468    /// Maximum value
469    Max,
470    /// Percentiles
471    Percentile,
472    /// Histogram
473    Histogram,
474    /// Standard deviation
475    StdDev,
476    /// Variance
477    Variance,
478}
479
480impl Default for AggregationConfig {
481    fn default() -> Self {
482        Self {
483            enabled_aggregations: vec![
484                AggregationType::Sum,
485                AggregationType::Count,
486                AggregationType::Average,
487                AggregationType::Min,
488                AggregationType::Max,
489                AggregationType::Percentile,
490            ],
491            percentiles: default_percentiles(),
492            histogram_buckets: None,
493            incremental: true,
494            min_samples: default_min_samples(),
495            max_samples: default_max_samples(),
496            remove_outliers: false,
497            outlier_threshold: default_outlier_threshold(),
498        }
499    }
500}
501
502impl AggregationConfig {
503    /// Validate aggregation configuration
504    pub fn validate(&self) -> Result<()> {
505        if self.enabled_aggregations.is_empty() {
506            return Err(ProcessorError::Configuration {
507                source: "at least one aggregation type must be enabled".into(),
508            });
509        }
510
511        for &p in &self.percentiles {
512            if p > 100 {
513                return Err(ProcessorError::Configuration {
514                    source: format!("invalid percentile: {}, must be 0-100", p).into(),
515                });
516            }
517        }
518
519        if let Some(buckets) = &self.histogram_buckets {
520            if buckets.is_empty() {
521                return Err(ProcessorError::Configuration {
522                    source: "histogram_buckets cannot be empty".into(),
523                });
524            }
525
526            // Check if buckets are sorted
527            for i in 1..buckets.len() {
528                if buckets[i] <= buckets[i - 1] {
529                    return Err(ProcessorError::Configuration {
530                        source: "histogram_buckets must be in ascending order".into(),
531                    });
532                }
533            }
534        }
535
536        if self.min_samples > self.max_samples {
537            return Err(ProcessorError::Configuration {
538                source: "min_samples cannot be greater than max_samples".into(),
539            });
540        }
541
542        if self.outlier_threshold <= 0.0 {
543            return Err(ProcessorError::Configuration {
544                source: "outlier_threshold must be positive".into(),
545            });
546        }
547
548        Ok(())
549    }
550
551    /// Check if a specific aggregation type is enabled
552    pub fn is_enabled(&self, agg_type: AggregationType) -> bool {
553        self.enabled_aggregations.contains(&agg_type)
554    }
555}
556
557/// Time-series normalization configuration
558///
559/// Provides configuration for normalizing time-series data by aligning events to fixed
560/// intervals, filling gaps, and handling out-of-order events.
561///
562/// Normalization is applied before windowing to ensure consistent event time alignment
563/// and reduce skew in aggregations.
564///
565/// # Example
566///
567/// ```rust
568/// use processor::config::{NormalizationConfig, FillStrategy, AlignmentStrategy};
569/// use std::time::Duration;
570///
571/// let config = NormalizationConfig::new()
572///     .enabled()
573///     .with_interval(Duration::from_secs(5))
574///     .with_fill_strategy(FillStrategy::LinearInterpolation)
575///     .with_alignment(AlignmentStrategy::Start)
576///     .with_buffer_size(1000);
577/// ```
578#[derive(Debug, Clone, Serialize, Deserialize)]
579pub struct NormalizationConfig {
580    /// Enable time-series normalization
581    #[serde(default)]
582    pub enabled: bool,
583
584    /// Fixed interval for time alignment (e.g., 1s, 5s, 1min)
585    #[serde(
586        default = "default_normalization_interval",
587        serialize_with = "serialize_duration",
588        deserialize_with = "deserialize_duration"
589    )]
590    pub interval: Duration,
591
592    /// Strategy for filling missing data points
593    #[serde(default)]
594    pub fill_strategy: FillStrategy,
595
596    /// Alignment strategy for timestamps (start, end, or center of interval)
597    #[serde(default)]
598    pub alignment: AlignmentStrategy,
599
600    /// Buffer size for handling out-of-order events
601    #[serde(default = "default_normalization_buffer_size")]
602    pub buffer_size: usize,
603
604    /// Maximum gap to interpolate (larger gaps use fill strategy)
605    #[serde(
606        default = "default_interpolation_max_gap",
607        serialize_with = "serialize_duration",
608        deserialize_with = "deserialize_duration"
609    )]
610    pub interpolation_max_gap: Duration,
611
612    /// Enable outlier detection and removal
613    #[serde(default)]
614    pub drop_outliers: bool,
615
616    /// Outlier threshold in standard deviations (z-score)
617    #[serde(default = "default_normalization_outlier_threshold")]
618    pub outlier_threshold: f64,
619
620    /// Enable metrics emission for normalization operations
621    #[serde(default = "default_true")]
622    pub emit_metrics: bool,
623}
624
625impl Default for NormalizationConfig {
626    fn default() -> Self {
627        Self {
628            enabled: false,
629            interval: default_normalization_interval(),
630            fill_strategy: FillStrategy::default(),
631            alignment: AlignmentStrategy::default(),
632            buffer_size: default_normalization_buffer_size(),
633            interpolation_max_gap: default_interpolation_max_gap(),
634            drop_outliers: false,
635            outlier_threshold: default_normalization_outlier_threshold(),
636            emit_metrics: true,
637        }
638    }
639}
640
641impl NormalizationConfig {
642    /// Create a new normalization configuration with defaults
643    pub fn new() -> Self {
644        Self::default()
645    }
646
647    /// Enable normalization
648    pub fn enabled(mut self) -> Self {
649        self.enabled = true;
650        self
651    }
652
653    /// Set the normalization interval
654    ///
655    /// # Example
656    ///
657    /// ```rust
658    /// use processor::config::NormalizationConfig;
659    /// use std::time::Duration;
660    ///
661    /// let config = NormalizationConfig::new()
662    ///     .with_interval(Duration::from_secs(5)); // Normalize to 5-second intervals
663    /// ```
664    pub fn with_interval(mut self, interval: Duration) -> Self {
665        self.interval = interval;
666        self
667    }
668
669    /// Set the fill strategy
670    pub fn with_fill_strategy(mut self, strategy: FillStrategy) -> Self {
671        self.fill_strategy = strategy;
672        self
673    }
674
675    /// Set the alignment strategy
676    pub fn with_alignment(mut self, alignment: AlignmentStrategy) -> Self {
677        self.alignment = alignment;
678        self
679    }
680
681    /// Set the buffer size for out-of-order events
682    pub fn with_buffer_size(mut self, size: usize) -> Self {
683        self.buffer_size = size;
684        self
685    }
686
687    /// Set the maximum gap for interpolation
688    pub fn with_interpolation_max_gap(mut self, gap: Duration) -> Self {
689        self.interpolation_max_gap = gap;
690        self
691    }
692
693    /// Enable outlier detection and removal
694    pub fn with_outlier_detection(mut self, threshold: f64) -> Self {
695        self.drop_outliers = true;
696        self.outlier_threshold = threshold;
697        self
698    }
699
700    /// Enable or disable metrics emission
701    pub fn with_metrics(mut self, enabled: bool) -> Self {
702        self.emit_metrics = enabled;
703        self
704    }
705
706    /// Validate normalization configuration
707    pub fn validate(&self) -> Result<()> {
708        if !self.enabled {
709            return Ok(());
710        }
711
712        if self.interval.as_millis() == 0 {
713            return Err(ProcessorError::Configuration {
714                source: "normalization interval must be greater than 0".into(),
715            });
716        }
717
718        if self.buffer_size == 0 {
719            return Err(ProcessorError::Configuration {
720                source: "normalization buffer_size must be greater than 0".into(),
721            });
722        }
723
724        if self.interpolation_max_gap.as_millis() == 0 {
725            return Err(ProcessorError::Configuration {
726                source: "interpolation_max_gap must be greater than 0".into(),
727            });
728        }
729
730        if self.outlier_threshold <= 0.0 {
731            return Err(ProcessorError::Configuration {
732                source: "outlier_threshold must be positive".into(),
733            });
734        }
735
736        // Validate that interpolation gap is reasonable relative to interval
737        if self.interpolation_max_gap < self.interval {
738            return Err(ProcessorError::Configuration {
739                source: "interpolation_max_gap should be >= interval".into(),
740            });
741        }
742
743        Ok(())
744    }
745
746    /// Get interval in milliseconds
747    pub fn interval_ms(&self) -> u64 {
748        self.interval.as_millis() as u64
749    }
750
751    /// Get interpolation max gap in milliseconds
752    pub fn interpolation_max_gap_ms(&self) -> u64 {
753        self.interpolation_max_gap.as_millis() as u64
754    }
755}
756
757/// Strategy for filling missing data points in time-series normalization
758#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
759#[serde(rename_all = "snake_case")]
760pub enum FillStrategy {
761    /// Fill with zero value
762    Zero,
763
764    /// Forward fill - use last known value
765    ForwardFill,
766
767    /// Backward fill - use next known value
768    BackwardFill,
769
770    /// Linear interpolation between known values
771    LinearInterpolation,
772
773    /// Drop missing intervals (no fill)
774    Drop,
775
776    /// Fill with a specific constant value
777    Constant,
778
779    /// Use mean of surrounding values
780    Mean,
781}
782
783impl Default for FillStrategy {
784    fn default() -> Self {
785        Self::ForwardFill
786    }
787}
788
789/// Strategy for aligning event timestamps to intervals
790#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
791#[serde(rename_all = "snake_case")]
792pub enum AlignmentStrategy {
793    /// Align to the start of the interval
794    /// Example: 10:23:47 with 5s interval -> 10:23:45
795    Start,
796
797    /// Align to the end of the interval
798    /// Example: 10:23:47 with 5s interval -> 10:23:50
799    End,
800
801    /// Align to the center of the interval
802    /// Example: 10:23:47 with 5s interval -> 10:23:47.5
803    Center,
804
805    /// Round to nearest interval boundary
806    /// Example: 10:23:47 with 5s interval -> 10:23:45 (closer to 45 than 50)
807    Nearest,
808}
809
810impl Default for AlignmentStrategy {
811    fn default() -> Self {
812        Self::Start
813    }
814}
815
816/// Event deduplication configuration
817///
818/// Provides configuration for detecting and filtering duplicate events in the stream.
819/// When enabled, events are deduplicated based on their content hash using a Redis-backed
820/// bloom filter or exact matching store.
821///
822/// # Example
823///
824/// ```rust
825/// use processor::config::{DeduplicationConfig, DeduplicationStrategy};
826/// use std::time::Duration;
827///
828/// let config = DeduplicationConfig {
829///     enabled: true,
830///     ttl: Duration::from_secs(3600),
831///     strategy: DeduplicationStrategy::ContentHash,
832///     redis_key_prefix: "dedup".to_string(),
833///     max_retries: 3,
834///     cleanup_interval: Duration::from_secs(300),
835/// };
836/// ```
837#[derive(Debug, Clone, Serialize, Deserialize)]
838pub struct DeduplicationConfig {
839    /// Enable event deduplication
840    #[serde(default)]
841    pub enabled: bool,
842
843    /// Time-to-live for deduplication records
844    /// Events with the same signature within this window are considered duplicates
845    #[serde(
846        default = "default_deduplication_ttl",
847        serialize_with = "serialize_duration",
848        deserialize_with = "deserialize_duration"
849    )]
850    pub ttl: Duration,
851
852    /// Deduplication strategy to use
853    #[serde(default)]
854    pub strategy: DeduplicationStrategy,
855
856    /// Redis key prefix for deduplication storage
857    #[serde(default = "default_redis_key_prefix")]
858    pub redis_key_prefix: String,
859
860    /// Maximum number of retries for Redis operations
861    #[serde(default = "default_max_retries")]
862    pub max_retries: u32,
863
864    /// Interval for cleanup of expired deduplication records
865    #[serde(
866        default = "default_cleanup_interval",
867        serialize_with = "serialize_duration",
868        deserialize_with = "deserialize_duration"
869    )]
870    pub cleanup_interval: Duration,
871}
872
873impl Default for DeduplicationConfig {
874    fn default() -> Self {
875        Self {
876            enabled: false,
877            ttl: default_deduplication_ttl(),
878            strategy: DeduplicationStrategy::default(),
879            redis_key_prefix: default_redis_key_prefix(),
880            max_retries: default_max_retries(),
881            cleanup_interval: default_cleanup_interval(),
882        }
883    }
884}
885
886impl DeduplicationConfig {
887    /// Create a new deduplication configuration with defaults
888    pub fn new() -> Self {
889        Self::default()
890    }
891
892    /// Enable deduplication
893    pub fn enabled(mut self) -> Self {
894        self.enabled = true;
895        self
896    }
897
898    /// Set the TTL for deduplication records
899    pub fn with_ttl(mut self, ttl: Duration) -> Self {
900        self.ttl = ttl;
901        self
902    }
903
904    /// Set the deduplication strategy
905    pub fn with_strategy(mut self, strategy: DeduplicationStrategy) -> Self {
906        self.strategy = strategy;
907        self
908    }
909
910    /// Set the Redis key prefix
911    pub fn with_redis_key_prefix<S: Into<String>>(mut self, prefix: S) -> Self {
912        self.redis_key_prefix = prefix.into();
913        self
914    }
915
916    /// Set maximum retries for Redis operations
917    pub fn with_max_retries(mut self, retries: u32) -> Self {
918        self.max_retries = retries;
919        self
920    }
921
922    /// Set cleanup interval
923    pub fn with_cleanup_interval(mut self, interval: Duration) -> Self {
924        self.cleanup_interval = interval;
925        self
926    }
927
928    /// Validate deduplication configuration
929    pub fn validate(&self) -> Result<()> {
930        if !self.enabled {
931            return Ok(());
932        }
933
934        if self.ttl.as_secs() == 0 {
935            return Err(ProcessorError::Configuration {
936                source: "deduplication ttl must be greater than 0".into(),
937            });
938        }
939
940        if self.redis_key_prefix.is_empty() {
941            return Err(ProcessorError::Configuration {
942                source: "deduplication redis_key_prefix cannot be empty".into(),
943            });
944        }
945
946        if self.cleanup_interval.as_secs() == 0 {
947            return Err(ProcessorError::Configuration {
948                source: "deduplication cleanup_interval must be greater than 0".into(),
949            });
950        }
951
952        Ok(())
953    }
954
955    /// Get TTL in milliseconds
956    pub fn ttl_ms(&self) -> u64 {
957        self.ttl.as_millis() as u64
958    }
959
960    /// Get cleanup interval in milliseconds
961    pub fn cleanup_interval_ms(&self) -> u64 {
962        self.cleanup_interval.as_millis() as u64
963    }
964}
965
966/// Strategy for event deduplication
967#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
968#[serde(rename_all = "snake_case")]
969pub enum DeduplicationStrategy {
970    /// Use content-based hashing (SHA-256) for deduplication
971    /// Deduplicates based on the full event payload
972    ContentHash,
973
974    /// Use event ID for exact matching
975    /// Requires events to have an ID field
976    EventId,
977
978    /// Use composite key (source + type + timestamp)
979    /// Useful for structured events with known fields
980    CompositeKey,
981}
982
983impl Default for DeduplicationStrategy {
984    fn default() -> Self {
985        Self::ContentHash
986    }
987}
988
989// Helper functions for Duration serialization/deserialization
990fn serialize_duration<S>(duration: &Duration, serializer: S) -> std::result::Result<S::Ok, S::Error>
991where
992    S: serde::Serializer,
993{
994    serializer.serialize_u64(duration.as_millis() as u64)
995}
996
997fn deserialize_duration<'de, D>(deserializer: D) -> std::result::Result<Duration, D::Error>
998where
999    D: serde::Deserializer<'de>,
1000{
1001    let millis = u64::deserialize(deserializer)?;
1002    Ok(Duration::from_millis(millis))
1003}
1004
1005// Default value functions
1006fn default_parallelism() -> usize {
1007    num_cpus::get()
1008}
1009
1010fn default_buffer_size() -> usize {
1011    10_000
1012}
1013
1014fn default_allowed_lateness() -> u64 {
1015    5_000 // 5 seconds
1016}
1017
1018fn default_max_delay() -> u64 {
1019    10_000 // 10 seconds
1020}
1021
1022fn default_idle_timeout() -> u64 {
1023    60_000 // 1 minute
1024}
1025
1026fn default_update_interval() -> u64 {
1027    1_000 // 1 second
1028}
1029
1030fn default_max_clock_skew() -> u64 {
1031    30_000 // 30 seconds
1032}
1033
1034fn default_checkpoint_interval() -> u64 {
1035    60_000 // 1 minute
1036}
1037
1038fn default_checkpoint_retention() -> usize {
1039    3
1040}
1041
1042fn default_percentiles() -> Vec<u8> {
1043    vec![50, 95, 99]
1044}
1045
1046fn default_min_samples() -> usize {
1047    1
1048}
1049
1050fn default_max_samples() -> usize {
1051    100_000
1052}
1053
1054fn default_outlier_threshold() -> f64 {
1055    3.0
1056}
1057
1058fn default_true() -> bool {
1059    true
1060}
1061
1062fn default_deduplication_ttl() -> Duration {
1063    Duration::from_secs(3600) // 1 hour
1064}
1065
1066fn default_redis_key_prefix() -> String {
1067    "dedup".to_string()
1068}
1069
1070fn default_max_retries() -> u32 {
1071    3
1072}
1073
1074fn default_cleanup_interval() -> Duration {
1075    Duration::from_secs(300) // 5 minutes
1076}
1077
1078fn default_normalization_interval() -> Duration {
1079    Duration::from_secs(1) // 1 second
1080}
1081
1082fn default_normalization_buffer_size() -> usize {
1083    1000
1084}
1085
1086fn default_interpolation_max_gap() -> Duration {
1087    Duration::from_secs(5) // 5 seconds
1088}
1089
1090fn default_normalization_outlier_threshold() -> f64 {
1091    3.0 // 3 standard deviations
1092}
1093
1094#[cfg(test)]
1095mod tests {
1096    use super::*;
1097
1098    #[test]
1099    fn test_default_processor_config() {
1100        let config = ProcessorConfig::default();
1101        assert!(config.validate().is_ok());
1102    }
1103
1104    #[test]
1105    fn test_tumbling_window_config() {
1106        let config = WindowConfig::tumbling(60_000);
1107        assert_eq!(config.window_type, WindowType::Tumbling);
1108        assert_eq!(config.size_ms, Some(60_000));
1109        assert!(config.validate().is_ok());
1110    }
1111
1112    #[test]
1113    fn test_sliding_window_config() {
1114        let config = WindowConfig::sliding(60_000, 30_000);
1115        assert_eq!(config.window_type, WindowType::Sliding);
1116        assert_eq!(config.size_ms, Some(60_000));
1117        assert_eq!(config.slide_ms, Some(30_000));
1118        assert!(config.validate().is_ok());
1119    }
1120
1121    #[test]
1122    fn test_invalid_sliding_window() {
1123        let mut config = WindowConfig::sliding(30_000, 60_000);
1124        assert!(config.validate().is_err());
1125    }
1126
1127    #[test]
1128    fn test_session_window_config() {
1129        let config = WindowConfig::session(5_000);
1130        assert_eq!(config.window_type, WindowType::Session);
1131        assert_eq!(config.gap_ms, Some(5_000));
1132        assert!(config.validate().is_ok());
1133    }
1134
1135    #[test]
1136    fn test_watermark_config_validation() {
1137        let config = WatermarkConfig::default();
1138        assert!(config.validate().is_ok());
1139
1140        let mut invalid_config = config.clone();
1141        invalid_config.update_interval_ms = 0;
1142        assert!(invalid_config.validate().is_err());
1143    }
1144
1145    #[test]
1146    fn test_state_config_validation() {
1147        let config = StateConfig::default();
1148        assert!(config.validate().is_ok());
1149
1150        let mut fs_config = config.clone();
1151        fs_config.backend = StateBackend::FileSystem;
1152        assert!(fs_config.validate().is_err()); // Missing storage_path
1153
1154        fs_config.storage_path = Some(PathBuf::from("/tmp/state"));
1155        assert!(fs_config.validate().is_ok());
1156    }
1157
1158    #[test]
1159    fn test_aggregation_config_validation() {
1160        let config = AggregationConfig::default();
1161        assert!(config.validate().is_ok());
1162
1163        let mut invalid_config = config.clone();
1164        invalid_config.percentiles.push(101);
1165        assert!(invalid_config.validate().is_err());
1166    }
1167
1168    #[test]
1169    fn test_aggregation_is_enabled() {
1170        let config = AggregationConfig::default();
1171        assert!(config.is_enabled(AggregationType::Sum));
1172        assert!(config.is_enabled(AggregationType::Average));
1173        assert!(!config.is_enabled(AggregationType::StdDev));
1174    }
1175
1176    #[test]
1177    fn test_histogram_bucket_validation() {
1178        let mut config = AggregationConfig::default();
1179        config.histogram_buckets = Some(vec![1.0, 5.0, 10.0, 50.0, 100.0]);
1180        assert!(config.validate().is_ok());
1181
1182        // Unsorted buckets
1183        config.histogram_buckets = Some(vec![1.0, 10.0, 5.0]);
1184        assert!(config.validate().is_err());
1185
1186        // Empty buckets
1187        config.histogram_buckets = Some(vec![]);
1188        assert!(config.validate().is_err());
1189    }
1190
1191    #[test]
1192    fn test_duration_conversions() {
1193        let watermark_config = WatermarkConfig::default();
1194        assert_eq!(
1195            watermark_config.max_delay(),
1196            Duration::from_millis(default_max_delay())
1197        );
1198
1199        let state_config = StateConfig::default();
1200        assert_eq!(
1201            state_config.checkpoint_interval(),
1202            Duration::from_millis(default_checkpoint_interval())
1203        );
1204    }
1205
1206    #[test]
1207    fn test_default_deduplication_config() {
1208        let config = DeduplicationConfig::default();
1209        assert!(!config.enabled);
1210        assert_eq!(config.ttl, Duration::from_secs(3600));
1211        assert_eq!(config.strategy, DeduplicationStrategy::ContentHash);
1212        assert_eq!(config.redis_key_prefix, "dedup");
1213        assert_eq!(config.max_retries, 3);
1214        assert!(config.validate().is_ok());
1215    }
1216
1217    #[test]
1218    fn test_deduplication_config_builder() {
1219        let config = DeduplicationConfig::new()
1220            .enabled()
1221            .with_ttl(Duration::from_secs(7200))
1222            .with_strategy(DeduplicationStrategy::EventId)
1223            .with_redis_key_prefix("custom_dedup")
1224            .with_max_retries(5)
1225            .with_cleanup_interval(Duration::from_secs(600));
1226
1227        assert!(config.enabled);
1228        assert_eq!(config.ttl, Duration::from_secs(7200));
1229        assert_eq!(config.strategy, DeduplicationStrategy::EventId);
1230        assert_eq!(config.redis_key_prefix, "custom_dedup");
1231        assert_eq!(config.max_retries, 5);
1232        assert_eq!(config.cleanup_interval, Duration::from_secs(600));
1233        assert!(config.validate().is_ok());
1234    }
1235
1236    #[test]
1237    fn test_deduplication_config_validation() {
1238        // Valid disabled config
1239        let config = DeduplicationConfig::default();
1240        assert!(config.validate().is_ok());
1241
1242        // Invalid TTL
1243        let mut invalid_config = DeduplicationConfig::new().enabled();
1244        invalid_config.ttl = Duration::from_secs(0);
1245        assert!(invalid_config.validate().is_err());
1246
1247        // Invalid Redis key prefix
1248        let mut invalid_config = DeduplicationConfig::new().enabled();
1249        invalid_config.redis_key_prefix = String::new();
1250        assert!(invalid_config.validate().is_err());
1251
1252        // Invalid cleanup interval
1253        let mut invalid_config = DeduplicationConfig::new().enabled();
1254        invalid_config.cleanup_interval = Duration::from_secs(0);
1255        assert!(invalid_config.validate().is_err());
1256    }
1257
1258    #[test]
1259    fn test_deduplication_strategy_default() {
1260        let strategy = DeduplicationStrategy::default();
1261        assert_eq!(strategy, DeduplicationStrategy::ContentHash);
1262    }
1263
1264    #[test]
1265    fn test_deduplication_config_ttl_ms() {
1266        let config = DeduplicationConfig::new()
1267            .with_ttl(Duration::from_secs(60));
1268        assert_eq!(config.ttl_ms(), 60_000);
1269    }
1270
1271    #[test]
1272    fn test_deduplication_config_cleanup_interval_ms() {
1273        let config = DeduplicationConfig::new()
1274            .with_cleanup_interval(Duration::from_secs(300));
1275        assert_eq!(config.cleanup_interval_ms(), 300_000);
1276    }
1277
1278    #[test]
1279    fn test_processor_config_with_deduplication() {
1280        let config = ProcessorConfig::default();
1281        assert!(!config.deduplication.enabled);
1282        assert!(config.validate().is_ok());
1283    }
1284
1285    #[test]
1286    fn test_default_normalization_config() {
1287        let config = NormalizationConfig::default();
1288        assert!(!config.enabled);
1289        assert_eq!(config.interval, Duration::from_secs(1));
1290        assert_eq!(config.fill_strategy, FillStrategy::ForwardFill);
1291        assert_eq!(config.alignment, AlignmentStrategy::Start);
1292        assert_eq!(config.buffer_size, 1000);
1293        assert!(!config.drop_outliers);
1294        assert_eq!(config.outlier_threshold, 3.0);
1295        assert!(config.validate().is_ok());
1296    }
1297
1298    #[test]
1299    fn test_normalization_config_builder() {
1300        let config = NormalizationConfig::new()
1301            .enabled()
1302            .with_interval(Duration::from_secs(5))
1303            .with_fill_strategy(FillStrategy::LinearInterpolation)
1304            .with_alignment(AlignmentStrategy::End)
1305            .with_buffer_size(2000)
1306            .with_interpolation_max_gap(Duration::from_secs(15))
1307            .with_outlier_detection(2.5)
1308            .with_metrics(false);
1309
1310        assert!(config.enabled);
1311        assert_eq!(config.interval, Duration::from_secs(5));
1312        assert_eq!(config.fill_strategy, FillStrategy::LinearInterpolation);
1313        assert_eq!(config.alignment, AlignmentStrategy::End);
1314        assert_eq!(config.buffer_size, 2000);
1315        assert_eq!(config.interpolation_max_gap, Duration::from_secs(15));
1316        assert!(config.drop_outliers);
1317        assert_eq!(config.outlier_threshold, 2.5);
1318        assert!(!config.emit_metrics);
1319        assert!(config.validate().is_ok());
1320    }
1321
1322    #[test]
1323    fn test_normalization_config_validation() {
1324        // Valid disabled config
1325        let config = NormalizationConfig::default();
1326        assert!(config.validate().is_ok());
1327
1328        // Invalid zero interval
1329        let mut invalid_config = NormalizationConfig::new().enabled();
1330        invalid_config.interval = Duration::from_secs(0);
1331        assert!(invalid_config.validate().is_err());
1332
1333        // Invalid zero buffer size
1334        let mut invalid_config = NormalizationConfig::new().enabled();
1335        invalid_config.buffer_size = 0;
1336        assert!(invalid_config.validate().is_err());
1337
1338        // Invalid zero interpolation gap
1339        let mut invalid_config = NormalizationConfig::new().enabled();
1340        invalid_config.interpolation_max_gap = Duration::from_secs(0);
1341        assert!(invalid_config.validate().is_err());
1342
1343        // Invalid negative outlier threshold
1344        let mut invalid_config = NormalizationConfig::new().enabled();
1345        invalid_config.outlier_threshold = -1.0;
1346        assert!(invalid_config.validate().is_err());
1347
1348        // Invalid interpolation gap < interval
1349        let mut invalid_config = NormalizationConfig::new()
1350            .enabled()
1351            .with_interval(Duration::from_secs(10))
1352            .with_interpolation_max_gap(Duration::from_secs(5));
1353        assert!(invalid_config.validate().is_err());
1354    }
1355
1356    #[test]
1357    fn test_fill_strategy_default() {
1358        let strategy = FillStrategy::default();
1359        assert_eq!(strategy, FillStrategy::ForwardFill);
1360    }
1361
1362    #[test]
1363    fn test_alignment_strategy_default() {
1364        let strategy = AlignmentStrategy::default();
1365        assert_eq!(strategy, AlignmentStrategy::Start);
1366    }
1367
1368    #[test]
1369    fn test_normalization_config_interval_ms() {
1370        let config = NormalizationConfig::new()
1371            .with_interval(Duration::from_secs(5));
1372        assert_eq!(config.interval_ms(), 5_000);
1373    }
1374
1375    #[test]
1376    fn test_normalization_config_interpolation_max_gap_ms() {
1377        let config = NormalizationConfig::new()
1378            .with_interpolation_max_gap(Duration::from_secs(10));
1379        assert_eq!(config.interpolation_max_gap_ms(), 10_000);
1380    }
1381
1382    #[test]
1383    fn test_processor_config_with_normalization() {
1384        let config = ProcessorConfig::default();
1385        assert!(!config.normalization.enabled);
1386        assert!(config.validate().is_ok());
1387    }
1388
1389    #[test]
1390    fn test_fill_strategies() {
1391        assert_eq!(FillStrategy::Zero, FillStrategy::Zero);
1392        assert_ne!(FillStrategy::Zero, FillStrategy::ForwardFill);
1393
1394        let strategies = vec![
1395            FillStrategy::Zero,
1396            FillStrategy::ForwardFill,
1397            FillStrategy::BackwardFill,
1398            FillStrategy::LinearInterpolation,
1399            FillStrategy::Drop,
1400            FillStrategy::Constant,
1401            FillStrategy::Mean,
1402        ];
1403
1404        for strategy in strategies {
1405            let serialized = serde_json::to_string(&strategy).unwrap();
1406            let deserialized: FillStrategy = serde_json::from_str(&serialized).unwrap();
1407            assert_eq!(strategy, deserialized);
1408        }
1409    }
1410
1411    #[test]
1412    fn test_alignment_strategies() {
1413        let strategies = vec![
1414            AlignmentStrategy::Start,
1415            AlignmentStrategy::End,
1416            AlignmentStrategy::Center,
1417            AlignmentStrategy::Nearest,
1418        ];
1419
1420        for strategy in strategies {
1421            let serialized = serde_json::to_string(&strategy).unwrap();
1422            let deserialized: AlignmentStrategy = serde_json::from_str(&serialized).unwrap();
1423            assert_eq!(strategy, deserialized);
1424        }
1425    }
1426}
llm_optimizer_processor/config.rs

llm_optimizer_processor/
config.rs
