rust_ethernet_ip/
monitoring.rs

1use crate::error::EtherNetIpError;
2use serde::{Deserialize, Serialize};
3use std::sync::Arc;
4use std::time::{Duration, Instant, SystemTime};
5use tokio::sync::RwLock;
6
7/// Production monitoring metrics for the EtherNet/IP library
8#[derive(Debug, Clone, Serialize, Deserialize)]
9pub struct MonitoringMetrics {
10    /// Connection statistics
11    pub connections: ConnectionMetrics,
12    /// Operation statistics
13    pub operations: OperationMetrics,
14    /// Performance statistics
15    pub performance: PerformanceMetrics,
16    /// Error statistics
17    pub errors: ErrorMetrics,
18    /// System health
19    pub health: HealthMetrics,
20}
21
22impl MonitoringMetrics {
23    /// Returns true because CPU/memory metrics in this legacy monitor are placeholders.
24    pub fn system_metrics_are_placeholders(&self) -> bool {
25        true
26    }
27}
28
29#[derive(Debug, Clone, Serialize, Deserialize)]
30pub struct ConnectionMetrics {
31    pub active_connections: u32,
32    pub total_connections: u64,
33    pub failed_connections: u64,
34    pub connection_uptime_avg: Duration,
35    pub last_connection_time: Option<SystemTime>,
36}
37
38#[derive(Debug, Clone, Serialize, Deserialize)]
39pub struct OperationMetrics {
40    pub total_reads: u64,
41    pub total_writes: u64,
42    pub successful_reads: u64,
43    pub successful_writes: u64,
44    pub failed_reads: u64,
45    pub failed_writes: u64,
46    pub batch_operations: u64,
47    pub subscription_updates: u64,
48    pub partial_batch_failures: u64,
49    pub last_successful_read_time: Option<SystemTime>,
50    pub last_failed_read_time: Option<SystemTime>,
51    pub last_successful_write_time: Option<SystemTime>,
52    pub last_failed_write_time: Option<SystemTime>,
53}
54
55#[derive(Debug, Clone, Serialize, Deserialize)]
56pub struct PerformanceMetrics {
57    pub avg_read_latency_ms: f64,
58    pub avg_write_latency_ms: f64,
59    pub max_read_latency_ms: f64,
60    pub max_write_latency_ms: f64,
61    pub reads_per_second: f64,
62    pub writes_per_second: f64,
63    pub memory_usage_mb: f64,
64    pub cpu_usage_percent: f64,
65}
66
67#[derive(Debug, Clone, Serialize, Deserialize)]
68pub struct ErrorMetrics {
69    pub network_errors: u64,
70    pub protocol_errors: u64,
71    pub timeout_errors: u64,
72    pub tag_not_found_errors: u64,
73    pub data_type_errors: u64,
74    pub session_errors: u64,
75    pub route_path_errors: u64,
76    pub embedded_service_errors: u64,
77    pub known_controller_limitation_errors: u64,
78    pub retriable_errors: u64,
79    pub non_retriable_errors: u64,
80    pub last_error_time: Option<SystemTime>,
81    pub last_error_message: Option<String>,
82    pub last_error_category: Option<ErrorCategory>,
83    pub last_retriable_error_time: Option<SystemTime>,
84}
85
86#[derive(Debug, Clone, Serialize, Deserialize)]
87pub struct HealthMetrics {
88    pub overall_health: HealthStatus,
89    pub last_health_check: SystemTime,
90    pub health_mode: HealthCheckMode,
91    pub last_verified_health_check: Option<SystemTime>,
92    pub consecutive_failures: u32,
93    pub recovery_attempts: u32,
94    pub system_uptime: Duration,
95    pub last_success_time: Option<SystemTime>,
96    pub last_failure_time: Option<SystemTime>,
97}
98
99#[derive(Debug, Clone, Serialize, Deserialize)]
100#[non_exhaustive]
101pub enum HealthStatus {
102    Healthy,
103    Warning,
104    Critical,
105    Unknown,
106}
107
108#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
109#[non_exhaustive]
110pub enum HealthCheckMode {
111    Passive,
112    Verified,
113}
114
115#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
116#[non_exhaustive]
117pub enum ErrorCategory {
118    Network,
119    Timeout,
120    Session,
121    RoutePath,
122    CipProtocol,
123    BatchEmbeddedService,
124    KnownControllerLimitation,
125    DataType,
126    NotFound,
127    Unknown,
128}
129
130impl ErrorCategory {
131    pub fn is_retriable(self) -> bool {
132        matches!(
133            self,
134            ErrorCategory::Network | ErrorCategory::Timeout | ErrorCategory::Session
135        )
136    }
137}
138
139#[derive(Debug, Clone, Serialize, Deserialize)]
140pub struct DiagnosticsSnapshot {
141    pub captured_at: SystemTime,
142    pub connections: ConnectionMetrics,
143    pub operations: OperationMetrics,
144    pub performance: PerformanceMetrics,
145    pub errors: ErrorMetrics,
146    pub health: HealthMetrics,
147    pub system_metrics_are_placeholders: bool,
148}
149
150/// Production monitoring system for EtherNet/IP operations
151#[deprecated(
152    since = "1.2.0",
153    note = "ProductionMonitor is a standalone placeholder not wired into EipClient; use EipClient diagnostics snapshots instead. The type will be removed in 2.0."
154)]
155pub struct ProductionMonitor {
156    metrics: Arc<RwLock<MonitoringMetrics>>,
157    start_time: Instant,
158}
159
160#[expect(
161    deprecated,
162    reason = "CODEX-AQ keeps ProductionMonitor compatibility until 2.0 removal"
163)]
164impl Default for ProductionMonitor {
165    fn default() -> Self {
166        Self::new()
167    }
168}
169
170#[expect(
171    deprecated,
172    reason = "CODEX-AQ keeps ProductionMonitor compatibility until 2.0 removal"
173)]
174impl ProductionMonitor {
175    pub fn new() -> Self {
176        Self {
177            metrics: Arc::new(RwLock::new(MonitoringMetrics {
178                connections: ConnectionMetrics {
179                    active_connections: 0,
180                    total_connections: 0,
181                    failed_connections: 0,
182                    connection_uptime_avg: Duration::ZERO,
183                    last_connection_time: None,
184                },
185                operations: OperationMetrics {
186                    total_reads: 0,
187                    total_writes: 0,
188                    successful_reads: 0,
189                    successful_writes: 0,
190                    failed_reads: 0,
191                    failed_writes: 0,
192                    batch_operations: 0,
193                    subscription_updates: 0,
194                    partial_batch_failures: 0,
195                    last_successful_read_time: None,
196                    last_failed_read_time: None,
197                    last_successful_write_time: None,
198                    last_failed_write_time: None,
199                },
200                performance: PerformanceMetrics {
201                    avg_read_latency_ms: 0.0,
202                    avg_write_latency_ms: 0.0,
203                    max_read_latency_ms: 0.0,
204                    max_write_latency_ms: 0.0,
205                    reads_per_second: 0.0,
206                    writes_per_second: 0.0,
207                    memory_usage_mb: 0.0,
208                    cpu_usage_percent: 0.0,
209                },
210                errors: ErrorMetrics {
211                    network_errors: 0,
212                    protocol_errors: 0,
213                    timeout_errors: 0,
214                    tag_not_found_errors: 0,
215                    data_type_errors: 0,
216                    session_errors: 0,
217                    route_path_errors: 0,
218                    embedded_service_errors: 0,
219                    known_controller_limitation_errors: 0,
220                    retriable_errors: 0,
221                    non_retriable_errors: 0,
222                    last_error_time: None,
223                    last_error_message: None,
224                    last_error_category: None,
225                    last_retriable_error_time: None,
226                },
227                health: HealthMetrics {
228                    overall_health: HealthStatus::Unknown,
229                    last_health_check: SystemTime::now(),
230                    health_mode: HealthCheckMode::Passive,
231                    last_verified_health_check: None,
232                    consecutive_failures: 0,
233                    recovery_attempts: 0,
234                    system_uptime: Duration::ZERO,
235                    last_success_time: None,
236                    last_failure_time: None,
237                },
238            })),
239            start_time: Instant::now(),
240        }
241    }
242
243    /// Record a successful read operation
244    pub async fn record_read_success(&self, latency: Duration) {
245        let mut metrics = self.metrics.write().await;
246        metrics.operations.total_reads += 1;
247        metrics.operations.successful_reads += 1;
248        let now = SystemTime::now();
249        metrics.operations.last_successful_read_time = Some(now);
250        metrics.health.last_success_time = Some(now);
251        metrics.health.consecutive_failures = 0;
252
253        // Update latency metrics
254        let latency_ms = latency.as_millis() as f64;
255        metrics.performance.avg_read_latency_ms = (metrics.performance.avg_read_latency_ms
256            * (metrics.operations.successful_reads - 1) as f64
257            + latency_ms)
258            / metrics.operations.successful_reads as f64;
259
260        if latency_ms > metrics.performance.max_read_latency_ms {
261            metrics.performance.max_read_latency_ms = latency_ms;
262        }
263    }
264
265    /// Record a failed read operation
266    pub async fn record_read_failure(&self, error_type: &str) {
267        let mut metrics = self.metrics.write().await;
268        metrics.operations.total_reads += 1;
269        metrics.operations.failed_reads += 1;
270        metrics.operations.last_failed_read_time = Some(SystemTime::now());
271        self.record_error(&mut metrics, error_type);
272    }
273
274    /// Record a successful write operation
275    pub async fn record_write_success(&self, latency: Duration) {
276        let mut metrics = self.metrics.write().await;
277        metrics.operations.total_writes += 1;
278        metrics.operations.successful_writes += 1;
279        let now = SystemTime::now();
280        metrics.operations.last_successful_write_time = Some(now);
281        metrics.health.last_success_time = Some(now);
282        metrics.health.consecutive_failures = 0;
283
284        // Update latency metrics
285        let latency_ms = latency.as_millis() as f64;
286        metrics.performance.avg_write_latency_ms = (metrics.performance.avg_write_latency_ms
287            * (metrics.operations.successful_writes - 1) as f64
288            + latency_ms)
289            / metrics.operations.successful_writes as f64;
290
291        if latency_ms > metrics.performance.max_write_latency_ms {
292            metrics.performance.max_write_latency_ms = latency_ms;
293        }
294    }
295
296    /// Record a failed write operation
297    pub async fn record_write_failure(&self, error_type: &str) {
298        let mut metrics = self.metrics.write().await;
299        metrics.operations.total_writes += 1;
300        metrics.operations.failed_writes += 1;
301        metrics.operations.last_failed_write_time = Some(SystemTime::now());
302        self.record_error(&mut metrics, error_type);
303    }
304
305    /// Record a partial batch failure without losing successful values.
306    pub async fn record_partial_batch_failure(&self, error_type: &str) {
307        let mut metrics = self.metrics.write().await;
308        metrics.operations.batch_operations += 1;
309        metrics.operations.partial_batch_failures += 1;
310        self.record_error(&mut metrics, error_type);
311    }
312
313    /// Record a connection event
314    pub async fn record_connection(&self, success: bool) {
315        let mut metrics = self.metrics.write().await;
316        if success {
317            metrics.connections.total_connections += 1;
318            metrics.connections.active_connections += 1;
319            metrics.connections.last_connection_time = Some(SystemTime::now());
320        } else {
321            metrics.connections.failed_connections += 1;
322        }
323    }
324
325    /// Record a disconnection event
326    pub async fn record_disconnection(&self) {
327        let mut metrics = self.metrics.write().await;
328        if metrics.connections.active_connections > 0 {
329            metrics.connections.active_connections -= 1;
330        }
331    }
332
333    /// Record an error
334    fn record_error(&self, metrics: &mut MonitoringMetrics, error_type: &str) {
335        let category = Self::classify_error_type(error_type);
336        let now = SystemTime::now();
337
338        match category {
339            ErrorCategory::Network => metrics.errors.network_errors += 1,
340            ErrorCategory::Timeout => metrics.errors.timeout_errors += 1,
341            ErrorCategory::Session => metrics.errors.session_errors += 1,
342            ErrorCategory::RoutePath => metrics.errors.route_path_errors += 1,
343            ErrorCategory::CipProtocol => metrics.errors.protocol_errors += 1,
344            ErrorCategory::BatchEmbeddedService => {
345                metrics.errors.protocol_errors += 1;
346                metrics.errors.embedded_service_errors += 1;
347            }
348            ErrorCategory::KnownControllerLimitation => {
349                metrics.errors.protocol_errors += 1;
350                metrics.errors.known_controller_limitation_errors += 1;
351            }
352            ErrorCategory::DataType => metrics.errors.data_type_errors += 1,
353            ErrorCategory::NotFound => metrics.errors.tag_not_found_errors += 1,
354            ErrorCategory::Unknown => {}
355        }
356
357        if category.is_retriable() {
358            metrics.errors.retriable_errors += 1;
359            metrics.errors.last_retriable_error_time = Some(now);
360        } else {
361            metrics.errors.non_retriable_errors += 1;
362        }
363
364        metrics.errors.last_error_time = Some(now);
365        metrics.errors.last_error_message = Some(error_type.to_string());
366        metrics.errors.last_error_category = Some(category);
367        metrics.health.consecutive_failures += 1;
368        metrics.health.last_failure_time = Some(now);
369    }
370
371    pub fn classify_error(error: &EtherNetIpError) -> ErrorCategory {
372        match error {
373            EtherNetIpError::Io(_) => ErrorCategory::Network,
374            EtherNetIpError::Timeout(_) => ErrorCategory::Timeout,
375            EtherNetIpError::Connection(_) | EtherNetIpError::ConnectionLost(_) => {
376                ErrorCategory::Session
377            }
378            EtherNetIpError::TagNotFound(_) => ErrorCategory::NotFound,
379            EtherNetIpError::DataTypeMismatch { .. } => ErrorCategory::DataType,
380            EtherNetIpError::CipError { code, message }
381            | EtherNetIpError::ReadError {
382                status: code,
383                message,
384            }
385            | EtherNetIpError::WriteError {
386                status: code,
387                message,
388            } => Self::classify_status_and_message(Some(*code), message),
389            EtherNetIpError::Protocol(message)
390            | EtherNetIpError::InvalidResponse { reason: message }
391            | EtherNetIpError::Other(message)
392            | EtherNetIpError::Tag(message)
393            | EtherNetIpError::Subscription(message)
394            | EtherNetIpError::Udt(message)
395            | EtherNetIpError::Permission(message)
396            | EtherNetIpError::InvalidString { reason: message } => {
397                Self::classify_status_and_message(None, message)
398            }
399            EtherNetIpError::Unsupported { .. } => ErrorCategory::CipProtocol,
400            EtherNetIpError::StringTooLong { .. } => ErrorCategory::DataType,
401            EtherNetIpError::Utf8(_) => ErrorCategory::DataType,
402        }
403    }
404
405    pub fn classify_error_type(error_type: &str) -> ErrorCategory {
406        match error_type {
407            "network" => ErrorCategory::Network,
408            "timeout" => ErrorCategory::Timeout,
409            "tag_not_found" => ErrorCategory::NotFound,
410            "data_type" => ErrorCategory::DataType,
411            "session" => ErrorCategory::Session,
412            "route_path" => ErrorCategory::RoutePath,
413            "embedded_service" => ErrorCategory::BatchEmbeddedService,
414            "known_controller_limitation" => ErrorCategory::KnownControllerLimitation,
415            "protocol" => ErrorCategory::CipProtocol,
416            other => Self::classify_status_and_message(None, other),
417        }
418    }
419
420    fn classify_status_and_message(status: Option<u8>, message: &str) -> ErrorCategory {
421        let lower = message.to_ascii_lowercase();
422
423        if status == Some(0x1E) || lower.contains("embedded service error") {
424            return ErrorCategory::BatchEmbeddedService;
425        }
426        if lower.contains("controller rejected")
427            || lower.contains("does not support writing to udt array element members")
428        {
429            return ErrorCategory::KnownControllerLimitation;
430        }
431        if status == Some(0x04) || lower.contains("path segment error") || lower.contains("route") {
432            return ErrorCategory::RoutePath;
433        }
434        if lower.contains("timed out") || lower.contains("timeout") {
435            return ErrorCategory::Timeout;
436        }
437        if lower.contains("connection lost")
438            || lower.contains("plc unreachable")
439            || lower.contains("session")
440            || lower.contains("keep-alive")
441        {
442            return ErrorCategory::Session;
443        }
444        if lower.contains("tag not found") {
445            return ErrorCategory::NotFound;
446        }
447        if lower.contains("data type")
448            || lower.contains("data-type")
449            || lower.contains("0x2107")
450            || lower.contains("invalid string")
451            || lower.contains("utf-8")
452        {
453            return ErrorCategory::DataType;
454        }
455        if lower.contains("io error") || lower.contains("network") {
456            return ErrorCategory::Network;
457        }
458        if status.is_some() || lower.contains("cip error") || lower.contains("protocol") {
459            return ErrorCategory::CipProtocol;
460        }
461
462        ErrorCategory::Unknown
463    }
464
465    /// Get current metrics
466    pub async fn get_metrics(&self) -> MonitoringMetrics {
467        let mut metrics = self.metrics.read().await.clone();
468
469        // Update system uptime
470        metrics.health.system_uptime = self.start_time.elapsed();
471
472        // Calculate operations per second
473        let total_time = metrics.health.system_uptime.as_secs_f64();
474        if total_time > 0.0 {
475            metrics.performance.reads_per_second =
476                metrics.operations.successful_reads as f64 / total_time;
477            metrics.performance.writes_per_second =
478                metrics.operations.successful_writes as f64 / total_time;
479        }
480
481        // Update health status
482        metrics.health.overall_health = self.calculate_health_status(&metrics);
483        metrics.health.last_health_check = SystemTime::now();
484        if metrics.health.last_verified_health_check.is_none() {
485            metrics.health.health_mode = HealthCheckMode::Passive;
486        }
487
488        metrics
489    }
490
491    /// Get a stable diagnostics snapshot for wrappers and service layers.
492    pub async fn get_diagnostics_snapshot(&self) -> DiagnosticsSnapshot {
493        let metrics = self.get_metrics().await;
494        DiagnosticsSnapshot {
495            captured_at: SystemTime::now(),
496            connections: metrics.connections,
497            operations: metrics.operations,
498            performance: metrics.performance,
499            errors: metrics.errors,
500            health: metrics.health,
501            system_metrics_are_placeholders: true,
502        }
503    }
504
505    /// Calculate overall health status
506    fn calculate_health_status(&self, metrics: &MonitoringMetrics) -> HealthStatus {
507        let error_rate = if metrics.operations.total_reads + metrics.operations.total_writes > 0 {
508            (metrics.operations.failed_reads + metrics.operations.failed_writes) as f64
509                / (metrics.operations.total_reads + metrics.operations.total_writes) as f64
510        } else {
511            0.0
512        };
513
514        if error_rate > 0.1 || metrics.health.consecutive_failures > 10 {
515            HealthStatus::Critical
516        } else if error_rate > 0.05 || metrics.health.consecutive_failures > 5 {
517            HealthStatus::Warning
518        } else if metrics.connections.active_connections > 0 {
519            HealthStatus::Healthy
520        } else {
521            HealthStatus::Unknown
522        }
523    }
524
525    /// Start monitoring background tasks
526    pub async fn start_monitoring(&self) {
527        tracing::warn!(
528            "ProductionMonitor::start_monitoring is deprecated and no longer spawns a placeholder metrics task"
529        );
530    }
531
532    /// Reset consecutive failures (call after successful recovery)
533    pub async fn reset_consecutive_failures(&self) {
534        let mut metrics = self.metrics.write().await;
535        metrics.health.consecutive_failures = 0;
536        metrics.health.recovery_attempts += 1;
537    }
538
539    /// Record the outcome of an active, verified health check.
540    pub async fn record_verified_health_check(&self, is_healthy: bool) {
541        let mut metrics = self.metrics.write().await;
542        let now = SystemTime::now();
543        metrics.health.health_mode = HealthCheckMode::Verified;
544        metrics.health.last_verified_health_check = Some(now);
545        metrics.health.last_health_check = now;
546
547        if is_healthy {
548            metrics.health.last_success_time = Some(now);
549            metrics.health.consecutive_failures = 0;
550        } else {
551            metrics.health.last_failure_time = Some(now);
552            metrics.health.consecutive_failures += 1;
553        }
554    }
555}
556
557#[expect(
558    deprecated,
559    reason = "CODEX-AQ keeps ProductionMonitor compatibility until 2.0 removal"
560)]
561impl Clone for ProductionMonitor {
562    fn clone(&self) -> Self {
563        Self {
564            metrics: Arc::clone(&self.metrics),
565            start_time: self.start_time,
566        }
567    }
568}
569
570#[cfg(test)]
571#[expect(
572    deprecated,
573    reason = "CODEX-AQ keeps ProductionMonitor unit coverage until 2.0 removal"
574)]
575mod tests {
576    use super::*;
577    use crate::error::EtherNetIpError;
578
579    #[test]
580    fn classify_timeout_and_route_path_errors() {
581        assert_eq!(
582            ProductionMonitor::classify_error(&EtherNetIpError::Timeout(Duration::from_secs(1))),
583            ErrorCategory::Timeout
584        );
585        assert_eq!(
586            ProductionMonitor::classify_error(&EtherNetIpError::Protocol(
587                "Path segment error while resolving route".to_string()
588            )),
589            ErrorCategory::RoutePath
590        );
591    }
592
593    #[test]
594    fn classify_known_controller_limitation_and_embedded_service() {
595        assert_eq!(
596            ProductionMonitor::classify_error(&EtherNetIpError::Protocol(
597                "Read/Write Tag data-type mismatch extended error: 0x2107".to_string()
598            )),
599            ErrorCategory::DataType
600        );
601        assert_eq!(
602            ProductionMonitor::classify_error(&EtherNetIpError::WriteError {
603                status: 0x1E,
604                message: "Embedded service error".to_string(),
605            }),
606            ErrorCategory::BatchEmbeddedService
607        );
608    }
609
610    #[tokio::test]
611    async fn diagnostics_snapshot_distinguishes_verified_health() {
612        let monitor = ProductionMonitor::new();
613        monitor.record_read_success(Duration::from_millis(10)).await;
614
615        let passive = monitor.get_diagnostics_snapshot().await;
616        assert_eq!(passive.health.health_mode, HealthCheckMode::Passive);
617        assert!(passive.health.last_verified_health_check.is_none());
618        assert!(passive.operations.last_successful_read_time.is_some());
619
620        monitor.record_verified_health_check(true).await;
621        let verified = monitor.get_diagnostics_snapshot().await;
622        assert_eq!(verified.health.health_mode, HealthCheckMode::Verified);
623        assert!(verified.health.last_verified_health_check.is_some());
624        assert!(verified.system_metrics_are_placeholders);
625    }
626}
rust_ethernet_ip/monitoring.rs

rust_ethernet_ip/
monitoring.rs
